[core] Handle query auth wrappers in split consumers

paimon-core/src/main/java/org/apache/paimon/crosspartition/IndexBootstrap.java

@@ -24,6 +24,7 @@
 import org.apache.paimon.data.InternalRow;
 import org.apache.paimon.data.JoinedRow;
 import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.options.Options;
 import org.apache.paimon.reader.RecordReader;
 import org.apache.paimon.schema.TableSchema;
 import org.apache.paimon.table.FileStoreTable;
@@ -41,12 +42,12 @@
 import java.io.Serializable;
 import java.time.Duration;
 import java.util.ArrayList;
-import java.util.Collections;
 import java.util.List;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
+import static org.apache.paimon.CoreOptions.QUERY_AUTH_ENABLED;
 import static org.apache.paimon.CoreOptions.SCAN_MODE;
 import static org.apache.paimon.CoreOptions.StartupMode.LATEST;
 import static org.apache.paimon.io.SplitsParallelReadUtil.parallelExecute;
@@ -80,11 +81,12 @@ public RecordReader<InternalRow> bootstrap(int numAssigners, int assignId) throw
                         .mapToInt(Integer::intValue)
                         .toArray();
 
-        // force using the latest scan mode
+        // Force using the latest scan mode and bypass query auth for this internal index read.
+        Options bootstrapOptions = new Options();
+        bootstrapOptions.set(SCAN_MODE, LATEST);
+        bootstrapOptions.set(QUERY_AUTH_ENABLED, false);
         ReadBuilder readBuilder =
-                table.copy(Collections.singletonMap(SCAN_MODE.key(), LATEST.toString()))
-                        .newReadBuilder()
-                        .withProjection(keyProjection);
+                table.copy(bootstrapOptions.toMap()).newReadBuilder().withProjection(keyProjection);
 
         DataTableScan tableScan = (DataTableScan) readBuilder.newScan();
         List<Split> splits =
@@ -119,7 +121,7 @@ public RecordReader<InternalRow> bootstrap(int numAssigners, int assignId) throw
                 options.pageSize(),
                 options.crossPartitionUpsertBootstrapParallelism(),
                 split -> {
-                    DataSplit dataSplit = ((DataSplit) split);
+                    DataSplit dataSplit = (DataSplit) split;
                     int bucket = dataSplit.bucket();
                     return partBucketConverter.toGenericRow(
                             new JoinedRow(dataSplit.partition(), GenericRow.of(bucket)));

paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java

@@ -37,6 +37,7 @@
 import org.apache.paimon.table.source.DataSplit;
 import org.apache.paimon.table.source.DataTableScan;
 import org.apache.paimon.table.source.InnerTableScan;
+import org.apache.paimon.table.source.QueryAuthSplit;
 import org.apache.paimon.table.source.Split;
 import org.apache.paimon.types.RowType;
 import org.apache.paimon.utils.Filter;
@@ -332,12 +333,17 @@ private Optional<GlobalIndexResult> evalGlobalIndex() {
     public static Plan wrapToIndexSplits(
             List<Split> splits, RowRangeIndex rowRangeIndex, ScoreGetter scoreGetter) {
         List<Split> indexedSplits = new ArrayList<>();
-        Function<Split, List<IndexedSplit>> process =
-                split ->
-                        Collections.singletonList(
-                                split instanceof IndexedSplit
-                                        ? (IndexedSplit) split
-                                        : wrap((DataSplit) split, rowRangeIndex, scoreGetter));
+        Function<Split, List<Split>> process =
+                split -> {
+                    Split wrappedSplit = QueryAuthSplit.unwrap(split);
+                    if (wrappedSplit instanceof IndexedSplit) {
+                        return Collections.singletonList(split);
+                    }
+                    IndexedSplit indexedSplit =
+                            wrap((DataSplit) wrappedSplit, rowRangeIndex, scoreGetter);
+                    return Collections.singletonList(
+                            QueryAuthSplit.retainAuth(split, indexedSplit));
+                };
         randomlyExecuteSequentialReturn(process, splits, null).forEachRemaining(indexedSplits::add);
         return () -> indexedSplits;
     }

paimon-core/src/main/java/org/apache/paimon/table/ChainGroupReadTable.java

@@ -19,6 +19,8 @@
 package org.apache.paimon.table;
 
 import org.apache.paimon.CoreOptions;
+import org.apache.paimon.annotation.VisibleForTesting;
+import org.apache.paimon.catalog.TableQueryAuthResult;
 import org.apache.paimon.codegen.CodeGenUtils;
 import org.apache.paimon.codegen.RecordComparator;
 import org.apache.paimon.data.BinaryRow;
@@ -36,7 +38,9 @@
 import org.apache.paimon.table.source.DataSplit;
 import org.apache.paimon.table.source.DataTableScan;
 import org.apache.paimon.table.source.InnerTableRead;
+import org.apache.paimon.table.source.QueryAuthSplit;
 import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.source.TableQueryAuth;
 import org.apache.paimon.table.source.TableRead;
 import org.apache.paimon.types.RowType;
 import org.apache.paimon.utils.ChainPartitionProjector;
@@ -45,6 +49,8 @@
 import org.apache.paimon.utils.Pair;
 import org.apache.paimon.utils.RowDataToObjectArrayConverter;
 
+import javax.annotation.Nullable;
+
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -119,6 +125,13 @@ private DataTableScan newDeltaScan(Function<FileStoreTable, DataTableScan> scanC
         return scanCreator.apply(other());
     }
 
+    static FileStoreTable withoutQueryAuth(FileStoreTable table) {
+        if (!table.coreOptions().queryAuthEnabled()) {
+            return table;
+        }
+        return table.copy(Collections.singletonMap(CoreOptions.QUERY_AUTH_ENABLED.key(), "false"));
+    }
+
     @Override
     public FileStoreTable copy(Map<String, String> dynamicOptions) {
         Map<String, String> wrappedOptions =
@@ -210,6 +223,8 @@ public static class ChainTableBatchScan extends FallbackReadScan {
         private final ChainGroupReadTable chainGroupReadTable;
         private final RecordComparator chainPartitionComparator;
         private final ChainPartitionProjector partitionProjector;
+        private final TableQueryAuth queryAuth;
+        @Nullable private RowType readType;
         private Predicate dataPredicate;
         private Filter<Integer> bucketFilter;
         protected boolean preloadTargetSnapshot = true;
@@ -219,17 +234,19 @@ public ChainTableBatchScan(
             this(tableSchema, chainGroupReadTable, FileStoreTable::newScan);
         }
 
-        private ChainTableBatchScan(
+        @VisibleForTesting
+        ChainTableBatchScan(
                 TableSchema tableSchema,
                 ChainGroupReadTable chainGroupReadTable,
                 Function<FileStoreTable, DataTableScan> scanCreator) {
             super(
                     chainGroupReadTable.wrapped,
                     chainGroupReadTable.other(),
                     tableSchema,
-                    scanCreator);
+                    physicalScanCreator(scanCreator));
             this.options = CoreOptions.fromMap(tableSchema.options());
             this.chainGroupReadTable = chainGroupReadTable;
+            this.queryAuth = chainGroupReadTable.catalogEnvironment().tableQueryAuth(this.options);
             this.partitionConverter =
                     new RowDataToObjectArrayConverter(tableSchema.logicalPartitionType());
             this.partitionComparator =
@@ -248,6 +265,19 @@ private ChainTableBatchScan(
                             partitionProjector.chainPartitionType().getFieldTypes());
         }
 
+        private static Function<FileStoreTable, DataTableScan> physicalScanCreator(
+                Function<FileStoreTable, DataTableScan> scanCreator) {
+            return table -> scanCreator.apply(ChainGroupReadTable.withoutQueryAuth(table));
+        }
+
+        @Override
+        public ChainTableBatchScan withReadType(@Nullable RowType readType) {
+            this.readType = readType;
+            mainScan.withReadType(readType);
+            fallbackScan.withReadType(readType);
+            return this;
+        }
+
         @Override
         public ChainTableBatchScan withFilter(Predicate predicate) {
             super.withFilter(predicate);
@@ -265,6 +295,18 @@ public ChainTableBatchScan withFilter(Predicate predicate) {
             return this;
         }
 
+        @Override
+        public ChainTableBatchScan withLimit(int limit) {
+            // Query authorization is applied only after the physical chain inputs have been
+            // assembled. Pushing a limit into those raw inputs could exhaust the limit with rows
+            // which are later removed by authorization. ReadBuilderImpl applies the logical limit
+            // after query authorization instead.
+            if (!options.queryAuthEnabled()) {
+                super.withLimit(limit);
+            }
+            return this;
+        }
+
         @Override
         public ChainTableBatchScan withPartitionFilter(Map<String, String> partitionSpec) {
             super.withPartitionFilter(partitionSpec);
@@ -332,6 +374,8 @@ public FallbackReadScan skipPreloadTargetSnapshot() {
          */
         @Override
         public Plan plan() {
+            TableQueryAuthResult queryAuthResult =
+                    queryAuth.auth(readType == null ? null : readType.getFieldNames());
             List<Split> splits = new ArrayList<>();
             PredicateBuilder builder = new PredicateBuilder(tableSchema.logicalPartitionType());
             Set<BinaryRow> snapshotPartitions = preloadTargetSnapshotSplits(splits);
@@ -359,6 +403,7 @@ public Plan plan() {
                 for (List<BinaryRow> deltaPartitionsInGroup : groupedDeltaPartitions.values()) {
 
                     // Sort delta by chain dimension ascending.
+                    // chainPartitionForCompare avoids copying BinaryRow in the comparator hot path.
                     deltaPartitionsInGroup.sort(
                             (a, b) ->
                                     chainPartitionComparator.compare(
@@ -430,18 +475,12 @@ public Plan plan() {
                             deltaScan.withPartitionFilter(selectedDeltaPartitions);
                         }
 
-                        List<DataSplit> deltaSubSplits =
-                                deltaScan.plan().splits().stream()
-                                        .map(s -> (DataSplit) s)
-                                        .collect(Collectors.toList());
-                        List<DataSplit> snapshotSubSplits = new ArrayList<>();
+                        List<Split> deltaSubSplits = deltaScan.plan().splits();
+                        List<Split> snapshotSubSplits = Collections.emptyList();
                         if (partitionPairs.getValue() != null) {
                             snapshotScan.withPartitionFilter(
                                     Collections.singletonList(partitionPairs.getValue()));
-                            snapshotSubSplits =
-                                    snapshotScan.plan().splits().stream()
-                                            .map(s -> (DataSplit) s)
-                                            .collect(Collectors.toList());
+                            snapshotSubSplits = snapshotScan.plan().splits();
                         }
                         splits.addAll(
                                 ChainTableUtils.buildChainSplits(
@@ -453,7 +492,8 @@ public Plan plan() {
                     }
                 }
             }
-            return new DataFilePlan<>(splits);
+            Plan plan = new DataFilePlan<>(splits);
+            return queryAuthResult == null ? plan : queryAuthResult.convertPlan(plan);
         }
 
         @Override
@@ -494,11 +534,11 @@ private Set<BinaryRow> preloadTargetSnapshotSplits(List<Split> splits) {
             }
 
             for (Split split : mainScan.plan().splits()) {
-                DataSplit dataSplit = (DataSplit) split;
+                DataSplit dataSplit = QueryAuthSplit.unwrapDataSplit(split);
                 HashMap<String, String> fileBucketPathMapping = new HashMap<>();
                 HashMap<String, String> fileBranchMapping = new HashMap<>();
                 for (DataFileMeta file : dataSplit.dataFiles()) {
-                    fileBucketPathMapping.put(file.fileName(), ((DataSplit) split).bucketPath());
+                    fileBucketPathMapping.put(file.fileName(), dataSplit.bucketPath());
                     fileBranchMapping.put(file.fileName(), options.scanFallbackSnapshotBranch());
                 }
                 splits.add(
@@ -526,6 +566,9 @@ private DataTableScan newFilteredScan(boolean snapshot) {
             if (bucketFilter != null) {
                 scan.withBucketFilter(bucketFilter);
             }
+            if (readType != null) {
+                scan.withReadType(readType);
+            }
             return scan;
         }
     }
@@ -582,7 +625,8 @@ public TableRead withIOManager(IOManager ioManager) {
 
         @Override
         public RecordReader<InternalRow> createReader(Split split) throws IOException {
-            if (split instanceof ChainSplit || split instanceof DataSplit) {
+            Split wrappedSplit = QueryAuthSplit.unwrap(split);
+            if (wrappedSplit instanceof ChainSplit || wrappedSplit instanceof DataSplit) {
                 return fallbackRead.createReader(split);
             }
             throw new IllegalArgumentException(

paimon-core/src/main/java/org/apache/paimon/table/ChainTableFileStoreTable.java

@@ -28,6 +28,7 @@
 import org.apache.paimon.table.source.ChainSplit;
 import org.apache.paimon.table.source.DataSplit;
 import org.apache.paimon.table.source.InnerTableRead;
+import org.apache.paimon.table.source.QueryAuthSplit;
 import org.apache.paimon.table.source.Split;
 import org.apache.paimon.table.source.StreamDataTableScan;
 import org.apache.paimon.table.source.TableRead;
@@ -209,7 +210,8 @@ public RecordReader<InternalRow> createReader(Split split) throws IOException {
                 // FallbackSplit (including FallbackDataSplit): use inherited fallback read logic
                 return fallbackRead.createReader(split);
             }
-            if (split instanceof ChainSplit || split instanceof DataSplit) {
+            Split wrappedSplit = QueryAuthSplit.unwrap(split);
+            if (wrappedSplit instanceof ChainSplit || wrappedSplit instanceof DataSplit) {
                 return chainGroupRead.createReader(split);
             }
             // Other split types: use inherited fallback read logic