This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 8ef3e0cde7 [core] Fallback and chain queries fail when unreferenced
columns are denied (#10725)
8ef3e0cde7 is described below
commit 8ef3e0cde70b24a7fb1b5f93d77b48f413ee38e3
Author: Ran Tao <[email protected]>
AuthorDate: Mon Oct 26 12:19:59 2026 +0800
[core] Fallback and chain queries fail when unreferenced columns are denied
(#10725)
---
.../paimon/globalindex/DataEvolutionBatchScan.java | 6 +
.../apache/paimon/table/ChainGroupReadTable.java | 51 ++++--
.../apache/paimon/table/ChainTableStreamScan.java | 22 ++-
.../paimon/table/FallbackReadFileStoreTable.java | 50 +++++-
.../paimon/table/source/AbstractDataTableScan.java | 10 ++
.../apache/paimon/table/source/InnerTableScan.java | 23 +++
.../paimon/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 4fdae682e5..ae6cbf01f8 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 class DataEvolutionBatchScan implements
DataTableScan {
return this;
}
+ @Override
+ public InnerTableScan withInternalPartitionFilter(PartitionPredicate
partitionPredicate) {
+ batchScan.withInternalPartitionFilter(partitionPredicate);
+ return this;
+ }
+
@Override
public InnerTableScan withBucketFilter(Filter<Integer> 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 5564260d8e..2a18a8fdd9 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 class ChainGroupReadTable extends
FallbackReadFileStoreTable {
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 class ChainGroupReadTable extends
FallbackReadFileStoreTable {
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 class ChainGroupReadTable extends
FallbackReadFileStoreTable {
// 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 class ChainGroupReadTable extends
FallbackReadFileStoreTable {
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 class ChainGroupReadTable extends
FallbackReadFileStoreTable {
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 class ChainGroupReadTable extends
FallbackReadFileStoreTable {
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 class ChainGroupReadTable extends
FallbackReadFileStoreTable {
@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 class ChainGroupReadTable extends
FallbackReadFileStoreTable {
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 @@ public class ChainGroupReadTable extends
FallbackReadFileStoreTable {
}
splits.addAll(auth.reapply(chainSplits));
- snapshotPartitions.addAll(
- newChainPartitionListingScan(true,
getMainPartitionPredicate())
- .listPartitions());
+
snapshotPartitions.addAll(newChainPartitionListingScan(true).listPartitions());
return snapshotPartitions;
}
@@ -554,6 +566,7 @@ public class ChainGroupReadTable extends
FallbackReadFileStoreTable {
?
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 e9f181f47f..b8d3ed96a6 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.Split;
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 @@ public class ChainTableStreamScan implements
StreamDataTableScan {
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 @@ public class ChainTableStreamScan implements
StreamDataTableScan {
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 @@ public class ChainTableStreamScan implements
StreamDataTableScan {
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 @@ public class ChainTableStreamScan implements
StreamDataTableScan {
}
/**
- * 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 2863320050..46c7cfcfc2 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.metrics.MetricRegistry;
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.Filter;
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.IOException;
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 class FallbackReadFileStoreTable extends
DelegatedFileStoreTable {
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 class FallbackReadFileStoreTable extends
DelegatedFileStoreTable {
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 class FallbackReadFileStoreTable extends
DelegatedFileStoreTable {
@Override
public FallbackReadScan withReadType(@Nullable RowType readType) {
+ this.readType = readType;
mainScan.withReadType(readType);
fallbackScan.withReadType(readType);
return this;
@@ -524,6 +532,14 @@ public class FallbackReadFileStoreTable extends
DelegatedFileStoreTable {
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 class FallbackReadFileStoreTable extends
DelegatedFileStoreTable {
.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 class FallbackReadFileStoreTable extends
DelegatedFileStoreTable {
}
protected void setPartitionPredicate(PartitionPredicate predicate) {
- this.mainPartitionPredicate = predicate;
- this.fallbackPartitionPredicate = predicate;
+ setPartitionPredicate(predicate, predicate);
}
protected void setPartitionPredicate(
@@ -646,6 +664,7 @@ public class FallbackReadFileStoreTable extends
DelegatedFileStoreTable {
PartitionPredicate fallbackPartitionPredicate) {
this.mainPartitionPredicate = mainPartitionPredicate;
this.fallbackPartitionPredicate = fallbackPartitionPredicate;
+ this.internalPartitionPredicate = null;
}
protected PartitionPredicate getMainPartitionPredicate() {
@@ -658,12 +677,35 @@ public class FallbackReadFileStoreTable extends
DelegatedFileStoreTable {
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 9cea4edfe4..4ac5cda74f 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 @@ abstract class AbstractDataTableScan implements
DataTableScan {
() -> 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 6c4b18ec9b..db4fa29f94 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 @@ public interface InnerTableScan extends TableScan {
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;
}
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 a651cfc2c9..2c9ab22db2 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.Test;
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.Optional;
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 class ChainTableFileStoreTableTest {
});
}
+ @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<GenericRow> 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<String> 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<GenericRow> rows = new ArrayList<>();
+ InternalRowSerializer serializer = new
InternalRowSerializer(readBuilder.readType());
+ try (RecordReader<InternalRow> 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<GenericRow> readProjectedSplits(ReadBuilder readBuilder,
List<Split> splits)
+ throws Exception {
+ InternalRowSerializer serializer = new
InternalRowSerializer(readBuilder.readType());
+ List<GenericRow> rows = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
readBuilder.newRead().createReader(splits)) {
+ reader.forEachRemaining(row -> rows.add((GenericRow)
serializer.copy(row)));
+ }
+ return rows;
+ }
+
private static final Map<String, String> 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 @@ public class ChainTableFileStoreTableTest {
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 @@ public class ChainTableFileStoreTableTest {
* circuits to "no rules", so no split is ever wrapped.
*/
private CatalogEnvironment
queryAuthEnvironment(Supplier<TableQueryAuthResult> authResult) {
+ return queryAuthEnvironment(select -> authResult.get());
+ }
+
+ private CatalogEnvironment queryAuthEnvironment(
+ Function<List<String>, 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 8c93914299..6b8727e0c6 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.sink.StreamTableWrite;
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.BeforeEach;
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 class FallbackReadFileStoreTableTest {
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<String> allowedColumns = new
ArrayList<>(Collections.singletonList(projectedColumn));
+ Catalog catalog = Mockito.mock(Catalog.class);
+ Mockito.when(catalog.authTableQuery(Mockito.any(), Mockito.any()))
+ .thenAnswer(
+ invocation -> {
+ List<String> 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<Integer> rows = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
+
readBuilder.newRead().createReader(readBuilder.newScan().plan().splits())) {
+ reader.forEachRemaining(row -> rows.add(row.getInt(0)));
+ }
+ List<Integer> 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<Integer> internalRows = new ArrayList<>();
+ try (RecordReader<InternalRow> 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<Integer> filteredRows = new ArrayList<>();
+ try (RecordReader<InternalRow> 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 @@ public class FallbackReadFileStoreTableTest {
}
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 @@ public class FallbackReadFileStoreTableTest {
ROW_TYPE.getFields(),
Collections.singletonList("pt"),
Collections.emptyList(),
- Collections.emptyMap(),
+ options.toMap(),
""));
return new AppendOnlyFileStoreTable(fileIO, tablePath, tableSchema);
}
