Copilot commented on code in PR #69673:
URL: https://github.com/apache/doris/pull/69673#discussion_r4386059349
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java:
##########
@@ -535,6 +541,15 @@ public void alterJob(AlterJobCommand alterJobCommand)
throws AnalysisException,
List<String> logParts = new ArrayList<>();
validateComputeGroupProperty(alterJobCommand.getProperties());
+ List<String> updatedTables = null;
Review Comment:
A separate offset-only ALTER after adding table B clears the pending
snapshot round through `updateOffset()`, while B remains in `syncTables`.
Resuming then skips B's historical rows. Reject offset changes while a
table-change round is pending, restoring persisted provider state before
checking so the restriction also holds after an FE restart.
##########
fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProvider.java:
##########
@@ -604,19 +677,27 @@ public void replayIfNeed(StreamingInsertJob job) throws
JobException {
getJobId(),
binlogOffsetPersist == null ? 0 :
binlogOffsetPersist.size(),
chunkHighWatermarkMap == null ? 0 :
chunkHighWatermarkMap.size());
- if (MapUtils.isNotEmpty(binlogOffsetPersist)) {
+ boolean tableChangePending = hasPendingIncludeTablesChange();
+ if (hasCommittedIncrementalOffset() && !tableChangePending) {
currentOffset = new JdbcOffset();
currentOffset.setSplits(Collections.singletonList(new
BinlogSplit(binlogOffsetPersist)));
} else {
Map<String, List<SnapshotSplit>> snapshotSplits =
StreamingJobUtils.restoreSplitsToJob(job.getJobId());
- if (MapUtils.isNotEmpty(chunkHighWatermarkMap) &&
MapUtils.isNotEmpty(snapshotSplits)) {
- List<SnapshotSplit> lastSnapshotSplits =
- recalculateRemainingSplits(chunkHighWatermarkMap,
snapshotSplits);
+ if (tableChangePending) {
+ snapshotSplits.entrySet().removeIf(entry ->
!addedTables.contains(getTableName(entry.getKey())));
+ }
+ // Added-table chunks must also be restored before their first
watermark is committed.
+ if (tableChangePending
+ || (MapUtils.isNotEmpty(chunkHighWatermarkMap) &&
MapUtils.isNotEmpty(snapshotSplits))) {
+ List<SnapshotSplit> nextSnapshotSplits =
recalculateRemainingSplits(
+ chunkHighWatermarkMap == null ? new HashMap<>() :
chunkHighWatermarkMap, snapshotSplits);
resumeCdcSplitProgressFromSplits();
Review Comment:
If FE restarts after an added table's complete chunk list is persisted but
before its first snapshot commit, replay restores the chunks without its schema
and schedules no further schema fetch. Ordinary snapshot commits do not return
the locally discovered schema, so subsequent CDC reloads the incomplete FE
baseline. Make the added-table schema recoverable with the chunks or return it
through snapshot commits, and test this crash window before allowing
incremental handoff.
--
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]
