ORC: Supported nested identity partition data

data/src/main/java/org/apache/iceberg/data/orc/GenericOrcReader.java

@@ -39,7 +39,7 @@
 import org.apache.iceberg.data.GenericRecord;
 import org.apache.iceberg.data.Record;
 import org.apache.iceberg.orc.ORCSchemaUtil;
-import org.apache.iceberg.orc.OrcValueReader;
+import org.apache.iceberg.orc.OrcRowReader;
 import org.apache.iceberg.types.Types;
 import org.apache.orc.TypeDescription;
 import org.apache.orc.storage.ql.exec.vector.BytesColumnVector;
@@ -56,7 +56,7 @@
 /**
  * ORC reader for Generic Record.
  */
-public class GenericOrcReader implements OrcValueReader<Record> {
+public class GenericOrcReader implements OrcRowReader<Record> {
 
   private final Schema schema;
   private final List<TypeDescription> columns;
@@ -82,7 +82,7 @@ private Converter[] buildConverters() {
     return newConverters;
   }
 
-  public static OrcValueReader<Record> buildReader(Schema expectedSchema, TypeDescription fileSchema) {
+  public static OrcRowReader<Record> buildReader(Schema expectedSchema, TypeDescription fileSchema) {
     return new GenericOrcReader(expectedSchema, fileSchema);
   }
 

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

@@ -123,7 +123,7 @@ public static class ReadBuilder {
     private Long start = null;
     private Long length = null;
 
-    private Function<TypeDescription, OrcValueReader<?>> readerFunc;
+    private Function<TypeDescription, OrcRowReader<?>> readerFunc;
 
     private ReadBuilder(InputFile file) {
       Preconditions.checkNotNull(file, "Input file cannot be null");
@@ -163,7 +163,7 @@ public ReadBuilder config(String property, String value) {
       return this;
     }
 
-    public ReadBuilder createReaderFunc(Function<TypeDescription, OrcValueReader<?>> readerFunction) {
+    public ReadBuilder createReaderFunc(Function<TypeDescription, OrcRowReader<?>> readerFunction) {
       this.readerFunc = readerFunction;
       return this;
     }

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

@@ -308,7 +308,7 @@ private static TypeDescription buildOrcProjection(Integer fieldId, Type type, bo
           orcType = convert(fieldId, type, false);
         }
     }
-
+    orcType.setAttribute(ICEBERG_ID_ATTRIBUTE, fieldId.toString());
     return orcType;
   }
 
@@ -382,6 +382,12 @@ private static Optional<Integer> icebergID(TypeDescription orcType) {
         .map(Integer::parseInt);
   }
 
+  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);
+  }
+
   private static boolean isRequired(TypeDescription orcType) {
     String isRequiredStr = orcType.getAttributeValue(ICEBERG_REQUIRED_ATTRIBUTE);
     if (isRequiredStr != null) {

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

@@ -42,11 +42,11 @@ class OrcIterable<T> extends CloseableGroup implements CloseableIterable<T> {
   private final InputFile file;
   private final Long start;
   private final Long length;
-  private final Function<TypeDescription, OrcValueReader<?>> readerFunction;
+  private final Function<TypeDescription, OrcRowReader<?>> readerFunction;
 
   OrcIterable(InputFile file, Configuration config, Schema schema,
               Long start, Long length,
-              Function<TypeDescription, OrcValueReader<?>> readerFunction) {
+              Function<TypeDescription, OrcRowReader<?>> readerFunction) {
     this.schema = schema;
     this.readerFunction = readerFunction;
     this.file = file;
@@ -91,9 +91,9 @@ private static class OrcIterator<T> implements Iterator<T> {
     private VectorizedRowBatch current;
 
     private final VectorizedRowBatchIterator batchIter;
-    private final OrcValueReader<T> reader;
+    private final OrcRowReader<T> reader;
 
-    OrcIterator(VectorizedRowBatchIterator batchIter, OrcValueReader<T> reader) {
+    OrcIterator(VectorizedRowBatchIterator batchIter, OrcRowReader<T> reader) {
       this.batchIter = batchIter;
       this.reader = reader;
       current = null;

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

@@ -0,0 +1,34 @@
+/*
+ * 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.orc;
+
+import org.apache.orc.storage.ql.exec.vector.VectorizedRowBatch;
+
+/**
+ * Used for implementing ORC row readers.
+ */
+public interface OrcRowReader<T> {
+
+  /**
+   * Reads a row.
+   */
+  T read(VectorizedRowBatch batch, int row);
+
+}

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

@@ -0,0 +1,89 @@
+/*
+ * 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.orc;
+
+import com.google.common.collect.Lists;
+import java.util.List;
+import org.apache.iceberg.types.Type;
+import org.apache.iceberg.types.Types;
+import org.apache.orc.TypeDescription;
+
+
+public abstract class OrcSchemaWithTypeVisitor<T> {
+  public static <T> T visit(
+      org.apache.iceberg.Schema iSchema, TypeDescription schema, OrcSchemaWithTypeVisitor<T> visitor) {
+    return visit(iSchema.asStruct(), schema, visitor);
+  }
+
+  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);
+
+      case UNION:
+        throw new UnsupportedOperationException("Cannot handle " + schema);
+
+      case LIST:
+        Types.ListType list = iType != null ? iType.asListType() : null;
+        return visitor.list(
+            list, schema,
+            visit(list.elementType(), schema.getChildren().get(0), visitor));
+
+      case MAP:
+        Types.MapType map = iType != null ? iType.asMapType() : null;
+        return visitor.map(
+            map, schema,
+            visit(map != null ? map.keyType() : null, schema.getChildren().get(0), visitor),
+            visit(map != null ? map.valueType() : null, schema.getChildren().get(1), visitor));
+
+      default:
+        return visitor.primitive(iType != null ? iType.asPrimitiveType() : null, schema);
+    }
+  }
+
+  private static <T> T visitRecord(
+      Types.StructType struct, TypeDescription record, OrcSchemaWithTypeVisitor<T> visitor) {
+    List<TypeDescription> fields = record.getChildren();
+    List<String> names = record.getFieldNames();
+    List<T> results = Lists.newArrayListWithExpectedSize(fields.size());
+    for (TypeDescription field : fields) {
+      int fieldId = ORCSchemaUtil.fieldId(field);
+      Types.NestedField iField = struct != null ? struct.field(fieldId) : null;
+      results.add(visit(iField != null ? iField.type() : null, field, visitor));
+    }
+    return visitor.record(struct, record, names, results);
+  }
+
+  public T record(Types.StructType iStruct, TypeDescription record, List<String> names, List<T> fields) {
+    return null;
+  }
+
+  public T list(Types.ListType iList, TypeDescription array, T element) {
+    return null;
+  }
+
+  public T map(Types.MapType iMap, TypeDescription map, T key, T value) {
+    return null;
+  }
+
+  public T primitive(Type.PrimitiveType iPrimitive, TypeDescription primitive) {
+    return null;
+  }
+}

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

@@ -19,16 +19,18 @@
 
 package org.apache.iceberg.orc;
 
-import org.apache.orc.storage.ql.exec.vector.VectorizedRowBatch;
+import org.apache.orc.storage.ql.exec.vector.ColumnVector;
 
-/**
- * Used for implementing ORC value readers.
- */
-public interface OrcValueReader<T> {
 
-  /**
-   * Reads a value in row.
-   */
-  T read(VectorizedRowBatch batch, int row);
+public interface OrcValueReader<T> {
+  default T read(ColumnVector vector, int row) {
+    int rowIndex = vector.isRepeating ? 0 : row;
+    if (!vector.noNulls && vector.isNull[rowIndex]) {
+      return null;
+    } else {
+      return nonNullRead(vector, rowIndex);
+    }
+  }
 
+  T nonNullRead(ColumnVector vector, int row);
 }