diff --git a/data/src/test/java/org/apache/iceberg/data/orc/TestGenericOrcReaderIdBinding.java b/data/src/test/java/org/apache/iceberg/data/orc/TestGenericOrcReaderIdBinding.java new file mode 100644 index 0000000000..1a6a668537 --- /dev/null +++ b/data/src/test/java/org/apache/iceberg/data/orc/TestGenericOrcReaderIdBinding.java @@ -0,0 +1,220 @@ +/* + * 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.util.List; +import java.util.Map; +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.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.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 writeAndRead( + String desc, Schema writeSchema, Schema readSchema, List records) throws IOException { + return writeAndRead(desc, writeSchema, readSchema, records, ImmutableMap.of()); + } + + private List writeAndRead( + String desc, + Schema writeSchema, + Schema readSchema, + List records, + Map idToConstant) + throws IOException { + File file = temp.newFile(desc + ".orc"); + Assert.assertTrue("Delete should succeed", file.delete()); + + try (FileAppender 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 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 testMetadataColumns() throws IOException { + Schema writeSchema = + new Schema( + required(1, "id", Types.LongType.get()), optional(2, "data", Types.StringType.get())); + + List 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 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")); + } +} diff --git a/orc/src/main/java/org/apache/iceberg/data/orc/GenericOrcReader.java b/orc/src/main/java/org/apache/iceberg/data/orc/GenericOrcReader.java index f4d816baf2..f394d64bf1 100644 --- a/orc/src/main/java/org/apache/iceberg/data/orc/GenericOrcReader.java +++ b/orc/src/main/java/org/apache/iceberg/data/orc/GenericOrcReader.java @@ -76,7 +76,7 @@ public OrcValueReader record( TypeDescription record, List names, List> 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(); diff --git a/orc/src/main/java/org/apache/iceberg/data/orc/GenericOrcReaders.java b/orc/src/main/java/org/apache/iceberg/data/orc/GenericOrcReaders.java index 18ce07ac8b..506c738015 100644 --- a/orc/src/main/java/org/apache/iceberg/data/orc/GenericOrcReaders.java +++ b/orc/src/main/java/org/apache/iceberg/data/orc/GenericOrcReaders.java @@ -39,6 +39,7 @@ import org.apache.iceberg.types.Types; import org.apache.iceberg.util.DateTimeUtil; import org.apache.iceberg.util.UUIDUtil; +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.DecimalColumnVector; @@ -51,11 +52,25 @@ public class GenericOrcReaders { private GenericOrcReaders() {} + /** + * @deprecated Use {@link #struct(TypeDescription, List, Types.StructType, Map)} instead. This + * method uses position-based binding which may cause field misalignment in MOR and lineage + * scenarios. + */ + @Deprecated public static OrcValueReader struct( List> readers, Types.StructType struct, Map idToConstant) { return new StructReader(readers, struct, idToConstant); } + public static OrcValueReader struct( + TypeDescription orcType, + List> readers, + Types.StructType struct, + Map idToConstant) { + return new StructReader(orcType, readers, struct, idToConstant); + } + public static OrcValueReader> array(OrcValueReader elementReader) { return new ListReader(elementReader); } @@ -204,12 +219,27 @@ public ByteBuffer nonNullRead(ColumnVector vector, int row) { private static class StructReader extends OrcValueReaders.StructReader { private final GenericRecord template; + /** + * @deprecated Use {@link #StructReader(TypeDescription, List, Types.StructType, Map)} instead. + * This constructor uses position-based binding which may cause field misalignment in MOR + * and lineage scenarios. + */ + @Deprecated protected StructReader( List> readers, Types.StructType structType, Map idToConstant) { super(readers, structType, idToConstant); - this.template = structType != null ? GenericRecord.create(structType) : null; + this.template = GenericRecord.create(structType); + } + + protected StructReader( + TypeDescription orcType, + List> readers, + Types.StructType structType, + Map idToConstant) { + super(orcType, readers, structType, idToConstant); + this.template = GenericRecord.create(structType); } @Override diff --git a/orc/src/main/java/org/apache/iceberg/orc/OrcValueReaders.java b/orc/src/main/java/org/apache/iceberg/orc/OrcValueReaders.java index 18fef29403..e13689eb20 100644 --- a/orc/src/main/java/org/apache/iceberg/orc/OrcValueReaders.java +++ b/orc/src/main/java/org/apache/iceberg/orc/OrcValueReaders.java @@ -22,7 +22,10 @@ import java.util.List; import java.util.Map; import org.apache.iceberg.MetadataColumns; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.apache.iceberg.relocated.com.google.common.collect.Maps; import org.apache.iceberg.types.Types; +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.DoubleColumnVector; @@ -134,12 +137,27 @@ public byte[] nonNullRead(ColumnVector vector, int row) { public abstract static class StructReader implements OrcValueReader { private final OrcValueReader[] readers; private final boolean[] isConstantOrMetadataField; - + // Maps each projected struct field position to the matching child index in the ORC schema. + // This allows fields to be read by Iceberg field ID when the projected struct order differs + // from the file schema. + private final int[] orcFieldIndex; + + /** + * @param readers readers for each field + * @param struct struct type + * @param idToConstant constant values by field id + * @deprecated Use {@link #StructReader(TypeDescription, List, Types.StructType, Map)} instead. + * This constructor uses position-based binding which may cause field misalignment in MOR + * scenarios. + */ + @Deprecated protected StructReader( List> readers, Types.StructType struct, Map idToConstant) { List fields = struct.fields(); this.readers = new OrcValueReader[fields.size()]; this.isConstantOrMetadataField = new boolean[fields.size()]; + this.orcFieldIndex = null; + for (int pos = 0, readerIndex = 0; pos < fields.size(); pos += 1) { Types.NestedField field = fields.get(pos); if (idToConstant.containsKey(field.fieldId())) { @@ -152,7 +170,6 @@ protected StructReader( this.isConstantOrMetadataField[pos] = true; this.readers[pos] = constants(false); } else if (MetadataColumns.isMetadataColumn(field.name())) { - // in case of any other metadata field, fill with nulls this.isConstantOrMetadataField[pos] = true; this.readers[pos] = constants(null); } else { @@ -161,6 +178,73 @@ protected StructReader( } } + protected StructReader( + TypeDescription orcType, + List> readers, + Types.StructType struct, + Map idToConstant) { + List fields = struct.fields(); + this.readers = new OrcValueReader[fields.size()]; + this.isConstantOrMetadataField = new boolean[fields.size()]; + this.orcFieldIndex = new int[fields.size()]; + + Map> readersById = readersByFieldId(orcType, readers); + Map fieldIdToOrcIndex = buildFieldIdToOrcIndex(orcType); + + for (int pos = 0; pos < fields.size(); pos += 1) { + Types.NestedField field = fields.get(pos); + OrcValueReader fileReader = readersById.get(field.fieldId()); + + if (idToConstant.containsKey(field.fieldId())) { + this.isConstantOrMetadataField[pos] = true; + this.readers[pos] = constants(idToConstant.get(field.fieldId())); + } else if (field.equals(MetadataColumns.ROW_POSITION)) { + this.isConstantOrMetadataField[pos] = true; + this.readers[pos] = new RowPositionReader(); + } else if (field.equals(MetadataColumns.IS_DELETED)) { + this.isConstantOrMetadataField[pos] = true; + this.readers[pos] = constants(false); + } else if (fileReader != null) { + this.isConstantOrMetadataField[pos] = false; + this.orcFieldIndex[pos] = fieldIdToOrcIndex.getOrDefault(field.fieldId(), -1); + this.readers[pos] = fileReader; + } else if (MetadataColumns.isMetadataColumn(field.name())) { + this.isConstantOrMetadataField[pos] = true; + this.readers[pos] = constants(null); + } else { + throw new IllegalArgumentException( + String.format("Missing ORC reader for field %s (%s)", field.name(), field.fieldId())); + } + } + } + + private Map buildFieldIdToOrcIndex(TypeDescription orcType) { + List children = orcType.getChildren(); + Map mapping = Maps.newHashMap(); + for (int i = 0; i < children.size(); i++) { + mapping.put(ORCSchemaUtil.fieldId(children.get(i)), i); + } + + return mapping; + } + + private Map> readersByFieldId( + TypeDescription orcType, List> readerList) { + List children = orcType.getChildren(); + Preconditions.checkState( + children.size() == readerList.size(), + "Invalid ORC reader binding: children=%s readers=%s", + children.size(), + readerList.size()); + + Map> readersById = Maps.newHashMap(); + for (int i = 0; i < children.size(); i += 1) { + readersById.put(ORCSchemaUtil.fieldId(children.get(i)), readerList.get(i)); + } + + return readersById; + } + protected abstract T create(); protected abstract void set(T struct, int pos, Object value); @@ -176,14 +260,17 @@ public T nonNullRead(ColumnVector vector, int row) { } private T readInternal(T struct, ColumnVector[] columnVectors, int row) { - for (int c = 0, vectorIndex = 0; c < readers.length; ++c) { + int vectorIndex = 0; + for (int c = 0; c < readers.length; ++c) { ColumnVector vector; if (isConstantOrMetadataField[c]) { vector = null; + } else if (orcFieldIndex != null) { + vector = columnVectors[orcFieldIndex[c]]; } else { - vector = columnVectors[vectorIndex]; - vectorIndex++; + vector = columnVectors[vectorIndex++]; } + set(struct, c, reader(c).read(vector, row)); } return struct; diff --git a/spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcReader.java b/spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcReader.java index 78db137054..3d53233538 100644 --- a/spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcReader.java +++ b/spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcReader.java @@ -77,7 +77,7 @@ public OrcValueReader record( TypeDescription record, List names, List> fields) { - return SparkOrcValueReaders.struct(fields, expected, idToConstant); + return SparkOrcValueReaders.struct(record, fields, expected, idToConstant); } @Override diff --git a/spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcValueReaders.java b/spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcValueReaders.java index 9e9b3e53bb..fabb0868f8 100644 --- a/spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcValueReaders.java +++ b/spark/v3.1/spark/src/main/java/org/apache/iceberg/spark/data/SparkOrcValueReaders.java @@ -26,6 +26,7 @@ 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; 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.DecimalColumnVector; @@ -63,11 +64,25 @@ public static OrcValueReader decimals(int precision, int scale) { } } + /** + * @deprecated Use {@link #struct(TypeDescription, List, Types.StructType, Map)} instead. This + * method uses position-based binding which may cause field misalignment in MOR and lineage + * scenarios. + */ + @Deprecated static OrcValueReader struct( List> readers, Types.StructType struct, Map idToConstant) { return new StructReader(readers, struct, idToConstant); } + static OrcValueReader struct( + TypeDescription orcType, + List> readers, + Types.StructType struct, + Map idToConstant) { + return new StructReader(orcType, readers, struct, idToConstant); + } + static OrcValueReader array(OrcValueReader elementReader) { return new ArrayReader(elementReader); } @@ -136,12 +151,27 @@ public void setBatchContext(long batchOffsetInFile) { static class StructReader extends OrcValueReaders.StructReader { private final int numFields; + /** + * @deprecated Use {@link #StructReader(TypeDescription, List, Types.StructType, Map)} instead. + * This constructor uses position-based binding which may cause field misalignment in MOR + * and lineage scenarios. + */ + @Deprecated protected StructReader( List> readers, Types.StructType struct, Map idToConstant) { super(readers, struct, idToConstant); this.numFields = struct.fields().size(); } + protected StructReader( + TypeDescription orcType, + List> readers, + Types.StructType struct, + Map idToConstant) { + super(orcType, readers, struct, idToConstant); + this.numFields = struct.fields().size(); + } + @Override protected InternalRow create() { return new GenericInternalRow(numFields);