Flink: Add Orc value reader, writer implementations

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

@@ -98,7 +98,7 @@ public OrcValueWriter<?> primitive(Type.PrimitiveType iPrimitive, TypeDescriptio
         case UUID:
           return GenericOrcWriters.uuids();
         case FIXED:
-          return GenericOrcWriters.fixed();
+          return GenericOrcWriters.byteArrays();
         case BINARY:
           return GenericOrcWriters.byteBuffers();
         case DECIMAL:

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

@@ -56,6 +56,14 @@ public static OrcValueWriter<Boolean> booleans() {
     return BooleanWriter.INSTANCE;
   }
 
+  public static OrcValueWriter<Byte> bytes() {
+    return ByteWriter.INSTANCE;
+  }
+
+  public static OrcValueWriter<Short> shorts() {
+    return ShortWriter.INSTANCE;
+  }
+
   public static OrcValueWriter<Integer> ints() {
     return IntWriter.INSTANCE;
   }
@@ -88,8 +96,8 @@ public static OrcValueWriter<UUID> uuids() {
     return UUIDWriter.INSTANCE;
   }
 
-  public static OrcValueWriter<byte[]> fixed() {
-    return FixedWriter.INSTANCE;
+  public static OrcValueWriter<byte[]> byteArrays() {
+    return ByteArrayWriter.INSTANCE;
   }
 
   public static OrcValueWriter<LocalDate> dates() {
@@ -136,6 +144,34 @@ public void nonNullWrite(int rowId, Boolean data, ColumnVector output) {
     }
   }
 
+  private static class ByteWriter implements OrcValueWriter<Byte> {
+    private static final OrcValueWriter<Byte> INSTANCE = new ByteWriter();
+
+    @Override
+    public Class<Byte> getJavaClass() {
+      return Byte.class;
+    }
+
+    @Override
+    public void nonNullWrite(int rowId, Byte data, ColumnVector output) {
+      ((LongColumnVector) output).vector[rowId] = data;
+    }
+  }
+
+  private static class ShortWriter implements OrcValueWriter<Short> {
+    private static final OrcValueWriter<Short> INSTANCE = new ShortWriter();
+
+    @Override
+    public Class<Short> getJavaClass() {
+      return Short.class;
+    }
+
+    @Override
+    public void nonNullWrite(int rowId, Short data, ColumnVector output) {
+      ((LongColumnVector) output).vector[rowId] = data;
+    }
+  }
+
   private static class IntWriter implements OrcValueWriter<Integer> {
     private static final OrcValueWriter<Integer> INSTANCE = new IntWriter();
 
@@ -252,8 +288,8 @@ public void nonNullWrite(int rowId, UUID data, ColumnVector output) {
     }
   }
 
-  private static class FixedWriter implements OrcValueWriter<byte[]> {
-    private static final OrcValueWriter<byte[]> INSTANCE = new FixedWriter();
+  private static class ByteArrayWriter implements OrcValueWriter<byte[]> {
+    private static final OrcValueWriter<byte[]> INSTANCE = new ByteArrayWriter();
 
     @Override
     public Class<byte[]> getJavaClass() {

flink/src/main/java/org/apache/iceberg/flink/RowDataTaskWriterFactory.java

@@ -32,6 +32,7 @@
 import org.apache.iceberg.avro.Avro;
 import org.apache.iceberg.encryption.EncryptionManager;
 import org.apache.iceberg.flink.data.FlinkAvroWriter;
+import org.apache.iceberg.flink.data.FlinkOrcWriter;
 import org.apache.iceberg.io.FileAppender;
 import org.apache.iceberg.io.FileAppenderFactory;
 import org.apache.iceberg.io.FileIO;
@@ -40,6 +41,7 @@
 import org.apache.iceberg.io.OutputFileFactory;
 import org.apache.iceberg.io.TaskWriter;
 import org.apache.iceberg.io.UnpartitionedWriter;
+import org.apache.iceberg.orc.ORC;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 
 class RowDataTaskWriterFactory implements TaskWriterFactory<RowData> {
@@ -137,8 +139,16 @@ public FileAppender<RowData> newAppender(OutputFile outputFile, FileFormat forma
                 .schema(schema)
                 .overwrite()
                 .build();
-          case PARQUET:
+
           case ORC:
+            return ORC.write(outputFile)
+                .createWriterFunc((iSchema, typDesc) -> FlinkOrcWriter.buildWriter(flinkSchema, iSchema))
+                .setAll(props)
+                .schema(schema)
+                .overwrite()
+                .build();
+
+          case PARQUET:
           default:
             throw new UnsupportedOperationException("Cannot write unknown file format: " + format);
         }

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

@@ -0,0 +1,129 @@
+/*
+ * 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 java.util.Map;
+import org.apache.flink.table.data.ArrayData;
+import org.apache.flink.table.data.MapData;
+import org.apache.flink.table.data.RowData;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.orc.OrcRowReader;
+import org.apache.iceberg.orc.OrcSchemaWithTypeVisitor;
+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.types.Type;
+import org.apache.iceberg.types.Types;
+import org.apache.orc.TypeDescription;
+import org.apache.orc.storage.ql.exec.vector.StructColumnVector;
+import org.apache.orc.storage.ql.exec.vector.VectorizedRowBatch;
+
+public class FlinkOrcReader implements OrcRowReader<RowData> {
+  private final OrcValueReader<?> reader;
+
+  private FlinkOrcReader(Schema iSchema, TypeDescription readSchema) {
+    this(iSchema, readSchema, ImmutableMap.of());
+  }
+
+  private FlinkOrcReader(Schema iSchema, TypeDescription readSchema, Map<Integer, ?> idToConstant) {
+    this.reader = OrcSchemaWithTypeVisitor.visit(iSchema, readSchema, new ReadBuilder(idToConstant));
+  }
+
+  public static OrcRowReader<RowData> buildReader(Schema schema, TypeDescription readSchema) {
+    return new FlinkOrcReader(schema, readSchema);
+  }
+
+  @Override
+  public RowData read(VectorizedRowBatch batch, int row) {
+    return (RowData) reader.read(new StructColumnVector(batch.size, batch.cols), row);
+  }
+
+  @Override
+  public void setBatchContext(long batchOffsetInFile) {
+    reader.setBatchContext(batchOffsetInFile);
+  }
+
+  private static class ReadBuilder extends OrcSchemaWithTypeVisitor<OrcValueReader<?>> {
+    private final Map<Integer, ?> idToConstant;
+
+    private ReadBuilder(Map<Integer, ?> idToConstant) {
+      this.idToConstant = idToConstant;
+    }
+
+    @Override
+    public OrcValueReader<RowData> record(Types.StructType iStruct, TypeDescription record, List<String> names,
+                                          List<OrcValueReader<?>> fields) {
+      return FlinkOrcReaders.struct(fields, iStruct, idToConstant);
+    }
+
+    @Override
+    public OrcValueReader<ArrayData> list(Types.ListType iList, TypeDescription array,
+                                          OrcValueReader<?> elementReader) {
+      return FlinkOrcReaders.array(elementReader);
+    }
+
+    @Override
+    public OrcValueReader<MapData> map(Types.MapType iMap, TypeDescription map,
+                                       OrcValueReader<?> keyReader,
+                                       OrcValueReader<?> valueReader) {
+      return FlinkOrcReaders.map(keyReader, valueReader);
+    }
+
+    @Override
+    public OrcValueReader<?> primitive(Type.PrimitiveType iPrimitive, TypeDescription primitive) {
+      switch (iPrimitive.typeId()) {
+        case BOOLEAN:
+          return OrcValueReaders.booleans();
+        case INTEGER:
+          return OrcValueReaders.ints();
+        case LONG:
+          return OrcValueReaders.longs();
+        case FLOAT:
+          return OrcValueReaders.floats();
+        case DOUBLE:
+          return OrcValueReaders.doubles();
+        case DATE:
+          return FlinkOrcReaders.dates();
+        case TIME:
+          return FlinkOrcReaders.times();
+        case TIMESTAMP:
+          Types.TimestampType timestampType = (Types.TimestampType) iPrimitive;
+          if (timestampType.shouldAdjustToUTC()) {
+            return FlinkOrcReaders.timestampTzs();
+          } else {
+            return FlinkOrcReaders.timestamps();
+          }
+        case STRING:
+          return FlinkOrcReaders.strings();
+        case UUID:
+        case FIXED:
+        case BINARY:
+          return OrcValueReaders.bytes();
+        case DECIMAL:
+          Types.DecimalType decimalType = (Types.DecimalType) iPrimitive;
+          return FlinkOrcReaders.decimals(decimalType.precision(), decimalType.scale());
+        default:
+          throw new IllegalArgumentException(String.format("Invalid iceberg type %s corresponding to ORC type %s",
+              iPrimitive, primitive));
+      }
+    }
+  }
+}