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 4c3b7f56b6 [common] Normalize row-id IN literals inside Range.toRanges
(#9630)
4c3b7f56b6 is described below
commit 4c3b7f56b65a59f4065be884bd3aca24b978c55f
Author: YangJie <[email protected]>
AuthorDate: Fri Sep 11 03:15:57 2026 -0400
[common] Normalize row-id IN literals inside Range.toRanges (#9630)
---
.../paimon/predicate/RowIdPredicateVisitor.java | 11 +++--
.../main/java/org/apache/paimon/utils/Range.java | 17 ++++++-
.../predicate/RowIdPredicateVisitorTest.java | 48 +++++++++++++++++++
.../paimon/table/DataEvolutionTableTest.java | 56 ++++++++++++++++++++++
4 files changed, 127 insertions(+), 5 deletions(-)
diff --git
a/paimon-common/src/main/java/org/apache/paimon/predicate/RowIdPredicateVisitor.java
b/paimon-common/src/main/java/org/apache/paimon/predicate/RowIdPredicateVisitor.java
index d0ccf6600f..6d145aa8b2 100644
---
a/paimon-common/src/main/java/org/apache/paimon/predicate/RowIdPredicateVisitor.java
+++
b/paimon-common/src/main/java/org/apache/paimon/predicate/RowIdPredicateVisitor.java
@@ -61,9 +61,14 @@ public class RowIdPredicateVisitor implements
PredicateVisitor<Optional<List<Ran
return Optional.of(Range.toRanges(rowIds));
} else if (function instanceof Between) {
List<Object> literals = predicate.literals();
- return Optional.of(
- Collections.singletonList(
- new Range((Long) literals.get(0), (Long)
literals.get(1))));
+ long from = (Long) literals.get(0);
+ long to = (Long) literals.get(1);
+ // BETWEEN with inverted bounds (SQL allows BETWEEN 10 AND 5)
is empty.
+ if (from > to) {
+ // Mutable: the Or union path accumulates into the
returned list.
+ return Optional.of(new ArrayList<>());
+ }
+ return Optional.of(Collections.singletonList(new Range(from,
to)));
}
}
return Optional.empty();
diff --git a/paimon-common/src/main/java/org/apache/paimon/utils/Range.java
b/paimon-common/src/main/java/org/apache/paimon/utils/Range.java
index 4457a00254..cdebaa342c 100644
--- a/paimon-common/src/main/java/org/apache/paimon/utils/Range.java
+++ b/paimon-common/src/main/java/org/apache/paimon/utils/Range.java
@@ -121,7 +121,8 @@ public class Range implements Serializable {
public static List<Range> sortAndMergeOverlap(List<Range> ranges, boolean
adjacent) {
if (ranges == null || ranges.isEmpty()) {
- return Collections.emptyList();
+ // Mutable: callers accumulate into the result across children.
+ return new ArrayList<>();
}
if (ranges.size() == 1) {
@@ -184,9 +185,18 @@ public class Range implements Serializable {
return result;
}
+ /**
+ * Groups row ids into ascending ranges, merging consecutive ids. The ids
may arrive in any
+ * order and may repeat; both are normalized here, since the returned list
has to be sorted and
+ * non-overlapping for {@link #and(List, List)} to intersect it correctly.
+ */
public static List<Range> toRanges(Iterable<Long> ids) {
+ List<Long> sorted = new ArrayList<>();
+ ids.forEach(sorted::add);
+ Collections.sort(sorted);
+
List<Range> ranges = new ArrayList<>();
- Iterator<Long> iterator = ids.iterator();
+ Iterator<Long> iterator = sorted.iterator();
if (!iterator.hasNext()) {
return ranges;
@@ -197,6 +207,9 @@ public class Range implements Serializable {
while (iterator.hasNext()) {
long current = iterator.next();
+ if (current == rangeEnd) {
+ continue;
+ }
if (current != rangeEnd + 1) {
// Save the current range and start a new one
ranges.add(new Range(rangeStart, rangeEnd));
diff --git
a/paimon-common/src/test/java/org/apache/paimon/predicate/RowIdPredicateVisitorTest.java
b/paimon-common/src/test/java/org/apache/paimon/predicate/RowIdPredicateVisitorTest.java
index 7d4db36ace..34d04cd6dd 100644
---
a/paimon-common/src/test/java/org/apache/paimon/predicate/RowIdPredicateVisitorTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/predicate/RowIdPredicateVisitorTest.java
@@ -26,6 +26,7 @@ import org.apache.paimon.utils.Range;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.List;
@@ -82,6 +83,53 @@ public class RowIdPredicateVisitorTest {
assertThat(unrecognized.visit(visitor)).isEmpty();
}
+ @Test
+ public void testUnsortedInLiteralsIntersectCorrectly() {
+ // IN literals in descending engine order: unsorted input to
Range.and's
+ // two-pointer intersection silently drops rows (e.g. only [5,5]
instead of
+ // [1,2] and [5,6]).
+ // >20 literals so PredicateBuilder keeps a real In leaf (smaller INs
become
+ // an OR of equals): descending 25..5 exercises the In branch unsorted.
+ List<Object> descending = new ArrayList<>();
+ for (long v = 25; v >= 5; v--) {
+ descending.add(v);
+ }
+ Predicate inUnsorted = builder.in(rowIdIndex, descending);
+ Predicate between1To6 = builder.between(rowIdIndex, 1L, 6L);
+ Predicate and = PredicateBuilder.and(inUnsorted, between1To6);
+ Optional<List<Range>> result = and.visit(visitor);
+ assertThat(result).isPresent();
+ assertThat(result.get()).containsExactly(new Range(5, 6));
+
+ // Duplicates in the IN list must not produce overlapping ranges.
+ List<Object> duplicates = new ArrayList<>();
+ for (int i = 0; i < 21; i++) {
+ duplicates.add(15L);
+ }
+ Predicate inDuplicates = builder.in(rowIdIndex, duplicates);
+ Predicate between10To20 = builder.between(rowIdIndex, 10L, 20L);
+ Optional<List<Range>> dedup =
+ PredicateBuilder.and(inDuplicates,
between10To20).visit(visitor);
+ assertThat(dedup).isPresent();
+ assertThat(dedup.get()).containsExactly(new Range(15, 15));
+ }
+
+ @Test
+ public void testInvertedBetweenIsEmpty() {
+ // SQL allows BETWEEN 10 AND 5; it matches nothing and must not
fabricate an
+ // inverted Range that breaks the from <= to invariant.
+ Predicate inverted = builder.between(rowIdIndex, 10L, 5L);
+ Optional<List<Range>> result = inverted.visit(visitor);
+ assertThat(result).isPresent();
+ assertThat(result.get()).isEmpty();
+
+ // Inverted BETWEEN under OR must not break the union accumulation.
+ Predicate equal7 = builder.equal(rowIdIndex, 7L);
+ Optional<List<Range>> orResult = PredicateBuilder.or(inverted,
equal7).visit(visitor);
+ assertThat(orResult).isPresent();
+ assertThat(orResult.get()).containsExactly(new Range(7, 7));
+ }
+
@Test
public void testCompoundedPredicate() {
// Test AND intersection
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 2cd688c9e9..c10e8ad575 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
@@ -601,6 +601,62 @@ public class DataEvolutionTableTest extends
DataEvolutionTestBase {
assertThat(plannedFirstRowIds(plan)).isEqualTo(Arrays.asList(0L, 1L));
}
+ @Test
+ public void testDescendingRowIdInIntersectBetweenReadsCorrectRows() throws
Exception {
+ // Table-level regression: a descending _ROW_ID IN list intersected
with a BETWEEN used to
+ // drop ranges (Range.toRanges/Range.and need ascending, deduped
input), and the dropped
+ // ranges are rows that are never read. This is the TableRead
equivalent of
+ // RowIdPredicateVisitorTest#testUnsortedInLiteralsIntersectCorrectly.
+ write(30); // one batch per column group; row id i <-> f0 == i
+ Schema schema = schemaDefault();
+ PredicateBuilder pb = new PredicateBuilder(rowTypeWithRowId(schema));
+ int rowIdIndex = schema.rowType().getFieldCount();
+
+ // IN (25,24,...,5) is 21 descending literals (> 20, so
PredicateBuilder keeps a real In
+ // leaf) intersected with BETWEEN 3 AND 8 -> {5,6,7,8}.
+ Predicate filter =
+ PredicateBuilder.and(
+ pb.in(rowIdIndex, descendingRowIds(25L, 5L)),
+ pb.between(rowIdIndex, 3L, 8L));
+ assertThat(readF0WithFilter(filter)).isEqualTo(Arrays.asList(5, 6, 7,
8));
+ }
+
+ @Test
+ public void testEmptyRowIdIntersectionUnderOrReadsOtherBranch() throws
Exception {
+ // The empty branch (disjoint IN ∩ BETWEEN) has to yield a mutable
empty range list so the
+ // Or union can accumulate the other branch into it; before the fix
this threw
+ // UnsupportedOperationException while planning the scan.
+ write(30);
+ Schema schema = schemaDefault();
+ PredicateBuilder pb = new PredicateBuilder(rowTypeWithRowId(schema));
+ int rowIdIndex = schema.rowType().getFieldCount();
+
+ Predicate emptyIntersection =
+ PredicateBuilder.and(
+ pb.in(rowIdIndex, descendingRowIds(25L, 5L)),
+ pb.between(rowIdIndex, 100L, 110L)); // disjoint from
the IN -> empty
+ Predicate filter = PredicateBuilder.or(emptyIntersection,
pb.between(rowIdIndex, 10L, 12L));
+ assertThat(readF0WithFilter(filter)).isEqualTo(Arrays.asList(10, 11,
12));
+ }
+
+ private List<Integer> readF0WithFilter(Predicate filter) throws Exception {
+ ReadBuilder rb = getTableDefault().newReadBuilder().withFilter(filter);
+ List<Integer> f0 = new ArrayList<>();
+ try (RecordReader<InternalRow> reader =
rb.newRead().createReader(rb.newScan().plan())) {
+ reader.forEachRemaining(r -> f0.add(r.getInt(0)));
+ }
+ Collections.sort(f0);
+ return f0;
+ }
+
+ private static List<Object> descendingRowIds(long hi, long lo) {
+ List<Object> ids = new ArrayList<>();
+ for (long v = hi; v >= lo; v--) {
+ ids.add(v);
+ }
+ return ids;
+ }
+
@Test
public void testLimitPushDownWithoutFilter() throws Exception {
createTableDefault();
