github-actions[bot] commented on code in PR #64878:
URL: https://github.com/apache/doris/pull/64878#discussion_r3611748304
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/parser/LogicalPlanBuilder.java:
##########
@@ -9309,7 +9315,8 @@ public LogicalPlan
visitAlterRoutineLoad(DorisParser.AlterRoutineLoadContext ctx
}
}
- return new AlterRoutineLoadCommand(labelNameInfo, loadPropertyMap,
properties, dataSourceMapProperties);
+ return new AlterRoutineLoadCommand(labelNameInfo, targetTableName,
Review Comment:
[P1] Preserve and validate the explicit FROM type. The grammar records
ctx.type, but this builder drops it; checkDataSourceProperties() then creates
the source-property object from the job's existing type, so
RoutineLoadManager's later comparison is always the job type against itself.
For example, a Kafka job accepts SET TARGET TABLE = "B" FROM
KINESIS("property.client.id"="x") as KafkaDataSourceProperties because
property.* is legal there, despite the explicit KINESIS token. Carry the parsed
type into the command and reject unknown or non-matching types before
constructing the property object, with cross-type negative tests.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterRoutineLoadCommand.java:
##########
@@ -164,15 +186,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);
Review Comment:
[P1] Plan the target with the statement's effective job properties. This
call validates B using the job's current update mode because
analyzedJobProperties is never overlaid. A paused UPSERT job on MoW table A can
therefore switch to non-MoW B while setting
unique_key_update_mode=UPDATE_FIXED_COLUMNS: B is planned as UPSERT and
accepted, both job implementations install the fixed-update mode and then B's
ID, and the first resumed task fails with Only unique key merge on write
support partial update. Build one staged task definition containing the target
and analyzed property delta, validate that definition, and add this
non-concurrent combined case.
##########
fe/fe-core/src/main/java/org/apache/doris/load/routineload/kinesis/KinesisRoutineLoadJob.java:
##########
@@ -687,9 +687,12 @@ public void modifyProperties(AlterRoutineLoadCommand
command) throws UserExcepti
}
modifyPropertiesInternal(jobProperties, dataSourceProperties);
Review Comment:
[P1] Apply the same failure atomicity to Kinesis. modifyPropertiesInternal()
merges custom properties and replaces customKinesisShards before checkShards(),
while the target assignment and EditLog call happen only afterward. A combined
target ALTER that supplies property.client.timeout plus a nonexistent explicit
shard therefore fails the membership check but leaves the live job's custom
properties and pinned shard list changed, with the old target and no journal;
resume and failover then observe different definitions. Stage source
properties, shard lists/progress, common fields, and the target off-object,
validate everything, and only then publish and journal the complete delta.
##########
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);
Review Comment:
[P1] Keep a failed combined ALTER side-effect-free.
modifyPropertiesInternal() runs before the target assignment and journal, and
it publishes custom Kafka properties before checkPartitions(). On a job
consuming partition 0, SET TARGET TABLE = "B" FROM
KAFKA("property.client.id"="new", "kafka_partitions"="999",
"kafka_offsets"="0") installs client.id, then throws because partition 999 is
absent; the command reports failure with the old target and no EditLog record,
but the live job has changed until failover/restart. Validate immutable staged
copies first, then publish the complete target/job/source delta atomically, and
assert every field is unchanged on failure.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
