Add Iceberg support for name-based mapping schema

apache · Dec 6, 2024 · dcedd33 · dcedd33
1 parent d24d838
commit dcedd33
Showing 1 changed file with 6 additions and 0 deletions.
diff --git a/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ScanTaskReader.java b/sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/ScanTaskReader.java
@@ -40,6 +40,8 @@
 import org.apache.iceberg.io.CloseableIterator;
 import org.apache.iceberg.io.FileIO;
 import org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.mapping.MappingUtil;
+import org.apache.iceberg.mapping.NameMapping;
 import org.apache.iceberg.orc.ORC;
 import org.apache.iceberg.parquet.Parquet;
 import org.checkerframework.checker.nullness.qual.NonNull;
@@ -88,6 +90,7 @@ public boolean advance() throws IOException {
     // which are not null-safe.
     @SuppressWarnings("nullness")
     org.apache.iceberg.@NonNull Schema project = this.project;
+    NameMapping nameMapping = MappingUtil.create(project);
 
     do {
       // If our current iterator is working... do that.
@@ -123,6 +126,7 @@ public boolean advance() throws IOException {
                   .project(project)
                   .createReaderFunc(fileSchema -> GenericOrcReader.buildReader(project, fileSchema))
                   .filter(fileTask.residual())
+                  .withNameMapping(nameMapping)
                   .build();
           break;
         case PARQUET:
@@ -134,6 +138,7 @@ public boolean advance() throws IOException {
                   .createReaderFunc(
                       fileSchema -> GenericParquetReaders.buildReader(project, fileSchema))
                   .filter(fileTask.residual())
+                  .withNameMapping(nameMapping)
                   .build();
           break;
         case AVRO:
@@ -143,6 +148,7 @@ public boolean advance() throws IOException {
                   .split(fileTask.start(), fileTask.length())
                   .project(project)
                   .createReaderFunc(DataReader::create)
+                  .withNameMapping(nameMapping)
                   .build();
           break;
         default: