Support pos-delete in vectorized read.

arrow/src/main/java/org/apache/iceberg/arrow/vectorized/BaseBatchReader.java

@@ -42,7 +42,7 @@ protected BaseBatchReader(List<VectorizedReader<?>> readers) {
   }
 
   @Override
-  public final void setRowGroupInfo(
+  public void setRowGroupInfo(
       PageReadStore pageStore, Map<ColumnPath, ColumnChunkMetaData> metaData, long rowPosition) {
     for (VectorizedArrowReader reader : readers) {
       if (reader != null) {

arrow/src/main/java/org/apache/iceberg/arrow/vectorized/VectorizedReaderBuilder.java

@@ -94,6 +94,10 @@ public VectorizedReader<?> message(
         reorderedFields.add(VectorizedArrowReader.nulls());
       }
     }
+    return vectorizedReader(reorderedFields);
+  }
+
+  protected VectorizedReader<?> vectorizedReader(List<VectorizedReader<?>> reorderedFields) {
     return readerFactory.apply(reorderedFields);
   }
 

build.gradle

@@ -215,6 +215,7 @@ project(':iceberg-core') {
     implementation "com.fasterxml.jackson.core:jackson-databind"
     implementation "com.fasterxml.jackson.core:jackson-core"
     implementation "com.github.ben-manes.caffeine:caffeine"
+    implementation "org.roaringbitmap:RoaringBitmap"
     compileOnly("org.apache.hadoop:hadoop-client") {
       exclude group: 'org.apache.avro', module: 'avro'
       exclude group: 'org.slf4j', module: 'slf4j-log4j12'

core/src/main/java/org/apache/iceberg/deletes/BitmapPositionDeleteIndex.java

@@ -0,0 +1,45 @@
+/*
+ * 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.iceberg.deletes;
+
+import org.roaringbitmap.longlong.Roaring64Bitmap;
+
+class BitmapPositionDeleteIndex implements PositionDeleteIndex {
+  private final Roaring64Bitmap roaring64Bitmap;
+
+  BitmapPositionDeleteIndex() {
+    roaring64Bitmap = new Roaring64Bitmap();
+  }
+
+  @Override
+  public void delete(long position) {
+    roaring64Bitmap.add(position);
+  }
+
+  @Override
+  public void delete(long posStart, long posEnd) {
+    roaring64Bitmap.add(posStart, posEnd);
+  }
+
+  @Override
+  public boolean deleted(long position) {
+    return roaring64Bitmap.contains(position);
+  }
+}

core/src/main/java/org/apache/iceberg/deletes/Deletes.java

@@ -107,6 +107,24 @@ public static Set<Long> toPositionSet(CloseableIterable<Long> posDeletes) {
     }
   }
 
+  public static <T extends StructLike> PositionDeleteIndex toPositionBitmap(CharSequence dataLocation,
+                                                                            List<CloseableIterable<T>> deleteFiles) {
+    DataFileFilter<T> locationFilter = new DataFileFilter<>(dataLocation);
+    List<CloseableIterable<Long>> positions = Lists.transform(deleteFiles, deletes ->
+        CloseableIterable.transform(locationFilter.filter(deletes), row -> (Long) POSITION_ACCESSOR.get(row)));
+    return toPositionBitmap(CloseableIterable.concat(positions));
+  }
+
+  public static PositionDeleteIndex toPositionBitmap(CloseableIterable<Long> posDeletes) {
+    try (CloseableIterable<Long> deletes = posDeletes) {
+      PositionDeleteIndex positionDeleteIndex = new BitmapPositionDeleteIndex();
+      deletes.forEach(positionDeleteIndex::delete);
+      return positionDeleteIndex;
+    } catch (IOException e) {
+      throw new UncheckedIOException("Failed to close position delete source", e);
+    }
+  }
+
   public static <T> CloseableIterable<T> streamingFilter(CloseableIterable<T> rows,
                                                          Function<T, Long> rowToPosition,
                                                          CloseableIterable<Long> posDeletes) {

core/src/main/java/org/apache/iceberg/deletes/PositionDeleteIndex.java

@@ -0,0 +1,42 @@
+/*
+ * 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.iceberg.deletes;
+
+public interface PositionDeleteIndex {
+  /**
+   * Set a deleted row position.
+   * @param position the deleted row position
+   */
+  void delete(long position);
+
+  /**
+   * Set a range of deleted row positions.
+   * @param posStart inclusive beginning of position range
+   * @param posEnd exclusive ending of position range
+   */
+  void delete(long posStart, long posEnd);
+
+  /**
+   * Checks whether a row at the position is deleted.
+   * @param position deleted row position
+   * @return whether the position is deleted
+   */
+  boolean deleted(long position);
+}

core/src/main/java/org/apache/iceberg/util/TableScanUtil.java

@@ -23,6 +23,7 @@
 import org.apache.iceberg.BaseCombinedScanTask;
 import org.apache.iceberg.CombinedScanTask;
 import org.apache.iceberg.ContentFile;
+import org.apache.iceberg.FileContent;
 import org.apache.iceberg.FileScanTask;
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
@@ -37,6 +38,15 @@ public static boolean hasDeletes(CombinedScanTask task) {
     return task.files().stream().anyMatch(TableScanUtil::hasDeletes);
   }
 
+  /**
+   * This is temporarily introduced since we plan to support pos-delete vectorized read first, then get to the
+   * equality-delete support. We will remove this method once both are supported.
+   */
+  public static boolean hasEqDeletes(CombinedScanTask task) {
+    return task.files().stream().anyMatch(
+        t -> t.deletes().stream().anyMatch(deleteFile -> deleteFile.content().equals(FileContent.EQUALITY_DELETES)));
+  }
+
   public static boolean hasDeletes(FileScanTask task) {
     return !task.deletes().isEmpty();
   }

data/src/main/java/org/apache/iceberg/data/DeleteFilter.java

@@ -37,6 +37,7 @@
 import org.apache.iceberg.data.orc.GenericOrcReader;
 import org.apache.iceberg.data.parquet.GenericParquetReaders;
 import org.apache.iceberg.deletes.Deletes;
+import org.apache.iceberg.deletes.PositionDeleteIndex;
 import org.apache.iceberg.expressions.Expressions;
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.io.InputFile;
@@ -69,6 +70,8 @@ public abstract class DeleteFilter<T> {
   private final Schema requiredSchema;
   private final Accessor<StructLike> posAccessor;
 
+  private PositionDeleteIndex deleteRowPositions = null;
+
   protected DeleteFilter(FileScanTask task, Schema tableSchema, Schema requestedSchema) {
     this.setFilterThreshold = DEFAULT_SET_FILTER_THRESHOLD;
     this.dataFile = task.file();
@@ -98,6 +101,10 @@ public Schema requiredSchema() {
     return requiredSchema;
   }
 
+  public boolean hasPosDeletes() {
+    return !posDeletes.isEmpty();
+  }
+
   Accessor<StructLike> posAccessor() {
     return posAccessor;
   }
@@ -185,6 +192,18 @@ protected boolean shouldKeep(T item) {
     return remainingRowsFilter.filter(records);
   }
 
+  public PositionDeleteIndex deletedRowPositions() {
+    if (posDeletes.isEmpty()) {
+      return null;
+    }
+
+    if (deleteRowPositions == null) {
+      List<CloseableIterable<Record>> deletes = Lists.transform(posDeletes, this::openPosDeletes);
+      deleteRowPositions = Deletes.toPositionBitmap(dataFile.path(), deletes);
+    }
+    return deleteRowPositions;
+  }
+
   private CloseableIterable<T> applyPosDeletes(CloseableIterable<T> records) {
     if (posDeletes.isEmpty()) {
       return records;

data/src/test/java/org/apache/iceberg/data/DeleteReadTests.java

@@ -75,9 +75,9 @@ public abstract class DeleteReadTests {
   protected String dateTableName = null;
   protected Table table = null;
   protected Table dateTable = null;
-  private List<Record> records = null;
+  protected List<Record> records = null;
   private List<Record> dateRecords = null;
-  private DataFile dataFile = null;
+  protected DataFile dataFile = null;
 
   @Before
   public void writeTestDataFile() throws IOException {
@@ -298,6 +298,39 @@ public void testPositionDeletes() throws IOException {
     Assert.assertEquals("Table should contain expected rows", expected, actual);
   }
 
+  @Test
+  public void testMultiplePosDeleteFiles() throws IOException {
+    List<Pair<CharSequence, Long>> deletes = Lists.newArrayList(
+        Pair.of(dataFile.path(), 0L), // id = 29
+        Pair.of(dataFile.path(), 3L) // id = 89
+    );
+
+    Pair<DeleteFile, CharSequenceSet> posDeletes = FileHelpers.writeDeleteFile(
+        table, Files.localOutput(temp.newFile()), Row.of(0), deletes);
+
+    table.newRowDelta()
+        .addDeletes(posDeletes.first())
+        .validateDataFilesExist(posDeletes.second())
+        .commit();
+
+    deletes = Lists.newArrayList(
+        Pair.of(dataFile.path(), 6L) // id = 122
+    );
+
+    posDeletes = FileHelpers.writeDeleteFile(
+        table, Files.localOutput(temp.newFile()), Row.of(0), deletes);
+
+    table.newRowDelta()
+        .addDeletes(posDeletes.first())
+        .validateDataFilesExist(posDeletes.second())
+        .commit();
+
+    StructLikeSet expected = rowSetWithoutIds(table, records, 29, 89, 122);
+    StructLikeSet actual = rowSet(tableName, table, "*");
+
+    Assert.assertEquals("Table should contain expected rows", expected, actual);
+  }
+
   @Test
   public void testMixedPositionAndEqualityDeletes() throws IOException {
     Schema dataSchema = table.schema().select("data");
@@ -411,7 +444,7 @@ private StructLikeSet selectColumns(StructLikeSet rows, String... columns) {
     return set;
   }
 
-  private static StructLikeSet rowSetWithoutIds(Table table, List<Record> recordList, int... idsToRemove) {
+  protected static StructLikeSet rowSetWithoutIds(Table table, List<Record> recordList, int... idsToRemove) {
     Set<Integer> deletedIds = Sets.newHashSet(ArrayUtil.toIntList(idsToRemove));
     StructLikeSet set = StructLikeSet.create(table.schema().asStruct());
     recordList.stream()

spark/v2.4/build.gradle

@@ -65,6 +65,7 @@ project(':iceberg-spark:iceberg-spark2') {
     compileOnly "org.apache.avro:avro"
     compileOnly("org.apache.spark:spark-hive_2.11") {
       exclude group: 'org.apache.avro', module: 'avro'
+      exclude group: 'org.roaringbitmap'
     }
 
     implementation("org.apache.orc:orc-core::nohive") {
@@ -159,6 +160,7 @@ project(':iceberg-spark:iceberg-spark-runtime') {
     relocate 'org.apache.arrow', 'org.apache.iceberg.shaded.org.apache.arrow'
     relocate 'com.carrotsearch', 'org.apache.iceberg.shaded.com.carrotsearch'
     relocate 'org.threeten.extra', 'org.apache.iceberg.shaded.org.threeten.extra'
+    relocate 'org.roaringbitmap', 'org.apache.iceberg.shaded.org.roaringbitmap'
 
     classifier null
   }

spark/v3.0/build.gradle

@@ -64,6 +64,7 @@ project(':iceberg-spark:iceberg-spark3') {
     compileOnly("org.apache.spark:spark-hive_2.12:${sparkVersion}") {
       exclude group: 'org.apache.avro', module: 'avro'
       exclude group: 'org.apache.arrow'
+      exclude group: 'org.roaringbitmap'
     }
 
     implementation("org.apache.orc:orc-core::nohive") {
@@ -134,6 +135,7 @@ project(":iceberg-spark:iceberg-spark3-extensions") {
     compileOnly("org.apache.spark:spark-hive_2.12:${sparkVersion}") {
       exclude group: 'org.apache.avro', module: 'avro'
       exclude group: 'org.apache.arrow'
+      exclude group: 'org.roaringbitmap'
     }
 
     testImplementation project(path: ':iceberg-hive-metastore', configuration: 'testArtifacts')
@@ -242,6 +244,7 @@ project(':iceberg-spark:iceberg-spark3-runtime') {
     relocate 'org.threeten.extra', 'org.apache.iceberg.shaded.org.threeten.extra'
     // relocate Antlr runtime and related deps to shade Iceberg specific version
     relocate 'org.antlr.v4', 'org.apache.iceberg.shaded.org.antlr.v4'
+    relocate 'org.roaringbitmap', 'org.apache.iceberg.shaded.org.roaringbitmap'
 
     classifier null
   }

spark/v3.1/build.gradle

@@ -64,6 +64,7 @@ project(':iceberg-spark:iceberg-spark-3.1') {
     compileOnly("org.apache.spark:spark-hive_2.12:${sparkVersion}") {
       exclude group: 'org.apache.avro', module: 'avro'
       exclude group: 'org.apache.arrow'
+      exclude group: 'org.roaringbitmap'
     }
 
     implementation("org.apache.orc:orc-core::nohive") {
@@ -134,6 +135,7 @@ project(":iceberg-spark:iceberg-spark-3.1-extensions") {
     compileOnly("org.apache.spark:spark-hive_2.12:${sparkVersion}") {
       exclude group: 'org.apache.avro', module: 'avro'
       exclude group: 'org.apache.arrow'
+      exclude group: 'org.roaringbitmap'
     }
 
     testImplementation project(path: ':iceberg-hive-metastore', configuration: 'testArtifacts')
@@ -242,6 +244,7 @@ project(':iceberg-spark:iceberg-spark-3.1-runtime') {
     relocate 'org.threeten.extra', 'org.apache.iceberg.shaded.org.threeten.extra'
     // relocate Antlr runtime and related deps to shade Iceberg specific version
     relocate 'org.antlr.v4', 'org.apache.iceberg.shaded.org.antlr.v4'
+    relocate 'org.roaringbitmap', 'org.apache.iceberg.shaded.org.roaringbitmap'
 
     classifier null
   }

spark/v3.2/build.gradle

@@ -66,6 +66,7 @@ project(':iceberg-spark:iceberg-spark-3.2') {
       exclude group: 'org.apache.arrow'
       // to make sure io.netty.buffer only comes from project(':iceberg-arrow')
       exclude group: 'io.netty', module: 'netty-buffer'
+      exclude group: 'org.roaringbitmap'
     }
 
     implementation("org.apache.orc:orc-core::nohive") {
@@ -138,6 +139,7 @@ project(":iceberg-spark:iceberg-spark-3.2-extensions") {
       exclude group: 'org.apache.arrow'
       // to make sure io.netty.buffer only comes from project(':iceberg-arrow')
       exclude group: 'io.netty', module: 'netty-buffer'
+      exclude group: 'org.roaringbitmap'
     }
 
     testImplementation project(path: ':iceberg-hive-metastore', configuration: 'testArtifacts')
@@ -244,6 +246,7 @@ project(':iceberg-spark:iceberg-spark-3.2-runtime') {
     relocate 'org.threeten.extra', 'org.apache.iceberg.shaded.org.threeten.extra'
     // relocate Antlr runtime and related deps to shade Iceberg specific version
     relocate 'org.antlr.v4', 'org.apache.iceberg.shaded.org.antlr.v4'
+    relocate 'org.roaringbitmap', 'org.apache.iceberg.shaded.org.roaringbitmap'
 
     classifier null
   }

spark/v3.2/spark/src/jmh/java/org/apache/iceberg/spark/source/IcebergSourceBenchmark.java

@@ -26,6 +26,7 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.iceberg.FileFormat;
 import org.apache.iceberg.Table;
 import org.apache.iceberg.TableProperties;
 import org.apache.iceberg.UpdateProperties;
@@ -187,4 +188,8 @@ protected void withTableProperties(Map<String, String> props, Action action) {
       restoreProperties.commit();
     }
   }
+
+  protected FileFormat fileFormat() {
+    throw new UnsupportedOperationException("Unsupported file format");
+  }
 }