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 113b31188b [fix] Apply row filter auth during scan planning (#8447)
113b31188b is described below
commit 113b31188b3adec8fac22e7d5cb98122006eb696
Author: Jiajia Li <[email protected]>
AuthorDate: Sat Jul 4 09:32:12 2026 +0800
[fix] Apply row filter auth during scan planning (#8447)
---
.../paimon/catalog/TableQueryAuthResult.java | 12 +-
.../apache/paimon/operation/ManifestsReader.java | 30 ++-
.../paimon/table/source/AbstractDataTableScan.java | 38 ++++
.../paimon/table/source/DataTableBatchScan.java | 15 +-
.../paimon/table/source/ReadBuilderImpl.java | 112 ++++++++++-
.../org/apache/paimon/rest/RESTCatalogTest.java | 208 ++++++++++++++++++-
.../org/apache/paimon/flink/RESTCatalogITCase.java | 43 ++++
.../paimon/spark/SparkCatalogWithRestTest.java | 220 +++++++++++++++++++++
8 files changed, 666 insertions(+), 12 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 b0bb9b1d52..0524eea1b0 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
@@ -106,6 +106,16 @@ public class TableQueryAuthResult implements Serializable {
return rowFilter;
}
+ /**
+ * Remap the auth predicate's field references by name to positional
indices of {@code rowType}.
+ * The auth server sends field-id-based {@link
org.apache.paimon.predicate.FieldRef}s, which
+ * must be resolved by name before positional evaluation.
+ */
+ @Nullable
+ public static Predicate remapPredicate(Predicate predicate, RowType
rowType) {
+ return predicate.visit(new PredicateRemapper(rowType));
+ }
+
public Map<String, Transform> extractColumnMasking() {
Map<String, Transform> result = new TreeMap<>();
if (columnMasking != null && !columnMasking.isEmpty()) {
@@ -129,7 +139,7 @@ public class TableQueryAuthResult implements Serializable {
RecordReader<InternalRow> reader, RowType outputRowType) {
Predicate rowFilter = extractPredicate();
if (rowFilter != null) {
- Predicate remappedFilter = rowFilter.visit(new
PredicateRemapper(outputRowType));
+ Predicate remappedFilter = remapPredicate(rowFilter,
outputRowType);
if (remappedFilter != null) {
reader = reader.filter(remappedFilter::test);
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestsReader.java
b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestsReader.java
index e5727cc0c8..46bbbd17f6 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/ManifestsReader.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/ManifestsReader.java
@@ -34,6 +34,7 @@ import org.apache.paimon.utils.SnapshotManager;
import javax.annotation.Nullable;
import javax.annotation.concurrent.ThreadSafe;
+import java.util.Arrays;
import java.util.Collections;
import java.util.List;
import java.util.Map;
@@ -54,6 +55,9 @@ public class ManifestsReader {
@Nullable private Integer specifiedBucket = null;
@Nullable private Integer specifiedLevel = null;
@Nullable private PartitionPredicate partitionFilter = null;
+ // Auth partition filter (ANDed with partitionFilter); kept separate so it
can be reset each
+ // plan without touching the base partitionFilter.
+ @Nullable private PartitionPredicate authPartitionFilter = null;
@Nullable private BiFilter<Integer, Integer> levelMinMaxFilter = null;
@Nullable protected RowRangeIndex rowRangeIndex = null;
@@ -108,6 +112,12 @@ public class ManifestsReader {
return this;
}
+ /** Overwrites the auth-derived partition filter; {@code null} clears it.
*/
+ public ManifestsReader withAuthPartitionFilter(@Nullable
PartitionPredicate predicate) {
+ this.authPartitionFilter = predicate;
+ return this;
+ }
+
public ManifestsReader withRowRangeIndex(RowRangeIndex rowRangeIndex) {
this.rowRangeIndex = rowRangeIndex;
return this;
@@ -115,7 +125,13 @@ public class ManifestsReader {
@Nullable
public PartitionPredicate partitionFilter() {
- return partitionFilter;
+ if (partitionFilter == null) {
+ return authPartitionFilter;
+ }
+ if (authPartitionFilter == null) {
+ return partitionFilter;
+ }
+ return PartitionPredicate.and(Arrays.asList(partitionFilter,
authPartitionFilter));
}
public Result read(@Nullable Snapshot specifiedSnapshot, ScanMode
scanMode) {
@@ -128,9 +144,12 @@ public class ManifestsReader {
manifests = readManifests(snapshot, scanMode);
}
+ // Compute the effective partition filter once (it ANDs the base and
auth slots) instead of
+ // rebuilding it per manifest.
+ PartitionPredicate effectivePartitionFilter = partitionFilter();
List<ManifestFileMeta> filtered =
manifests.stream()
- .filter(this::filterManifestFileMeta)
+ .filter(m -> filterManifestFileMeta(m,
effectivePartitionFilter))
.collect(Collectors.toList());
return new Result(snapshot, manifests, filtered);
}
@@ -163,7 +182,8 @@ public class ManifestsReader {
}
/** Note: Keep this thread-safe. */
- private boolean filterManifestFileMeta(ManifestFileMeta manifest) {
+ private boolean filterManifestFileMeta(
+ ManifestFileMeta manifest, @Nullable PartitionPredicate
effectivePartitionFilter) {
Integer minBucket = manifest.minBucket();
Integer maxBucket = manifest.maxBucket();
if (minBucket != null && maxBucket != null) {
@@ -188,9 +208,9 @@ public class ManifestsReader {
}
}
- if (partitionFilter != null) {
+ if (effectivePartitionFilter != null) {
SimpleStats stats = manifest.partitionStats();
- if (!partitionFilter.test(
+ if (!effectivePartitionFilter.test(
manifest.numAddedFiles() + manifest.numDeletedFiles(),
stats.minValues(),
stats.maxValues(),
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 adec8a9175..7bf5e68603 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
@@ -69,6 +69,7 @@ import javax.annotation.Nullable;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Optional;
import java.util.TimeZone;
@@ -88,6 +89,11 @@ abstract class AbstractDataTableScan implements
DataTableScan {
private final TableQueryAuth queryAuth;
@Nullable private RowType readType;
+ // Last applied auth predicate; guards redundant re-application across
plan()s.
+ @Nullable private Predicate appliedAuthPredicate;
+ // Whether the auth predicate has a non-partition part (enforced only at
read time). Used by
+ // DataTableBatchScan to disable limit push down; not pushed through
withFilter.
+ protected boolean authHasNonPartitionFilter;
protected AbstractDataTableScan(
TableSchema schema,
@@ -103,6 +109,8 @@ abstract class AbstractDataTableScan implements
DataTableScan {
@Override
public final TableScan.Plan plan() {
TableQueryAuthResult queryAuthResult = authQuery();
+ // Always apply/clear the auth filter so removing auth leaves no stale
partition pruning.
+ applyAuthFilter(queryAuthResult == null ? null :
queryAuthResult.extractPredicate());
Plan plan = planWithoutAuth();
if (queryAuthResult != null) {
plan = queryAuthResult.convertPlan(plan);
@@ -112,6 +120,36 @@ abstract class AbstractDataTableScan implements
DataTableScan {
protected abstract TableScan.Plan planWithoutAuth();
+ private void applyAuthFilter(@Nullable Predicate authPredicate) {
+ if (Objects.equals(authPredicate, appliedAuthPredicate)) {
+ return;
+ }
+ appliedAuthPredicate = authPredicate;
+
+ PartitionPredicate authPartitionFilter = null;
+ boolean hasNonPartitionPart = false;
+ if (authPredicate != null) {
+ // Remap field-id FieldRefs to positional indices by name (as
doAuth does on read), so
+ // pruning stays correct across schema evolution.
+ Predicate remappedAuth =
+ TableQueryAuthResult.remapPredicate(authPredicate,
schema.logicalRowType());
+ if (remappedAuth != null) {
+ Pair<Optional<PartitionPredicate>, List<Predicate>> split =
+
PartitionPredicate.splitPartitionPredicatesAndDataPredicates(
+ remappedAuth, schema.logicalRowType(),
schema.partitionKeys());
+ authPartitionFilter = split.getLeft().orElse(null);
+ hasNonPartitionPart = !split.getRight().isEmpty();
+ }
+ }
+
+ // 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);
+ // A non-partition auth part is enforced only at read time, so limit
push down is unsafe
+ // (DataTableBatchScan reads this). Kept off SnapshotReader since it
is not a pushed filter.
+ this.authHasNonPartitionFilter = hasNonPartitionPart;
+ }
+
@Override
public InnerTableScan withFilter(Predicate predicate) {
snapshotReader.withFilter(predicate);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/DataTableBatchScan.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/DataTableBatchScan.java
index 5293fad4ff..7d528db6c3 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/DataTableBatchScan.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/DataTableBatchScan.java
@@ -97,8 +97,8 @@ public class DataTableBatchScan extends AbstractDataTableScan
{
@Override
public InnerTableScan withLimit(int limit) {
+ // Record it; applyPushDownLimit pushes the file-store limit only when
safe.
this.pushDownLimit = limit;
- snapshotReader.withLimit(limit);
return this;
}
@@ -144,9 +144,14 @@ public class DataTableBatchScan extends
AbstractDataTableScan {
}
private Optional<StartingScanner.Result> applyPushDownLimit() {
- if (pushDownLimit == null || snapshotReader.hasNonPartitionFilter()) {
+ // A read-time filter (WHERE or auth) drops rows after scanning, so
only push the limit down
+ // when neither is present.
+ if (pushDownLimit == null
+ || snapshotReader.hasNonPartitionFilter()
+ || authHasNonPartitionFilter) {
return Optional.empty();
}
+ snapshotReader.withLimit(pushDownLimit);
StartingScanner.Result result = startingScanner.scan(snapshotReader);
if (!(result instanceof ScannedResult)) {
@@ -183,7 +188,11 @@ public class DataTableBatchScan extends
AbstractDataTableScan {
}
private Optional<StartingScanner.Result> applyPushDownTopN() {
- if (topN == null || pushDownLimit != null ||
!schema.primaryKeys().isEmpty()) {
+ // Auth drops rows at read time, so split-level TopN pruning could
drop authorized rows.
+ if (topN == null
+ || pushDownLimit != null
+ || authHasNonPartitionFilter
+ || !schema.primaryKeys().isEmpty()) {
return Optional.empty();
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/ReadBuilderImpl.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/ReadBuilderImpl.java
index 553a848749..6dbc364002 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/ReadBuilderImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/ReadBuilderImpl.java
@@ -19,10 +19,15 @@
package org.apache.paimon.table.source;
import org.apache.paimon.CoreOptions;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.metrics.MetricRegistry;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.predicate.TopN;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.reader.RecordReader.RecordIterator;
import org.apache.paimon.table.InnerTable;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.Filter;
@@ -31,6 +36,7 @@ import org.apache.paimon.utils.RowRangeIndex;
import javax.annotation.Nullable;
+import java.io.IOException;
import java.util.List;
import java.util.Map;
import java.util.Objects;
@@ -47,6 +53,7 @@ public class ReadBuilderImpl implements ReadBuilder {
private final InnerTable table;
private final RowType partitionType;
private final String defaultPartitionName;
+ private final boolean queryAuthEnabled;
private Predicate filter;
@@ -69,7 +76,9 @@ public class ReadBuilderImpl implements ReadBuilder {
public ReadBuilderImpl(InnerTable table) {
this.table = table;
this.partitionType = table.rowType().project(table.partitionKeys());
- this.defaultPartitionName = new
CoreOptions(table.options()).partitionDefaultName();
+ CoreOptions options = new CoreOptions(table.options());
+ this.defaultPartitionName = options.partitionDefaultName();
+ this.queryAuthEnabled = options.queryAuthEnabled();
}
@Override
@@ -234,6 +243,14 @@ public class ReadBuilderImpl implements ReadBuilder {
if (readType != null) {
read.withReadType(readType);
}
+ if (queryAuthEnabled) {
+ // Skip TopN (engine re-applies it); apply the limit after auth
only without a TopN,
+ // else an unordered limit could drop sorted rows.
+ if (topN == null && limit != null) {
+ return new LimitTableRead(read, limit);
+ }
+ return read;
+ }
if (topN != null) {
read.withTopN(topN);
}
@@ -264,4 +281,97 @@ public class ReadBuilderImpl implements ReadBuilder {
result = 31 * result + Objects.hash(readType);
return result;
}
+
+ /**
+ * Limits the delegate's output to {@code limit} rows. Used under query
auth to apply the limit
+ * after read-time auth filtering, not before (which could drop authorized
rows).
+ */
+ private static class LimitTableRead implements TableRead {
+
+ private final TableRead delegate;
+ private final int limit;
+
+ private LimitTableRead(TableRead delegate, int limit) {
+ this.delegate = delegate;
+ this.limit = limit;
+ }
+
+ @Override
+ public TableRead withMetricRegistry(MetricRegistry registry) {
+ delegate.withMetricRegistry(registry);
+ return this;
+ }
+
+ @Override
+ public TableRead executeFilter() {
+ delegate.executeFilter();
+ return this;
+ }
+
+ @Override
+ public TableRead withIOManager(IOManager ioManager) {
+ delegate.withIOManager(ioManager);
+ return this;
+ }
+
+ @Override
+ public RecordReader<InternalRow> createReader(Split split) throws
IOException {
+ return limit(delegate.createReader(split));
+ }
+
+ @Override
+ public RecordReader<InternalRow> createReader(List<Split> splits)
throws IOException {
+ // Limit the merged reader so the cap is global, not per split.
+ return limit(delegate.createReader(splits));
+ }
+
+ @Override
+ public RecordReader<InternalRow> createReader(TableScan.Plan plan)
throws IOException {
+ return limit(delegate.createReader(plan));
+ }
+
+ private RecordReader<InternalRow> limit(RecordReader<InternalRow>
reader) {
+ // Stop reading once the limit is reached (return EOF), rather
than filtering and
+ // draining the rest of the data.
+ return new RecordReader<InternalRow>() {
+ private int count;
+
+ @Nullable
+ @Override
+ public RecordIterator<InternalRow> readBatch() throws
IOException {
+ if (count >= limit) {
+ return null;
+ }
+ RecordIterator<InternalRow> batch = reader.readBatch();
+ if (batch == null) {
+ return null;
+ }
+ return new RecordIterator<InternalRow>() {
+ @Nullable
+ @Override
+ public InternalRow next() throws IOException {
+ if (count >= limit) {
+ return null;
+ }
+ InternalRow row = batch.next();
+ if (row != null) {
+ count++;
+ }
+ return row;
+ }
+
+ @Override
+ public void releaseBatch() {
+ batch.releaseBatch();
+ }
+ };
+ }
+
+ @Override
+ public void close() throws IOException {
+ reader.close();
+ }
+ };
+ }
+ }
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
index 818b6e301c..4cac0d13c9 100644
--- a/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/rest/RESTCatalogTest.java
@@ -59,6 +59,7 @@ import org.apache.paimon.predicate.GreaterThan;
import org.apache.paimon.predicate.LeafPredicate;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
+import org.apache.paimon.predicate.TopN;
import org.apache.paimon.predicate.Transform;
import org.apache.paimon.predicate.UpperTransform;
import org.apache.paimon.reader.RecordReader;
@@ -83,6 +84,7 @@ import org.apache.paimon.table.sink.CommitMessageImpl;
import org.apache.paimon.table.sink.StreamTableCommit;
import org.apache.paimon.table.sink.StreamTableWrite;
import org.apache.paimon.table.sink.TableWriteImpl;
+import org.apache.paimon.table.source.InnerTableScan;
import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.table.source.TableRead;
@@ -138,6 +140,8 @@ import static org.apache.paimon.CoreOptions.TYPE;
import static org.apache.paimon.TableType.OBJECT_TABLE;
import static org.apache.paimon.catalog.Catalog.SYSTEM_DATABASE_NAME;
import static org.apache.paimon.data.BinaryRow.EMPTY_ROW;
+import static org.apache.paimon.predicate.SortValue.NullOrdering.NULLS_LAST;
+import static org.apache.paimon.predicate.SortValue.SortDirection.DESCENDING;
import static org.apache.paimon.rest.RESTApi.PAGE_TOKEN;
import static org.apache.paimon.rest.RESTCatalogOptions.DLF_OSS_ENDPOINT;
import static org.apache.paimon.rest.RESTCatalogOptions.IO_CACHE_ENABLED;
@@ -3898,6 +3902,169 @@ public abstract class RESTCatalogTest extends
CatalogTestBase {
assertThat(rows.get(0).getString(1).toString()).isIn("Alice", "Bob",
"Charlie", "David");
}
+ @Test
+ void testRowFilterWithLimitReadsPastUnauthorizedFiles() throws Exception {
+ Identifier identifier = Identifier.create("test_table_db",
"auth_table_limit");
+ catalog.createDatabase(identifier.getDatabaseName(), true);
+
+ List<DataField> fields = new ArrayList<>();
+ fields.add(new DataField(0, "id", DataTypes.INT()));
+ fields.add(new DataField(1, "secret", DataTypes.INT()));
+ catalog.createTable(
+ identifier,
+ new Schema(
+ fields,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.singletonMap(QUERY_AUTH_ENABLED.key(),
"true"),
+ ""),
+ true);
+
+ Table table = catalog.getTable(identifier);
+
+ // Two files, each with an authorized (secret=1) and an unauthorized
(secret=0) row.
+ commitRows(table, GenericRow.of(1, 1), GenericRow.of(2, 0));
+ commitRows(table, GenericRow.of(3, 1), GenericRow.of(4, 0));
+
+ // secret is a data column, so the filter is enforced at read time,
not pushed to the store.
+ LeafPredicate secretFilter =
+ LeafPredicate.of(
+ new FieldTransform(new FieldRef(1, "secret",
DataTypes.INT())),
+ Equal.INSTANCE,
+ Collections.singletonList(1));
+ setRowFilter(identifier, Collections.singletonList(secretFilter));
+
+ // The file-store limit must be disabled under the auth filter, else
the scan stops at the
+ // first file and misses the authorized row in the second.
+ List<String> result = batchReadWithLimit(table, 1);
+ assertThat(result).contains("+I[1, 1]", "+I[3, 1]");
+ }
+
+ @Test
+ void testRowFilterReadBuilderLimitAfterAuth() throws Exception {
+ Identifier identifier = Identifier.create("test_table_db",
"auth_table_read_limit");
+ catalog.createDatabase(identifier.getDatabaseName(), true);
+
+ List<DataField> fields = new ArrayList<>();
+ fields.add(new DataField(0, "id", DataTypes.INT()));
+ fields.add(new DataField(1, "v", DataTypes.INT()));
+ catalog.createTable(
+ identifier,
+ new Schema(
+ fields,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.singletonMap(QUERY_AUTH_ENABLED.key(),
"true"),
+ ""),
+ true);
+
+ Table table = catalog.getTable(identifier);
+
+ // Rows 1,2 unauthorized and come first; rows 3,4 authorized.
+ commitRows(
+ table,
+ GenericRow.of(1, 10),
+ GenericRow.of(2, 20),
+ GenericRow.of(3, 30),
+ GenericRow.of(4, 40));
+
+ LeafPredicate idGt2 =
+ LeafPredicate.of(
+ new FieldTransform(new FieldRef(0, "id",
DataTypes.INT())),
+ GreaterThan.INSTANCE,
+ Collections.singletonList(2));
+ setRowFilter(identifier, Collections.singletonList(idGt2));
+
+ // withLimit(1) must return exactly one authorized row: skipping the
limit returns 2,
+ // applying it before auth returns 0.
+ List<String> result = batchReadWithReadLimit(table, 1);
+ assertThat(result).hasSize(1);
+ assertThat(result.get(0)).isIn("+I[3, 30]", "+I[4, 40]");
+ }
+
+ @Test
+ void testRowFilterReadLimitSkippedWithTopN() throws Exception {
+ Identifier identifier = Identifier.create("test_table_db",
"auth_table_read_limit_topn");
+ catalog.createDatabase(identifier.getDatabaseName(), true);
+
+ List<DataField> fields = new ArrayList<>();
+ fields.add(new DataField(0, "id", DataTypes.INT()));
+ fields.add(new DataField(1, "score", DataTypes.INT()));
+ catalog.createTable(
+ identifier,
+ new Schema(
+ fields,
+ Collections.emptyList(),
+ Collections.emptyList(),
+ Collections.singletonMap(QUERY_AUTH_ENABLED.key(),
"true"),
+ ""),
+ true);
+
+ Table table = catalog.getTable(identifier);
+
+ // Rows 1,2 unauthorized; 3,4 authorized. The highest-sorted
authorized row (id=4) is last.
+ commitRows(
+ table,
+ GenericRow.of(1, 10),
+ GenericRow.of(2, 20),
+ GenericRow.of(3, 30),
+ GenericRow.of(4, 40));
+
+ LeafPredicate idGt2 =
+ LeafPredicate.of(
+ new FieldTransform(new FieldRef(0, "id",
DataTypes.INT())),
+ GreaterThan.INSTANCE,
+ Collections.singletonList(2));
+ setRowFilter(identifier, Collections.singletonList(idGt2));
+
+ // With a TopN pushed, the read-side limit must be skipped so all
authorized rows reach the
+ // engine's ORDER BY; else id=4 (last in scan order) gets truncated.
+ TopN topN = new TopN(new FieldRef(1, "score", DataTypes.INT()),
DESCENDING, NULLS_LAST, 1);
+ ReadBuilder readBuilder =
table.newReadBuilder().withTopN(topN).withLimit(1);
+ List<String> result = batchRead(table,
readBuilder.newScan().plan().splits(), readBuilder);
+ assertThat(result).containsExactlyInAnyOrder("+I[3, 30]", "+I[4, 40]");
+ }
+
+ @Test
+ void testRowFilterWithTopNKeepsAuthorizedSplits() throws Exception {
+ Identifier identifier = Identifier.create("test_table_db",
"auth_table_topn");
+ catalog.createDatabase(identifier.getDatabaseName(), true);
+
+ // Partitioned so each row is its own split; TopN pruning works at
split granularity.
+ List<DataField> fields = new ArrayList<>();
+ fields.add(new DataField(0, "pt", DataTypes.INT()));
+ fields.add(new DataField(1, "score", DataTypes.INT()));
+ fields.add(new DataField(2, "secret", DataTypes.INT()));
+ catalog.createTable(
+ identifier,
+ new Schema(
+ fields,
+ Collections.singletonList("pt"),
+ Collections.emptyList(),
+ Collections.singletonMap(QUERY_AUTH_ENABLED.key(),
"true"),
+ ""),
+ true);
+
+ Table table = catalog.getTable(identifier);
+
+ // The split with the highest score is unauthorized; the authorized
row has a lower score.
+ commitRows(table, GenericRow.of(1, 100, 0));
+ commitRows(table, GenericRow.of(2, 50, 1));
+
+ LeafPredicate secretFilter =
+ LeafPredicate.of(
+ new FieldTransform(new FieldRef(2, "secret",
DataTypes.INT())),
+ Equal.INSTANCE,
+ Collections.singletonList(1));
+ setRowFilter(identifier, Collections.singletonList(secretFilter));
+
+ // ORDER BY score DESC LIMIT 1: TopN split pruning must be disabled
under the auth filter,
+ // else it keeps the top unauthorized split and drops the authorized
row.
+ TopN topN = new TopN(new FieldRef(1, "score", DataTypes.INT()),
DESCENDING, NULLS_LAST, 1);
+ List<String> result = batchReadWithTopN(table, topN);
+ assertThat(result).contains("+I[2, 50, 1]");
+ }
+
@Test
public void testConflictRollback() throws Exception {
doTestConflictRollback(false);
@@ -4067,9 +4234,46 @@ public abstract class RESTCatalogTest extends
CatalogTestBase {
commit.close();
}
+ protected void commitRows(Table table, GenericRow... rows) throws
Exception {
+ BatchWriteBuilder writeBuilder = table.newBatchWriteBuilder();
+ BatchTableWrite write = writeBuilder.newWrite();
+ for (GenericRow row : rows) {
+ write.write(row);
+ }
+ BatchTableCommit commit = writeBuilder.newCommit();
+ commit.commit(write.prepareCommit());
+ write.close();
+ commit.close();
+ }
+
protected List<String> batchRead(Table table) throws IOException {
- ReadBuilder readBuilder = table.newReadBuilder();
- List<Split> splits = readBuilder.newScan().plan().splits();
+ return batchRead(table,
table.newReadBuilder().newScan().plan().splits());
+ }
+
+ protected List<String> batchReadWithLimit(Table table, int limit) throws
IOException {
+ // Limit only the scan (file pruning), not the read, so the result
reflects the kept files.
+ InnerTableScan scan = (InnerTableScan)
table.newReadBuilder().newScan();
+ return batchRead(table, scan.withLimit(limit).plan().splits());
+ }
+
+ protected List<String> batchReadWithTopN(Table table, TopN topN) throws
IOException {
+ // Apply TopN only on the scan (split pruning), so the result reflects
the kept splits.
+ InnerTableScan scan = (InnerTableScan)
table.newReadBuilder().newScan();
+ return batchRead(table, scan.withTopN(topN).plan().splits());
+ }
+
+ protected List<String> batchReadWithReadLimit(Table table, int limit)
throws IOException {
+ // Exercises the ReadBuilder.withLimit path, which limits both the
scan and the read side.
+ ReadBuilder readBuilder = table.newReadBuilder().withLimit(limit);
+ return batchRead(table, readBuilder.newScan().plan().splits(),
readBuilder);
+ }
+
+ private List<String> batchRead(Table table, List<Split> splits) throws
IOException {
+ return batchRead(table, splits, table.newReadBuilder());
+ }
+
+ private List<String> batchRead(Table table, List<Split> splits,
ReadBuilder readBuilder)
+ throws IOException {
TableRead read = readBuilder.newRead();
RecordReader<InternalRow> reader = read.createReader(splits);
List<String> result = new ArrayList<>();
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RESTCatalogITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RESTCatalogITCase.java
index de93679e1a..5be454a9d0 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RESTCatalogITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/RESTCatalogITCase.java
@@ -472,6 +472,49 @@ class RESTCatalogITCase extends RESTCatalogITCaseBase {
DATABASE_NAME, filterTable)))
.containsExactlyInAnyOrder(Row.of(3, "Charlie", 35, "IT"));
+ String partitionFilterTable = "partition_row_filter_table";
+ batchSql(
+ String.format(
+ "CREATE TABLE %s.%s (id INT, name STRING, dtpart
STRING) PARTITIONED BY (dtpart) WITH ('query-auth.enabled' = 'true')",
+ DATABASE_NAME, partitionFilterTable));
+ batchSql(
+ String.format(
+ "INSERT INTO %s.%s VALUES (1, 'blocked',
'2026-07-03'), (2, 'allowed', '2026-07-02')",
+ DATABASE_NAME, partitionFilterTable));
+ Predicate partitionPredicate =
+ LeafPredicate.of(
+ new FieldTransform(new FieldRef(2, "dtpart",
DataTypes.STRING())),
+ Equal.INSTANCE,
+
Collections.singletonList(BinaryString.fromString("2026-07-02")));
+ restCatalogServer.setRowFilterAuth(
+ Identifier.create(DATABASE_NAME, partitionFilterTable),
+ Collections.singletonList(partitionPredicate));
+
+ assertThat(
+ batchSql(
+ String.format(
+ "SELECT * FROM %s.%s WHERE dtpart =
'2026-07-02'",
+ DATABASE_NAME, partitionFilterTable)))
+ .containsExactly(Row.of(2, "allowed", "2026-07-02"));
+ assertThat(
+ batchSql(
+ String.format(
+ "SELECT * FROM %s.%s",
+ DATABASE_NAME, partitionFilterTable)))
+ .containsExactly(Row.of(2, "allowed", "2026-07-02"));
+ assertThat(
+ batchSql(
+ String.format(
+ "SELECT id, dtpart FROM %s.%s LIMIT 1",
+ DATABASE_NAME, partitionFilterTable)))
+ .containsExactly(Row.of(2, "2026-07-02"));
+ assertThat(
+ batchSql(
+ String.format(
+ "SELECT id FROM %s.%s WHERE dtpart =
'2026-07-03'",
+ DATABASE_NAME, partitionFilterTable)))
+ .isEmpty();
+
// Test JOIN with row filter
String joinTable = "join_table";
batchSql(
diff --git
a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkCatalogWithRestTest.java
b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkCatalogWithRestTest.java
index 4a8f43e61b..fcde06202e 100644
---
a/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkCatalogWithRestTest.java
+++
b/paimon-spark/paimon-spark-ut/src/test/java/org/apache/paimon/spark/SparkCatalogWithRestTest.java
@@ -33,6 +33,7 @@ import org.apache.paimon.predicate.FieldRef;
import org.apache.paimon.predicate.FieldTransform;
import org.apache.paimon.predicate.GreaterOrEqual;
import org.apache.paimon.predicate.GreaterThan;
+import org.apache.paimon.predicate.IsNull;
import org.apache.paimon.predicate.LeafPredicate;
import org.apache.paimon.predicate.LessThan;
import org.apache.paimon.predicate.Predicate;
@@ -443,6 +444,70 @@ public class SparkCatalogWithRestTest {
.toString())
.isEqualTo("[[3,Charlie,35,IT]]");
+ spark.sql(
+ "CREATE TABLE t_partition_row_filter (id INT, name STRING,
dtpart STRING) PARTITIONED BY (dtpart)"
+ + " TBLPROPERTIES ('query-auth.enabled'='true')");
+ spark.sql(
+ "INSERT INTO t_partition_row_filter VALUES (1, 'blocked',
'2026-07-03'), (2, 'allowed', '2026-07-02')");
+ Predicate partitionPredicate =
+ LeafPredicate.of(
+ new FieldTransform(new FieldRef(2, "dtpart",
DataTypes.STRING())),
+ Equal.INSTANCE,
+
Collections.singletonList(BinaryString.fromString("2026-07-02")));
+ restCatalogServer.setRowFilterAuth(
+ Identifier.create("db2", "t_partition_row_filter"),
+ Collections.singletonList(partitionPredicate));
+
+ assertThat(
+ spark.sql(
+ "SELECT * FROM t_partition_row_filter
WHERE dtpart = '2026-07-02'")
+ .collectAsList()
+ .toString())
+ .isEqualTo("[[2,allowed,2026-07-02]]");
+ assertThat(spark.sql("SELECT * FROM
t_partition_row_filter").collectAsList().toString())
+ .isEqualTo("[[2,allowed,2026-07-02]]");
+ assertThat(
+ spark.sql("SELECT id, dtpart FROM
t_partition_row_filter LIMIT 1")
+ .collectAsList()
+ .toString())
+ .isEqualTo("[[2,2026-07-02]]");
+ assertThat(
+ spark.sql(
+ "SELECT id FROM t_partition_row_filter
WHERE dtpart = '2026-07-03'")
+ .collectAsList()
+ .toString())
+ .isEqualTo("[]");
+
+ // After DROP COLUMN, a partition key's field id no longer equals its
position; pruning must
+ // remap by name, else it mis-projects onto p2 and drops visible rows.
+ spark.sql(
+ "CREATE TABLE t_evolved_row_filter (a INT, b STRING, p1
STRING, p2 STRING)"
+ + " PARTITIONED BY (p1, p2) TBLPROPERTIES
('query-auth.enabled'='true')");
+ spark.sql("ALTER TABLE t_evolved_row_filter DROP COLUMN a");
+ spark.sql(
+ "INSERT INTO t_evolved_row_filter VALUES ('x', 'keep', 'A'),
('y', 'keep', 'B'),"
+ + " ('z', 'drop', 'A')");
+ // p1 keeps field id 2 though it now sits at position 1 after the drop.
+ Predicate evolvedPredicate =
+ LeafPredicate.of(
+ new FieldTransform(new FieldRef(2, "p1",
DataTypes.STRING())),
+ Equal.INSTANCE,
+
Collections.singletonList(BinaryString.fromString("keep")));
+ restCatalogServer.setRowFilterAuth(
+ Identifier.create("db2", "t_evolved_row_filter"),
+ Collections.singletonList(evolvedPredicate));
+
+ assertThat(
+ spark.sql("SELECT b FROM t_evolved_row_filter ORDER BY
b")
+ .collectAsList()
+ .toString())
+ .isEqualTo("[[x], [y]]");
+ assertThat(
+ spark.sql("SELECT b FROM t_evolved_row_filter WHERE p2
= 'B'")
+ .collectAsList()
+ .toString())
+ .isEqualTo("[[y]]");
+
// Test JOIN with row filter
spark.sql("CREATE TABLE t_join2 (id INT, salary DOUBLE)");
spark.sql(
@@ -541,6 +606,161 @@ public class SparkCatalogWithRestTest {
.isEqualTo("[[4]]");
}
+ @Test
+ public void testRowFilterPrimaryKeyTable() {
+ // Row filter on a PK table must respect merge-on-read: id=2 survives
on its merged value,
+ // not dropped on the stale file value.
+ spark.sql(
+ "CREATE TABLE t_pk_row_filter (id INT, name STRING, age INT)
TBLPROPERTIES"
+ + " ('primary-key'='id', 'bucket'='2',
'query-auth.enabled'='true')");
+ spark.sql(
+ "INSERT INTO t_pk_row_filter VALUES (1, 'Alice', 25), (2,
'Bob', 30), (3, 'Charlie', 35)");
+ // Later snapshot updates id=2 from age 30 to 40.
+ spark.sql("INSERT INTO t_pk_row_filter VALUES (2, 'Bob', 40)");
+
+ // Filter on a value (non-key) column across the update.
+ Predicate ageGt30Predicate =
+ LeafPredicate.of(
+ new FieldTransform(new FieldRef(2, "age",
DataTypes.INT())),
+ GreaterThan.INSTANCE,
+ Collections.singletonList(30));
+ restCatalogServer.setRowFilterAuth(
+ Identifier.create("db2", "t_pk_row_filter"),
+ Collections.singletonList(ageGt30Predicate));
+ assertThat(
+ spark.sql("SELECT * FROM t_pk_row_filter ORDER BY id")
+ .collectAsList()
+ .toString())
+ .isEqualTo("[[2,Bob,40], [3,Charlie,35]]");
+
+ // Filter on the primary-key column.
+ Predicate idGe2Predicate =
+ LeafPredicate.of(
+ new FieldTransform(new FieldRef(0, "id",
DataTypes.INT())),
+ GreaterOrEqual.INSTANCE,
+ Collections.singletonList(2));
+ restCatalogServer.setRowFilterAuth(
+ Identifier.create("db2", "t_pk_row_filter"),
+ Collections.singletonList(idGe2Predicate));
+ assertThat(
+ spark.sql("SELECT * FROM t_pk_row_filter ORDER BY id")
+ .collectAsList()
+ .toString())
+ .isEqualTo("[[2,Bob,40], [3,Charlie,35]]");
+ }
+
+ @Test
+ public void testRowFilterNullAndDefaultPartition() {
+ // IS NULL row filter on a regular column returns only rows whose
value is null.
+ spark.sql(
+ "CREATE TABLE t_null_row_filter (id INT, grade STRING)
TBLPROPERTIES"
+ + " ('query-auth.enabled'='true')");
+ spark.sql(
+ "INSERT INTO t_null_row_filter VALUES (1, 'A'), (2, CAST(NULL
AS STRING)), (3, 'B'),"
+ + " (4, CAST(NULL AS STRING))");
+ Predicate gradeIsNull =
+ LeafPredicate.of(
+ new FieldTransform(new FieldRef(1, "grade",
DataTypes.STRING())),
+ IsNull.INSTANCE,
+ Collections.emptyList());
+ restCatalogServer.setRowFilterAuth(
+ Identifier.create("db2", "t_null_row_filter"),
+ Collections.singletonList(gradeIsNull));
+ assertThat(
+ spark.sql("SELECT id FROM t_null_row_filter ORDER BY
id")
+ .collectAsList()
+ .toString())
+ .isEqualTo("[[2], [4]]");
+
+ // Partition column with NULL values goes to the default partition.
+ spark.sql(
+ "CREATE TABLE t_null_partition_filter (id INT, dtpart STRING)
PARTITIONED BY (dtpart)"
+ + " TBLPROPERTIES ('query-auth.enabled'='true')");
+ spark.sql(
+ "INSERT INTO t_null_partition_filter VALUES (1, '2026-07-02'),"
+ + " (2, CAST(NULL AS STRING)), (3, '2026-07-03'), (4,
CAST(NULL AS STRING))");
+
+ // Equality on the partition column must exclude the NULL (default)
partition.
+ Predicate dtpartEq =
+ LeafPredicate.of(
+ new FieldTransform(new FieldRef(1, "dtpart",
DataTypes.STRING())),
+ Equal.INSTANCE,
+
Collections.singletonList(BinaryString.fromString("2026-07-02")));
+ restCatalogServer.setRowFilterAuth(
+ Identifier.create("db2", "t_null_partition_filter"),
+ Collections.singletonList(dtpartEq));
+ assertThat(
+ spark.sql("SELECT id FROM t_null_partition_filter
ORDER BY id")
+ .collectAsList()
+ .toString())
+ .isEqualTo("[[1]]");
+
+ // IS NULL on the partition column must keep only the default (NULL)
partition.
+ Predicate dtpartIsNull =
+ LeafPredicate.of(
+ new FieldTransform(new FieldRef(1, "dtpart",
DataTypes.STRING())),
+ IsNull.INSTANCE,
+ Collections.emptyList());
+ restCatalogServer.setRowFilterAuth(
+ Identifier.create("db2", "t_null_partition_filter"),
+ Collections.singletonList(dtpartIsNull));
+ assertThat(
+ spark.sql("SELECT id FROM t_null_partition_filter
ORDER BY id")
+ .collectAsList()
+ .toString())
+ .isEqualTo("[[2], [4]]");
+ }
+
+ @Test
+ public void testRowFilterBucketKeyTable() {
+ // Row filter on the bucket-key column flows through bucket selection;
must stay correct.
+ spark.sql(
+ "CREATE TABLE t_bucket_row_filter (id INT, name STRING)
TBLPROPERTIES"
+ + " ('bucket'='4', 'bucket-key'='id',
'query-auth.enabled'='true')");
+ spark.sql("INSERT INTO t_bucket_row_filter VALUES (1, 'a'), (2, 'b'),
(3, 'c'), (4, 'd')");
+ Predicate idEq3Predicate =
+ LeafPredicate.of(
+ new FieldTransform(new FieldRef(0, "id",
DataTypes.INT())),
+ Equal.INSTANCE,
+ Collections.singletonList(3));
+ restCatalogServer.setRowFilterAuth(
+ Identifier.create("db2", "t_bucket_row_filter"),
+ Collections.singletonList(idEq3Predicate));
+ assertThat(
+ spark.sql("SELECT * FROM t_bucket_row_filter ORDER BY
id")
+ .collectAsList()
+ .toString())
+ .isEqualTo("[[3,c]]");
+ }
+
+ @Test
+ public void testRowFilterDeletionVectorsTable() {
+ // Deletion-vectors table: deleted rows excluded via the deletion
vector, then filtered.
+ spark.sql(
+ "CREATE TABLE t_dv_row_filter (id INT, name STRING, age INT)
TBLPROPERTIES"
+ + " ('primary-key'='id', 'bucket'='2',
'deletion-vectors.enabled'='true',"
+ + " 'query-auth.enabled'='true')");
+ spark.sql(
+ "INSERT INTO t_dv_row_filter VALUES (1, 'Alice', 25), (2,
'Bob', 30),"
+ + " (3, 'Charlie', 35), (4, 'David', 45)");
+ // Delete id=4 (writes a deletion vector). id=4 passes the age>=35
filter, so if the
+ // deletion vector were ignored it would resurface in the result and
fail the test.
+ spark.sql("DELETE FROM t_dv_row_filter WHERE id = 4");
+ Predicate ageGe35Predicate =
+ LeafPredicate.of(
+ new FieldTransform(new FieldRef(2, "age",
DataTypes.INT())),
+ GreaterOrEqual.INSTANCE,
+ Collections.singletonList(35));
+ restCatalogServer.setRowFilterAuth(
+ Identifier.create("db2", "t_dv_row_filter"),
+ Collections.singletonList(ageGe35Predicate));
+ assertThat(
+ spark.sql("SELECT * FROM t_dv_row_filter ORDER BY id")
+ .collectAsList()
+ .toString())
+ .isEqualTo("[[3,Charlie,35]]");
+ }
+
@Test
public void testColumnMaskingAndRowFilter() {
spark.sql(
