Flink: Support writing shredded variant in Flink

docs/docs/flink-configuration.md

@@ -160,6 +160,8 @@ INSERT INTO tableName /*+ OPTIONS('upsert-enabled'='true') */
 | compression-strategy                    | Table write.orc.compression-strategy       | Overrides this table's compression strategy for ORC tables for this write                                                                       |
 | write-parallelism                       | Upstream operator parallelism              | Overrides the writer parallelism                                                                                                                |
 | uid-suffix                              | As per table property                      | Overrides the uid suffix used in the underlying IcebergSink for this table                                                                      |
+| shred-variants                          | Table write.parquet.shred-variants         | Overrides this table's shred variants for this write |
+| variant-inference-buffer-size           | Table write.parquet.variant-inference-buffer-size | Overrides this table's variant inference buffer size this write |
 
 #### Range distribution statistics type
 

flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/FlinkWriteConf.java

@@ -262,4 +262,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.1/flink/src/main/java/org/apache/iceberg/flink/FlinkWriteOptions.java

@@ -105,4 +105,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.1/flink/src/main/java/org/apache/iceberg/flink/data/FlinkFormatModels.java

@@ -19,6 +19,7 @@
 package org.apache.iceberg.flink.data;
 
 import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.typeutils.RowDataSerializer;
 import org.apache.flink.table.types.logical.RowType;
 import org.apache.iceberg.avro.AvroFormatModel;
 import org.apache.iceberg.formats.FormatModelRegistry;
@@ -33,7 +34,9 @@ public static void register() {
             RowType.class,
             FlinkParquetWriters::buildWriter,
             (icebergSchema, fileSchema, engineSchema, idToConstant) ->
-                FlinkParquetReaders.buildReader(icebergSchema, fileSchema, idToConstant)));
+                FlinkParquetReaders.buildReader(icebergSchema, fileSchema, idToConstant),
+            new FlinkVariantShreddingAnalyzer(),
+            rowType -> new RowDataSerializer(rowType)::copy));
 
     FormatModelRegistry.register(
         AvroFormatModel.create(

flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/data/FlinkVariantShreddingAnalyzer.java

@@ -0,0 +1,72 @@
+/*
+ * 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.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.util.List;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.types.variant.BinaryVariant;
+import org.apache.flink.types.variant.Variant;
+import org.apache.iceberg.parquet.VariantShreddingAnalyzer;
+import org.apache.iceberg.relocated.com.google.common.collect.Lists;
+import org.apache.iceberg.variants.VariantMetadata;
+import org.apache.iceberg.variants.VariantValue;
+
+/**
+ * Analyzes Variant fields in Flink {@link RowData} and converts Flink's binary Variant
+ * representation to Iceberg {@link VariantValue} instances for Variant shredding.
+ */
+public class FlinkVariantShreddingAnalyzer extends VariantShreddingAnalyzer<RowData, RowType> {
+
+  @Override
+  protected List<VariantValue> extractVariantValues(
+      List<RowData> bufferedRows, int variantFieldIndex) {
+    List<VariantValue> values = Lists.newArrayList();
+
+    for (RowData row : bufferedRows) {
+      if (!row.isNullAt(variantFieldIndex)) {
+        Variant flinkVariant = row.getVariant(variantFieldIndex);
+        if (flinkVariant != null) {
+          if (flinkVariant instanceof BinaryVariant binaryVariant) {
+            VariantValue variantValue =
+                VariantValue.from(
+                    VariantMetadata.from(
+                        ByteBuffer.wrap(binaryVariant.getMetadata())
+                            .order(ByteOrder.LITTLE_ENDIAN)),
+                    ByteBuffer.wrap(binaryVariant.getValue()).order(ByteOrder.LITTLE_ENDIAN));
+
+            values.add(variantValue);
+          } else {
+            throw new UnsupportedOperationException(
+                "Not a supported type: " + flinkVariant.getClass());
+          }
+        }
+      }
+    }
+
+    return values;
+  }
+
+  @Override
+  protected int resolveColumnIndex(RowType flinkSchema, String columnName) {
+    return flinkSchema.getFieldIndex(columnName);
+  }
+}

flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/SinkUtil.java

@@ -24,6 +24,8 @@
 import static org.apache.iceberg.TableProperties.ORC_COMPRESSION_STRATEGY;
 import static org.apache.iceberg.TableProperties.PARQUET_COMPRESSION;
 import static org.apache.iceberg.TableProperties.PARQUET_COMPRESSION_LEVEL;
+import static org.apache.iceberg.TableProperties.PARQUET_SHRED_VARIANTS;
+import static org.apache.iceberg.TableProperties.PARQUET_VARIANT_BUFFER_SIZE;
 
 import java.util.List;
 import java.util.Map;
@@ -128,6 +130,10 @@ public static Map<String, String> writeProperties(
           writeProperties.put(PARQUET_COMPRESSION_LEVEL, parquetCompressionLevel);
         }
 
+        writeProperties.put(PARQUET_SHRED_VARIANTS, String.valueOf(conf.parquetShredVariants()));
+        writeProperties.put(
+            PARQUET_VARIANT_BUFFER_SIZE, String.valueOf(conf.parquetVariantInferenceBufferSize()));
+
         break;
       case AVRO:
         writeProperties.put(AVRO_COMPRESSION, conf.avroCompressionCodec());