0AyanamiRei commented on code in PR #64878:
URL: https://github.com/apache/doris/pull/64878#discussion_r3610398814
##########
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:
done
--
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]
