HIVE-30102: Iceberg: LLAP cannot read an ORC file that has positional deletes

iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/hive/vector/HiveVectorizedReader.java

@@ -61,7 +61,6 @@
 import org.apache.iceberg.io.CloseableIterable;
 import org.apache.iceberg.io.CloseableIterator;
 import org.apache.iceberg.io.FileIO;
-import org.apache.iceberg.mr.InputFormatConfig;
 import org.apache.iceberg.mr.hive.HiveIcebergInputFormat;
 import org.apache.iceberg.mr.mapred.MapredIcebergInputFormat;
 import org.apache.iceberg.orc.VectorizedReadUtils;
@@ -220,6 +219,8 @@ private static RecordReader<NullWritable, VectorizedRowBatch> orcRecordReader(Jo
     // Need to turn positional schema evolution off since we use column name based schema evolution for projection
     // and Iceberg will make a mapping between the file schema and the current reading schema.
     job.setBoolean(OrcConf.FORCE_POSITIONAL_EVOLUTION.getHiveConfName(), false);
+    // Row positions are derived from the batch index, and ORC's row-level filter can drop rows and whole batches.
+    job.setBoolean(OrcConf.ALLOW_SARG_TO_FILTER.getHiveConfName(), false);
 
     // Metadata information has to be passed along in the OrcSplit. Without specifying this, the vectorized
     // reader will assume that the ORC file ends at the task's start + length, and might fail reading the tail..
@@ -230,9 +231,8 @@ private static RecordReader<NullWritable, VectorizedRowBatch> orcRecordReader(Jo
         VectorizedReadUtils.deserializeToShadedOrcTail(serializedOrcTail).getSchema(), residual);
 
     // If LLAP enabled, try to retrieve an LLAP record reader - this might yield to null in some special cases
-    // TODO: add support for reading files with positional deletes with LLAP (LLAP would need to provide file row num)
     if (HiveConf.getBoolVar(job, HiveConf.ConfVars.LLAP_IO_ENABLED, LlapProxy.isDaemon()) &&
-        LlapProxy.getIo() != null && task.deletes().isEmpty() && !InputFormatConfig.fetchVirtualColumns(job)) {
+        LlapProxy.getIo() != null) {
       recordReader = LlapProxy.getIo().orc().vectorizedReaderForPath(fileId, path, null, readColumnIds,
           job, start, length, reporter);
     }

iceberg/iceberg-handler/src/main/java/org/apache/iceberg/mr/mapreduce/IcebergRecordReader.java

@@ -34,6 +34,7 @@
 import org.apache.hadoop.hive.ql.io.IOContextAwareRecordReader;
 import org.apache.hadoop.mapreduce.InputSplit;
 import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.hive.iceberg.org.apache.orc.OrcConf;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.DataTask;
 import org.apache.iceberg.FileFormat;
@@ -262,6 +263,8 @@ private CloseableIterable<T> newOrcIterable(
     ORC.ReadBuilder orcReadBuilder = ORC.read(inputFile)
         .project(readSchemaWithoutConstantAndMetadataFields)
         .filter(residual)
+        // Row positions are derived from the batch index, and ORC's row-level filter can drop rows and whole batches.
+        .config(OrcConf.ALLOW_SARG_TO_FILTER.getHiveConfName(), "false")
         .caseSensitive(isCaseSensitive())
         .split(task.start(), task.length());
 

iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/TestHiveIcebergCRUD.java

@@ -215,6 +215,32 @@ public void testReadAndWriteFormatV2Partitioned_PosDelete() throws IOException {
     Assert.assertArrayEquals(new Object[] {2L, "Trudy", "Pink"}, objects.get(3));
   }
 
+  /**
+   * ORC can filter rows by the pushed down predicate; in row mode, row positions and deletes must still apply to
+   * the right rows.
+   */
+  @Test
+  public void testPositionDeleteWithOrcRowFiltering() {
+    Assume.assumeTrue(fileFormat == FileFormat.ORC && formatVersion >= 2 && !isVectorized);
+
+    Schema schema = new Schema(optional(1, "customer_id", Types.LongType.get()));
+    List<Record> records = TestHelper.generateRandomRecords(schema, 30000, 0L);
+    for (int i = 0; i < records.size(); ++i) {
+      records.get(i).setField("customer_id", (long) i);
+    }
+    testTables.createTable(shell, "customers", schema, PartitionSpec.unpartitioned(), fileFormat, records,
+        formatVersion);
+    shell.executeStatement("DELETE FROM customers WHERE customer_id = 25000");
+
+    shell.setHiveSessionValue("orc.sarg.to.filter", true);
+    shell.setHiveSessionValue("orc.filter.use.selected", true);
+    List<Object[]> result = shell.executeStatement(
+        "SELECT customer_id, ROW__POSITION FROM customers WHERE customer_id IN (25000, 25001)");
+
+    Assert.assertEquals(1, result.size());
+    Assert.assertArrayEquals(new Object[] {25001L, 25001L}, result.get(0));
+  }
+
   /**
    * Tests reading/consuming legacy Iceberg V2 tables containing position delete files with embedded
    * row data. Iceberg 1.12.0 removed writer support for row data in position deletes, but Hive

iceberg/iceberg-handler/src/test/java/org/apache/iceberg/mr/hive/vector/TestHiveIcebergVectorization.java

@@ -219,6 +219,31 @@ private void testVectorizedReadWithDeleteFilter(Map<String, String> props) {
     validation.apply(1501);
   }
 
+  /**
+   * ORC can filter the rows of a batch by the pushed down predicate; row positions and deletes must still apply to
+   * the right rows.
+   */
+  @Test
+  public void testHiveDeleteFilterWithOrcRowFiltering() {
+    Assume.assumeTrue(fileFormat == FileFormat.ORC);
+
+    Schema schema = new Schema(optional(1, "customer_id", Types.LongType.get()));
+    List<Record> records = TestHelper.generateRandomRecords(schema, 30000, 0L);
+    for (int i = 0; i < records.size(); ++i) {
+      records.get(i).setField("customer_id", (long) i);
+    }
+    testTables.createTable(shell, "vectordelete", schema, PartitionSpec.unpartitioned(), fileFormat, records, 2);
+    shell.executeStatement("DELETE FROM vectordelete WHERE customer_id = 25000");
+
+    shell.setHiveSessionValue("orc.sarg.to.filter", true);
+    shell.setHiveSessionValue("orc.filter.use.selected", true);
+    List<Object[]> result = shell.executeStatement(
+        "select customer_id, ROW__POSITION from vectordelete where customer_id in (25000, 25001)");
+
+    Assert.assertEquals(1, result.size());
+    Assert.assertArrayEquals(new Object[] { 25001L, 25001L }, result.get(0));
+  }
+
   @Test
   public void testHiveDeleteFilterWithFilteredParquetBlock() {
     Assume.assumeTrue(fileFormat == FileFormat.PARQUET);

iceberg/iceberg-handler/src/test/queries/positive/llap_iceberg_read_orc_deletes.q

@@ -0,0 +1,26 @@
+-- ORC files with position deletes are read through LLAP IO (see the LLAP IO counters): the deletes and ROW__POSITION
+-- apply to the right rows in a later row group and across a batch boundary (the third row group starts at file row
+-- 20000, its sixth batch at 25120)
+set hive.llap.io.enabled=true;
+set hive.vectorized.execution.enabled=true;
+set hive.fetch.task.conversion=none;
+set hive.optimize.index.filter=true;
+
+-- V2: position delete files
+CREATE EXTERNAL TABLE ice_orc_v2 (id INT) STORED BY ICEBERG STORED AS ORC
+TBLPROPERTIES ('format-version'='2');
+INSERT INTO ice_orc_v2 SELECT pos FROM (SELECT posexplode(split(space(29999), ' ')) AS (pos, val)) t;
+DELETE FROM ice_orc_v2 WHERE id IN (25119, 25121);
+
+-- V3: deletion vectors
+CREATE EXTERNAL TABLE ice_orc_v3 (id INT) STORED BY ICEBERG STORED AS ORC
+TBLPROPERTIES ('format-version'='3');
+INSERT INTO ice_orc_v3 SELECT pos FROM (SELECT posexplode(split(space(29999), ' ')) AS (pos, val)) t;
+DELETE FROM ice_orc_v3 WHERE id IN (25119, 25121);
+
+SET hive.exec.post.hooks=org.apache.hadoop.hive.ql.hooks.PostExecTezSummaryPrinter;
+SELECT id, ROW__POSITION FROM ice_orc_v2 WHERE id BETWEEN 25117 AND 25123 ORDER BY id;
+SELECT id, ROW__POSITION FROM ice_orc_v3 WHERE id BETWEEN 25117 AND 25123 ORDER BY id;
+
+DROP TABLE ice_orc_v2;
+DROP TABLE ice_orc_v3;

iceberg/iceberg-handler/src/test/results/positive/llap/llap_iceberg_read_orc_deletes.q.out

@@ -0,0 +1,138 @@
+PREHOOK: query: CREATE EXTERNAL TABLE ice_orc_v2 (id INT) STORED BY ICEBERG STORED AS ORC
+TBLPROPERTIES ('format-version'='2')
+PREHOOK: type: CREATETABLE
+PREHOOK: Output: database:default
+PREHOOK: Output: default@ice_orc_v2
+POSTHOOK: query: CREATE EXTERNAL TABLE ice_orc_v2 (id INT) STORED BY ICEBERG STORED AS ORC
+TBLPROPERTIES ('format-version'='2')
+POSTHOOK: type: CREATETABLE
+POSTHOOK: Output: database:default
+POSTHOOK: Output: default@ice_orc_v2
+PREHOOK: query: INSERT INTO ice_orc_v2 SELECT pos FROM (SELECT posexplode(split(space(29999), ' ')) AS (pos, val)) t
+PREHOOK: type: QUERY
+PREHOOK: Input: _dummy_database@_dummy_table
+PREHOOK: Output: default@ice_orc_v2
+POSTHOOK: query: INSERT INTO ice_orc_v2 SELECT pos FROM (SELECT posexplode(split(space(29999), ' ')) AS (pos, val)) t
+POSTHOOK: type: QUERY
+POSTHOOK: Input: _dummy_database@_dummy_table
+POSTHOOK: Output: default@ice_orc_v2
+PREHOOK: query: DELETE FROM ice_orc_v2 WHERE id IN (25119, 25121)
+PREHOOK: type: QUERY
+PREHOOK: Input: default@ice_orc_v2
+PREHOOK: Output: default@ice_orc_v2
+POSTHOOK: query: DELETE FROM ice_orc_v2 WHERE id IN (25119, 25121)
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@ice_orc_v2
+POSTHOOK: Output: default@ice_orc_v2
+PREHOOK: query: CREATE EXTERNAL TABLE ice_orc_v3 (id INT) STORED BY ICEBERG STORED AS ORC
+TBLPROPERTIES ('format-version'='3')
+PREHOOK: type: CREATETABLE
+PREHOOK: Output: database:default
+PREHOOK: Output: default@ice_orc_v3
+POSTHOOK: query: CREATE EXTERNAL TABLE ice_orc_v3 (id INT) STORED BY ICEBERG STORED AS ORC
+TBLPROPERTIES ('format-version'='3')
+POSTHOOK: type: CREATETABLE
+POSTHOOK: Output: database:default
+POSTHOOK: Output: default@ice_orc_v3
+PREHOOK: query: INSERT INTO ice_orc_v3 SELECT pos FROM (SELECT posexplode(split(space(29999), ' ')) AS (pos, val)) t
+PREHOOK: type: QUERY
+PREHOOK: Input: _dummy_database@_dummy_table
+PREHOOK: Output: default@ice_orc_v3
+POSTHOOK: query: INSERT INTO ice_orc_v3 SELECT pos FROM (SELECT posexplode(split(space(29999), ' ')) AS (pos, val)) t
+POSTHOOK: type: QUERY
+POSTHOOK: Input: _dummy_database@_dummy_table
+POSTHOOK: Output: default@ice_orc_v3
+PREHOOK: query: DELETE FROM ice_orc_v3 WHERE id IN (25119, 25121)
+PREHOOK: type: QUERY
+PREHOOK: Input: default@ice_orc_v3
+PREHOOK: Output: default@ice_orc_v3
+POSTHOOK: query: DELETE FROM ice_orc_v3 WHERE id IN (25119, 25121)
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@ice_orc_v3
+POSTHOOK: Output: default@ice_orc_v3
+PREHOOK: query: SELECT id, ROW__POSITION FROM ice_orc_v2 WHERE id BETWEEN 25117 AND 25123 ORDER BY id
+PREHOOK: type: QUERY
+PREHOOK: Input: default@ice_orc_v2
+#### A masked pattern was here ####
+Stage-1 FILE SYSTEM COUNTERS:
+Stage-1 HIVE COUNTERS:
+   CREATED_FILES: 1
+   DESERIALIZE_ERRORS: 0
+   RECORDS_IN_Map_1: 9998
+   RECORDS_OUT_0: 5
+   RECORDS_OUT_INTERMEDIATE_Map_1: 5
+   RECORDS_OUT_INTERMEDIATE_Reducer_2: 0
+   RECORDS_OUT_OPERATOR_FIL_7: 5
+   RECORDS_OUT_OPERATOR_FS_11: 5
+   RECORDS_OUT_OPERATOR_MAP_0: 0
+   RECORDS_OUT_OPERATOR_RS_9: 5
+   RECORDS_OUT_OPERATOR_SEL_10: 5
+   RECORDS_OUT_OPERATOR_SEL_8: 5
+   RECORDS_OUT_OPERATOR_TS_0: 9998
+   TOTAL_TABLE_ROWS_WRITTEN: 0
+Stage-1 LLAP IO COUNTERS:
+   CACHE_HIT_BYTES: 229
+   CACHE_MISS_BYTES: 0
+   METADATA_CACHE_HIT: 2
+   NUM_DECODED_BATCHES: 1
+   NUM_VECTOR_BATCHES: 10
+   ROWS_EMITTED: 10000
+   SELECTED_ROWGROUPS: 1
+Stage-1 INPUT COUNTERS:
+   GROUPED_INPUT_SPLITS_Map_1: 1
+   INPUT_DIRECTORIES_Map_1: 1
+   INPUT_FILES_Map_1: 1
+   RAW_INPUT_SPLITS_Map_1: 1
+25117	25117
+25118	25118
+25120	25120
+25122	25122
+25123	25123
+PREHOOK: query: SELECT id, ROW__POSITION FROM ice_orc_v3 WHERE id BETWEEN 25117 AND 25123 ORDER BY id
+PREHOOK: type: QUERY
+PREHOOK: Input: default@ice_orc_v3
+#### A masked pattern was here ####
+Stage-1 FILE SYSTEM COUNTERS:
+Stage-1 HIVE COUNTERS:
+   CREATED_FILES: 1
+   DESERIALIZE_ERRORS: 0
+   RECORDS_IN_Map_1: 9998
+   RECORDS_OUT_0: 5
+   RECORDS_OUT_INTERMEDIATE_Map_1: 5
+   RECORDS_OUT_INTERMEDIATE_Reducer_2: 0
+   RECORDS_OUT_OPERATOR_FIL_7: 5
+   RECORDS_OUT_OPERATOR_FS_11: 5
+   RECORDS_OUT_OPERATOR_MAP_0: 0
+   RECORDS_OUT_OPERATOR_RS_9: 5
+   RECORDS_OUT_OPERATOR_SEL_10: 5
+   RECORDS_OUT_OPERATOR_SEL_8: 5
+   RECORDS_OUT_OPERATOR_TS_0: 9998
+   TOTAL_TABLE_ROWS_WRITTEN: 0
+Stage-1 LLAP IO COUNTERS:
+   CACHE_HIT_BYTES: 229
+   CACHE_MISS_BYTES: 0
+   METADATA_CACHE_HIT: 2
+   NUM_DECODED_BATCHES: 1
+   NUM_VECTOR_BATCHES: 10
+   ROWS_EMITTED: 10000
+   SELECTED_ROWGROUPS: 1
+Stage-1 INPUT COUNTERS:
+   GROUPED_INPUT_SPLITS_Map_1: 1
+   INPUT_DIRECTORIES_Map_1: 1
+   INPUT_FILES_Map_1: 1
+   RAW_INPUT_SPLITS_Map_1: 1
+25117	25117
+25118	25118
+25120	25120
+25122	25122
+25123	25123
+PREHOOK: query: DROP TABLE ice_orc_v2
+PREHOOK: type: DROPTABLE
+PREHOOK: Input: default@ice_orc_v2
+PREHOOK: Output: database:default
+PREHOOK: Output: default@ice_orc_v2
+PREHOOK: query: DROP TABLE ice_orc_v3
+PREHOOK: type: DROPTABLE
+PREHOOK: Input: default@ice_orc_v3
+PREHOOK: Output: database:default
+PREHOOK: Output: default@ice_orc_v3

itests/src/test/resources/testconfiguration.properties

@@ -423,6 +423,7 @@ iceberg.llap.query.files=\
   iceberg_merge_files.q,\
   iceberg_row_mode_virtual_columns.q,\
   llap_iceberg_read_orc.q,\
+  llap_iceberg_read_orc_deletes.q,\
   llap_iceberg_read_parquet.q,\
   puffin_col_stats_with_time_travel.q,\
   vectorized_iceberg_read_mixed.q,\
@@ -483,6 +484,7 @@ iceberg.llap.only.query.files=\
   iceberg_merge_files.q,\
   iceberg_row_mode_virtual_columns.q,\
   llap_iceberg_read_orc.q,\
+  llap_iceberg_read_orc_deletes.q,\
   llap_iceberg_read_parquet.q,\
   puffin_col_stats_with_time_travel.q
 

llap-server/src/java/org/apache/hadoop/hive/llap/io/api/impl/ColumnVectorBatch.java

@@ -34,6 +34,8 @@ public class ColumnVectorBatch {
   public MutableFilterContext filterContext;
   public ColumnVector[] cols;
   public int size;
+  /** File row position of the first row; Long.MIN_VALUE when the producer does not track positions. */
+  private long startRowInFile = Long.MIN_VALUE;
 
   public ColumnVectorBatch(int columnCount) {
     this(columnCount, VectorizedRowBatch.DEFAULT_SIZE);
@@ -51,6 +53,13 @@ public void swapColumnVector(int ix, ColumnVector[] other, int otherIx) {
     cols[ix] = old;
   }
 
+  public long getStartRowInFile() {
+    return startRowInFile;
+  }
+
+  public void setStartRowInFile(long startRowInFile) {
+    this.startRowInFile = startRowInFile;
+  }
 
   @Override
   public String toString() {
@@ -76,7 +85,6 @@ public String toString() {
     }
     b.append('\n');
 
-
     for (int i = 0; i < size; i++) {
       b.append('[');
       for (int k = 0; k < cols.length; k++) {
@@ -85,12 +93,10 @@ public String toString() {
           b.append(", ");
         }
         if (cv == null) continue;
-        if (cv != null) {
-          try {
-            cv.stringifyValue(b, i);
-          } catch (Exception ex) {
-            b.append("invalid");
-          }
+        try {
+          cv.stringifyValue(b, i);
+        } catch (Exception ex) {
+          b.append("invalid");
         }
       }
       b.append(']');
@@ -101,4 +107,4 @@ public String toString() {
 
     return b.toString();
   }
-}
+}