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 > > > > >> > >> > > > > >> > > > > > > >> > > > > > > >
