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

fluss-client/src/main/java/org/apache/fluss/client/Connection.java

@@ -19,6 +19,7 @@
 
 import org.apache.fluss.annotation.PublicEvolving;
 import org.apache.fluss.client.admin.Admin;
+import org.apache.fluss.client.table.MultiTable;
 import org.apache.fluss.client.table.Table;
 import org.apache.fluss.config.Configuration;
 import org.apache.fluss.metadata.TablePath;
@@ -56,6 +57,16 @@ public interface Connection extends AutoCloseable {
     /** Retrieve a new Table client to operate data in table. */
     Table getTable(TablePath tablePath);
 
+    /**
+     * Retrieve a {@link MultiTable} client to scan and write data across multiple tables in a
+     * single unified API. Unlike {@link #getTable(TablePath)} which is bound to one table, {@code
+     * MultiTable} is table-agnostic: tables are identified per scan subscription and per write
+     * record.
+     *
+     * <p>{@link MultiTable} instances are light-weight and NOT thread-safe; obtain per-thread.
+     */
+    MultiTable getMultiTable();
+
     /** Close the connection and release all resources. */
     @Override
     void close() throws Exception;

fluss-client/src/main/java/org/apache/fluss/client/FlussConnection.java

@@ -22,6 +22,8 @@
 import org.apache.fluss.client.lookup.LookupClient;
 import org.apache.fluss.client.metadata.MetadataUpdater;
 import org.apache.fluss.client.table.FlussTable;
+import org.apache.fluss.client.table.MultiTable;
+import org.apache.fluss.client.table.MultiTableImpl;
 import org.apache.fluss.client.table.Table;
 import org.apache.fluss.client.table.scanner.RemoteFileDownloader;
 import org.apache.fluss.client.token.DefaultSecurityTokenManager;
@@ -108,6 +110,11 @@ public Table getTable(TablePath tablePath) {
         return new FlussTable(this, tablePath, admin.getTableInfo(tablePath).join());
     }
 
+    @Override
+    public MultiTable getMultiTable() {
+        return new MultiTableImpl(this);
+    }
+
     public MetadataUpdater getMetadataUpdater() {
         return metadataUpdater;
     }

fluss-client/src/main/java/org/apache/fluss/client/metrics/MultiTableWriterMetricGroup.java

@@ -0,0 +1,45 @@
+/*
+ * 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.metrics;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.client.write.WriterClient;
+import org.apache.fluss.rpc.metrics.ClientMetricGroup;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * Metrics for {@link WriterClient}s owned by multi-table writers.
+ *
+ * <p>Unlike the connection-level shared writer client which keeps the {@code client.writer} metric
+ * identity, each multi-table writer owns a dedicated {@link WriterClient}, so multiple instances
+ * may coexist under the same {@link ClientMetricGroup}. This group therefore uses the dedicated
+ * {@code multi_table_writer} scope plus a unique {@code writer_instance_id} variable so that
+ * reporters derive distinct metric identities (e.g. JMX ObjectNames, Prometheus label sets) instead
+ * of colliding.
+ */
+@Internal
+public class MultiTableWriterMetricGroup extends WriterMetricGroup {
+
+    private static final String NAME = "multi_table_writer";
+    private static final AtomicLong NEXT_WRITER_INSTANCE_ID = new AtomicLong();
+
+    public MultiTableWriterMetricGroup(ClientMetricGroup parent) {
+        super(parent, NAME, String.valueOf(NEXT_WRITER_INSTANCE_ID.incrementAndGet()));
+    }
+}

fluss-client/src/main/java/org/apache/fluss/client/metrics/WriterMetricGroup.java

@@ -29,14 +29,30 @@
 import org.apache.fluss.metrics.groups.AbstractMetricGroup;
 import org.apache.fluss.rpc.metrics.ClientMetricGroup;
 
+import javax.annotation.Nullable;
+
+import java.util.Map;
+
 import static org.apache.fluss.metrics.utils.MetricGroupUtils.makeScope;
 
-/** Metrics for {@link WriterClient}. */
+/**
+ * Metrics for {@link WriterClient}.
+ *
+ * <p>The connection-level shared writer client uses the {@code client.writer} scope without any
+ * extra variables. Writer clients that may have multiple instances under the same {@link
+ * ClientMetricGroup} should use a subclass with a dedicated scope and a disambiguating {@code
+ * writer_instance_id} variable, see {@link MultiTableWriterMetricGroup}.
+ */
 @Internal
 public class WriterMetricGroup extends AbstractMetricGroup {
     private static final String name = "writer";
     private static final int WINDOW_SIZE = 1024;
 
+    private final String groupName;
+
+    /** Only set for subclasses whose instances may coexist under the same parent. */
+    @Nullable private final String writerInstanceId;
+
     private final Counter recordsRetryTotal;
     private final Counter recordsSendTotal;
     private final Counter bytesSendTotal;
@@ -47,7 +63,16 @@ public class WriterMetricGroup extends AbstractMetricGroup {
     private volatile long batchQueueTimeMs = -1;
 
     public WriterMetricGroup(ClientMetricGroup parent) {
-        super(parent.getMetricRegistry(), makeScope(parent, name), parent);
+        this(parent, name, null);
+    }
+
+    protected WriterMetricGroup(
+            ClientMetricGroup parent, String groupName, @Nullable String writerInstanceId) {
+        super(parent.getMetricRegistry(), makeScope(parent, groupName), parent);
+        this.groupName = groupName;
+        // must be assigned before the first metric registration below, as reporters resolve
+        // getAllVariables() (which reads writerInstanceId) when a metric is added
+        this.writerInstanceId = writerInstanceId;
 
         gauge(MetricNames.WRITER_BATCH_QUEUE_TIME_MS, () -> batchQueueTimeMs);
 
@@ -99,6 +124,13 @@ public Histogram recordPerBatch() {
 
     @Override
     protected String getGroupName(CharacterFilter filter) {
-        return name;
+        return groupName;
+    }
+
+    @Override
+    protected final void putVariables(Map<String, String> variables) {
+        if (writerInstanceId != null) {
+            variables.put("writer_instance_id", writerInstanceId);
+        }
     }
 }

fluss-client/src/main/java/org/apache/fluss/client/table/MultiTable.java

@@ -0,0 +1,48 @@
+/*
+ * 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;
+
+import org.apache.fluss.annotation.PublicEvolving;
+import org.apache.fluss.client.Connection;
+import org.apache.fluss.client.table.scanner.MultiTableScan;
+import org.apache.fluss.client.table.writer.MultiTableWrite;
+import org.apache.fluss.metadata.TablePath;
+
+/**
+ * Used to communicate with multiple Fluss tables through a single client. Obtain an instance from
+ * {@link Connection#getMultiTable()}.
+ *
+ * <p>{@code MultiTable} is the cross-table counterpart of {@link Table}. Useful for CDC ingestion,
+ * cross-table streaming pipelines, and multi-table catalog sinks.
+ *
+ * <p>Unlike {@link Connection#getTable(TablePath)} which is bound to one table, {@code MultiTable}
+ * is table-agnostic: tables are identified per scan subscription and per write record.
+ *
+ * <p>{@code MultiTable} instances are light-weight and NOT thread-safe; obtain per-thread.
+ *
+ * @since 0.7
+ */
+@PublicEvolving
+public interface MultiTable {
+
+    /** Build a scanner that reads data from multiple tables simultaneously. */
+    MultiTableScan newMultiTableScan();
+
+    /** Build a writer that writes data (with change types) to multiple tables. */
+    MultiTableWrite newMultiTableWrite();
+}

fluss-client/src/main/java/org/apache/fluss/client/table/MultiTableImpl.java

@@ -0,0 +1,56 @@
+/*
+ * 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;
+
+import org.apache.fluss.annotation.Internal;
+import org.apache.fluss.client.FlussConnection;
+import org.apache.fluss.client.table.scanner.MultiTableScan;
+import org.apache.fluss.client.table.scanner.MultiTableScanImpl;
+import org.apache.fluss.client.table.writer.MultiTableWrite;
+import org.apache.fluss.client.table.writer.MultiTableWriteImpl;
+
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
+/**
+ * Default implementation of {@link MultiTable}.
+ *
+ * <p>{@code MultiTableImpl} is light-weight and NOT thread-safe; the underlying {@link
+ * FlussConnection} is shared and thread-safe, but each {@code MultiTableImpl} instance should be
+ * obtained per-thread.
+ *
+ * @since 0.7
+ */
+@Internal
+public class MultiTableImpl implements MultiTable {
+
+    private final FlussConnection connection;
+
+    public MultiTableImpl(FlussConnection connection) {
+        this.connection = checkNotNull(connection, "connection");
+    }
+
+    @Override
+    public MultiTableScan newMultiTableScan() {
+        return new MultiTableScanImpl(connection);
+    }
+
+    @Override
+    public MultiTableWrite newMultiTableWrite() {
+        return new MultiTableWriteImpl(connection);
+    }
+}

fluss-client/src/main/java/org/apache/fluss/client/table/scanner/MultiTableBatchScanner.java

@@ -0,0 +1,31 @@
+/*
+ * 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;
+
+import org.apache.fluss.annotation.PublicEvolving;
+
+/**
+ * Reads current data from multiple tables through a single client.
+ *
+ * <p>NOTE: planned for a future release; declared as a placeholder for forward compatibility. There
+ * is no concrete implementation today.
+ *
+ * @since 0.7
+ */
+@PublicEvolving
+public interface MultiTableBatchScanner extends AutoCloseable {}