github-actions[bot] commented on code in PR #64878:
URL: https://github.com/apache/doris/pull/64878#discussion_r3610163485
##########
fe/fe-core/src/main/java/org/apache/doris/load/routineload/kafka/KafkaRoutineLoadJob.java:
##########
@@ -766,91 +786,95 @@ public Map<String, String> getCustomProperties() {
return getMaskedCustomProperties("property.");
}
+ @Override
+ public boolean appliesRoutineLoadDescAtomically() {
+ return true;
+ }
+
@Override
public void modifyProperties(AlterRoutineLoadCommand command) throws
UserException {
- Map<String, String> jobProperties = command.getAnalyzedJobProperties();
KafkaDataSourceProperties dataSourceProperties =
(KafkaDataSourceProperties) command.getDataSourceProperties();
- if (null != dataSourceProperties) {
- // if the partition offset is set by timestamp, convert it to real
offset
- convertOffset(dataSourceProperties);
- }
writeLock();
try {
if (getState() != JobState.PAUSED) {
throw new DdlException("Only supports modification of PAUSED
jobs");
}
+ if (command.getRoutineLoadDesc() != null &&
command.getLoadPropertyTableId() != tableId) {
+ throw new DdlException("Routine load target table changed
while validating load properties; "
+ + "please retry ALTER ROUTINE LOAD");
+ }
+ if (!command.hasTargetTable()) {
Review Comment:
[P1] Revalidate the saved target against the current job state under this
lock. A target-only ALTER can validate non-MoW B while the job is UPSERT; while
it waits, another ALTER can change paused MoW job A to fixed partial update.
This path then skips mutation-time validation solely because it carries a
target and installs B, so resume later fails the partial-update/MoW planner
check. The existing target-ID guard applies only to a command carrying a load
clause and does not close this inverse race. Record/check a job-definition
generation, or recompute the effective mode and revalidate B before assigning
tableId; cover the two-ALTER interleaving.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterRoutineLoadCommand.java:
##########
@@ -315,33 +356,64 @@ private void checkJobProperties() throws UserException {
}
}
- private void checkDataSourceProperties() throws UserException {
+ private void checkDataSourceProperties(RoutineLoadJob job) throws
UserException {
if (MapUtils.isEmpty(dataSourceMapProperties)) {
return;
}
- RoutineLoadJob job = Env.getCurrentEnv().getRoutineLoadManager()
- .getJob(getDbName(), getJobName());
+ String effectiveDataSourceType = dataSourceType == null ?
job.getDataSourceType().name() : dataSourceType;
+ if
(!effectiveDataSourceType.equalsIgnoreCase(job.getDataSourceType().name())) {
+ throw new AnalysisException("The specified job type is not: " +
effectiveDataSourceType);
+ }
this.dataSourceProperties = RoutineLoadDataSourcePropertyFactory
- .createDataSource(job.getDataSourceType().name(),
dataSourceMapProperties, job.isMultiTable());
+ .createDataSource(effectiveDataSourceType,
dataSourceMapProperties, job.isMultiTable());
dataSourceProperties.setAlter(true);
- dataSourceProperties.setTimezone(job.getTimezone());
+ dataSourceProperties.setTimezone(analyzedJobProperties.getOrDefault(
+ CreateRoutineLoadInfo.TIMEZONE, job.getTimezone()));
dataSourceProperties.analyze();
}
- private void checkPartialUpdate() throws UserException {
- if (!isPartialUpdate) {
+ private void validateAlterJobProperties(RoutineLoadJob job,
+ TUniqueKeyUpdateMode effectiveUniqueKeyUpdateMode) throws
UserException {
+ if (effectiveUniqueKeyUpdateMode == TUniqueKeyUpdateMode.UPSERT) {
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.");
+ throw new AnalysisException("Partial update is not supported in
multi-table load.");
}
Database db =
Env.getCurrentInternalCatalog().getDbOrAnalysisException(job.getDbFullName());
Table table = db.getTableOrAnalysisException(job.getTableName());
- if (isPartialUpdate && !((OlapTable)
table).getEnableUniqueKeyMergeOnWrite()) {
- throw new AnalysisException("load by PARTIAL_COLUMNS is only
supported in unique table MoW");
+ job.validateAlterJobProperties((OlapTable) table,
analyzedJobProperties, effectiveUniqueKeyUpdateMode);
+ }
+
+ private void validateTargetTable(ConnectContext ctx, RoutineLoadJob job,
+ TUniqueKeyUpdateMode effectiveUniqueKeyUpdateMode) 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");
+ }
+ Database db =
Env.getCurrentInternalCatalog().getDbOrAnalysisException(job.getDbFullName());
+ Table table = db.getTableOrAnalysisException(targetTableName);
Review Comment:
[P1] Authorize before resolving the target metadata. doRun() completes
validate() before RoutineLoadManager checks access to the current job, and this
lookup plus the non-OLAP/temporary checks also precede the target LOAD check. A
limited-RBAC user can therefore distinguish a missing target from an
unauthorized ordinary table, and can classify non-OLAP or temporary targets
from their specific errors, even without access to the current job. This
violates the threat model's object non-discovery property. Authorize the
current job first, then check LOAD for the requested target name before
resolving it; unauthorized existing and nonexistent targets should follow the
same access-denied path without metadata lookup.
--
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]
