This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 0952d147bd4edf190af6cda35b48bfe43fc57fea Author: Jiajia Li <[email protected]> AuthorDate: Fri Sep 18 12:02:30 2026 +0800 [core] Preserve query authorization when narrowing chain table splits (#9929) --- .../paimon/catalog/TableQueryAuthResult.java | 33 +++ .../apache/paimon/table/ChainGroupReadTable.java | 93 ++++-- .../paimon/table/ChainTableFileStoreTable.java | 5 +- .../apache/paimon/table/ChainTableStreamScan.java | 49 +++- .../paimon/table/source/AbstractDataTableScan.java | 22 +- .../apache/paimon/table/source/DataTableScan.java | 12 + .../apache/paimon/table/source/QueryAuthSplit.java | 10 + .../paimon/catalog/TableQueryAuthResultTest.java | 85 ++++++ .../paimon/table/ChainTableFileStoreTableTest.java | 314 ++++++++++++++++++++- 9 files changed, 585 insertions(+), 38 deletions(-) diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/TableQueryAuthResult.java b/paimon-core/src/main/java/org/apache/paimon/catalog/TableQueryAuthResult.java index 6ab47711dd..cf3bfdf049 100644 --- a/paimon-core/src/main/java/org/apache/paimon/catalog/TableQueryAuthResult.java +++ b/paimon-core/src/main/java/org/apache/paimon/catalog/TableQueryAuthResult.java @@ -55,6 +55,7 @@ import java.util.HashMap; import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.Objects; import java.util.Optional; import java.util.Set; import java.util.TreeMap; @@ -95,6 +96,38 @@ public class TableQueryAuthResult implements Serializable { return extractPredicate() != null || !extractColumnMasking().isEmpty(); } + /** + * A catalog builds a fresh instance per authorization call and deserialization builds another, + * so identity says nothing about whether two results agree. + * + * <p>Compared as the rules parse out rather than as they arrive, the filter as the set of its + * conjuncts, so neither the JSON text nor the order the catalog lists them in matters. + */ + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + if (!(o instanceof TableQueryAuthResult)) { + return false; + } + TableQueryAuthResult that = (TableQueryAuthResult) o; + return parsedFilterConjuncts().equals(that.parsedFilterConjuncts()) + && extractColumnMasking().equals(that.extractColumnMasking()); + } + + @Override + public int hashCode() { + return Objects.hash(parsedFilterConjuncts(), extractColumnMasking()); + } + + private Set<Predicate> parsedFilterConjuncts() { + Predicate predicate = extractPredicate(); + return predicate == null + ? Collections.emptySet() + : new HashSet<>(PredicateBuilder.splitAnd(predicate)); + } + /** * Rejects a search on a query-auth table. Called from the methods that produce a scan or a * read, not from the builder constructors: the builders are serializable, and deserialization 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 b0f4278666..c141d9b8e2 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 @@ -19,6 +19,7 @@ package org.apache.paimon.table; import org.apache.paimon.CoreOptions; +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,6 +37,7 @@ import org.apache.paimon.table.source.DataFilePlan; 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.TableRead; import org.apache.paimon.types.RowType; @@ -45,6 +47,8 @@ import org.apache.paimon.utils.Filter; 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; @@ -405,8 +409,11 @@ public class ChainGroupReadTable extends FallbackReadFileStoreTable { tableSchema.logicalPartitionType(), snapshotSearchPred); // 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. List<BinaryRow> snapshotPartitionsInGroup = newChainPartitionListingScan(true, snapshotAnchorPredicate) + .withoutAuthPartitionPushdown() .listPartitions(); // Find delta → snapshot mapping (for each delta partition, find the nearest @@ -452,29 +459,25 @@ public class ChainGroupReadTable extends FallbackReadFileStoreTable { deltaScan.withPartitionFilter(selectedDeltaPartitions); } - List<DataSplit> deltaSubSplits = - deltaScan.plan().splits().stream() - .map(s -> (DataSplit) s) - .collect(Collectors.toList()); + 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())); - snapshotSubSplits = - snapshotScan.plan().splits().stream() - .map(s -> (DataSplit) s) - .collect(Collectors.toList()); + snapshotSubSplits = auth.unwrap(snapshotScan.plan().splits()); } splits.addAll( - ChainTableUtils.buildChainSplits( - partitionPairs.getKey(), - snapshotSubSplits, - deltaSubSplits, - options.scanFallbackSnapshotBranch(), - options.scanFallbackDeltaBranch(), - keyComparator, - targetSplitSize, - openFileCost)); + auth.reapply( + ChainTableUtils.buildChainSplits( + partitionPairs.getKey(), + snapshotSubSplits, + deltaSubSplits, + options.scanFallbackSnapshotBranch(), + options.scanFallbackDeltaBranch(), + keyComparator, + targetSplitSize, + openFileCost))); } } } @@ -518,10 +521,12 @@ public class ChainGroupReadTable extends FallbackReadFileStoreTable { return snapshotPartitions; } - for (Split split : mainScan.plan().splits()) { - DataSplit dataSplit = (DataSplit) split; - splits.add(ChainSplit.from(dataSplit, options.scanFallbackSnapshotBranch())); + ChainAuth auth = new ChainAuth(); + List<Split> chainSplits = new ArrayList<>(); + for (DataSplit dataSplit : auth.unwrap(mainScan.plan().splits())) { + chainSplits.add(ChainSplit.from(dataSplit, options.scanFallbackSnapshotBranch())); } + splits.addAll(auth.reapply(chainSplits)); snapshotPartitions.addAll( newChainPartitionListingScan(true, getMainPartitionPredicate()) @@ -531,9 +536,10 @@ public class ChainGroupReadTable extends FallbackReadFileStoreTable { private DataTableScan newFilteredScan(boolean snapshot) { DataTableScan scan = - snapshot - ? chainGroupReadTable.newSnapshotScan(scanCreator) - : chainGroupReadTable.newDeltaScan(scanCreator); + (snapshot + ? chainGroupReadTable.newSnapshotScan(scanCreator) + : chainGroupReadTable.newDeltaScan(scanCreator)) + .withoutAuthPartitionPushdown(); if (dataPredicate != null) { scan.withFilter(dataPredicate); } @@ -603,11 +609,48 @@ public class ChainGroupReadTable extends FallbackReadFileStoreTable { @Override public RecordReader<InternalRow> createReader(Split split) throws IOException { - if (split instanceof ChainSplit || split instanceof DataSplit) { + // fallbackRead unwraps and applies the rules itself, so it gets the wrapper untouched. + Split inner = QueryAuthSplit.unwrap(split); + if (inner instanceof ChainSplit || inner instanceof DataSplit) { return fallbackRead.createReader(split); } throw new IllegalArgumentException( - "Unsupported split type for chain table read: " + split.getClass().getName()); + "Unsupported split type for chain table read: " + inner.getClass().getName()); + } + } + + /** + * Carries the branch scans' authorization across chain split building; dropping it would turn + * the failed cast into unrestricted reads. Both branches authorize the same logical table, so + * one result covers the splits built from either side. + */ + private static final class ChainAuth { + + @Nullable private TableQueryAuthResult authResult; + + List<DataSplit> unwrap(List<Split> splits) { + List<DataSplit> dataSplits = new ArrayList<>(splits.size()); + for (Split split : splits) { + if (split instanceof QueryAuthSplit) { + TableQueryAuthResult next = ((QueryAuthSplit) split).authResult(); + // Enforced rather than assumed: a divergence would apply one branch's rules to + // the other's rows. + checkArgument( + authResult == null || authResult.equals(next), + "Branch scans of the same chain table returned different authorization rules."); + authResult = next; + } + dataSplits.add((DataSplit) QueryAuthSplit.unwrap(split)); + } + return dataSplits; + } + + List<Split> reapply(List<? extends Split> splits) { + List<Split> result = new ArrayList<>(splits.size()); + for (Split split : splits) { + result.add(authResult == null ? split : new QueryAuthSplit(split, authResult)); + } + return result; } } } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/ChainTableFileStoreTable.java b/paimon-core/src/main/java/org/apache/paimon/table/ChainTableFileStoreTable.java index ba8f8bf48c..8daed52edc 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/ChainTableFileStoreTable.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/ChainTableFileStoreTable.java @@ -29,6 +29,7 @@ import org.apache.paimon.schema.TableSchema; 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; @@ -217,7 +218,9 @@ public class ChainTableFileStoreTable extends FallbackReadFileStoreTable { // FallbackSplit (including FallbackDataSplit): use inherited fallback read logic return fallbackRead.createReader(split); } - if (split instanceof ChainSplit || split instanceof DataSplit) { + // Route on what the wrapper carries, so the wrapper still reaches the read. + Split inner = QueryAuthSplit.unwrap(split); + if (inner instanceof ChainSplit || inner instanceof DataSplit) { return chainGroupRead.createReader(split); } // Other split types: use inherited fallback read logic 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 5812a2611b..e9f181f47f 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 @@ -20,6 +20,7 @@ package org.apache.paimon.table; import org.apache.paimon.CoreOptions; import org.apache.paimon.Snapshot; +import org.apache.paimon.catalog.TableQueryAuthResult; import org.apache.paimon.codegen.CodeGenUtils; import org.apache.paimon.codegen.RecordComparator; import org.apache.paimon.data.BinaryRow; @@ -33,6 +34,7 @@ import org.apache.paimon.table.source.DataSplit; import org.apache.paimon.table.source.DataTableScan; import org.apache.paimon.table.source.DataTableStreamScan; import org.apache.paimon.table.source.InnerTableScan; +import org.apache.paimon.table.source.QueryAuthSplit; import org.apache.paimon.table.source.SnapshotNotExistPlan; import org.apache.paimon.table.source.Split; import org.apache.paimon.table.source.StreamDataTableScan; @@ -57,6 +59,8 @@ import java.util.Map; import java.util.Objects; import java.util.Set; +import static org.apache.paimon.utils.Preconditions.checkArgument; + /** * Streaming scan for chain tables with a two-phase design: * @@ -104,6 +108,9 @@ public class ChainTableStreamScan implements StreamDataTableScan { /** Whether the starting plan (Phase 1) has been completed. */ private boolean startingDone = false; + /** Authorization removed while grouping the starting splits; re-applied before returning. */ + @Nullable private TableQueryAuthResult startingAuthResult; + /** * Predicates, shard, and bucket filter for applying to local scans in {@link #planStarting()}. */ @@ -190,6 +197,9 @@ public class ChainTableStreamScan implements StreamDataTableScan { * allowing streaming readers to see cross-branch deletions and updates. */ private TableScan.Plan planStarting() { + // restore() can re-run this on the same instance; a stale result would re-apply rules + // revoked in between. + startingAuthResult = null; FileStoreTable deltaTable = chainGroupReadTable.other(); String deltaBranch = deltaTable.coreOptions().branch(); String snapshotBranch = chainGroupReadTable.wrapped.coreOptions().branch(); @@ -241,7 +251,7 @@ public class ChainTableStreamScan implements StreamDataTableScan { Map<BinaryRow, List<DataSplit>> deltaSplitsByPartition; if (deltaLatestId != null) { FileStoreTable pinnedDelta = deltaTable.copy(pinnedOptions(deltaLatestId)); - DataTableScan pinnedDeltaScan = pinnedDelta.newScan(); + DataTableScan pinnedDeltaScan = pinnedDelta.newScan().withoutAuthPartitionPushdown(); applyPredicatesShardAndBucket(pinnedDeltaScan); deltaSplitsByPartition = groupByPartition(pinnedDeltaScan); } else { @@ -256,7 +266,8 @@ public class ChainTableStreamScan implements StreamDataTableScan { FileStoreTable pinnedSnapshot = null; if (snapshotLatestId != null) { pinnedSnapshot = chainGroupReadTable.wrapped.copy(pinnedOptions(snapshotLatestId)); - DataTableScan partitionListingScan = pinnedSnapshot.newScan(); + DataTableScan partitionListingScan = + pinnedSnapshot.newScan().withoutAuthPartitionPushdown(); for (BinaryRow partition : partitionListingScan.listPartitions()) { Object groupKey = toGroupKey(partition); BinaryRow existingLatest = latestChainPartitionPerGroup.get(groupKey); @@ -275,7 +286,7 @@ public class ChainTableStreamScan implements StreamDataTableScan { List<BinaryRow> latestPartitions = new ArrayList<>(latestChainPartitionPerGroup.values()); Map<BinaryRow, List<DataSplit>> snapshotSplitsByPartition; if (!latestPartitions.isEmpty() && pinnedSnapshot != null) { - DataTableScan snapshotScan = pinnedSnapshot.newScan(); + DataTableScan snapshotScan = pinnedSnapshot.newScan().withoutAuthPartitionPushdown(); snapshotScan.withPartitionFilter(latestPartitions); applyPredicatesShardAndBucket(snapshotScan); snapshotSplitsByPartition = groupByPartition(snapshotScan); @@ -316,7 +327,7 @@ public class ChainTableStreamScan implements StreamDataTableScan { allSplits.size()); startingDone = true; - return new DataFilePlan<>(allSplits); + return new DataFilePlan<>(reapplyStartingAuth(allSplits)); } /** @@ -476,16 +487,42 @@ public class ChainTableStreamScan implements StreamDataTableScan { return latestId; } - /** Plans a scan and groups the resulting splits by partition. */ - private static Map<BinaryRow, List<DataSplit>> groupByPartition(DataTableScan scan) { + /** + * Authorization is stripped here and kept for {@link #reapplyStartingAuth}; dropping it would + * silently discard the rules. + */ + private Map<BinaryRow, List<DataSplit>> groupByPartition(DataTableScan scan) { Map<BinaryRow, List<DataSplit>> grouped = new LinkedHashMap<>(); for (Split s : scan.plan().splits()) { + if (s instanceof QueryAuthSplit) { + QueryAuthSplit authSplit = (QueryAuthSplit) s; + TableQueryAuthResult next = authSplit.authResult(); + // The delta and the snapshot scan authorize separately. Replacing one result with + // the other would widen what the reader sees. + checkArgument( + startingAuthResult == null || startingAuthResult.equals(next), + "Branch scans of the same chain table returned different authorization rules."); + startingAuthResult = next; + s = authSplit.split(); + } DataSplit ds = (DataSplit) s; grouped.computeIfAbsent(ds.partition(), k -> new ArrayList<>()).add(ds); } return grouped; } + /** Puts back the authorization removed by {@link #groupByPartition}. */ + private List<Split> reapplyStartingAuth(List<Split> splits) { + if (startingAuthResult == null) { + return splits; + } + List<Split> wrapped = new ArrayList<>(splits.size()); + for (Split split : splits) { + wrapped.add(new QueryAuthSplit(split, startingAuthResult)); + } + return wrapped; + } + /** * Extracts a stable group key from a full partition row. When there is no group partition (all * fields are chain keys), returns a shared singleton to avoid zero-field {@link BinaryRow} 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 8be73caff6..9cea4edfe4 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 @@ -103,6 +103,9 @@ abstract class AbstractDataTableScan implements DataTableScan { // Whether the auth predicate has a non-partition part (enforced only at read time). Used by // AbstractBatchTableScan to disable limit push down; not pushed through withFilter. protected boolean authHasNonPartitionFilter; + private boolean authPartitionPushdown = true; + // Turning the pushdown off after a filter was pushed has to rebuild it, rules unchanged. + private boolean appliedAuthPartitionPushdown = true; // auth state, refreshed each plan(). The filter is pushed once, without the conjuncts on // masked columns, whose raw statistics a mask invalidates; the partition fields are // replaced on each push, as ManifestsReader#withPartitionFilter overwrites rather than ands. @@ -160,10 +163,12 @@ abstract class AbstractDataTableScan implements DataTableScan { protected abstract TableScan.Plan planWithoutAuth(); private void applyAuthFilter(@Nullable Predicate authPredicate) { - if (Objects.equals(authPredicate, appliedAuthPredicate)) { + if (Objects.equals(authPredicate, appliedAuthPredicate) + && authPartitionPushdown == appliedAuthPartitionPushdown) { return; } appliedAuthPredicate = authPredicate; + appliedAuthPartitionPushdown = authPartitionPushdown; PartitionPredicate authPartitionFilter = null; boolean hasNonPartitionPart = false; @@ -183,11 +188,20 @@ abstract class AbstractDataTableScan implements DataTableScan { // Push only the partition part, to a dedicated slot overwritten/cleared each plan() so a // changed/removed auth leaves no stale pruning. The full filter is enforced at read time. - snapshotReader.manifestsReader().withAuthPartitionFilter(authPartitionFilter); + snapshotReader + .manifestsReader() + .withAuthPartitionFilter(authPartitionPushdown ? authPartitionFilter : null); // A non-partition auth part is enforced only at read time, so limit push down is unsafe // (AbstractBatchTableScan reads this). Kept off SnapshotReader since it is not a pushed - // filter. - this.authHasNonPartitionFilter = hasNonPartitionPart; + // filter. Without the partition push down the whole rule is read time. + this.authHasNonPartitionFilter = + authPartitionPushdown ? hasNonPartitionPart : authPredicate != null; + } + + @Override + public AbstractDataTableScan withoutAuthPartitionPushdown() { + this.authPartitionPushdown = false; + return this; } @Override diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/DataTableScan.java b/paimon-core/src/main/java/org/apache/paimon/table/source/DataTableScan.java index 29c825b0be..173e21e34c 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/DataTableScan.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/DataTableScan.java @@ -23,4 +23,16 @@ public interface DataTableScan extends InnerTableScan { /** Specify the shard to be read, and allocate sharded files to read records. */ DataTableScan withShard(int indexOfThisSubtask, int numberOfParallelSubtasks); + + /** + * Stops the partition part of an authorization rule from pruning here, for a scan whose + * physical partitions are not the ones the rule speaks about — a chain table maps a branch + * partition onto a different logical one. + * + * <p>Only for a scan whose splits are then read, which applies the whole rule anyway. {@link + * #listPartitionEntries} answers the caller directly and must keep the pruning. + */ + default DataTableScan withoutAuthPartitionPushdown() { + return this; + } } diff --git a/paimon-core/src/main/java/org/apache/paimon/table/source/QueryAuthSplit.java b/paimon-core/src/main/java/org/apache/paimon/table/source/QueryAuthSplit.java index 3f6106d015..f38fbc500f 100644 --- a/paimon-core/src/main/java/org/apache/paimon/table/source/QueryAuthSplit.java +++ b/paimon-core/src/main/java/org/apache/paimon/table/source/QueryAuthSplit.java @@ -54,6 +54,16 @@ public class QueryAuthSplit implements Split { return split; } + /** + * The split this wrapper carries, or {@code split} itself when it is not one. Narrowing a + * planned split to a concrete type must go through this, or the cast fails once the table + * carries a rule. A caller handing the split to a read passes the wrapper on instead, or the + * authorization never reaches the reader. + */ + public static Split unwrap(Split split) { + return split instanceof QueryAuthSplit ? ((QueryAuthSplit) split).split() : split; + } + @Nullable public TableQueryAuthResult authResult() { return authResult; diff --git a/paimon-core/src/test/java/org/apache/paimon/catalog/TableQueryAuthResultTest.java b/paimon-core/src/test/java/org/apache/paimon/catalog/TableQueryAuthResultTest.java index 5b7f64fe8a..6ff67de361 100644 --- a/paimon-core/src/test/java/org/apache/paimon/catalog/TableQueryAuthResultTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/catalog/TableQueryAuthResultTest.java @@ -24,7 +24,9 @@ import org.apache.paimon.predicate.Equal; import org.apache.paimon.predicate.FieldRef; import org.apache.paimon.predicate.FieldTransform; import org.apache.paimon.predicate.LeafPredicate; +import org.apache.paimon.predicate.LowerTransform; import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.predicate.UpperTransform; import org.apache.paimon.types.DataTypes; import org.apache.paimon.types.RowType; import org.apache.paimon.utils.JsonSerdeUtil; @@ -162,4 +164,87 @@ public class TableQueryAuthResultTest { Map<String, String> masking = Collections.singletonMap("display", maskJson()); assertThat(new TableQueryAuthResult(null, masking).hasRules()).isTrue(); } + + /** + * Chain table planning aborts a query whose branches disagree, so a difference that is not a + * difference in the rules would fail a query it should have served. + */ + @Test + public void testEqualsComparesRulesNotTheTransportShapeTheyArriveIn() { + Map<String, String> masking = Collections.singletonMap("display", maskJson()); + + // the same conjuncts listed in the other order + assertThat( + new TableQueryAuthResult( + Arrays.asList(filterJson(), otherFilterJson()), masking)) + .isEqualTo( + new TableQueryAuthResult( + Arrays.asList(otherFilterJson(), filterJson()), masking)) + .hasSameHashCodeAs( + new TableQueryAuthResult( + Arrays.asList(otherFilterJson(), filterJson()), masking)); + + // an absent rule and an empty one + assertThat(new TableQueryAuthResult(Collections.singletonList(filterJson()), null)) + .isEqualTo( + new TableQueryAuthResult( + Collections.singletonList(filterJson()), Collections.emptyMap())); + assertThat(new TableQueryAuthResult(null, masking)) + .isEqualTo(new TableQueryAuthResult(Collections.emptyList(), masking)); + + assertThat(new TableQueryAuthResult(null, masking)) + .isEqualTo( + new TableQueryAuthResult( + null, Collections.singletonMap("display", maskJson()))); + + // JSON spaced out by a different serializer + assertThat(new TableQueryAuthResult(Collections.singletonList(filterJson()), masking)) + .isEqualTo( + new TableQueryAuthResult( + Collections.singletonList(spacedOut(filterJson())), + Collections.singletonMap("display", spacedOut(maskJson())))); + + // same shape, same column, still two different masks + assertThat(new TableQueryAuthResult(null, Collections.singletonMap("display", upperJson()))) + .isNotEqualTo( + new TableQueryAuthResult( + null, Collections.singletonMap("display", lowerJson()))); + + // rules that really do differ + assertThat(new TableQueryAuthResult(Collections.singletonList(filterJson()), null)) + .isNotEqualTo( + new TableQueryAuthResult( + Collections.singletonList(otherFilterJson()), null)) + .isNotEqualTo(new TableQueryAuthResult(null, null)); + assertThat(new TableQueryAuthResult(null, masking)) + .isNotEqualTo(new TableQueryAuthResult(null, null)) + .isNotEqualTo( + new TableQueryAuthResult( + null, Collections.singletonMap("other", maskJson()))); + } + + /** The same JSON, with another serializer's whitespace. */ + private static String spacedOut(String json) { + return json.replace(",", ", ").replace(":", ": "); + } + + private static String upperJson() { + return JsonSerdeUtil.toFlatJson( + new UpperTransform( + Collections.singletonList(new FieldRef(1, "extra", DataTypes.STRING())))); + } + + private static String lowerJson() { + return JsonSerdeUtil.toFlatJson( + new LowerTransform( + Collections.singletonList(new FieldRef(1, "extra", DataTypes.STRING())))); + } + + private static String otherFilterJson() { + return JsonSerdeUtil.toFlatJson( + LeafPredicate.of( + new FieldTransform(new FieldRef(1, "extra", DataTypes.STRING())), + Equal.INSTANCE, + Collections.singletonList(BinaryString.fromString("y")))); + } } 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 66b82195f0..b65b2e6ec0 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 @@ -19,6 +19,9 @@ package org.apache.paimon.table; import org.apache.paimon.CoreOptions; +import org.apache.paimon.catalog.Catalog; +import org.apache.paimon.catalog.Identifier; +import org.apache.paimon.catalog.TableQueryAuthResult; import org.apache.paimon.data.BinaryString; import org.apache.paimon.data.GenericRow; import org.apache.paimon.data.InternalRow; @@ -27,6 +30,11 @@ import org.apache.paimon.disk.IOManagerImpl; import org.apache.paimon.fs.Path; import org.apache.paimon.fs.local.LocalFileIO; import org.apache.paimon.options.Options; +import org.apache.paimon.predicate.FieldRef; +import org.apache.paimon.predicate.FieldTransform; +import org.apache.paimon.predicate.Predicate; +import org.apache.paimon.predicate.PredicateBuilder; +import org.apache.paimon.predicate.Transform; import org.apache.paimon.reader.RecordReader; import org.apache.paimon.schema.FileSystemSchemaManager; import org.apache.paimon.schema.Schema; @@ -37,13 +45,17 @@ import org.apache.paimon.table.sink.CommitMessage; import org.apache.paimon.table.sink.InnerTableWrite; import org.apache.paimon.table.sink.StreamTableCommit; import org.apache.paimon.table.source.ChainSplit; +import org.apache.paimon.table.source.DataTableScan; import org.apache.paimon.table.source.DeletionFile; +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.TableScan; import org.apache.paimon.types.DataTypes; import org.apache.paimon.types.RowKind; import org.apache.paimon.types.RowType; +import org.apache.paimon.utils.JsonSerdeUtil; import org.apache.paimon.shade.guava30.com.google.common.collect.ImmutableMap; @@ -53,6 +65,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.MethodSource; +import org.mockito.Mockito; import java.util.ArrayList; import java.util.Arrays; @@ -63,6 +76,8 @@ import java.util.Objects; import java.util.Optional; import java.util.UUID; import java.util.function.Consumer; +import java.util.function.Supplier; +import java.util.stream.Collectors; import java.util.stream.Stream; import static org.apache.paimon.CoreOptions.BUCKET; @@ -350,7 +365,303 @@ public class ChainTableFileStoreTableTest { assertThat(foundChainSplit).as("Should have found at least one ChainSplit").isTrue(); } + @Test + public void testQueryAuthBatchPlanKeepsChainSplitsWrapped() throws Exception { + createChainTableWithQueryAuth(); + TableQueryAuthResult authResult = maskValueAndFilterKey(); + + FileStoreTable table = loadTable(queryAuthEnvironment(this::maskValueAndFilterKey)); + List<Split> splits = planQueriedPartition(table.newScan()); + + assertChainSplitsCarryAuth(splits, authResult); + } + + @Test + public void testQueryAuthScanWithSnapshotReaderFactoryKeepsChainSplitsWrapped() + throws Exception { + createChainTableWithQueryAuth(); + TableQueryAuthResult authResult = maskValueAndFilterKey(); + + FileStoreTable table = loadTable(queryAuthEnvironment(this::maskValueAndFilterKey)); + List<Split> splits = planQueriedPartition(table.newScan(FileStoreTable::newSnapshotReader)); + + assertChainSplitsCarryAuth(splits, authResult); + } + + @Test + public void testQueryAuthRulesAreAppliedWhenReadingChainSplits() throws Exception { + createChainTableWithQueryAuth(); + + // the same query without any rule: raw values, no row dropped + assertThat(getResult(loadTable(), QUERIED_PARTITION)) + .containsExactlyInAnyOrder( + row(1L, 2L, "1-1", "CN", "20250810", "21"), + row(2L, 1L, "2", "CN", "20250810", "21"), + row(4L, 1L, "4", "CN", "20250810", "21")); + + FileStoreTable table = loadTable(queryAuthEnvironment(this::maskValueAndFilterKey)); + + // 'v' comes back as the region it is masked with, and k=4 is dropped by the row filter + assertThat(getResult(table, QUERIED_PARTITION)) + .containsExactlyInAnyOrder( + row(1L, 2L, "CN", "CN", "20250810", "21"), + row(2L, 1L, "CN", "CN", "20250810", "21")); + } + + /** + * A chain partition key names the partition a row is reported under, not the branch partition + * it is stored in: k=2 sits in the snapshot branch under hour 20, and the chain merge hands it + * to the reader under the queried hour 21. A rule on 'hour' therefore cannot prune branch + * partitions, or the row it does admit disappears. + */ + @Test + public void testQueryAuthChainPartitionRuleKeepsRowsOfAnEarlierBranchPartition() + throws Exception { + createChainTableWithQueryAuth(); + + FileStoreTable table = loadTable(queryAuthEnvironment(this::keepQueriedChainPartition)); + + assertThat(getResult(table, QUERIED_PARTITION)) + .containsExactlyInAnyOrder( + row(1L, 2L, "1-1", "CN", "20250810", "21"), + row(2L, 1L, "2", "CN", "20250810", "21"), + row(4L, 1L, "4", "CN", "20250810", "21")); + } + + /** + * The other direction: a partition the snapshot branch holds in full is read from it directly, + * branch partition and queried partition being the same row of values. A rule excluding that + * partition still has to prune it away, not hand the reader a split it filters down to nothing. + */ + @Test + public void testQueryAuthPartitionRuleStillPrunesASnapshotBranchPartition() throws Exception { + createChainTableWithQueryAuth(); + Map<String, String> snapshotPartition = ImmutableMap.of("dt", "20250810", "hour", "20"); + + assertThat(planPartition(loadTable().newScan(), snapshotPartition)).isNotEmpty(); + + // the rule admits hour 21 only, so nothing of the queried hour 20 is left to read + FileStoreTable table = loadTable(queryAuthEnvironment(this::keepQueriedChainPartition)); + assertThat(planPartition(table.newScan(), snapshotPartition)).isEmpty(); + assertThat(getResult(table, snapshotPartition)).isEmpty(); + } + + /** + * Partition listing has no read to follow it, so an excluded partition would be reported — with + * its row and file counts — to a caller that cannot read a row of it. + */ + @Test + public void testQueryAuthPartitionRulePrunesTheListedPartitions() throws Exception { + createChainTableWithQueryAuth(); + + assertThat(listedHours(loadTable())).containsExactlyInAnyOrder("20", "21"); + + FileStoreTable table = loadTable(queryAuthEnvironment(this::keepQueriedChainPartition)); + assertThat(listedHours(table)).containsExactly("21"); + } + + private List<String> listedHours(FileStoreTable table) { + return table.newScan().listPartitions().stream() + .map(partition -> partition.getString(2).toString()) + .collect(Collectors.toList()); + } + + /** + * The authorization has not changed when the pushdown is turned off, so the cheap way out of + * reapplying it would leave the pruning the caller just asked to drop. + */ + @Test + public void testQueryAuthPartitionPushdownIsDroppedAfterOneWasPushed() throws Exception { + createChainTableWithQueryAuth(); + FileStoreTable snapshotBranch = + loadBranchTable( + SNAPSHOT_BRANCH, queryAuthEnvironment(this::keepQueriedChainPartition)); + + // the rule admits hour 21 and this branch holds only hour 20, so the push prunes it away + DataTableScan scan = snapshotBranch.newScan(); + assertThat(scan.listPartitions()).isEmpty(); + + assertThat(scan.withoutAuthPartitionPushdown().plan().splits()).isNotEmpty(); + } + + /** The branch as an ordinary table. switchToBranch would build another chain table. */ + private FileStoreTable loadBranchTable(String branch, CatalogEnvironment catalogEnvironment) { + Path tablePath = new Path(tempDir.toUri().toString(), tableName); + LocalFileIO fileIO = LocalFileIO.create(); + Optional<TableSchema> schemaOpt = + new FileSystemSchemaManager(fileIO, tablePath, branch).latest(); + assertThat(schemaOpt.isPresent()).isTrue(); + Options options = new Options(schemaOpt.get().options()); + options.set(CHAIN_TABLE_ENABLED, false); + options.set(CoreOptions.BRANCH, branch); + return FileStoreTableFactory.create( + fileIO, tablePath, schemaOpt.get().copy(options.toMap()), catalogEnvironment); + } + + /** Keeps only the rows reported under the chain partition hour = '21'. */ + private TableQueryAuthResult keepQueriedChainPartition() { + RowType rowType = loadTable().schema().logicalRowType(); + Predicate rowFilter = new PredicateBuilder(rowType).equal(5, BinaryString.fromString("21")); + return new TableQueryAuthResult( + Collections.singletonList(JsonSerdeUtil.toFlatJson(rowFilter)), null); + } + + @Test + public void testQueryAuthWithoutRulesLeavesPlanAndReadUntouched() throws Exception { + createChainTableWithQueryAuth(); + List<GenericRow> withoutQueryAuth = getResult(loadTable(), QUERIED_PARTITION); + + // authorization answers with neither a row filter nor a column mask + FileStoreTable table = + loadTable(queryAuthEnvironment(() -> new TableQueryAuthResult(null, null))); + List<Split> splits = planQueriedPartition(table.newScan()); + + assertThat(splits).isNotEmpty(); + assertThat(splits) + .allSatisfy( + split -> + assertThat( + ((FallbackReadFileStoreTable.FallbackSplit) split) + .wrapped()) + .isInstanceOf(ChainSplit.class)); + assertThat(getResult(table, QUERIED_PARTITION)) + .containsExactlyInAnyOrderElementsOf(withoutQueryAuth); + } + + @Test + public void testQueryAuthStreamingStartingPlanKeepsChainSplitsWrapped() throws Exception { + createChainTableWithQueryAuth(); + TableQueryAuthResult authResult = maskValueAndFilterKey(); + + FileStoreTable table = loadTable(queryAuthEnvironment(this::maskValueAndFilterKey)); + List<Split> splits = table.newStreamScan().plan().splits(); + + assertThat(splits).isNotEmpty(); + assertThat(splits) + .allSatisfy( + split -> { + assertThat(split).isInstanceOf(QueryAuthSplit.class); + QueryAuthSplit authSplit = (QueryAuthSplit) split; + assertThat(authSplit.authResult()).isEqualTo(authResult); + assertThat(authSplit.split()).isInstanceOf(ChainSplit.class); + }); + + // phase 1 reads the latest snapshot partition plus the later delta partitions, so the + // same key shows up under both; what matters here is that both carry the rules + List<GenericRow> rows = readSplits(table, splits); + assertThat(rows).isNotEmpty(); + assertThat(rows) + .allSatisfy( + r -> { + assertThat(r.getLong(0)).isLessThan(3L); + assertThat(r.getString(2)).isEqualTo(r.getString(3)); + }); + } + + private static final Map<String, String> QUERIED_PARTITION = + ImmutableMap.of("dt", "20250810", "hour", "21"); + + private void createChainTableWithQueryAuth() throws Exception { + createChainTable( + options -> { + options.set(CoreOptions.QUERY_AUTH_ENABLED, true); + // a chain read that cannot handle its own split otherwise falls through to the + // main branch with nothing but a log line, which would hide a routing mistake + 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")); + writeWithCommit( + chainTable.switchToBranch(DELTA_BRANCH), + row(1L, 2L, "1-1", "CN", "20250810", "21"), + row(4L, 1L, "4", "CN", "20250810", "21")); + } + + /** Masks column 'v' with the value of column 'region' and keeps only the rows with k < 3. */ + private TableQueryAuthResult maskValueAndFilterKey() { + RowType rowType = loadTable().schema().logicalRowType(); + Predicate rowFilter = new PredicateBuilder(rowType).lessThan(0, 3L); + Transform mask = new FieldTransform(new FieldRef(3, "region", DataTypes.STRING())); + return new TableQueryAuthResult( + Collections.singletonList(JsonSerdeUtil.toFlatJson(rowFilter)), + Collections.singletonMap("v", JsonSerdeUtil.toFlatJson(mask))); + } + + /** + * A catalog environment whose authorization answers with {@code authResult}. The table, its + * branch scans and its reads are all real; only the catalog call fetching the rules is stubbed. + * {@link CatalogEnvironment#empty()} cannot express this: without a catalog loader it short + * circuits to "no rules", so no split is ever wrapped. + */ + private CatalogEnvironment queryAuthEnvironment(Supplier<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()); + // 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))) + .thenThrow(new UnsupportedOperationException()); + } catch (Exception e) { + throw new RuntimeException(e); + } + return new CatalogEnvironment( + Identifier.create("default", tableName), + null, + () -> catalog, + null, + null, + null, + false, + false); + } + + private List<Split> planQueriedPartition(DataTableScan scan) { + return planPartition(scan, QUERIED_PARTITION); + } + + private List<Split> planPartition(DataTableScan scan, Map<String, String> partition) { + return scan.withPartitionFilter(partition).plan().splits(); + } + + private void assertChainSplitsCarryAuth(List<Split> splits, TableQueryAuthResult authResult) { + assertThat(splits).isNotEmpty(); + assertThat(splits) + .allSatisfy( + split -> { + assertThat(split) + .isInstanceOf(FallbackReadFileStoreTable.FallbackSplit.class); + Split wrapped = + ((FallbackReadFileStoreTable.FallbackSplit) split).wrapped(); + assertThat(wrapped).isInstanceOf(QueryAuthSplit.class); + QueryAuthSplit authSplit = (QueryAuthSplit) wrapped; + assertThat(authSplit.authResult()).isEqualTo(authResult); + assertThat(authSplit.split()).isInstanceOf(ChainSplit.class); + }); + } + + private List<GenericRow> readSplits(FileStoreTable table, List<Split> splits) throws Exception { + InternalRowSerializer serializer = + new InternalRowSerializer(table.schema().logicalRowType()); + InnerTableRead read = table.newRead(); + List<GenericRow> result = new ArrayList<>(); + for (Split split : splits) { + try (RecordReader<InternalRow> reader = read.createReader(split)) { + reader.forEachRemaining(row -> result.add((GenericRow) serializer.copy(row))); + } + } + return result; + } + private FileStoreTable loadTable() { + return loadTable(CatalogEnvironment.empty()); + } + + private FileStoreTable loadTable(CatalogEnvironment catalogEnvironment) { Path tablePath = new Path(tempDir.toUri().toString(), tableName); LocalFileIO fileIO = LocalFileIO.create(); Options options = new Options(); @@ -359,8 +670,7 @@ public class ChainTableFileStoreTableTest { Optional<TableSchema> schemaOpt = new FileSystemSchemaManager(fileIO, tablePath, branchName).latest(); assertThat(schemaOpt.isPresent()).isTrue(); - return FileStoreTableFactory.create( - fileIO, tablePath, schemaOpt.get(), CatalogEnvironment.empty()); + return FileStoreTableFactory.create(fileIO, tablePath, schemaOpt.get(), catalogEnvironment); } private void createChainTable(Consumer<Options> optionCustomizer) throws Exception {
