github-actions[bot] commented on code in PR #64878:
URL: https://github.com/apache/doris/pull/64878#discussion_r3610840399
##########
fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java:
##########
@@ -2128,23 +2232,49 @@ private void validateFlexiblePartialUpdateForAlter()
throws UserException {
}
OlapTable olapTable = (OlapTable) table;
- // Validate table-level constraints (MoW, skip_bitmap,
light_schema_change, variant columns)
olapTable.validateForFlexiblePartialUpdate();
+ String format =
this.jobProperties.getOrDefault(FileFormatProperties.PROP_FORMAT, "csv");
+ if (!"json".equalsIgnoreCase(format)) {
+ throw new DdlException("Flexible partial update only supports JSON
format, but current job uses: "
+ + format);
+ }
+ if (Boolean.parseBoolean(this.jobProperties.getOrDefault(
+ JsonFileFormatProperties.PROP_FUZZY_PARSE, "false"))) {
+ throw new DdlException("Flexible partial update does not support
fuzzy_parse");
+ }
+ String jsonPaths = getJsonPaths();
+ if (jsonPaths != null && !jsonPaths.isEmpty()) {
+ throw new DdlException("Flexible partial update does not support
jsonpaths");
+ }
+ if (columnDescs != null && !columnDescs.descs.isEmpty()) {
+ throw new DdlException("Flexible partial update does not support
COLUMNS specification");
+ }
+ }
+
+ /**
+ * Validate flexible partial update constraints when altering routine load
job.
+ */
+ private void validateFlexiblePartialUpdateForAlter(OlapTable targetTable,
+ Map<String, String> alteredJobProperties) throws UserException {
+ // Validate table-level constraints (MoW, skip_bitmap,
light_schema_change, variant columns)
+ targetTable.validateForFlexiblePartialUpdate();
+ Map<String, String> effectiveJobProperties =
Maps.newHashMap(jobProperties);
+ effectiveJobProperties.putAll(alteredJobProperties);
Review Comment:
[P1] Validate the staged load descriptor here. A paused JSON UPSERT job with
no current column mapping can issue ALTER ... COLUMNS(k1)
PROPERTIES("unique_key_update_mode"="UPDATE_FLEXIBLE_COLUMNS"); both analysis
and the mutation-time recheck still see the old empty columnDescs, then Kafka
applies and journals the flexible mode plus COLUMNS together. The statement
succeeds with a definition that CREATE ROUTINE LOAD explicitly rejects as
unsupported. Please overlay command.getRoutineLoadDesc() into the effective
definition under the job lock before validating flexible-update constraints.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/AlterRoutineLoadCommand.java:
##########
@@ -315,33 +356,65 @@ 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) {
Review Comment:
[P1] Align this guard with the advertised Kinesis support. The PR
description and release note say that SET TARGET TABLE can be combined with
Kafka/Kinesis source properties and require only that the explicit source type
match the existing job, but every paused single-table Kinesis job is rejected
here. KinesisRoutineLoadJob also has no target-ID/descriptor mutation or replay
path, so the documented statement cannot work. Please either implement atomic
live/replay support for Kinesis or explicitly narrow the release note and
documentation to Kafka-only target switching.
##########
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()) {
+ validateAlterJobPropertiesForMutation(command);
+ }
- modifyPropertiesInternal(jobProperties, dataSourceProperties);
-
- AlterRoutineLoadJobOperationLog log = new
AlterRoutineLoadJobOperationLog(this.id,
- jobProperties, dataSourceProperties);
- Env.getCurrentEnv().getEditLog().logAlterRoutineLoadJob(log);
+ // Kafka offset resolution may perform external I/O. Keep it
outside DB/table metadata locks.
+ PreparedKafkaProperties preparedDataSourceProperties =
+ prepareDataSourceProperties(dataSourceProperties, false);
+ unprotectApplyAlter(command, dataSourceProperties,
preparedDataSourceProperties);
} finally {
writeUnlock();
}
}
- private void convertOffset(KafkaDataSourceProperties dataSourceProperties)
throws UserException {
+ private void unprotectApplyAlter(AlterRoutineLoadCommand command,
+ KafkaDataSourceProperties dataSourceProperties,
+ PreparedKafkaProperties preparedDataSourceProperties) throws
UserException {
+ modifyPropertiesInternal(command.getAnalyzedJobProperties(),
dataSourceProperties,
+ preparedDataSourceProperties);
+ if (command.hasTargetTable()) {
+ tableId = command.getTargetTableId();
+ }
+ setRoutineLoadDesc(command.getRoutineLoadDesc());
+ AlterRoutineLoadJobOperationLog log = new
AlterRoutineLoadJobOperationLog(id,
+ command.getAnalyzedJobProperties(), dataSourceProperties,
command.getTargetTableId(),
+ command.getRoutineLoadDesc());
+ Env.getCurrentEnv().getEditLog().logAlterRoutineLoadJob(log);
+ }
+
+ private List<Pair<Integer, Long>> resolveOffsets(KafkaDataSourceProperties
dataSourceProperties,
+ KafkaDataSourceSnapshot dataSourceSnapshot) throws UserException {
List<Pair<Integer, Long>> partitionOffsets =
dataSourceProperties.getKafkaPartitionOffsets();
if (partitionOffsets.isEmpty()) {
- return;
+ return partitionOffsets;
}
- List<Pair<Integer, Long>> newOffsets;
if (dataSourceProperties.isOffsetsForTimes()) {
- newOffsets = KafkaUtil.getOffsetsForTimes(brokerList, topic,
convertedCustomProperties, partitionOffsets);
- } else {
- newOffsets = KafkaUtil.getRealOffsets(brokerList, topic,
convertedCustomProperties, partitionOffsets);
+ return KafkaUtil.getOffsetsForTimes(dataSourceSnapshot.brokerList,
dataSourceSnapshot.topic,
+ dataSourceSnapshot.convertedCustomProperties,
partitionOffsets);
}
- dataSourceProperties.setKafkaPartitionOffsets(newOffsets);
+ return KafkaUtil.getRealOffsets(dataSourceSnapshot.brokerList,
dataSourceSnapshot.topic,
+ dataSourceSnapshot.convertedCustomProperties,
partitionOffsets);
}
private void modifyPropertiesInternal(Map<String, String> jobProperties,
KafkaDataSourceProperties
dataSourceProperties)
throws UserException {
- if (null != dataSourceProperties) {
- List<Pair<Integer, Long>> kafkaPartitionOffsets =
Lists.newArrayList();
- Map<String, String> customKafkaProperties = Maps.newHashMap();
-
- if
(MapUtils.isNotEmpty(dataSourceProperties.getOriginalDataSourceProperties())) {
- kafkaPartitionOffsets =
dataSourceProperties.getKafkaPartitionOffsets();
- customKafkaProperties =
dataSourceProperties.getCustomKafkaProperties();
- }
-
- // convertCustomProperties and check partitions before reset
progress to make modify operation atomic
- if (!customKafkaProperties.isEmpty()) {
- this.customProperties.putAll(customKafkaProperties);
- convertCustomProperties(true);
- }
-
- if (!kafkaPartitionOffsets.isEmpty()) {
- ((KafkaProgress)
progress).checkPartitions(kafkaPartitionOffsets);
- }
+ PreparedKafkaProperties preparedDataSourceProperties =
prepareDataSourceProperties(dataSourceProperties, true);
+ modifyPropertiesInternal(jobProperties, dataSourceProperties,
preparedDataSourceProperties);
+ }
- if (Config.isCloudMode()) {
- Cloud.ResetRLProgressRequest.Builder builder =
Cloud.ResetRLProgressRequest.newBuilder()
-
.setRequestIp(FrontendOptions.getLocalHostAddressCached());
- builder.setCloudUniqueId(Config.cloud_unique_id);
- builder.setDbId(dbId);
- builder.setJobId(id);
- if (!kafkaPartitionOffsets.isEmpty()) {
- Map<Integer, Long> partitionOffsetMap = new HashMap<>();
- for (Pair<Integer, Long> pair : kafkaPartitionOffsets) {
- // The reason why the value recorded in MS in cloud
mode needs to be subtracted by one is
- // this value will be incremented
- // when pulling MS persistent progress data and
updating memory
- // in routineLoadJob.updateCloudProgress().
- partitionOffsetMap.put(pair.first, pair.second - 1);
- }
- builder.putAllPartitionToOffset(partitionOffsetMap);
- }
- resetCloudProgress(builder);
+ private void modifyPropertiesInternal(Map<String, String> jobProperties,
+ KafkaDataSourceProperties
dataSourceProperties,
+ PreparedKafkaProperties
preparedDataSourceProperties)
+ throws UserException {
+ if (null != dataSourceProperties) {
+ Preconditions.checkNotNull(preparedDataSourceProperties);
+ KafkaDataSourceSnapshot dataSourceSnapshot =
preparedDataSourceProperties.dataSourceSnapshot;
+ List<Pair<Integer, Long>> kafkaPartitionOffsets =
preparedDataSourceProperties.kafkaPartitionOffsets;
+
dataSourceProperties.setKafkaPartitionOffsets(kafkaPartitionOffsets);
+
+ if (dataSourceSnapshot.customPropertiesChanged) {
+ this.customProperties.clear();
+
this.customProperties.putAll(dataSourceSnapshot.customProperties);
+ this.convertedCustomProperties.clear();
+
this.convertedCustomProperties.putAll(dataSourceSnapshot.convertedCustomProperties);
+ this.kafkaDefaultOffSet =
dataSourceSnapshot.kafkaDefaultOffset;
}
// It is necessary to reset the Kafka progress cache if topic
change,
// and should reset cache before modifying partition offset.
- if (!Strings.isNullOrEmpty(dataSourceProperties.getTopic())) {
- this.topic = dataSourceProperties.getTopic();
+ if (dataSourceSnapshot.resetProgress) {
+ this.topic = dataSourceSnapshot.topic;
this.progress = new KafkaProgress();
Review Comment:
[P1] Re-seed pinned partitions on a topic-only switch. A job created with
kafka_partitions=0 can be paused and altered with only kafka_topic=B; this
branch clears progress but leaves customKafkaPartitions/currentKafkaPartitions
at [0]. On resume the pinned branch skips Kafka refresh and returns false, so
updateNewPartitionProgress() is never called, and divideRoutineLoadJob() puts a
null offset into the new task even when B has partition 0. Please either
require explicit offsets for a pinned topic switch or validate the retained IDs
against the staged topic and seed them from the effective default before
committing.
--
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]
