Repository: apache/iceberg (language: python)

Issue:
Flink: Add table.exec.iceberg.use-v2-sink option
This PR adds a `table.exec.iceberg.use-v2-sink` configuration option allowing to use Flink's Sink v2 API described in the [FLIP-143](https://cwiki.apache.org/confluence/display/FLINK/FLIP-143%3A+Unified+Sink+API) document.

The configuration option is by default set to `false`.

This PR is the follow-up of discussion in #10179 and in #11219


Discussion / hints:
@rodmeneses , @pvary please have a look at this PR. In the change, I added the configuration option that we talked about. I also added the documentation. In the docs I showed differences between the interfaces that I found while working on unifying the unit tests for both sinks. 

This is still a draft, as I have no idea how to show in unit tests, that this configuration option does anything 😅. I've only checked in the debugger that the code is entering the correct code blocks. I could add a test on `IcebergTableSink.getSinkRuntimeProvider` level that would check which implementation is chosen but that doesn't seem like a good idea.

Do you know of any differences in the behaviour of these sinks that I could show in the test? I see that there are plans to add small file compaction in SinkV2 based implementation. Perhaps we should add this test when this feature will be available. WDYT?
> This is still a draft, as I have no idea how to show in unit tests, that this configuration option does anything 😅. I've only checked in the debugger that the code is entering the correct code blocks. I could add a test on `IcebergTableSink.getSinkRuntimeProvider` level that would check which implementation is chosen but that doesn't seem like a good idea.

I found out the way to verify if the correct API used. I added the test. This change is ready for review.
@rodmeneses, @stevenzwu: Please review the PR. Especially the documentation part.
@rodmeneses @pvary I've done all fixes that were discussed in comments. Can we move forward with this change? Do we need more eyes to take a look at the docs part? @stevenzwu can you take a look? 
This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.
Hi @arkadius is this still needed/relevant ? please advise, as it will be closed due to inactivity soon
cc @pvary 
I think that this option would benefit this project in a good transitional migration to the new interface in Table API / Flink SQL. What is the alternative plan for the migration between the old implementation and the new one? 
We definitely need this.
@arkadius: Could you please make sure the tests are run? Just change the commit message of the last commit and force push the branch, or something?
> @arkadius: Could you please make sure the tests are run? Just change the commit message of the last commit and force push the branch, or something?

Sure, I rebased the branch to main to make sure that all tests will be run on the state that will be present after merge
Thanks for the PR @arkadius and @rodmeneses for the review!
@arkadius: Could you please backport the change to the relevant older Flink branches?

Fix patch:
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/FlinkConfigOptions.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/FlinkConfigOptions.java
index 7c7afd24ed8e..6362bc447634 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/FlinkConfigOptions.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/FlinkConfigOptions.java
@@ -91,6 +91,12 @@ private FlinkConfigOptions() {}
           .defaultValue(false)
           .withDescription("Use the FLIP-27 based Iceberg source implementation.");
 
+  public static final ConfigOption<Boolean> TABLE_EXEC_ICEBERG_USE_V2_SINK =
+      ConfigOptions.key("table.exec.iceberg.use-v2-sink")
+          .booleanType()
+          .defaultValue(false)
+          .withDescription("Use the SinkV2 API based Iceberg sink implementation.");
+
   public static final ConfigOption<SplitAssignerType> TABLE_EXEC_SPLIT_ASSIGNER_TYPE =
       ConfigOptions.key("table.exec.iceberg.split-assigner-type")
           .enumType(SplitAssignerType.class)
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/IcebergTableSink.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/IcebergTableSink.java
index 1b9268569d9a..c2c5a6706e92 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/IcebergTableSink.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/IcebergTableSink.java
@@ -35,6 +35,7 @@
 import org.apache.flink.types.RowKind;
 import org.apache.flink.util.Preconditions;
 import org.apache.iceberg.flink.sink.FlinkSink;
+import org.apache.iceberg.flink.sink.IcebergSink;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
 
 public class IcebergTableSink implements DynamicTableSink, SupportsPartitioning, SupportsOverwrite {
@@ -77,14 +78,25 @@ public SinkRuntimeProvider getSinkRuntimeProvider(Context context) {
       @Override
       public DataStreamSink<?> consumeDataStream(
           ProviderContext providerContext, DataStream<RowData> dataStream) {
-        return FlinkSink.forRowData(dataStream)
-            .tableLoader(tableLoader)
-            .tableSchema(tableSchema)
-            .equalityFieldColumns(equalityColumns)
-            .overwrite(overwrite)
-            .setAll(writeProps)
-            .flinkConf(readableConfig)
-            .append();
+        if (readableConfig.get(FlinkConfigOptions.TABLE_EXEC_ICEBERG_USE_V2_SINK)) {
+          return IcebergSink.forRowData(dataStream)
+              .tableLoader(tableLoader)
+              .tableSchema(tableSchema)
+              .equalityFieldColumns(equalityColumns)
+              .overwrite(overwrite)
+              .setAll(writeProps)
+              .flinkConf(readableConfig)
+              .append();
+        } else {
+          return FlinkSink.forRowData(dataStream)
+              .tableLoader(tableLoader)
+              .tableSchema(tableSchema)
+              .equalityFieldColumns(equalityColumns)
+              .overwrite(overwrite)
+              .setAll(writeProps)
+              .flinkConf(readableConfig)
+              .append();
+        }
       }
     };
   }
diff --git a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
index c53431490984..3f1fef6580ef 100644
--- a/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
+++ b/flink/v1.19/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java
@@ -41,7 +41,7 @@
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.datastream.DataStreamSink;
 import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
-import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
+import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
 import org.apache.flink.table.api.TableSchema;
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.data.util.DataFormatConverters;
@@ -381,7 +381,7 @@ public Builder toBranch(String branch) {
       return this;
     }
 
-    private <T> DataStreamSink<T> chainIcebergOperators() {
+    private DataStreamSink<Void> chainIcebergOperators() {
       Preconditions.checkArgument(
           inputCreator != null,
           "Please use forRowData() or forMapperOutputType() to initialize the input DataStream.");
@@ -472,12 +472,10 @@ List<Integer> checkAndGetEqualityFieldIds() {
       return equalityFieldIds;
     }
 
-    @SuppressWarnings("unchecked")
-    private <T> DataStreamSink<T> appendDummySink(
-        SingleOutputStreamOperator<Void> committerStream) {
-      DataStreamSink<T> resultStream =
+    private DataStreamSink<Void> appendDummySink(SingleOutputStreamOperator<Void> committerStream) {
+      DataStreamSink<Void> resultStream =
           committerStream
-              .addSink(new DiscardingSink())
+              .sinkTo(new DiscardingSink<>())
               .name(operatorName(String.format("IcebergSink %s", this.table.name())))
               .setParallelism(1);
       if (uidPrefix != null) {


Test patch:
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSink.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSink.java
index 2978a92945a2..fad30f9c1e67 100644
--- a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSink.java
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSink.java
@@ -40,7 +40,6 @@
 
 public class TestFlinkTableSink extends CatalogTestBase {
 
-  private static final String SOURCE_TABLE = "default_catalog.default_database.bounded_source";
   private static final String TABLE_NAME = "test_table";
   private TableEnvironment tEnv;
   private Table icebergTable;
@@ -51,7 +50,11 @@ public class TestFlinkTableSink extends CatalogTestBase {
   @Parameter(index = 3)
   private boolean isStreamingJob;
 
-  @Parameters(name = "catalogName={0}, baseNamespace={1}, format={2}, isStreaming={3}")
+  @Parameter(index = 4)
+  private boolean useV2Sink;
+
+  @Parameters(
+      name = "catalogName={0}, baseNamespace={1}, format={2}, isStreaming={3}, useV2Sink={4}")
   public static List<Object[]> parameters() {
     List<Object[]> parameters = Lists.newArrayList();
     for (FileFormat format :
@@ -60,10 +63,24 @@ public static List<Object[]> parameters() {
         for (Object[] catalogParams : CatalogTestBase.parameters()) {
           String catalogName = (String) catalogParams[0];
           Namespace baseNamespace = (Namespace) catalogParams[1];
-          parameters.add(new Object[] {catalogName, baseNamespace, format, isStreaming});
+          parameters.add(
+              new Object[] {
+                catalogName, baseNamespace, format, isStreaming, false /* don't use v2 sink */
+              });
         }
       }
     }
+
+    for (FileFormat format :
+        new FileFormat[] {FileFormat.ORC, FileFormat.AVRO, FileFormat.PARQUET}) {
+      for (Boolean isStreaming : new Boolean[] {true, false}) {
+        String catalogName = "testhadoop_basenamespace";
+        Namespace baseNamespace = Namespace.of("l0", "l1");
+        parameters.add(
+            new Object[] {catalogName, baseNamespace, format, isStreaming, true /* use v2 sink */});
+      }
+    }
+
     return parameters;
   }
 
@@ -87,6 +104,11 @@ protected TableEnvironment getTableEnv() {
         }
       }
     }
+
+    tEnv.getConfig()
+        .getConfiguration()
+        .set(FlinkConfigOptions.TABLE_EXEC_ICEBERG_USE_V2_SINK, useV2Sink);
+
     return tEnv;
   }
 
diff --git a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSinkExtended.java b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSinkExtended.java
index b63547d433a4..3afabf6e0795 100644
--- a/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSinkExtended.java
+++ b/flink/v1.19/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSinkExtended.java
@@ -31,6 +31,8 @@
 import java.util.stream.IntStream;
 import org.apache.flink.api.dag.Transformation;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
+import org.apache.flink.streaming.api.functions.sink.v2.DiscardingSink;
+import org.apache.flink.streaming.api.transformations.SinkTransformation;
 import org.apache.flink.table.api.EnvironmentSettings;
 import org.apache.flink.table.api.TableEnvironment;
 import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
@@ -52,6 +54,7 @@
 import org.apache.iceberg.TableProperties;
 import org.apache.iceberg.catalog.Namespace;
 import org.apache.iceberg.catalog.TableIdentifier;
+import org.apache.iceberg.flink.sink.IcebergSink;
 import org.apache.iceberg.flink.source.BoundedTableFactory;
 import org.apache.iceberg.hadoop.HadoopCatalog;
 import org.apache.iceberg.relocated.com.google.common.collect.ImmutableList;
@@ -91,11 +94,20 @@ public class TestFlinkTableSinkExtended extends SqlBase {
 
   private TableEnvironment tEnv;
 
-  @Parameter protected boolean isStreamingJob;
+  @Parameter(index = 0)
+  protected boolean isStreamingJob;
 
-  @Parameters(name = "isStreamingJob={0}")
+  @Parameter(index = 1)
+  protected Boolean useV2Sink;
+
+  @Parameters(name = "isStreamingJob={0}, useV2Sink={1}")
   protected static List<Object[]> parameters() {
-    return Arrays.asList(new Boolean[] {true}, new Boolean[] {false});
+    return Arrays.asList(
+        new Object[] {true, false},
+        new Object[] {false, false},
+        new Object[] {true, true},
+        new Object[] {false, true},
+        new Object[] {true, null});
   }
 
   protected synchronized TableEnvironment getTableEnv() {
@@ -115,6 +127,13 @@ protected synchronized TableEnvironment getTableEnv() {
         tEnv = TableEnvironment.create(settingsBuilder.build());
       }
     }
+
+    if (useV2Sink != null) {
+      tEnv.getConfig()
+          .getConfiguration()
+          .set(FlinkConfigOptions.TABLE_EXEC_ICEBERG_USE_V2_SINK, useV2Sink);
+    }
+
     return tEnv;
   }
 
@@ -146,6 +165,32 @@ public void clean() throws Exception {
     catalog.close();
   }
 
+  @TestTemplate
+  public void testUsedFlinkSinkInterface() {
+    String dataId = BoundedTableFactory.registerDataSet(Collections.emptyList());
+    sql(
+        "CREATE TABLE %s(id INT NOT NULL, data STRING NOT NULL)"
+            + " WITH ('connector'='BoundedSource', 'data-id'='%s')",
+        SOURCE_TABLE, dataId);
+
+    PlannerBase planner = (PlannerBase) ((TableEnvironmentImpl) getTableEnv()).getPlanner();
+    String insertSQL = String.format("INSERT INTO %s SELECT * FROM %s", TABLE, SOURCE_TABLE);
+    ModifyOperation operation = (ModifyOperation) planner.getParser().parse(insertSQL).get(0);
+    Transformation<?> transformation =
+        planner.translate(Collections.singletonList(operation)).get(0);
+    assertThat(transformation).as("Should use SinkV2 API").isInstanceOf(SinkTransformation.class);
+    SinkTransformation<?, ?> sinkTransformation = (SinkTransformation<?, ?>) transformation;
+    if (useV2Sink != null && useV2Sink) {
+      assertThat(sinkTransformation.getSink())
+          .as("Should use SinkV2 API based implementation")
+          .isInstanceOf(IcebergSink.class);
+    } else {
+      assertThat(sinkTransformation.getSink())
+          .as("Should use custom chain of StreamOperators terminated by DiscardingSink")
+          .isInstanceOf(DiscardingSink.class);
+    }
+  }
+
   @TestTemplate
   public void testWriteParallelism() {
     List<Row> dataSet =
@@ -165,18 +210,25 @@ public void testWriteParallelism() {
             "INSERT INTO %s /*+ OPTIONS('write-parallelism'='1') */ SELECT * FROM %s",
             TABLE, SOURCE_TABLE);
     ModifyOperation operation = (ModifyOperation) planner.getParser().parse(insertSQL).get(0);
-    Transformation<?> dummySink = planner.translate(Collections.singletonList(operation)).get(0);
-    Transformation<?> committer = dummySink.getInputs().get(0);
-    Transformation<?> writer = committer.getInputs().get(0);
-
-    assertThat(writer.getParallelism()).as("Should have the expected 1 parallelism.").isEqualTo(1);
-    writer
-        .getInputs()
-        .forEach(
-            input ->
-                assertThat(input.getParallelism())
-                    .as("Should have the expected parallelism.")
-                    .isEqualTo(isStreamingJob ? 2 : 4));
+    Transformation<?> sink = planner.translate(Collections.singletonList(operation)).get(0);
+    if (useV2Sink != null && useV2Sink) {
+      assertThat(sink.getParallelism()).as("Should have the expected 1 parallelism.").isEqualTo(1);
+      Transformation<?> writerInput = sink.getInputs().get(0);
+      assertThat(writerInput.getParallelism())
+          .as("Should have the expected parallelism.")
+          .isEqualTo(isStreamingJob ? 2 : 4);
+    } else {
+      Transformation<?> committer = sink.getInputs().get(0);
+      Transformation<?> writer = committer.getInputs().get(0);
+
+      assertThat(writer.getParallelism())
+          .as("Should have the expected 1 parallelism.")
+          .isEqualTo(1);
+      Transformation<?> writerInput = writer.getInputs().get(0);
+      assertThat(writerInput.getParallelism())
+          .as("Should have the expected parallelism.")
+          .isEqualTo(isStreamingJob ? 2 : 4);
+    }
   }
 
   @TestTemplate