This is an automated email from the ASF dual-hosted git repository. JingsongLi pushed a commit to branch release-2.0 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit 1d817ddd42459681a8d91bf9900c00da2df15f32 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; + } + } +}
