fxbing commented on code in PR #3140: URL: https://github.com/apache/fluss/pull/3140#discussion_r3526724973
########## fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/AbstractLogScanner.java: ########## @@ -0,0 +1,207 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.client.table.scanner.log; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.client.metrics.ScannerMetricGroup; +import org.apache.fluss.exception.WakeupException; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.io.IOException; +import java.time.Duration; +import java.util.ConcurrentModificationException; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; + +/** + * Base class that centralizes the shared poll loop, lightweight single-thread guard, wakeup and + * close lifecycle for log scanners. Concrete scanners supply the result type {@code R} and an + * enrichment step from raw {@link ScanRecords} to that type. + * + * <p>This class is NOT thread-safe. The lightweight guard only detects accidental concurrent use + * and throws {@link ConcurrentModificationException}; it does not serialize callers. + */ +@Internal +abstract class AbstractLogScanner<R> implements AutoCloseable { + + private static final Logger LOG = LoggerFactory.getLogger(AbstractLogScanner.class); + + private static final long NO_CURRENT_THREAD = -1L; + + protected final String name; + protected final LogScannerStatus logScannerStatus; + protected final LogFetcher logFetcher; + protected final ScannerMetricGroup scannerMetricGroup; + + private volatile boolean closed = false; + + // currentThread holds the threadId of the current thread accessing the scanner + // and is used to prevent multithreaded access. + private final AtomicLong currentThread = new AtomicLong(NO_CURRENT_THREAD); + // refCount is used to allow reentrant access by the thread that has acquired currentThread. + private final AtomicInteger refCount = new AtomicInteger(0); + + protected AbstractLogScanner( + String name, + LogScannerStatus logScannerStatus, + LogFetcher logFetcher, + ScannerMetricGroup scannerMetricGroup) { + this.name = name; + this.logScannerStatus = logScannerStatus; + this.logFetcher = logFetcher; + this.scannerMetricGroup = scannerMetricGroup; + } + + /** + * Template poll loop shared by all concrete scanners. + * + * <ul> + * <li>If {@link #pollForFetches()} returns a non-empty batch, {@link #toResult(ScanRecords)} + * is invoked to produce the concrete result type. + * <li>If the batch is empty, we wait on {@link LogFetcher#awaitNotEmpty(long)} until either + * data arrives, the timeout elapses, or {@link #wakeup()} is called. + * <li>Any empty/timeout/wakeup path returns {@link #emptyResult()}. + * </ul> + */ + public final R poll(Duration timeout) { + acquireAndEnsureOpen(); + try { + if (!logScannerStatus.prepareToPoll()) { + throw new IllegalStateException(notSubscribedMessage()); + } + + scannerMetricGroup.recordPollStart(System.currentTimeMillis()); + long timeoutNanos = timeout.toNanos(); + long startNanos = System.nanoTime(); + do { + ScanRecords scanRecords = pollForFetches(); + if (scanRecords.isEmpty()) { Review Comment: Should progress-only results be returned here? `ScanRecords` may contain buckets with empty record lists but advanced offsets; `isEmpty()` drops that distinction. ########## fluss-common/src/main/java/org/apache/fluss/record/LogRecordReadContext.java: ########## @@ -487,4 +502,45 @@ private static FieldGetter[] buildProjectedFieldGetters(RowType rowType, int[] s } return fieldGetters; } + + // ------------------------------------------------------------------------- + // ReadTarget + // ------------------------------------------------------------------------- + + /** + * Encapsulates the target output configuration for reading log records. Contains the target + * schema identity, data row type, selected field indices, and pre-built field getters. + * + * <p>When a {@code ReadTarget} is provided, field getters are fixed and schema evolution may be + * enabled. When absent ({@code null}), the reader dynamically adapts to each batch's schema so + * that columns added via schema evolution are visible. + */ + static class ReadTarget { + /** The schema id this configuration was built for. */ + final int schemaId; + + /** The row type of the data (may be server-projected for ARROW format). */ + final RowType dataRowType; + + /** The field indices to select from the data row type. */ + final int[] selectedFields; + + /** Pre-built field getters corresponding to {@link #selectedFields}. */ + final FieldGetter[] fieldGetters; + + /** Whether to project records from other schemas to this target schema. */ + final boolean readAsTargetSchema; Review Comment: Is `readAsTargetSchema` still needed? It is written but I could not find a read. ########## fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/MultiTableLogScannerImpl.java: ########## @@ -0,0 +1,356 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.client.table.scanner.log; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.client.admin.Admin; +import org.apache.fluss.client.metadata.ClientSchemaGetter; +import org.apache.fluss.client.metadata.MetadataUpdater; +import org.apache.fluss.client.metrics.ScannerMetricGroup; +import org.apache.fluss.client.table.scanner.MultiTableRecord; +import org.apache.fluss.client.table.scanner.RemoteFileDownloader; +import org.apache.fluss.client.table.scanner.ScanRecord; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.metadata.Schema; +import org.apache.fluss.metadata.SchemaGetter; +import org.apache.fluss.metadata.TableBucket; +import org.apache.fluss.metadata.TableInfo; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.record.LogRecordReadContext; +import org.apache.fluss.rpc.metrics.ClientMetricGroup; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Collections; +import java.util.ConcurrentModificationException; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.ExecutionException; + +/** + * Default implementation of {@link MultiTableLogScanner}. + * + * <p>This scanner subscribes to log data from multiple tables simultaneously. Tables are registered + * dynamically on the first {@code subscribe(...)} call to a given {@link TablePath}: the scanner + * resolves the {@link TableInfo} via the admin client and registers the table on the underlying + * {@link LogFetcher} with no projection / filter (phase 1 surface). + * + * <p>Internally, it delegates to a single shared {@link LogFetcher} that manages per-table read + * contexts. + * + * <p>{@code MultiTableLogScannerImpl} is NOT thread-safe. It is the responsibility of the user to + * ensure that multithreaded access is properly synchronized. Un-synchronized access will result in + * {@link ConcurrentModificationException}. + * + * @since 0.7 + */ +@Internal +public class MultiTableLogScannerImpl extends AbstractLogScanner<MultiTableRecords> + implements MultiTableLogScanner { + + private static final Logger LOG = LoggerFactory.getLogger(MultiTableLogScannerImpl.class); + + private final MetadataUpdater metadataUpdater; + private final Admin admin; + + /** Per-table {@link TableInfo}, populated lazily on first subscribe to the table. */ + private final Map<TablePath, TableInfo> registeredByPath = new HashMap<>(); + + /** tableId -> tablePath, for poll-time record enrichment. */ + private final Map<Long, TablePath> pathByTableId = new HashMap<>(); + + private final Map<Long, SchemaGetter> schemaGetters = new HashMap<>(); + + /** + * Currently subscribed buckets per table. When the last bucket of a table is unsubscribed, the + * table registration is evicted so that a subsequent {@code subscribe(...)} re-resolves a fresh + * {@link TableInfo}. This is what lets callers recover from a drop+recreate (tableId change) by + * unsubscribing all buckets of the stale table and subscribing again. + */ + private final Map<TablePath, Set<TableBucket>> subscribedBucketsByPath = new HashMap<>(); + + public MultiTableLogScannerImpl( + String scannerName, + MetadataUpdater metadataUpdater, + Configuration conf, + ClientMetricGroup clientMetricGroup, + RemoteFileDownloader remoteFileDownloader, + Admin admin) { + this( + scannerName, + metadataUpdater, + conf, + remoteFileDownloader, + admin, + new LogScannerStatus(), + // Use a synthetic TablePath for scanner-level metrics scoping. + new ScannerMetricGroup( + clientMetricGroup, TablePath.of("multi-table", scannerName))); + } + + private MultiTableLogScannerImpl( + String scannerName, + MetadataUpdater metadataUpdater, + Configuration conf, + RemoteFileDownloader remoteFileDownloader, + Admin admin, + LogScannerStatus logScannerStatus, + ScannerMetricGroup scannerMetricGroup) { + super( + scannerName, + logScannerStatus, + new LogFetcher( + scannerName, + logScannerStatus, + conf, + metadataUpdater, + scannerMetricGroup, + remoteFileDownloader, + LogRecordReadContext.SchemaResolution.DYNAMIC), + scannerMetricGroup); + this.metadataUpdater = metadataUpdater; + this.admin = admin; + } + + @Override + protected MultiTableRecords emptyResult() { + return MultiTableRecords.EMPTY; + } + + @Override + protected MultiTableRecords toResult(ScanRecords scanRecords) { + return enrich(scanRecords); + } + + @Override + protected String notSubscribedMessage() { + return "MultiTableLogScanner is not subscribed to any buckets."; + } + + @Override + public void subscribe(TablePath tablePath, int bucket, long offset) { + acquireAndEnsureOpen(); + try { + TableInfo tableInfo = registerIfAbsent(tablePath); + if (tableInfo.isPartitioned()) { + throw new IllegalStateException( + "Table " + + tablePath + + " is a partitioned table, please use " + + "\"subscribe(TablePath, long partitionId, int bucket, long offset)\" " + + "to subscribe a partitioned bucket instead."); + } + TableBucket tableBucket = new TableBucket(tableInfo.getTableId(), bucket); + metadataUpdater.checkAndUpdateTableMetadata(Collections.singleton(tablePath)); + logScannerStatus.assignScanBuckets(Collections.singletonMap(tableBucket, offset)); + trackSubscription(tablePath, tableBucket); + } finally { + release(); + } + } + + @Override + public void unsubscribe(TablePath tablePath, int bucket) { + acquireAndEnsureOpen(); + try { + TableInfo tableInfo = requireRegistered(tablePath); + if (tableInfo.isPartitioned()) { + throw new IllegalStateException( + "Table " + + tablePath + + " is a partitioned table, please use " + + "\"unsubscribe(TablePath, long partitionId, int bucket)\" " + + "to unsubscribe a partitioned bucket instead."); + } + TableBucket tableBucket = new TableBucket(tableInfo.getTableId(), bucket); + logScannerStatus.unassignScanBuckets(Collections.singletonList(tableBucket)); + untrackSubscription(tablePath, tableBucket); + } finally { + release(); + } + } + + @Override + public void subscribe(TablePath tablePath, long partitionId, int bucket, long offset) { + acquireAndEnsureOpen(); + try { + TableInfo tableInfo = registerIfAbsent(tablePath); + if (!tableInfo.isPartitioned()) { + throw new IllegalStateException( + "Table " + + tablePath + + " is not a partitioned table, please use " + + "\"subscribe(TablePath, int bucket, long offset)\" " + + "to subscribe a non-partitioned bucket instead."); + } + TableBucket tableBucket = new TableBucket(tableInfo.getTableId(), partitionId, bucket); + metadataUpdater.checkAndUpdateTableMetadata(Collections.singleton(tablePath)); + metadataUpdater.checkAndUpdatePartitionMetadata( + tablePath, Collections.singleton(partitionId)); + logScannerStatus.assignScanBuckets(Collections.singletonMap(tableBucket, offset)); + trackSubscription(tablePath, tableBucket); + } finally { + release(); + } + } + + @Override + public void unsubscribe(TablePath tablePath, long partitionId, int bucket) { + acquireAndEnsureOpen(); + try { + TableInfo tableInfo = requireRegistered(tablePath); + if (!tableInfo.isPartitioned()) { + throw new IllegalStateException( + "Table " + tablePath + " is not a partitioned table."); + } + TableBucket tableBucket = new TableBucket(tableInfo.getTableId(), partitionId, bucket); + logScannerStatus.unassignScanBuckets(Collections.singletonList(tableBucket)); + untrackSubscription(tablePath, tableBucket); + } finally { + release(); + } + } + + /** + * Register the table on the underlying {@link LogFetcher} if not yet registered. Must be called + * under the acquire-lock. Phase 1 always registers with no projection / filter. + */ + private TableInfo registerIfAbsent(TablePath tablePath) { + TableInfo cached = registeredByPath.get(tablePath); + if (cached != null) { + checkTableIdNotChanged(tablePath, cached); + return cached; + } + + TableInfo tableInfo = resolveTableInfo(tablePath); + TableScanSpec spec = new TableScanSpec(tableInfo, null, null); + SchemaGetter schemaGetter = + new ClientSchemaGetter(tablePath, tableInfo.getSchemaInfo(), admin); + logFetcher.registerTable(spec, schemaGetter); + registeredByPath.put(tablePath, tableInfo); + pathByTableId.put(tableInfo.getTableId(), tablePath); + schemaGetters.put(tableInfo.getTableId(), schemaGetter); + return tableInfo; + } + + /** + * Record a freshly assigned bucket against its table. Must be called under the acquire-lock. + */ + private void trackSubscription(TablePath tablePath, TableBucket tableBucket) { + subscribedBucketsByPath.computeIfAbsent(tablePath, k -> new HashSet<>()).add(tableBucket); + } + + /** + * Drop a bucket from its table's subscription set. When the table has no remaining subscribed + * buckets, evict its cached registration so a later {@code subscribe(...)} re-resolves the + * table (picking up a new tableId after a drop+recreate). Must be called under the + * acquire-lock. + */ + private void untrackSubscription(TablePath tablePath, TableBucket tableBucket) { + Set<TableBucket> buckets = subscribedBucketsByPath.get(tablePath); + if (buckets == null) { + return; + } + buckets.remove(tableBucket); + if (buckets.isEmpty()) { Review Comment: Should this also unregister/close the corresponding `LogFetcher` table context when the last bucket is removed? ########## fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/MultiTableLogScannerImpl.java: ########## @@ -0,0 +1,356 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.client.table.scanner.log; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.client.admin.Admin; +import org.apache.fluss.client.metadata.ClientSchemaGetter; +import org.apache.fluss.client.metadata.MetadataUpdater; +import org.apache.fluss.client.metrics.ScannerMetricGroup; +import org.apache.fluss.client.table.scanner.MultiTableRecord; +import org.apache.fluss.client.table.scanner.RemoteFileDownloader; +import org.apache.fluss.client.table.scanner.ScanRecord; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.metadata.Schema; +import org.apache.fluss.metadata.SchemaGetter; +import org.apache.fluss.metadata.TableBucket; +import org.apache.fluss.metadata.TableInfo; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.record.LogRecordReadContext; +import org.apache.fluss.rpc.metrics.ClientMetricGroup; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Collections; +import java.util.ConcurrentModificationException; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.ExecutionException; + +/** + * Default implementation of {@link MultiTableLogScanner}. + * + * <p>This scanner subscribes to log data from multiple tables simultaneously. Tables are registered + * dynamically on the first {@code subscribe(...)} call to a given {@link TablePath}: the scanner + * resolves the {@link TableInfo} via the admin client and registers the table on the underlying + * {@link LogFetcher} with no projection / filter (phase 1 surface). + * + * <p>Internally, it delegates to a single shared {@link LogFetcher} that manages per-table read + * contexts. + * + * <p>{@code MultiTableLogScannerImpl} is NOT thread-safe. It is the responsibility of the user to + * ensure that multithreaded access is properly synchronized. Un-synchronized access will result in + * {@link ConcurrentModificationException}. + * + * @since 0.7 + */ +@Internal +public class MultiTableLogScannerImpl extends AbstractLogScanner<MultiTableRecords> + implements MultiTableLogScanner { + + private static final Logger LOG = LoggerFactory.getLogger(MultiTableLogScannerImpl.class); + + private final MetadataUpdater metadataUpdater; + private final Admin admin; + + /** Per-table {@link TableInfo}, populated lazily on first subscribe to the table. */ + private final Map<TablePath, TableInfo> registeredByPath = new HashMap<>(); + + /** tableId -> tablePath, for poll-time record enrichment. */ + private final Map<Long, TablePath> pathByTableId = new HashMap<>(); + + private final Map<Long, SchemaGetter> schemaGetters = new HashMap<>(); + + /** + * Currently subscribed buckets per table. When the last bucket of a table is unsubscribed, the + * table registration is evicted so that a subsequent {@code subscribe(...)} re-resolves a fresh + * {@link TableInfo}. This is what lets callers recover from a drop+recreate (tableId change) by + * unsubscribing all buckets of the stale table and subscribing again. + */ + private final Map<TablePath, Set<TableBucket>> subscribedBucketsByPath = new HashMap<>(); + + public MultiTableLogScannerImpl( + String scannerName, + MetadataUpdater metadataUpdater, + Configuration conf, + ClientMetricGroup clientMetricGroup, + RemoteFileDownloader remoteFileDownloader, + Admin admin) { + this( + scannerName, + metadataUpdater, + conf, + remoteFileDownloader, + admin, + new LogScannerStatus(), + // Use a synthetic TablePath for scanner-level metrics scoping. + new ScannerMetricGroup( + clientMetricGroup, TablePath.of("multi-table", scannerName))); Review Comment: Should we add per-table metrics for different subscribed tables here? ########## fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/MultiTableLogScannerImpl.java: ########## @@ -0,0 +1,356 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.fluss.client.table.scanner.log; + +import org.apache.fluss.annotation.Internal; +import org.apache.fluss.client.admin.Admin; +import org.apache.fluss.client.metadata.ClientSchemaGetter; +import org.apache.fluss.client.metadata.MetadataUpdater; +import org.apache.fluss.client.metrics.ScannerMetricGroup; +import org.apache.fluss.client.table.scanner.MultiTableRecord; +import org.apache.fluss.client.table.scanner.RemoteFileDownloader; +import org.apache.fluss.client.table.scanner.ScanRecord; +import org.apache.fluss.config.Configuration; +import org.apache.fluss.metadata.Schema; +import org.apache.fluss.metadata.SchemaGetter; +import org.apache.fluss.metadata.TableBucket; +import org.apache.fluss.metadata.TableInfo; +import org.apache.fluss.metadata.TablePath; +import org.apache.fluss.record.LogRecordReadContext; +import org.apache.fluss.rpc.metrics.ClientMetricGroup; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.Collections; +import java.util.ConcurrentModificationException; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Optional; +import java.util.Set; +import java.util.concurrent.ExecutionException; + +/** + * Default implementation of {@link MultiTableLogScanner}. + * + * <p>This scanner subscribes to log data from multiple tables simultaneously. Tables are registered + * dynamically on the first {@code subscribe(...)} call to a given {@link TablePath}: the scanner + * resolves the {@link TableInfo} via the admin client and registers the table on the underlying + * {@link LogFetcher} with no projection / filter (phase 1 surface). + * + * <p>Internally, it delegates to a single shared {@link LogFetcher} that manages per-table read + * contexts. + * + * <p>{@code MultiTableLogScannerImpl} is NOT thread-safe. It is the responsibility of the user to + * ensure that multithreaded access is properly synchronized. Un-synchronized access will result in + * {@link ConcurrentModificationException}. + * + * @since 0.7 + */ +@Internal +public class MultiTableLogScannerImpl extends AbstractLogScanner<MultiTableRecords> + implements MultiTableLogScanner { + + private static final Logger LOG = LoggerFactory.getLogger(MultiTableLogScannerImpl.class); + + private final MetadataUpdater metadataUpdater; + private final Admin admin; + + /** Per-table {@link TableInfo}, populated lazily on first subscribe to the table. */ + private final Map<TablePath, TableInfo> registeredByPath = new HashMap<>(); + + /** tableId -> tablePath, for poll-time record enrichment. */ + private final Map<Long, TablePath> pathByTableId = new HashMap<>(); + + private final Map<Long, SchemaGetter> schemaGetters = new HashMap<>(); + + /** + * Currently subscribed buckets per table. When the last bucket of a table is unsubscribed, the + * table registration is evicted so that a subsequent {@code subscribe(...)} re-resolves a fresh + * {@link TableInfo}. This is what lets callers recover from a drop+recreate (tableId change) by + * unsubscribing all buckets of the stale table and subscribing again. + */ + private final Map<TablePath, Set<TableBucket>> subscribedBucketsByPath = new HashMap<>(); + + public MultiTableLogScannerImpl( + String scannerName, + MetadataUpdater metadataUpdater, + Configuration conf, + ClientMetricGroup clientMetricGroup, + RemoteFileDownloader remoteFileDownloader, + Admin admin) { + this( + scannerName, + metadataUpdater, + conf, + remoteFileDownloader, + admin, + new LogScannerStatus(), + // Use a synthetic TablePath for scanner-level metrics scoping. + new ScannerMetricGroup( + clientMetricGroup, TablePath.of("multi-table", scannerName))); + } + + private MultiTableLogScannerImpl( + String scannerName, + MetadataUpdater metadataUpdater, + Configuration conf, + RemoteFileDownloader remoteFileDownloader, + Admin admin, + LogScannerStatus logScannerStatus, + ScannerMetricGroup scannerMetricGroup) { + super( + scannerName, + logScannerStatus, + new LogFetcher( + scannerName, + logScannerStatus, + conf, + metadataUpdater, + scannerMetricGroup, + remoteFileDownloader, + LogRecordReadContext.SchemaResolution.DYNAMIC), + scannerMetricGroup); + this.metadataUpdater = metadataUpdater; + this.admin = admin; + } + + @Override + protected MultiTableRecords emptyResult() { + return MultiTableRecords.EMPTY; + } + + @Override + protected MultiTableRecords toResult(ScanRecords scanRecords) { + return enrich(scanRecords); + } + + @Override + protected String notSubscribedMessage() { + return "MultiTableLogScanner is not subscribed to any buckets."; + } + + @Override + public void subscribe(TablePath tablePath, int bucket, long offset) { + acquireAndEnsureOpen(); + try { + TableInfo tableInfo = registerIfAbsent(tablePath); Review Comment: Should the partitioned-table check happen before `registerIfAbsent()`? A failed wrong-overload subscribe seems to leave registration state behind. ########## fluss-common/src/main/java/org/apache/fluss/record/LogRecordBatch.java: ########## @@ -186,6 +186,9 @@ default ArrowBatchData loadArrowBatch(ReadContext context) { /** The read context of a {@link LogRecordBatch} to read records. */ interface ReadContext { + /** Gets the table id, or -1 if not available. */ + long getTableId(); Review Comment: Is `ReadContext#getTableId()` currently unused? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
