Arrow: add support for null vectors

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

@@ -50,7 +50,6 @@ public final ColumnarBatch read(ColumnarBatch reuse, int numRowsToRead) {
           "Number of rows in the vector %s didn't match expected %s ",
           numRowsInVector,
           numRowsToRead);
-      // Handle null vector for constant case
       columnVectors[i] = new ColumnVector(vectorHolders[i]);
     }
     return new ColumnarBatch(numRowsToRead, columnVectors);

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

@@ -35,6 +35,7 @@
 import org.apache.arrow.vector.Float4Vector;
 import org.apache.arrow.vector.Float8Vector;
 import org.apache.arrow.vector.IntVector;
+import org.apache.arrow.vector.NullVector;
 import org.apache.arrow.vector.TimeMicroVector;
 import org.apache.arrow.vector.TimeStampMicroTZVector;
 import org.apache.arrow.vector.TimeStampMicroVector;
@@ -220,8 +221,11 @@ private ArrowVectorAccessor<DecimalT, Utf8StringT, ArrayT, ChildVectorT> getPlai
       }
       return new FixedSizeBinaryAccessor<>(
           (FixedSizeBinaryVector) vector, stringFactorySupplier.get());
+    } else if (vector instanceof NullVector) {
+      return new NullAccessor<>((NullVector) vector);
     }
-    throw new UnsupportedOperationException("Unsupported vector: " + vector.getClass());
+    String vectorName = (vector == null) ? "null" : vector.getClass().toString();
+    throw new UnsupportedOperationException("Unsupported vector: " + vectorName);
   }
 
   private static boolean isDecimal(PrimitiveType primitive) {
@@ -244,6 +248,18 @@ public final boolean getBoolean(int rowId) {
     }
   }
 
+  private static class NullAccessor<
+          DecimalT, Utf8StringT, ArrayT, ChildVectorT extends AutoCloseable>
+      extends ArrowVectorAccessor<DecimalT, Utf8StringT, ArrayT, ChildVectorT> {
+
+    private final NullVector vector;
+
+    NullAccessor(NullVector vector) {
+      super(vector);
+      this.vector = vector;
+    }
+  }
+
   private static class IntAccessor<
           DecimalT, Utf8StringT, ArrayT, ChildVectorT extends AutoCloseable>
       extends ArrowVectorAccessor<DecimalT, Utf8StringT, ArrayT, ChildVectorT> {

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

@@ -131,6 +131,20 @@ public boolean isDummy() {
     return vector == null;
   }
 
+  public static class NullVectorHolder extends VectorHolder {
+    private final int numRows;
+
+    public NullVectorHolder(FieldVector vec, Types.NestedField field, int numRows) {
+      super(vec, field, new NullabilityHolder(numRows));
+      this.numRows = numRows;
+    }
+
+    @Override
+    public int numValues() {
+      return this.numRows;
+    }
+  }
+
   /**
    * A Vector Holder which does not actually produce values, consumers of this class should use the
    * constantValue to populate their ColumnVector implementation.

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

@@ -30,6 +30,7 @@
 import org.apache.arrow.vector.Float4Vector;
 import org.apache.arrow.vector.Float8Vector;
 import org.apache.arrow.vector.IntVector;
+import org.apache.arrow.vector.NullVector;
 import org.apache.arrow.vector.TimeMicroVector;
 import org.apache.arrow.vector.TimeStampMicroTZVector;
 import org.apache.arrow.vector.TimeStampMicroVector;
@@ -451,10 +452,6 @@ public String toString() {
     return columnDescriptor.toString();
   }
 
-  public static VectorizedArrowReader nulls() {
-    return NullVectorReader.INSTANCE;
-  }
-
   public static VectorizedArrowReader positions() {
     return new PositionVectorReader(false);
   }
@@ -463,12 +460,16 @@ public static VectorizedArrowReader positionsWithSetArrowValidityVector() {
     return new PositionVectorReader(true);
   }
 
-  private static final class NullVectorReader extends VectorizedArrowReader {
-    private static final NullVectorReader INSTANCE = new NullVectorReader();
+  public static final class NullVectorReader extends VectorizedArrowReader {
+
+    public NullVectorReader(Types.NestedField icebergField) {
+      super(icebergField);
+    }
 
     @Override
     public VectorHolder read(VectorHolder reuse, int numValsToRead) {
-      return VectorHolder.dummyHolder(numValsToRead);
+      NullVector vector = new NullVector(icebergField().name(), numValsToRead);
+      return new VectorHolder.NullVectorHolder(vector, icebergField(), numValsToRead);
     }
 
     @Override

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

@@ -28,6 +28,7 @@
 import org.apache.iceberg.arrow.ArrowAllocation;
 import org.apache.iceberg.arrow.vectorized.VectorizedArrowReader.ConstantVectorReader;
 import org.apache.iceberg.arrow.vectorized.VectorizedArrowReader.DeletedVectorReader;
+import org.apache.iceberg.arrow.vectorized.VectorizedArrowReader.NullVectorReader;
 import org.apache.iceberg.parquet.TypeWithSchemaVisitor;
 import org.apache.iceberg.parquet.VectorizedReader;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
@@ -97,7 +98,7 @@ public VectorizedReader<?> message(
       } else if (reader != null) {
         reorderedFields.add(reader);
       } else {
-        reorderedFields.add(VectorizedArrowReader.nulls());
+        reorderedFields.add(new NullVectorReader(field));
       }
     }
     return vectorizedReader(reorderedFields);

arrow/src/test/java/org/apache/iceberg/arrow/vectorized/ArrowReaderTest.java

@@ -19,7 +19,7 @@
 package org.apache.iceberg.arrow.vectorized;
 
 import static org.apache.iceberg.Files.localInput;
-import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.*;
 
 import java.io.File;
 import java.io.IOException;
@@ -59,6 +59,7 @@
 import org.apache.arrow.vector.types.pojo.ArrowType;
 import org.apache.arrow.vector.types.pojo.Field;
 import org.apache.arrow.vector.types.pojo.FieldType;
+import org.apache.iceberg.AppendFiles;
 import org.apache.iceberg.DataFile;
 import org.apache.iceberg.DataFiles;
 import org.apache.iceberg.FileFormat;
@@ -262,6 +263,89 @@ public void testReadColumnFilter2() throws Exception {
         scan, NUM_ROWS_PER_MONTH, 12 * NUM_ROWS_PER_MONTH, ImmutableList.of("timestamp"));
   }
 
+  @Test
+  public void testReadColumnThatDoesNotExistInParquetSchema() throws Exception {
+    setMaxStackTraceElementsDisplayed(15);
+    rowsWritten = Lists.newArrayList();
+    tables = new HadoopTables();
+
+    Schema schema =
+        new Schema(
+            Types.NestedField.required(1, "a", Types.IntegerType.get()),
+            Types.NestedField.optional(2, "b", Types.IntegerType.get()));
+
+    PartitionSpec spec = PartitionSpec.builderFor(schema).build();
+    Table table1 = tables.create(schema, spec, tableLocation);
+
+    // Add one record to the table
+    GenericRecord rec = GenericRecord.create(schema);
+    rec.setField("a", 1);
+    List<GenericRecord> genericRecords = Lists.newArrayList();
+    genericRecords.add(rec);
+
+    AppendFiles appendFiles = table1.newAppend();
+    appendFiles.appendFile(writeParquetFile(table1, genericRecords));
+    appendFiles.commit();
+
+    // Alter the table schema by adding a new, optional column.
+    // Do not add any data for this new column in the one existing row in the table
+    // and do not insert any new rows into the table.
+    Table table = tables.load(tableLocation);
+    table.updateSchema().addColumn("z", Types.IntegerType.get()).commit();
+
+    // Select all columns, all rows from the table
+    TableScan scan = table.newScan().select("*");
+
+    List<String> columns = ImmutableList.of("a", "b", "z");
+    // Read the data and verify that the returned ColumnarBatches match expected rows.
+    int rowIndex = 0;
+    try (VectorizedTableScanIterable itr = new VectorizedTableScanIterable(scan, 1, false)) {
+      for (ColumnarBatch batch : itr) {
+        List<GenericRecord> expectedRows = rowsWritten.subList(rowIndex, rowIndex + 1);
+
+        Map<String, Integer> columnNameToIndex = Maps.newHashMap();
+        for (int i = 0; i < columns.size(); i++) {
+          columnNameToIndex.put(columns.get(i), i);
+        }
+        Set<String> columnSet = columnNameToIndex.keySet();
+
+        assertThat(batch.numRows()).isEqualTo(1);
+        assertThat(batch.numCols()).isEqualTo(columns.size());
+
+        checkColumnarArrayValues(
+            1,
+            expectedRows,
+            batch,
+            0,
+            columnSet,
+            "a",
+            (records, i) -> records.get(i).getField("a"),
+            ColumnVector::getInt);
+        checkColumnarArrayValues(
+            1,
+            expectedRows,
+            batch,
+            1,
+            columnSet,
+            "b",
+            (records, i) -> records.get(i).getField("b"),
+            (array, i) -> array.isNullAt(i) ? null : array.getInt(i));
+        checkColumnarArrayValues(
+            1,
+            expectedRows,
+            batch,
+            2,
+            columnSet,
+            "z",
+            (records, i) -> records.get(i).getField("z"),
+            (array, i) -> array.isNullAt(i) ? null : array.getInt(i));
+        rowIndex += 1;
+      }
+    }
+    // Read the data and verify that the returned Arrow VectorSchemaRoots match expected rows.
+    readAndCheckArrowResult(scan, 1, 1, columns);
+  }
+
   /**
    * The test asserts that {@link CloseableIterator#hasNext()} returned by the {@link ArrowReader}
    * is idempotent.

arrow/src/test/java/org/apache/iceberg/arrow/vectorized/GenericArrowVectorAccessorFactoryTest.java

@@ -0,0 +1,86 @@
+/*
+ * 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.arrow.vectorized;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+import java.math.BigDecimal;
+import java.util.function.Supplier;
+import org.apache.arrow.vector.IntVector;
+import org.apache.iceberg.types.Types;
+import org.apache.parquet.column.ColumnDescriptor;
+import org.apache.parquet.schema.PrimitiveType;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.MockitoAnnotations;
+
+class GenericArrowVectorAccessorFactoryTest {
+  @Mock
+  Supplier<GenericArrowVectorAccessorFactory.DecimalFactory<BigDecimal>> decimalFactorySupplier;
+
+  @Mock Supplier<GenericArrowVectorAccessorFactory.StringFactory<String>> stringFactorySupplier;
+
+  @Mock
+  Supplier<GenericArrowVectorAccessorFactory.StructChildFactory<Integer>>
+      structChildFactorySupplier;
+
+  @Mock
+  Supplier<GenericArrowVectorAccessorFactory.ArrayFactory<Integer, Integer[]>> arrayFactorySupplier;
+
+  @InjectMocks GenericArrowVectorAccessorFactory genericArrowVectorAccessorFactory;
+
+  @BeforeEach
+  void before() {
+    MockitoAnnotations.openMocks(this);
+  }
+
+  @Test
+  void testGetVectorAccessorWithIntVector() {
+    IntVector vector = mock(IntVector.class);
+    when(vector.get(0)).thenReturn(88);
+
+    Types.NestedField nestedField = Types.NestedField.optional(0, "a1", Types.IntegerType.get());
+    ColumnDescriptor columnDescriptor =
+        new ColumnDescriptor(
+            new String[] {nestedField.name()}, PrimitiveType.PrimitiveTypeName.INT32, 0, 1);
+    NullabilityHolder nullabilityHolder = new NullabilityHolder(10000);
+    VectorHolder vectorHolder =
+        new VectorHolder(columnDescriptor, vector, false, null, nullabilityHolder, nestedField);
+    ArrowVectorAccessor actual = genericArrowVectorAccessorFactory.getVectorAccessor(vectorHolder);
+    assertThat(actual).isNotNull();
+    assertThat(actual).isInstanceOf(ArrowVectorAccessor.class);
+    int intValue = actual.getInt(0);
+    assertThat(intValue).isEqualTo(88);
+  }
+
+  @Test
+  void testGetVectorAccessorWithNullVector() {
+    assertThatThrownBy(
+            () -> {
+              genericArrowVectorAccessorFactory.getVectorAccessor(VectorHolder.dummyHolder(1));
+            })
+        .isInstanceOf(UnsupportedOperationException.class)
+        .hasMessage("Unsupported vector: null");
+  }
+}