Flink: Add table.exec.iceberg.use-v2-sink option #11244

arkadius · 2024-10-01T14:51:54Z

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 document.

The configuration option is by default set to false.

This PR is the follow-up of discussion in #10179 and in #11219

arkadius · 2024-10-01T16:32:54Z

@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?

arkadius · 2024-10-03T10:47:31Z

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.

pvary · 2024-10-04T07:22:46Z

docs/docs/flink-writes.md

+The previous [SinkV1 interface](https://cwiki.apache.org/confluence/display/FLINK/FLIP-143%3A+Unified+Sink+API)
+had some limitations - for example it created a lot of small files when writing to it. This problem is called
+the `small-file-compaction` problem in
+the [FLIP-191 document](https://cwiki.apache.org/confluence/display/FLINK/FLIP-191%3A+Extend+unified+Sink+interface+to+support+small+file+compaction).


The table maintenance is not available yet, so I would not mention it in this PR yet.

The previous [SinkV1 interface](https://cwiki.apache.org/confluence/display/FLINK/FLIP-143%3A+Unified+Sink+API) will be deprecated and removed from Flink.

Ok, I see your point, but I think we need to change it even more, I'm currently thinking about how to do this. The problem with the proposed changes in the introducing section are:

Flink had 3 approaches to defining sink: SinkFunction, SinkV1 (FLIP-143) and SinkV2 (FLIP-191). In FlinkSink we used the first one approach, not the SinkV1 but ...

... It is not so important because the main logic of sinking logic is done in operators before the sink. For the sink is used DiscardingSink which BTW can be replaced by SinkV2 version to avoid problems with the removal of the previous interface (I've just did it)

IMO we should give some introduction explaining why we have 2 implementations and it is problematic without exposing future plans. Do you see some other idea?

I've rewritten the introduction part. Please correct me if I've twisted something or if you see it differently.

pvary · 2024-10-04T07:23:37Z

docs/docs/flink-writes.md

+## Writing with DataStream
+
+To use SinkV2 based implementation, replace `FlinkSink` with `IcebergSink` in the provided snippets.
+Warning: some settings are not available in this class (e.g. `rangeDistributionStatisticsType`),


The `RANGE` distribution mode is not yet available for the `IcebergSink`.

pvary · 2024-10-04T07:24:28Z

docs/docs/flink-writes.md

+
+To use SinkV2 based implementation, replace `FlinkSink` with `IcebergSink` in the provided snippets.
+Warning: some settings are not available in this class (e.g. `rangeDistributionStatisticsType`),
+some others are slightly different (e.g. there is a `uidSuffix` method instead of `uidPrefix`).


When using IcebergSink use uidPrefix instead of the uidSuffix

👍 with one little change, should be opposite: IcebergSink has the uidSuffix method and FlinkSink has the uidPrefix method

flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSink.java

pvary · 2024-10-04T07:27:16Z

flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSinkExtended.java

@@ -115,6 +125,9 @@ protected synchronized TableEnvironment getTableEnv() {
        tEnv = TableEnvironment.create(settingsBuilder.build());
      }
    }
+    tEnv.getConfig()


nit: we add an empty line between blocks

pvary · 2024-10-04T07:28:29Z

@rodmeneses, @stevenzwu: Please review the PR. Especially the documentation part.

rodmeneses · 2024-10-07T18:23:56Z

flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java

@@ -381,7 +381,7 @@ public Builder toBranch(String branch) {
      return this;
    }

-    private <T> DataStreamSink<T> chainIcebergOperators() {
+    private DataStreamSink<Void> chainIcebergOperators() {


why is this change being introduced?

This is related to this discussion: #11244 (comment). I realize that we don't need to use the legacy, deprecated interface in the FlinkSink implementation as the sink part of the chain is not important in this chain of transformations - discarding sink behaves the same when it is used either as legacy SinkFunction or the new, SinkV2 interface.

rodmeneses · 2024-10-07T18:24:06Z

flink/v1.20/flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java

          committerStream
-              .addSink(new DiscardingSink())
+              .sinkTo(new DiscardingSink<>())


why is this change being introduced?

The same as above

rodmeneses · 2024-10-07T18:30:23Z

flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSink.java

@@ -60,10 +63,23 @@ 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});
+          boolean doNotUseV2Sink = false;


we have a kind of double negation doNotUseV2Sink = false , but below in the other block we have boolean useV2Sink = true;
could we just do

parameters.add( new Object[] {catalogName, baseNamespace, format, isStreaming, false});

ie: removing this local variable?

In Contribute guide is written that boolean arguments should be named. This rule shouldn't be applied in this place as well? Or do you prefer the comment instead?

I replaced it with comment, take a look now

rodmeneses · 2024-10-07T18:41:08Z

flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSinkExtended.java

@@ -115,6 +126,9 @@ protected synchronized TableEnvironment getTableEnv() {
        tEnv = TableEnvironment.create(settingsBuilder.build());
      }
    }
+    tEnv.getConfig()
+        .getConfiguration()
+        .set(FlinkConfigOptions.TABLE_EXEC_ICEBERG_USE_V2_SINK, useV2Sink);


nit:
I see we added this block on this PR, and we provide testTemplate with useV2Sink=true and useV2Sink=false.
I wonder if it would be a good idea to also test the scenarios when we dont explicitly set this ConfigOption? In my mind, it should behave identically as when user does .set(FlinkConfigOptions.TABLE_EXEC_ICEBERG_USE_V2_SINK, false);

But I may be overcomplicating this. what do you think? @pvary

Good point, I'll add a test case for that.

docs/docs/flink-writes.md

rodmeneses · 2024-10-10T16:13:02Z

flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSinkExtended.java

-                assertThat(input.getParallelism())
-                    .as("Should have the expected parallelism.")
-                    .isEqualTo(isStreamingJob ? 2 : 4));
+    Transformation<?> sink = planner.translate(Collections.singletonList(operation)).get(0);


do you plan to leverage the marker interface for the sink builder in #11219
for these tests?

I don't think that it is possible as we don't have builders here. Instead, we have already built a chain of operators. This test works on the table connectors level - builders are hidden behind the table connector API as an implementation detail.

arkadius · 2024-10-21T08:39:31Z

@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?

github-actions · 2024-11-21T00:16:18Z

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 [email protected] list. Thank you for your contributions.

rodmeneses · 2024-11-21T01:42:19Z

Hi @arkadius is this still needed/relevant ? please advise, as it will be closed due to inactivity soon
cc @pvary

arkadius · 2024-11-21T06:35:14Z

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?

pvary · 2024-11-21T13:52:39Z

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 · 2024-11-21T14:49:46Z

@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

pvary · 2024-11-25T12:23:26Z

Thanks for the PR @arkadius and @rodmeneses for the review!

pvary · 2024-11-25T12:24:16Z

@arkadius: Could you please backport the change to the relevant older Flink branches?

…e-v2-sink option)

arkadius · 2024-11-27T11:54:06Z

@arkadius: Could you please backport the change to the relevant older Flink branches?

Sure, here is backport for Flink 1.19: #11665
The SinkV2 API was added in Flink 1.19 so it is only relevant for this version.

…ink option) (#11665)

…e-v2-sink option) (apache#11665)

github-actions bot added the flink label Oct 1, 2024

arkadius changed the title ~~Add table.exec.iceberg.use-v2-sink option~~ Fflink: Add table.exec.iceberg.use-v2-sink option Oct 1, 2024

github-actions bot added the docs label Oct 1, 2024

arkadius force-pushed the sinkv2-configuration-option branch from 70857ec to 213668f Compare October 2, 2024 07:47

arkadius marked this pull request as ready for review October 3, 2024 10:46

pvary reviewed Oct 4, 2024

View reviewed changes

flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSink.java Show resolved Hide resolved

pvary reviewed Oct 4, 2024

View reviewed changes

flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSink.java Show resolved Hide resolved

pvary reviewed Oct 4, 2024

View reviewed changes

flink/v1.20/flink/src/test/java/org/apache/iceberg/flink/TestFlinkTableSink.java Show resolved Hide resolved

pvary reviewed Oct 4, 2024

View reviewed changes

arkadius force-pushed the sinkv2-configuration-option branch 2 times, most recently from 23fa5e8 to 525b6ee Compare October 4, 2024 14:28

arkadius requested a review from pvary October 7, 2024 07:23

rodmeneses reviewed Oct 7, 2024

View reviewed changes

docs/docs/flink-writes.md Show resolved Hide resolved

arkadius requested a review from rodmeneses October 7, 2024 21:17

arkadius changed the title ~~Fflink: Add table.exec.iceberg.use-v2-sink option~~ Flink: Add table.exec.iceberg.use-v2-sink option Oct 8, 2024

arkadius mentioned this pull request Oct 10, 2024

Flink: Tests alignment for the Flink Sink v2-based implemenation (IcebergSink) #11219

Closed

rodmeneses reviewed Oct 10, 2024

View reviewed changes

arkadius requested a review from rodmeneses October 21, 2024 08:37

github-actions bot added the stale label Nov 21, 2024

Arek Burdach added 6 commits November 21, 2024 15:48

Add table.exec.iceberg.use-v2-sink option

5ab0cce

Added test for correct sink interface usage

18e1996

Review fixes

032b71a

FlinkSink using v2 sinks instead of SinkFunction

8591ba0

introduction rewritten

839fe1f

review fixes

799de35

arkadius force-pushed the sinkv2-configuration-option branch from ee48c23 to 799de35 Compare November 21, 2024 14:48

github-actions bot removed the stale label Nov 22, 2024

pvary approved these changes Nov 22, 2024

View reviewed changes

pvary merged commit fa47f31 into apache:main Nov 25, 2024
20 checks passed

arkadius added a commit to arkadius/iceberg that referenced this pull request Nov 27, 2024

Flink: Backport apache#11244 to Flink 1.19 (Add table.exec.iceberg.us…

dd91c94

…e-v2-sink option)

pvary pushed a commit that referenced this pull request Nov 27, 2024

Flink: Backport #11244 to Flink 1.19 (Add table.exec.iceberg.use-v2-s…

57527d7

…ink option) (#11665)

zachdisc pushed a commit to zachdisc/iceberg that referenced this pull request Dec 23, 2024

Flink: Add table.exec.iceberg.use-v2-sink option (apache#11244)

c1a432d

zachdisc pushed a commit to zachdisc/iceberg that referenced this pull request Dec 23, 2024

Flink: Backport apache#11244 to Flink 1.19 (Add table.exec.iceberg.us…

4134bb6

…e-v2-sink option) (apache#11665)

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Flink: Add table.exec.iceberg.use-v2-sink option #11244

Flink: Add table.exec.iceberg.use-v2-sink option #11244

arkadius commented Oct 1, 2024 •

edited

Loading

arkadius commented Oct 1, 2024

arkadius commented Oct 3, 2024

pvary Oct 4, 2024

arkadius Oct 4, 2024 •

edited

Loading

arkadius Oct 4, 2024

pvary Oct 4, 2024

pvary Oct 4, 2024

arkadius Oct 4, 2024

pvary Oct 4, 2024

pvary commented Oct 4, 2024

rodmeneses Oct 7, 2024

arkadius Oct 7, 2024

rodmeneses Oct 7, 2024

arkadius Oct 7, 2024

rodmeneses Oct 7, 2024

arkadius Oct 7, 2024 •

edited

Loading

arkadius Oct 7, 2024

rodmeneses Oct 7, 2024

arkadius Oct 7, 2024

rodmeneses Oct 10, 2024

arkadius Oct 10, 2024

arkadius commented Oct 21, 2024

github-actions bot commented Nov 21, 2024

rodmeneses commented Nov 21, 2024

arkadius commented Nov 21, 2024

pvary commented Nov 21, 2024

arkadius commented Nov 21, 2024

pvary commented Nov 25, 2024

pvary commented Nov 25, 2024

arkadius commented Nov 27, 2024 •

edited

Loading

Flink: Add table.exec.iceberg.use-v2-sink option #11244

Flink: Add table.exec.iceberg.use-v2-sink option #11244

Conversation

arkadius commented Oct 1, 2024 • edited Loading

arkadius commented Oct 1, 2024

arkadius commented Oct 3, 2024

Choose a reason for hiding this comment

arkadius Oct 4, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pvary commented Oct 4, 2024

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

arkadius Oct 7, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

arkadius commented Oct 21, 2024

github-actions bot commented Nov 21, 2024

rodmeneses commented Nov 21, 2024

arkadius commented Nov 21, 2024

pvary commented Nov 21, 2024

arkadius commented Nov 21, 2024

pvary commented Nov 25, 2024

pvary commented Nov 25, 2024

arkadius commented Nov 27, 2024 • edited Loading

arkadius commented Oct 1, 2024 •

edited

Loading

arkadius Oct 4, 2024 •

edited

Loading

arkadius Oct 7, 2024 •

edited

Loading

arkadius commented Nov 27, 2024 •

edited

Loading