[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,44 @@ protected void setRoutineLoadDesc(RoutineLoadDesc routineLoadDesc) {
         }
     }
 
+    public void validateTargetTable(Database db, OlapTable targetTable,
+            Map<String, String> alteredJobProperties) 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));
+
+        targetTable.readLock();
+        try {
+            NereidsRoutineLoadTaskInfo taskInfo = toNereidsRoutineLoadTaskInfo();
+            taskInfo.applyAlterPropertiesForValidation(
+                    alteredJobProperties, getEffectiveUniqueKeyUpdateMode(alteredJobProperties));
+            NereidsStreamLoadPlanner planner = new NereidsStreamLoadPlanner(db, targetTable, taskInfo);
+            planner.plan(new TUniqueId(0, 0));
+        } finally {
+            targetTable.readUnlock();
+        }
+    }
+
+    private TUniqueKeyUpdateMode getEffectiveUniqueKeyUpdateMode(Map<String, String> alteredJobProperties) {
+        TUniqueKeyUpdateMode effectiveUniqueKeyUpdateMode = uniqueKeyUpdateMode;
+        if (alteredJobProperties.containsKey(CreateRoutineLoadInfo.UNIQUE_KEY_UPDATE_MODE)) {
+            effectiveUniqueKeyUpdateMode = TUniqueKeyUpdateMode.valueOf(
+                    alteredJobProperties.get(CreateRoutineLoadInfo.UNIQUE_KEY_UPDATE_MODE));
+        }
+        if (alteredJobProperties.containsKey(CreateRoutineLoadInfo.PARTIAL_COLUMNS)
+                && Boolean.parseBoolean(alteredJobProperties.get(CreateRoutineLoadInfo.PARTIAL_COLUMNS))
+                && effectiveUniqueKeyUpdateMode == TUniqueKeyUpdateMode.UPSERT) {
+            effectiveUniqueKeyUpdateMode = TUniqueKeyUpdateMode.UPDATE_FIXED_COLUMNS;
+        }
+        return effectiveUniqueKeyUpdateMode;
+    }
+
     @Override
     public long getId() {
         return id;

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

@@ -782,9 +782,12 @@ public void modifyProperties(AlterRoutineLoadCommand command) throws UserExcepti
             }
 
             modifyPropertiesInternal(jobProperties, dataSourceProperties);
+            if (command.hasTargetTable()) {
+                tableId = command.getTargetTableId();
+            }
 
             AlterRoutineLoadJobOperationLog log = new AlterRoutineLoadJobOperationLog(this.id,
-                    jobProperties, dataSourceProperties);
+                    jobProperties, dataSourceProperties, command.getTargetTableId());
             Env.getCurrentEnv().getEditLog().logAlterRoutineLoadJob(log);
         } finally {
             writeUnlock();
@@ -908,6 +911,9 @@ private void resetCloudProgress(Cloud.ResetRLProgressRequest.Builder builder) th
     public void replayModifyProperties(AlterRoutineLoadJobOperationLog log) {
         try {
             modifyPropertiesInternal(log.getJobProperties(), (KafkaDataSourceProperties) log.getDataSourceProperties());
+            if (log.getTargetTableId() != 0) {
+                tableId = log.getTargetTableId();
+            }
         } catch (UserException e) {
             // should not happen
             LOG.error("failed to replay modify kafka routine load job: {}", id, e);

fe/fe-core/src/main/java/org/apache/doris/nereids/load/NereidsRoutineLoadTaskInfo.java

@@ -121,6 +121,13 @@ public int calTimeoutSec() {
         return realTimeoutSec;
     }
 
+    /** Apply analyzed ALTER properties to this validation-only task. */
+    public void applyAlterPropertiesForValidation(Map<String, String> alteredJobProperties,
+            TUniqueKeyUpdateMode effectiveUniqueKeyUpdateMode) {
+        jobProperties.putAll(alteredJobProperties);
+        uniquekeyUpdateMode = effectiveUniqueKeyUpdateMode;
+    }
+
     @Override
     public int getTimeout() {
         return this.timeoutSec;

fe/fe-core/src/main/java/org/apache/doris/nereids/parser/LogicalPlanBuilder.java

@@ -9361,6 +9361,8 @@ public LogicalPlan visitAlterRoutineLoad(DorisParser.AlterRoutineLoadContext ctx
             throw new ParseException("only support [<db>.]<job_name>", ctx.name);
         }
         LabelNameInfo labelNameInfo = new LabelNameInfo(dbName, jobName);
+        String targetTableName = ctx.targetTable == null
+                ? null : SqlLiteralUtils.parseStringLiteral(ctx.targetTable.getText());
 
         Map<String, String> properties = new HashMap<>();
         if (ctx.properties != null) {
@@ -9387,7 +9389,8 @@ public LogicalPlan visitAlterRoutineLoad(DorisParser.AlterRoutineLoadContext ctx
             }
         }
 
-        return new AlterRoutineLoadCommand(labelNameInfo, loadPropertyMap, properties, dataSourceMapProperties);
+        return new AlterRoutineLoadCommand(labelNameInfo, targetTableName,
+                loadPropertyMap, properties, dataSourceMapProperties);
     }
 
     @Override

fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterRoutineLoadCommand.java

@@ -21,18 +21,24 @@
 import org.apache.doris.catalog.Database;
 import org.apache.doris.catalog.Env;
 import org.apache.doris.catalog.OlapTable;
+import org.apache.doris.catalog.RandomDistributionInfo;
 import org.apache.doris.catalog.Table;
 import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
 import org.apache.doris.common.FeNameFormat;
 import org.apache.doris.common.UserException;
 import org.apache.doris.common.util.TimeUtils;
 import org.apache.doris.common.util.Util;
+import org.apache.doris.datasource.InternalCatalog;
 import org.apache.doris.datasource.property.fileformat.CsvFileFormatProperties;
 import org.apache.doris.datasource.property.fileformat.JsonFileFormatProperties;
 import org.apache.doris.load.RoutineLoadDesc;
 import org.apache.doris.load.routineload.AbstractDataSourceProperties;
+import org.apache.doris.load.routineload.LoadDataSourceType;
 import org.apache.doris.load.routineload.RoutineLoadDataSourcePropertyFactory;
 import org.apache.doris.load.routineload.RoutineLoadJob;
+import org.apache.doris.mysql.privilege.PrivPredicate;
 import org.apache.doris.nereids.trees.plans.PlanType;
 import org.apache.doris.nereids.trees.plans.commands.info.CreateRoutineLoadInfo;
 import org.apache.doris.nereids.trees.plans.commands.info.LabelNameInfo;
@@ -52,6 +58,7 @@
 
 /**
  * ALTER ROUTINE LOAD db.label
+ * SET TARGET TABLE = "table"
  * PROPERTIES(
  * ...
  * )
@@ -85,10 +92,12 @@ public class AlterRoutineLoadCommand extends AlterCommand {
             .build();
 
     private final LabelNameInfo labelNameInfo;
+    private final String targetTableName;
     private final Map<String, LoadProperty> loadPropertyMap;
     private RoutineLoadDesc routineLoadDesc;
     private final Map<String, String> jobProperties;
     private final Map<String, String> dataSourceMapProperties;
+    private long targetTableId;
     private boolean isPartialUpdate;
 
     // save analyzed job properties.
@@ -100,6 +109,7 @@ public class AlterRoutineLoadCommand extends AlterCommand {
      * AlterRoutineLoadCommand
      */
     public AlterRoutineLoadCommand(LabelNameInfo labelNameInfo,
+                                   String targetTableName,
                                    Map<String, LoadProperty> loadPropertyMap,
                                    Map<String, String> jobProperties,
                                    Map<String, String> dataSourceMapProperties) {
@@ -108,6 +118,7 @@ public AlterRoutineLoadCommand(LabelNameInfo labelNameInfo,
         Objects.requireNonNull(jobProperties, "jobProperties is null");
         Objects.requireNonNull(dataSourceMapProperties, "dataSourceMapProperties is null");
         this.labelNameInfo = labelNameInfo;
+        this.targetTableName = targetTableName;
         this.loadPropertyMap = loadPropertyMap == null ? Maps.newHashMap() : loadPropertyMap;
         this.jobProperties = jobProperties;
         this.dataSourceMapProperties = dataSourceMapProperties;
@@ -118,7 +129,7 @@ public AlterRoutineLoadCommand(LabelNameInfo labelNameInfo,
     public AlterRoutineLoadCommand(LabelNameInfo labelNameInfo,
                                    Map<String, String> jobProperties,
                                    Map<String, String> dataSourceMapProperties) {
-        this(labelNameInfo, Maps.newHashMap(), jobProperties, dataSourceMapProperties);
+        this(labelNameInfo, null, Maps.newHashMap(), jobProperties, dataSourceMapProperties);
     }
 
     public String getDbName() {
@@ -133,6 +144,18 @@ public Map<String, String> getAnalyzedJobProperties() {
         return analyzedJobProperties;
     }
 
+    public boolean hasTargetTable() {
+        return targetTableName != null;
+    }
+
+    public String getTargetTableName() {
+        return targetTableName;
+    }
+
+    public long getTargetTableId() {
+        return targetTableId;
+    }
+
     public boolean hasDataSourceProperty() {
         return MapUtils.isNotEmpty(dataSourceMapProperties);
     }
@@ -164,15 +187,22 @@ public void validate(ConnectContext ctx) throws UserException {
         // check routine load job properties include desired concurrent number etc.
         checkJobProperties();
         // check load properties
-        RoutineLoadJob job = Env.getCurrentEnv().getRoutineLoadManager()
-                .getJob(getDbName(), getJobName());
-        this.routineLoadDesc = CreateRoutineLoadInfo.checkLoadProperties(ctx, loadPropertyMap,
-                job.getDbFullName(), job.getTableName(), job.isMultiTable(), job.getMergeType());
+        RoutineLoadJob job = hasTargetTable()
+                ? Env.getCurrentEnv().getRoutineLoadManager().checkPrivAndGetJob(getDbName(), getJobName())
+                : Env.getCurrentEnv().getRoutineLoadManager().getJob(getDbName(), getJobName());
+        if (MapUtils.isNotEmpty(loadPropertyMap)) {
+            this.routineLoadDesc = CreateRoutineLoadInfo.checkLoadProperties(ctx, loadPropertyMap,
+                    job.getDbFullName(), job.getTableName(), job.isMultiTable(), job.getMergeType());
+        }
         // check data source properties
         checkDataSourceProperties();
-        checkPartialUpdate();
+        if (hasTargetTable()) {
+            validateTargetTable(ctx, job);
+        } else {
+            checkPartialUpdate(job);
+        }
         if (analyzedJobProperties.isEmpty() && MapUtils.isEmpty(dataSourceMapProperties)
-                && routineLoadDesc == null) {
+                && routineLoadDesc == null && !hasTargetTable()) {
             throw new AnalysisException("No properties are specified");
         }
     }
@@ -328,12 +358,10 @@ private void checkDataSourceProperties() throws UserException {
         dataSourceProperties.analyze();
     }
 
-    private void checkPartialUpdate() throws UserException {
+    private void checkPartialUpdate(RoutineLoadJob job) throws UserException {
         if (!isPartialUpdate) {
             return;
         }
-        RoutineLoadJob job = Env.getCurrentEnv().getRoutineLoadManager()
-                .getJob(getDbName(), getDbName());
         if (job.isMultiTable()) {
             throw new AnalysisException("load by PARTIAL_COLUMNS is not supported in multi-table load.");
         }
@@ -344,6 +372,37 @@ private void checkPartialUpdate() throws UserException {
         }
     }
 
+    private void validateTargetTable(ConnectContext ctx, RoutineLoadJob job) throws UserException {
+        if (job.getDataSourceType() != LoadDataSourceType.KAFKA) {
+            throw new AnalysisException("ALTER ROUTINE LOAD target table change only supports Kafka jobs");
+        }
+        if (job.isMultiTable()) {
+            throw new AnalysisException("ALTER ROUTINE LOAD target table change only supports single-table job");
+        }
+        String dbFullName = job.getDbFullName();
+        if (!Env.getCurrentEnv().getAccessManager().checkTblPriv(ctx, InternalCatalog.INTERNAL_CATALOG_NAME,
+                dbFullName, targetTableName, PrivPredicate.LOAD)) {
+            ErrorReport.reportAnalysisException(ErrorCode.ERR_TABLEACCESS_DENIED_ERROR, "LOAD",
+                    ctx.getQualifiedUser(), ctx.getRemoteIP(), dbFullName + ": " + targetTableName);
+        }
+        Database db = Env.getCurrentInternalCatalog().getDbOrAnalysisException(dbFullName);
+        Table table = db.getTableOrAnalysisException(targetTableName);
+        if (!(table instanceof OlapTable)) {
+            throw new AnalysisException("ALTER ROUTINE LOAD target table only supports OLAP table");
+        }
+        OlapTable olapTable = (OlapTable) table;
+        if (olapTable.isTemporary()) {
+            throw new AnalysisException("Do not support load for temporary table " + olapTable.getDisplayName());
+        }
+        if (job.isLoadToSingleTablet()
+                && !(olapTable.getDefaultDistributionInfo() instanceof RandomDistributionInfo)) {
+            throw new AnalysisException(
+                    "if load_to_single_tablet set to true, the olap table must be with random distribution");
+        }
+        job.validateTargetTable(db, olapTable, analyzedJobProperties);
+        this.targetTableId = olapTable.getId();
+    }
+
     @Override
     public <R, C> R accept(PlanVisitor<R, C> visitor, C context) {
         return visitor.visitAlterRoutineLoadCommand(this, context);

fe/fe-core/src/main/java/org/apache/doris/persist/AlterRoutineLoadJobOperationLog.java

@@ -37,12 +37,20 @@ public class AlterRoutineLoadJobOperationLog  implements Writable {
     private Map<String, String> jobProperties;
     @SerializedName(value = "dataSourceProperties")
     private AbstractDataSourceProperties dataSourceProperties;
+    @SerializedName(value = "targetTableId")
+    private long targetTableId;
 
     public AlterRoutineLoadJobOperationLog(long jobId, Map<String, String> jobProperties,
             AbstractDataSourceProperties dataSourceProperties) {
+        this(jobId, jobProperties, dataSourceProperties, 0L);
+    }
+
+    public AlterRoutineLoadJobOperationLog(long jobId, Map<String, String> jobProperties,
+            AbstractDataSourceProperties dataSourceProperties, long targetTableId) {
         this.jobId = jobId;
         this.jobProperties = jobProperties;
         this.dataSourceProperties = dataSourceProperties;
+        this.targetTableId = targetTableId;
     }
 
     public long getJobId() {
@@ -57,6 +65,10 @@ public AbstractDataSourceProperties getDataSourceProperties() {
         return dataSourceProperties;
     }
 
+    public long getTargetTableId() {
+        return targetTableId;
+    }
+
     public static AlterRoutineLoadJobOperationLog read(DataInput in) throws IOException {
         String json = Text.readString(in);
         return GsonUtils.GSON.fromJson(json, AlterRoutineLoadJobOperationLog.class);