[client] Support log scanner and writer for multiple tables

Author: loserwang1024

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;
@@ -106,6 +108,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/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 {}

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

@@ -0,0 +1,117 @@
+/*
+ * 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;
+import org.apache.fluss.client.table.scanner.log.MultiTableLogScanner;
+import org.apache.fluss.client.table.writer.MultiTableWriter;
+import org.apache.fluss.metadata.Schema;
+import org.apache.fluss.metadata.TablePath;
+import org.apache.fluss.record.ChangeType;
+import org.apache.fluss.row.InternalRow;
+
+import static org.apache.fluss.utils.Preconditions.checkNotNull;
+
+/**
+ * A record produced by a {@link MultiTableLogScanner}. Wraps an unchanged {@link ScanRecord} and
+ * enriches it with the source table identity (path, id) and schema (schema object, schema id).
+ *
+ * <p>The underlying {@link ScanRecord} structure is NOT modified.
+ *
+ * <p>Read-side only. For writes via {@link MultiTableWriter}, use {@link
+ * org.apache.fluss.client.table.writer.MultiTableWriteRecord} which carries only the write-relevant
+ * fields.
+ *
+ * @since 0.7
+ */
+@PublicEvolving
+public final class MultiTableRecord {
+
+    private final TablePath tablePath;
+    private final long tableId;
+    private final Schema schema;
+    private final int schemaId;
+    private final ScanRecord scanRecord;
+
+    public MultiTableRecord(
+            TablePath tablePath, long tableId, int schemaId, Schema schema, ScanRecord scanRecord) {
+        this.tablePath = checkNotNull(tablePath, "tablePath");
+        this.tableId = tableId;
+        this.schema = checkNotNull(schema, "schema");
+        this.schemaId = schemaId;
+        this.scanRecord = checkNotNull(scanRecord, "scanRecord");
+    }
+
+    // ---- table identity ----
+
+    public TablePath getTablePath() {
+        return tablePath;
+    }
+
+    public long getTableId() {
+        return tableId;
+    }
+
+    // ---- schema info ----
+
+    public Schema getSchema() {
+        return schema;
+    }
+
+    public int getSchemaId() {
+        return schemaId;
+    }
+
+    // ---- underlying record (unchanged) ----
+
+    public ScanRecord getScanRecord() {
+        return scanRecord;
+    }
+
+    // ---- convenience delegates ----
+
+    public long logOffset() {
+        return scanRecord.logOffset();
+    }
+
+    public long timestamp() {
+        return scanRecord.timestamp();
+    }
+
+    public ChangeType getChangeType() {
+        return scanRecord.getChangeType();
+    }
+
+    public InternalRow getRow() {
+        return scanRecord.getRow();
+    }
+
+    @Override
+    public String toString() {
+        return "MultiTableRecord{"
+                + "tablePath="
+                + tablePath
+                + ", tableId="
+                + tableId
+                + ", schemaId="
+                + schemaId
+                + ", scanRecord="
+                + scanRecord
+                + '}';
+    }
+}

…ss/client/table/scanner/log/LogScan.java …client/table/scanner/MultiTableScan.javafluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogScan.java renamed to fluss-client/src/main/java/org/apache/fluss/client/table/scanner/MultiTableScan.java fluss-client/src/main/java/org/apache/fluss/client/table/scanner/log/LogScan.java renamed to fluss-client/src/main/java/org/apache/fluss/client/table/scanner/MultiTableScan.java

@@ -15,42 +15,27 @@
  * limitations under the License.
  */
 
-package org.apache.fluss.client.table.scanner.log;
+package org.apache.fluss.client.table.scanner;
 
 import org.apache.fluss.annotation.PublicEvolving;
-
-import javax.annotation.Nullable;
+import org.apache.fluss.client.table.scanner.log.MultiTableLogScanner;
 
 /**
- * Used to describe the operation to scan log data by {@link LogScanner} to a table.
+ * Builder for a multi-table scanner.
+ *
+ * <p>Phase 1 surface intentionally exposes no per-table builder configuration: tables are
+ * registered dynamically on the created {@link MultiTableLogScanner} when {@code subscribe(...)} is
+ * first called for a {@code TablePath}. Future phases may add {@code withTable(...)} overloads to
+ * pre-declare tables together with projection / filter pushdown.
  *
- * @since 0.1
+ * @since 0.7
  */
 @PublicEvolving
-public class LogScan {
-
-    /** The projected fields to do projection. No projection if is null. */
-    @Nullable private final int[] projectedFields;
-
-    public LogScan() {
-        this(null);
-    }
-
-    private LogScan(@Nullable int[] projectedFields) {
-        this.projectedFields = projectedFields;
-    }
+public interface MultiTableScan {
 
     /**
-     * Returns a new instance of LogScan description with column projection.
-     *
-     * @param projectedFields the projection fields
+     * Creates a {@link MultiTableLogScanner} to continuously read log data from multiple tables.
+     * Tables are registered dynamically on first {@code subscribe(...)} call.
      */
-    public LogScan withProjectedFields(int[] projectedFields) {
-        return new LogScan(projectedFields);
-    }
-
-    @Nullable
-    public int[] getProjectedFields() {
-        return projectedFields;
-    }
+    MultiTableLogScanner createLogScanner();
 }