Hi Anton, Thanks for your advice. I have updated the FIP based on our discussion.
Best, Hongshun On Sat, May 23, 2026 at 12:36 AM Anton Borisov <[email protected]> wrote: > Hi Hongshun, > > Thanks for the clarification. Most points are clear now. I have a few > remaining comments. > > 1. Older schemaId on writes > > I think option (1) matches the current code better. > Today, older schemaId means an additive older schema: SchemaUpdate only > allows ADD COLUMN LAST, while DROP / RENAME / MODIFY are rejected. > KvTablet.validateSchemaId also rejects only newer/negative schemaIds, and > PaddingRow already pads shorter rows for the KV merge. > > So I do not think we need to reject older schemaIds in Phase 1. The writer > can: > > - send directly if record.schemaId == cached schemaId; > - otherwise flush, refresh metadata, and: > - fail if record.schemaId > refreshed schemaId; > - send with record.schemaId if record.schemaId <= refreshed schemaId. > > This also keeps the multi-table writer aligned with the existing > single-table writer, where the cached schemaId may lag and writes still > work. > If non-additive schema evolution is added later, we can revisit this > contract then. Let me know if I'm missing something with my reasoning. > > 2. Cross-table fairness > > Ack. The bucket rotation makes sense to me. A short note is more than > enough, > thank you > > 3. Drop+recreate > > The protection does not seem to be there, at least in the PoC code. > MultiTableLogScannerImpl.registerIfAbsent() short-circuits on the > TablePath cache and does not re-resolve the table. admin.getTableInfo(...) > only runs on the first subscribe. So a resubscribe after drop+recreate can > silently reuse the stale TableInfo and assign a TableBucket with the old > tableId. > > PR #2747 does not cover this case either. It only touches client/write/* > and aborts batches in RecordAccumulator.drainBatchesForOneNode, so it does > not affect the scanner/fetch path. > I will help to review this PR, it seems, thank you for referencing it! > > For Phase 1, we might re-resolve on every subscribe(TablePath, ...) > and throw if the tableId changed > and on the fetch path invalidate the cachedTableInfo and throw from > the next poll() once a metadata refresh confirms > the bucket's tableId is gone - the read-side mirror of #2747 in essence. > > -- Anton > > пт, 22 мая 2026 г. в 09:28, Hongshun Wang <[email protected]>: > > > > Hi Anton, > > > > > > Thanks for the detailed review! Replies inline below. > > > > --- > > > > > For example, a pipeline may read a record at schemaId = 3 and write it > > back after the target table has moved to schemaId = 5. > > > > > > Good point. I had the same concern. In the current Fluss server design, > > writing with an older schemaId actually works — the server tolerates it > and > > pads missing columns with `null`. The open question is whether we should: > > > > 1. **Allow** the writer to pass an older schemaId and let the server > > reconcile it (current behavior), or > > 2. **Force** the caller to resolve schema evolution outside the writer > > client (i.e., the writer must always be called with the latest schemaId). > > > > I lean toward option (2) for safety — silently padding `null` for newly > > added columns is fine, but for more complex evolutions (e.g., `ALTER > COLUMN > > TYPE`) it can cause subtle data corruption. Pushing schema reconciliation > > to the pipeline layer makes the contract more explicit. Happy to hear > > thoughts here. > > > > --- > > > > > Please add a "Scope (Phase 1)" to make the tradeoffs and scope > explicit. > > > > > > Agreed. For partial-update / target-columns on writes, and projection / > > filter on multi-table reads — these are not yet required for the primary > > use case driving this FIP (Flink CDC whole-database sync), and they > involve > > non-trivial coupling with the SQL optimizer layer for the single-table > > path. I will explicitly mark them as **out of scope for Phase 1** in the > > FIP and add a follow-up section noting we can revisit them when a > concrete > > use case shows up. > > > > --- > > > > > The FIP does not specify what fairness guarantees MultiTableLogScanner > > provides when subscribed buckets span multiple tables on the same leader. > > > > > > This FIP does not change the existing fetch protocol. The fetch behavior > is > > identical to the single-table case: > > > > - Each fetch request to a leader includes **all subscribed buckets** that > > reside on that leader (regardless of which table they belong to). > > - The server processes buckets in request order, so earlier buckets in > the > > request may be served first within a single round. > > - On the client side, after a `CompletedFetch` is consumed, that bucket > is > > **moved to the tail** of the fetch queue, so over time fairness is > achieved > > at **bucket granularity** (not table granularity). > > > > So there is no per-table fairness guarantee — fairness is per-bucket, > same > > as today's single-table scanner. I will add a clarifying note about this > in > > the FIP. > > > > Still some unfair problems such as what kafka never solves in > > https://issues.apache.org/jira/browse/KAFKA-19415. We can discuss > another > > issue. > > > > --- > > > > > The FIP does not specify what happens if a table is dropped and > recreated > > under the same TablePath while a scanner is subscribed. > > > > The behavior is consistent with the single-table scanner today, and > > detection happens at **two levels**: > > > > 1. **At `subscribe(...)` time** — if a stale `TablePath → tableId` > mapping > > is detected for an already-subscribed table, the call throws an > exception. > > The scanner will not silently start reading the new table's data. > > 2. **At fetch time** — even if the scanner already holds buckets from the > > old `tableId`, the fetch path will surface a `not leader or follower` / > > `tableId changed` error, so stale subscriptions cannot silently consume > > data from a recreated table. This protection was added in [#2747]( > > https://github.com/apache/fluss/pull/2747). > > > > So silently mixing data from two different `tableId`s under the same > > `TablePath` is impossible by design — both the subscribe path and the > fetch > > path enforce it. > > --- > > > > > > Thanks again for the careful review! > > > > Best, > > Hongshun > > > > On Thu, May 21, 2026 at 9:40 PM Jark Wu <[email protected]> wrote: > > > > > Thanks Hongshun for the updates. > > > > > > The current design looks good to me, and I agree with Anton's points. > > > Once his comments are addressed, I think we can start the vote. > > > > > > Best, > > > Jark > > > > > > On Thu, 21 May 2026 at 19:13, Anton Borisov <[email protected]> > wrote: > > > > > > > > Hi Hongshun, > > > > > > > > Thanks for iterating on this. I like the current shape(+1) and think > > > > it's a very much needed addition. > > > > I studied the FIP and have some thoughts, hope you find it useful. > > > > > > > > 1. Historical schemaId semantics on writes > > > > The FIP specifies fail-fast behavior when a record's schemaId is > newer > > > > than the server's current schema, but does not specify what happens > > > > when the record's schemaId is older. > > > > This matters for the CDC use case in section 2. For example, a > > > > pipeline may read a record at schemaId = 3 and write it back after > the > > > > target table has moved to schemaId = 5. > > > > > > > > If this is out of scope for phase 1, for example because it requires > > > > partial-update / target-columns semantics, please state that > > > > explicitly. Otherwise, please specify the expected writer behavior > for > > > > historical schemaIds. > > > > > > > > 2. Phase-1 scope > > > > From the PoC, phase 1 appears to exclude: > > > > - partial-update / target-columns on writes, implying > MergeMode.DEFAULT > > > only; > > > > - projection / filter on multi-table reads; > > > > - bounded multi-table reads, since MultiTableBatchScanner is > currently > > > > only a placeholder. > > > > > > > > The FIP already refers to "Phase 1" in the WriteResult code block. > > > > Please add a "Scope (Phase 1)" to make the tradeoffs and scope > > > > explicit. > > > > > > > > 3. Cross-table fetch fairness > > > > The FIP does not specify what fairness guarantees > MultiTableLogScanner > > > > provides when subscribed buckets span multiple tables on the same > > > > leader. > > > > Please clarify whether the intended contract is per-bucket fairness > > > > only, per-table fairness, or something else, and whether there is any > > > > related configuration. > > > > > > > > One follow-up item: > > > > the FIP does not specify what happens if a table is dropped and > > > > recreated under the same TablePath while a scanner is subscribed. > > > > Since Fluss identifies tables by tableId rather than path, recreate > > > > produces a new tableId under a stable path. It would be useful to > > > > document the recovery behavior or list this as a known limitation. > > > > > > > > What do you think? > > > > > > > > -- Anton > > > > > > > > чт, 21 мая 2026 г. в 10:20, Hongshun Wang <[email protected]>: > > > > > > > > > > Hi devs, > > > > > > > > > > I have modified this FLIP[1] again: > > > > > 1. Rename MultiTableScannerRecord -> MultiTableScanRecord. > > > > > 2. remove the optional schema ID behavior and require users to > always > > > > > specify a schema ID when writing. > > > > > > > > > > Look forward to hearing from you. And I will start a vote soon if > > > there is > > > > > no problem any more. > > > > > > > > > > > > > > > Best, > > > > > Hongshun > > > > > > > > > > On Wed, May 20, 2026 at 11:17 AM Hongshun Wang < > > > [email protected]> > > > > > wrote: > > > > > > > > > > > Hi jark, > > > > > > The design intention is to simplify usage. When no schema > > > > > > ID is specified, the writer defaults to using the table schema > that > > > was > > > > > > cached during the first subscribe call. During write time, it > > > validates > > > > > > whether the number of columns in the row matches the cached > schema. > > > > > > > > > > > > The current "no schema ID" approach is admittedly coarse-grained > — > > > it only > > > > > > checks that the column count matches, but does not verify that > the > > > columns > > > > > > themselves are semantically correct. For example, if we later > > > support ALTER > > > > > > COLUMN TYPE, a row could pass the column-count check but > actually > > > have > > > > > > mismatched column semantics, leading to silent data corruption. > > > > > > > > > > > > > > > > > > Given this, I think it's also reasonable to remove the optional > > > schema ID > > > > > > behavior and require users to always specify a schema ID when > > > writing. This > > > > > > would make the contract stricter and safer, especially as schema > > > evolution > > > > > > capabilities grow. > > > > > > > > > > > > WDYT? > > > > > > > > > > > > Best, > > > > > > Hongshun > > > > > > > > > > > > > > > > > > On Tue, May 19, 2026 at 12:13 PM Jark Wu <[email protected]> > wrote: > > > > > > > > > > > >> Hi Hongshun, > > > > > >> > > > > > >> The new FIP looks good to me in general. I only have some minor > > > comments. > > > > > >> > > > > > >> 1. Rename MultiTableScannerRecord -> MultiTableScannRecord, > > > > > >> MultiTableScannerRecords -> MultiTableScanRecords to align with > > > > > >> ScanRecord. > > > > > >> > > > > > >> 2. MultiTableWriter supports MultiTableWriteRecord without > schema > > > ID. > > > > > >> What's the behavior when there is no schema id on the write > record? > > > > > >> What happens when the write record has different schema? > > > > > >> > > > > > >> Best, > > > > > >> Jark > > > > > >> > > > > > >> On Mon, 18 May 2026 at 20:39, Hongshun Wang < > > > [email protected]> > > > > > >> wrote: > > > > > >> > > > > > > >> > Hi Devs, > > > > > >> > > > > > > >> > I have modified this FLIP[1]: tables are registered > dynamically > > > on the > > > > > >> > first MultiTableLogScanner::subscribe(...) call rather than > > > > > >> > pre-defined in MultiTableScan:: > > > > > >> > withTable before MultiTableLogScanner is created. And also , > the > > > > > >> > multiple table writer is added. > > > > > >> > > > > > > >> > Poc code is here: https://github.com/apache/fluss/pull/3140 > > > > > >> > > > > > > >> > > > > > > >> > [1] > > > > > >> > > > > > > >> > > > > https://cwiki.apache.org/confluence/display/FLUSS/FIP-43++Fluss+Multi-Table+Subscription+API > > > > > >> > > > > > > >> > Best, > > > > > >> > Hongshun > > > > > >> > > > > > > >> > On Thu, May 14, 2026 at 2:14 PM Hongshun Wang < > > > [email protected]> > > > > > >> > wrote: > > > > > >> > > > > > > >> > > Hi Jark, > > > > > >> > > I agree with you, the new interface is better. The only > issue > > > is that > > > > > >> > > `MultiTableScan withTable(TablePath tablePath)` and > > > `MultiTableScan > > > > > >> > > withTable(TablePath tablePath, int[] projectedColumns)` is > > > > > >> unnecessary, > > > > > >> > > because multi-table subscription often involves dynamically > > > adding new > > > > > >> > > tables to LogFetcher during runtime, rather than declaring > them > > > in > > > > > >> advance. > > > > > >> > > > > > > > >> > > Best, > > > > > >> > > Hongshun > > > > > >> > > > > > > > >> > > On Tue, May 12, 2026 at 7:21 PM Jark Wu <[email protected]> > > > wrote: > > > > > >> > > > > > > > >> > >> Thanks for the proposal. > > > > > >> > >> > > > > > >> > >> Overall, I think this is a very strong FIP that further > aligns > > > our > > > > > >> API > > > > > >> > >> capabilities with KafkaConsumer, particularly regarding > > > multi-topic > > > > > >> > >> consumption. > > > > > >> > >> > > > > > >> > >> However, I have several concerns regarding the current API > > > design: > > > > > >> > >> > > > > > >> > >> 1. The Connection serves as the entry point for the Fluss > API, > > > > > >> > >> providing interfaces for Admin (DDL) and Table (per-table > > > > > >> read/write). > > > > > >> > >> However, LogScannerGroup does not appear to be an object at > > > the same > > > > > >> > >> hierarchical level as Admin or Table. > > > > > >> > >> > > > > > >> > >> 2. I have frequently encountered user requests in the > > > community for a > > > > > >> > >> single API to support multi-table writes in Fluss. I > believe we > > > > > >> should > > > > > >> > >> consider the design of multi-table write APIs alongside > > > multi-table > > > > > >> > >> read APIs to avoid potential redesigns in the future. > > > > > >> > >> > > > > > >> > >> 3. The use cases for multi-table read/write operations > differ > > > > > >> > >> significantly from those for single-table operations. I > suggest > > > > > >> > >> isolating these APIs to allow them to evolve independently > > > rather > > > > > >> than > > > > > >> > >> making them interdependent. For instance, query push-down > is > > > > > >> currently > > > > > >> > >> less critical for multi-table scenarios, and having > > > LogScannerGroup > > > > > >> > >> depend on LogScanner feels awkward from a usability > > > perspective. > > > > > >> > >> > > > > > >> > >> 4. We should minimize changes to foundational APIs such as > > > LogRecord > > > > > >> > >> and ScanRecord to avoid significant impacts on existing > > > pipelines, > > > > > >> > >> particularly concerning performance and compatibility. > > > > > >> > >> > > > > > >> > >> Therefore, I propose introducing a MultiTable interface as > the > > > entry > > > > > >> > >> point for multi-table read/write operations, corresponding > to > > > the > > > > > >> > >> single-table Table interface. Additionally, we can provide > a > > > > > >> > >> MultiTableLogScanner to support core multi-table > subscription > > > > > >> > >> functionality. I have drafted a design outline [1] for your > > > > > >> reference. > > > > > >> > >> > > > > > >> > >> What do you think? > > > > > >> > >> > > > > > >> > >> Best, > > > > > >> > >> Jark > > > > > >> > >> > > > > > >> > >> [1] > > > https://gist.github.com/wuchong/d4bebe0f3fe610f1de730a05ab520c7d > > > > > >> > >> > > > > > >> > >> On Fri, 8 May 2026 at 16:54, Hongshun Wang < > > > [email protected]> > > > > > >> > >> wrote: > > > > > >> > >> > > > > > > >> > >> > Hi devs, > > > > > >> > >> > > > > > > >> > >> > Recently, I have been working on the Fluss YAML Source to > > > support > > > > > >> > >> > full-database synchronization (whole-database CDC). > During > > > the > > > > > >> > >> > implementation, I ran into two fundamental limitations > in the > > > > > >> > >> > current LogScanner API: > > > > > >> > >> > > > > > > >> > >> > 1. LogScanner can only subscribe to a single table. When > a > > > business > > > > > >> > >> needs > > > > > >> > >> > to consume change logs from multiple tables > simultaneously, > > > users > > > > > >> must > > > > > >> > >> > create a separate LogScanner for each table and manually > > > manage > > > > > >> > >> > multiplexing at the application layer. This leads to > thread > > > > > >> > >> proliferation, > > > > > >> > >> > resource waste (duplicate connections to the same > > > TabletServer), > > > > > >> and the > > > > > >> > >> > inability to unify consumption into a single poll call. > > > > > >> > >> > > > > > > >> > >> > 2. LogScanner truncates logs by expected schema, losing > > > schemaId > > > > > >> and raw > > > > > >> > >> > data. The current implementation projects records > according > > > to the > > > > > >> > >> schema > > > > > >> > >> > provided at scanner creation time, which strips the > schemaId > > > > > >> metadata > > > > > >> > >> and > > > > > >> > >> > discards columns added by subsequent schema changes. For > CDC > > > > > >> scenarios, > > > > > >> > >> we > > > > > >> > >> > need to faithfully forward the original record data along > > > with its > > > > > >> > >> schema > > > > > >> > >> > version to downstream systems. > > > > > >> > >> > > > > > > >> > >> > To address these issues, I'd like to propose FIP-43: > > > Multi-Table > > > > > >> > >> > Subscription API[1] to solve both problems. > > > > > >> > >> > > > > > > >> > >> > Looking forward to your feedback and suggestions! > > > > > >> > >> > > > > > > >> > >> > > > > > > >> > >> > Best, > > > > > >> > >> > > > > > > >> > >> > Hongshun > > > > > >> > >> > > > > > > >> > >> > [1] > > > > > >> > >> > > > > > > >> > >> > > > > > >> > > > > https://cwiki.apache.org/confluence/display/FLUSS/FIP-43++Fluss+Multi-Table+Subscription+API > > > > > >> > >> > > > > > >> > > > > > > > >> > > > > > > > > > >
