ryanmeowy commented on issue #10203: URL: https://github.com/apache/seatunnel/issues/10203#issuecomment-6206801855
@SEZ9 thanks — follow-up to my earlier note, answering your six points. One thing that note did not consider is #11206, which has since grown into the same area, so I start there with how the two relate and what that changes. **Relation to #11206.** #11206 already lands two of the three modes in this space; this design is the third, not a parallel implementation. * `scan.binlog.newly-added-table.enabled` (live, DDL-driven) — the option description and the docs both say it "does not snapshot historical data for those tables". * `scan.newly-added-table.enabled` (snapshots history) — evaluated only on restore, and only through the hybrid branch of `IncrementalSource.restore()` (`:378-395`); a checkpoint taken in the incremental phase never reaches that diff. * this design — live **and** backfilled, which is the intersection the other two miss. | mode | live trigger | backfills pre-existing rows | survives restore | needs exactly-once | | --- | --- | --- | --- | --- | | `scan.binlog.newly-added-table.enabled` | yes | no (documented) | writers restored, no new snapshot | no | | `scan.newly-added-table.enabled` | no — restore only | yes, via the snapshot phase | yes | same caveat as any snapshot | | this design | yes | yes | from the registry | yes | What we would reuse rather than rebuild: its reader DDL channel (`MySqlSchemaChangeResolver` / `CreateTableEvent`) as one of the two discovery triggers, its runtime sink-writer registration (which takes the sink side out of scope here), and its `MySqlDialect.getPrimaryKey` query fallback, which the snapshot of a table absent from `tableMap` needs. What this design adds: admission **while the job runs**; a backfill with an exact `W_high`, which is why exactly-once is a hard precondition here (the live path can afford at-least-once only because it never needs that boundary); the incremental-phase re-drive below; and the registry dedup. Both changes edit `IncrementalSource.java:435` — they are compatible (ours subtracts the registry inside their conditional) but should not be reviewed as one, so I would land #11206 first and deliver the registry + subtraction separately. **1. Enabled on a non-exactly-once job: reject at config validation.** `exactly_once` is a submission-time option (`SourceOptions.java:114-118`, key `exactly_once`, default `false`), so the rule can reject `discovery enabled && !exactly_once` before anything starts; warning-and-disabling means the user believes discovery is on and silently misses rows. It is a real guard, not cosmetics: without exactly-once the backfill read is skipped (`MySqlSnapshotFetchTask.java:86-92`) and `shouldEmit()` degenerates to `position.isAtOrAfter(splitStartWatermark)` (`IncrementalSourceStreamFetcher.java:269-278`), so a newly admitted table has no correct `>= W_high` boundary. It has to be a `ConditionExtension` reading the option, not a second `conditional(...)`: `exactly_once` is already a conditional member of `STARTUP_MODE = INITIAL` (`MySqlIncrementalSourceFactory.java:105-108`) and `OptionRule.Builder` rejects double registration (`OptionRule.java:548-575`). **2. `database.server.id`: stop/restart for v1**, plus the case you asked about. The collision is structural — the live client and a backfill client in the same subtask derive the same id from `serverIdRange.getServerId(subtaskId)` (`MySqlSourceConfigFactory.java:99-102`) and MySQL drops the older one; a concurrent client needs a reserved slot (`2*id`/`2*id+1`) and a `ServerIdRange` change every connector shares. On "dies after the snapshot read but before the backfill completes": `[low, high]` is deliberately **not** persisted — it is re-derived from a fresh `currentBinlogOffset()` on every attempt (`MySqlSnapshotSplitReadTask.doExecute` `:137`/`:149`). What is persisted, in one checkpoint, is `pendingSnapshotTables[T] → admittedTables[T]` and `tableHighWatermarks[T] = W_high`. Before that checkpoint T is an ordinary unfinished snapshot split that the enumerator re-creates and re-runs from a fresh `[low', high']`, with nothing yet emitted; after it, restore rebuilds the fil ter from `W_high`, drops `< W_high`, and starts no second snapshot. The invariant: T enters `tableIds` only in the checkpoint that also carries its watermark. The honest cost is the pause you named — and the reserved slot does not remove it, because the reader is still snapshotting while the stream client buffers, so the bounded queue can stall every table anyway. **3. Cadence: 60s default, enumerator-owned, dedicated timer** — unchanged. Worth adding that #11206's own docs support the "poller is the correctness mechanism" reading: the DDL channel exists only while a stream fetcher is running (`pollSplitRecords()` only polls when `streamFetchTask.isRunning() || isBoundedReadFinished()`, `IncrementalSourceStreamFetcher.java:157`), so DDL during the snapshot phase, a restart, or between the admission decision and the next checkpoint is invisible to it. Tying the timer to `notifyCheckpointComplete()` would make a job without `checkpoint.interval` silently never discover. **4. Discovery and delivery: the enumerator owns it.** Discovery runs in `IncrementalSourceEnumerator` on the existing `run()` → `assignSplits()` path, and the registry is persisted as enumerator state alongside `IncrementalPhaseState` / `HybridPendingSplitsState`, so it survives a subtask restart and is rebuilt on restore. Delivery is deliberately not a reader-side patch: `IncrementalSourceSplitReader.checkSplitOrStartNext()` returns early while a live `IncrementalSourceStreamFetcher` exists, so the only way a newly admitted table reaches a reader is inside the reassembled `IncrementalSplit.tableIds`, re-routed by the existing assignment path; per-table filtering then stays where it already is (`configureFilter` / `shouldEmit`). Suppression is the union of the two existing lists and the registry (`alreadyProcessedTables ∪ remainingTables ∪ registry.*`), which is what makes re-discovery idempotent for the whole job rather than per attempt. Scope: MySQL only in the first vers ion — the design reuses the MySQL DDL channel and `MySqlDialect`; OceanBase should be a separate pass over the same registry, since its binlog protocol and server-id allocation are different implementations. **5. Backward compatibility (your point 5).** With discovery off the registry is empty and the expression reduces to today's `capturedTables − checkpointCapturedTables` unchanged; it only ever removes tables this feature admitted mid-run, which is precisely the set the unchanged expression would re-snapshot. Config-added table semantics are not touched. One honest qualification: the incremental-branch gap in the next point exists on dev today for any table captured but still pending at restore — the design closes it, it does not create it. **6. The part my earlier note got wrong.** (a) The subtraction basis has to include the registry — `newTables = capturedTables − (alreadyProcessedTables ∪ remainingTables ∪ registry.admittedTables ∪ registry.pendingSnapshotTables)` — otherwise a table admitted mid-run, snapshot done and in neither list, gets snapshotted twice on the next hybrid restore. (b) The re-drive is required, not optional: `restore()` runs only on the hybrid branch, so on the incremental branch a table that is still pending is picked up by the rebuilt filter (tableIds come from live `capturedTables`) and streamed from the old `startupOffset` **without ever reading its existing rows** — missing rows, not duplicates. **Sequencing (not one of your six).** #12567 is closed, so the condition you set has no owner: the author closed it unmerged on 2026-10-15, after you and @DanielLeens asked for the minimal Option A sketch inside the PR. The defect it targeted is real and independent of discovery — the restored fallback watermark is the split-level minimum rather than each table's own offset, so at incremental parallelism > 1 a restored table can re-emit under exactly-once. Two questions: 1. Should this work take that fix over as its first, separately-reviewable commit — `addSplits()` restoring `tableWatermarks.put(tableId, startupOffset)` instead of reading `incrementalSplit.getTableStartupOffsets()`, `createIncrementalSplit()` keeping each captured table's own offset, reader-side per-table tracking dropped — or should it keep waiting for whoever picks #12567 up? Either way the registry + restore subtraction is the small part that must land. 2. Third-mode naming: inside #11206's namespace (`scan.binlog.newly-added-table.snapshot.enabled` + `.interval`) or a separate `table.discovery.*`? I lean to the former; it makes the "snapshots what the non-snapshot mode skips" relationship explicit in the option name. Still a design write-up, no PR from me yet. Happy to expand any of the above — the state machine, the test matrix, or the `ServerIdRange` alternative — before we settle on a shape. -- 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]
