From 9badcbf0e97632ed0b41c8bbd6bf98385f4e455b Mon Sep 17 00:00:00 2001 From: Ran Tao Date: Wed, 21 Oct 2026 14:19:20 +0800 Subject: [PATCH] [core] Fallback and chain queries require access to unreferenced columns --- .../globalindex/DataEvolutionBatchScan.java | 6 + .../paimon/table/ChainGroupReadTable.java | 51 +++-- .../paimon/table/ChainTableStreamScan.java | 22 +- .../table/FallbackReadFileStoreTable.java | 50 ++++- .../table/source/AbstractDataTableScan.java | 10 + .../paimon/table/source/InnerTableScan.java | 23 +++ .../table/ChainTableFileStoreTableTest.java | 193 +++++++++++++++++- .../table/FallbackReadFileStoreTableTest.java | 185 ++++++++++++++++- 8 files changed, 508 insertions(+), 32 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java b/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java index 4fdae682e5fb..ae6cbf01f8b4 100644 --- a/paimon-core/src/main/java/org/apache/paimon/globalindex/DataEvolutionBatchScan.java +++ b/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 bucketFilter) { batchScan.withBucketFilter(bucketFilter); diff --git a/paimon-core/src/main/java/org/apache/paimon/table/ChainGroupReadTable.java b/paimon-core/src/main/java/org/apache/paimon/table/ChainGroupReadTable.java index 5564260d8e88..2a18a8fdd9a5 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/ChainGroupReadTable.java +++ b/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 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 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 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 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 deltaSubSplits = auth.unwrap(deltaScan.plan().splits()); List 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 listPartitionEntries() { - DataTableScan snapshotScan = - newChainPartitionListingScan(true, getMainPartitionPredicate()); - DataTableScan deltaScan = - newChainPartitionListingScan(false, getFallbackPartitionPredicate()); + DataTableScan snapshotScan = newChainPartitionListingScan(true); + DataTableScan deltaScan = newChainPartitionListingScan(false); List partitionEntries = new ArrayList<>(snapshotScan.listPartitionEntries()); Set partitions = @@ -517,15 +534,12 @@ public List 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 preloadTargetSnapshotSplits(List 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); } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/ChainTableStreamScan.java b/paimon-core/src/main/java/org/apache/paimon/table/ChainTableStreamScan.java index e9f181f47f07..b8d3ed96a69a 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/ChainTableStreamScan.java +++ b/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 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> 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 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); } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/FallbackReadFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/FallbackReadFileStoreTable.java index 2863320050e0..46c7cfcfc205 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/FallbackReadFileStoreTable.java +++ b/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 scanCreator; protected final DataTableScan mainScan; protected final DataTableScan fallbackScan; + @Nullable protected RowType readType; + private Set 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, List> 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,8 +656,7 @@ public List topNPartitions(int num, int partitionFieldCount) { } protected void setPartitionPredicate(PartitionPredicate predicate) { - this.mainPartitionPredicate = predicate; - this.fallbackPartitionPredicate = predicate; + setPartitionPredicate(predicate, predicate); } protected void setPartitionPredicate( @@ -646,6 +664,7 @@ protected void setPartitionPredicate( 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); + } } } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableScan.java b/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableScan.java index 9cea4edfe477..4ac5cda74fe0 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/AbstractDataTableScan.java +++ b/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( diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/InnerTableScan.java b/paimon-core/src/main/java/org/apache/paimon/table/source/InnerTableScan.java index 6c4b18ec9b76..db4fa29f9477 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/InnerTableScan.java +++ b/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. + * + *

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. + * + *

Existing user-filter authorization requirements are preserved. Columns referenced only by + * this internal filter are not added to authorization. + * + *

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; } diff --git a/paimon-core/src/test/java/org/apache/paimon/table/ChainTableFileStoreTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/ChainTableFileStoreTableTest.java index a651cfc2c9d1..2c9ab22db2b8 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/ChainTableFileStoreTableTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/ChainTableFileStoreTableTest.java @@ -68,6 +68,7 @@ import org.junit.jupiter.api.io.TempDir; import org.junit.jupiter.params.ParameterizedTest; import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.CsvSource; import org.junit.jupiter.params.provider.MethodSource; import org.junit.jupiter.params.provider.ValueSource; import org.mockito.Mockito; @@ -83,6 +84,7 @@ import java.util.Random; import java.util.UUID; import java.util.function.Consumer; +import java.util.function.Function; import java.util.function.Supplier; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -987,10 +989,184 @@ public void testQueryAuthStreamingStartingPlanKeepsChainSplitsWrapped() throws E }); } + @ParameterizedTest + @CsvSource({ + "false,false,true", + "true,false,true", + "true,true,true", + "false,false,false", + "true,false,false", + "true,true,false" + }) + public void testQueryAuthWithProjectedColumns( + boolean streaming, boolean streamingMergeSnapshot, boolean withSnapshot) + throws Exception { + createChainTableWithQueryAuth(withSnapshot); + FileStoreTable table = + loadTable( + queryAuthEnvironment( + select -> { + if (select == null + || !Arrays.asList("k", "dt", "hour") + .containsAll(select)) { + throw new SecurityException( + "No permission to read unselected columns"); + } + return new TableQueryAuthResult(null, null); + })) + .copy( + Collections.singletonMap( + CoreOptions.CHAIN_TABLE_STREAMING_MERGE_SNAPSHOT.key(), + String.valueOf(streamingMergeSnapshot))); + ReadBuilder readBuilder = + table.newReadBuilder().withReadType(table.rowType().project("k", "dt", "hour")); + TableScan scan = streaming ? readBuilder.newStreamScan() : readBuilder.newScan(); + List expected = + new ArrayList<>( + Arrays.asList( + row(1L, "20250810", "20"), row(2L, "20250810", "20"), + row(1L, "20250810", "21"), row(4L, "20250810", "21"))); + if (!streaming) { + expected.add(row(2L, "20250810", "21")); + } + if (streamingMergeSnapshot) { + expected = + Arrays.asList( + row(1L, "20250810", "21"), + row(2L, "20250810", "21"), + row(4L, "20250810", "21")); + } + if (!withSnapshot) { + expected = Arrays.asList(row(1L, "20250810", "21"), row(4L, "20250810", "21")); + } + assertThat(readProjectedSplits(readBuilder, scan.plan().splits())) + .containsExactlyInAnyOrderElementsOf(expected); + if (streaming) { + writeWithCommit( + loadTable().switchToBranch(DELTA_BRANCH), + row(5L, 1L, "5", "CN", "20250810", "22")); + assertThat(readProjectedSplits(readBuilder, scan.plan().splits())) + .containsExactly(row(5L, "20250810", "22")); + } + + if (!streaming) { + assertThat(readBuilder.newScan().listPartitionEntries()).hasSize(withSnapshot ? 2 : 1); + ReadBuilder partitionFiltered = + table.newReadBuilder() + .withReadType(readBuilder.readType()) + .withPartitionFilter(Collections.singletonMap("region", "CN")); + assertThatThrownBy(() -> partitionFiltered.newScan().plan()) + .hasMessageContaining("No permission"); + assertThatThrownBy(() -> partitionFiltered.newScan().listPartitionEntries()) + .hasMessageContaining("No permission"); + ReadBuilder filtered = + table.newReadBuilder() + .withReadType(readBuilder.readType()) + .withFilter( + new PredicateBuilder(table.rowType()) + .equal(3, BinaryString.fromString("CN"))); + assertThatThrownBy(() -> filtered.newScan().plan()) + .hasMessageContaining("No permission"); + } + + ReadBuilder denied = table.newReadBuilder().withReadType(table.rowType().project("v")); + assertThatThrownBy(() -> (streaming ? denied.newStreamScan() : denied.newScan()).plan()) + .hasMessageContaining("No permission"); + readBuilder.withFilter( + new PredicateBuilder(table.rowType()).equal(2, BinaryString.fromString("secret"))); + assertThatThrownBy( + () -> + (streaming ? readBuilder.newStreamScan() : readBuilder.newScan()) + .plan()) + .hasMessageContaining("No permission"); + } + + @ParameterizedTest + @CsvSource({ + "false,false,false", + "false,false,true", + "false,true,false", + "false,true,true", + "true,false,false", + "true,false,true", + "true,true,false", + "true,true,true" + }) + public void testQueryAuthWithUnprojectedValueFilter( + boolean skipPreload, boolean allowValue, boolean withSnapshot) throws Exception { + createChainTableWithQueryAuth(withSnapshot); + List allowedFields = new ArrayList<>(Arrays.asList("k", "dt", "hour")); + if (allowValue) { + allowedFields.add("v"); + } + FallbackReadFileStoreTable table = + (FallbackReadFileStoreTable) + loadTable( + queryAuthEnvironment( + select -> { + if (select == null + || !allowedFields.containsAll(select)) { + throw new SecurityException("No permission"); + } + return new TableQueryAuthResult(null, null); + })); + FileStoreTable chainGroup = table.other(); + // The snapshot value "1" is replaced by "1-1" in delta. Pushing the value filter + // into either merge input could discard the update and resurrect the old value. + for (String value : Arrays.asList("1", "1-1")) { + ReadBuilder readBuilder = + chainGroup + .newReadBuilder() + .withReadType(chainGroup.rowType().project("k", "dt", "hour")) + .withFilter( + new PredicateBuilder(chainGroup.rowType()) + .equal(2, BinaryString.fromString(value))); + ChainGroupReadTable.ChainTableBatchScan scan = + (ChainGroupReadTable.ChainTableBatchScan) readBuilder.newScan(); + if (skipPreload) { + scan.skipPreloadTargetSnapshot(); + } + if (!allowValue) { + assertThatThrownBy(scan::plan).hasMessageContaining("No permission"); + assertThatThrownBy(scan::listPartitionEntries) + .hasMessageContaining("No permission"); + continue; + } + + List rows = new ArrayList<>(); + InternalRowSerializer serializer = new InternalRowSerializer(readBuilder.readType()); + try (RecordReader reader = + readBuilder.newRead().executeFilter().createReader(scan.plan().splits())) { + reader.forEachRemaining(row -> rows.add((GenericRow) serializer.copy(row))); + } + if ("1-1".equals(value)) { + assertThat(rows).containsExactly(row(1L, "20250810", "21")); + } else if (withSnapshot && !skipPreload) { + assertThat(rows).containsExactly(row(1L, "20250810", "20")); + } else { + assertThat(rows).isEmpty(); + } + } + } + + private List readProjectedSplits(ReadBuilder readBuilder, List splits) + throws Exception { + InternalRowSerializer serializer = new InternalRowSerializer(readBuilder.readType()); + List rows = new ArrayList<>(); + try (RecordReader reader = readBuilder.newRead().createReader(splits)) { + reader.forEachRemaining(row -> rows.add((GenericRow) serializer.copy(row))); + } + return rows; + } + private static final Map QUERIED_PARTITION = ImmutableMap.of("dt", "20250810", "hour", "21"); private void createChainTableWithQueryAuth() throws Exception { + createChainTableWithQueryAuth(true); + } + + private void createChainTableWithQueryAuth(boolean withSnapshot) throws Exception { createChainTable( options -> { options.set(CoreOptions.QUERY_AUTH_ENABLED, true); @@ -999,10 +1175,12 @@ private void createChainTableWithQueryAuth() throws Exception { options.set(CoreOptions.SCAN_FALLBACK_BRANCH_READ_FAIL_FAST, true); }); FileStoreTable chainTable = loadTable(); - writeWithCommit( - chainTable.switchToBranch(SNAPSHOT_BRANCH), - row(1L, 1L, "1", "CN", "20250810", "20"), - row(2L, 1L, "2", "CN", "20250810", "20")); + if (withSnapshot) { + writeWithCommit( + chainTable.switchToBranch(SNAPSHOT_BRANCH), + row(1L, 1L, "1", "CN", "20250810", "20"), + row(2L, 1L, "2", "CN", "20250810", "20")); + } writeWithCommit( chainTable.switchToBranch(DELTA_BRANCH), row(1L, 2L, "1-1", "CN", "20250810", "21"), @@ -1026,11 +1204,16 @@ private TableQueryAuthResult maskValueAndFilterKey() { * circuits to "no rules", so no split is ever wrapped. */ private CatalogEnvironment queryAuthEnvironment(Supplier authResult) { + return queryAuthEnvironment(select -> authResult.get()); + } + + private CatalogEnvironment queryAuthEnvironment( + Function, TableQueryAuthResult> authResult) { Catalog catalog = Mockito.mock(Catalog.class); try { // a fresh instance per call, as RESTCatalog#authTableQuery builds one per scan Mockito.when(catalog.authTableQuery(Mockito.any(), Mockito.any())) - .thenAnswer(invocation -> authResult.get()); + .thenAnswer(invocation -> authResult.apply(invocation.getArgument(1))); // the snapshot manager takes the same catalog; let it fall back to the file system // instead of answering "no snapshot" for every branch Mockito.when(catalog.loadSnapshot(Mockito.any(Identifier.class))) diff --git a/paimon-core/src/test/java/org/apache/paimon/table/FallbackReadFileStoreTableTest.java b/paimon-core/src/test/java/org/apache/paimon/table/FallbackReadFileStoreTableTest.java index 8c93914299b6..6b8727e0c6dc 100644 --- a/paimon-core/src/test/java/org/apache/paimon/table/FallbackReadFileStoreTableTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/table/FallbackReadFileStoreTableTest.java @@ -45,6 +45,7 @@ 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.ReadBuilder; import org.apache.paimon.table.source.Split; import org.apache.paimon.table.source.TableRead; import org.apache.paimon.types.DataType; @@ -59,12 +60,14 @@ import org.junit.jupiter.api.Test; import org.junit.jupiter.api.io.TempDir; import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; import org.junit.jupiter.params.provider.ValueSource; import org.mockito.AdditionalAnswers; import org.mockito.Mockito; import java.io.IOException; import java.util.ArrayList; +import java.util.Arrays; import java.util.Collections; import java.util.List; import java.util.UUID; @@ -119,6 +122,179 @@ public void testScanForwardsReadType() { Mockito.verify(fallbackScan).withReadType(readType); } + @ParameterizedTest + @CsvSource({ + "scan.fallback-branch,pt,false", + "scan.primary-branch,pt,false", + "scan.fallback-branch,a,false", + "scan.primary-branch,a,false", + "scan.fallback-branch,pt,true", + "scan.primary-branch,pt,true", + "scan.fallback-branch,a,true", + "scan.primary-branch,a,true" + }) + void testQueryAuthWithProjectedColumns( + String branchOption, String projectedColumn, boolean dataEvolution) throws Exception { + FileStoreTable localTable = createTable(dataEvolution); + localTable.createBranch("backup"); + writeDataIntoTable(localTable, 0, rowData(1, 10), rowData(3, 50)); + writeDataIntoTable( + createTableFromBranch(localTable, "backup"), + 0, + rowData(1, 20), + rowData(1, 30), + rowData(2, 40)); + + List allowedColumns = new ArrayList<>(Collections.singletonList(projectedColumn)); + Catalog catalog = Mockito.mock(Catalog.class); + Mockito.when(catalog.authTableQuery(Mockito.any(), Mockito.any())) + .thenAnswer( + invocation -> { + List select = invocation.getArgument(1); + if (select == null || !allowedColumns.containsAll(select)) { + throw new SecurityException( + "No permission to read unselected columns"); + } + return new TableQueryAuthResult(null, null); + }); + Mockito.when(catalog.loadSnapshot(Mockito.any(Identifier.class))) + .thenThrow(new UnsupportedOperationException()); + CatalogEnvironment environment = + new CatalogEnvironment( + Identifier.create("mydb", "mytable"), + null, + () -> catalog, + null, + null, + null, + false, + false); + Options options = new Options(); + options.set(CoreOptions.QUERY_AUTH_ENABLED, true); + options.set(CoreOptions.SCAN_FALLBACK_BRANCH_READ_FAIL_FAST, true); + options.set(branchOption, "backup"); + FileStoreTable table = + FileStoreTableFactory.create( + fileIO, tablePath, localTable.schema(), options, environment); + ReadBuilder readBuilder = + table.newReadBuilder().withReadType(table.rowType().project(projectedColumn)); + List rows = new ArrayList<>(); + try (RecordReader reader = + readBuilder.newRead().createReader(readBuilder.newScan().plan().splits())) { + reader.forEachRemaining(row -> rows.add(row.getInt(0))); + } + List expected; + if ("pt".equals(projectedColumn)) { + expected = + "scan.primary-branch".equals(branchOption) + ? Arrays.asList(1, 1, 2, 3) + : Arrays.asList(1, 2, 3); + } else { + expected = + "scan.primary-branch".equals(branchOption) + ? Arrays.asList(20, 30, 40, 50) + : Arrays.asList(10, 40, 50); + } + assertThat(rows).containsExactlyInAnyOrderElementsOf(expected); + assertThat(readBuilder.newScan().listPartitionEntries()).hasSize(3); + + String deniedColumn = "pt".equals(projectedColumn) ? "a" : "pt"; + assertThatThrownBy( + () -> + table.newReadBuilder() + .withReadType(table.rowType().project(deniedColumn)) + .newScan() + .plan()) + .hasMessageContaining("No permission"); + readBuilder.withFilter( + new PredicateBuilder(table.rowType()) + .equal(table.rowType().getFieldIndex(deniedColumn), 1)); + assertThatThrownBy(() -> readBuilder.newScan().plan()) + .hasMessageContaining("No permission"); + assertThatThrownBy(() -> readBuilder.newScan().listPartitionEntries()) + .hasMessageContaining("No permission"); + if ("a".equals(projectedColumn)) { + DataTableScan internalScan = table.newScan(); + internalScan.withReadType(table.rowType().project("a")); + internalScan.withInternalPartitionFilter( + PartitionPredicate.fromPredicate( + table.schema().logicalPartitionType(), + new PredicateBuilder(table.schema().logicalPartitionType()) + .equal(0, 2))); + assertThat(internalScan.listPartitions()) + .extracting(partition -> partition.getInt(0)) + .containsExactly(2); + List internalRows = new ArrayList<>(); + try (RecordReader reader = + table.newReadBuilder() + .withReadType(table.rowType().project("a")) + .newRead() + .createReader(internalScan.plan().splits())) { + reader.forEachRemaining(row -> internalRows.add(row.getInt(0))); + } + assertThat(internalRows).containsExactly(40); + + ReadBuilder partitionFiltered = + table.newReadBuilder() + .withReadType(table.rowType().project("a")) + .withPartitionFilter(Collections.singletonMap("pt", "1")); + assertThatThrownBy(() -> partitionFiltered.newScan().plan()) + .hasMessageContaining("No permission"); + assertThatThrownBy(() -> partitionFiltered.newScan().listPartitionEntries()) + .hasMessageContaining("No permission"); + + // Preserve user operands in both the underlying scan and newly created wrapper scans. + FileStoreTable underlyingTable = + FileStoreTableFactory.createWithoutFallbackBranch( + fileIO, tablePath, localTable.schema(), options, environment); + for (DataTableScan scan : Arrays.asList(underlyingTable.newScan(), table.newScan())) { + scan.withReadType(table.rowType().project("a")); + scan.withPartitionFilter(Collections.singletonMap("pt", "1")); + scan.withInternalPartitionFilter(PartitionPredicate.ALWAYS_TRUE); + assertThatThrownBy(scan::plan).hasMessageContaining("No permission"); + assertThatThrownBy(scan::listPartitionEntries) + .hasMessageContaining("No permission"); + + // Unlike internal planning, a new user filter replaces the old user operands. + scan.withPartitionFilter(PartitionPredicate.ALWAYS_TRUE); + assertThat(scan.plan().splits()).isNotEmpty(); + assertThat(scan.listPartitionEntries()).isNotEmpty(); + scan.withPartitionFilter( + PartitionPredicate.fromPredicate( + table.schema().logicalPartitionType(), + new PredicateBuilder(table.schema().logicalPartitionType()) + .equal(0, 1))); + assertThatThrownBy(scan::plan).hasMessageContaining("No permission"); + assertThatThrownBy(scan::listPartitionEntries) + .hasMessageContaining("No permission"); + } + } + + if ("pt".equals(projectedColumn)) { + allowedColumns.add("a"); + // A value filter must be authorized without affecting which branch owns a partition. + // In fallback mode, a = 20 in backup must not replace a = 10 in main's pt = 1. + ReadBuilder valueFiltered = + table.newReadBuilder() + .withReadType(table.rowType().project("pt")) + .withFilter(new PredicateBuilder(table.rowType()).equal(1, 20)); + assertThat(valueFiltered.newScan().listPartitionEntries()).hasSize(3); + List filteredRows = new ArrayList<>(); + try (RecordReader reader = + valueFiltered + .newRead() + .executeFilter() + .createReader(valueFiltered.newScan().plan().splits())) { + reader.forEachRemaining(row -> filteredRows.add(row.getInt(0))); + } + assertThat(filteredRows) + .containsExactlyElementsOf( + "scan.primary-branch".equals(branchOption) + ? Collections.singletonList(1) + : Collections.emptyList()); + } + } + @Test public void testPlanAndReadWithQueryAuthSplit() throws Exception { FileStoreTable mainTable = createTable(); @@ -611,6 +787,13 @@ private void writeDataIntoTable( } private AppendOnlyFileStoreTable createTable() throws Exception { + return createTable(false); + } + + private AppendOnlyFileStoreTable createTable(boolean dataEvolution) throws Exception { + Options options = new Options(); + options.set(CoreOptions.DATA_EVOLUTION_ENABLED, dataEvolution); + options.set(CoreOptions.ROW_TRACKING_ENABLED, dataEvolution); TableSchema tableSchema = SchemaUtils.forceCommit( new FileSystemSchemaManager(LocalFileIO.create(), tablePath), @@ -618,7 +801,7 @@ private AppendOnlyFileStoreTable createTable() throws Exception { ROW_TYPE.getFields(), Collections.singletonList("pt"), Collections.emptyList(), - Collections.emptyMap(), + options.toMap(), "")); return new AppendOnlyFileStoreTable(fileIO, tablePath, tableSchema); }