Flink: Add streaming upsert write option. by Reo-LEI · Pull Request #2863 · apache/iceberg

Reo-LEI · 2021-07-25T16:01:21Z

fork from #1996

… use UPSERT.

# Conflicts: # flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java # flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergStreamWriter.java

Reo-LEI · 2021-07-25T16:08:10Z

      case INSERT:
+        if (upsert) {
+          writer.delete(row);
+        }


I think we could only delete row on INSERT. I don't think there will be only have UPDATE_AFTER row and lost UPDATE_BEFORE situation. @openinx please check this.

For an update operation in flink, the UPDATE_AFTER event will must be emitted to the downstream, while the UPDATE_BEFORE is a best effort behavior or an configured behavior from the upstream flink source. You can take a look at this GroupAggFunction, if the flink source is configured to produce UPDATE_AFTER only, then it won't emit any UPDATE_BEFORE to the downstream.

For the downstream iceberg sink, we need to handle all the UPDATE_AFTER as UPSERT. That also means we need to do nothing for the UPDATE_BEFORE because we will remove the previous key in the next UPDATE_AFTER events.

I think we can only delete row on UPDATE_AFTER and keep UPDATE_BEFORE do nothing to prevent delete one row twice

By name, I thought INSERT is "add a new row". Then we don't need to add a delete for it. But I guess it actually means "append a row (new or updated)".

As @openinx mentioned in #1996 (comment), we need to transform INSERT/UPDATE_AFTER to be UPSERT(delete + insert). If we don't add a delete on INSERT row when upsert mode is enable, we will get duplicate rows for same primary key.

Reo-LEI · 2021-07-25T16:12:01Z

I pick #1996 up on this PR, could you take a look? @rdblue @openinx

coolderli · 2021-07-28T13:51:29Z

@Reo-LEI any update?

…mode

Reo-LEI · 2021-07-28T16:31:11Z

@Reo-LEI any update?

Yep, I will keep push this PR. But I only have time on night, so the progress will be slow.

Reo-LEI · 2021-07-31T02:12:30Z

ping @openinx @rdblue

# Conflicts: # flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java # flink/src/main/java/org/apache/iceberg/flink/sink/RowDataTaskWriterFactory.java # flink/src/main/java/org/apache/iceberg/flink/source/RowDataRewriter.java # flink/src/test/java/org/apache/iceberg/flink/sink/TestDeltaTaskWriter.java # flink/src/test/java/org/apache/iceberg/flink/sink/TestTaskWriters.java

Reo-LEI · 2021-08-23T14:28:51Z

This PR have been open for some time, I think this feature is very importent for user, and just waiting someone to review and merge that. Has anybody to take a look of this? @rdblue @openinx @aokolnychyi @stevenzwu @kbendick

…e when upsert is enable.

stevenzwu · 2021-08-25T16:31:10Z

+            "OVERWRITE mode shouldn't be enable when configuring to use UPSERT data stream.");
+        Preconditions.checkState(!equalityFieldIds.isEmpty(),
+            "Equality field columns shouldn't be empty when configuring to use UPSERT data stream.");
+        if (!table.spec().isUnpartitioned()) {


for my own learning, does partition field must be included in equality fields?

e.g., we can have an equality field (like user_id) and table can be partitioned by hour. would that be a valid scenario?

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

Line 222 in 4d33f18

* a subset of equality fields, otherwise the old row that located in partition-A could not be deleted by the

As @openinx comment as above, we shoule restrict the partition fields is a subset of equality fields to ensure we can delete the old data in same partition.

e.g., we can have an equality field (like user_id) and table can be partitioned by hour. would that be a valid scenario?

I think that is not a valid scenario, to keep user_id unique in all different hour parition is make no sense.

If we have a table with user_id and hour, the business primary key is user_id, which mean the table should have at most one row for each given user_id. Now let's take about the partition strategy.

If we just partition the table by hour field, that means two different hour partitions may have the same user_id, because people may insert the user_id in hour=01 and hour=02. If we wanna to keep the primary key semantics, then we will need to delete the old user_id in the hour=01 first, then insert the new user_id in the hour=02. But when an INSERT come, we don't know which partition has the specific user_id, then we have to broadcast the DELETE to all the partitions, which is quite inefficient.

Reo-LEI · 2021-09-05T06:13:13Z

I have fixed the unittest, could you retry the ci? :) @openinx

openinx · 2021-09-08T01:39:34Z

Retry to run CI ...

openinx · 2021-09-08T02:48:00Z

LGTM if the travis CI says OK ! Thanks for @Reo-LEI for picking this up !

kbendick

A couple questions for my own understanding.

Overall this looks good to me, but I'll leave it for others until I have a chance to look more closely as Flink is an area I'm somewhat less specialized in. Thanks for working on this @Reo-LEI and @openinx!

kbendick · 2021-09-08T20:53:22Z

+  public static final String UPSERT_MODE_ENABLE = "write.upsert.enable";
+  public static final boolean UPSERT_MODE_ENABLE_DEFAULT = false;


Two questions, one that's somewhat unrelated:

Is this only used in streaming mode now? Or does this work with Flink batch sink as well?

(Somewhat unrelated / thinking out loud) If we have this new write.upsert.enabled flag, could we possibly use it to add our own support for CDC on top of Spark Structured Streaming?

Is this only used in streaming mode now? Or does this work with Flink batch sink as well?

Yes, it's only used for streaming mode right now. The batch upsert semantic has been implemented correctly by the MERGE INTO clause.

could we possibly use it to add our own support for CDC on top of Spark Structured Streaming?

In theory, it's possible to add the CDC support for spark sturctured streaming, though the spark structured streaming does not support CDC event natively (I mean flink support INSERT/DELETE/UPDATE_BEFORE/UPDATE_AFTER events natively while Spark streaming doesn't unless we add extra field to indicate what's the operation type it is). I think @XuQianJin-Stars @chenjunjiedada 's team are working on this issue in their own repo.

kbendick · 2021-09-08T20:56:09Z

+        if (upsert) {
+          break;  // UPDATE_BEFORE is not necessary for UPDATE, we do nothing to prevent delete one row twice
+        }


Non-blocking question:

Are there possible concerns with events coming out of order for some reason? I guess since the table commits are serializable, this isn't a concern as the same row for these equality fields shouldn't be updated twice in the same commit?

It's a good question, @kbendick ! Let's describe the out-of-order in two dimension:

Is possible to produce disordered events in a single iceberg transaction ? First of all, if we want to maintain the correct data semantics between the source table and iceberg sink table, the records consumed from source table must be the correct order. Second, the streaming job will need to shuffle based on the equality fields so that the records with same key are dispatched to the specialized parallelism task, otherwise the out-of-order issue happen if different tasks write the records with same equality fields to the iceberg table. In this way, the order in a single transaction is guaranteed.

The out-of-order issue between two continues transaction. In our flink stream integration, we have guaranteed the exact commit order even if a failover happen. For the spark streaming, I think we will need more consideration to this issue.

Hopefully, I've answered your question, @kbendick :-)

kbendick · 2021-09-08T21:01:11Z

+      if (upsertMode) {
+        Preconditions.checkState(!overwrite,
+            "OVERWRITE mode shouldn't be enable when configuring to use UPSERT data stream.");


Can you add a test that verifies the builder doesn't allow overwrite and upsert / upsertMode?

Maybe I missed it, but seems like an important thing to have a test for in case of future code refractors.

I have added a unittest to cover this, thanks for your reminder！:) @kbendick

Reo-LEI · 2021-09-12T14:24:39Z

@openinx @stevenzwu @kbendick do you have another other concerns for this PR? I think this feature is very important for flink user, we should merge this PR as soon as possible. :)

openinx · 2021-09-13T02:47:23Z

@Reo-LEI , The PR looks good to me now, thanks for the patient contribution, thanks all for reviewing (@kbendick & @stevenzwu ) . I will get this merged once the travis CI says okay !

openinx · 2021-09-13T08:58:41Z

Reopened PR to retry the CI.

cross-batch duplicates In upsert mode, INSERT did not call `deleteKey()` before `write()`, unlike Flink's `BaseDeltaTaskWriter` (established in PR apache#2863, refined in PR apache#4364). This caused a "shielding" bug: when INSERT and UPDATE for the same key arrived in the same batch, INSERT's `write()` populated `insertedRowMap`, causing UPDATE's `deleteKey()` to take the position-delete path instead of emitting the equality delete needed to remove the prior batch's row. The fix adds `deleteKey(keyProjection.wrap(row))` before `write()` for INSERT when upsert is enabled - matching Flink's behavior. This produces an equality delete for every INSERT in upsert mode, even for genuinely new keys (no-op at read time, resolved by compaction). This is the same trade-off Flink makes. Adds table-level integration tests (`TestCDCDeltaWriterTableLevel`) that commit to real Iceberg tables and read back row data, following the Flink `TestDeltaTaskWriter` pattern. These tests verify that equality deletes actually mask prior-batch rows - something the existing mock-based unit tests could not validate. Also updates documentation with a delete behavior matrix, exactly-once semantics explanation, and operational guidance for CDC mode.

openinx and others added 6 commits December 28, 2020 16:43

Flink: Flat INSERT as one DELETE following one INSERT if configure to…

b047073

… use UPSERT.

Minor changes.

bab1b22

Merge remote-tracking branch 'openinx/flink-upsert' into flink-upsert

9d3d1ed

# Conflicts: # flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java # flink/src/test/java/org/apache/iceberg/flink/sink/TestIcebergStreamWriter.java

Fix unittest

49835e1

Only delete record on INSERT row kind

34f02c7

Enable upsert mode on flink sql when primary key is not null

dcdd403

github-actions Bot added the flink label Jul 25, 2021

Reo-LEI mentioned this pull request Jul 25, 2021

Flink: Transform INSERT as one DELETE following one INSERT if configure to use UPSERT #1996

Closed

Reo-LEI commented Jul 25, 2021

View reviewed changes

openinx reviewed Jul 27, 2021

View reviewed changes

Comment thread flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java

openinx reviewed Jul 27, 2021

View reviewed changes

Comment thread flink/src/main/java/org/apache/iceberg/flink/sink/FlinkSink.java Outdated

Reo-LEI added 2 commits July 29, 2021 00:21

Delete row on UPDATE_AFTER and do nothing on UPDATE_BEFORE in upsert …

28c18f3

…mode

make upsert mode configurable

8e397e3

github-actions Bot added the core label Jul 28, 2021

stevenzwu reviewed Aug 24, 2021

View reviewed changes

Comment thread flink/src/main/java/org/apache/iceberg/flink/IcebergTableSink.java Outdated

extract upsert mode prop in build() and add a validation for overwrit…

4d33f18

…e when upsert is enable.

stevenzwu reviewed Aug 25, 2021

View reviewed changes

Comment thread core/src/main/java/org/apache/iceberg/TableProperties.java Outdated

stevenzwu reviewed Aug 25, 2021

View reviewed changes

Reo-LEI added 2 commits August 28, 2021 23:38

Addressing comments.

56822c8

Fix unittest.

9ae91cc

Reo-LEI requested a review from openinx September 5, 2021 06:13

openinx closed this Sep 8, 2021

openinx reopened this Sep 8, 2021

openinx changed the title ~~Flink: Transform INSERT as one DELETE following one INSERT~~ Flink: Add streaming upsert write option. Sep 8, 2021

kbendick reviewed Sep 8, 2021

View reviewed changes

Add unittest to cover upser mode check.

8a8a812

openinx closed this Sep 13, 2021

openinx reopened this Sep 13, 2021

openinx merged commit 3763952 into apache:master Sep 13, 2021

openinx mentioned this pull request Sep 13, 2021

Docs: Add document how to export records from CDC/Upsert Stream into apache iceberg table. #3105

Closed

Reo-LEI deleted the flink-upsert branch September 13, 2021 14:58

openinx mentioned this pull request Sep 14, 2021

after set 'write.upsert.enable'='true' in flink sql, using flink sql read iceberg table will get exception: java.lang.IllegalArgumentException: Row arity: 3, but serializer arity: 2 #3114

Closed

openinx mentioned this pull request Nov 9, 2021

Flink: Rename upsert table property #3504

Merged

openinx mentioned this pull request Nov 25, 2021

Duplicate records with primary key in iceberg 0.12.0 when use flinkSQL #3599

Closed

izchen pushed a commit to izchen/iceberg that referenced this pull request Dec 7, 2021

Flink: Add streaming upsert write option. (apache#2863)

497a8f5

Initial-neko pushed a commit to Initial-neko/iceberg that referenced this pull request Dec 17, 2021

Flink: Add streaming upsert write option. apache#2863(merged)

fbc9ce9

hameizi mentioned this pull request Feb 10, 2022

change delete logic #3834

Merged

t3hw mentioned this pull request Mar 15, 2026

Implement Iceberg Kafka Connect with Delta Writer Support in DV Mode for in-batch deduplication #14797

Open

		public static final String UPSERT_MODE_ENABLE = "write.upsert.enable";
		public static final boolean UPSERT_MODE_ENABLE_DEFAULT = false;

Conversation

Reo-LEI commented Jul 25, 2021

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Reo-LEI commented Jul 25, 2021

Uh oh!

Uh oh!

Uh oh!

coolderli commented Jul 28, 2021

Uh oh!

Reo-LEI commented Jul 28, 2021

Uh oh!

Reo-LEI commented Jul 31, 2021

Uh oh!

Reo-LEI commented Aug 23, 2021

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reo-LEI commented Sep 5, 2021

Uh oh!

openinx commented Sep 8, 2021

Uh oh!

openinx commented Sep 8, 2021

Uh oh!

kbendick left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reo-LEI Sep 9, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reo-LEI commented Sep 12, 2021

Uh oh!

openinx commented Sep 13, 2021

Uh oh!

openinx commented Sep 13, 2021

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

5 participants

Reo-LEI Sep 9, 2021 •

edited

Loading