Implement the parquet value reader & writer for apache flink

build.gradle

@@ -261,6 +261,8 @@ project(':iceberg-flink') {
     testCompile("org.apache.flink:flink-test-utils_2.12") {
       exclude group: "org.apache.curator", module: 'curator-test'
     }
+
+    testCompile project(path: ':iceberg-api', configuration: 'testArtifacts')
   }
 }
 

data/src/main/java/org/apache/iceberg/data/parquet/GenericParquetReaders.java

@@ -30,7 +30,6 @@
 import java.util.Map;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.data.GenericRecord;
-import org.apache.iceberg.data.Record;
 import org.apache.iceberg.parquet.ParquetSchemaUtil;
 import org.apache.iceberg.parquet.ParquetValueReader;
 import org.apache.iceberg.parquet.ParquetValueReaders;
@@ -67,26 +66,34 @@ private GenericParquetReaders() {
 
   public static ParquetValueReader<GenericRecord> buildReader(Schema expectedSchema,
                                                               MessageType fileSchema) {
-    return buildReader(expectedSchema, fileSchema, ImmutableMap.of());
+    return buildReader(expectedSchema, fileSchema, ImmutableMap.of(), RecordReader::new);
   }
-  @SuppressWarnings("unchecked")
+
   public static ParquetValueReader<GenericRecord> buildReader(Schema expectedSchema,
                                                               MessageType fileSchema,
                                                               Map<Integer, ?> idToConstant) {
+    return buildReader(expectedSchema, fileSchema, idToConstant, RecordReader::new);
+  }
+
+  @SuppressWarnings("unchecked")
+  public static <T> ParquetValueReader<T> buildReader(Schema expectedSchema,
+                                                      MessageType fileSchema,
+                                                      Map<Integer, ?> idToConstant,
+                                                      StructReaderFactory<T> structReaderFactory) {
     if (ParquetSchemaUtil.hasIds(fileSchema)) {
-      return (ParquetValueReader<GenericRecord>)
+      return (ParquetValueReader<T>)
           TypeWithSchemaVisitor.visit(expectedSchema.asStruct(), fileSchema,
-              new ReadBuilder(fileSchema, idToConstant));
+              new ReadBuilder(fileSchema, idToConstant, structReaderFactory));
     } else {
-      return (ParquetValueReader<GenericRecord>)
+      return (ParquetValueReader<T>)
           TypeWithSchemaVisitor.visit(expectedSchema.asStruct(), fileSchema,
-              new FallbackReadBuilder(fileSchema, idToConstant));
+              new FallbackReadBuilder(fileSchema, idToConstant, structReaderFactory));
     }
   }
 
   private static class FallbackReadBuilder extends ReadBuilder {
-    FallbackReadBuilder(MessageType type, Map<Integer, ?> idToConstant) {
-      super(type, idToConstant);
+    FallbackReadBuilder(MessageType type, Map<Integer, ?> idToConstant, StructReaderFactory structReaderFactory) {
+      super(type, idToConstant, structReaderFactory);
     }
 
     @Override
@@ -111,17 +118,19 @@ public ParquetValueReader<?> struct(StructType expected, GroupType struct,
         types.add(fieldType);
       }
 
-      return new RecordReader(types, newFields, expected);
+      return structReaderFactory().create(types, newFields, expected);
     }
   }
 
   private static class ReadBuilder extends TypeWithSchemaVisitor<ParquetValueReader<?>> {
     private final MessageType type;
     private final Map<Integer, ?> idToConstant;
+    private final StructReaderFactory structReaderFactory;
 
-    ReadBuilder(MessageType type, Map<Integer, ?> idToConstant) {
+    ReadBuilder(MessageType type, Map<Integer, ?> idToConstant, StructReaderFactory structReaderFactory) {
       this.type = type;
       this.idToConstant = idToConstant;
+      this.structReaderFactory = structReaderFactory;
     }
 
     @Override
@@ -168,7 +177,7 @@ public ParquetValueReader<?> struct(StructType expected, GroupType struct,
         }
       }
 
-      return new RecordReader(types, reorderedFields, expected);
+      return structReaderFactory.create(types, reorderedFields, expected);
     }
 
     @Override
@@ -298,6 +307,10 @@ public ParquetValueReader<?> primitive(org.apache.iceberg.types.Type.PrimitiveTy
     MessageType type() {
       return type;
     }
+
+    StructReaderFactory structReaderFactory() {
+      return structReaderFactory;
+    }
   }
 
   private static final OffsetDateTime EPOCH = Instant.ofEpochSecond(0).atOffset(ZoneOffset.UTC);
@@ -396,7 +409,12 @@ public byte[] read(byte[] reuse) {
     }
   }
 
-  static class RecordReader extends StructReader<Record, Record> {
+  public interface StructReaderFactory<T> {
+
+    StructReader<T, T> create(List<Type> types, List<ParquetValueReader<?>> readers, StructType struct);
+  }
+
+  static class RecordReader extends StructReader<GenericRecord, GenericRecord> {
     private final StructType structType;
 
     RecordReader(List<Type> types,
@@ -407,7 +425,7 @@ static class RecordReader extends StructReader<Record, Record> {
     }
 
     @Override
-    protected Record newStructData(Record reuse) {
+    protected GenericRecord newStructData(GenericRecord reuse) {
       if (reuse != null) {
         return reuse;
       } else {
@@ -417,17 +435,17 @@ protected Record newStructData(Record reuse) {
 
     @Override
     @SuppressWarnings("unchecked")
-    protected Object getField(Record intermediate, int pos) {
+    protected Object getField(GenericRecord intermediate, int pos) {
       return intermediate.get(pos);
     }
 
     @Override
-    protected Record buildStruct(Record struct) {
+    protected GenericRecord buildStruct(GenericRecord struct) {
       return struct;
     }
 
     @Override
-    protected void set(Record struct, int pos, Object value) {
+    protected void set(GenericRecord struct, int pos, Object value) {
       struct.set(pos, value);
     }
   }

data/src/main/java/org/apache/iceberg/data/parquet/GenericParquetWriter.java

@@ -51,14 +51,21 @@ private GenericParquetWriter() {
 
   @SuppressWarnings("unchecked")
   public static <T> ParquetValueWriter<T> buildWriter(MessageType type) {
-    return (ParquetValueWriter<T>) ParquetTypeVisitor.visit(type, new WriteBuilder(type));
+    return buildWriter(type, RecordWriter::new);
+  }
+
+  @SuppressWarnings("unchecked")
+  public static <T> ParquetValueWriter<T> buildWriter(MessageType type, StructWriterFactory structWriterFactory) {
+    return (ParquetValueWriter<T>) ParquetTypeVisitor.visit(type, new WriteBuilder(type, structWriterFactory));
   }
 
   private static class WriteBuilder extends ParquetTypeVisitor<ParquetValueWriter<?>> {
     private final MessageType type;
+    private final StructWriterFactory structWriterFactory;
 
-    WriteBuilder(MessageType type) {
+    WriteBuilder(MessageType type, StructWriterFactory structWriterFactory) {
       this.type = type;
+      this.structWriterFactory = structWriterFactory;
     }
 
     @Override
@@ -78,7 +85,7 @@ public ParquetValueWriter<?> struct(GroupType struct,
         writers.add(ParquetValueWriters.option(fieldType, fieldD, fieldWriters.get(i)));
       }
 
-      return new RecordWriter(writers);
+      return structWriterFactory.create(writers);
     }
 
     @Override
@@ -284,6 +291,11 @@ public void write(int repetitionLevel, byte[] value) {
     }
   }
 
+  public interface StructWriterFactory {
+
+    StructWriter<?> create(List<ParquetValueWriter<?>> writers);
+  }
+
   private static class RecordWriter extends StructWriter<Record> {
     private RecordWriter(List<ParquetValueWriter<?>> writers) {
       super(writers);

flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetReaders.java

@@ -0,0 +1,75 @@
+/*
+ * 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.flink.data;
+
+import java.util.List;
+import org.apache.flink.types.Row;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.data.parquet.GenericParquetReaders;
+import org.apache.iceberg.parquet.ParquetValueReader;
+import org.apache.iceberg.parquet.ParquetValueReaders;
+import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap;
+import org.apache.iceberg.types.Types;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.Type;
+
+public class FlinkParquetReaders {
+  private FlinkParquetReaders() {
+
+  }
+
+  public static ParquetValueReader<Row> buildReader(Schema expectedSchema,
+                                                    MessageType fileSchema) {
+    return GenericParquetReaders.buildReader(expectedSchema, fileSchema, ImmutableMap.of(), RowReader::new);
+  }
+
+  static class RowReader extends ParquetValueReaders.StructReader<Row, Row> {
+    private final Types.StructType structType;
+
+    RowReader(List<Type> types, List<ParquetValueReader<?>> readers, Types.StructType struct) {
+      super(types, readers);
+      this.structType = struct;
+    }
+
+    @Override
+    protected Row newStructData(Row reuse) {
+      if (reuse != null) {
+        return reuse;
+      } else {
+        return new Row(structType.fields().size());
+      }
+    }
+
+    @Override
+    protected Object getField(Row row, int pos) {
+      return row.getField(pos);
+    }
+
+    @Override
+    protected Row buildStruct(Row row) {
+      return row;
+    }
+
+    @Override
+    protected void set(Row row, int pos, Object value) {
+      row.setField(pos, value);
+    }
+  }
+}

flink/src/main/java/org/apache/iceberg/flink/data/FlinkParquetWriters.java

@@ -0,0 +1,49 @@
+/*
+ * 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.flink.data;
+
+import java.util.List;
+import org.apache.flink.types.Row;
+import org.apache.iceberg.data.parquet.GenericParquetWriter;
+import org.apache.iceberg.parquet.ParquetValueWriter;
+import org.apache.iceberg.parquet.ParquetValueWriters;
+import org.apache.parquet.schema.MessageType;
+
+public class FlinkParquetWriters {
+  private FlinkParquetWriters() {
+  }
+
+  @SuppressWarnings("unchecked")
+  public static <T> ParquetValueWriter<T> buildWriter(MessageType type) {
+    return GenericParquetWriter.buildWriter(type, RowWriter::new);
+  }
+
+  private static class RowWriter extends ParquetValueWriters.StructWriter<Row> {
+
+    private RowWriter(List<ParquetValueWriter<?>> writers) {
+      super(writers);
+    }
+
+    @Override
+    protected Object get(Row row, int index) {
+      return row.getField(index);
+    }
+  }
+}