[common] Support variant type

Author: swuferhong

docs/variant-shredding-design.md

@@ -27,6 +27,7 @@
 
 import java.io.IOException;
 import java.util.List;
+import java.util.Map;
 
 /**
  * Used to configure and create a scanner to scan data for a table.
@@ -60,6 +61,19 @@ public interface Scan {
      */
     Scan limit(int rowNumber);
 
+    /**
+     * Returns a new scan from this with variant sub-field projection hints. When a Variant column
+     * has been shredded, specifying which sub-fields are needed allows the server to send only the
+     * relevant typed_value children, reducing network transfer.
+     *
+     * <p>The map key is the column index (in the projected or full schema), and the value is the
+     * list of field names to project within that Variant column.
+     *
+     * @param columnToFields mapping from Variant column index to desired field names, or null to
+     *     disable sub-field projection
+     */
+    Scan variantFieldProjection(@Nullable Map<Integer, List<String>> columnToFields);
+
     /**
      * Creates a {@link LogScanner} to continuously read log data for this scan.
      *

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

@@ -40,6 +40,7 @@
 
 import java.io.IOException;
 import java.util.List;
+import java.util.Map;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
@@ -53,27 +54,32 @@ public class TableScan implements Scan {
     @Nullable private final int[] projectedColumns;
     /** The limited row number to read. No limit if is null. */
     @Nullable private final Integer limit;
+    /** Variant sub-field projection hints. Null means no sub-field projection. */
+    @Nullable private final Map<Integer, List<String>> variantFieldProjection;
 
     public TableScan(FlussConnection conn, TableInfo tableInfo, SchemaGetter schemaGetter) {
-        this(conn, tableInfo, schemaGetter, null, null);
+        this(conn, tableInfo, schemaGetter, null, null, null);
     }
 
     private TableScan(
             FlussConnection conn,
             TableInfo tableInfo,
             SchemaGetter schemaGetter,
             @Nullable int[] projectedColumns,
-            @Nullable Integer limit) {
+            @Nullable Integer limit,
+            @Nullable Map<Integer, List<String>> variantFieldProjection) {
         this.conn = conn;
         this.tableInfo = tableInfo;
         this.projectedColumns = projectedColumns;
         this.limit = limit;
         this.schemaGetter = schemaGetter;
+        this.variantFieldProjection = variantFieldProjection;
     }
 
     @Override
     public Scan project(@Nullable int[] projectedColumns) {
-        return new TableScan(conn, tableInfo, schemaGetter, projectedColumns, limit);
+        return new TableScan(
+                conn, tableInfo, schemaGetter, projectedColumns, limit, variantFieldProjection);
     }
 
     @Override
@@ -92,12 +98,20 @@ public Scan project(List<String> projectedColumnNames) {
             }
             columnIndexes[i] = index;
         }
-        return new TableScan(conn, tableInfo, schemaGetter, columnIndexes, limit);
+        return new TableScan(
+                conn, tableInfo, schemaGetter, columnIndexes, limit, variantFieldProjection);
     }
 
     @Override
     public Scan limit(int rowNumber) {
-        return new TableScan(conn, tableInfo, schemaGetter, projectedColumns, rowNumber);
+        return new TableScan(
+                conn, tableInfo, schemaGetter, projectedColumns, rowNumber, variantFieldProjection);
+    }
+
+    @Override
+    public Scan variantFieldProjection(@Nullable Map<Integer, List<String>> columnToFields) {
+        return new TableScan(
+                conn, tableInfo, schemaGetter, projectedColumns, limit, columnToFields);
     }
 
     @Override
@@ -116,6 +130,7 @@ public LogScanner createLogScanner() {
                 conn.getClientMetricGroup(),
                 conn.getOrCreateRemoteFileDownloader(),
                 projectedColumns,
+                variantFieldProjection,
                 schemaGetter);
     }
 

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

@@ -50,6 +50,7 @@
 import org.apache.fluss.rpc.messages.PbFetchLogReqForTable;
 import org.apache.fluss.rpc.messages.PbFetchLogRespForBucket;
 import org.apache.fluss.rpc.messages.PbFetchLogRespForTable;
+import org.apache.fluss.rpc.messages.PbVariantFieldProjection;
 import org.apache.fluss.rpc.protocol.ApiError;
 import org.apache.fluss.rpc.protocol.Errors;
 import org.apache.fluss.utils.IOUtils;
@@ -93,6 +94,12 @@ public class LogFetcher implements Closeable {
     //  bytes from remote file.
     private final LogRecordReadContext remoteReadContext;
     @Nullable private final Projection projection;
+    /**
+     * Variant sub-field projection hints. Maps Variant column index to desired sub-field names.
+     * Null means no sub-field projection (initial version).
+     */
+    @Nullable private final Map<Integer, List<String>> variantFieldProjection;
+
     private final int maxFetchBytes;
     private final int maxBucketFetchBytes;
     private final int minFetchBytes;
@@ -121,13 +128,36 @@ public LogFetcher(
             ScannerMetricGroup scannerMetricGroup,
             RemoteFileDownloader remoteFileDownloader,
             SchemaGetter schemaGetter) {
+        this(
+                tableInfo,
+                projection,
+                null,
+                logScannerStatus,
+                conf,
+                metadataUpdater,
+                scannerMetricGroup,
+                remoteFileDownloader,
+                schemaGetter);
+    }
+
+    public LogFetcher(
+            TableInfo tableInfo,
+            @Nullable Projection projection,
+            @Nullable Map<Integer, List<String>> variantFieldProjection,
+            LogScannerStatus logScannerStatus,
+            Configuration conf,
+            MetadataUpdater metadataUpdater,
+            ScannerMetricGroup scannerMetricGroup,
+            RemoteFileDownloader remoteFileDownloader,
+            SchemaGetter schemaGetter) {
         this.tablePath = tableInfo.getTablePath();
         this.isPartitioned = tableInfo.isPartitioned();
         this.readContext =
                 LogRecordReadContext.createReadContext(tableInfo, false, projection, schemaGetter);
         this.remoteReadContext =
                 LogRecordReadContext.createReadContext(tableInfo, true, projection, schemaGetter);
         this.projection = projection;
+        this.variantFieldProjection = variantFieldProjection;
         this.logScannerStatus = logScannerStatus;
         this.maxFetchBytes =
                 (int) conf.get(ConfigOptions.CLIENT_SCANNER_LOG_FETCH_MAX_BYTES).getBytes();
@@ -528,12 +558,23 @@ Map<Integer, FetchLogRequest> prepareFetchLogRequests() {
                                 new PbFetchLogReqForTable().setTableId(finalTableId);
                         if (readContext.isProjectionPushDowned()) {
                             assert projection != null;
+                            int[] projectedFields = projection.getProjectionInOrder();
                             reqForTable
                                     .setProjectionPushdownEnabled(true)
-                                    .setProjectedFields(projection.getProjectionInOrder());
+                                    .setProjectedFields(projectedFields);
                         } else {
                             reqForTable.setProjectionPushdownEnabled(false);
                         }
+                        // Serialize variant sub-field projection hints if present
+                        if (variantFieldProjection != null && !variantFieldProjection.isEmpty()) {
+                            variantFieldProjection.forEach(
+                                    (colIdx, fieldNames) -> {
+                                        PbVariantFieldProjection vfp =
+                                                reqForTable.addVariantFieldProjection();
+                                        vfp.setColumnIndex(colIdx);
+                                        vfp.addAllFieldNames(fieldNames);
+                                    });
+                        }
                         reqForTable.addAllBucketsReqs(reqForBuckets);
                         fetchLogRequest.addAllTablesReqs(Collections.singletonList(reqForTable));
                         fetchLogRequests.put(leaderId, fetchLogRequest);

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

@@ -21,6 +21,9 @@
 
 import javax.annotation.Nullable;
 
+import java.util.List;
+import java.util.Map;
+
 /**
  * Used to describe the operation to scan log data by {@link LogScanner} to a table.
  *
@@ -32,12 +35,21 @@ public class LogScan {
     /** The projected fields to do projection. No projection if is null. */
     @Nullable private final int[] projectedFields;
 
+    /**
+     * Variant sub-field projection hints. Maps Variant column index to the list of sub-field names
+     * that should be projected. Null means no sub-field projection.
+     */
+    @Nullable private final Map<Integer, List<String>> variantFieldProjection;
+
     public LogScan() {
-        this(null);
+        this(null, null);
     }
 
-    private LogScan(@Nullable int[] projectedFields) {
+    private LogScan(
+            @Nullable int[] projectedFields,
+            @Nullable Map<Integer, List<String>> variantFieldProjection) {
         this.projectedFields = projectedFields;
+        this.variantFieldProjection = variantFieldProjection;
     }
 
     /**
@@ -46,11 +58,26 @@ private LogScan(@Nullable int[] projectedFields) {
      * @param projectedFields the projection fields
      */
     public LogScan withProjectedFields(int[] projectedFields) {
-        return new LogScan(projectedFields);
+        return new LogScan(projectedFields, variantFieldProjection);
+    }
+
+    /**
+     * Returns a new instance of LogScan description with variant sub-field projection hints.
+     *
+     * @param variantFieldProjection mapping from Variant column index to desired field names
+     */
+    public LogScan withVariantFieldProjection(
+            @Nullable Map<Integer, List<String>> variantFieldProjection) {
+        return new LogScan(projectedFields, variantFieldProjection);
     }
 
     @Nullable
     public int[] getProjectedFields() {
         return projectedFields;
     }
+
+    @Nullable
+    public Map<Integer, List<String>> getVariantFieldProjection() {
+        return variantFieldProjection;
+    }
 }

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

@@ -85,6 +85,26 @@ public LogScannerImpl(
             RemoteFileDownloader remoteFileDownloader,
             @Nullable int[] projectedFields,
             SchemaGetter schemaGetter) {
+        this(
+                conf,
+                tableInfo,
+                metadataUpdater,
+                clientMetricGroup,
+                remoteFileDownloader,
+                projectedFields,
+                null,
+                schemaGetter);
+    }
+
+    public LogScannerImpl(
+            Configuration conf,
+            TableInfo tableInfo,
+            MetadataUpdater metadataUpdater,
+            ClientMetricGroup clientMetricGroup,
+            RemoteFileDownloader remoteFileDownloader,
+            @Nullable int[] projectedFields,
+            @Nullable Map<Integer, List<String>> variantFieldProjection,
+            SchemaGetter schemaGetter) {
         this.tablePath = tableInfo.getTablePath();
         this.tableId = tableInfo.getTableId();
         this.isPartitionedTable = tableInfo.isPartitioned();
@@ -98,6 +118,7 @@ public LogScannerImpl(
                 new LogFetcher(
                         tableInfo,
                         projection,
+                        variantFieldProjection,
                         logScannerStatus,
                         conf,
                         metadataUpdater,
@@ -112,6 +133,8 @@ public LogScannerImpl(
      */
     @Nullable
     private Projection sanityProjection(@Nullable int[] projectedFields, TableInfo tableInfo) {
+        // Validate against the user-visible row type (excludes internal shredded columns like $v.x)
+        // so that projection indices from callers (e.g. Flink) stay within user-visible bounds.
         RowType tableRowType = tableInfo.getRowType();
         if (projectedFields != null) {
             for (int projectedField : projectedFields) {