[feature](routineload) Add routine load target-table alter support

fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java

@@ -19,6 +19,7 @@
 
 import org.apache.doris.analysis.Expr;
 import org.apache.doris.analysis.ExprToSqlVisitor;
+import org.apache.doris.analysis.ImportColumnDesc;
 import org.apache.doris.analysis.Separator;
 import org.apache.doris.analysis.ToSqlParams;
 import org.apache.doris.analysis.UserIdentity;
@@ -470,6 +471,97 @@ protected void setRoutineLoadDesc(RoutineLoadDesc routineLoadDesc) {
         }
     }
 
+    public void validateTargetTable(Database db, OlapTable targetTable) throws UserException {
+        validateTargetTable(db, targetTable, Maps.newHashMap(), uniqueKeyUpdateMode);
+    }
+
+    public void validateTargetTable(Database db, OlapTable targetTable,
+            Map<String, String> alteredJobProperties, TUniqueKeyUpdateMode effectiveUniqueKeyUpdateMode)
+            throws UserException {
+        if (isMultiTable) {
+            throw new AnalysisException("ALTER ROUTINE LOAD target table change only supports single-table job");
+        }
+        List<ImportColumnDesc> columnsInfo = null;
+        if (columnDescs != null && !columnDescs.descs.isEmpty()) {
+            columnsInfo = new ArrayList<>(columnDescs.descs);
+        }
+        checkMeta(targetTable, new RoutineLoadDesc(columnSeparator, lineDelimiter, columnsInfo, precedingFilter,
+                whereExpr, partitionNamesInfo, deleteCondition, mergeType, sequenceCol));
+        validateAlterJobProperties(targetTable, alteredJobProperties, effectiveUniqueKeyUpdateMode);
+
+        targetTable.readLock();
+        try {
+            NereidsRoutineLoadTaskInfo taskInfo = toNereidsRoutineLoadTaskInfo();
+            taskInfo.applyAlterPropertiesForValidation(alteredJobProperties, effectiveUniqueKeyUpdateMode);
+            NereidsStreamLoadPlanner planner = new NereidsStreamLoadPlanner(db, targetTable, taskInfo);
+            planner.plan(new TUniqueId(0, 0));
+        } finally {
+            targetTable.readUnlock();
+        }
+    }
+
+    public void validateAlterJobProperties(OlapTable targetTable, Map<String, String> alteredJobProperties,
+            TUniqueKeyUpdateMode effectiveUniqueKeyUpdateMode) throws UserException {
+        if (effectiveUniqueKeyUpdateMode == TUniqueKeyUpdateMode.UPSERT) {
+            return;
+        }
+        if (isMultiTable) {
+            throw new AnalysisException("Partial update is not supported in multi-table load.");
+        }
+        if (effectiveUniqueKeyUpdateMode == TUniqueKeyUpdateMode.UPDATE_FIXED_COLUMNS) {
+            if (!targetTable.getEnableUniqueKeyMergeOnWrite()) {
+                throw new AnalysisException("load by PARTIAL_COLUMNS is only supported in unique table MoW");
+            }
+            return;
+        }
+        Preconditions.checkState(effectiveUniqueKeyUpdateMode == TUniqueKeyUpdateMode.UPDATE_FLEXIBLE_COLUMNS,
+                effectiveUniqueKeyUpdateMode);
+        validateFlexiblePartialUpdateForAlter(targetTable, alteredJobProperties);
+    }
+
+    public static TUniqueKeyUpdateMode getEffectiveUniqueKeyUpdateMode(TUniqueKeyUpdateMode currentUniqueKeyUpdateMode,
+            Map<String, String> alteredJobProperties) {
+        if (alteredJobProperties.containsKey(CreateRoutineLoadInfo.UNIQUE_KEY_UPDATE_MODE)) {
+            return TUniqueKeyUpdateMode.valueOf(
+                    alteredJobProperties.get(CreateRoutineLoadInfo.UNIQUE_KEY_UPDATE_MODE));
+        }
+        if (alteredJobProperties.containsKey(CreateRoutineLoadInfo.PARTIAL_COLUMNS)
+                && Boolean.parseBoolean(alteredJobProperties.get(CreateRoutineLoadInfo.PARTIAL_COLUMNS))
+                && currentUniqueKeyUpdateMode == TUniqueKeyUpdateMode.UPSERT) {
+            return TUniqueKeyUpdateMode.UPDATE_FIXED_COLUMNS;
+        }
+        return currentUniqueKeyUpdateMode;
+    }
+
+    protected void validateAlterJobPropertiesForMutation(AlterRoutineLoadCommand command) throws UserException {
+        Map<String, String> alteredJobProperties = command.getAnalyzedJobProperties();
+        TUniqueKeyUpdateMode effectiveUniqueKeyUpdateMode = getEffectiveUniqueKeyUpdateMode(
+                uniqueKeyUpdateMode, alteredJobProperties);
+        if (!command.hasTargetTable() && effectiveUniqueKeyUpdateMode == TUniqueKeyUpdateMode.UPSERT) {
+            return;
+        }
+
+        Database db = Env.getCurrentInternalCatalog().getDbNullable(dbId);
+        if (db == null) {
+            throw new DdlException("Database not found: " + dbId);
+        }
+        long effectiveTableId = command.hasTargetTable() ? command.getTargetTableId() : tableId;
+        Table table = db.getTableNullable(effectiveTableId);
+        if (table == null) {
+            throw new DdlException("Table not found: " + effectiveTableId);
+        }
+        if (!(table instanceof OlapTable)) {
+            throw new DdlException(command.hasTargetTable()
+                    ? "Routine load target table must be an OLAP table"
+                    : "Partial update is only supported for OLAP tables");
+        }
+        if (command.hasTargetTable()) {
+            validateTargetTable(db, (OlapTable) table, alteredJobProperties, effectiveUniqueKeyUpdateMode);
+            return;
+        }
+        validateAlterJobProperties((OlapTable) table, alteredJobProperties, effectiveUniqueKeyUpdateMode);
+    }
+
     @Override
     public long getId() {
         return id;
@@ -1689,11 +1781,10 @@ public String getStateReason() {
     }
 
     public List<String> getShowInfo() {
-        Optional<Database> database = Env.getCurrentInternalCatalog().getDb(dbId);
-        Optional<Table> table = database.flatMap(db -> db.getTable(tableId));
-
         readLock();
         try {
+            Optional<Database> database = Env.getCurrentInternalCatalog().getDb(dbId);
+            Optional<Table> table = database.flatMap(db -> db.getTable(tableId));
             List<String> row = Lists.newArrayList();
             row.add(String.valueOf(id));
             row.add(name);
@@ -1756,6 +1847,15 @@ public List<List<String>> getTasksShowInfo() throws AnalysisException {
     }
 
     public String getShowCreateInfo() {
+        readLock();
+        try {
+            return unprotectGetShowCreateInfo();
+        } finally {
+            readUnlock();
+        }
+    }
+
+    private String unprotectGetShowCreateInfo() {
         Optional<Database> database = Env.getCurrentInternalCatalog().getDb(dbId);
         Optional<Table> table = database.flatMap(db -> db.getTable(tableId));
         StringBuilder sb = new StringBuilder();
@@ -2040,7 +2140,57 @@ public void gsonPostProcess() throws IOException {
         }
     }
 
-    public abstract void modifyProperties(AlterRoutineLoadCommand command) throws UserException;
+    public void modifyProperties(AlterRoutineLoadCommand command) throws UserException {
+        writeLock();
+        try {
+            if (getState() != JobState.PAUSED) {
+                throw new DdlException("Only supports modification of PAUSED jobs");
+            }
+            if (!command.hasTargetTable()) {
+                unprotectApplyAlter(command);
+                return;
+            }
+
+            Database db = Env.getCurrentInternalCatalog().getDbNullable(dbId);
+            if (db == null) {
+                throw new DdlException("Database not found: " + dbId);
+            }
+            db.readLock();
+            try {
+                Table table = db.getTableNullable(command.getTargetTableId());
+                if (table == null) {
+                    throw new DdlException("Table not found: " + command.getTargetTableId());
+                }
+                if (!(table instanceof OlapTable)) {
+                    throw new DdlException("Routine load target table must be an OLAP table");
+                }
+                table.readLock();
+                try {
+                    unprotectApplyAlter(command);
+                } finally {
+                    table.readUnlock();
+                }
+            } finally {
+                db.readUnlock();
+            }
+        } finally {
+            writeUnlock();
+        }
+    }
+
+    private void unprotectApplyAlter(AlterRoutineLoadCommand command) throws UserException {
+        validateAlterJobPropertiesForMutation(command);
+        unprotectModifyProperties(command);
+        if (command.hasTargetTable()) {
+            this.tableId = command.getTargetTableId();
+        }
+
+        AlterRoutineLoadJobOperationLog log = new AlterRoutineLoadJobOperationLog(this.id,
+                command.getAnalyzedJobProperties(), command.getDataSourceProperties(), command.getTargetTableId());
+        Env.getCurrentEnv().getEditLog().logAlterRoutineLoadJob(log);
+    }
+
+    protected abstract void unprotectModifyProperties(AlterRoutineLoadCommand command) throws UserException;
 
     public abstract void replayModifyProperties(AlterRoutineLoadJobOperationLog log);
 
@@ -2079,17 +2229,16 @@ protected void modifyCommonJobProperties(Map<String, String> jobProperties) thro
                     jobProperties.remove(CreateRoutineLoadInfo.MAX_BATCH_SIZE_PROPERTY));
         }
 
-        if (jobProperties.containsKey(CreateRoutineLoadInfo.UNIQUE_KEY_UPDATE_MODE)) {
+        boolean hasExplicitUniqueKeyUpdateMode = jobProperties.containsKey(
+                CreateRoutineLoadInfo.UNIQUE_KEY_UPDATE_MODE);
+        if (hasExplicitUniqueKeyUpdateMode) {
             String modeStr = jobProperties.remove(CreateRoutineLoadInfo.UNIQUE_KEY_UPDATE_MODE);
             TUniqueKeyUpdateMode newMode = CreateRoutineLoadInfo.parseAndValidateUniqueKeyUpdateMode(modeStr);
-            // Validate flexible partial update constraints when changing to UPDATE_FLEXIBLE_COLUMNS
-            if (newMode == TUniqueKeyUpdateMode.UPDATE_FLEXIBLE_COLUMNS) {
-                validateFlexiblePartialUpdateForAlter();
-            }
             this.uniqueKeyUpdateMode = newMode;
             this.isPartialUpdate = (uniqueKeyUpdateMode == TUniqueKeyUpdateMode.UPDATE_FIXED_COLUMNS);
             this.jobProperties.put(CreateRoutineLoadInfo.UNIQUE_KEY_UPDATE_MODE, uniqueKeyUpdateMode.name());
             this.jobProperties.put(CreateRoutineLoadInfo.PARTIAL_COLUMNS, String.valueOf(isPartialUpdate));
+            jobProperties.remove(CreateRoutineLoadInfo.PARTIAL_COLUMNS);
         }
 
         if (jobProperties.containsKey(CreateRoutineLoadInfo.PARTIAL_COLUMNS)) {
@@ -2108,43 +2257,27 @@ protected void modifyCommonJobProperties(Map<String, String> jobProperties) thro
     /**
      * Validate flexible partial update constraints when altering routine load job.
      */
-    private void validateFlexiblePartialUpdateForAlter() throws UserException {
-        // Multi-table load does not support flexible partial update
-        if (isMultiTable) {
-            throw new DdlException("Flexible partial update is not supported in multi-table load");
-        }
-
-        // Get the table to check table-level constraints
-        Database db = Env.getCurrentInternalCatalog().getDbNullable(dbId);
-        if (db == null) {
-            throw new DdlException("Database not found: " + dbId);
-        }
-        Table table = db.getTableNullable(tableId);
-        if (table == null) {
-            throw new DdlException("Table not found: " + tableId);
-        }
-        if (!(table instanceof OlapTable)) {
-            throw new DdlException("Flexible partial update is only supported for OLAP tables");
-        }
-        OlapTable olapTable = (OlapTable) table;
-
+    private void validateFlexiblePartialUpdateForAlter(OlapTable targetTable,
+            Map<String, String> alteredJobProperties) throws UserException {
         // Validate table-level constraints (MoW, skip_bitmap, light_schema_change, variant columns)
-        olapTable.validateForFlexiblePartialUpdate();
+        targetTable.validateForFlexiblePartialUpdate();
+        Map<String, String> effectiveJobProperties = Maps.newHashMap(jobProperties);
+        effectiveJobProperties.putAll(alteredJobProperties);
 
         // Routine load specific validations
         // Must use JSON format
-        String format = this.jobProperties.getOrDefault(FileFormatProperties.PROP_FORMAT, "csv");
+        String format = effectiveJobProperties.getOrDefault(FileFormatProperties.PROP_FORMAT, "csv");
         if (!"json".equalsIgnoreCase(format)) {
             throw new DdlException("Flexible partial update only supports JSON format, but current job uses: "
                     + format);
         }
         // Cannot use fuzzy_parse
-        if (Boolean.parseBoolean(this.jobProperties.getOrDefault(
+        if (Boolean.parseBoolean(effectiveJobProperties.getOrDefault(
                 JsonFileFormatProperties.PROP_FUZZY_PARSE, "false"))) {
             throw new DdlException("Flexible partial update does not support fuzzy_parse");
         }
         // Cannot use jsonpaths
-        String jsonPaths = getJsonPaths();
+        String jsonPaths = effectiveJobProperties.get(JsonFileFormatProperties.PROP_JSON_PATHS);
         if (jsonPaths != null && !jsonPaths.isEmpty()) {
             throw new DdlException("Flexible partial update does not support jsonpaths");
         }

fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaDataSourceProperties.java

@@ -105,6 +105,14 @@ public class KafkaDataSourceProperties extends AbstractDataSourceProperties {
                     .add(KafkaConfiguration.KAFKA_TEXT_TABLE_NAME_FIELD_INDEX.getName())
                     .build();
 
+    private static final ImmutableSet<String> CREATE_ONLY_MULTI_TABLE_PROPERTIES_SET =
+            new ImmutableSet.Builder<String>()
+                    .add(KafkaConfiguration.KAFKA_TABLE_NAME_LOCATION.getName())
+                    .add(KafkaConfiguration.KAFKA_TABLE_NAME_FORMAT.getName())
+                    .add(KafkaConfiguration.KAFKA_TEXT_TABLE_NAME_FIELD_DELIMITER.getName())
+                    .add(KafkaConfiguration.KAFKA_TEXT_TABLE_NAME_FIELD_INDEX.getName())
+                    .build();
+
     public KafkaDataSourceProperties(Map<String, String> dataSourceProperties, boolean multiLoad) {
         super(dataSourceProperties, multiLoad);
     }
@@ -132,6 +140,14 @@ public void convertAndCheckDataSourceProperties() throws UserException {
         if (optional.isPresent()) {
             throw new AnalysisException(optional.get() + " is invalid kafka property or can not be set");
         }
+        if (isAlter()) {
+            Optional<String> createOnlyProperty = originalDataSourceProperties.keySet().stream()
+                    .filter(CREATE_ONLY_MULTI_TABLE_PROPERTIES_SET::contains).findFirst();
+            if (createOnlyProperty.isPresent()) {
+                throw new AnalysisException(createOnlyProperty.get() + " can only be set when creating "
+                        + "a multi-table routine load job");
+            }
+        }
 
         this.brokerList = KafkaConfiguration.KAFKA_BROKER_LIST.getParameterValue(originalDataSourceProperties
                 .get(KafkaConfiguration.KAFKA_BROKER_LIST.getName()));
@@ -166,24 +182,41 @@ public void convertAndCheckDataSourceProperties() throws UserException {
         //check offset
         List<String> offsets = KafkaConfiguration.KAFKA_OFFSETS.getParameterValue(originalDataSourceProperties
                 .get(KafkaConfiguration.KAFKA_OFFSETS.getName()));
-        String defaultOffsetString = originalDataSourceProperties
+        boolean hasCustomDefaultOffset = customKafkaProperties
+                .containsKey(KafkaConfiguration.KAFKA_DEFAULT_OFFSETS.getName());
+        String defaultOffsetString = customKafkaProperties.get(KafkaConfiguration.KAFKA_DEFAULT_OFFSETS.getName());
+        boolean hasDirectDefaultOffset = originalDataSourceProperties
+                .containsKey(KafkaConfiguration.KAFKA_DEFAULT_OFFSETS.getName());
+        String directDefaultOffsetString = originalDataSourceProperties
                 .get(KafkaConfiguration.KAFKA_DEFAULT_OFFSETS.getName());
-        if (CollectionUtils.isNotEmpty(offsets) && StringUtils.isNotBlank(defaultOffsetString)) {
+        if (hasDirectDefaultOffset) {
+            if (hasCustomDefaultOffset) {
+                throw new AnalysisException("Only one of " + KafkaConfiguration.KAFKA_DEFAULT_OFFSETS.getName()
+                        + " and property." + KafkaConfiguration.KAFKA_DEFAULT_OFFSETS.getName() + " can be set.");
+            }
+            defaultOffsetString = directDefaultOffsetString;
+            customKafkaProperties.put(KafkaConfiguration.KAFKA_DEFAULT_OFFSETS.getName(), defaultOffsetString);
+        }
+        boolean hasDefaultOffset = hasCustomDefaultOffset || hasDirectDefaultOffset;
+        if (CollectionUtils.isNotEmpty(offsets) && hasDefaultOffset) {
             throw new AnalysisException("Only one of " + KafkaConfiguration.KAFKA_OFFSETS.getName() + " and "
                     + KafkaConfiguration.KAFKA_DEFAULT_OFFSETS.getName() + " can be set.");
         }
         if (multiTable) {
             checkAndSetMultiLoadProperties();
         }
         if (isAlter() && CollectionUtils.isNotEmpty(partitions) && CollectionUtils.isEmpty(offsets)
-                && StringUtils.isBlank(defaultOffsetString)) {
+                && !hasDefaultOffset) {
             // if this is an alter operation, the partition and (default)offset must be set together.
             throw new AnalysisException("Must set offset or default offset with partition property");
         }
         if (CollectionUtils.isNotEmpty(offsets)) {
             this.isOffsetsForTimes = analyzeKafkaOffsetProperty(offsets);
             return;
         }
+        if (isAlter() && CollectionUtils.isEmpty(partitions) && !hasDefaultOffset) {
+            return;
+        }
         this.isOffsetsForTimes = analyzeKafkaDefaultOffsetProperty();
         if (CollectionUtils.isNotEmpty(kafkaPartitionOffsets)) {
             defaultOffsetString = customKafkaProperties.get(KafkaConfiguration.KAFKA_DEFAULT_OFFSETS.getName());