This is an automated email from the ASF dual-hosted git repository.
pvary pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/iceberg.git
The following commit(s) were added to refs/heads/main by this push:
new 89a6a8776a Spark: Backport forEachInRange vectorized delete checks to
3.5, 4.0 and 4.1 (#18207)
89a6a8776a is described below
commit 89a6a8776a251f5113770697b2c1d634c1f5fae0
Author: JeonDaehong <[email protected]>
AuthorDate: Wed Sep 23 16:28:51 2026 +0900
Spark: Backport forEachInRange vectorized delete checks to 3.5, 4.0 and 4.1
(#18207)
Backport of #18027
---
.../spark/data/vectorized/ColumnarBatchUtil.java | 107 +++++++++++++
.../data/vectorized/TestColumnarBatchUtil.java | 166 ++++++++++++++++-----
.../spark/data/vectorized/ColumnarBatchUtil.java | 107 +++++++++++++
.../data/vectorized/TestColumnarBatchUtil.java | 166 ++++++++++++++++-----
.../spark/data/vectorized/ColumnarBatchUtil.java | 107 +++++++++++++
.../data/vectorized/TestColumnarBatchUtil.java | 166 ++++++++++++++++-----
6 files changed, 708 insertions(+), 111 deletions(-)
diff --git
a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchUtil.java
b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchUtil.java
index 89fe4538bd..bddc144fb2 100644
---
a/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchUtil.java
+++
b/spark/v3.5/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchUtil.java
@@ -19,6 +19,7 @@
package org.apache.iceberg.spark.data.vectorized;
import java.util.Arrays;
+import java.util.function.LongConsumer;
import java.util.function.Predicate;
import org.apache.iceberg.data.DeleteFilter;
import org.apache.iceberg.deletes.PositionDeleteIndex;
@@ -64,6 +65,15 @@ public class ColumnarBatchUtil {
}
PositionDeleteIndex deletedPositions = deletes.deletedRowPositions();
+
+ if (!deletes.hasEqDeletes()) {
+ // positions in a batch form a contiguous ascending range, so the index
can be traversed once
+ // for the whole range instead of being probed once per row
+ return deletedPositions == null
+ ? null
+ : buildRowIdMapping(deletedPositions, deletes, rowStartPosInBatch,
batchSize);
+ }
+
Predicate<InternalRow> eqDeleteFilter = deletes.eqDeletedRowFilter();
ColumnarBatchRow row = new ColumnarBatchRow(columnVectors);
int[] rowIdMapping = new int[batchSize];
@@ -83,6 +93,21 @@ public class ColumnarBatchUtil {
return liveRowId == batchSize ? null : Pair.of(rowIdMapping, liveRowId);
}
+ // builds the mapping from the deleted positions in the batch range, without
probing per row
+ private static Pair<int[], Integer> buildRowIdMapping(
+ PositionDeleteIndex deletedPositions,
+ DeleteFilter<InternalRow> deletes,
+ long rowStartPosInBatch,
+ int batchSize) {
+ RowIdMappingCollector collector =
+ new RowIdMappingCollector(deletes, rowStartPosInBatch, batchSize);
+ deletedPositions.forEachInRange(rowStartPosInBatch, rowStartPosInBatch +
batchSize, collector);
+ collector.appendRemainingLiveRows();
+
+ int liveRowId = collector.liveRowCount();
+ return liveRowId == batchSize ? null : Pair.of(collector.rowIdMapping(),
liveRowId);
+ }
+
/**
* Builds a boolean array to indicate if a row is deleted or not.
*
@@ -119,6 +144,20 @@ public class ColumnarBatchUtil {
}
PositionDeleteIndex deletedPositions = deletes.deletedRowPositions();
+
+ if (!deletes.hasEqDeletes()) {
+ // positions in a batch form a contiguous ascending range, so the index
can be traversed once
+ // for the whole range instead of being probed once per row
+ if (deletedPositions != null) {
+ IsDeletedCollector collector =
+ new IsDeletedCollector(deletes, isDeleted, rowStartPosInBatch);
+ deletedPositions.forEachInRange(
+ rowStartPosInBatch, rowStartPosInBatch + batchSize, collector);
+ }
+
+ return isDeleted;
+ }
+
Predicate<InternalRow> eqDeleteFilter = deletes.eqDeletedRowFilter();
ColumnarBatchRow row = new ColumnarBatchRow(columnVectors);
@@ -178,4 +217,72 @@ public class ColumnarBatchUtil {
return columnVectors;
}
}
+
+ /**
+ * Consumes deleted positions in ascending order, filling the gaps between
them with live row IDs.
+ */
+ private static class RowIdMappingCollector implements LongConsumer {
+ private final DeleteFilter<InternalRow> deletes;
+ private final long rowStartPosInBatch;
+ private final int batchSize;
+ private final int[] rowIdMapping;
+ private int nextRowId = 0;
+ private int liveRowId = 0;
+
+ RowIdMappingCollector(
+ DeleteFilter<InternalRow> deletes, long rowStartPosInBatch, int
batchSize) {
+ this.deletes = deletes;
+ this.rowStartPosInBatch = rowStartPosInBatch;
+ this.batchSize = batchSize;
+ this.rowIdMapping = new int[batchSize];
+ }
+
+ @Override
+ public void accept(long pos) {
+ int deletedRowId = (int) (pos - rowStartPosInBatch);
+ for (int rowId = nextRowId; rowId < deletedRowId; rowId++) {
+ rowIdMapping[liveRowId] = rowId;
+ liveRowId++;
+ }
+
+ deletes.incrementDeleteCount();
+ this.nextRowId = deletedRowId + 1;
+ }
+
+ /** Appends the live rows that follow the last deleted position in the
batch. */
+ void appendRemainingLiveRows() {
+ for (int rowId = nextRowId; rowId < batchSize; rowId++) {
+ rowIdMapping[liveRowId] = rowId;
+ liveRowId++;
+ }
+ }
+
+ int liveRowCount() {
+ return liveRowId;
+ }
+
+ int[] rowIdMapping() {
+ return rowIdMapping;
+ }
+ }
+
+ /** Consumes deleted positions in a batch range, marking them in the given
array. */
+ private static class IsDeletedCollector implements LongConsumer {
+ private final DeleteFilter<InternalRow> deletes;
+ private final boolean[] isDeleted;
+ private final long rowStartPosInBatch;
+
+ IsDeletedCollector(
+ DeleteFilter<InternalRow> deletes, boolean[] isDeleted, long
rowStartPosInBatch) {
+ this.deletes = deletes;
+ this.isDeleted = isDeleted;
+ this.rowStartPosInBatch = rowStartPosInBatch;
+ }
+
+ @Override
+ public void accept(long pos) {
+ isDeleted[(int) (pos - rowStartPosInBatch)] = true;
+ deletes.incrementDeleteCount();
+ }
+ }
}
diff --git
a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/data/vectorized/TestColumnarBatchUtil.java
b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/data/vectorized/TestColumnarBatchUtil.java
index dadbe3e788..468242f610 100644
---
a/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/data/vectorized/TestColumnarBatchUtil.java
+++
b/spark/v3.5/spark/src/test/java/org/apache/iceberg/spark/data/vectorized/TestColumnarBatchUtil.java
@@ -21,14 +21,21 @@ package org.apache.iceberg.spark.data.vectorized;
import static java.util.Collections.nCopies;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.util.Arrays;
+import java.util.List;
+import java.util.Random;
import java.util.function.Predicate;
import java.util.stream.Stream;
import org.apache.iceberg.Schema;
import org.apache.iceberg.data.DeleteFilter;
+import org.apache.iceberg.deletes.Deletes;
import org.apache.iceberg.deletes.PositionDeleteIndex;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.spark.sql.catalyst.InternalRow;
import org.apache.spark.sql.vectorized.ColumnVector;
import org.junit.jupiter.api.BeforeEach;
@@ -48,13 +55,7 @@ public class TestColumnarBatchUtil {
@Test
public void testBuildRowIdMappingNoDeletes() {
when(deleteFilter.hasPosDeletes()).thenReturn(true);
- PositionDeleteIndex deletedRowPos = mock(PositionDeleteIndex.class);
-
- for (long i = 0; i <= 10; i++) {
- when(deletedRowPos.isDeleted(i)).thenReturn(false);
- }
-
- when(deleteFilter.deletedRowPositions()).thenReturn(deletedRowPos);
+ when(deleteFilter.deletedRowPositions()).thenReturn(positionIndex());
var rowIdMapping = ColumnarBatchUtil.buildRowIdMapping(columnVectors,
deleteFilter, 0, 10);
assertThat(rowIdMapping).isNull();
}
@@ -62,14 +63,8 @@ public class TestColumnarBatchUtil {
@Test
public void testBuildRowIdMappingPositionDeletesOnly() {
when(deleteFilter.hasPosDeletes()).thenReturn(true);
- PositionDeleteIndex deletedRowPos = mock(PositionDeleteIndex.class);
-
// 5 position deletes
- for (long i = 98; i < 103; i++) {
- when(deletedRowPos.isDeleted(i)).thenReturn(true);
- }
-
- when(deleteFilter.deletedRowPositions()).thenReturn(deletedRowPos);
+ when(deleteFilter.deletedRowPositions()).thenReturn(positionIndex(98, 99,
100, 101, 102));
var rowIdMapping = ColumnarBatchUtil.buildRowIdMapping(columnVectors,
deleteFilter, 0, 200);
assertThat(rowIdMapping).isNotNull();
@@ -96,6 +91,7 @@ public class TestColumnarBatchUtil {
// Mock DeleteFilter
when(deleteFilter.hasPosDeletes()).thenReturn(false);
+ when(deleteFilter.hasEqDeletes()).thenReturn(true);
when(deleteFilter.deletedRowPositions()).thenReturn(null);
when(deleteFilter.eqDeletedRowFilter()).thenReturn(eqDeletePredicate);
@@ -121,11 +117,9 @@ public class TestColumnarBatchUtil {
Stream.of(rawEqDelete).map(Predicate::negate).reduce(Predicate::and).orElse(t
-> true);
when(deleteFilter.eqDeletedRowFilter()).thenReturn(eqDeletePredicate);
- PositionDeleteIndex deletedRowPos = mock(PositionDeleteIndex.class);
- when(deletedRowPos.isDeleted(1)).thenReturn(true); // 41
- when(deletedRowPos.isDeleted(4)).thenReturn(true); // 44
when(deleteFilter.hasPosDeletes()).thenReturn(true);
- when(deleteFilter.deletedRowPositions()).thenReturn(deletedRowPos);
+ when(deleteFilter.hasEqDeletes()).thenReturn(true);
+ when(deleteFilter.deletedRowPositions()).thenReturn(positionIndex(1, 4));
// 41 and 44
var rowIdMapping = ColumnarBatchUtil.buildRowIdMapping(columnVectors,
deleteFilter, 0, 5);
@@ -141,11 +135,8 @@ public class TestColumnarBatchUtil {
void testBuildRowIdMappingEmptyColumVectors() {
ColumnVector[] columnVectorsZero = new ColumnVector[0];
- PositionDeleteIndex deletedRowPos = mock(PositionDeleteIndex.class);
- when(deletedRowPos.isDeleted(1)).thenReturn(true);
- when(deletedRowPos.isDeleted(4)).thenReturn(true);
when(deleteFilter.hasPosDeletes()).thenReturn(true);
- when(deleteFilter.deletedRowPositions()).thenReturn(deletedRowPos);
+ when(deleteFilter.deletedRowPositions()).thenReturn(positionIndex(1, 4));
var rowIdMapping = ColumnarBatchUtil.buildRowIdMapping(columnVectorsZero,
deleteFilter, 0, 0);
@@ -164,12 +155,9 @@ public class TestColumnarBatchUtil {
Stream.of(rawEqDelete).map(Predicate::negate).reduce(Predicate::and).orElse(t
-> true);
when(deleteFilter.eqDeletedRowFilter()).thenReturn(eqDeletePredicate);
- PositionDeleteIndex deletedRowPos = mock(PositionDeleteIndex.class);
- when(deletedRowPos.isDeleted(0)).thenReturn(true); // 40
- when(deletedRowPos.isDeleted(1)).thenReturn(true); // 41
- when(deletedRowPos.isDeleted(4)).thenReturn(true); // 44
when(deleteFilter.hasPosDeletes()).thenReturn(true);
- when(deleteFilter.deletedRowPositions()).thenReturn(deletedRowPos);
+ when(deleteFilter.hasEqDeletes()).thenReturn(true);
+ when(deleteFilter.deletedRowPositions()).thenReturn(positionIndex(0, 1,
4)); // 40, 41, 44
var rowIdMapping = ColumnarBatchUtil.buildRowIdMapping(columnVectors,
deleteFilter, 0, 5);
@@ -184,12 +172,7 @@ public class TestColumnarBatchUtil {
@Test
void testBuildIsDeletedPositionDeletes() {
- PositionDeleteIndex deletedRowPos = mock(PositionDeleteIndex.class);
- when(deleteFilter.deletedRowPositions()).thenReturn(deletedRowPos);
-
- for (long i = 98; i < 100; i++) {
- when(deletedRowPos.isDeleted(i)).thenReturn(true);
- }
+ when(deleteFilter.deletedRowPositions()).thenReturn(positionIndex(98, 99));
var isDeleted = ColumnarBatchUtil.buildIsDeleted(columnVectors,
deleteFilter, 0, 100);
@@ -213,6 +196,7 @@ public class TestColumnarBatchUtil {
// Mimic real eqDeletedRowFilter(): keep row only if it does NOT match
delete condition
Predicate<InternalRow> eqDeletePredicate =
Stream.of(rawEqDelete).map(Predicate::negate).reduce(Predicate::and).orElse(t
-> true);
+ when(deleteFilter.hasEqDeletes()).thenReturn(true);
when(deleteFilter.eqDeletedRowFilter()).thenReturn(eqDeletePredicate);
var isDeleted = ColumnarBatchUtil.buildIsDeleted(columnVectors,
deleteFilter, 0, 5);
@@ -236,11 +220,9 @@ public class TestColumnarBatchUtil {
Stream.of(rawEqDelete).map(Predicate::negate).reduce(Predicate::and).orElse(t
-> true);
when(deleteFilter.eqDeletedRowFilter()).thenReturn(eqDeletePredicate);
- PositionDeleteIndex deletedRowPos = mock(PositionDeleteIndex.class);
- when(deletedRowPos.isDeleted(1)).thenReturn(true); // 41
- when(deletedRowPos.isDeleted(4)).thenReturn(true); // 44
when(deleteFilter.hasPosDeletes()).thenReturn(true);
- when(deleteFilter.deletedRowPositions()).thenReturn(deletedRowPos);
+ when(deleteFilter.hasEqDeletes()).thenReturn(true);
+ when(deleteFilter.deletedRowPositions()).thenReturn(positionIndex(1, 4));
// 41 and 44
var isDeleted = ColumnarBatchUtil.buildIsDeleted(columnVectors,
deleteFilter, 0, 5);
@@ -288,6 +270,116 @@ public class TestColumnarBatchUtil {
assertThat(result.length).isEqualTo(3);
}
+ @Test
+ void testBuildRowIdMappingNonZeroBatchStart() {
+ // batches after the first one start at a non-zero position in the file
+ long rowStartPosInBatch = 1_000_000L;
+ when(deleteFilter.hasPosDeletes()).thenReturn(true);
+ when(deleteFilter.deletedRowPositions())
+ .thenReturn(positionIndex(1_000_000L, 1_000_003L, 1_000_009L));
+
+ var rowIdMapping =
+ ColumnarBatchUtil.buildRowIdMapping(columnVectors, deleteFilter,
rowStartPosInBatch, 10);
+
+ assertThat(rowIdMapping).isNotNull();
+ int[] rowIds = (int[]) rowIdMapping.first();
+ int liveRows = (Integer) rowIdMapping.second();
+
+ assertThat(liveRows).isEqualTo(7);
+ assertThat(Arrays.copyOf(rowIds, liveRows)).containsExactly(1, 2, 4, 5, 6,
7, 8);
+ verify(deleteFilter, times(3)).incrementDeleteCount();
+ }
+
+ @Test
+ void testBuildRowIdMappingDeletesOutsideBatch() {
+ // the index covers the whole file, so most batches see no deletes at all
+ when(deleteFilter.hasPosDeletes()).thenReturn(true);
+ when(deleteFilter.deletedRowPositions()).thenReturn(positionIndex(5L, 6L,
500L, 501L));
+
+ var rowIdMapping = ColumnarBatchUtil.buildRowIdMapping(columnVectors,
deleteFilter, 100L, 100);
+
+ assertThat(rowIdMapping).isNull();
+ verify(deleteFilter, times(0)).incrementDeleteCount();
+ }
+
+ @Test
+ void testBuildRowIdMappingAllRowsDeletedByPosition() {
+ when(deleteFilter.hasPosDeletes()).thenReturn(true);
+ when(deleteFilter.deletedRowPositions()).thenReturn(positionIndex(0L, 1L,
2L, 3L, 4L));
+
+ var rowIdMapping = ColumnarBatchUtil.buildRowIdMapping(columnVectors,
deleteFilter, 0, 5);
+
+ assertThat(rowIdMapping).isNotNull();
+ assertThat((Integer) rowIdMapping.second()).isEqualTo(0);
+ verify(deleteFilter, times(5)).incrementDeleteCount();
+ }
+
+ @Test
+ void testBuildIsDeletedNonZeroBatchStart() {
+
when(deleteFilter.deletedRowPositions()).thenReturn(positionIndex(1_000_002L,
1_000_004L));
+
+ var isDeleted = ColumnarBatchUtil.buildIsDeleted(columnVectors,
deleteFilter, 1_000_000L, 5);
+
+ assertThat(isDeleted).containsExactly(false, false, true, false, true);
+ verify(deleteFilter, times(2)).incrementDeleteCount();
+ }
+
+ @Test
+ void testPositionOnlyPathMatchesPerRowPath() {
+ // the range traversal must agree with a straightforward per-row probe on
every batch
+ Random random = new Random(20260818L);
+ long fileSize = 300_000L;
+ List<Long> deletedPositions = Lists.newArrayList();
+ for (long pos = 0; pos < fileSize; pos++) {
+ if (random.nextInt(1000) < 7) {
+ deletedPositions.add(pos);
+ }
+ }
+
+ PositionDeleteIndex index = positionIndex(deletedPositions);
+ int batchSize = 5000;
+
+ for (long batchStart = 0; batchStart < fileSize; batchStart += batchSize) {
+ DeleteFilter<InternalRow> filter = mock(DeleteFilter.class);
+ when(filter.deletedRowPositions()).thenReturn(index);
+
+ var rowIdMapping =
+ ColumnarBatchUtil.buildRowIdMapping(columnVectors, filter,
batchStart, batchSize);
+
+ int[] expected = new int[batchSize];
+ int expectedLiveRows = 0;
+ for (int rowId = 0; rowId < batchSize; rowId++) {
+ if (!index.isDeleted(batchStart + rowId)) {
+ expected[expectedLiveRows] = rowId;
+ expectedLiveRows++;
+ }
+ }
+
+ if (expectedLiveRows == batchSize) {
+ assertThat(rowIdMapping).as("batch at %s", batchStart).isNull();
+ } else {
+ assertThat(rowIdMapping).as("batch at %s", batchStart).isNotNull();
+ assertThat((Integer)
rowIdMapping.second()).isEqualTo(expectedLiveRows);
+ assertThat(Arrays.copyOf((int[]) rowIdMapping.first(),
expectedLiveRows))
+ .as("batch at %s", batchStart)
+ .containsExactly(Arrays.copyOf(expected, expectedLiveRows));
+ verify(filter, times(batchSize -
expectedLiveRows)).incrementDeleteCount();
+ }
+ }
+ }
+
+ private static PositionDeleteIndex positionIndex(long... positions) {
+ List<Long> list = Lists.newArrayList();
+ for (long position : positions) {
+ list.add(position);
+ }
+ return positionIndex(list);
+ }
+
+ private static PositionDeleteIndex positionIndex(List<Long> positions) {
+ return Deletes.toPositionIndex(CloseableIterable.withNoopClose(positions));
+ }
+
private ColumnVector[] mockColumnVector() {
// Create a mocked Int column vector with values: 40, 41, 42, 43, 44
ColumnVector intVector = mock(ColumnVector.class);
diff --git
a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchUtil.java
b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchUtil.java
index 89fe4538bd..bddc144fb2 100644
---
a/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchUtil.java
+++
b/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchUtil.java
@@ -19,6 +19,7 @@
package org.apache.iceberg.spark.data.vectorized;
import java.util.Arrays;
+import java.util.function.LongConsumer;
import java.util.function.Predicate;
import org.apache.iceberg.data.DeleteFilter;
import org.apache.iceberg.deletes.PositionDeleteIndex;
@@ -64,6 +65,15 @@ public class ColumnarBatchUtil {
}
PositionDeleteIndex deletedPositions = deletes.deletedRowPositions();
+
+ if (!deletes.hasEqDeletes()) {
+ // positions in a batch form a contiguous ascending range, so the index
can be traversed once
+ // for the whole range instead of being probed once per row
+ return deletedPositions == null
+ ? null
+ : buildRowIdMapping(deletedPositions, deletes, rowStartPosInBatch,
batchSize);
+ }
+
Predicate<InternalRow> eqDeleteFilter = deletes.eqDeletedRowFilter();
ColumnarBatchRow row = new ColumnarBatchRow(columnVectors);
int[] rowIdMapping = new int[batchSize];
@@ -83,6 +93,21 @@ public class ColumnarBatchUtil {
return liveRowId == batchSize ? null : Pair.of(rowIdMapping, liveRowId);
}
+ // builds the mapping from the deleted positions in the batch range, without
probing per row
+ private static Pair<int[], Integer> buildRowIdMapping(
+ PositionDeleteIndex deletedPositions,
+ DeleteFilter<InternalRow> deletes,
+ long rowStartPosInBatch,
+ int batchSize) {
+ RowIdMappingCollector collector =
+ new RowIdMappingCollector(deletes, rowStartPosInBatch, batchSize);
+ deletedPositions.forEachInRange(rowStartPosInBatch, rowStartPosInBatch +
batchSize, collector);
+ collector.appendRemainingLiveRows();
+
+ int liveRowId = collector.liveRowCount();
+ return liveRowId == batchSize ? null : Pair.of(collector.rowIdMapping(),
liveRowId);
+ }
+
/**
* Builds a boolean array to indicate if a row is deleted or not.
*
@@ -119,6 +144,20 @@ public class ColumnarBatchUtil {
}
PositionDeleteIndex deletedPositions = deletes.deletedRowPositions();
+
+ if (!deletes.hasEqDeletes()) {
+ // positions in a batch form a contiguous ascending range, so the index
can be traversed once
+ // for the whole range instead of being probed once per row
+ if (deletedPositions != null) {
+ IsDeletedCollector collector =
+ new IsDeletedCollector(deletes, isDeleted, rowStartPosInBatch);
+ deletedPositions.forEachInRange(
+ rowStartPosInBatch, rowStartPosInBatch + batchSize, collector);
+ }
+
+ return isDeleted;
+ }
+
Predicate<InternalRow> eqDeleteFilter = deletes.eqDeletedRowFilter();
ColumnarBatchRow row = new ColumnarBatchRow(columnVectors);
@@ -178,4 +217,72 @@ public class ColumnarBatchUtil {
return columnVectors;
}
}
+
+ /**
+ * Consumes deleted positions in ascending order, filling the gaps between
them with live row IDs.
+ */
+ private static class RowIdMappingCollector implements LongConsumer {
+ private final DeleteFilter<InternalRow> deletes;
+ private final long rowStartPosInBatch;
+ private final int batchSize;
+ private final int[] rowIdMapping;
+ private int nextRowId = 0;
+ private int liveRowId = 0;
+
+ RowIdMappingCollector(
+ DeleteFilter<InternalRow> deletes, long rowStartPosInBatch, int
batchSize) {
+ this.deletes = deletes;
+ this.rowStartPosInBatch = rowStartPosInBatch;
+ this.batchSize = batchSize;
+ this.rowIdMapping = new int[batchSize];
+ }
+
+ @Override
+ public void accept(long pos) {
+ int deletedRowId = (int) (pos - rowStartPosInBatch);
+ for (int rowId = nextRowId; rowId < deletedRowId; rowId++) {
+ rowIdMapping[liveRowId] = rowId;
+ liveRowId++;
+ }
+
+ deletes.incrementDeleteCount();
+ this.nextRowId = deletedRowId + 1;
+ }
+
+ /** Appends the live rows that follow the last deleted position in the
batch. */
+ void appendRemainingLiveRows() {
+ for (int rowId = nextRowId; rowId < batchSize; rowId++) {
+ rowIdMapping[liveRowId] = rowId;
+ liveRowId++;
+ }
+ }
+
+ int liveRowCount() {
+ return liveRowId;
+ }
+
+ int[] rowIdMapping() {
+ return rowIdMapping;
+ }
+ }
+
+ /** Consumes deleted positions in a batch range, marking them in the given
array. */
+ private static class IsDeletedCollector implements LongConsumer {
+ private final DeleteFilter<InternalRow> deletes;
+ private final boolean[] isDeleted;
+ private final long rowStartPosInBatch;
+
+ IsDeletedCollector(
+ DeleteFilter<InternalRow> deletes, boolean[] isDeleted, long
rowStartPosInBatch) {
+ this.deletes = deletes;
+ this.isDeleted = isDeleted;
+ this.rowStartPosInBatch = rowStartPosInBatch;
+ }
+
+ @Override
+ public void accept(long pos) {
+ isDeleted[(int) (pos - rowStartPosInBatch)] = true;
+ deletes.incrementDeleteCount();
+ }
+ }
}
diff --git
a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/data/vectorized/TestColumnarBatchUtil.java
b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/data/vectorized/TestColumnarBatchUtil.java
index dadbe3e788..468242f610 100644
---
a/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/data/vectorized/TestColumnarBatchUtil.java
+++
b/spark/v4.0/spark/src/test/java/org/apache/iceberg/spark/data/vectorized/TestColumnarBatchUtil.java
@@ -21,14 +21,21 @@ package org.apache.iceberg.spark.data.vectorized;
import static java.util.Collections.nCopies;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.util.Arrays;
+import java.util.List;
+import java.util.Random;
import java.util.function.Predicate;
import java.util.stream.Stream;
import org.apache.iceberg.Schema;
import org.apache.iceberg.data.DeleteFilter;
+import org.apache.iceberg.deletes.Deletes;
import org.apache.iceberg.deletes.PositionDeleteIndex;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.spark.sql.catalyst.InternalRow;
import org.apache.spark.sql.vectorized.ColumnVector;
import org.junit.jupiter.api.BeforeEach;
@@ -48,13 +55,7 @@ public class TestColumnarBatchUtil {
@Test
public void testBuildRowIdMappingNoDeletes() {
when(deleteFilter.hasPosDeletes()).thenReturn(true);
- PositionDeleteIndex deletedRowPos = mock(PositionDeleteIndex.class);
-
- for (long i = 0; i <= 10; i++) {
- when(deletedRowPos.isDeleted(i)).thenReturn(false);
- }
-
- when(deleteFilter.deletedRowPositions()).thenReturn(deletedRowPos);
+ when(deleteFilter.deletedRowPositions()).thenReturn(positionIndex());
var rowIdMapping = ColumnarBatchUtil.buildRowIdMapping(columnVectors,
deleteFilter, 0, 10);
assertThat(rowIdMapping).isNull();
}
@@ -62,14 +63,8 @@ public class TestColumnarBatchUtil {
@Test
public void testBuildRowIdMappingPositionDeletesOnly() {
when(deleteFilter.hasPosDeletes()).thenReturn(true);
- PositionDeleteIndex deletedRowPos = mock(PositionDeleteIndex.class);
-
// 5 position deletes
- for (long i = 98; i < 103; i++) {
- when(deletedRowPos.isDeleted(i)).thenReturn(true);
- }
-
- when(deleteFilter.deletedRowPositions()).thenReturn(deletedRowPos);
+ when(deleteFilter.deletedRowPositions()).thenReturn(positionIndex(98, 99,
100, 101, 102));
var rowIdMapping = ColumnarBatchUtil.buildRowIdMapping(columnVectors,
deleteFilter, 0, 200);
assertThat(rowIdMapping).isNotNull();
@@ -96,6 +91,7 @@ public class TestColumnarBatchUtil {
// Mock DeleteFilter
when(deleteFilter.hasPosDeletes()).thenReturn(false);
+ when(deleteFilter.hasEqDeletes()).thenReturn(true);
when(deleteFilter.deletedRowPositions()).thenReturn(null);
when(deleteFilter.eqDeletedRowFilter()).thenReturn(eqDeletePredicate);
@@ -121,11 +117,9 @@ public class TestColumnarBatchUtil {
Stream.of(rawEqDelete).map(Predicate::negate).reduce(Predicate::and).orElse(t
-> true);
when(deleteFilter.eqDeletedRowFilter()).thenReturn(eqDeletePredicate);
- PositionDeleteIndex deletedRowPos = mock(PositionDeleteIndex.class);
- when(deletedRowPos.isDeleted(1)).thenReturn(true); // 41
- when(deletedRowPos.isDeleted(4)).thenReturn(true); // 44
when(deleteFilter.hasPosDeletes()).thenReturn(true);
- when(deleteFilter.deletedRowPositions()).thenReturn(deletedRowPos);
+ when(deleteFilter.hasEqDeletes()).thenReturn(true);
+ when(deleteFilter.deletedRowPositions()).thenReturn(positionIndex(1, 4));
// 41 and 44
var rowIdMapping = ColumnarBatchUtil.buildRowIdMapping(columnVectors,
deleteFilter, 0, 5);
@@ -141,11 +135,8 @@ public class TestColumnarBatchUtil {
void testBuildRowIdMappingEmptyColumVectors() {
ColumnVector[] columnVectorsZero = new ColumnVector[0];
- PositionDeleteIndex deletedRowPos = mock(PositionDeleteIndex.class);
- when(deletedRowPos.isDeleted(1)).thenReturn(true);
- when(deletedRowPos.isDeleted(4)).thenReturn(true);
when(deleteFilter.hasPosDeletes()).thenReturn(true);
- when(deleteFilter.deletedRowPositions()).thenReturn(deletedRowPos);
+ when(deleteFilter.deletedRowPositions()).thenReturn(positionIndex(1, 4));
var rowIdMapping = ColumnarBatchUtil.buildRowIdMapping(columnVectorsZero,
deleteFilter, 0, 0);
@@ -164,12 +155,9 @@ public class TestColumnarBatchUtil {
Stream.of(rawEqDelete).map(Predicate::negate).reduce(Predicate::and).orElse(t
-> true);
when(deleteFilter.eqDeletedRowFilter()).thenReturn(eqDeletePredicate);
- PositionDeleteIndex deletedRowPos = mock(PositionDeleteIndex.class);
- when(deletedRowPos.isDeleted(0)).thenReturn(true); // 40
- when(deletedRowPos.isDeleted(1)).thenReturn(true); // 41
- when(deletedRowPos.isDeleted(4)).thenReturn(true); // 44
when(deleteFilter.hasPosDeletes()).thenReturn(true);
- when(deleteFilter.deletedRowPositions()).thenReturn(deletedRowPos);
+ when(deleteFilter.hasEqDeletes()).thenReturn(true);
+ when(deleteFilter.deletedRowPositions()).thenReturn(positionIndex(0, 1,
4)); // 40, 41, 44
var rowIdMapping = ColumnarBatchUtil.buildRowIdMapping(columnVectors,
deleteFilter, 0, 5);
@@ -184,12 +172,7 @@ public class TestColumnarBatchUtil {
@Test
void testBuildIsDeletedPositionDeletes() {
- PositionDeleteIndex deletedRowPos = mock(PositionDeleteIndex.class);
- when(deleteFilter.deletedRowPositions()).thenReturn(deletedRowPos);
-
- for (long i = 98; i < 100; i++) {
- when(deletedRowPos.isDeleted(i)).thenReturn(true);
- }
+ when(deleteFilter.deletedRowPositions()).thenReturn(positionIndex(98, 99));
var isDeleted = ColumnarBatchUtil.buildIsDeleted(columnVectors,
deleteFilter, 0, 100);
@@ -213,6 +196,7 @@ public class TestColumnarBatchUtil {
// Mimic real eqDeletedRowFilter(): keep row only if it does NOT match
delete condition
Predicate<InternalRow> eqDeletePredicate =
Stream.of(rawEqDelete).map(Predicate::negate).reduce(Predicate::and).orElse(t
-> true);
+ when(deleteFilter.hasEqDeletes()).thenReturn(true);
when(deleteFilter.eqDeletedRowFilter()).thenReturn(eqDeletePredicate);
var isDeleted = ColumnarBatchUtil.buildIsDeleted(columnVectors,
deleteFilter, 0, 5);
@@ -236,11 +220,9 @@ public class TestColumnarBatchUtil {
Stream.of(rawEqDelete).map(Predicate::negate).reduce(Predicate::and).orElse(t
-> true);
when(deleteFilter.eqDeletedRowFilter()).thenReturn(eqDeletePredicate);
- PositionDeleteIndex deletedRowPos = mock(PositionDeleteIndex.class);
- when(deletedRowPos.isDeleted(1)).thenReturn(true); // 41
- when(deletedRowPos.isDeleted(4)).thenReturn(true); // 44
when(deleteFilter.hasPosDeletes()).thenReturn(true);
- when(deleteFilter.deletedRowPositions()).thenReturn(deletedRowPos);
+ when(deleteFilter.hasEqDeletes()).thenReturn(true);
+ when(deleteFilter.deletedRowPositions()).thenReturn(positionIndex(1, 4));
// 41 and 44
var isDeleted = ColumnarBatchUtil.buildIsDeleted(columnVectors,
deleteFilter, 0, 5);
@@ -288,6 +270,116 @@ public class TestColumnarBatchUtil {
assertThat(result.length).isEqualTo(3);
}
+ @Test
+ void testBuildRowIdMappingNonZeroBatchStart() {
+ // batches after the first one start at a non-zero position in the file
+ long rowStartPosInBatch = 1_000_000L;
+ when(deleteFilter.hasPosDeletes()).thenReturn(true);
+ when(deleteFilter.deletedRowPositions())
+ .thenReturn(positionIndex(1_000_000L, 1_000_003L, 1_000_009L));
+
+ var rowIdMapping =
+ ColumnarBatchUtil.buildRowIdMapping(columnVectors, deleteFilter,
rowStartPosInBatch, 10);
+
+ assertThat(rowIdMapping).isNotNull();
+ int[] rowIds = (int[]) rowIdMapping.first();
+ int liveRows = (Integer) rowIdMapping.second();
+
+ assertThat(liveRows).isEqualTo(7);
+ assertThat(Arrays.copyOf(rowIds, liveRows)).containsExactly(1, 2, 4, 5, 6,
7, 8);
+ verify(deleteFilter, times(3)).incrementDeleteCount();
+ }
+
+ @Test
+ void testBuildRowIdMappingDeletesOutsideBatch() {
+ // the index covers the whole file, so most batches see no deletes at all
+ when(deleteFilter.hasPosDeletes()).thenReturn(true);
+ when(deleteFilter.deletedRowPositions()).thenReturn(positionIndex(5L, 6L,
500L, 501L));
+
+ var rowIdMapping = ColumnarBatchUtil.buildRowIdMapping(columnVectors,
deleteFilter, 100L, 100);
+
+ assertThat(rowIdMapping).isNull();
+ verify(deleteFilter, times(0)).incrementDeleteCount();
+ }
+
+ @Test
+ void testBuildRowIdMappingAllRowsDeletedByPosition() {
+ when(deleteFilter.hasPosDeletes()).thenReturn(true);
+ when(deleteFilter.deletedRowPositions()).thenReturn(positionIndex(0L, 1L,
2L, 3L, 4L));
+
+ var rowIdMapping = ColumnarBatchUtil.buildRowIdMapping(columnVectors,
deleteFilter, 0, 5);
+
+ assertThat(rowIdMapping).isNotNull();
+ assertThat((Integer) rowIdMapping.second()).isEqualTo(0);
+ verify(deleteFilter, times(5)).incrementDeleteCount();
+ }
+
+ @Test
+ void testBuildIsDeletedNonZeroBatchStart() {
+
when(deleteFilter.deletedRowPositions()).thenReturn(positionIndex(1_000_002L,
1_000_004L));
+
+ var isDeleted = ColumnarBatchUtil.buildIsDeleted(columnVectors,
deleteFilter, 1_000_000L, 5);
+
+ assertThat(isDeleted).containsExactly(false, false, true, false, true);
+ verify(deleteFilter, times(2)).incrementDeleteCount();
+ }
+
+ @Test
+ void testPositionOnlyPathMatchesPerRowPath() {
+ // the range traversal must agree with a straightforward per-row probe on
every batch
+ Random random = new Random(20260818L);
+ long fileSize = 300_000L;
+ List<Long> deletedPositions = Lists.newArrayList();
+ for (long pos = 0; pos < fileSize; pos++) {
+ if (random.nextInt(1000) < 7) {
+ deletedPositions.add(pos);
+ }
+ }
+
+ PositionDeleteIndex index = positionIndex(deletedPositions);
+ int batchSize = 5000;
+
+ for (long batchStart = 0; batchStart < fileSize; batchStart += batchSize) {
+ DeleteFilter<InternalRow> filter = mock(DeleteFilter.class);
+ when(filter.deletedRowPositions()).thenReturn(index);
+
+ var rowIdMapping =
+ ColumnarBatchUtil.buildRowIdMapping(columnVectors, filter,
batchStart, batchSize);
+
+ int[] expected = new int[batchSize];
+ int expectedLiveRows = 0;
+ for (int rowId = 0; rowId < batchSize; rowId++) {
+ if (!index.isDeleted(batchStart + rowId)) {
+ expected[expectedLiveRows] = rowId;
+ expectedLiveRows++;
+ }
+ }
+
+ if (expectedLiveRows == batchSize) {
+ assertThat(rowIdMapping).as("batch at %s", batchStart).isNull();
+ } else {
+ assertThat(rowIdMapping).as("batch at %s", batchStart).isNotNull();
+ assertThat((Integer)
rowIdMapping.second()).isEqualTo(expectedLiveRows);
+ assertThat(Arrays.copyOf((int[]) rowIdMapping.first(),
expectedLiveRows))
+ .as("batch at %s", batchStart)
+ .containsExactly(Arrays.copyOf(expected, expectedLiveRows));
+ verify(filter, times(batchSize -
expectedLiveRows)).incrementDeleteCount();
+ }
+ }
+ }
+
+ private static PositionDeleteIndex positionIndex(long... positions) {
+ List<Long> list = Lists.newArrayList();
+ for (long position : positions) {
+ list.add(position);
+ }
+ return positionIndex(list);
+ }
+
+ private static PositionDeleteIndex positionIndex(List<Long> positions) {
+ return Deletes.toPositionIndex(CloseableIterable.withNoopClose(positions));
+ }
+
private ColumnVector[] mockColumnVector() {
// Create a mocked Int column vector with values: 40, 41, 42, 43, 44
ColumnVector intVector = mock(ColumnVector.class);
diff --git
a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchUtil.java
b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchUtil.java
index 89fe4538bd..bddc144fb2 100644
---
a/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchUtil.java
+++
b/spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ColumnarBatchUtil.java
@@ -19,6 +19,7 @@
package org.apache.iceberg.spark.data.vectorized;
import java.util.Arrays;
+import java.util.function.LongConsumer;
import java.util.function.Predicate;
import org.apache.iceberg.data.DeleteFilter;
import org.apache.iceberg.deletes.PositionDeleteIndex;
@@ -64,6 +65,15 @@ public class ColumnarBatchUtil {
}
PositionDeleteIndex deletedPositions = deletes.deletedRowPositions();
+
+ if (!deletes.hasEqDeletes()) {
+ // positions in a batch form a contiguous ascending range, so the index
can be traversed once
+ // for the whole range instead of being probed once per row
+ return deletedPositions == null
+ ? null
+ : buildRowIdMapping(deletedPositions, deletes, rowStartPosInBatch,
batchSize);
+ }
+
Predicate<InternalRow> eqDeleteFilter = deletes.eqDeletedRowFilter();
ColumnarBatchRow row = new ColumnarBatchRow(columnVectors);
int[] rowIdMapping = new int[batchSize];
@@ -83,6 +93,21 @@ public class ColumnarBatchUtil {
return liveRowId == batchSize ? null : Pair.of(rowIdMapping, liveRowId);
}
+ // builds the mapping from the deleted positions in the batch range, without
probing per row
+ private static Pair<int[], Integer> buildRowIdMapping(
+ PositionDeleteIndex deletedPositions,
+ DeleteFilter<InternalRow> deletes,
+ long rowStartPosInBatch,
+ int batchSize) {
+ RowIdMappingCollector collector =
+ new RowIdMappingCollector(deletes, rowStartPosInBatch, batchSize);
+ deletedPositions.forEachInRange(rowStartPosInBatch, rowStartPosInBatch +
batchSize, collector);
+ collector.appendRemainingLiveRows();
+
+ int liveRowId = collector.liveRowCount();
+ return liveRowId == batchSize ? null : Pair.of(collector.rowIdMapping(),
liveRowId);
+ }
+
/**
* Builds a boolean array to indicate if a row is deleted or not.
*
@@ -119,6 +144,20 @@ public class ColumnarBatchUtil {
}
PositionDeleteIndex deletedPositions = deletes.deletedRowPositions();
+
+ if (!deletes.hasEqDeletes()) {
+ // positions in a batch form a contiguous ascending range, so the index
can be traversed once
+ // for the whole range instead of being probed once per row
+ if (deletedPositions != null) {
+ IsDeletedCollector collector =
+ new IsDeletedCollector(deletes, isDeleted, rowStartPosInBatch);
+ deletedPositions.forEachInRange(
+ rowStartPosInBatch, rowStartPosInBatch + batchSize, collector);
+ }
+
+ return isDeleted;
+ }
+
Predicate<InternalRow> eqDeleteFilter = deletes.eqDeletedRowFilter();
ColumnarBatchRow row = new ColumnarBatchRow(columnVectors);
@@ -178,4 +217,72 @@ public class ColumnarBatchUtil {
return columnVectors;
}
}
+
+ /**
+ * Consumes deleted positions in ascending order, filling the gaps between
them with live row IDs.
+ */
+ private static class RowIdMappingCollector implements LongConsumer {
+ private final DeleteFilter<InternalRow> deletes;
+ private final long rowStartPosInBatch;
+ private final int batchSize;
+ private final int[] rowIdMapping;
+ private int nextRowId = 0;
+ private int liveRowId = 0;
+
+ RowIdMappingCollector(
+ DeleteFilter<InternalRow> deletes, long rowStartPosInBatch, int
batchSize) {
+ this.deletes = deletes;
+ this.rowStartPosInBatch = rowStartPosInBatch;
+ this.batchSize = batchSize;
+ this.rowIdMapping = new int[batchSize];
+ }
+
+ @Override
+ public void accept(long pos) {
+ int deletedRowId = (int) (pos - rowStartPosInBatch);
+ for (int rowId = nextRowId; rowId < deletedRowId; rowId++) {
+ rowIdMapping[liveRowId] = rowId;
+ liveRowId++;
+ }
+
+ deletes.incrementDeleteCount();
+ this.nextRowId = deletedRowId + 1;
+ }
+
+ /** Appends the live rows that follow the last deleted position in the
batch. */
+ void appendRemainingLiveRows() {
+ for (int rowId = nextRowId; rowId < batchSize; rowId++) {
+ rowIdMapping[liveRowId] = rowId;
+ liveRowId++;
+ }
+ }
+
+ int liveRowCount() {
+ return liveRowId;
+ }
+
+ int[] rowIdMapping() {
+ return rowIdMapping;
+ }
+ }
+
+ /** Consumes deleted positions in a batch range, marking them in the given
array. */
+ private static class IsDeletedCollector implements LongConsumer {
+ private final DeleteFilter<InternalRow> deletes;
+ private final boolean[] isDeleted;
+ private final long rowStartPosInBatch;
+
+ IsDeletedCollector(
+ DeleteFilter<InternalRow> deletes, boolean[] isDeleted, long
rowStartPosInBatch) {
+ this.deletes = deletes;
+ this.isDeleted = isDeleted;
+ this.rowStartPosInBatch = rowStartPosInBatch;
+ }
+
+ @Override
+ public void accept(long pos) {
+ isDeleted[(int) (pos - rowStartPosInBatch)] = true;
+ deletes.incrementDeleteCount();
+ }
+ }
}
diff --git
a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/vectorized/TestColumnarBatchUtil.java
b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/vectorized/TestColumnarBatchUtil.java
index dadbe3e788..468242f610 100644
---
a/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/vectorized/TestColumnarBatchUtil.java
+++
b/spark/v4.1/spark/src/test/java/org/apache/iceberg/spark/data/vectorized/TestColumnarBatchUtil.java
@@ -21,14 +21,21 @@ package org.apache.iceberg.spark.data.vectorized;
import static java.util.Collections.nCopies;
import static org.assertj.core.api.Assertions.assertThat;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.util.Arrays;
+import java.util.List;
+import java.util.Random;
import java.util.function.Predicate;
import java.util.stream.Stream;
import org.apache.iceberg.Schema;
import org.apache.iceberg.data.DeleteFilter;
+import org.apache.iceberg.deletes.Deletes;
import org.apache.iceberg.deletes.PositionDeleteIndex;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
import org.apache.spark.sql.catalyst.InternalRow;
import org.apache.spark.sql.vectorized.ColumnVector;
import org.junit.jupiter.api.BeforeEach;
@@ -48,13 +55,7 @@ public class TestColumnarBatchUtil {
@Test
public void testBuildRowIdMappingNoDeletes() {
when(deleteFilter.hasPosDeletes()).thenReturn(true);
- PositionDeleteIndex deletedRowPos = mock(PositionDeleteIndex.class);
-
- for (long i = 0; i <= 10; i++) {
- when(deletedRowPos.isDeleted(i)).thenReturn(false);
- }
-
- when(deleteFilter.deletedRowPositions()).thenReturn(deletedRowPos);
+ when(deleteFilter.deletedRowPositions()).thenReturn(positionIndex());
var rowIdMapping = ColumnarBatchUtil.buildRowIdMapping(columnVectors,
deleteFilter, 0, 10);
assertThat(rowIdMapping).isNull();
}
@@ -62,14 +63,8 @@ public class TestColumnarBatchUtil {
@Test
public void testBuildRowIdMappingPositionDeletesOnly() {
when(deleteFilter.hasPosDeletes()).thenReturn(true);
- PositionDeleteIndex deletedRowPos = mock(PositionDeleteIndex.class);
-
// 5 position deletes
- for (long i = 98; i < 103; i++) {
- when(deletedRowPos.isDeleted(i)).thenReturn(true);
- }
-
- when(deleteFilter.deletedRowPositions()).thenReturn(deletedRowPos);
+ when(deleteFilter.deletedRowPositions()).thenReturn(positionIndex(98, 99,
100, 101, 102));
var rowIdMapping = ColumnarBatchUtil.buildRowIdMapping(columnVectors,
deleteFilter, 0, 200);
assertThat(rowIdMapping).isNotNull();
@@ -96,6 +91,7 @@ public class TestColumnarBatchUtil {
// Mock DeleteFilter
when(deleteFilter.hasPosDeletes()).thenReturn(false);
+ when(deleteFilter.hasEqDeletes()).thenReturn(true);
when(deleteFilter.deletedRowPositions()).thenReturn(null);
when(deleteFilter.eqDeletedRowFilter()).thenReturn(eqDeletePredicate);
@@ -121,11 +117,9 @@ public class TestColumnarBatchUtil {
Stream.of(rawEqDelete).map(Predicate::negate).reduce(Predicate::and).orElse(t
-> true);
when(deleteFilter.eqDeletedRowFilter()).thenReturn(eqDeletePredicate);
- PositionDeleteIndex deletedRowPos = mock(PositionDeleteIndex.class);
- when(deletedRowPos.isDeleted(1)).thenReturn(true); // 41
- when(deletedRowPos.isDeleted(4)).thenReturn(true); // 44
when(deleteFilter.hasPosDeletes()).thenReturn(true);
- when(deleteFilter.deletedRowPositions()).thenReturn(deletedRowPos);
+ when(deleteFilter.hasEqDeletes()).thenReturn(true);
+ when(deleteFilter.deletedRowPositions()).thenReturn(positionIndex(1, 4));
// 41 and 44
var rowIdMapping = ColumnarBatchUtil.buildRowIdMapping(columnVectors,
deleteFilter, 0, 5);
@@ -141,11 +135,8 @@ public class TestColumnarBatchUtil {
void testBuildRowIdMappingEmptyColumVectors() {
ColumnVector[] columnVectorsZero = new ColumnVector[0];
- PositionDeleteIndex deletedRowPos = mock(PositionDeleteIndex.class);
- when(deletedRowPos.isDeleted(1)).thenReturn(true);
- when(deletedRowPos.isDeleted(4)).thenReturn(true);
when(deleteFilter.hasPosDeletes()).thenReturn(true);
- when(deleteFilter.deletedRowPositions()).thenReturn(deletedRowPos);
+ when(deleteFilter.deletedRowPositions()).thenReturn(positionIndex(1, 4));
var rowIdMapping = ColumnarBatchUtil.buildRowIdMapping(columnVectorsZero,
deleteFilter, 0, 0);
@@ -164,12 +155,9 @@ public class TestColumnarBatchUtil {
Stream.of(rawEqDelete).map(Predicate::negate).reduce(Predicate::and).orElse(t
-> true);
when(deleteFilter.eqDeletedRowFilter()).thenReturn(eqDeletePredicate);
- PositionDeleteIndex deletedRowPos = mock(PositionDeleteIndex.class);
- when(deletedRowPos.isDeleted(0)).thenReturn(true); // 40
- when(deletedRowPos.isDeleted(1)).thenReturn(true); // 41
- when(deletedRowPos.isDeleted(4)).thenReturn(true); // 44
when(deleteFilter.hasPosDeletes()).thenReturn(true);
- when(deleteFilter.deletedRowPositions()).thenReturn(deletedRowPos);
+ when(deleteFilter.hasEqDeletes()).thenReturn(true);
+ when(deleteFilter.deletedRowPositions()).thenReturn(positionIndex(0, 1,
4)); // 40, 41, 44
var rowIdMapping = ColumnarBatchUtil.buildRowIdMapping(columnVectors,
deleteFilter, 0, 5);
@@ -184,12 +172,7 @@ public class TestColumnarBatchUtil {
@Test
void testBuildIsDeletedPositionDeletes() {
- PositionDeleteIndex deletedRowPos = mock(PositionDeleteIndex.class);
- when(deleteFilter.deletedRowPositions()).thenReturn(deletedRowPos);
-
- for (long i = 98; i < 100; i++) {
- when(deletedRowPos.isDeleted(i)).thenReturn(true);
- }
+ when(deleteFilter.deletedRowPositions()).thenReturn(positionIndex(98, 99));
var isDeleted = ColumnarBatchUtil.buildIsDeleted(columnVectors,
deleteFilter, 0, 100);
@@ -213,6 +196,7 @@ public class TestColumnarBatchUtil {
// Mimic real eqDeletedRowFilter(): keep row only if it does NOT match
delete condition
Predicate<InternalRow> eqDeletePredicate =
Stream.of(rawEqDelete).map(Predicate::negate).reduce(Predicate::and).orElse(t
-> true);
+ when(deleteFilter.hasEqDeletes()).thenReturn(true);
when(deleteFilter.eqDeletedRowFilter()).thenReturn(eqDeletePredicate);
var isDeleted = ColumnarBatchUtil.buildIsDeleted(columnVectors,
deleteFilter, 0, 5);
@@ -236,11 +220,9 @@ public class TestColumnarBatchUtil {
Stream.of(rawEqDelete).map(Predicate::negate).reduce(Predicate::and).orElse(t
-> true);
when(deleteFilter.eqDeletedRowFilter()).thenReturn(eqDeletePredicate);
- PositionDeleteIndex deletedRowPos = mock(PositionDeleteIndex.class);
- when(deletedRowPos.isDeleted(1)).thenReturn(true); // 41
- when(deletedRowPos.isDeleted(4)).thenReturn(true); // 44
when(deleteFilter.hasPosDeletes()).thenReturn(true);
- when(deleteFilter.deletedRowPositions()).thenReturn(deletedRowPos);
+ when(deleteFilter.hasEqDeletes()).thenReturn(true);
+ when(deleteFilter.deletedRowPositions()).thenReturn(positionIndex(1, 4));
// 41 and 44
var isDeleted = ColumnarBatchUtil.buildIsDeleted(columnVectors,
deleteFilter, 0, 5);
@@ -288,6 +270,116 @@ public class TestColumnarBatchUtil {
assertThat(result.length).isEqualTo(3);
}
+ @Test
+ void testBuildRowIdMappingNonZeroBatchStart() {
+ // batches after the first one start at a non-zero position in the file
+ long rowStartPosInBatch = 1_000_000L;
+ when(deleteFilter.hasPosDeletes()).thenReturn(true);
+ when(deleteFilter.deletedRowPositions())
+ .thenReturn(positionIndex(1_000_000L, 1_000_003L, 1_000_009L));
+
+ var rowIdMapping =
+ ColumnarBatchUtil.buildRowIdMapping(columnVectors, deleteFilter,
rowStartPosInBatch, 10);
+
+ assertThat(rowIdMapping).isNotNull();
+ int[] rowIds = (int[]) rowIdMapping.first();
+ int liveRows = (Integer) rowIdMapping.second();
+
+ assertThat(liveRows).isEqualTo(7);
+ assertThat(Arrays.copyOf(rowIds, liveRows)).containsExactly(1, 2, 4, 5, 6,
7, 8);
+ verify(deleteFilter, times(3)).incrementDeleteCount();
+ }
+
+ @Test
+ void testBuildRowIdMappingDeletesOutsideBatch() {
+ // the index covers the whole file, so most batches see no deletes at all
+ when(deleteFilter.hasPosDeletes()).thenReturn(true);
+ when(deleteFilter.deletedRowPositions()).thenReturn(positionIndex(5L, 6L,
500L, 501L));
+
+ var rowIdMapping = ColumnarBatchUtil.buildRowIdMapping(columnVectors,
deleteFilter, 100L, 100);
+
+ assertThat(rowIdMapping).isNull();
+ verify(deleteFilter, times(0)).incrementDeleteCount();
+ }
+
+ @Test
+ void testBuildRowIdMappingAllRowsDeletedByPosition() {
+ when(deleteFilter.hasPosDeletes()).thenReturn(true);
+ when(deleteFilter.deletedRowPositions()).thenReturn(positionIndex(0L, 1L,
2L, 3L, 4L));
+
+ var rowIdMapping = ColumnarBatchUtil.buildRowIdMapping(columnVectors,
deleteFilter, 0, 5);
+
+ assertThat(rowIdMapping).isNotNull();
+ assertThat((Integer) rowIdMapping.second()).isEqualTo(0);
+ verify(deleteFilter, times(5)).incrementDeleteCount();
+ }
+
+ @Test
+ void testBuildIsDeletedNonZeroBatchStart() {
+
when(deleteFilter.deletedRowPositions()).thenReturn(positionIndex(1_000_002L,
1_000_004L));
+
+ var isDeleted = ColumnarBatchUtil.buildIsDeleted(columnVectors,
deleteFilter, 1_000_000L, 5);
+
+ assertThat(isDeleted).containsExactly(false, false, true, false, true);
+ verify(deleteFilter, times(2)).incrementDeleteCount();
+ }
+
+ @Test
+ void testPositionOnlyPathMatchesPerRowPath() {
+ // the range traversal must agree with a straightforward per-row probe on
every batch
+ Random random = new Random(20260818L);
+ long fileSize = 300_000L;
+ List<Long> deletedPositions = Lists.newArrayList();
+ for (long pos = 0; pos < fileSize; pos++) {
+ if (random.nextInt(1000) < 7) {
+ deletedPositions.add(pos);
+ }
+ }
+
+ PositionDeleteIndex index = positionIndex(deletedPositions);
+ int batchSize = 5000;
+
+ for (long batchStart = 0; batchStart < fileSize; batchStart += batchSize) {
+ DeleteFilter<InternalRow> filter = mock(DeleteFilter.class);
+ when(filter.deletedRowPositions()).thenReturn(index);
+
+ var rowIdMapping =
+ ColumnarBatchUtil.buildRowIdMapping(columnVectors, filter,
batchStart, batchSize);
+
+ int[] expected = new int[batchSize];
+ int expectedLiveRows = 0;
+ for (int rowId = 0; rowId < batchSize; rowId++) {
+ if (!index.isDeleted(batchStart + rowId)) {
+ expected[expectedLiveRows] = rowId;
+ expectedLiveRows++;
+ }
+ }
+
+ if (expectedLiveRows == batchSize) {
+ assertThat(rowIdMapping).as("batch at %s", batchStart).isNull();
+ } else {
+ assertThat(rowIdMapping).as("batch at %s", batchStart).isNotNull();
+ assertThat((Integer)
rowIdMapping.second()).isEqualTo(expectedLiveRows);
+ assertThat(Arrays.copyOf((int[]) rowIdMapping.first(),
expectedLiveRows))
+ .as("batch at %s", batchStart)
+ .containsExactly(Arrays.copyOf(expected, expectedLiveRows));
+ verify(filter, times(batchSize -
expectedLiveRows)).incrementDeleteCount();
+ }
+ }
+ }
+
+ private static PositionDeleteIndex positionIndex(long... positions) {
+ List<Long> list = Lists.newArrayList();
+ for (long position : positions) {
+ list.add(position);
+ }
+ return positionIndex(list);
+ }
+
+ private static PositionDeleteIndex positionIndex(List<Long> positions) {
+ return Deletes.toPositionIndex(CloseableIterable.withNoopClose(positions));
+ }
+
private ColumnVector[] mockColumnVector() {
// Create a mocked Int column vector with values: 40, 41, 42, 43, 44
ColumnVector intVector = mock(ColumnVector.class);
