Flink: Upgrade to latest supported Flink versions 2.2 and 2.3 without dropping 2.1

.github/workflows/cve-scan.yml

@@ -154,18 +154,24 @@ jobs:
             :iceberg-flink:iceberg-flink-runtime-1.20:shadowJar
           scan-path: flink/v1.20/flink-runtime/build/libs
           unpack: false
-        - distribution: flink-runtime-2.0
-          build-task: >-
-            -DflinkVersions=2.0
-            :iceberg-flink:iceberg-flink-runtime-2.0:shadowJar
-          scan-path: flink/v2.0/flink-runtime/build/libs
-          unpack: false
         - distribution: flink-runtime-2.1
           build-task: >-
             -DflinkVersions=2.1
             :iceberg-flink:iceberg-flink-runtime-2.1:shadowJar
           scan-path: flink/v2.1/flink-runtime/build/libs
           unpack: false
+        - distribution: flink-runtime-2.2
+          build-task: >-
+            -DflinkVersions=2.2
+            :iceberg-flink:iceberg-flink-runtime-2.2:shadowJar
+          scan-path: flink/v2.2/flink-runtime/build/libs
+          unpack: false
+        - distribution: flink-runtime-2.3
+          build-task: >-
+            -DflinkVersions=2.3
+            :iceberg-flink:iceberg-flink-runtime-2.3:shadowJar
+          scan-path: flink/v2.3/flink-runtime/build/libs
+          unpack: false
         - distribution: open-api-test-fixtures-runtime
           build-task: :iceberg-open-api:shadowJar
           scan-path: open-api/build/libs

.github/workflows/flink-ci.yml

@@ -87,7 +87,7 @@ jobs:
           # Pull requests run the baseline JDK only.
           - event_name: pull_request
             jvm: 21
-        flink: ['1.20', '2.0', '2.1']
+        flink: ['1.20', '2.1', '2.2', '2.3']
     env:
       SPARK_LOCAL_IP: localhost
     steps:

dev/stage-binaries.sh

@@ -19,7 +19,7 @@
 #
 
 SCALA_VERSION=2.12
-FLINK_VERSIONS=1.20,2.0,2.1
+FLINK_VERSIONS=1.20,2.1,2.2,2.3
 SPARK_VERSIONS=3.5,4.0,4.1
 KAFKA_VERSIONS=3
 

flink/build.gradle

@@ -23,10 +23,14 @@ if (flinkVersions.contains("1.20")) {
   apply from: file("$projectDir/v1.20/build.gradle")
 }
 
-if (flinkVersions.contains("2.0")) {
-  apply from: file("$projectDir/v2.0/build.gradle")
-}
-
 if (flinkVersions.contains("2.1")) {
   apply from: file("$projectDir/v2.1/build.gradle")
+}
+
+if (flinkVersions.contains("2.2")) {
+  apply from: file("$projectDir/v2.2/build.gradle")
+}
+
+if (flinkVersions.contains("2.3")) {
+  apply from: file("$projectDir/v2.3/build.gradle")
 }

flink/v2.0/build.gradle → flink/v2.2/build.gradle

@@ -17,7 +17,7 @@
  * under the License.
  */
 
-String flinkMajorVersion = '2.0'
+String flinkMajorVersion = '2.2'
 String scalaVersion = System.getProperty("scalaVersion") != null ? System.getProperty("scalaVersion") : System.getProperty("defaultScalaVersion")
 
 project(":iceberg-flink:iceberg-flink-${flinkMajorVersion}") {
@@ -32,16 +32,16 @@ project(":iceberg-flink:iceberg-flink-${flinkMajorVersion}") {
     implementation project(':iceberg-parquet')
     implementation project(':iceberg-hive-metastore')
 
-    compileOnly libs.flink20.avro
+    compileOnly libs.flink22.avro
     compileOnly libs.joda.time
     // dropwizard histogram metrics (optional in Flink)
-    compileOnly libs.flink20.metrics.dropwizard
-    compileOnly libs.flink20.streaming.java
-    compileOnly "${libs.flink20.streaming.java.get().module}:${libs.flink20.streaming.java.get().getVersion()}:tests"
-    compileOnly libs.flink20.table.api.java.bridge
-    compileOnly "org.apache.flink:flink-table-planner_${scalaVersion}:${libs.versions.flink20.get()}"
-    compileOnly libs.flink20.connector.base
-    compileOnly libs.flink20.connector.files
+    compileOnly libs.flink22.metrics.dropwizard
+    compileOnly libs.flink22.streaming.java
+    compileOnly "${libs.flink22.streaming.java.get().module}:${libs.flink22.streaming.java.get().getVersion()}:tests"
+    compileOnly libs.flink22.table.api.java.bridge
+    compileOnly "org.apache.flink:flink-table-planner_${scalaVersion}:${libs.versions.flink22.get()}"
+    compileOnly libs.flink22.connector.base
+    compileOnly libs.flink22.connector.files
 
     compileOnly libs.hadoop3.hdfs
     compileOnly libs.hadoop3.common
@@ -70,13 +70,13 @@ project(":iceberg-flink:iceberg-flink-${flinkMajorVersion}") {
     implementation libs.datasketches
     implementation libs.roaringbitmap
 
-    testImplementation libs.flink20.connector.test.utils
-    testImplementation libs.flink20.core
-    testImplementation libs.flink20.runtime
-    testImplementation(libs.flink20.test.utilsjunit) {
+    testImplementation libs.flink22.connector.test.utils
+    testImplementation libs.flink22.core
+    testImplementation libs.flink22.runtime
+    testImplementation(libs.flink22.test.utilsjunit) {
       exclude group: 'junit'
     }
-    testImplementation(libs.flink20.test.utils) {
+    testImplementation(libs.flink22.test.utils) {
       exclude group: "org.apache.curator", module: 'curator-test'
       exclude group: 'junit'
     }
@@ -180,13 +180,13 @@ project(":iceberg-flink:iceberg-flink-runtime-${flinkMajorVersion}") {
     integrationImplementation project(path: ":iceberg-flink:iceberg-flink-${flinkMajorVersion}", configuration: "testArtifacts")
     integrationImplementation project(path: ':iceberg-api', configuration: 'testArtifacts')
     integrationImplementation project(path: ':iceberg-hive-metastore', configuration: 'testArtifacts')
-    integrationImplementation(libs.flink20.test.utils) {
+    integrationImplementation(libs.flink22.test.utils) {
       exclude group: "org.apache.curator", module: 'curator-test'
       exclude group: 'junit'
     }
 
-    integrationImplementation libs.flink20.table.api.java.bridge
-    integrationImplementation "org.apache.flink:flink-table-planner_${scalaVersion}:${libs.versions.flink20.get()}"
+    integrationImplementation libs.flink22.table.api.java.bridge
+    integrationImplementation "org.apache.flink:flink-table-planner_${scalaVersion}:${libs.versions.flink22.get()}"
 
     integrationImplementation libs.hadoop3.common
     integrationImplementation libs.hadoop3.hdfs

flink/v2.0/flink-runtime/LICENSE → flink/v2.2/flink-runtime/LICENSE

@@ -528,6 +528,7 @@ This product includes code from Apache Flink.
 * RowData to Avro conversion logic in RowDataToAvroConverters.java
 * Avro schema conversion logic in AvroSchemaConverter.java
 * Joda optional dependency encapsulation in JodaConverter.java
+* Binary variant accessor utility methods in BinaryVariantAccessorUtils.java
 
 Copyright: 1999-2022 The Apache Software Foundation.
 Project URL: https://flink.apache.org/

flink/v2.2/flink/src/main/java/org/apache/flink/types/variant/BinaryVariantAccessorUtils.java

@@ -0,0 +1,86 @@
+/*
+ * 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.flink.types.variant;
+
+import java.util.List;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+
+/**
+ * These Accessor utils should be part of Flink Variant interface itself. Added in the PR, <a
+ * href="https://github.com/apache/flink/pull/27600">...</a>. This is a workaround until we upgrade
+ * Flink to a release version with above changes.
+ */
+public class BinaryVariantAccessorUtils extends BinaryVariantUtil {
+  /**
+   * Get the size of an array variant.
+   *
+   * @return Number of elements if this variant is an instance of BinaryVariant and an array
+   * @throws IllegalArgumentException If this variant is not an instance of BinaryVariant or not an
+   *     array.
+   */
+  public static int arraySize(Variant variant) {
+    Preconditions.checkArgument(
+        variant instanceof BinaryVariant,
+        "Invalid variant implementation instance:%s. Only BinaryVariant is supported.",
+        variant.getClass().getName());
+
+    Preconditions.checkArgument(
+        variant.getType() == Variant.Type.ARRAY,
+        "Invalid variant type:%s. Expecting ARRAY variant type.",
+        variant.getType());
+
+    BinaryVariant binaryArrayVariant = (BinaryVariant) variant;
+    return handleArray(
+        binaryArrayVariant.getValue(), 0, (size, offsetSize, offsetStart, dataStart) -> size);
+  }
+
+  /**
+   * Get the field names of an object variant only at top level. Doesn't include the nested fields.
+   *
+   * @return List of field names if this is an instance of BinaryVariant and is an object
+   * @throws IllegalArgumentException If this variant is not an instance of BinaryVariant or not an
+   *     object.
+   */
+  public static List<String> fieldNames(Variant variant) {
+    Preconditions.checkArgument(
+        variant instanceof BinaryVariant,
+        "Invalid variant implementation instance:%s. Only BinaryVariant is supported.",
+        variant.getClass().getName());
+
+    Preconditions.checkArgument(
+        variant.getType() == Variant.Type.OBJECT,
+        "Invalid variant type:%s. Expecting OBJECT variant type.",
+        variant.getType());
+
+    BinaryVariant binaryArrayVariant = (BinaryVariant) variant;
+    return BinaryVariantUtil.handleObject(
+        binaryArrayVariant.getValue(),
+        0,
+        (size, idSize, offsetSize, idStart, offsetStart, dataStart) -> {
+          List<String> fieldNames = Lists.newArrayList();
+          for (int i = 0; i < size; i++) {
+            int id = readUnsigned(binaryArrayVariant.getValue(), idStart + idSize * i, idSize);
+            String fieldName = getMetadataKey(binaryArrayVariant.getMetadata(), id);
+            fieldNames.add(fieldName);
+          }
+          return fieldNames;
+        });
+  }
+}

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkCreateTableOptions.java → flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/FlinkCreateTableOptions.java

@@ -18,11 +18,13 @@
  */
 package org.apache.iceberg.flink;
 
+import org.apache.flink.annotation.Internal;
 import org.apache.flink.configuration.ConfigOption;
 import org.apache.flink.configuration.ConfigOptions;
 import org.apache.iceberg.util.JsonUtil;
 
-class FlinkCreateTableOptions {
+@Internal
+public class FlinkCreateTableOptions {
   private final String catalogName;
   private final String catalogDb;
   private final String catalogTable;

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkTypeToType.java → flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/FlinkTypeToType.java

@@ -40,6 +40,7 @@
 import org.apache.flink.table.types.logical.TinyIntType;
 import org.apache.flink.table.types.logical.VarBinaryType;
 import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.flink.table.types.logical.VariantType;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.types.Type;
 import org.apache.iceberg.types.Types;
@@ -206,4 +207,9 @@ public Type visit(RowType rowType) {
 
     return Types.StructType.of(newFields);
   }
+
+  @Override
+  public Type visit(VariantType variantType) {
+    return Types.VariantType.get();
+  }
 }

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkWriteConf.java → flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/FlinkWriteConf.java

@@ -271,4 +271,22 @@ public Duration tableRefreshInterval() {
         .flinkConfig(FlinkWriteOptions.TABLE_REFRESH_INTERVAL)
         .parseOptional();
   }
+
+  public boolean parquetShredVariants() {
+    return confParser
+        .booleanConf()
+        .option(FlinkWriteOptions.SHRED_VARIANTS.key())
+        .tableProperty(TableProperties.PARQUET_SHRED_VARIANTS)
+        .defaultValue(TableProperties.PARQUET_SHRED_VARIANTS_DEFAULT)
+        .parse();
+  }
+
+  public int parquetVariantInferenceBufferSize() {
+    return confParser
+        .intConf()
+        .option(FlinkWriteOptions.VARIANT_INFERENCE_BUFFER_SIZE.key())
+        .tableProperty(TableProperties.PARQUET_VARIANT_BUFFER_SIZE)
+        .defaultValue(TableProperties.PARQUET_VARIANT_BUFFER_SIZE_DEFAULT)
+        .parse();
+  }
 }

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkWriteOptions.java → flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/FlinkWriteOptions.java

@@ -111,4 +111,10 @@ private FlinkWriteOptions() {}
   //  specify the uidSuffix to be used for the underlying IcebergSink
   public static final ConfigOption<String> UID_SUFFIX =
       ConfigOptions.key("uid-suffix").stringType().defaultValue("");
+
+  public static final ConfigOption<Boolean> SHRED_VARIANTS =
+      ConfigOptions.key("shred-variants").booleanType().defaultValue(false);
+
+  public static final ConfigOption<Integer> VARIANT_INFERENCE_BUFFER_SIZE =
+      ConfigOptions.key("variant-inference-buffer-size").intType().noDefaultValue();
 }

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/IcebergTableSink.java → flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/IcebergTableSink.java

@@ -21,6 +21,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
+import org.apache.flink.configuration.Configuration;
 import org.apache.flink.configuration.ReadableConfig;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.datastream.DataStreamSink;
@@ -285,8 +286,9 @@ private DynamicTableRecordGenerator createDynamicRecordGenerator(String generato
           DynConstructors.builder(DynamicTableRecordGenerator.class)
               .loader(IcebergTableSink.class.getClassLoader())
               .impl(generatorImpl, RowType.class)
+              .impl(generatorImpl, RowType.class, Map.class, Configuration.class)
               .buildChecked();
-      return ctor.newInstance(rowType);
+      return ctor.newInstance(rowType, writeProps, fromReadableConfig());
     } catch (ClassCastException e) {
       throw new IllegalArgumentException(
           String.format("Class %s does not implement DynamicRecordGeneratorSQL", generatorImpl), e);
@@ -295,4 +297,10 @@ private DynamicTableRecordGenerator createDynamicRecordGenerator(String generato
           String.format("Failed to instantiate DynamicRecordGeneratorSQL %s", generatorImpl), e);
     }
   }
+
+  private Configuration fromReadableConfig() {
+    return readableConfig instanceof Configuration
+        ? (Configuration) readableConfig
+        : Configuration.fromMap(readableConfig.toMap());
+  }
 }

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/TypeToFlinkType.java → flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/TypeToFlinkType.java

@@ -37,6 +37,7 @@
 import org.apache.flink.table.types.logical.TimestampType;
 import org.apache.flink.table.types.logical.VarBinaryType;
 import org.apache.flink.table.types.logical.VarCharType;
+import org.apache.flink.table.types.logical.VariantType;
 import org.apache.iceberg.Schema;
 import org.apache.iceberg.relocated.com.google.common.collect.Lists;
 import org.apache.iceberg.types.Type;
@@ -83,6 +84,11 @@ public LogicalType map(Types.MapType map, LogicalType keyResult, LogicalType val
     return new MapType(keyResult.copy(false), valueResult.copy(map.isValueOptional()));
   }
 
+  @Override
+  public LogicalType variant(Types.VariantType variant) {
+    return new VariantType();
+  }
+
   @Override
   public LogicalType primitive(Type.PrimitiveType primitive) {
     switch (primitive.typeId()) {

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/data/AvroWithFlinkSchemaVisitor.java → flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/data/AvroWithFlinkSchemaVisitor.java

@@ -24,13 +24,19 @@
 import org.apache.flink.table.types.logical.MapType;
 import org.apache.flink.table.types.logical.NullType;
 import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.table.types.logical.VariantType;
 import org.apache.iceberg.avro.AvroWithPartnerByStructureVisitor;
 import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
 import org.apache.iceberg.util.Pair;
 
 public abstract class AvroWithFlinkSchemaVisitor<T>
     extends AvroWithPartnerByStructureVisitor<LogicalType, T> {
 
+  @Override
+  protected boolean isVariantType(LogicalType logicalType) {
+    return logicalType instanceof VariantType;
+  }
+
   @Override
   protected boolean isStringType(LogicalType logicalType) {
     return logicalType.getTypeRoot().getFamilies().contains(LogicalTypeFamily.CHARACTER_STRING);

flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/data/FlinkAvroWriter.java → flink/v2.2/flink/src/main/java/org/apache/iceberg/flink/data/FlinkAvroWriter.java

@@ -110,6 +110,12 @@ public ValueWriter<?> map(
           keyWriter, mapKeyType(sMap), valueWriter, mapValueType(sMap));
     }
 
+    @Override
+    public ValueWriter<?> variant(
+        LogicalType variant, ValueWriter<?> metadataWriter, ValueWriter<?> valueWriter) {
+      return FlinkValueWriters.variants();
+    }
+
     @Override
     public ValueWriter<?> primitive(LogicalType type, Schema primitive) {
       org.apache.avro.LogicalType logicalType = primitive.getLogicalType();