Support default value read for ORC format in spark

orc/src/main/java/org/apache/iceberg/orc/ORCSchemaUtil.java

@@ -270,6 +270,9 @@ private static TypeDescription buildOrcProjection(Integer fieldId, Type type, bo
           // Using suffix _r to avoid potential underlying issues in ORC reader
           // with reused column names between ORC and Iceberg;
           // e.g. renaming column c -> d and adding new column d
+          if (mapping.get(nestedField.fieldId()) == null && nestedField.hasDefaultValue()) {
+            continue;
+          }
           String name = Optional.ofNullable(mapping.get(nestedField.fieldId()))
               .map(OrcField::name)
               .orElseGet(() -> nestedField.name() + "_r" + nestedField.fieldId());
@@ -387,7 +390,7 @@ static Optional<Integer> icebergID(TypeDescription orcType) {
         .map(Integer::parseInt);
   }
 
-  static int fieldId(TypeDescription orcType) {
+  public static int fieldId(TypeDescription orcType) {
     String idStr = orcType.getAttributeValue(ICEBERG_ID_ATTRIBUTE);
     Preconditions.checkNotNull(idStr, "Missing expected '%s' property", ICEBERG_ID_ATTRIBUTE);
     return Integer.parseInt(idStr);

orc/src/main/java/org/apache/iceberg/orc/OrcSchemaWithTypeVisitor.java

@@ -35,7 +35,7 @@ public static <T> T visit(
   public static <T> T visit(Type iType, TypeDescription schema, OrcSchemaWithTypeVisitor<T> visitor) {
     switch (schema.getCategory()) {
       case STRUCT:
-        return visitRecord(iType != null ? iType.asStructType() : null, schema, visitor);
+        return visitor.visitRecord(iType != null ? iType.asStructType() : null, schema, visitor);
 
       case UNION:
         throw new UnsupportedOperationException("Cannot handle " + schema);
@@ -58,7 +58,7 @@ public static <T> T visit(Type iType, TypeDescription schema, OrcSchemaWithTypeV
     }
   }
 
-  private static <T> T visitRecord(
+  protected T visitRecord(
       Types.StructType struct, TypeDescription record, OrcSchemaWithTypeVisitor<T> visitor) {
     List<TypeDescription> fields = record.getChildren();
     List<String> names = record.getFieldNames();

spark/src/main/java/org/apache/iceberg/spark/OrcSchemaWithTypeVisitorSpark.java

@@ -0,0 +1,110 @@
+/*
+ * 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.spark;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import org.apache.iceberg.MetadataColumns;
+import org.apache.iceberg.orc.ORCSchemaUtil;
+import org.apache.iceberg.orc.OrcSchemaWithTypeVisitor;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.Types;
+import org.apache.orc.TypeDescription;
+
+public abstract class OrcSchemaWithTypeVisitorSpark<T> extends OrcSchemaWithTypeVisitor<T> {
+
+  private final Map<Integer, Object> idToConstant;
+
+  public Map<Integer, Object> getIdToConstant() {
+    return idToConstant;
+  }
+
+  protected OrcSchemaWithTypeVisitorSpark(Map<Integer, ?> idToConstant) {
+    this.idToConstant = new HashMap<>();
+    this.idToConstant.putAll(idToConstant);
+  }
+
+  @Override
+  protected T visitRecord(
+          Types.StructType struct, TypeDescription record, OrcSchemaWithTypeVisitor<T> visitor) {
+    Preconditions.checkState(
+            checkIcebergAndOrcSchemaAlignment(struct, record),
+            "Iceberg schema and ORC schema doesn't align, please call ORCSchemaUtil.buildOrcProjection" +
+                    "to get an aligned ORC schema first!"
+    );
+    List<Types.NestedField> iFields = struct.fields();
+    List<TypeDescription> fields = record.getChildren();
+    List<String> names = record.getFieldNames();
+    List<T> results = Lists.newArrayListWithExpectedSize(fields.size());
+
+    for (int i = 0, j = 0; i < iFields.size(); i++) {
+      Types.NestedField iField = iFields.get(i);
+      TypeDescription field = j < fields.size() ? fields.get(j) : null;
+      if (field == null || (iField.fieldId() != ORCSchemaUtil.fieldId(field))) {
+        // there are 3 cases where we need to use idToConstant for an iField
+        // 1. The field is MetadataColumns.ROW_POSITION, we build a RowPositionReader
+        // 2. The field is a partition column, we build a ConstantReader
+        // 3. The field should be read using the default value, where we build a ConstantReader
+        // Here we should only need to update idToConstant when it's the 3rd case,
+        // because the first 2 cases have been handled by logic elsewhere.
+        if (!iField.equals(MetadataColumns.ROW_POSITION) &&
+                !idToConstant.containsKey(iField.fieldId())) {
+          idToConstant.put(iField.fieldId(), iField.getDefaultValue());
+        }
+      } else {
+        results.add(visit(iField.type(), field, visitor));
+        j++;
+      }
+    }
+    return visitor.record(struct, record, names, results);
+  }
+
+  private static boolean checkIcebergAndOrcSchemaAlignment(Types.StructType struct, TypeDescription record) {
+    List<Integer> icebergIDList = struct.fields().stream().map(Types.NestedField::fieldId).collect(Collectors.toList());
+    List<Integer> orcIDList = record.getChildren().stream().map(ORCSchemaUtil::fieldId).collect(Collectors.toList());
+
+    // icebergIDList should be a superset of orcIDList, and the overlapping ids should appear
+    // in the same order in these 2 lists
+    return checkTwoListAlignmentHelper(icebergIDList, orcIDList);
+  }
+
+  private static boolean checkTwoListAlignmentHelper(List<Integer> list1, List<Integer> list2) {
+    if (list1.size() < list2.size()) {
+      return false;
+    }
+
+    for (int i = 0, j = 0; j < list2.size(); j++) {
+      if (i >= list1.size()) {
+        return false;
+      }
+      while (!list1.get(i).equals(list2.get(j))) {
+        i++;
+        if (i >= list1.size()) {
+          return false;
+        }
+      }
+      i++;
+    }
+    return true;
+  }
+}

spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcReader.java

@@ -26,6 +26,7 @@
 import org.apache.iceberg.orc.OrcValueReader;
 import org.apache.iceberg.orc.OrcValueReaders;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.spark.OrcSchemaWithTypeVisitorSpark;
 import org.apache.iceberg.types.Type;
 import org.apache.iceberg.types.Types;
 import org.apache.orc.TypeDescription;
@@ -62,17 +63,16 @@ public void setBatchContext(long batchOffsetInFile) {
     reader.setBatchContext(batchOffsetInFile);
   }
 
-  private static class ReadBuilder extends OrcSchemaWithTypeVisitor<OrcValueReader<?>> {
-    private final Map<Integer, ?> idToConstant;
+  public static class ReadBuilder extends OrcSchemaWithTypeVisitorSpark<OrcValueReader<?>> {
 
     private ReadBuilder(Map<Integer, ?> idToConstant) {
-      this.idToConstant = idToConstant;
+      super(idToConstant);
     }
 
     @Override
     public OrcValueReader<?> record(
         Types.StructType expected, TypeDescription record, List<String> names, List<OrcValueReader<?>> fields) {
-      return SparkOrcValueReaders.struct(fields, expected, idToConstant);
+      return SparkOrcValueReaders.struct(fields, expected, getIdToConstant());
     }
 
     @Override

spark/src/main/java/org/apache/iceberg/spark/data/vectorized/ConstantColumnVector.java

@@ -109,6 +109,9 @@ public Decimal getDecimal(int rowId, int precision, int scale) {
 
   @Override
   public UTF8String getUTF8String(int rowId) {
+    if (constant instanceof String) {
+      return UTF8String.fromString((String) constant);
+    }
     return (UTF8String) constant;
   }
 

spark/src/main/java/org/apache/iceberg/spark/data/vectorized/VectorizedSparkOrcReaders.java

@@ -29,6 +29,7 @@
 import org.apache.iceberg.orc.OrcValueReader;
 import org.apache.iceberg.orc.OrcValueReaders;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.spark.OrcSchemaWithTypeVisitorSpark;
 import org.apache.iceberg.spark.SparkSchemaUtil;
 import org.apache.iceberg.spark.data.SparkOrcValueReaders;
 import org.apache.iceberg.types.Type;
@@ -80,17 +81,16 @@ ColumnVector convert(org.apache.orc.storage.ql.exec.vector.ColumnVector columnVe
                          long batchOffsetInFile);
   }
 
-  private static class ReadBuilder extends OrcSchemaWithTypeVisitor<Converter> {
-    private final Map<Integer, ?> idToConstant;
+  private static class ReadBuilder extends OrcSchemaWithTypeVisitorSpark<Converter> {
 
     private ReadBuilder(Map<Integer, ?> idToConstant) {
-      this.idToConstant = idToConstant;
+      super(idToConstant);
     }
 
     @Override
     public Converter record(Types.StructType iStruct, TypeDescription record, List<String> names,
                             List<Converter> fields) {
-      return new StructConverter(iStruct, fields, idToConstant);
+      return new StructConverter(iStruct, fields, getIdToConstant());
     }
 
     @Override

spark/src/test/java/org/apache/iceberg/spark/data/TestSparkOrcReaderForFieldsWithDefaultValue.java

@@ -0,0 +1,130 @@
+/*
+ * 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.spark.data;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.Iterator;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.iceberg.Files;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.orc.ORC;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.relocated.com.google.common.collect.Iterators;
+import org.apache.iceberg.spark.data.vectorized.VectorizedSparkOrcReaders;
+import org.apache.iceberg.types.Types;
+import org.apache.orc.OrcFile;
+import org.apache.orc.TypeDescription;
+import org.apache.orc.Writer;
+import org.apache.orc.storage.ql.exec.vector.LongColumnVector;
+import org.apache.orc.storage.ql.exec.vector.VectorizedRowBatch;
+import org.apache.spark.sql.catalyst.InternalRow;
+import org.apache.spark.sql.catalyst.expressions.GenericInternalRow;
+import org.apache.spark.sql.vectorized.ColumnarBatch;
+import org.apache.spark.unsafe.types.UTF8String;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import static org.apache.iceberg.spark.data.TestHelpers.assertEquals;
+
+
+public class TestSparkOrcReaderForFieldsWithDefaultValue {
+
+  @Rule
+  public TemporaryFolder temp = new TemporaryFolder();
+
+  @Test
+  public void testOrcDefaultValues() throws IOException {
+    final int numRows = 10;
+
+    final InternalRow expectedFirstRow = new GenericInternalRow(2);
+    expectedFirstRow.update(0, 0);
+    expectedFirstRow.update(1, "foo");
+
+    final InternalRow expectedFirstRowFromBatch = expectedFirstRow.copy();
+    expectedFirstRowFromBatch.update(1, UTF8String.fromString("foo"));
+
+    TypeDescription orcSchema =
+            TypeDescription.fromString("struct<col1:int>");
+
+    Schema readSchema = new Schema(
+            Types.NestedField.required(1, "col1", Types.IntegerType.get()),
+            Types.NestedField.required(2, "col2", Types.StringType.get(), "foo", null)
+    );
+
+    Configuration conf = new Configuration();
+
+    File orcFile = temp.newFile();
+    Path orcFilePath = new Path(orcFile.getPath());
+
+    Writer writer = OrcFile.createWriter(orcFilePath,
+            OrcFile.writerOptions(conf).setSchema(orcSchema).overwrite(true));
+
+    VectorizedRowBatch batch = orcSchema.createRowBatch();
+    LongColumnVector firstCol = (LongColumnVector) batch.cols[0];
+    for (int r = 0; r < numRows; ++r) {
+      int row = batch.size++;
+      firstCol.vector[row] = r;
+      // If the batch is full, write it out and start over.
+      if (batch.size == batch.getMaxSize()) {
+        writer.addRowBatch(batch);
+        batch.reset();
+      }
+    }
+    if (batch.size != 0) {
+      writer.addRowBatch(batch);
+      batch.reset();
+    }
+    writer.close();
+
+    // try to read the data using the readSchema, which is an evolved
+    // schema that contains a new column with default value
+
+    // non-vectorized read
+    try (CloseableIterable<InternalRow> reader = ORC.read(Files.localInput(orcFile))
+            .project(readSchema)
+            .createReaderFunc(readOrcSchema -> new SparkOrcReader(readSchema, readOrcSchema))
+            .build()) {
+      final Iterator<InternalRow> actualRows = reader.iterator();
+      final InternalRow actualFirstRow = actualRows.next();
+
+      assertEquals(readSchema, expectedFirstRow, actualFirstRow);
+    }
+
+    // vectorized-read
+    try (CloseableIterable<ColumnarBatch> reader = ORC.read(Files.localInput(orcFile))
+            .project(readSchema)
+            .createBatchedReaderFunc(readOrcSchema ->
+                    VectorizedSparkOrcReaders.buildReader(readSchema, readOrcSchema, ImmutableMap.of()))
+            .build()) {
+      final Iterator<InternalRow> actualRows = batchesToRows(reader.iterator());
+      final InternalRow actualFirstRow = actualRows.next();
+
+      assertEquals(readSchema, expectedFirstRowFromBatch, actualFirstRow);
+    }
+  }
+
+  private Iterator<InternalRow> batchesToRows(Iterator<ColumnarBatch> batches) {
+    return Iterators.concat(Iterators.transform(batches, ColumnarBatch::rowIterator));
+  }
+}