github-actions[bot] commented on code in PR #64878:
URL: https://github.com/apache/doris/pull/64878#discussion_r3600200426
##########
fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java:
##########
@@ -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();
Review Comment:
[P1] Revalidate every pre-lock tableId decision. tableId is now mutable, but
RESUME/STOP/ALTER and SHOW/SHOW CREATE authorize the old target before
acquiring the job lock, while update() checks the old table's existence before
its write lock. A concurrent switch here can therefore let an old-target user
act on or render the new-target job, or let a stale missing-old lookup
permanently cancel a job already switched to a live table. Bind each
privilege/existence check and its consumer to the same job read/write lock and
current tableId.
##########
fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java:
##########
@@ -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());
Review Comment:
[P1] Recheck LOAD after resolving the target ID. The command checks LOAD
against targetTableName during analysis, but this later critical section
follows only the saved table ID. A concurrent RENAME preserves the ID while
changing the name, so a user authorized for the old name can journal the
renamed table without LOAD on its current name. Recheck LOAD for
table.getName() while these database/table locks are held, before validation or
mutation.
##########
fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java:
##########
@@ -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));
Review Comment:
[P1] Validate with the job's execution context. This direct planner call
runs as the ALTER session, while the real task path first installs
getUserIdentity() and the job's cloud cluster/compute group.
Privilege-sensitive rewrites such as EncryptKeyRef therefore can pass here
under the caller and fail immediately on resume under the stored job user (or
be rejected in the reverse grant split). Share the runtime context setup for
validation and restore the caller context afterward.
##########
fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java:
##########
@@ -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);
Review Comment:
[P1] Apply the same precedence fix to Kinesis. This removal makes an
explicit unique_key_update_mode win in the common/Kafka path, but
KinesisRoutineLoadJob.modifyPropertiesInternal() still rereads the original
partial_columns value and overwrites isPartialUpdate. UPSERT plus
partial_columns=true therefore leaves live application and edit-log replay with
UPSERT/isPartialUpdate=true, while a checkpoint image reload reconstructs false
from jobProperties. Remove the Kinesis override and add the parallel precedence
test.
##########
fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java:
##########
@@ -885,6 +886,51 @@ private void modifyPropertiesInternal(Map<String, String>
jobProperties,
this.id, jobProperties, dataSourceProperties);
}
+ private KafkaDataSourceSnapshot
stageDataSourceProperties(KafkaDataSourceProperties dataSourceProperties)
+ throws DdlException {
+ Map<String, String> stagedCustomProperties =
Maps.newHashMap(customProperties);
+ Map<String, String> stagedConvertedCustomProperties =
Maps.newHashMap(convertedCustomProperties);
+ String stagedKafkaDefaultOffset = kafkaDefaultOffSet;
+ Map<String, String> alteredCustomProperties =
dataSourceProperties.getCustomKafkaProperties();
+ boolean customPropertiesChanged =
MapUtils.isNotEmpty(alteredCustomProperties);
+ if (customPropertiesChanged) {
+ stagedCustomProperties.putAll(alteredCustomProperties);
Review Comment:
[P1] Replace the datetime origin together with the default. If the job
currently has a datetime default, customProperties contains both
kafka_default_offsets and kafka_origin_default_offsets. A later ALTER using the
newly supported direct kafka_default_offsets=OFFSET_BEGINNING/END spelling adds
only the new default here, leaving the old origin;
buildConvertedCustomProperties() then prefers that stale origin, so the ALTER
is ineffective for future partitions. Remove the staged origin whenever a new
default is supplied without one, and cover datetime-to-symbolic ALTER.
--
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]
