public void mergeLoadDescToOriginStatement(RoutineLoadDesc routineLoadDesc) {
        if (origStmt == null) {
            return;
        }

        RoutineLoadDesc originLoadDesc = CreateRoutineLoadStmt.getLoadDesc(origStmt, sessionVariables);
        if (originLoadDesc == null) {
            originLoadDesc = new RoutineLoadDesc();
        }
        if (routineLoadDesc.getColumnSeparator() != null) {
            originLoadDesc.setColumnSeparator(routineLoadDesc.getColumnSeparator());
        }
        if (routineLoadDesc.getRowDelimiter() != null) {
            originLoadDesc.setRowDelimiter(routineLoadDesc.getRowDelimiter());
        }
        if (routineLoadDesc.getColumnsInfo() != null) {
            originLoadDesc.setColumnsInfo(routineLoadDesc.getColumnsInfo());
        }
        if (routineLoadDesc.getWherePredicate() != null) {
            originLoadDesc.setWherePredicate(routineLoadDesc.getWherePredicate());
        }
        if (routineLoadDesc.getPartitionNames() != null) {
            originLoadDesc.setPartitionNames(routineLoadDesc.getPartitionNames());
        }

        String tableName = null;
        try {
            tableName = getTableName();
        } catch (Exception e) {
            LOG.warn("get table name failed", e);
            tableName = "unknown";
        }

        // we use sql to persist the load properties, so we just put the load properties to sql.
        String sql = String.format("CREATE ROUTINE LOAD %s ON %s %s" +
                        " PROPERTIES (\"desired_concurrent_number\"=\"1\")" +
                        " FROM KAFKA (\"kafka_topic\" = \"my_topic\")",
                name, tableName, originLoadDesc.toSql());
        LOG.debug("merge result: {}", sql);
        origStmt = new OriginStatement(sql, 0);
    }
@Test
    public void testMergeLoadDescToOriginStatement() throws Exception {
        KafkaRoutineLoadJob routineLoadJob = new KafkaRoutineLoadJob(1L, "job",
                2L, 3L, "192.168.1.2:10000", "topic");
        String originStmt = "CREATE ROUTINE LOAD job ON unknown " +
                "PROPERTIES (\"desired_concurrent_number\"=\"1\") " +
                "FROM KAFKA (\"kafka_topic\" = \"my_topic\")";
        routineLoadJob.setOrigStmt(new OriginStatement(originStmt, 0));

        // alter columns terminator
        RoutineLoadDesc loadDesc = CreateRoutineLoadStmt.getLoadDesc(new OriginStatement(
                "ALTER ROUTINE LOAD FOR job " +
                        "COLUMNS TERMINATED BY ';'", 0), null);
        routineLoadJob.mergeLoadDescToOriginStatement(loadDesc);
        Assert.assertEquals("CREATE ROUTINE LOAD job ON unknown " +
                "COLUMNS TERMINATED BY ';' " +
                "PROPERTIES (\"desired_concurrent_number\"=\"1\") " +
                "FROM KAFKA (\"kafka_topic\" = \"my_topic\")", routineLoadJob.getOrigStmt().originStmt);

        // alter rows terminator
        loadDesc = CreateRoutineLoadStmt.getLoadDesc(new OriginStatement(
                "ALTER ROUTINE LOAD FOR job " +
                        "ROWS TERMINATED BY '\n'", 0), null);
        routineLoadJob.mergeLoadDescToOriginStatement(loadDesc);
        Assert.assertEquals("CREATE ROUTINE LOAD job ON unknown " +
                "COLUMNS TERMINATED BY ';', " +
                "ROWS TERMINATED BY '\n' " +
                "PROPERTIES (\"desired_concurrent_number\"=\"1\") " +
                "FROM KAFKA (\"kafka_topic\" = \"my_topic\")", routineLoadJob.getOrigStmt().originStmt);

        // alter columns
        loadDesc = CreateRoutineLoadStmt.getLoadDesc(new OriginStatement(
                "ALTER ROUTINE LOAD FOR job " +
                        "COLUMNS(`a`, `b`, `c`=1)", 0), null);
        routineLoadJob.mergeLoadDescToOriginStatement(loadDesc);
        Assert.assertEquals("CREATE ROUTINE LOAD job ON unknown " +
                "COLUMNS TERMINATED BY ';', " +
                "ROWS TERMINATED BY '\n', " +
                "COLUMNS(`a`, `b`, `c` = 1) " +
                "PROPERTIES (\"desired_concurrent_number\"=\"1\") " +
                "FROM KAFKA (\"kafka_topic\" = \"my_topic\")", routineLoadJob.getOrigStmt().originStmt);

        // alter partition
        loadDesc = CreateRoutineLoadStmt.getLoadDesc(new OriginStatement(
                "ALTER ROUTINE LOAD FOR job " +
                        "TEMPORARY PARTITION(`p1`, `p2`)", 0), null);
        routineLoadJob.mergeLoadDescToOriginStatement(loadDesc);
        Assert.assertEquals("CREATE ROUTINE LOAD job ON unknown " +
                "COLUMNS TERMINATED BY ';', " +
                "ROWS TERMINATED BY '\n', " +
                "COLUMNS(`a`, `b`, `c` = 1), " +
                "TEMPORARY PARTITION(`p1`, `p2`) " +
                "PROPERTIES (\"desired_concurrent_number\"=\"1\") " +
                "FROM KAFKA (\"kafka_topic\" = \"my_topic\")", routineLoadJob.getOrigStmt().originStmt);

        // alter where
        loadDesc = CreateRoutineLoadStmt.getLoadDesc(new OriginStatement(
                "ALTER ROUTINE LOAD FOR job " +
                        "WHERE a = 1", 0), null);
        routineLoadJob.mergeLoadDescToOriginStatement(loadDesc);
        Assert.assertEquals("CREATE ROUTINE LOAD job ON unknown " +
                "COLUMNS TERMINATED BY ';', " +
                "ROWS TERMINATED BY '\n', " +
                "COLUMNS(`a`, `b`, `c` = 1), " +
                "TEMPORARY PARTITION(`p1`, `p2`), " +
                "WHERE `a` = 1 " +
                "PROPERTIES (\"desired_concurrent_number\"=\"1\") " +
                "FROM KAFKA (\"kafka_topic\" = \"my_topic\")", routineLoadJob.getOrigStmt().originStmt);

        // alter columns terminator again
        loadDesc = CreateRoutineLoadStmt.getLoadDesc(new OriginStatement(
                "ALTER ROUTINE LOAD FOR job " +
                        "COLUMNS TERMINATED BY '\t'", 0), null);
        routineLoadJob.mergeLoadDescToOriginStatement(loadDesc);
        Assert.assertEquals("CREATE ROUTINE LOAD job ON unknown " +
                "COLUMNS TERMINATED BY '\t', " +
                "ROWS TERMINATED BY '\n', " +
                "COLUMNS(`a`, `b`, `c` = 1), " +
                "TEMPORARY PARTITION(`p1`, `p2`), " +
                "WHERE `a` = 1 " +
                "PROPERTIES (\"desired_concurrent_number\"=\"1\") " +
                "FROM KAFKA (\"kafka_topic\" = \"my_topic\")", routineLoadJob.getOrigStmt().originStmt);

        // alter rows terminator again
        loadDesc = CreateRoutineLoadStmt.getLoadDesc(new OriginStatement(
                "ALTER ROUTINE LOAD FOR job " +
                        "ROWS TERMINATED BY 'a'", 0), null);
        routineLoadJob.mergeLoadDescToOriginStatement(loadDesc);
        Assert.assertEquals("CREATE ROUTINE LOAD job ON unknown " +
                "COLUMNS TERMINATED BY '\t', " +
                "ROWS TERMINATED BY 'a', " +
                "COLUMNS(`a`, `b`, `c` = 1), " +
                "TEMPORARY PARTITION(`p1`, `p2`), " +
                "WHERE `a` = 1 " +
                "PROPERTIES (\"desired_concurrent_number\"=\"1\") " +
                "FROM KAFKA (\"kafka_topic\" = \"my_topic\")", routineLoadJob.getOrigStmt().originStmt);

        // alter columns again
        loadDesc = CreateRoutineLoadStmt.getLoadDesc(new OriginStatement(
                "ALTER ROUTINE LOAD FOR job " +
                        "COLUMNS(`a`)", 0), null);
        routineLoadJob.mergeLoadDescToOriginStatement(loadDesc);
        Assert.assertEquals("CREATE ROUTINE LOAD job ON unknown " +
                "COLUMNS TERMINATED BY '\t', " +
                "ROWS TERMINATED BY 'a', " +
                "COLUMNS(`a`), " +
                "TEMPORARY PARTITION(`p1`, `p2`), " +
                "WHERE `a` = 1 " +
                "PROPERTIES (\"desired_concurrent_number\"=\"1\") " +
                "FROM KAFKA (\"kafka_topic\" = \"my_topic\")", routineLoadJob.getOrigStmt().originStmt);
        // alter partition again
        loadDesc = CreateRoutineLoadStmt.getLoadDesc(new OriginStatement(
                "ALTER ROUTINE LOAD FOR job " +
                        " PARTITION(`p1`, `p2`)", 0), null);
        routineLoadJob.mergeLoadDescToOriginStatement(loadDesc);
        Assert.assertEquals("CREATE ROUTINE LOAD job ON unknown " +
                "COLUMNS TERMINATED BY '\t', " +
                "ROWS TERMINATED BY 'a', " +
                "COLUMNS(`a`), " +
                "PARTITION(`p1`, `p2`), " +
                "WHERE `a` = 1 " +
                "PROPERTIES (\"desired_concurrent_number\"=\"1\") " +
                "FROM KAFKA (\"kafka_topic\" = \"my_topic\")", routineLoadJob.getOrigStmt().originStmt);

        // alter where again
        loadDesc = CreateRoutineLoadStmt.getLoadDesc(new OriginStatement(
                "ALTER ROUTINE LOAD FOR job " +
                        "WHERE a = 5", 0), null);
        routineLoadJob.mergeLoadDescToOriginStatement(loadDesc);
        Assert.assertEquals("CREATE ROUTINE LOAD job ON unknown " +
                "COLUMNS TERMINATED BY '\t', " +
                "ROWS TERMINATED BY 'a', " +
                "COLUMNS(`a`), " +
                "PARTITION(`p1`, `p2`), " +
                "WHERE `a` = 5 " +
                "PROPERTIES (\"desired_concurrent_number\"=\"1\") " +
                "FROM KAFKA (\"kafka_topic\" = \"my_topic\")", routineLoadJob.getOrigStmt().originStmt);

        // alter where again
        loadDesc = CreateRoutineLoadStmt.getLoadDesc(new OriginStatement(
                "ALTER ROUTINE LOAD FOR job " +
                        "WHERE a = 5 and b like 'c1%' and c between 1 and 100 and substring(d,1,5) = 'cefd' ", 0), null);
        routineLoadJob.mergeLoadDescToOriginStatement(loadDesc);
        Assert.assertEquals("CREATE ROUTINE LOAD job ON unknown " +
                "COLUMNS TERMINATED BY '\t', " +
                "ROWS TERMINATED BY 'a', " +
                "COLUMNS(`a`), " +
                "PARTITION(`p1`, `p2`), " +
                "WHERE (((`a` = 5) " +
                "AND (`b` LIKE 'c1%')) " +
                "AND (`c` BETWEEN 1 AND 100)) " +
                "AND (substring(`d`, 1, 5) = 'cefd') " +
                "PROPERTIES (\"desired_concurrent_number\"=\"1\") " +
                "FROM KAFKA (\"kafka_topic\" = \"my_topic\")", routineLoadJob.getOrigStmt().originStmt);
    }