[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,27 @@ protected void setRoutineLoadDesc(RoutineLoadDesc routineLoadDesc) {
         }
     }
 
+    public void validateTargetTable(Database db, OlapTable targetTable) 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 {
+            NereidsStreamLoadPlanner planner = new NereidsStreamLoadPlanner(db, targetTable,
+                    toNereidsRoutineLoadTaskInfo());
+            planner.plan(new TUniqueId(0, 0));
+        } finally {
+            targetTable.readUnlock();
+        }
+    }
+
     @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/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,21 @@ 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);
+        }
         if (analyzedJobProperties.isEmpty() && MapUtils.isEmpty(dataSourceMapProperties)
-                && routineLoadDesc == null) {
+                && routineLoadDesc == null && !hasTargetTable()) {
             throw new AnalysisException("No properties are specified");
         }
     }
@@ -344,6 +373,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);
+        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);

fe/fe-core/src/test/java/org/apache/doris/load/routineload/KafkaRoutineLoadJobTest.java

@@ -40,10 +40,13 @@
 import org.apache.doris.load.routineload.kafka.KafkaRoutineLoadJob;
 import org.apache.doris.load.routineload.kafka.KafkaTaskInfo;
 import org.apache.doris.mysql.privilege.MockedAuth;
+import org.apache.doris.nereids.trees.plans.commands.AlterRoutineLoadCommand;
 import org.apache.doris.nereids.trees.plans.commands.info.CreateRoutineLoadInfo;
 import org.apache.doris.nereids.trees.plans.commands.info.LabelNameInfo;
 import org.apache.doris.nereids.trees.plans.commands.load.LoadProperty;
 import org.apache.doris.nereids.trees.plans.commands.load.LoadSeparator;
+import org.apache.doris.persist.AlterRoutineLoadJobOperationLog;
+import org.apache.doris.persist.EditLog;
 import org.apache.doris.qe.ConnectContext;
 import org.apache.doris.thrift.TResourceInfo;
 
@@ -57,6 +60,7 @@
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
+import org.mockito.ArgumentCaptor;
 import org.mockito.MockedStatic;
 import org.mockito.Mockito;
 
@@ -237,6 +241,101 @@ public void testUpdateLagRebuildsConvertedPropertiesAfterReplay() throws UserExc
         }
     }
 
+    @Test
+    public void testModifyTargetTableWithJobAndDataSourceProperties() throws Exception {
+        KafkaRoutineLoadJob routineLoadJob = new KafkaRoutineLoadJob(1L, "kafka_routine_load_job", 1L,
+                101L, "127.0.0.1:9020", "topic1", UserIdentity.ADMIN);
+        Deencapsulation.setField(routineLoadJob, "state", RoutineLoadJob.JobState.PAUSED);
+        KafkaProgress progress = new KafkaProgress(Maps.newHashMap());
+        Deencapsulation.setField(routineLoadJob, "progress", progress);
+
+        Map<String, String> sourceProperties = Maps.newHashMap();
+        sourceProperties.put("property.client.id", "target-switch");
+        KafkaDataSourceProperties dataSourceProperties = new KafkaDataSourceProperties(sourceProperties);
+        dataSourceProperties.setAlter(true);
+        dataSourceProperties.setTimezone("UTC");
+        dataSourceProperties.analyze();
+
+        Map<String, String> jobProperties = Maps.newHashMap();
+        jobProperties.put(CreateRoutineLoadInfo.MAX_ERROR_NUMBER_PROPERTY, "10");
+        AlterRoutineLoadCommand command = Mockito.mock(AlterRoutineLoadCommand.class);
+        Mockito.when(command.hasTargetTable()).thenReturn(true);
+        Mockito.when(command.getTargetTableId()).thenReturn(202L);
+        Mockito.when(command.getAnalyzedJobProperties()).thenReturn(jobProperties);
+        Mockito.when(command.getDataSourceProperties()).thenReturn(dataSourceProperties);
+
+        Env env = Mockito.mock(Env.class);
+        EditLog editLog = Mockito.mock(EditLog.class);
+        Mockito.when(env.getEditLog()).thenReturn(editLog);
+        try (MockedStatic<Env> envStatic = Mockito.mockStatic(Env.class)) {
+            envStatic.when(Env::getCurrentEnv).thenReturn(env);
+            routineLoadJob.modifyProperties(command);
+        }
+
+        Assert.assertEquals(202L, routineLoadJob.getTableId());
+        Assert.assertSame(progress, routineLoadJob.getProgress());
+        Assert.assertEquals(10L, ((Long) Deencapsulation.getField(routineLoadJob, "maxErrorNum")).longValue());
+        Assert.assertEquals("target-switch",
+                routineLoadJob.getCustomProperties().get("property.client.id"));
+        ArgumentCaptor<AlterRoutineLoadJobOperationLog> logCaptor =
+                ArgumentCaptor.forClass(AlterRoutineLoadJobOperationLog.class);
+        Mockito.verify(editLog).logAlterRoutineLoadJob(logCaptor.capture());
+        Assert.assertEquals(202L, logCaptor.getValue().getTargetTableId());
+    }
+
+    @Test
+    public void testModifyTargetTableOnly() throws Exception {
+        KafkaRoutineLoadJob routineLoadJob = new KafkaRoutineLoadJob(1L, "kafka_routine_load_job", 1L,
+                101L, "127.0.0.1:9020", "topic1", UserIdentity.ADMIN);
+        Deencapsulation.setField(routineLoadJob, "state", RoutineLoadJob.JobState.PAUSED);
+        KafkaProgress progress = new KafkaProgress(Maps.newHashMap());
+        Deencapsulation.setField(routineLoadJob, "progress", progress);
+
+        AlterRoutineLoadCommand command = Mockito.mock(AlterRoutineLoadCommand.class);
+        Mockito.when(command.hasTargetTable()).thenReturn(true);
+        Mockito.when(command.getTargetTableId()).thenReturn(202L);
+        Mockito.when(command.getAnalyzedJobProperties()).thenReturn(Maps.newHashMap());
+        Mockito.when(command.getDataSourceProperties()).thenReturn(null);
+
+        Env env = Mockito.mock(Env.class);
+        EditLog editLog = Mockito.mock(EditLog.class);
+        Mockito.when(env.getEditLog()).thenReturn(editLog);
+        try (MockedStatic<Env> envStatic = Mockito.mockStatic(Env.class)) {
+            envStatic.when(Env::getCurrentEnv).thenReturn(env);
+            routineLoadJob.modifyProperties(command);
+        }
+
+        Assert.assertEquals(202L, routineLoadJob.getTableId());
+        Assert.assertSame(progress, routineLoadJob.getProgress());
+        ArgumentCaptor<AlterRoutineLoadJobOperationLog> logCaptor =
+                ArgumentCaptor.forClass(AlterRoutineLoadJobOperationLog.class);
+        Mockito.verify(editLog).logAlterRoutineLoadJob(logCaptor.capture());
+        Assert.assertEquals(202L, logCaptor.getValue().getTargetTableId());
+        Assert.assertTrue(logCaptor.getValue().getJobProperties().isEmpty());
+        Assert.assertNull(logCaptor.getValue().getDataSourceProperties());
+    }
+
+    @Test
+    public void testReplayModifyPropertiesSwitchesTargetTableWithoutResettingProgress() {
+        KafkaRoutineLoadJob routineLoadJob = new KafkaRoutineLoadJob(1L, "kafka_routine_load_job", 1L,
+                101L, "127.0.0.1:9020", "topic1", UserIdentity.ADMIN);
+        Map<Integer, Long> partitionToOffset = Maps.newHashMap();
+        partitionToOffset.put(1, 123L);
+        KafkaProgress progress = new KafkaProgress(partitionToOffset);
+        Deencapsulation.setField(routineLoadJob, "progress", progress);
+
+        routineLoadJob.replayModifyProperties(new AlterRoutineLoadJobOperationLog(
+                1L, Maps.newHashMap(), null));
+        Assert.assertEquals(101L, routineLoadJob.getTableId());
+
+        routineLoadJob.replayModifyProperties(new AlterRoutineLoadJobOperationLog(
+                1L, Maps.newHashMap(), null, 202L));
+        Assert.assertEquals(202L, routineLoadJob.getTableId());
+        Assert.assertSame(progress, routineLoadJob.getProgress());
+        Assert.assertEquals(Long.valueOf(123L),
+                ((KafkaProgress) routineLoadJob.getProgress()).getOffsetByPartition(1));
+    }
+
     @Test
     public void testUpdateProgressWarnsWhenReadCommittedTaskHasZeroRowsAndLag() throws UserException {
         KafkaRoutineLoadJob routineLoadJob = new KafkaRoutineLoadJob(1L, "kafka_routine_load_job", 1L,