This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit fc9658dbcd721f4438903fbaacd0c5153f4bb95a 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();
