ORC: bind struct fields by Iceberg field id in generic and Spark 3.5 readers

data/src/test/java/org/apache/iceberg/data/orc/TestGenericOrcReaderIdBinding.java

@@ -0,0 +1,334 @@
+/*
+ * 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.data.orc;
+
+import static org.apache.iceberg.types.Types.NestedField.optional;
+import static org.apache.iceberg.types.Types.NestedField.required;
+
+import java.io.File;
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+import java.util.Map;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.iceberg.Files;
+import org.apache.iceberg.MetadataColumns;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.data.GenericRecord;
+import org.apache.iceberg.data.Record;
+import org.apache.iceberg.io.CloseableIterable;
+import org.apache.iceberg.io.FileAppender;
+import org.apache.iceberg.orc.ORC;
+import org.apache.iceberg.orc.ORCSchemaUtil;
+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.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.types.TypeUtil;
+import org.apache.iceberg.types.Types;
+import org.apache.orc.OrcFile;
+import org.apache.orc.Reader;
+import org.apache.orc.TypeDescription;
+import org.apache.orc.storage.ql.exec.vector.BytesColumnVector;
+import org.apache.orc.storage.ql.exec.vector.ColumnVector;
+import org.apache.orc.storage.ql.exec.vector.LongColumnVector;
+import org.apache.orc.storage.ql.exec.vector.StructColumnVector;
+import org.apache.orc.storage.ql.exec.vector.VectorizedRowBatch;
+import org.junit.Assert;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+/**
+ * Tests specifically targeting the id-based field binding path of {@link GenericOrcReaders}. These
+ * exercise branches of the id-binding {@code StructReader} constructor that the shared projection
+ * tests do not, and assert that id-binding produces the same results positional binding would.
+ */
+public class TestGenericOrcReaderIdBinding {
+
+  @Rule public TemporaryFolder temp = new TemporaryFolder();
+
+  private List<Record> writeAndRead(
+      String desc, Schema writeSchema, Schema readSchema, List<Record> records) throws IOException {
+    return writeAndRead(desc, writeSchema, readSchema, records, ImmutableMap.of());
+  }
+
+  private List<Record> writeAndRead(
+      String desc,
+      Schema writeSchema,
+      Schema readSchema,
+      List<Record> records,
+      Map<Integer, ?> idToConstant)
+      throws IOException {
+    File file = temp.newFile(desc + ".orc");
+    Assert.assertTrue("Delete should succeed", file.delete());
+
+    try (FileAppender<Record> appender =
+        ORC.write(Files.localOutput(file))
+            .schema(writeSchema)
+            .createWriterFunc(GenericOrcWriter::buildWriter)
+            .build()) {
+      appender.addAll(records);
+    }
+
+    // Project only fields that physically exist in the file (excluding metadata columns) so that
+    // buildOrcProjection succeeds, but hand the reader the full read schema so the id-binding
+    // constructor resolves metadata/constant fields.
+    Schema projection = TypeUtil.selectNot(readSchema, MetadataColumns.metadataFieldIds());
+    try (CloseableIterable<Record> reader =
+        ORC.read(Files.localInput(file))
+            .project(projection)
+            .createReaderFunc(
+                fileSchema -> GenericOrcReader.buildReader(readSchema, fileSchema, idToConstant))
+            .build()) {
+      return Lists.newArrayList(reader);
+    }
+  }
+
+  @Test
+  public void testTypePromotion() throws IOException {
+    Schema writeSchema =
+        new Schema(
+            required(1, "id", Types.IntegerType.get()),
+            optional(2, "f", Types.FloatType.get()),
+            required(3, "dec", Types.DecimalType.of(9, 2)));
+
+    Record record = GenericRecord.create(writeSchema.asStruct());
+    record.setField("id", 42);
+    record.setField("f", 1.5f);
+    record.setField("dec", new BigDecimal("123.45"));
+
+    Schema promotedSchema =
+        new Schema(
+            required(1, "id", Types.LongType.get()),
+            optional(2, "f", Types.DoubleType.get()),
+            required(3, "dec", Types.DecimalType.of(11, 2)));
+
+    Record projected =
+        writeAndRead("type_promotion", writeSchema, promotedSchema, Lists.newArrayList(record))
+            .get(0);
+
+    Assert.assertEquals("int should be promoted to long", 42L, projected.getField("id"));
+    Assert.assertEquals(
+        "float should be promoted to double", 1.5d, (double) projected.getField("f"), 0.0d);
+    Assert.assertEquals(
+        "decimal precision should widen", new BigDecimal("123.45"), projected.getField("dec"));
+  }
+
+  @Test
+  public void testReorderedProjectionValues() throws IOException {
+    Schema writeSchema =
+        new Schema(
+            required(1, "a", Types.LongType.get()),
+            optional(2, "b", Types.StringType.get()),
+            required(3, "c", Types.IntegerType.get()));
+
+    Record record = GenericRecord.create(writeSchema.asStruct());
+    record.setField("a", 10L);
+    record.setField("b", "hello");
+    record.setField("c", 7);
+
+    // Read schema requests the fields in a different order than the file's physical column order.
+    Schema reordered =
+        new Schema(
+            required(3, "c", Types.IntegerType.get()),
+            required(1, "a", Types.LongType.get()),
+            optional(2, "b", Types.StringType.get()));
+
+    Record projected =
+        writeAndRead("reordered", writeSchema, reordered, Lists.newArrayList(record)).get(0);
+
+    Assert.assertEquals("c should bind by id", 7, projected.getField("c"));
+    Assert.assertEquals("a should bind by id", 10L, projected.getField("a"));
+    Assert.assertEquals("b should bind by id", "hello", projected.getField("b").toString());
+  }
+
+  @Test
+  public void testNonIdentityFieldIdMapping() {
+    Schema expectedSchema =
+        new Schema(
+            required(10, "first", Types.LongType.get()),
+            required(20, "second", Types.StringType.get()));
+    Schema orcSchema =
+        new Schema(
+            required(20, "second", Types.StringType.get()),
+            required(10, "first", Types.LongType.get()));
+    TypeDescription orcType = ORCSchemaUtil.convert(orcSchema);
+
+    List<OrcValueReader<?>> readers =
+        Lists.newArrayList(GenericOrcReaders.strings(), OrcValueReaders.longs());
+    OrcValueReader<Record> reader =
+        GenericOrcReaders.struct(orcType, readers, expectedSchema.asStruct(), ImmutableMap.of());
+
+    BytesColumnVector second = new BytesColumnVector(1);
+    byte[] secondValue = "value".getBytes(StandardCharsets.UTF_8);
+    second.setRef(0, secondValue, 0, secondValue.length);
+    LongColumnVector first = new LongColumnVector(1);
+    first.vector[0] = 34L;
+    StructColumnVector vector = new StructColumnVector(1, new ColumnVector[] {second, first});
+
+    Record actual = reader.read(vector, 0);
+    Assert.assertEquals("first should bind from ORC index 1", 34L, actual.get(0));
+    Assert.assertEquals("second should bind from ORC index 0", "value", actual.get(1));
+  }
+
+  @Test
+  public void testMetadataColumns() throws IOException {
+    Schema writeSchema =
+        new Schema(
+            required(1, "id", Types.LongType.get()), optional(2, "data", Types.StringType.get()));
+
+    List<Record> records = Lists.newArrayList();
+    for (long i = 0; i < 5; i++) {
+      Record record = GenericRecord.create(writeSchema.asStruct());
+      record.setField("id", i);
+      record.setField("data", "row" + i);
+      records.add(record);
+    }
+
+    Schema readSchema =
+        new Schema(
+            required(1, "id", Types.LongType.get()),
+            optional(2, "data", Types.StringType.get()),
+            MetadataColumns.ROW_POSITION,
+            MetadataColumns.IS_DELETED);
+
+    List<Record> projected = writeAndRead("metadata_columns", writeSchema, readSchema, records);
+
+    Assert.assertEquals(5, projected.size());
+    for (int i = 0; i < projected.size(); i++) {
+      Record record = projected.get(i);
+      Assert.assertEquals("id should read from file", (long) i, record.getField("id"));
+      Assert.assertEquals(
+          "data should read from file", "row" + i, record.getField("data").toString());
+      Assert.assertEquals(
+          "_pos should be the row position",
+          (long) i,
+          record.getField(MetadataColumns.ROW_POSITION.name()));
+      Assert.assertEquals(
+          "_deleted should be false", false, record.getField(MetadataColumns.IS_DELETED.name()));
+    }
+  }
+
+  @Test
+  public void testIdToConstant() throws IOException {
+    Schema writeSchema =
+        new Schema(
+            required(1, "id", Types.LongType.get()), optional(2, "data", Types.StringType.get()));
+
+    Record record = GenericRecord.create(writeSchema.asStruct());
+    record.setField("id", 1L);
+    record.setField("data", "value");
+
+    // Read schema adds an identity-partition-style constant field (id 3) not present in the file.
+    Schema readSchema =
+        new Schema(
+            required(1, "id", Types.LongType.get()),
+            optional(2, "data", Types.StringType.get()),
+            optional(3, "part", Types.StringType.get()));
+
+    Record projected =
+        writeAndRead(
+                "id_to_constant",
+                writeSchema,
+                readSchema,
+                Lists.newArrayList(record),
+                ImmutableMap.of(3, "constant-partition"))
+            .get(0);
+
+    Assert.assertEquals("id should read from file", 1L, projected.getField("id"));
+    Assert.assertEquals(
+        "data should read from file", "value", projected.getField("data").toString());
+    Assert.assertEquals(
+        "part should resolve from idToConstant", "constant-partition", projected.getField("part"));
+  }
+
+  @Test
+  public void testIdLessFileBindsThroughNameMapping() throws IOException {
+    // Files migrated from Hive carry no Iceberg field ids. They are resolved by name mapping, which
+    // stamps ids onto the projection before the reader binds, so id-binding must still produce the
+    // values positional binding produced. A field absent from such a file must read null.
+    File file = writeIdLessFile("id_less");
+
+    Schema readSchema =
+        new Schema(
+            required(1, "id", Types.LongType.get()),
+            optional(2, "data", Types.StringType.get()),
+            optional(3, "added_after_migration", Types.StringType.get()));
+
+    List<Record> projected;
+    try (CloseableIterable<Record> reader =
+        ORC.read(Files.localInput(file))
+            .project(readSchema)
+            .createReaderFunc(fileSchema -> GenericOrcReader.buildReader(readSchema, fileSchema))
+            .build()) {
+      projected = Lists.newArrayList(reader);
+    }
+
+    Assert.assertEquals(3, projected.size());
+    for (int i = 0; i < projected.size(); i++) {
+      Record record = projected.get(i);
+      Assert.assertEquals("id should bind by name mapping", (long) i, record.getField("id"));
+      Assert.assertEquals(
+          "data should bind by name mapping", "row" + i, record.getField("data").toString());
+      Assert.assertNull(
+          "field absent from an id-less file should read null",
+          record.getField("added_after_migration"));
+    }
+  }
+
+  /**
+   * Writes an ORC file directly, without the Iceberg writer, so it carries no field id attributes.
+   */
+  private File writeIdLessFile(String desc) throws IOException {
+    File file = temp.newFile(desc + ".orc");
+    Assert.assertTrue("Delete should succeed", file.delete());
+
+    TypeDescription writerSchema = TypeDescription.fromString("struct<id:bigint,data:string>");
+    try (org.apache.orc.Writer writer =
+        OrcFile.createWriter(
+            new Path(file.toString()),
+            OrcFile.writerOptions(new Configuration()).setSchema(writerSchema))) {
+      VectorizedRowBatch batch = writerSchema.createRowBatch();
+      LongColumnVector ids = (LongColumnVector) batch.cols[0];
+      BytesColumnVector data = (BytesColumnVector) batch.cols[1];
+      for (long i = 0; i < 3; i++) {
+        int row = batch.size++;
+        ids.vector[row] = i;
+        data.setVal(row, ("row" + i).getBytes(StandardCharsets.UTF_8));
+      }
+      writer.addRowBatch(batch);
+    }
+
+    // Without this the test could silently exercise the id-bearing path it is meant to avoid.
+    try (Reader orcReader =
+        OrcFile.createReader(
+            new Path(file.toString()), OrcFile.readerOptions(new Configuration()))) {
+      for (TypeDescription child : orcReader.getSchema().getChildren()) {
+        Assert.assertNull(
+            "migrated file should carry no iceberg field ids",
+            child.getAttributeValue("iceberg.id"));
+      }
+    }
+
+    return file;
+  }
+}

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

@@ -76,7 +76,7 @@ public OrcValueReader<?> record(
         TypeDescription record,
         List<String> names,
         List<OrcValueReader<?>> fields) {
-      return GenericOrcReaders.struct(fields, expected, idToConstant);
+      return GenericOrcReaders.struct(record, fields, expected, idToConstant);
     }
 
     @Override
@@ -96,6 +96,10 @@ public OrcValueReader<?> map(
 
     @Override
     public OrcValueReader<?> primitive(Type.PrimitiveType iPrimitive, TypeDescription primitive) {
+      if (iPrimitive == null) {
+        return null;
+      }
+
       switch (primitive.getCategory()) {
         case BOOLEAN:
           return OrcValueReaders.booleans();