[core] Fallback and chain queries require access to unreferenced columns

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

@@ -221,6 +221,12 @@ public InnerTableScan withPartitionFilter(PartitionPredicate partitionPredicate)
         return this;
     }
 
+    @Override
+    public InnerTableScan withInternalPartitionFilter(PartitionPredicate partitionPredicate) {
+        batchScan.withInternalPartitionFilter(partitionPredicate);
+        return this;
+    }
+
     @Override
     public InnerTableScan withBucketFilter(Filter<Integer> bucketFilter) {
         batchScan.withBucketFilter(bucketFilter);

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

@@ -318,6 +318,13 @@ public ChainTableBatchScan withPartitionFilter(PartitionPredicate partitionPredi
             return this;
         }
 
+        @Override
+        public ChainTableBatchScan withInternalPartitionFilter(
+                PartitionPredicate partitionPredicate) {
+            super.withInternalPartitionFilter(partitionPredicate);
+            return this;
+        }
+
         @Override
         public ChainTableBatchScan withPartitionFilter(Predicate partitionPredicate) {
             super.withPartitionFilter(partitionPredicate);
@@ -376,8 +383,7 @@ public Plan plan() {
             long targetSplitSize = options.splitTargetSize();
             long openFileCost = options.splitOpenFileCost();
 
-            DataTableScan deltaPartitionScan =
-                    newChainPartitionListingScan(false, getFallbackPartitionPredicate());
+            DataTableScan deltaPartitionScan = newChainPartitionListingScan(false);
             List<BinaryRow> deltaPartitions =
                     deltaPartitionScan.listPartitions().stream()
                             .filter(p -> !snapshotPartitions.contains(p))
@@ -425,9 +431,11 @@ public Plan plan() {
                     // List snapshot partitions for this group, sorted by chain dimension.
                     // The anchor of an admitted logical partition sits in an earlier one, which a
                     // rule on a chain partition key excludes.
+                    // This internal lookup must not add group or chain keys to user authorization.
                     List<BinaryRow> snapshotPartitionsInGroup =
-                            newChainPartitionListingScan(true, snapshotAnchorPredicate)
+                            newChainPartitionListingScan(true)
                                     .withoutAuthPartitionPushdown()
+                                    .withInternalPartitionFilter(snapshotAnchorPredicate)
                                     .listPartitions();
 
                     // Find delta → snapshot mapping (for each delta partition, find the nearest
@@ -449,6 +457,8 @@ public Plan plan() {
                         DataTableScan snapshotScan = newFilteredScan(true);
                         DataTableScan deltaScan = newFilteredScan(false);
 
+                        // These internal ranges select the deltas needed to reconstruct the target
+                        // partition while preserving the user-filter authorization requirements.
                         if (snapshotPartition == null) {
                             List<Predicate> predicates = new ArrayList<>();
                             predicates.add(
@@ -461,7 +471,10 @@ public Plan plan() {
                             predicates.add(
                                     ChainTableUtils.createLinearPredicate(
                                             deltaPartition, partitionConverter, builder::equal));
-                            deltaScan.withPartitionFilter(PredicateBuilder.or(predicates));
+                            deltaScan.withInternalPartitionFilter(
+                                    PartitionPredicate.fromPredicate(
+                                            tableSchema.logicalPartitionType(),
+                                            PredicateBuilder.or(predicates)));
                         } else {
                             List<BinaryRow> selectedDeltaPartitions =
                                     ChainTableUtils.getDeltaPartitionsWithProjector(
@@ -470,15 +483,21 @@ public Plan plan() {
                                             options,
                                             chainPartitionComparator,
                                             partitionProjector);
-                            deltaScan.withPartitionFilter(selectedDeltaPartitions);
+                            deltaScan.withInternalPartitionFilter(
+                                    PartitionPredicate.fromMultiple(
+                                            tableSchema.logicalPartitionType(),
+                                            selectedDeltaPartitions));
                         }
 
                         ChainAuth auth = new ChainAuth();
                         List<DataSplit> deltaSubSplits = auth.unwrap(deltaScan.plan().splits());
                         List<DataSplit> snapshotSubSplits = new ArrayList<>();
                         if (partitionPairs.getValue() != null) {
-                            snapshotScan.withPartitionFilter(
-                                    Collections.singletonList(partitionPairs.getValue()));
+                            // Read the snapshot anchor selected for this delta partition.
+                            snapshotScan.withInternalPartitionFilter(
+                                    PartitionPredicate.fromMultiple(
+                                            tableSchema.logicalPartitionType(),
+                                            Collections.singletonList(partitionPairs.getValue())));
                             snapshotSubSplits = auth.unwrap(snapshotScan.plan().splits());
                         }
                         splits.addAll(
@@ -500,10 +519,8 @@ public Plan plan() {
 
         @Override
         public List<PartitionEntry> listPartitionEntries() {
-            DataTableScan snapshotScan =
-                    newChainPartitionListingScan(true, getMainPartitionPredicate());
-            DataTableScan deltaScan =
-                    newChainPartitionListingScan(false, getFallbackPartitionPredicate());
+            DataTableScan snapshotScan = newChainPartitionListingScan(true);
+            DataTableScan deltaScan = newChainPartitionListingScan(false);
             List<PartitionEntry> partitionEntries =
                     new ArrayList<>(snapshotScan.listPartitionEntries());
             Set<BinaryRow> partitions =
@@ -517,15 +534,12 @@ public List<PartitionEntry> listPartitionEntries() {
             return partitionEntries;
         }
 
-        private DataTableScan newChainPartitionListingScan(
-                boolean snapshot, PartitionPredicate scanPartitionPredicate) {
+        private DataTableScan newChainPartitionListingScan(boolean snapshot) {
             DataTableScan scan =
                     snapshot
                             ? chainGroupReadTable.newSnapshotScan(scanCreator)
                             : chainGroupReadTable.newDeltaScan(scanCreator);
-            if (scanPartitionPredicate != null) {
-                scan.withPartitionFilter(scanPartitionPredicate);
-            }
+            configureBranchScan(scan, snapshot);
             return scan;
         }
 
@@ -542,9 +556,7 @@ private Set<BinaryRow> preloadTargetSnapshotSplits(List<Split> splits) {
             }
             splits.addAll(auth.reapply(chainSplits));
 
-            snapshotPartitions.addAll(
-                    newChainPartitionListingScan(true, getMainPartitionPredicate())
-                            .listPartitions());
+            snapshotPartitions.addAll(newChainPartitionListingScan(true).listPartitions());
             return snapshotPartitions;
         }
 
@@ -554,6 +566,7 @@ private DataTableScan newFilteredScan(boolean snapshot) {
                                     ? chainGroupReadTable.newSnapshotScan(scanCreator)
                                     : chainGroupReadTable.newDeltaScan(scanCreator))
                             .withoutAuthPartitionPushdown();
+            configureBranchScan(scan, snapshot);
             if (keyPredicate != null) {
                 scan.withFilter(keyPredicate);
             }

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

@@ -40,6 +40,7 @@
 import org.apache.paimon.table.source.StreamDataTableScan;
 import org.apache.paimon.table.source.TableScan;
 import org.apache.paimon.table.source.snapshot.StartingContext;
+import org.apache.paimon.types.RowType;
 import org.apache.paimon.utils.ChainPartitionProjector;
 import org.apache.paimon.utils.ChainTableUtils;
 import org.apache.paimon.utils.Filter;
@@ -122,6 +123,8 @@ public class ChainTableStreamScan implements StreamDataTableScan {
 
     @Nullable private Filter<Integer> bucketFilter;
 
+    @Nullable private RowType readType;
+
     /** Maximum number of retries when race condition is detected during position capture. */
     private static final int MAX_RACE_RETRIES = 3;
 
@@ -268,6 +271,7 @@ private TableScan.Plan planStarting() {
             pinnedSnapshot = chainGroupReadTable.wrapped.copy(pinnedOptions(snapshotLatestId));
             DataTableScan partitionListingScan =
                     pinnedSnapshot.newScan().withoutAuthPartitionPushdown();
+            partitionListingScan.withReadType(readType);
             for (BinaryRow partition : partitionListingScan.listPartitions()) {
                 Object groupKey = toGroupKey(partition);
                 BinaryRow existingLatest = latestChainPartitionPerGroup.get(groupKey);
@@ -287,7 +291,10 @@ private TableScan.Plan planStarting() {
         Map<BinaryRow, List<DataSplit>> snapshotSplitsByPartition;
         if (!latestPartitions.isEmpty() && pinnedSnapshot != null) {
             DataTableScan snapshotScan = pinnedSnapshot.newScan().withoutAuthPartitionPushdown();
-            snapshotScan.withPartitionFilter(latestPartitions);
+            // Selecting the latest snapshot per group is internal to the streaming starting scan.
+            snapshotScan.withInternalPartitionFilter(
+                    PartitionPredicate.fromMultiple(
+                            chainGroupReadTable.schema().logicalPartitionType(), latestPartitions));
             applyPredicatesShardAndBucket(snapshotScan);
             snapshotSplitsByPartition = groupByPartition(snapshotScan);
         } else {
@@ -536,6 +543,14 @@ private Object toGroupKey(BinaryRow fullPartition) {
         return partitionProjector.extractGroupPartition(fullPartition);
     }
 
+    @Override
+    public InnerTableScan withReadType(@Nullable RowType readType) {
+        this.readType = readType;
+        batchScan.withReadType(readType);
+        deltaStreamScan.withReadType(readType);
+        return this;
+    }
+
     @Override
     public InnerTableScan withFilter(Predicate predicate) {
         if (predicate == null) {
@@ -620,10 +635,11 @@ private static Map<String, String> pinnedOptions(long snapshotId) {
     }
 
     /**
-     * Applies all previously set predicates, shard, and bucket filter to a newly created scan. Used
-     * for the pinned delta scan in {@link #planStarting()}.
+     * Applies the read type, predicates, shard, and bucket filter to a newly created scan in {@link
+     * #planStarting()}.
      */
     private void applyPredicatesShardAndBucket(DataTableScan scan) {
+        scan.withReadType(readType);
         for (Predicate p : predicates) {
             scan.withFilter(p);
         }

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

@@ -34,6 +34,7 @@
 import org.apache.paimon.options.Options;
 import org.apache.paimon.partition.PartitionPredicate;
 import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateVisitor;
 import org.apache.paimon.predicate.TopN;
 import org.apache.paimon.reader.ReadBatchSizer;
 import org.apache.paimon.reader.RecordReader;
@@ -54,6 +55,7 @@
 import org.apache.paimon.utils.Pair;
 import org.apache.paimon.utils.Preconditions;
 import org.apache.paimon.utils.SegmentsCache;
+import org.apache.paimon.utils.TypeUtils;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -64,6 +66,7 @@
 import java.io.ObjectInputStream;
 import java.io.ObjectOutputStream;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -422,8 +425,11 @@ public static class FallbackReadScan implements DataTableScan {
         protected final Function<FileStoreTable, DataTableScan> scanCreator;
         protected final DataTableScan mainScan;
         protected final DataTableScan fallbackScan;
+        @Nullable protected RowType readType;
+        private Set<String> userFilterFields = Collections.emptySet();
         private PartitionPredicate mainPartitionPredicate;
         private PartitionPredicate fallbackPartitionPredicate;
+        @Nullable private PartitionPredicate internalPartitionPredicate;
 
         public FallbackReadScan(
                 FileStoreTable wrappedTable,
@@ -449,6 +455,7 @@ public FallbackReadScan withShard(int indexOfThisSubtask, int numberOfParallelSu
         public FallbackReadScan withFilter(Predicate predicate) {
             mainScan.withFilter(predicate);
             fallbackScan.withFilter(predicate);
+            userFilterFields = PredicateVisitor.collectFieldNames(predicate);
             if (predicate != null) {
                 Pair<Optional<PartitionPredicate>, List<Predicate>> pair =
                         PartitionPredicate.splitPartitionPredicatesAndDataPredicates(
@@ -462,6 +469,7 @@ public FallbackReadScan withFilter(Predicate predicate) {
 
         @Override
         public FallbackReadScan withReadType(@Nullable RowType readType) {
+            this.readType = readType;
             mainScan.withReadType(readType);
             fallbackScan.withReadType(readType);
             return this;
@@ -524,6 +532,14 @@ public InnerTableScan withPartitionFilter(PartitionPredicate partitionPredicate)
             return this;
         }
 
+        @Override
+        public InnerTableScan withInternalPartitionFilter(PartitionPredicate partitionPredicate) {
+            mainScan.withInternalPartitionFilter(partitionPredicate);
+            fallbackScan.withInternalPartitionFilter(partitionPredicate);
+            this.internalPartitionPredicate = partitionPredicate;
+            return this;
+        }
+
         @Override
         public FallbackReadScan withPartitionFilter(Predicate partitionPredicate) {
             mainScan.withPartitionFilter(partitionPredicate);
@@ -601,7 +617,10 @@ public TableScan.Plan plan() {
                             .filter(p -> !completePartitions.contains(p))
                             .collect(Collectors.toList());
             if (!remainingPartitions.isEmpty()) {
-                fallbackScan.withPartitionFilter(remainingPartitions);
+                // Restrict fallback reads to partitions missing from the main branch.
+                fallbackScan.withInternalPartitionFilter(
+                        PartitionPredicate.fromMultiple(
+                                tableSchema.logicalPartitionType(), remainingPartitions));
                 for (Split split : fallbackScan.plan().splits()) {
                     splits.add(toFallbackSplit(split, true));
                 }
@@ -637,15 +656,15 @@ public List<BinaryRow> topNPartitions(int num, int partitionFieldCount) {
         }
 
         protected void setPartitionPredicate(PartitionPredicate predicate) {
-            this.mainPartitionPredicate = predicate;
-            this.fallbackPartitionPredicate = predicate;
+            setPartitionPredicate(predicate, predicate);
         }
 
         protected void setPartitionPredicate(
                 PartitionPredicate mainPartitionPredicate,
                 PartitionPredicate fallbackPartitionPredicate) {
             this.mainPartitionPredicate = mainPartitionPredicate;
             this.fallbackPartitionPredicate = fallbackPartitionPredicate;
+            this.internalPartitionPredicate = null;
         }
 
         protected PartitionPredicate getMainPartitionPredicate() {
@@ -658,12 +677,35 @@ protected PartitionPredicate getFallbackPartitionPredicate() {
 
         private DataTableScan newPartitionListingScan(boolean isMain) {
             DataTableScan scan = scanCreator.apply(isMain ? wrappedTable : fallbackTable);
+            configureBranchScan(scan, isMain);
+            return scan;
+        }
+
+        @Nullable
+        protected RowType branchReadType() {
+            // Include projected and user-filter columns in internal scans' authorization requests.
+            // Carry filter dependencies through the read type: pushing value predicates here could
+            // change fallback partition ownership or discard records needed for chain merging.
+            return readType == null
+                    ? null
+                    : TypeUtils.withMissingFields(
+                            tableSchema.logicalRowType(), readType, userFilterFields);
+        }
+
+        /** Applies the shared read type and partition conditions to a newly created branch scan. */
+        protected void configureBranchScan(DataTableScan scan, boolean isMain) {
+            scan.withReadType(branchReadType());
             if (isMain && getMainPartitionPredicate() != null) {
                 scan.withPartitionFilter(getMainPartitionPredicate());
             } else if (!isMain && getFallbackPartitionPredicate() != null) {
                 scan.withPartitionFilter(getFallbackPartitionPredicate());
             }
-            return scan;
+            if (internalPartitionPredicate != null) {
+                // Replace the physical partition filter while preserving the user partition-filter
+                // authorization fields registered above. Columns referenced only by the internal
+                // filter are not added to authorization.
+                scan.withInternalPartitionFilter(internalPartitionPredicate);
+            }
         }
     }
 

paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableScan.java

@@ -271,6 +271,16 @@ public AbstractDataTableScan withPartitionFilter(PartitionPredicate partitionPre
                 () -> snapshotReader.withPartitionFilter(partitionPredicate));
     }
 
+    @Override
+    public AbstractDataTableScan withInternalPartitionFilter(
+            PartitionPredicate partitionPredicate) {
+        // Replace the physical partition filter while preserving user-filter authorization
+        // fields. Internal pruning must not add its own columns to authorization.
+        return pushPartitionFilter(
+                partitionFilterFields,
+                () -> snapshotReader.withPartitionFilter(partitionPredicate));
+    }
+
     @Override
     public InnerTableScan withPartitionFilter(Predicate predicate) {
         return pushPartitionFilter(

paimon-core/src/main/java/org/apache/paimon/table/source/InnerTableScan.java

@@ -63,6 +63,29 @@ default InnerTableScan withPartitionFilter(PartitionPredicate partitionPredicate
         return this;
     }
 
+    /**
+     * Replaces the physical partition filter with one generated solely for internal scan planning.
+     *
+     * <p>Internal filters select the inputs needed to execute a query, such as missing fallback
+     * partitions or chain snapshot anchors. The new filter is not intersected with the existing
+     * physical partition filter.
+     *
+     * <p>Existing user-filter authorization requirements are preserved. Columns referenced only by
+     * this internal filter are not added to authorization.
+     *
+     * <p>Use {@link #withPartitionFilter(PartitionPredicate)} for user query conditions, including
+     * those rewritten or pushed down during planning. Internal filters require explicit support
+     * from the implementation; the default implementation throws {@link
+     * UnsupportedOperationException}.
+     *
+     * @param partitionPredicate the replacement physical partition filter
+     * @return this scan
+     * @throws UnsupportedOperationException if internal partition filtering is not supported
+     */
+    default InnerTableScan withInternalPartitionFilter(PartitionPredicate partitionPredicate) {
+        throw new UnsupportedOperationException("Internal partition filters are not supported");
+    }
+
     default InnerTableScan withPartitionFilter(Predicate predicate) {
         return this;
     }