HIVE-29857: Iceberg: Fix ROW__POSITION on vectorized VARIANT reads

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

@@ -266,9 +266,10 @@ private static RecordReader<NullWritable, VectorizedRowBatch> parquetRecordReade
 
     ParquetMetadata parquetMetadata = HiveParquetUtil.readFooter(task.file(), io, job, footerData);
     MessageType fileSchema = parquetMetadata.getFileMetaData().getSchema();
-    ParquetMetadata prunedMetadata =
-        VariantParquetFilters.pruneVariantRowGroups(parquetMetadata, fileSchema, residual);
-    inputFormat.setMetadata(prunedMetadata);
+    // The reader keeps the file's own footer, so row positions and scan statistics still describe the
+    // file. Row groups variant pruning ruled out are named alongside it; null means read them all.
+    inputFormat.setMetadata(parquetMetadata,
+        VariantParquetFilters.pickRowGroups(fileSchema, residual, parquetMetadata.getBlocks()));
 
     MessageType typeWithIds = null;
     Schema expectedSchema = task.spec().schema();
@@ -287,7 +288,7 @@ private static RecordReader<NullWritable, VectorizedRowBatch> parquetRecordReade
     inputFormat.seInitialColumnDefaults(initialColumnDefaults);
     RecordReader<NullWritable, VectorizedRowBatch> reader = inputFormat.getRecordReader(split, job, reporter);
     return ParquetVariantRecordReader
-        .tryWrap(reader, job, task, path, start, length, prunedMetadata)
+        .tryWrap(reader, job, task, path, start, length, parquetMetadata)
         .orElse(reader);
   }
 

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

@@ -23,6 +23,7 @@
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
 import java.util.function.ToIntFunction;
@@ -31,6 +32,7 @@
 import org.apache.hadoop.hive.ql.exec.vector.ColumnVector;
 import org.apache.hadoop.hive.ql.exec.vector.StructColumnVector;
 import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
+import org.apache.hadoop.hive.ql.io.RowPositionAwareVectorizedRecordReader;
 import org.apache.hadoop.hive.ql.io.parquet.ParquetRecordReaderBase;
 import org.apache.hadoop.io.NullWritable;
 import org.apache.hadoop.mapred.JobConf;
@@ -52,7 +54,8 @@
 import org.apache.parquet.schema.MessageType;
 import org.apache.parquet.schema.Type;
 
-final class ParquetVariantRecordReader implements RecordReader<NullWritable, VectorizedRowBatch> {
+final class ParquetVariantRecordReader
+    implements RecordReader<NullWritable, VectorizedRowBatch>, RowPositionAwareVectorizedRecordReader {
 
   private static final String INVALID_VARIANT_STRUCT = "Invalid Variant struct for column ";
 
@@ -137,11 +140,11 @@ private static List<BlockMetaData> blocksForSplit(
     // If the underlying Hive Parquet reader already computed row-group filtering (e.g. from SARG),
     // we must use the exact same blocks to keep this reader aligned with the delegate.
     if (delegate instanceof ParquetRecordReaderBase parquetDelegate) {
+      // The delegate's answer is the whole answer, including when it holds none: an empty list means it
+      // filtered every row group out, and null means it found none to read at all. Falling back to the
+      // split's own row groups would read one the delegate has already ruled out.
       List<BlockMetaData> filteredBlocks = parquetDelegate.getFilteredBlocks();
-      // Treat an empty list as authoritative (delegate filtered out all row groups).
-      if (filteredBlocks != null) {
-        return filteredBlocks;
-      }
+      return filteredBlocks != null ? filteredBlocks : Collections.emptyList();
     }
     // Fallback: compute blocks from split boundaries
     List<BlockMetaData> splitBlocks = Lists.newArrayList();
@@ -154,6 +157,20 @@ private static List<BlockMetaData> blocksForSplit(
     return splitBlocks;
   }
 
+  /**
+   * Row positions come from the reader this wraps. Without this the wrapper hides the delegate's
+   * {@link RowPositionAwareVectorizedRecordReader}, and every row of a VARIANT table is handed the unknown
+   * position marker instead - which ROW__POSITION, row lineage and positional deletes all rely on.
+   */
+  @Override
+  public long getRowNumber() throws IOException {
+    if (delegate instanceof RowPositionAwareVectorizedRecordReader positionAware) {
+      return positionAware.getRowNumber();
+    }
+    throw new UnsupportedOperationException(
+        "The reader under " + delegate.getClass().getName() + " cannot report row positions");
+  }
+
   @Override
   public boolean next(NullWritable key, VectorizedRowBatch value) throws IOException {
     boolean hasNext = delegate.next(key, value);

iceberg/iceberg-handler/src/main/java/org/apache/iceberg/parquet/ReadConf.java

@@ -101,14 +101,14 @@ class ReadConf<T> {
       bloomFilter = new ParquetBloomRowGroupFilter(expectedSchema, filter, caseSensitive);
     }
 
-    boolean[] variantRowGroupMayMatch =
-        VariantParquetFilters.variantRowGroupMayMatch(fileSchema, filter, rowGroups);
+    boolean[] mayMatch =
+        VariantParquetFilters.pickRowGroups(fileSchema, filter, rowGroups);
 
     long computedTotalValues = 0L;
     for (int i = 0; i < shouldSkip.length; i += 1) {
       BlockMetaData rowGroup = rowGroups.get(i);
       boolean shouldRead =
-          (variantRowGroupMayMatch == null || variantRowGroupMayMatch[i]) &&
+          (mayMatch == null || mayMatch[i]) &&
           (filter == null ||
               statsFilter.shouldRead(typeWithIds, rowGroup) &&
               dictFilter.shouldRead(typeWithIds, rowGroup, reader.getDictionaryReader(rowGroup)) &&

iceberg/iceberg-handler/src/main/java/org/apache/iceberg/parquet/VariantParquetFilters.java

@@ -41,7 +41,6 @@
 import org.apache.parquet.hadoop.metadata.BlockMetaData;
 import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData;
 import org.apache.parquet.hadoop.metadata.ColumnPath;
-import org.apache.parquet.hadoop.metadata.ParquetMetadata;
 import org.apache.parquet.io.api.Binary;
 import org.apache.parquet.schema.GroupType;
 import org.apache.parquet.schema.LogicalTypeAnnotation.StringLogicalTypeAnnotation;
@@ -95,7 +94,11 @@ private static ResolvedVariantFilter resolveVariantFilter(MessageType schema, Ex
     return new ResolvedVariantFilter(predicate, visitor.fallbackValueColumns());
   }
 
-  public static boolean[] variantRowGroupMayMatch(
+  /**
+   * Which of these row groups a VARIANT predicate could match, one flag each, or null when the predicate
+   * says nothing about them and every row group is to be read.
+   */
+  public static boolean[] pickRowGroups(
       MessageType fileSchema, Expression filter, List<BlockMetaData> rowGroups) {
     if (fileSchema == null || filter == null || rowGroups == null || rowGroups.isEmpty()) {
       return null;
@@ -135,54 +138,6 @@ private static boolean mayMatchViaFallback(BlockMetaData rowGroup, Set<ColumnPat
     return false;
   }
 
-  private static List<BlockMetaData> pruneVariantRowGroups(
-      MessageType fileSchema, Expression filter, List<BlockMetaData> rowGroups) {
-    boolean[] mayMatch = variantRowGroupMayMatch(fileSchema, filter, rowGroups);
-    if (mayMatch == null) {
-      return rowGroups;
-    }
-
-    List<BlockMetaData> kept = Lists.newArrayListWithCapacity(rowGroups.size());
-    for (int i = 0; i < rowGroups.size(); i++) {
-      if (mayMatch[i]) {
-        kept.add(rowGroups.get(i));
-      }
-    }
-
-    return kept.size() == rowGroups.size() ? rowGroups : kept;
-  }
-
-  /** Returns Parquet metadata with row groups pruned using best-effort VARIANT pruning. */
-  public static ParquetMetadata pruneVariantRowGroups(
-      ParquetMetadata parquetMetadata, MessageType fileSchema, Expression filter) {
-    if (parquetMetadata == null || filter == null) {
-      return parquetMetadata;
-    }
-
-    List<BlockMetaData> rowGroups = parquetMetadata.getBlocks();
-    if (rowGroups == null || rowGroups.isEmpty()) {
-      return parquetMetadata;
-    }
-
-    MessageType schema = fileSchema;
-    if (schema == null) {
-      if (parquetMetadata.getFileMetaData() == null) {
-        return parquetMetadata;
-      }
-      schema = parquetMetadata.getFileMetaData().getSchema();
-    }
-    if (schema == null) {
-      return parquetMetadata;
-    }
-
-    List<BlockMetaData> kept = pruneVariantRowGroups(schema, filter, rowGroups);
-    if (kept == rowGroups || parquetMetadata.getFileMetaData() == null) {
-      return parquetMetadata;
-    }
-
-    return new ParquetMetadata(parquetMetadata.getFileMetaData(), kept);
-  }
-
   private static boolean isColumnAllNull(ColumnChunkMetaData meta) {
     if (meta == null || meta.getStatistics() == null || !meta.getStatistics().isNumNullsSet()) {
       return false;

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

@@ -339,21 +339,31 @@ private static boolean mapVectorized(List<Object[]> explain) {
   private static void assertVectorizedParquetRowGroupsPruned(Path parquetPath, Expression filter) {
     assertParquetRowGroupsPruned(
         parquetPath, filter,
-        (parquetMetadata, fileSchema, expr) ->
-            // Simulate what HiveVectorizedReader.parquetRecordReader() does
-            VariantParquetFilters
-                .pruneVariantRowGroups(parquetMetadata, fileSchema, expr)
-                .getBlocks()
-                .size());
+        (parquetMetadata, fileSchema, expr) -> {
+          // Simulate what HiveVectorizedReader.parquetRecordReader() does: the footer stays whole and the
+          // row groups to read are named separately
+          boolean[] mayMatch = VariantParquetFilters
+              .pickRowGroups(fileSchema, expr, parquetMetadata.getBlocks());
+          if (mayMatch == null) {
+            return parquetMetadata.getBlocks().size();
+          }
+          int matching = 0;
+          for (boolean match : mayMatch) {
+            if (match) {
+              matching++;
+            }
+          }
+          return matching;
+        });
   }
 
   private static void assertNonVectorizedParquetRowGroupsPruned(Path parquetPath, Expression filter) {
     assertParquetRowGroupsPruned(
         parquetPath, filter,
         (parquetMetadata, fileSchema, expr) -> {
-          // Simulate what ReadConf does - uses variantRowGroupMayMatch to compute shouldSkip array
+          // Simulate what ReadConf does - uses pickRowGroups to compute shouldSkip array
           boolean[] mayMatch = VariantParquetFilters
-              .variantRowGroupMayMatch(fileSchema, expr, parquetMetadata.getBlocks());
+              .pickRowGroups(fileSchema, expr, parquetMetadata.getBlocks());
           int matching = 0;
           for (boolean match : mayMatch) {
             if (match) {

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

@@ -0,0 +1,95 @@
+-- Row positions are absolute within a data file. Variant row-group pruning must not shift them: a row
+-- group the predicate drops still occupies its rows, so every later row group keeps the position it had.
+set hive.explain.user=false;
+set hive.fetch.task.conversion=none;
+set hive.vectorized.execution.enabled=true;
+
+drop table if exists variant_row_position;
+
+CREATE EXTERNAL TABLE variant_row_position (
+  id INT,
+  data VARIANT
+) STORED BY ICEBERG
+TBLPROPERTIES (
+  'format-version'='3',
+  'variant.shredding.enabled'='true',
+  'write.parquet.row-group-size-bytes'='1024'
+);
+
+-- The first rows carry tier=bronze, the later ones tier=gold, and the small row group size puts them in
+-- different row groups. A predicate on tier drops the bronze ones, which is what shifts the positions of
+-- the gold ones when pruning is applied to the footer the reader counts over.
+INSERT INTO variant_row_position
+SELECT pos, parse_json(concat('{"tier": "', if(pos < 200, 'bronze', 'gold'), '", "n": ', pos, '}'))
+FROM (SELECT 1) x LATERAL VIEW posexplode(split(space(399), ' ')) e AS pos, val;
+
+-- ROW__POSITION of the surviving rows must match their id, which was written in file order.
+SELECT id, variant_row_position.ROW__POSITION
+FROM variant_row_position
+WHERE variant_get(data, '$.tier', 'string') = 'gold' AND id < 205
+ORDER BY id;
+
+-- the lowest surviving position is the first gold row, not zero
+SELECT min(variant_row_position.ROW__POSITION) AS first_gold_position,
+       max(variant_row_position.ROW__POSITION) AS last_gold_position,
+       count(*) AS gold_rows
+FROM variant_row_position
+WHERE variant_get(data, '$.tier', 'string') = 'gold';
+
+drop table variant_row_position;
+
+-- A file read as several splits: each split reports positions from the file's own start, so the row groups
+-- a later split reads must not be numbered as though its split began the file.
+drop table if exists variant_row_position_split;
+
+CREATE EXTERNAL TABLE variant_row_position_split (
+  id INT,
+  data VARIANT
+) STORED BY ICEBERG
+TBLPROPERTIES (
+  'format-version'='3',
+  'variant.shredding.enabled'='true',
+  'write.parquet.row-group-size-bytes'='1024',
+  'read.split.target-size'='1024'
+);
+
+INSERT INTO variant_row_position_split
+SELECT pos, parse_json(concat('{"tier": "', if(pos < 200, 'bronze', 'gold'), '", "n": ', pos, '}'))
+FROM (SELECT 1) x LATERAL VIEW posexplode(split(space(399), ' ')) e AS pos, val;
+
+SELECT min(variant_row_position_split.ROW__POSITION) AS first_gold_position,
+       max(variant_row_position_split.ROW__POSITION) AS last_gold_position,
+       count(*) AS gold_rows
+FROM variant_row_position_split
+WHERE variant_get(data, '$.tier', 'string') = 'gold';
+
+drop table variant_row_position_split;
+
+-- A positional delete addresses rows by position, so a position shifted by pruning deletes the wrong row.
+-- Here the delete predicate prunes row groups on the read side while the positions are being recorded.
+drop table if exists variant_row_position_del;
+
+CREATE EXTERNAL TABLE variant_row_position_del (
+  id INT,
+  data VARIANT
+) STORED BY ICEBERG
+TBLPROPERTIES (
+  'format-version'='3',
+  'variant.shredding.enabled'='true',
+  'write.parquet.row-group-size-bytes'='1024',
+  'write.delete.mode'='merge-on-read'
+);
+
+INSERT INTO variant_row_position_del
+SELECT pos, parse_json(concat('{"tier": "', if(pos < 200, 'bronze', 'gold'), '", "n": ', pos, '}'))
+FROM (SELECT 1) x LATERAL VIEW posexplode(split(space(399), ' ')) e AS pos, val;
+
+DELETE FROM variant_row_position_del
+WHERE variant_get(data, '$.tier', 'string') = 'gold' AND id < 205;
+
+-- exactly ids 200-204 are gone: 395 rows left, and the ids either side of the hole are untouched
+SELECT count(*) AS rows_left, min(id) AS lowest_id, max(id) AS highest_id FROM variant_row_position_del;
+
+SELECT id FROM variant_row_position_del WHERE id BETWEEN 197 AND 208 ORDER BY id;
+
+drop table variant_row_position_del;