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 dd0bbf6a3b [core][python] Fix mixed row-id OR pushdown (#8496)
dd0bbf6a3b is described below
commit dd0bbf6a3bd8a80ee659517598e8b8fc822b7ca8
Author: QuakeWang <[email protected]>
AuthorDate: Tue Jul 7 20:39:05 2026 +0800
[core][python] Fix mixed row-id OR pushdown (#8496)
Data evolution scans reused the full filter for row-range extraction,
stats pruning, and global index evaluation.
For mixed predicates such as `_ROW_ID = 1 OR f0 > 5`, the row-id visitor
cannot fully convert the predicate to row ranges. The old path could
still remove or push `_ROW_ID` branches into stats/global-index pruning,
which may narrow scan semantics, hit system-field handling bugs, or
incorrectly enable limit/top-n split pruning.
This PR keeps the full predicate for read-time filtering and limit/top-n
safety, while only pushing row-id-safe residual predicates into scan
pruning and global index evaluation. Python follows the same rule and
excludes `_ROW_ID` from stats predicates.
---
.../paimon/globalindex/DataEvolutionBatchScan.java | 49 +++++++---
.../table/source/snapshot/SnapshotReader.java | 9 ++
.../table/source/snapshot/SnapshotReaderImpl.java | 24 ++++-
.../apache/paimon/table/system/AuditLogTable.java | 10 ++
.../globalindex/DataEvolutionBatchScanTest.java | 73 +++++++++++++++
.../paimon/table/BtreeGlobalIndexTableTest.java | 21 +++++
.../paimon/table/DataEvolutionTableTest.java | 102 +++++++++++++++++++++
paimon-python/pypaimon/read/push_down_utils.py | 6 ++
.../pypaimon/read/scanner/file_scanner.py | 11 ++-
paimon-python/pypaimon/tests/range_test.py | 3 +
.../pypaimon/tests/reader_predicate_test.py | 23 ++++-
11 files changed, 305 insertions(+), 26 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 76ae6195d5..39634f012c 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
@@ -27,7 +27,9 @@ import org.apache.paimon.metrics.MetricRegistry;
import org.apache.paimon.partition.PartitionPredicate;
import org.apache.paimon.predicate.CompoundPredicate;
import org.apache.paimon.predicate.LeafPredicate;
+import org.apache.paimon.predicate.Or;
import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.predicate.RowIdPredicateVisitor;
import org.apache.paimon.predicate.TopN;
import org.apache.paimon.table.FileStoreTable;
@@ -85,39 +87,52 @@ public class DataEvolutionBatchScan implements
DataTableScan {
return this;
}
- predicate.visit(new
RowIdPredicateVisitor()).ifPresent(this::withRowRanges);
- predicate = removeRowIdFilter(predicate);
+ Optional<List<Range>> rowRanges = predicate.visit(new
RowIdPredicateVisitor());
+ if (rowRanges.isPresent()) {
+ withRowRanges(rowRanges.get());
+ }
this.filter = predicate;
- batchScan.withFilter(predicate);
+
+ batchScan.snapshotReader().withFilter(predicate,
rowIdSafeResidualFilter(predicate));
return this;
}
- private Predicate removeRowIdFilter(Predicate filter) {
+ private Predicate rowIdSafeResidualFilter(Predicate filter) {
if (filter instanceof LeafPredicate
&& ((LeafPredicate)
filter).fieldNames().contains(ROW_ID.name())) {
return null;
} else if (filter instanceof CompoundPredicate) {
CompoundPredicate compoundPredicate = (CompoundPredicate) filter;
+ if (compoundPredicate.function() instanceof Or) {
+ return containsRowId(compoundPredicate) ? null : filter;
+ }
List<Predicate> newChildren = new ArrayList<>();
for (Predicate child : compoundPredicate.children()) {
- Predicate newChild = removeRowIdFilter(child);
+ Predicate newChild = rowIdSafeResidualFilter(child);
if (newChild != null) {
newChildren.add(newChild);
}
}
- if (newChildren.isEmpty()) {
- return null;
- } else if (newChildren.size() == 1) {
- return newChildren.get(0);
- } else {
- return new CompoundPredicate(compoundPredicate.function(),
newChildren);
- }
+ return PredicateBuilder.andNullable(newChildren);
}
return filter;
}
+ private boolean containsRowId(Predicate filter) {
+ if (filter instanceof LeafPredicate) {
+ return ((LeafPredicate)
filter).fieldNames().contains(ROW_ID.name());
+ } else if (filter instanceof CompoundPredicate) {
+ for (Predicate child : ((CompoundPredicate) filter).children()) {
+ if (containsRowId(child)) {
+ return true;
+ }
+ }
+ }
+ return false;
+ }
+
@Override
public InnerTableScan withReadType(@Nullable RowType readType) {
batchScan.withReadType(readType);
@@ -270,19 +285,23 @@ public class DataEvolutionBatchScan implements
DataTableScan {
if (!options.globalIndexEnabled()) {
return Optional.empty();
}
+ Predicate globalIndexFilter = rowIdSafeResidualFilter(filter);
+ if (globalIndexFilter == null) {
+ return Optional.empty();
+ }
PartitionPredicate partitionFilter =
batchScan.snapshotReader().manifestsReader().partitionFilter();
Optional<GlobalIndexScanner> optionalScanner =
- GlobalIndexScanner.create(table, partitionFilter, filter);
+ GlobalIndexScanner.create(table, partitionFilter,
globalIndexFilter);
if (!optionalScanner.isPresent()) {
return Optional.empty();
}
try (GlobalIndexScanner scanner = optionalScanner.get()) {
- Optional<GlobalIndexResult> result = scanner.scan(filter);
+ Optional<GlobalIndexResult> result =
scanner.scan(globalIndexFilter);
if (result.isPresent()) {
LOG.info("Scan table '{}' with global index.", table.name());
- return
Optional.of(result.get().or(scanner.unindexedRows(filter)));
+ return
Optional.of(result.get().or(scanner.unindexedRows(globalIndexFilter)));
}
return Optional.empty();
} catch (IOException e) {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java
index 1ea0854eea..90e680e6d1 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReader.java
@@ -75,6 +75,15 @@ public interface SnapshotReader {
SnapshotReader withFilter(Predicate predicate);
+ /**
+ * Applies a full read-time filter and an optional scan-pruning filter.
+ *
+ * <p>{@code predicate} is used to determine whether the reader has a
non-partition filter which
+ * must still be evaluated at read time. {@code pushdownPredicate} is the
only predicate used
+ * for scan pruning such as partition, stats, and bucket pruning.
+ */
+ SnapshotReader withFilter(Predicate predicate, @Nullable Predicate
pushdownPredicate);
+
SnapshotReader withPartitionFilter(Map<String, String> partitionSpec);
SnapshotReader withPartitionFilter(Predicate predicate);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java
b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java
index 0a752f2564..839b4cef79 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/source/snapshot/SnapshotReaderImpl.java
@@ -233,17 +233,31 @@ public class SnapshotReaderImpl implements SnapshotReader
{
@Override
public SnapshotReader withFilter(Predicate predicate) {
+ return withFilter(predicate, predicate);
+ }
+
+ @Override
+ public SnapshotReader withFilter(Predicate predicate, @Nullable Predicate
pushdownPredicate) {
Pair<Optional<PartitionPredicate>, List<Predicate>> pair =
splitPartitionPredicatesAndDataPredicates(
predicate, tableSchema.logicalRowType(),
tableSchema.partitionKeys());
- if (pair.getLeft().isPresent()) {
- scan.withPartitionFilter(pair.getLeft().get());
- }
if (!pair.getRight().isEmpty()) {
this.hasNonPartitionFilter = true;
- nonPartitionFilterConsumer.accept(scan,
PredicateBuilder.and(pair.getRight()));
}
- scan.withCompleteFilter(predicate);
+
+ Pair<Optional<PartitionPredicate>, List<Predicate>> pushdownPair =
+ splitPartitionPredicatesAndDataPredicates(
+ pushdownPredicate,
+ tableSchema.logicalRowType(),
+ tableSchema.partitionKeys());
+ if (pushdownPair.getLeft().isPresent()) {
+ scan.withPartitionFilter(pushdownPair.getLeft().get());
+ }
+ if (!pushdownPair.getRight().isEmpty()) {
+ this.hasNonPartitionFilter = true;
+ nonPartitionFilterConsumer.accept(scan,
PredicateBuilder.and(pushdownPair.getRight()));
+ }
+ scan.withCompleteFilter(pushdownPredicate);
return this;
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
index 1e912d4068..71ca70d8fe 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/system/AuditLogTable.java
@@ -340,6 +340,16 @@ public class AuditLogTable implements DataTable,
ReadonlyTable {
return this;
}
+ @Override
+ public SnapshotReader withFilter(Predicate predicate, Predicate
pushdownPredicate) {
+ Optional<Predicate> converted = convert(predicate);
+ Optional<Predicate> convertedPushdown = convert(pushdownPredicate);
+ if (converted.isPresent()) {
+ wrapped.withFilter(converted.get(),
convertedPushdown.orElse(null));
+ }
+ return this;
+ }
+
@Override
public SnapshotReader withPartitionFilter(Map<String, String>
partitionSpec) {
wrapped.withPartitionFilter(partitionSpec);
diff --git
a/paimon-core/src/test/java/org/apache/paimon/globalindex/DataEvolutionBatchScanTest.java
b/paimon-core/src/test/java/org/apache/paimon/globalindex/DataEvolutionBatchScanTest.java
index 30744ae542..4c7cc83d07 100644
---
a/paimon-core/src/test/java/org/apache/paimon/globalindex/DataEvolutionBatchScanTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/globalindex/DataEvolutionBatchScanTest.java
@@ -20,12 +20,20 @@ package org.apache.paimon.globalindex;
import org.apache.paimon.data.BinaryRow;
import org.apache.paimon.io.DataFileMeta;
+import org.apache.paimon.predicate.Predicate;
+import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.DataTableBatchScan;
import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.source.snapshot.SnapshotReader;
+import org.apache.paimon.types.DataField;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.Range;
import org.apache.paimon.utils.RowRangeIndex;
import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
import java.util.ArrayList;
import java.util.Arrays;
@@ -34,11 +42,63 @@ import java.util.List;
import java.util.Random;
import static org.apache.paimon.stats.SimpleStats.EMPTY_STATS;
+import static org.apache.paimon.table.SpecialFields.ROW_ID;
import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.Mockito.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.same;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
/** Tests for {@link DataEvolutionBatchScan}. */
public class DataEvolutionBatchScanTest {
+ @Test
+ public void testWithFilterKeepsMixedOrWhenRowRangeExtractionFails() {
+ PredicateBuilder builder = new PredicateBuilder(rowTypeWithRowId());
+ Predicate predicate = PredicateBuilder.or(builder.equal(2, 1L),
builder.greaterThan(0, 5));
+
+ DataTableBatchScan batchScan = mock(DataTableBatchScan.class);
+ SnapshotReader snapshotReader = mockSnapshotReader(batchScan);
+ new DataEvolutionBatchScan(null, batchScan).withFilter(predicate);
+
+ verify(snapshotReader).withFilter(predicate, null);
+ verify(batchScan, never()).withFilter(any(Predicate.class));
+ }
+
+ @Test
+ public void testWithFilterRemovesRowIdAfterRowRangeExtractionSucceeds() {
+ PredicateBuilder builder = new PredicateBuilder(rowTypeWithRowId());
+ Predicate nonRowIdPredicate = builder.greaterThan(0, 5);
+ Predicate predicate = PredicateBuilder.and(builder.equal(2, 1L),
nonRowIdPredicate);
+
+ DataTableBatchScan batchScan = mock(DataTableBatchScan.class);
+ SnapshotReader snapshotReader = mockSnapshotReader(batchScan);
+ new DataEvolutionBatchScan(null, batchScan).withFilter(predicate);
+
+ ArgumentCaptor<Predicate> captor =
ArgumentCaptor.forClass(Predicate.class);
+ verify(snapshotReader).withFilter(same(predicate), captor.capture());
+ assertThat(captor.getValue()).isSameAs(nonRowIdPredicate);
+ }
+
+ @Test
+ public void testWithFilterDropsNestedMixedOrFromStatsResidual() {
+ PredicateBuilder builder = new PredicateBuilder(rowTypeWithRowId());
+ Predicate nonRowIdPredicate = builder.lessThan(0, 100);
+ Predicate mixedOr = PredicateBuilder.or(builder.equal(2, 1L),
builder.greaterThan(1, 5));
+ Predicate predicate =
+ PredicateBuilder.and(builder.between(2, 0L, 10L),
nonRowIdPredicate, mixedOr);
+
+ DataTableBatchScan batchScan = mock(DataTableBatchScan.class);
+ SnapshotReader snapshotReader = mockSnapshotReader(batchScan);
+ new DataEvolutionBatchScan(null, batchScan).withFilter(predicate);
+
+ ArgumentCaptor<Predicate> captor =
ArgumentCaptor.forClass(Predicate.class);
+ verify(snapshotReader).withFilter(same(predicate), captor.capture());
+ assertThat(captor.getValue()).isSameAs(nonRowIdPredicate);
+ }
+
@Test
public void testWrapToIndexSplitsRandomly() {
Random random = new Random();
@@ -137,6 +197,19 @@ public class DataEvolutionBatchScanTest {
.containsExactly(new Range(4200, 4450), new Range(4650, 4700));
}
+ private static RowType rowTypeWithRowId() {
+ return RowType.of(
+ new DataField(0, "f0", DataTypes.INT()),
+ new DataField(1, "f1", DataTypes.INT()),
+ new DataField(2, ROW_ID.name(), DataTypes.BIGINT()));
+ }
+
+ private static SnapshotReader mockSnapshotReader(DataTableBatchScan
batchScan) {
+ SnapshotReader snapshotReader = mock(SnapshotReader.class);
+ when(batchScan.snapshotReader()).thenReturn(snapshotReader);
+ return snapshotReader;
+ }
+
private static List<Range> expectedRanges(long min, long max, List<Range>
rowRanges) {
List<Range> expected = new ArrayList<>();
for (Range range : rowRanges) {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java
index 0723ee5239..9ae9666200 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/BtreeGlobalIndexTableTest.java
@@ -133,6 +133,27 @@ public class BtreeGlobalIndexTableTest extends
DataEvolutionTestBase {
assertThat(readF1).containsExactly("a200", "a300", "a400", "a56789");
}
+ @Test
+ public void testMixedRowIdOrSkipsGlobalIndexScan() throws Exception {
+ write(10L);
+ createIndex("f1");
+
+ FileStoreTable table = (FileStoreTable) catalog.getTable(identifier());
+ PredicateBuilder builder =
+ new
PredicateBuilder(SpecialFields.rowTypeWithRowId(table.rowType()));
+ int rowIdIndex = table.rowType().getFieldCount();
+ Predicate predicate =
+ PredicateBuilder.or(
+ builder.equal(rowIdIndex, 1L),
+ builder.equal(1, BinaryString.fromString("a7")));
+
+ ReadBuilder readBuilder = table.newReadBuilder().withFilter(predicate);
+ List<Split> splits = readBuilder.newScan().plan().splits();
+
+ assertThat(splits).isNotEmpty();
+ assertThat(splits).allMatch(split -> split instanceof DataSplit);
+ }
+
@Test
public void testBTreeGlobalIndexSearchModeControlsUnindexedData() throws
Exception {
write(500L);
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionTableTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionTableTest.java
index 89742874c1..51cf16a114 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionTableTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/DataEvolutionTableTest.java
@@ -51,6 +51,7 @@ import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.table.source.StreamTableScan;
import org.apache.paimon.table.source.TableScan;
+import org.apache.paimon.types.DataField;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.Range;
@@ -516,6 +517,84 @@ public class DataEvolutionTableTest extends
DataEvolutionTestBase {
assertThat(readBuilder.newScan().plan().splits().isEmpty()).isTrue();
}
+ @Test
+ public void testMixedRowIdOrFilterDoesNotPruneByUnsafeStatsResidual()
throws Exception {
+ createTableDefault();
+ Schema schema = schemaDefault();
+ FileStoreTable table = getTableDefault();
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ RowType writeType = schema.rowType().project(Arrays.asList("f0",
"f1"));
+
+ try (BatchTableWrite write =
builder.newWrite().withWriteType(writeType);
+ BatchTableCommit commit = builder.newCommit()) {
+ write.write(GenericRow.of(1, BinaryString.fromString("a")));
+ write.write(GenericRow.of(2, BinaryString.fromString("b")));
+ List<CommitMessage> commitables = write.prepareCommit();
+ setFirstRowId(commitables, 0L);
+ commit.commit(commitables);
+ }
+
+ try (BatchTableWrite write =
builder.newWrite().withWriteType(writeType);
+ BatchTableCommit commit = builder.newCommit()) {
+ write.write(GenericRow.of(6, BinaryString.fromString("c")));
+ write.write(GenericRow.of(7, BinaryString.fromString("d")));
+ List<CommitMessage> commitables = write.prepareCommit();
+ setFirstRowId(commitables, 2L);
+ commit.commit(commitables);
+ }
+
+ PredicateBuilder predicateBuilder = new
PredicateBuilder(rowTypeWithRowId(schema));
+ int rowIdIndex = schema.rowType().getFieldCount();
+ Predicate topLevelMixedOr =
+ PredicateBuilder.or(
+ predicateBuilder.equal(rowIdIndex, 1L),
predicateBuilder.greaterThan(0, 5));
+ assertThat(plannedFirstRowIds(table,
topLevelMixedOr)).isEqualTo(Arrays.asList(0L, 2L));
+
+ Predicate nestedMixedOr =
+ PredicateBuilder.and(
+ predicateBuilder.between(rowIdIndex, 0L, 10L),
+ PredicateBuilder.or(
+ predicateBuilder.equal(rowIdIndex, 1L),
+ predicateBuilder.greaterThan(0, 5)));
+ assertThat(plannedFirstRowIds(table,
nestedMixedOr)).isEqualTo(Arrays.asList(0L, 2L));
+ }
+
+ @Test
+ public void testMixedRowIdOrFilterDisablesUnsafeLimitPushDown() throws
Exception {
+ createTableDefault();
+ Schema schema = schemaDefault();
+ FileStoreTable table = getTableDefault();
+ BatchWriteBuilder builder = table.newBatchWriteBuilder();
+ RowType writeType = schema.rowType().project(Arrays.asList("f0",
"f1"));
+
+ try (BatchTableWrite write =
builder.newWrite().withWriteType(writeType);
+ BatchTableCommit commit = builder.newCommit()) {
+ write.write(GenericRow.of(1, BinaryString.fromString("a")));
+ List<CommitMessage> commitables = write.prepareCommit();
+ setFirstRowId(commitables, 0L);
+ commit.commit(commitables);
+ }
+
+ try (BatchTableWrite write =
builder.newWrite().withWriteType(writeType);
+ BatchTableCommit commit = builder.newCommit()) {
+ write.write(GenericRow.of(60, BinaryString.fromString("b")));
+ List<CommitMessage> commitables = write.prepareCommit();
+ setFirstRowId(commitables, 1L);
+ commit.commit(commitables);
+ }
+
+ PredicateBuilder predicateBuilder = new
PredicateBuilder(rowTypeWithRowId(schema));
+ int rowIdIndex = schema.rowType().getFieldCount();
+ Predicate predicate =
+ PredicateBuilder.or(
+ predicateBuilder.equal(rowIdIndex, 999L),
+ predicateBuilder.greaterThan(0, 50));
+
+ TableScan.Plan plan =
+
table.newReadBuilder().withFilter(predicate).withLimit(1).newScan().plan();
+ assertThat(plannedFirstRowIds(plan)).isEqualTo(Arrays.asList(0L, 1L));
+ }
+
@Test
public void testLimitPushDownWithoutFilter() throws Exception {
createTableDefault();
@@ -1987,6 +2066,29 @@ public class DataEvolutionTableTest extends
DataEvolutionTestBase {
.sum();
}
+ private static List<Long> plannedFirstRowIds(FileStoreTable table,
Predicate filter) {
+ return
plannedFirstRowIds(table.newReadBuilder().withFilter(filter).newScan().plan());
+ }
+
+ private static List<Long> plannedFirstRowIds(TableScan.Plan plan) {
+ return plan.splits().stream()
+ .flatMap(
+ split ->
+ (split instanceof DataSplit
+ ? ((DataSplit)
split).dataFiles()
+ : ((IndexedSplit)
split).dataSplit().dataFiles())
+ .stream())
+ .map(DataFileMeta::firstRowId)
+ .sorted()
+ .collect(Collectors.toList());
+ }
+
+ private static RowType rowTypeWithRowId(Schema schema) {
+ List<DataField> fields = new ArrayList<>(schema.rowType().getFields());
+ fields.add(SpecialFields.ROW_ID);
+ return new RowType(fields);
+ }
+
private static long countMatchingRows(ReadBuilder rb) throws Exception {
RecordReader<InternalRow> reader =
rb.newRead().createReader(rb.newScan().plan());
AtomicInteger cnt = new AtomicInteger(0);
diff --git a/paimon-python/pypaimon/read/push_down_utils.py
b/paimon-python/pypaimon/read/push_down_utils.py
index ee564a4c79..9c12872d9f 100644
--- a/paimon-python/pypaimon/read/push_down_utils.py
+++ b/paimon-python/pypaimon/read/push_down_utils.py
@@ -150,6 +150,12 @@ def remove_row_id_filter(predicate: Predicate) ->
Optional[Predicate]:
filtered.append(r)
return PredicateBuilder.and_predicates(filtered)
if predicate.method == "or":
+ fields = _get_all_fields(predicate)
+ if (
+ SpecialFields.ROW_ID.name in fields
+ and fields != {SpecialFields.ROW_ID.name}
+ ):
+ return predicate
new_children = []
for c in predicate.literals or []:
r = remove_row_id_filter(c)
diff --git a/paimon-python/pypaimon/read/scanner/file_scanner.py
b/paimon-python/pypaimon/read/scanner/file_scanner.py
index 57fe7e27b2..153b424b13 100755
--- a/paimon-python/pypaimon/read/scanner/file_scanner.py
+++ b/paimon-python/pypaimon/read/scanner/file_scanner.py
@@ -52,6 +52,7 @@ from pypaimon.read.scanner.primary_key_table_split_generator
import \
from pypaimon.read.split import DataSplit
from pypaimon.snapshot.snapshot import Snapshot
from pypaimon.table.bucket_mode import BucketMode
+from pypaimon.table.special_fields import SpecialFields
from pypaimon.table.source.deletion_file import DeletionFile
@@ -221,7 +222,15 @@ class FileScanner:
self.table: FileStoreTable = table
self.manifest_scanner = manifest_scanner
self.predicate = predicate
- self.predicate_for_stats = remove_row_id_filter(predicate) if
predicate else None
+ row_ranges = (
+ _row_ranges_from_predicate(predicate) if predicate else None
+ )
+ if predicate and row_ranges is not None:
+ self.predicate_for_stats = remove_row_id_filter(predicate)
+ else:
+ self.predicate_for_stats = predicate
+ self.predicate_for_stats = exclude_predicate_with_fields(
+ self.predicate_for_stats, {SpecialFields.ROW_ID.name})
# Partition columns aren't in data files, so skip them for value-stats
pruning.
self.predicate_for_stats = exclude_predicate_with_fields(
self.predicate_for_stats, set(self.table.partition_keys))
diff --git a/paimon-python/pypaimon/tests/range_test.py
b/paimon-python/pypaimon/tests/range_test.py
index 0956ddfaf4..500fe47de9 100644
--- a/paimon-python/pypaimon/tests/range_test.py
+++ b/paimon-python/pypaimon/tests/range_test.py
@@ -75,6 +75,9 @@ class RangeTest(unittest.TestCase):
pred_or = Predicate('or', None, None, [pred_eq5, pred_eq3])
assert _row_ranges_from_predicate(pred_or) == [Range(3, 3), Range(5,
5)]
+ pred_mixed_or = Predicate(
+ 'or', None, None, [pred_eq5, pred_other])
+ assert _row_ranges_from_predicate(pred_mixed_or) is None
pred_between_1_5 = Predicate('between', 0, SpecialFields.ROW_ID.name,
[1, 5])
pred_between_3_7 = Predicate('between', 0, SpecialFields.ROW_ID.name,
[3, 7])
pred_or_overlap = Predicate('or', None, None, [pred_between_1_5,
pred_between_3_7])
diff --git a/paimon-python/pypaimon/tests/reader_predicate_test.py
b/paimon-python/pypaimon/tests/reader_predicate_test.py
index 4508a2063c..89e4efef6a 100644
--- a/paimon-python/pypaimon/tests/reader_predicate_test.py
+++ b/paimon-python/pypaimon/tests/reader_predicate_test.py
@@ -104,12 +104,25 @@ class ReaderPredicateTest(unittest.TestCase):
self.assertIsNotNone(result)
self.assertEqual(result.field, 'f0')
self.assertEqual(result.method, 'greaterThan')
- or_mixed = pb.or_predicates([pb.equal('_ROW_ID', 1),
pb.greater_than('f0', 5)])
+ or_mixed = pb.or_predicates([
+ pb.equal('_ROW_ID', 1),
+ pb.greater_than('f0', 5),
+ ])
result = push_down_utils.remove_row_id_filter(or_mixed)
- self.assertIsNotNone(result, "OR: strip _ROW_ID child, keep f0>5 (same
as Java)")
- self.assertEqual(result.field, 'f0')
- self.assertEqual(result.method, 'greaterThan')
- or_no_row_id = pb.or_predicates([pb.greater_than('f0', 5),
pb.less_than('f0', 10)])
+ self.assertIsNotNone(result)
+ self.assertEqual(result.method, 'or')
+ self.assertEqual(len(result.literals), 2)
+ self.assertEqual(result.literals[0].field, '_ROW_ID')
+ self.assertEqual(result.literals[1].field, 'f0')
+ or_row_id = pb.or_predicates([
+ pb.equal('_ROW_ID', 1),
+ pb.equal('_ROW_ID', 2),
+ ])
+ self.assertIsNone(push_down_utils.remove_row_id_filter(or_row_id))
+ or_no_row_id = pb.or_predicates([
+ pb.greater_than('f0', 5),
+ pb.less_than('f0', 10),
+ ])
result = push_down_utils.remove_row_id_filter(or_no_row_id)
self.assertIsNotNone(result)
self.assertEqual(result.method, 'or')
