HIVE-29544: Fix Vectorized Parquet reading Struct columns with all fields null (#6408)

Author: ayushtkn

iceberg/iceberg-handler/src/test/results/positive/iceberg_default_column.q.out

@@ -174,7 +174,7 @@ POSTHOOK: Output: hdfs://### HDFS PATH ###
 3	{"x":100,"y":99}	unknown	25	50000.0	true	2024-01-01	2024-01-01 10:00:00	100.00	general	NULL
 4	{"x":100,"y":99}	NULL	25	50000.0	true	2024-01-01	2024-01-01 10:00:00	100.00	general	NULL
 5	{"x":100,"y":99}	custom_name	30	50000.0	true	2024-01-01	2024-01-01 10:00:00	100.00	general	NULL
-6	NULL	unknown	25	50000.0	true	2024-01-01	2024-01-01 10:00:00	100.00	general	NULL
+6	{"x":null,"y":null}	unknown	25	50000.0	true	2024-01-01	2024-01-01 10:00:00	100.00	general	NULL
 7	NULL	null	NULL	50000.0	true	2024-01-01	2024-01-01 10:00:00	100.00	general	NULL
 8	NULL	null	NULL	50000.0	true	2024-01-01	2024-01-01 10:00:00	100.00	general	{"name":"John","address":{"street":"Main St","city":"New York"}}
 9	NULL	null	NULL	50000.0	true	2024-01-01	2024-01-01 10:00:00	100.00	general	{"name":null,"address":{"street":null,"city":"Bangalore"}}

ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/BaseVectorizedColumnReader.java

@@ -87,6 +87,9 @@ public abstract class BaseVectorizedColumnReader implements VectorizedColumnRead
   protected int definitionLevel;
   protected int repetitionLevel;
 
+  protected int[] currentDefLevels;
+  protected int defLevelIndex = 0;
+
   /**
    * Repetition/Definition/Value readers.
    */
@@ -154,6 +157,9 @@ public BaseVectorizedColumnReader(
   protected void readRepetitionAndDefinitionLevels() {
     repetitionLevel = repetitionLevelColumn.nextInt();
     definitionLevel = definitionLevelColumn.nextInt();
+    if (currentDefLevels != null && defLevelIndex < currentDefLevels.length) {
+      currentDefLevels[defLevelIndex++] = definitionLevel;
+    }
     valuesRead++;
   }
 
@@ -309,4 +315,9 @@ int nextInt() {
       return 0;
     }
   }
+
+  @Override
+  public int[] getDefinitionLevels() {
+    return currentDefLevels;
+  }
 }

ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/VectorizedColumnReader.java

@@ -36,4 +36,8 @@ void readBatch(
     int total,
     ColumnVector column,
     TypeInfo columnType) throws IOException;
+
+  default int[] getDefinitionLevels() {
+    return null;
+  }
 }

ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/VectorizedParquetRecordReader.java

@@ -459,14 +459,14 @@ private void checkEndOfRowGroup() throws IOException {
           columnReaders[i] =
               buildVectorizedParquetReader(columnTypesList.get(colsToInclude.get(i)), types.get(i),
                   pages, requestedSchema.getColumns(), skipTimestampConversion, writerTimezone, skipProlepticConversion,
-                  legacyConversionEnabled, 0);
+                  legacyConversionEnabled, 0, 0);
         }
       }
     } else {
       for (int i = 0; i < types.size(); ++i) {
         columnReaders[i] = buildVectorizedParquetReader(columnTypesList.get(i), types.get(i), pages,
           requestedSchema.getColumns(), skipTimestampConversion, writerTimezone, skipProlepticConversion,
-          legacyConversionEnabled, 0);
+          legacyConversionEnabled, 0, 0);
       }
     }
 
@@ -522,7 +522,12 @@ private VectorizedColumnReader buildVectorizedParquetReader(
     ZoneId writerTimezone,
     boolean skipProlepticConversion,
     boolean legacyConversionEnabled,
-    int depth) throws IOException {
+    int depth, int currentDefLevel) throws IOException {
+
+    int typeDefLevel = currentDefLevel;
+    if (type.isRepetition(Type.Repetition.OPTIONAL) || type.isRepetition(Type.Repetition.REPEATED)) {
+      typeDefLevel++;
+    }
     List<ColumnDescriptor> descriptors =
       getAllColumnDescriptorByType(depth, type, columnDescriptors);
     // Support for schema evolution: if the column from the current
@@ -549,8 +554,8 @@ private VectorizedColumnReader buildVectorizedParquetReader(
       List<Type> types = type.asGroupType().getFields();
       for (int i = 0; i < fieldTypes.size(); i++) {
         VectorizedColumnReader r =
-          buildVectorizedParquetReader(fieldTypes.get(i), types.get(i), pages, descriptors,
-            skipTimestampConversion, writerTimezone, skipProlepticConversion, legacyConversionEnabled, depth + 1);
+            buildVectorizedParquetReader(fieldTypes.get(i), types.get(i), pages, descriptors, skipTimestampConversion,
+                writerTimezone, skipProlepticConversion, legacyConversionEnabled, depth + 1, typeDefLevel);
         if (r != null) {
           fieldReaders.add(r);
         } else {
@@ -559,7 +564,7 @@ private VectorizedColumnReader buildVectorizedParquetReader(
               .getTypeName() + " and Parquet type" + types.get(i).toString());
         }
       }
-      return new VectorizedStructColumnReader(fieldReaders);
+      return new VectorizedStructColumnReader(fieldReaders, typeDefLevel);
     case LIST:
       checkListColumnSupport(((ListTypeInfo) typeInfo).getListElementTypeInfo());
       if (columnDescriptors == null || columnDescriptors.isEmpty()) {

ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/VectorizedPrimitiveColumnReader.java

@@ -64,6 +64,8 @@ public void readBatch(
       int total,
       ColumnVector column,
       TypeInfo columnType) throws IOException {
+    this.currentDefLevels = new int[total];
+    this.defLevelIndex = 0;
     int rowId = 0;
     while (total > 0) {
       // Compute the number of values we want to read in this page.

ql/src/java/org/apache/hadoop/hive/ql/io/parquet/vector/VectorizedStructColumnReader.java

@@ -29,9 +29,11 @@
 public class VectorizedStructColumnReader implements VectorizedColumnReader {
 
   private final List<VectorizedColumnReader> fieldReaders;
+  private final int structDefLevel;
 
-  public VectorizedStructColumnReader(List<VectorizedColumnReader> fieldReaders) {
+  public VectorizedStructColumnReader(List<VectorizedColumnReader> fieldReaders, int structDefLevel) {
     this.fieldReaders = fieldReaders;
+    this.structDefLevel = structDefLevel;
   }
 
   @Override
@@ -46,14 +48,29 @@ public void readBatch(
       fieldReaders.get(i)
         .readBatch(total, vectors[i], structTypeInfo.getAllStructFieldTypeInfos().get(i));
       structColumnVector.isRepeating = structColumnVector.isRepeating && vectors[i].isRepeating;
+    }
+    int[] defLevels = getDefinitionLevels();
 
-      for (int j = 0; j < vectors[i].isNull.length; j++) {
-        structColumnVector.isNull[j] =
-          (i == 0) ? vectors[i].isNull[j] : structColumnVector.isNull[j] && vectors[i].isNull[j];
+    // Evaluate struct nullability using Parquet Definition Levels
+    if (defLevels != null) {
+      for (int j = 0; j < total; j++) {
+        if (defLevels[j] < structDefLevel) {
+          // The Definition Level boundary crossed the struct. The whole struct is null.
+          structColumnVector.isNull[j] = true;
+          structColumnVector.noNulls = false;
+        }
       }
-      structColumnVector.noNulls =
-        (i == 0) ? vectors[i].noNulls : structColumnVector.noNulls && vectors[i].noNulls;
     }
+  }
 
+  @Override
+  public int[] getDefinitionLevels() {
+    for (VectorizedColumnReader reader : fieldReaders) {
+      int[] defLevels = reader.getDefinitionLevels();
+      if (defLevels != null) {
+        return defLevels;
+      }
+    }
+    return null;
   }
 }

ql/src/test/org/apache/hadoop/hive/ql/io/parquet/VectorizedColumnReaderTestBase.java

@@ -317,14 +317,15 @@ protected static void writeData(ParquetWriter<Group> writer, boolean isDictionar
       g.addGroup("nsf").append("c", intVal).append("d", intVal);
       g.append("e", doubleVal);
 
-      Group some_null_g = group.addGroup("struct_field_some_null");
-      if (i % 2 != 0) {
-        some_null_g.append("f", intVal);
-      }
-      if (i % 3 != 0) {
-        some_null_g.append("g", doubleVal);
+      if (i % 2 != 0 || i % 3 != 0) {
+        Group structFieldWithNulls = group.addGroup("struct_field_some_null");
+        if (i % 2 != 0) {
+          structFieldWithNulls.append("f", intVal);
+        }
+        if (i % 3 != 0) {
+          structFieldWithNulls.append("g", doubleVal);
+        }
       }
-
       Group mapGroup = group.addGroup("map_field");
       if (i % 13 != 1) {
         mapGroup.addGroup("map").append("key", binary).append("value", "abc");

ql/src/test/queries/clientpositive/parquet_struct_with_null_vectorization.q

@@ -0,0 +1,31 @@
+-- SORT_QUERY_RESULTS
+SET hive.vectorized.execution.enabled=true;
+set hive.vectorized.execution.reduce.enabled=true;
+SET hive.fetch.task.conversion=none;
+
+CREATE TABLE test_parquet_struct_nulls (
+    id INT,
+    st_prim STRUCT<x:INT, y:INT>
+) STORED AS PARQUET;
+
+INSERT INTO test_parquet_struct_nulls VALUES
+    (1, named_struct('x', CAST(NULL AS INT), 'y', CAST(NULL AS INT))),
+    (2, if(1=0, named_struct('x', 1, 'y', 1), null)),
+    (3, named_struct('x', 3, 'y', CAST(NULL AS INT))),
+    (4, named_struct('x', 4, 'y', 4));
+
+-- Test A: Full table scan to check JSON representation
+SELECT * FROM test_parquet_struct_nulls;
+
+-- Test B: Verify IS NULL evaluates correctly
+SELECT id FROM test_parquet_struct_nulls WHERE st_prim IS NULL;
+
+-- Test C: Verify IS NOT NULL evaluates correctly
+SELECT id FROM test_parquet_struct_nulls WHERE st_prim IS NOT NULL;
+
+-- Test D: Verify field-level null evaluation inside a valid struct
+SELECT id FROM test_parquet_struct_nulls WHERE st_prim IS NOT NULL AND st_prim.x IS NULL;
+
+-- Validate without vectorization
+SET hive.vectorized.execution.enabled=false;
+SELECT * FROM test_parquet_struct_nulls;

ql/src/test/results/clientpositive/llap/parquet_struct_with_null_vectorization.q.out

@@ -0,0 +1,85 @@
+PREHOOK: query: CREATE TABLE test_parquet_struct_nulls (
+    id INT,
+    st_prim STRUCT<x:INT, y:INT>
+) STORED AS PARQUET
+PREHOOK: type: CREATETABLE
+PREHOOK: Output: database:default
+PREHOOK: Output: default@test_parquet_struct_nulls
+POSTHOOK: query: CREATE TABLE test_parquet_struct_nulls (
+    id INT,
+    st_prim STRUCT<x:INT, y:INT>
+) STORED AS PARQUET
+POSTHOOK: type: CREATETABLE
+POSTHOOK: Output: database:default
+POSTHOOK: Output: default@test_parquet_struct_nulls
+PREHOOK: query: INSERT INTO test_parquet_struct_nulls VALUES
+    (1, named_struct('x', CAST(NULL AS INT), 'y', CAST(NULL AS INT))),
+    (2, if(1=0, named_struct('x', 1, 'y', 1), null)),
+    (3, named_struct('x', 3, 'y', CAST(NULL AS INT))),
+    (4, named_struct('x', 4, 'y', 4))
+PREHOOK: type: QUERY
+PREHOOK: Input: _dummy_database@_dummy_table
+PREHOOK: Output: default@test_parquet_struct_nulls
+POSTHOOK: query: INSERT INTO test_parquet_struct_nulls VALUES
+    (1, named_struct('x', CAST(NULL AS INT), 'y', CAST(NULL AS INT))),
+    (2, if(1=0, named_struct('x', 1, 'y', 1), null)),
+    (3, named_struct('x', 3, 'y', CAST(NULL AS INT))),
+    (4, named_struct('x', 4, 'y', 4))
+POSTHOOK: type: QUERY
+POSTHOOK: Input: _dummy_database@_dummy_table
+POSTHOOK: Output: default@test_parquet_struct_nulls
+POSTHOOK: Lineage: test_parquet_struct_nulls.id SCRIPT []
+POSTHOOK: Lineage: test_parquet_struct_nulls.st_prim SCRIPT []
+PREHOOK: query: SELECT * FROM test_parquet_struct_nulls
+PREHOOK: type: QUERY
+PREHOOK: Input: default@test_parquet_struct_nulls
+#### A masked pattern was here ####
+POSTHOOK: query: SELECT * FROM test_parquet_struct_nulls
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@test_parquet_struct_nulls
+#### A masked pattern was here ####
+1	{"x":null,"y":null}
+2	NULL
+3	{"x":3,"y":null}
+4	{"x":4,"y":4}
+PREHOOK: query: SELECT id FROM test_parquet_struct_nulls WHERE st_prim IS NULL
+PREHOOK: type: QUERY
+PREHOOK: Input: default@test_parquet_struct_nulls
+#### A masked pattern was here ####
+POSTHOOK: query: SELECT id FROM test_parquet_struct_nulls WHERE st_prim IS NULL
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@test_parquet_struct_nulls
+#### A masked pattern was here ####
+2
+PREHOOK: query: SELECT id FROM test_parquet_struct_nulls WHERE st_prim IS NOT NULL
+PREHOOK: type: QUERY
+PREHOOK: Input: default@test_parquet_struct_nulls
+#### A masked pattern was here ####
+POSTHOOK: query: SELECT id FROM test_parquet_struct_nulls WHERE st_prim IS NOT NULL
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@test_parquet_struct_nulls
+#### A masked pattern was here ####
+1
+3
+4
+PREHOOK: query: SELECT id FROM test_parquet_struct_nulls WHERE st_prim IS NOT NULL AND st_prim.x IS NULL
+PREHOOK: type: QUERY
+PREHOOK: Input: default@test_parquet_struct_nulls
+#### A masked pattern was here ####
+POSTHOOK: query: SELECT id FROM test_parquet_struct_nulls WHERE st_prim IS NOT NULL AND st_prim.x IS NULL
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@test_parquet_struct_nulls
+#### A masked pattern was here ####
+1
+PREHOOK: query: SELECT * FROM test_parquet_struct_nulls
+PREHOOK: type: QUERY
+PREHOOK: Input: default@test_parquet_struct_nulls
+#### A masked pattern was here ####
+POSTHOOK: query: SELECT * FROM test_parquet_struct_nulls
+POSTHOOK: type: QUERY
+POSTHOOK: Input: default@test_parquet_struct_nulls
+#### A masked pattern was here ####
+1	{"x":null,"y":null}
+2	NULL
+3	{"x":3,"y":null}
+4	{"x":4,"y":4}