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 5aae9ddb63 [format] Fix DELETE and UPDATE targeting wrong rows after
Parquet predicate filtering (#8987)
5aae9ddb63 is described below
commit 5aae9ddb630b3fc77c24fa307e3fac3ed70014e4
Author: XiaoHongbo <[email protected]>
AuthorDate: Mon Aug 3 18:59:09 2026 +0800
[format] Fix DELETE and UPDATE targeting wrong rows after Parquet predicate
filtering (#8987)
---
.../format/parquet/reader/ColumnarBatch.java | 4 +
.../format/parquet/reader/RowIndexGenerator.java | 30 +++-
.../parquet/reader/RowIndexGeneratorTest.java | 199 +++++++++++++++++++++
3 files changed, 231 insertions(+), 2 deletions(-)
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/ColumnarBatch.java
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/ColumnarBatch.java
index 10b6c6de04..b1a6991f94 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/ColumnarBatch.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/ColumnarBatch.java
@@ -67,6 +67,10 @@ public class ColumnarBatch {
this.vectorizedColumnBatch.setNumRows(numRows);
}
+ int numRows() {
+ return vectorizedColumnBatch.getNumRows();
+ }
+
/** Returns the column at `ordinal`. */
public ColumnVector column(int ordinal) {
return columns[ordinal];
diff --git
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/RowIndexGenerator.java
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/RowIndexGenerator.java
index 19f20e719c..1a30360eb9 100644
---
a/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/RowIndexGenerator.java
+++
b/paimon-format/src/main/java/org/apache/paimon/format/parquet/reader/RowIndexGenerator.java
@@ -22,14 +22,19 @@ import org.apache.paimon.utils.LongIterator;
import org.apache.parquet.column.page.PageReadStore;
+import java.util.NoSuchElementException;
import java.util.PrimitiveIterator;
/** Generate row index for columnar batch. */
-public class RowIndexGenerator {
+public class RowIndexGenerator implements LongIterator {
private LongIterator rowIndexIterator;
+ private long pendingSkips;
+ private int remainingIndexes;
public void initFromPageReadStore(PageReadStore pageReadStore) {
+ pendingSkips = 0;
+ remainingIndexes = 0;
long startingRowIdx = pageReadStore.getRowIndexOffset().orElse(0L);
PrimitiveIterator.OfLong rowIndexes =
pageReadStore.getRowIndexes().orElse(null);
if (rowIndexes != null) {
@@ -53,6 +58,27 @@ public class RowIndexGenerator {
}
public void populateRowIndex(ColumnarBatch columnarBatch) {
- columnarBatch.resetPositions(rowIndexIterator);
+ pendingSkips += remainingIndexes;
+ remainingIndexes = columnarBatch.numRows();
+ columnarBatch.resetPositions(this);
+ }
+
+ @Override
+ public boolean hasNext() {
+ return remainingIndexes > 0 && rowIndexIterator.hasNext();
+ }
+
+ @Override
+ public long next() {
+ if (remainingIndexes == 0) {
+ throw new NoSuchElementException();
+ }
+ while (pendingSkips > 0) {
+ rowIndexIterator.next();
+ pendingSkips--;
+ }
+ long index = rowIndexIterator.next();
+ remainingIndexes--;
+ return index;
}
}
diff --git
a/paimon-format/src/test/java/org/apache/paimon/format/parquet/reader/RowIndexGeneratorTest.java
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/reader/RowIndexGeneratorTest.java
new file mode 100644
index 0000000000..468543e857
--- /dev/null
+++
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/reader/RowIndexGeneratorTest.java
@@ -0,0 +1,199 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.format.parquet.reader;
+
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.data.columnar.ColumnVector;
+import org.apache.paimon.data.columnar.ColumnarRowIterator;
+import org.apache.paimon.data.columnar.heap.HeapIntVector;
+import org.apache.paimon.data.columnar.heap.HeapLongVector;
+import org.apache.paimon.fs.Path;
+import org.apache.paimon.table.SpecialFields;
+
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.column.page.PageReadStore;
+import org.apache.parquet.column.page.PageReader;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+import java.util.Optional;
+import java.util.PrimitiveIterator;
+import java.util.stream.LongStream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests for {@link RowIndexGenerator}. */
+public class RowIndexGeneratorTest {
+
+ @Test
+ public void testLazyRowIdAcrossBatches() {
+ CountingRowIndexes indexes = new CountingRowIndexes(6);
+ HeapLongVector rowIds = new HeapLongVector(3);
+ rowIds.fillWithNulls();
+ ColumnarBatch batch =
+ new ColumnarBatch(
+ new Path("test"), new ColumnVector[] {new
HeapIntVector(3), rowIds}, null);
+ ColumnarRowIterator iterator = batch.vectorizedRowIterator;
+ iterator.assignRowTracking(
+ 500_000L, 1L,
Collections.singletonMap(SpecialFields.ROW_ID.name(), 1));
+
+ RowIndexGenerator generator = newGenerator(indexes);
+
+ batch.setNumRows(3);
+ generator.populateRowIndex(batch);
+ InternalRow row;
+ while ((row = iterator.next()) != null) {
+ row.getInt(0);
+ }
+
+ batch.setNumRows(3);
+ generator.populateRowIndex(batch);
+ assertThat(indexes.nextIndex).isZero();
+ row = iterator.next();
+ assertThat(row.getLong(1)).isEqualTo(500_003L);
+ assertThat(indexes.nextIndex).isEqualTo(4);
+ }
+
+ @Test
+ public void testPartiallyConsumedBatch() {
+ CountingRowIndexes indexes = new CountingRowIndexes(6);
+ RowIndexGenerator generator = newGenerator(indexes);
+ ColumnarBatch batch = newBatch();
+
+ generator.populateRowIndex(batch);
+ assertThat(generator.next()).isZero();
+
+ generator.populateRowIndex(batch);
+ assertThat(generator.next()).isEqualTo(3);
+ }
+
+ @Test
+ public void testMultipleUnconsumedBatchesStayLazy() {
+ CountingRowIndexes indexes = new CountingRowIndexes(9);
+ RowIndexGenerator generator = newGenerator(indexes);
+ ColumnarBatch batch = newBatch();
+
+ generator.populateRowIndex(batch);
+ generator.populateRowIndex(batch);
+ generator.populateRowIndex(batch);
+ assertThat(indexes.nextIndex).isZero();
+ assertThat(generator.next()).isEqualTo(6);
+ assertThat(indexes.nextIndex).isEqualTo(7);
+ }
+
+ @Test
+ public void testRowGroupResetDropsPendingPositions() {
+ CountingRowIndexes firstGroup = new CountingRowIndexes(3);
+ RowIndexGenerator generator = newGenerator(firstGroup);
+ ColumnarBatch batch = newBatch();
+ generator.populateRowIndex(batch);
+
+ CountingRowIndexes secondGroup = new CountingRowIndexes(3);
+ initGenerator(generator, secondGroup, 100);
+ generator.populateRowIndex(batch);
+
+ assertThat(firstGroup.nextIndex).isZero();
+ assertThat(generator.next()).isEqualTo(100);
+ }
+
+ @Test
+ public void testNonContinuousRowIndexes() {
+ PrimitiveIterator.OfLong indexes = LongStream.of(1, 4, 9, 12, 20,
30).iterator();
+ RowIndexGenerator generator = newGenerator(indexes, 6);
+ ColumnarBatch batch = newBatch();
+
+ generator.populateRowIndex(batch);
+ generator.populateRowIndex(batch);
+
+ assertThat(generator.next()).isEqualTo(12);
+ }
+
+ private static ColumnarBatch newBatch() {
+ ColumnarBatch batch =
+ new ColumnarBatch(
+ new Path("test"), new ColumnVector[] {new
HeapIntVector(3)}, null);
+ batch.setNumRows(3);
+ return batch;
+ }
+
+ private static RowIndexGenerator newGenerator(CountingRowIndexes indexes) {
+ return newGenerator(indexes, indexes.end);
+ }
+
+ private static RowIndexGenerator newGenerator(PrimitiveIterator.OfLong
indexes, long rowCount) {
+ RowIndexGenerator generator = new RowIndexGenerator();
+ initGenerator(generator, indexes, rowCount, 0);
+ return generator;
+ }
+
+ private static void initGenerator(
+ RowIndexGenerator generator, CountingRowIndexes indexes, long
rowIndexOffset) {
+ initGenerator(generator, indexes, indexes.end, rowIndexOffset);
+ }
+
+ private static void initGenerator(
+ RowIndexGenerator generator,
+ PrimitiveIterator.OfLong indexes,
+ long rowCount,
+ long rowIndexOffset) {
+ PageReadStore page =
+ new PageReadStore() {
+ @Override
+ public PageReader getPageReader(ColumnDescriptor
descriptor) {
+ return null;
+ }
+
+ @Override
+ public long getRowCount() {
+ return rowCount;
+ }
+
+ @Override
+ public Optional<Long> getRowIndexOffset() {
+ return Optional.of(rowIndexOffset);
+ }
+
+ @Override
+ public Optional<PrimitiveIterator.OfLong> getRowIndexes() {
+ return Optional.of(indexes);
+ }
+ };
+ generator.initFromPageReadStore(page);
+ }
+
+ private static class CountingRowIndexes implements
PrimitiveIterator.OfLong {
+
+ private final long end;
+ private long nextIndex;
+
+ private CountingRowIndexes(long end) {
+ this.end = end;
+ }
+
+ @Override
+ public long nextLong() {
+ return nextIndex++;
+ }
+
+ @Override
+ public boolean hasNext() {
+ return nextIndex < end;
+ }
+ }
+}
