[feature](routine-load) Support composable target table alteration

### What problem does this PR solve?

Issue Number: None

Related PR: #64878

Problem Summary: ALTER ROUTINE LOAD previously used the ON target-table syntax and could not combine a target switch with supported job or data-source properties. Add the explicit SET TARGET TABLE syntax, preserve Doris property constraints, validate and apply target/job/source changes atomically, and keep unspecified Kafka and Kinesis default positions unchanged.

### Release note

ALTER ROUTINE LOAD now uses SET TARGET TABLE = "table" and can combine a target table switch with supported PROPERTIES and FROM data-source properties.

### Check List (For Author)

- Test: Regression test / Unit Test / Build
    - FE routine-load and parser unit tests
    - load_p0/routine_load/test_routine_load_alter regression test
    - ./build.sh --fe -j48
- Behavior changed: Yes. ALTER ROUTINE LOAD target-table syntax and composability changed as described above.
- Does this need documentation: Yes (tracked by Related PR #64878)

Author: 0AyanamiRei

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

@@ -470,6 +470,12 @@ 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");
         }
@@ -479,17 +485,75 @@ public void validateTargetTable(Database db, OlapTable targetTable) throws UserE
         }
         checkMeta(targetTable, new RoutineLoadDesc(columnSeparator, lineDelimiter, columnsInfo, precedingFilter,
                 whereExpr, partitionNamesInfo, deleteCondition, mergeType, sequenceCol));
+        validateAlterJobProperties(targetTable, alteredJobProperties, effectiveUniqueKeyUpdateMode);
 
         targetTable.readLock();
         try {
-            NereidsStreamLoadPlanner planner = new NereidsStreamLoadPlanner(db, targetTable,
-                    toNereidsRoutineLoadTaskInfo());
+            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 (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("Partial update is only supported for OLAP tables");
+        }
+        validateAlterJobProperties((OlapTable) table, alteredJobProperties, effectiveUniqueKeyUpdateMode);
+    }
+
     @Override
     public long getId() {
         return id;
@@ -2093,10 +2157,6 @@ protected void modifyCommonJobProperties(Map<String, String> jobProperties) thro
         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());
@@ -2120,43 +2180,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());