[FIP-43][client] Support log scanner and writer for multiple tables

[loserwang1024]

Purpose

Linked issue: close #3139

Brief change log

Tests

API and Format

Documentation

[wuchong]

Thanks @loserwang1024. Supporting multi-table reads is a valuable enhancement. Currently, we can already read from multiple tables by creating separate LogScanner instances via connection.getTable(tablePath).newScan(..).createLogScanner(..).

I understand that the primary goal of this PR is to consolidate I/O within the LogFetcher, and maintaining individual LogScanners per table is not a problem. This optimization mirrors our approach on the writer side, where writes for multiple tables are merged into a unified sender instance to aggregate I/O. We can apply a similar strategy here by enabling multi-table support in the LogFetcher and sharing it at the Connection level.

This approach allows us to keep the user-facing API unchanged, leveraging the existing hierarchy of Connection -> Table -> LogScanner without introducing a new multi-table scanner abstraction. It also ensures consistency between reader and writer operations for both single-table and multi-table scenarios.

What do you think?

[loserwang1024]

leveraging the existing hierarchy of Connection -> Table -> LogScanner without introducing a new multi-table scanner abstraction. It also ensures consistency between reader and writer operations for both single-table and multi-table scenarios.

this approach introduces two critical issues:

• Blocking Latency vs. Asynchronous Writes: A single reader would need to maintain multiple LogScanner instances and poll them sequentially (e.g., calling scanner.poll(POLL_TIMEOUT) in a loop). If the tables at the beginning of the queue have low data volume, the poll operation will block for the entire timeout period, causing significant latency before subsequent tables are processed.
  Note: The reason this pattern works on the writer side is that writes are asynchronous; one write operation does not block others. In contrast, the synchronous nature of reading makes sequential polling inefficient.

• Memory Overhead: Requiring each reader to manage multiple LogScanner instances creates substantial resource overhead. Each LogScanner must independently maintain its own memory buffers for parsing Arrow records. Consequently, the aggregate memory footprint becomes significantly larger than necessary compared to a unified approach.

[fxbing]

Thanks for the work on this. I left a few small questions while reading through the scanner/fetcher paths; most are about expected semantics and cleanup boundaries.

[fxbing]

Should we add per-table metrics for different subscribed tables here?

[loserwang1024]

I want to it in another PR to support bucket level's metric, thus MutilTable and Single table can share same metrics.

[loserwang1024]

@leonardBang , would you like to help review this?

[leonardBang]

I did another focused review of the production paths, with particular attention to schema transitions, shared client state, asynchronous fetch lifecycles, and off-heap resource ownership. I left several inline questions describing the current behavior, the scenarios in which it may surface, and possible regression coverage. It would be helpful to clarify or address these cases before merging.

[leonardBang]

Would it make sense to enforce the schema boundary in the shared accumulator rather than relying on this flush?

Each MultiTableWriterImpl keeps its own schema cache, while writers created from the same connection share the same WriterClient and RecordAccumulator. Since WriterClient.flush() allows other threads to continue sending, one writer may still have a schema-1 log batch buffered when another writer first resolves schema 2 and appends to the same table/bucket. The KV batch checks schemaId, but the log batches currently do not, so the resulting batch may carry the old schema id while containing a row encoded with the new schema.

Could RecordAccumulator or WriteBatch.tryAppend() roll over when the table id, schema id, or write format changes? It may also be helpful to add a test with two writers sharing one connection, leaving the old-schema write outstanding before sending a new-schema record from the second writer.

[leonardBang]

Could we avoid taking the connection-wide flush and metadata-refresh path for every historical-schema record?

latestSchemaIdByPath continues to point to the table's latest schema after the historical state has been cached. Therefore, consecutive records carrying the same historical schema id still enter this branch repeatedly, invoking writerClient.flush() and forcing a metadata update for each row. This may prevent historical CDC records from batching and may also flush unrelated writers that share the connection.

Would it make sense to track the active write schema separately, or let the shared accumulator roll the batch when the schema changes? A test that sends several historical-schema records without waiting for each future could help verify that the refresh/flush is not repeated per row.

[leonardBang]

Could we clarify whether the original ChangeType is expected to be preserved for log tables?

INSERT is accepted here, but the generated WriteRecord does not carry the change type, and the row/Arrow log batches later append it as APPEND_ONLY. A scan-then-write pipeline may therefore observe a different changelog type from the input. UPDATE_BEFORE is also returned as a successful no-op before the target table kind or existence is resolved.

If log tables are intended to preserve changelog semantics, could the change type be carried through to the batch? Otherwise, would it be safer to narrow the API and reject unsupported values explicitly? A round-trip test asserting the scanned ChangeType, rather than only row contents, may help document the intended behavior.

[leonardBang]

Would it make sense to include the RPC callback threads in the synchronization/lifecycle contract here?

The scanner entry points are guarded, but gateway.fetchLog(...).whenComplete(...) reads tableReadContexts asynchronously while registerTable() and unregisterTable() mutate this plain HashMap. unregisterTable() may also close a context that is still referenced by an in-flight response, a buffered CompletedFetch, or a remote pending fetch. In addition, falling back to another table's read context when the original context is missing could decode the response with an unrelated schema.

Could the request capture a context snapshot or generation and defer closing it until buffered/in-flight references are released? A stale response with no matching context could then be discarded instead of using another table's context.

[leonardBang]

Could these read contexts reuse the ChunkedFactory owned by LogFetcher?

The five-argument createReadContext() overload creates a new factory internally, so the factory closed by LogFetcher.close() is not the one used by either the local or remote Arrow context. Closing the allocator releases its buffers, but an inaccessible factory may continue retaining active or free native chunks for reuse.

Would it make sense to pass chunkedFactory into both calls, as the previous single-table implementation did, or make each context explicitly own and close the factory it creates?

[leonardBang]

Could the progress entry also preserve the corresponding table and bucket in the result?

This call records the consumed offset, but the builder creates the tablePath -> bucket entry only when at least one materialized record is added below. A progress-only poll can therefore return hasProgress() == true and a non-null consumed offset while tablePaths() and buckets(tablePath) remain empty, which differs from their documented behavior.

Would it make sense for the progress builder method to receive tablePath and create an empty bucket entry? The IT helper could then require the expected table and bucket instead of falling back to an assertion on hasProgress() after the timeout.

[leonardBang]

Would it make sense to make this registration transactional with the remaining subscribe steps?

registerIfAbsent() has already populated the scanner maps and created the fetcher's read contexts, while the metadata/partition checks and bucket assignment below may still fail. If that happens, no bucket is added to subscribedBucketsByPath; a later unsubscribe() reaches untrackSubscription() with no bucket set and cannot remove the stale registration or close its contexts.

Could the metadata checks run before registration where possible, or could the exception path roll back the maps and fetcher context? A subscription using a nonexistent partition may be a useful regression test.

[leonardBang]

Could we keep the existing org.apache.fluss.client.table.scanner.log.LogScan class and add MultiTableScan as a separate file?

Git represents this change as a rename, but the two classes serve different purposes. Since LogScan has been a @PublicEvolving API since 0.1 and is still present on main, removing it may break existing source users and cause NoClassDefFoundError for binaries compiled against it.

Was the removal intentional? If not, retaining LogScan.java unchanged while adding this new interface separately would preserve compatibility.

[loserwang1024]

Yes, I remove it because no place has used it.

[leonardBang]

Could we keep org.apache.fluss.client.table.scanner.log.LogScan and add MultiTableScan as a separate class? Although LogScan is not currently referenced inside this repository, it has been a @PublicEvolving API since 0.1 with public constructors and projection methods. Removing its FQCN may cause existing clients to fail at compile time or with NoClassDefFoundError. A deprecation cycle would be safer if the old API is intended to be retired.

[leonardBang]

Would it make sense to treat a schema ID mismatch as a batch-compatibility boundary and return false here, similar to table ID and write format mismatches? Currently, writing a valid new-schema record while an old-schema batch is buffered throws synchronously, so CDC/schema-evolution callers must explicitly flush() at every schema boundary. Returning false would allow RecordAccumulator to close the old batch and create a schema-specific new batch automatically. It may also be helpful to change testSchemaSwitchWithoutFlushFails to verify that both records succeed without an explicit intermediate flush.

[leonardBang]

Could we keep org.apache.fluss.client.table.scanner.log.LogScan and add MultiTableScan as a separate class? Although LogScan is not currently referenced inside this repository, it has been a @PublicEvolving API since 0.1 with public constructors and projection methods. Removing its FQCN may cause existing clients to fail at compile time or with NoClassDefFoundError. A deprecation cycle would be safer if the old API is intended to be retired.

[leonardBang]

Could we assign this.writerMetricGroup before calling buildIdempotenceManager()? The latter may fail during configuration validation, in which case the field remains null. The catch block then calls close(), which unconditionally invokes writerMetricGroup.close() and introduces a NullPointerException that masks the original exception.

[leonardBang]

Thanks @loserwang1024 for the update, the PR looks good to me now, I only left one minor comment for potential NPE. Feel free to merge it once address it.