public TExecPlanFragmentParams plan(TUniqueId loadId) throws UserException {
        boolean isPrimaryKey = destTable.getKeysType() == KeysType.PRIMARY_KEYS;
        resetAnalyzer();
        // construct tuple descriptor, used for scanNode and dataSink
        TupleDescriptor tupleDesc = descTable.createTupleDescriptor("DstTableTuple");
        boolean negative = streamLoadInfo.getNegative();
        if (isPrimaryKey) {
            if (negative) {
                throw new DdlException("Primary key table does not support negative load");
            }
            if (destTable.hasRowStorageType() && streamLoadInfo.isPartialUpdate() &&
                    streamLoadInfo.getPartialUpdateMode() != TPartialUpdateMode.ROW_MODE) {
                throw new DdlException("column with row table only support row mode partial update");
            }
        } else {
            if (streamLoadInfo.isPartialUpdate()) {
                throw new DdlException("Only primary key table support partial update");
            }
        }
        List<Pair<Integer, ColumnDict>> globalDicts = Lists.newArrayList();
        List<Column> destColumns;
        List<Boolean> missAutoIncrementColumn = Lists.newArrayList();
        if (streamLoadInfo.isPartialUpdate()) {
            destColumns = Load.getPartialUpateColumns(destTable, streamLoadInfo.getColumnExprDescs(),
                    missAutoIncrementColumn);
        } else {
            destColumns = destTable.getFullSchema();
        }
        for (Column col : destColumns) {
            SlotDescriptor slotDesc = descTable.addSlotDescriptor(tupleDesc);
            slotDesc.setIsMaterialized(true);
            slotDesc.setColumn(col);
            slotDesc.setIsNullable(col.isAllowNull());
            if (negative && !col.isKey() && col.getAggregationType() != AggregateType.SUM) {
                throw new DdlException("Column is not SUM AggreateType. column:" + col.getName());
            }

            if (col.getType().isVarchar() && Config.enable_dict_optimize_stream_load &&
                    IDictManager.getInstance().hasGlobalDict(destTable.getId(),
                            col.getColumnId())) {
                Optional<ColumnDict> dict = IDictManager.getInstance().getGlobalDict(destTable.getId(), col.getColumnId());
                dict.ifPresent(columnDict -> globalDicts.add(new Pair<>(slotDesc.getId().asInt(), columnDict)));
            }
        }
        if (isPrimaryKey) {
            // add op type column
            SlotDescriptor slotDesc = descTable.addSlotDescriptor(tupleDesc);
            slotDesc.setIsMaterialized(true);
            slotDesc.setColumn(new Column(Load.LOAD_OP_COLUMN, Type.TINYINT));
            slotDesc.setIsNullable(false);
        }

        // create scan node
        StreamLoadScanNode scanNode =
                new StreamLoadScanNode(loadId, new PlanNodeId(0), tupleDesc, destTable, streamLoadInfo);
        scanNode.setUseVectorizedLoad(true);
        scanNode.init(analyzer);
        scanNode.finalizeStats(analyzer);
        scanNode.setWarehouseId(streamLoadInfo.getWarehouseId());

        descTable.computeMemLayout();

        // create dest sink
        TWriteQuorumType writeQuorum = destTable.writeQuorum();

        List<Long> partitionIds = getAllPartitionIds();
        boolean enableAutomaticPartition;
        if (streamLoadInfo.isSpecifiedPartitions()) {
            enableAutomaticPartition = false;
        } else {
            enableAutomaticPartition = destTable.supportedAutomaticPartition();
        }
        OlapTableSink olapTableSink = new OlapTableSink(destTable, tupleDesc, partitionIds, writeQuorum,
                destTable.enableReplicatedStorage(), scanNode.nullExprInAutoIncrement(),
                enableAutomaticPartition, streamLoadInfo.getWarehouseId());
        if (missAutoIncrementColumn.size() == 1 && missAutoIncrementColumn.get(0) == Boolean.TRUE) {
            olapTableSink.setMissAutoIncrementColumn();
        }
        if (destTable.getAutomaticBucketSize() > 0) {
            olapTableSink.setAutomaticBucketSize(destTable.getAutomaticBucketSize());
        }
        olapTableSink.init(loadId, streamLoadInfo.getTxnId(), db.getId(), streamLoadInfo.getTimeout());
        Load.checkMergeCondition(streamLoadInfo.getMergeConditionStr(), destTable, destColumns,
                olapTableSink.missAutoIncrementColumn());
        olapTableSink.setPartialUpdateMode(streamLoadInfo.getPartialUpdateMode());
        olapTableSink.complete(streamLoadInfo.getMergeConditionStr());

        // for stream load, we only need one fragment, ScanNode -> DataSink.
        // OlapTableSink can dispatch data to corresponding node.
        PlanFragment fragment = new PlanFragment(new PlanFragmentId(0), scanNode, DataPartition.UNPARTITIONED);
        fragment.setSink(olapTableSink);
        // At present, we only support dop=1 for olap table sink.
        // because tablet writing needs to know the number of senders in advance
        // and guaranteed order of data writing
        // It can be parallel only in some scenes, for easy use 1 dop now.
        fragment.setPipelineDop(1);
        // After data loading, we need to check the global dict for low cardinality string column
        // whether update.
        fragment.setLoadGlobalDicts(globalDicts);

        fragment.createDataSink(TResultSinkType.MYSQL_PROTOCAL);

        TExecPlanFragmentParams params = new TExecPlanFragmentParams();
        params.setProtocol_version(InternalServiceVersion.V1);
        params.setFragment(fragment.toThrift());

        params.setDesc_tbl(analyzer.getDescTbl().toThrift());

        TPlanFragmentExecParams execParams = new TPlanFragmentExecParams();
        // user load id (streamLoadInfo.id) as query id
        execParams.setQuery_id(loadId);
        execParams.setFragment_instance_id(new TUniqueId(loadId.hi, loadId.lo + 1));
        execParams.per_exch_num_senders = Maps.newHashMap();
        execParams.destinations = Lists.newArrayList();
        Map<Integer, List<TScanRangeParams>> perNodeScanRange = Maps.newHashMap();
        List<TScanRangeParams> scanRangeParams = Lists.newArrayList();
        for (TScanRangeLocations locations : scanNode.getScanRangeLocations(0)) {
            scanRangeParams.add(new TScanRangeParams(locations.getScan_range()));
        }
        // For stream load, only one sender
        execParams.setSender_id(0);
        execParams.setNum_senders(1);
        perNodeScanRange.put(scanNode.getId().asInt(), scanRangeParams);
        execParams.setPer_node_scan_ranges(perNodeScanRange);
        params.setParams(execParams);
        TQueryOptions queryOptions = new TQueryOptions();
        queryOptions.setQuery_type(TQueryType.LOAD);
        queryOptions.setQuery_timeout(streamLoadInfo.getTimeout());
        queryOptions.setLoad_transmission_compression_type(streamLoadInfo.getTransmisionCompressionType());
        queryOptions.setLog_rejected_record_num(streamLoadInfo.getLogRejectedRecordNum());

        // Disable load_dop for LakeTable temporary, because BE's `LakeTabletsChannel` does not support
        // parallel send from a single sender.
        if (streamLoadInfo.getLoadParallelRequestNum() != 0 && !destTable.isCloudNativeTableOrMaterializedView()) {
            // only dup_keys can use parallel write since other table's the order of write is important
            if (destTable.getKeysType() == KeysType.DUP_KEYS) {
                queryOptions.setLoad_dop(streamLoadInfo.getLoadParallelRequestNum());
            } else {
                queryOptions.setLoad_dop(1);
            }
        }
        // for stream load, we use exec_mem_limit to limit the memory usage of load channel.
        queryOptions.setMem_limit(streamLoadInfo.getExecMemLimit());
        queryOptions.setLoad_mem_limit(streamLoadInfo.getLoadMemLimit());

        if (connectContext.getSessionVariable().isEnableLoadProfile()) {
            queryOptions.setEnable_profile(true);
            queryOptions.setLoad_profile_collect_second(Config.stream_load_profile_collect_second);
        }

        params.setQuery_options(queryOptions);
        TQueryGlobals queryGlobals = new TQueryGlobals();
        queryGlobals.setNow_string(DATE_FORMAT.format(new Date()));
        queryGlobals.setTimestamp_ms(new Date().getTime());
        queryGlobals.setTime_zone(streamLoadInfo.getTimezone());
        params.setQuery_globals(queryGlobals);

        // Since stream load has only one fragment,
        // the backend number can be directly assigned to 0
        params.setBackend_num(0);
        TNetworkAddress coordAddress = new TNetworkAddress(FrontendOptions.getLocalHostAddress(), Config.rpc_port);
        params.setCoord(coordAddress);

        LOG.info("load job id: {}, txn id: {}, parallel: {}, compress: {}, replicated: {}, quorum: {}",
                DebugUtil.printId(loadId), streamLoadInfo.getTxnId(), queryOptions.getLoad_dop(),
                queryOptions.getLoad_transmission_compression_type(), destTable.enableReplicatedStorage(), writeQuorum);
        this.execPlanFragmentParams = params;
        return params;
    }
@Test
    public void testNormalPlan() throws UserException {
        List<Column> columns = Lists.newArrayList();
        Column c1 = new Column("c1", Type.BIGINT, false);
        columns.add(c1);
        Column c2 = new Column("c2", Type.BIGINT, true);
        columns.add(c2);
        new Expectations() {
            {
                destTable.getBaseSchema();
                minTimes = 0;
                result = columns;
                destTable.getPartitions();
                minTimes = 0;
                result = Arrays.asList(partition);
                scanNode.init((Analyzer) any);
                minTimes = 0;
                scanNode.getChildren();
                minTimes = 0;
                result = Lists.newArrayList();
                scanNode.getId();
                minTimes = 0;
                result = new PlanNodeId(5);
                partition.getId();
                minTimes = 0;
                result = 0;
            }
        };
        TStreamLoadPutRequest request = new TStreamLoadPutRequest();
        request.setTxnId(1);
        request.setLoadId(new TUniqueId(2, 3));
        request.setFileType(TFileType.FILE_STREAM);
        request.setFormatType(TFileFormatType.FORMAT_CSV_PLAIN);
        request.setLoad_dop(2);
        request.setPayload_compression_type("LZ4_FRAME");
        StreamLoadInfo streamLoadInfo = StreamLoadInfo.fromTStreamLoadPutRequest(request, db);
        StreamLoadPlanner planner = new StreamLoadPlanner(db, destTable, streamLoadInfo);
        planner.plan(streamLoadInfo.getId());
        Assert.assertEquals(TCompressionType.LZ4_FRAME, streamLoadInfo.getPayloadCompressionType());
    }