Avro: Reading files using DataFileStream with ROW LINEAGE if the column isn't projected (#15508)

Author: ayushtkn

core/src/main/java/org/apache/iceberg/avro/ValueReaders.java

@@ -254,7 +254,10 @@ private static Pair<Integer, ValueReader<?>> fileFieldReader(
       Integer projectionPos,
       ValueReader<?> fieldReader,
       Map<Integer, ?> idToConstant) {
-    if (Objects.equals(fieldId, MetadataColumns.ROW_ID.fieldId())) {
+    if (projectionPos == null) {
+      // field is in the file but not projected; keep the reader only for skipping
+      return Pair.of(null, fieldReader);
+    } else if (Objects.equals(fieldId, MetadataColumns.ROW_ID.fieldId())) {
       Long firstRowId = (Long) idToConstant.get(fieldId);
       return Pair.of(projectionPos, ValueReaders.rowIds(firstRowId, fieldReader));
     } else if (Objects.equals(fieldId, MetadataColumns.LAST_UPDATED_SEQUENCE_NUMBER.fieldId())) {
@@ -273,7 +276,7 @@ private static Pair<Integer, ValueReader<?>> fieldReader(
       ValueReader<?> fieldReader,
       Map<Integer, ?> idToConstant) {
     Object constant = idToConstant.get(fieldId);
-    if (projectionPos != null && constant != null) {
+    if (constant != null) {
       return Pair.of(projectionPos, ValueReaders.replaceWithConstant(fieldReader, constant));
     }
 

core/src/test/java/org/apache/iceberg/data/avro/TestPlannedDataReader.java

@@ -23,6 +23,7 @@
 
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
+import java.io.File;
 import java.io.IOException;
 import java.time.LocalDateTime;
 import java.time.OffsetDateTime;
@@ -33,14 +34,18 @@
 import org.apache.avro.LogicalTypes;
 import org.apache.avro.Schema;
 import org.apache.avro.SchemaBuilder;
+import org.apache.avro.file.DataFileWriter;
 import org.apache.avro.generic.GenericData;
 import org.apache.avro.generic.GenericDatumWriter;
 import org.apache.avro.generic.GenericRecord;
 import org.apache.avro.io.BinaryDecoder;
 import org.apache.avro.io.BinaryEncoder;
 import org.apache.avro.io.DecoderFactory;
 import org.apache.avro.io.EncoderFactory;
+import org.apache.iceberg.Files;
 import org.apache.iceberg.MetadataColumns;
+import org.apache.iceberg.avro.Avro;
+import org.apache.iceberg.avro.AvroIterable;
 import org.apache.iceberg.data.Record;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
 import org.apache.iceberg.relocated.com.google.common.collect.Iterables;
@@ -305,6 +310,65 @@ public void testMixedRowLineageValues() throws IOException {
         .isEqualTo(10L);
   }
 
+  @Test
+  public void testLineageColumnsNotProjected() throws Exception {
+
+    org.apache.iceberg.Schema icebergSchema =
+        new org.apache.iceberg.Schema(
+            Types.NestedField.required(1, "data", Types.StringType.get()));
+
+    Schema fileSchema =
+        SchemaBuilder.record("test")
+            .fields()
+            .name("data")
+            .type()
+            .stringType()
+            .noDefault()
+            .name(MetadataColumns.ROW_ID.name())
+            .type()
+            .optional()
+            .longType()
+            .name(MetadataColumns.LAST_UPDATED_SEQUENCE_NUMBER.name())
+            .type()
+            .optional()
+            .longType()
+            .endRecord();
+
+    fileSchema.getField("data").addProp("field-id", 1);
+    fileSchema
+        .getField(MetadataColumns.ROW_ID.name())
+        .addProp("field-id", MetadataColumns.ROW_ID.fieldId());
+    fileSchema
+        .getField(MetadataColumns.LAST_UPDATED_SEQUENCE_NUMBER.name())
+        .addProp("field-id", MetadataColumns.LAST_UPDATED_SEQUENCE_NUMBER.fieldId());
+
+    File file = File.createTempFile("test", ".avro");
+
+    try (DataFileWriter<GenericRecord> writer =
+        new DataFileWriter<>(new GenericDatumWriter<>(fileSchema))) {
+
+      writer.create(fileSchema, file);
+
+      GenericRecord rec = new GenericData.Record(fileSchema);
+      rec.put("data", "a");
+      rec.put(MetadataColumns.ROW_ID.name(), 10L);
+      rec.put(MetadataColumns.LAST_UPDATED_SEQUENCE_NUMBER.name(), 5L);
+
+      writer.append(rec);
+    }
+
+    try (AvroIterable<Record> reader =
+        Avro.read(Files.localInput(file))
+            .createResolvingReader(schema -> PlannedDataReader.create(icebergSchema))
+            .project(icebergSchema)
+            .build()) {
+
+      List<Record> rows = Lists.newArrayList(reader);
+      assertThat(rows).hasSize(1);
+      assertThat(rows.get(0).getField("data")).isEqualTo("a");
+    }
+  }
+
   private Record readRecord(
       PlannedDataReader<Record> reader, Schema avroSchema, GenericRecord avroRecord)
       throws IOException {