You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@fluss.apache.org by Hongshun Wang <lo...@gmail.com> on 2026/05/08 08:54:11 UTC
[DISCUSS] FIP-43: Fluss Multi-Table Subscription API
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
Re: [DISCUSS] FIP-43: Fluss Multi-Table Subscription API
Posted by Hongshun Wang <lo...@gmail.com>.
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 <bo...@gmail.com> 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 <lo...@gmail.com>:
> >
> > 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 <im...@gmail.com> 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 <bo...@gmail.com>
> 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 <lo...@gmail.com>:
> > > > >
> > > > > 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 <
> > > loserwang1024@gmail.com>
> > > > > 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 <im...@gmail.com>
> 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 <
> > > loserwang1024@gmail.com>
> > > > > >> 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 <
> > > loserwang1024@gmail.com>
> > > > > >> > 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 <im...@gmail.com>
> > > 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 <
> > > loserwang1024@gmail.com>
> > > > > >> > >> 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
> > > > > >> > >>
> > > > > >> > >
> > > > > >>
> > > > > >
> > >
>
Re: [DISCUSS] FIP-43: Fluss Multi-Table Subscription API
Posted by Anton Borisov <bo...@gmail.com>.
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 <lo...@gmail.com>:
>
> 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 <im...@gmail.com> 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 <bo...@gmail.com> 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 <lo...@gmail.com>:
> > > >
> > > > 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 <
> > loserwang1024@gmail.com>
> > > > 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 <im...@gmail.com> 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 <
> > loserwang1024@gmail.com>
> > > > >> 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 <
> > loserwang1024@gmail.com>
> > > > >> > 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 <im...@gmail.com>
> > 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 <
> > loserwang1024@gmail.com>
> > > > >> > >> 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
> > > > >> > >>
> > > > >> > >
> > > > >>
> > > > >
> >
Re: [DISCUSS] FIP-43: Fluss Multi-Table Subscription API
Posted by Hongshun Wang <lo...@gmail.com>.
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 <im...@gmail.com> 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 <bo...@gmail.com> 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 <lo...@gmail.com>:
> > >
> > > 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 <
> loserwang1024@gmail.com>
> > > 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 <im...@gmail.com> 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 <
> loserwang1024@gmail.com>
> > > >> 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 <
> loserwang1024@gmail.com>
> > > >> > 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 <im...@gmail.com>
> 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 <
> loserwang1024@gmail.com>
> > > >> > >> 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
> > > >> > >>
> > > >> > >
> > > >>
> > > >
>
Re: [DISCUSS] FIP-43: Fluss Multi-Table Subscription API
Posted by Jark Wu <im...@gmail.com>.
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 <bo...@gmail.com> 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 <lo...@gmail.com>:
> >
> > 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 <lo...@gmail.com>
> > 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 <im...@gmail.com> 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 <lo...@gmail.com>
> > >> 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 <lo...@gmail.com>
> > >> > 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 <im...@gmail.com> 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 <lo...@gmail.com>
> > >> > >> 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
> > >> > >>
> > >> > >
> > >>
> > >
Re: [DISCUSS] FIP-43: Fluss Multi-Table Subscription API
Posted by Anton Borisov <bo...@gmail.com>.
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 <lo...@gmail.com>:
>
> 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 <lo...@gmail.com>
> 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 <im...@gmail.com> 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 <lo...@gmail.com>
> >> 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 <lo...@gmail.com>
> >> > 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 <im...@gmail.com> 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 <lo...@gmail.com>
> >> > >> 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
> >> > >>
> >> > >
> >>
> >
Re: [DISCUSS] FIP-43: Fluss Multi-Table Subscription API
Posted by Hongshun Wang <lo...@gmail.com>.
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 <lo...@gmail.com>
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 <im...@gmail.com> 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 <lo...@gmail.com>
>> 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 <lo...@gmail.com>
>> > 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 <im...@gmail.com> 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 <lo...@gmail.com>
>> > >> 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
>> > >>
>> > >
>>
>
Re: [DISCUSS] FIP-43: Fluss Multi-Table Subscription API
Posted by Hongshun Wang <lo...@gmail.com>.
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 <im...@gmail.com> 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 <lo...@gmail.com>
> 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 <lo...@gmail.com>
> > 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 <im...@gmail.com> 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 <lo...@gmail.com>
> > >> 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
> > >>
> > >
>
Re: [DISCUSS] FIP-43: Fluss Multi-Table Subscription API
Posted by Jark Wu <im...@gmail.com>.
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 <lo...@gmail.com> 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 <lo...@gmail.com>
> 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 <im...@gmail.com> 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 <lo...@gmail.com>
> >> 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
> >>
> >
Re: [DISCUSS] FIP-43: Fluss Multi-Table Subscription API
Posted by Hongshun Wang <lo...@gmail.com>.
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 <lo...@gmail.com>
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 <im...@gmail.com> 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 <lo...@gmail.com>
>> 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
>>
>
Re: [DISCUSS] FIP-43: Fluss Multi-Table Subscription API
Posted by Hongshun Wang <lo...@gmail.com>.
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 <im...@gmail.com> 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 <lo...@gmail.com>
> 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
>
Re: [DISCUSS] FIP-43: Fluss Multi-Table Subscription API
Posted by Jark Wu <im...@gmail.com>.
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 <lo...@gmail.com> 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