github-actions[bot] commented on code in PR #69673:
URL: https://github.com/apache/doris/pull/69673#discussion_r4386148913
##########
fe/fe-core/src/main/java/org/apache/doris/job/offset/jdbc/JdbcSourceOffsetProvider.java:
##########
@@ -160,6 +165,63 @@ public void ensureInitialized(Long jobId, Map<String,
String> newProps) throws J
DataSourceConfigKeys.SNAPSHOT_PARALLELISM_DEFAULT));
}
+ @Override
+ public boolean hasPendingIncludeTablesChange() {
+ return addedTables != null;
+ }
+
+ private boolean hasCommittedIncrementalOffset() {
+ return MapUtils.isNotEmpty(binlogOffsetPersist);
+ }
+
+ @Override
+ public void validateAlterIncludeTables(List<String> updatedTables) throws
JobException {
+ synchronized (splitsLock) {
+ if (!hasCommittedIncrementalOffset()) {
+ throw new JobException("Cannot alter include_tables before an
incremental offset is committed");
+ }
+ if (hasPendingIncludeTablesChange() &&
!updatedTables.containsAll(addedTables)) {
+ throw new JobException("Cannot remove newly added tables until
incremental synchronization resumes");
+ }
+ }
+ }
+
+ @Override
+ public void applyIncludeTablesChange(List<String> updatedTables,
List<String> addedTables) throws JobException {
+ synchronized (splitsLock) {
+ // Added-table schemas arrive with fetched splits; keep only
retained schemas here.
+ String retainedSchemas = filterTableSchemas(tableSchemas,
updatedTables);
+ if (!hasPendingIncludeTablesChange()) {
+ // Reset snapshot progress while preserving the committed
incremental offset for resumption.
+ clearSnapshotState();
+ this.addedTables = new ArrayList<>();
+ }
+ // Extend the pending snapshot without resetting its chunks,
watermarks or splitting cursor.
+ this.addedTables.addAll(addedTables);
+ cachedSyncTables = new ArrayList<>(updatedTables);
+ setTableSchemas(retainedSchemas);
+ hasMoreData = true;
+ }
+ }
+
+ private JdbcOffset createIncrementalOffset() {
+ Map<String, String> offset = new HashMap<>(binlogOffsetPersist);
+ return new JdbcOffset(Collections.singletonList(new
BinlogSplit(offset)));
Review Comment:
[P2] Exclude already-snapshotted DDL from the added table's replay
Consider a job capturing only `a`, paused at committed offset T0. While
paused, an existing excluded table `b(id INT PRIMARY KEY, v INT)` executes
`ALTER TABLE b MODIFY COLUMN v BIGINT`; then ALTER JOB adds `b`. Target
creation and the snapshot correctly use BIGINT, but this split resumes every
table from T0 without a per-table schema boundary. The historical MODIFY is
emitted again ([Debezium's ALTER listener signals it even against the current
definition](https://raw.githubusercontent.com/debezium/debezium/v1.9.8.Final/debezium-connector-mysql/src/main/java/io/debezium/connector/mysql/antlr/listener/AlterTableParserListener.java));
in evolve mode, `MySqlDebeziumJsonDeserializer:179-199` rejects MODIFY and
pauses the valid job as an unsupported schema change. An extra explicit RESUME
is then required despite no post-snapshot schema change. Preserve T0 for
retained tables while preventing pre-snapshot DDL for added tables from being
treated as new target changes.
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/controller/ClientController.java:
##########
@@ -66,8 +66,8 @@ public Object initSourceReader(@RequestBody JobBaseConfig
jobConfig) {
public Object fetchSplits(@RequestBody FetchTableSplitsRequest ftsReq) {
try {
SourceReader reader = Env.getCurrentEnv().getReader(ftsReq);
- List splits = reader.getSourceSplits(ftsReq);
- return RestResponse.success(splits);
+ FetchTableSplitsResult<AbstractSourceSplit> result =
reader.getSourceSplits(ftsReq);
+ return RestResponse.success(result);
Review Comment:
[P1] Preserve the split response during rolling upgrades
This changes `data` from an array to an object for every caller, including
requests with `includeTableSchema=false`. Old FE parses `List<SnapshotSplit>`,
while the new FE only parses `FetchTableSplitsResult`. Each BE runs its bundled
CDC client and forwards the JSON unchanged
(`cdc_client_mgr.cpp:615-616,796-833`), so either upgrade order leaves ordinary
initial/snapshot jobs failing to decode splits while FE and BE versions differ.
Negotiate the extended response and accept the legacy array in the new FE so
existing jobs keep working during an upgrade.
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/source/reader/JdbcIncrementalSourceReader.java:
##########
@@ -185,7 +192,7 @@ private ChunkSplitterState buildChunkSplitterState(
if (pkValues == null || pkValues.length == 0) {
return ChunkSplitterState.NO_SPLITTING_TABLE_STATE;
}
- TableChanges.TableChange tableChange =
getTableSchemas(ftsReq).get(tableId);
+ TableChanges.TableChange tableChange = getTableSchema(tableId, ftsReq);
Review Comment:
[P2] Refresh split metadata when a removed table is re-added
`getTableSchema` returns an existing cached entry unchanged. A metadata-only
reader on another BE survives PAUSE and ordinary snapshot cleanup, and
`rebuildReader` only refreshes readers handling write requests. For example, an
earlier multi-batch PostgreSQL snapshot caches `b` with `id BIGINT` as its
primary key and a `code VARCHAR` column. Remove `b`, switch its primary key to
nonnumeric `code` while excluded, then re-add it. [The fresh splitter uses the
current primary
key](https://raw.githubusercontent.com/apache/flink-cdc/release-3.6.0/flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/splitter/JdbcSourceChunkSplitter.java),
but a resumed split RPC on that old metadata BE gets the cached `id` key here;
`resolveSplitKeyClass` selects Long and `convertBounds` fails on the string
cursor. With enough rows for multiple batches, that split request fails and the
scheduler pauses the job. Use fresh schema fo
r split-cursor reconstruction or invalidate metadata schema/key-class caches
for each table-set generation, while retaining offset-aligned schemas for
record decoding.
--
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]
