Flink: flink read iceberg upsert data use streaming mode by hameizi · Pull Request #3095 · apache/iceberg

hameizi · 2021-09-10T02:46:36Z

Now, flink read iceberg use streaming mode ignore 'overwrite' snapshot, so user can't read the delete data real-time.
This PR first emit the eqdelete as delete rowdata(-U), and then emit the rowdata(+I) that join adddata and posdelete by function applyPosdelete. And then flink can keep one primary table from read iceberg primary table.

hameizi · 2021-09-10T02:47:42Z

@openinx @stevenzwu Could you help take a look?

openinx · 2021-09-13T11:47:56Z

Thanks for @hameizi for contribution, I will take a look tomorrow.

hameizi · 2021-10-18T08:33:58Z

@openinx Could you help take a look again?

openinx · 2021-10-18T11:43:52Z

Thanks for the work @hameizi , I think I will take a look at this tomorrow !

MOBIN-F · 2021-10-25T08:21:14Z

@hameizi
I merged this PR and did some testing and when I update a record, it actually inserts a new record

Flink SQL> CREATE TABLE iceberg_table_upsert8(

id BIGINT,
data STRING,
PRIMARY KEY (id) NOT ENFORCED
) WITH (
'connector'='iceberg',
'format-version' = '2',
'catalog-name'='hive_prod',
'catalog-type'='hive',
'write.upsert.enable'='true',
'uri'='thrift://localhost:9083',
'warehouse'='file:///tmp/iceberg/'
);

INSERT INTO iceberg_table_upsert8 VALUES(1,'qq');

INSERT INTO iceberg_table_upsert8 VALUES(1,'aa');

Flink SQL> SELECT * FROM iceberg_table_upsert8 /+ OPTIONS('streaming'='true', 'monitor-interval'='1s')/ ;
+----+----------------------+--------------------------------+
| op | id | data |
+----+----------------------+--------------------------------+
| -D | 1 | qq |
| +I | 1 | qq |
| -D | 1 | aa |
| +I | 1 | aa |

When I use Flink Batch query table ，it has tow records
Flink SQL> select * from iceberg_table_upsert8;
+----+----------------------+--------------------------------+
| op | id | data |
+----+----------------------+--------------------------------+
| +I | 1 | qq |
| +I | 1 | aa |
+----+----------------------+--------------------------------+
Received a total of 2 rows

hameizi · 2021-10-25T08:26:43Z

@MOBIN-F It's true, this PR translate update to -D and +I.

MOBIN-F · 2021-10-25T08:45:43Z

@MOBIN-F It's true, this PR translate update to -D and +I.

It will query the record(1,qq). but this record is from before the update, Is that right?

hameizi · 2021-10-25T08:48:24Z

@MOBIN-F It's true, this PR translate update to -D and +I.
It will query the record(1,qq). but this record is from before the update, Is that right?

If you don't query op of data, you will get only one result (1,aa).

xinbinhuang · 2021-11-03T19:21:18Z

Hi @hameizi, thanks for the great work! I'm really interested in this patch to support overwrite snapshots. In my company, 90% of the write to iceberg is overwrite by Spark, so it makes really hard/impossible to streaming read from these tables in Flink. I wonder if this patch requires Flink 1.13, or it'll also work for 1.12?

@openinx do you know if this PR can make it to the next release? if so, roughly when will be the next release or does it depends on the progress on the priority 1 items in the roadmap?

hameizi · 2021-11-04T02:07:16Z

I wonder if this patch requires Flink 1.13, or it'll also work for 1.12?

@xinbinhuang It work for both Flink 1.12 and 1.13.

Initial-neko · 2021-11-10T09:37:41Z

@MOBIN-F It's true, this PR translate update to -D and +I.
It will query the record(1,qq). but this record is from before the update, Is that right?
If you don't query op of data, you will get only one result (1,aa).
Sorry, I don't understand how to get only one result(1,aa). Could you tell me how to do that?

https://ci.apache.org/projects/flink/flink-docs-release-1.13/zh/docs/dev/table/sqlclient/

client mode can help you to do that

MOBIN-F · 2021-11-12T06:13:25Z

@MOBIN-F It's true, this PR translate update to -D and +I.
It will query the record(1,qq). but this record is from before the update, Is that right?
If you don't query op of data, you will get only one result (1,aa).
Sorry, I don't understand how to get only one result(1,aa). Could you tell me how to do that?
https://ci.apache.org/projects/flink/flink-docs-release-1.13/zh/docs/dev/table/sqlclient/
client mode can help you to do that

Thanks ,After testing, the PR is problematic, there will be duplicate data in batch read, not suitable for Retract stream and join scenario also have problems,
I'm trying to implement this feature according to the design documentation

hameizi · 2021-11-12T07:18:56Z

Thanks ,After testing, the PR is problematic, there will be duplicate data in batch read, not suitable for Upsert stream and join scenario also have problems,
I'm trying to implement this feature according to the design documentation

This PR is not relate to bath mode. And can you descript the detail of Upsert and join problem?

openinx · 2021-11-12T09:00:48Z

@hameizi , would you mind to resolve the conflicts ? I think I can take a look for the first round later.

hameizi · 2021-11-12T09:30:59Z

@hameizi , would you mind to resolve the conflicts ? I think I can take a look for the first round later.

@openinx I will resolve the conflicts , but mayby take some time because there is many change in flink moudle. When do you have time to review? today?

xianyouQ · 2022-01-19T07:22:34Z

I merged this PR and did some testing. i used flink cdc to consume binlog and wrote to iceberg table, then run some flink sql query(streaming) and compared with mysql original query , then the results were not matched.

After some debug I found 2 issues.

flink writing cdc data to iceberg will ignore UPDATE_BEFORE and treat UPDATE_AFTER as retract message, it's completely not correct. The relevant codes are as follows:

`

//org.apache.iceberg.flink.sink.BaseDeltaTaskWriter
switch (row.getRowKind()) {
case INSERT:
case UPDATE_AFTER:
if (upsert) {
writer.delete(row);
}
writer.write(row);
break;
case UPDATE_BEFORE:
if (upsert) {
break; // UPDATE_BEFORE is not necessary for UPDATE, we do nothing to prevent delete one row twice
}
writer.delete(row);
break;
case DELETE:
writer.delete(row);
break;
default:
throw new UnsupportedOperationException("Unknown row kind: " + row.getRowKind());
}

`

when there is only delete operation during the period of one snapshot，the snapshot of iceberg will contains only equality delete files and no data file(data file is deleted), flink will ignore equality delete files and miss all -D . function org.apache.iceberg.ManifestGroup.planStreamingFiles() create FileScanTask only when data file exists, so flink can not process equality delete files.

After fixed above 2 issues the test is passed.

hameizi · 2022-01-20T08:11:25Z

flink writing cdc data to iceberg will ignore UPDATE_BEFORE and treat UPDATE_AFTER as retract message, it's completely not correct. The relevant codes are as follows:

For this issue, config write.upsert.enabled is not suitable for CDC. So if your scene is CDC you can config it false that result upsert false.

2. when there is only delete operation during the period of one snapshot，the snapshot of iceberg will contains only equality delete files and no data file(data file is deleted), flink will ignore equality delete files and miss all -D . function org.apache.iceberg.ManifestGroup.planStreamingFiles() create FileScanTask only when data file exists, so flink can not process equality delete files.

I will do more test for this issue.
Thanks for your feedback.

xianyouQ · 2022-01-24T03:56:16Z

For this issue, config write.upsert.enabled is not suitable for CDC. So if your scene is CDC you can config it false that result upsert false.
Thanks , I also merged change-delete-logic , I think it's a better way to write delete rows. I am confused why delete row writing to postition delete file also write to equality delete file.

kingeasternsun · 2022-01-27T09:40:47Z

I merged this PR and did some testing. i used flink cdc to consume binlog and wrote to iceberg table, then run some flink sql query(streaming) and compared with mysql original query , then the results were not matched.
After some debug I found 2 issues.
flink writing cdc data to iceberg will ignore UPDATE_BEFORE and treat UPDATE_AFTER as retract message, it's completely not correct. The relevant codes are as follows:
`
//org.apache.iceberg.flink.sink.BaseDeltaTaskWriter
switch (row.getRowKind()) {
case INSERT:
case UPDATE_AFTER:
if (upsert) {
writer.delete(row);
}
writer.write(row);
break;
case UPDATE_BEFORE:
if (upsert) {
break; // UPDATE_BEFORE is not necessary for UPDATE, we do nothing to prevent delete one row twice
}
writer.delete(row);
break;
case DELETE:
writer.delete(row);
break;
default:
throw new UnsupportedOperationException("Unknown row kind: " + row.getRowKind());
}
`
when there is only delete operation during the period of one snapshot，the snapshot of iceberg will contains only equality delete files and no data file(data file is deleted), flink will ignore equality delete files and miss all -D . function org.apache.iceberg.ManifestGroup.planStreamingFiles() create FileScanTask only when data file exists, so flink can not process equality delete files.
After fixed above 2 issues the test is passed.

BaseDeltaTaskWriter

I think this code will be more clear if written like this

publicvoidwrite(RowDatarow) throwsIOException {
RowDataDeltaWriterwriter = route(row);
switch (row.getRowKind()) {
caseINSERT:
if (upsert) {
writer.delete(row);
}
writer.write(row);
break;
caseUPDATE_AFTER:
writer.write(row);
break;
caseUPDATE_BEFORE:
caseDELETE:
writer.delete(row);
break;
default:
thrownewUnsupportedOperationException("Unknown row kind: " + row.getRowKind());
}
}

liubo1022126 · 2022-02-18T03:22:37Z

@hameizi @openinx @rdblue hi，boys，We are currently using the flink upsert function. In which version is this PR planned to be released?

xianyouQ · 2022-03-16T07:41:13Z

Another issue, After run rewrite data file action and expire snapshot action(retain one Last snapshot), and then started a new flink streaming job, it did not read the old data. I checked metadata files , there was only one snapshot and all manifest file had a smaller sequenceNumber. what should i do if i want to read all data.

stevenzwu · 2022-04-19T04:04:45Z

+ * copied.
+ * @return a {@link CloseableIterable} of {@link FileScanTask}
+ */
+ public CloseableIterable<FileScanTask> planStreamingFiles() {


Streaming is an ambiguous name here. maybe planFilesForChangelog?

stevenzwu · 2022-04-19T04:05:16Z

+ * @return a table scan which can read append data for {@code snapshotId}
+ * exclusive and up to current snapshot inclusive
+ */
+ TableScan appendsCurrent(long snapshotId);


this change seems unnecessary

stevenzwu · 2022-04-19T04:06:26Z

+import org.apache.iceberg.util.PartitionUtil;
+
+@Internal
+public class StreamingRowDataFileScanTaskReader implements FileScanTaskReader<RowData> {


Similarly, maybe ChangelogRowDataFileScanTaskReader is more accurate

hameizi · 2022-04-19T05:45:13Z

@stevenzwu I will perfect this PR after #4580 finished, because there is some correlation between them.

GavinH1984 · 2022-05-16T08:03:16Z

@hameizi hihi, is there any plan for this PR? i saw the PR #4580 already get merge d into master,

hameizi · 2022-05-16T09:40:06Z

@hameizi hihi, is there any plan for this PR? i saw the PR #4580 already get merge d into master,

I will work on this these days.

iflytek-hmwang5 · 2022-08-02T03:06:03Z

@hameizi hihi, is there any plan for this PR? i saw the PR #4580 already get merge d into master,
I will work on this these days.
hameizi how about this PR?

Xiangakun · 2023-06-09T06:21:40Z

@hameizi @openinx Hello guys，what's the status of the issue now, please? Any blocking things? Thanks~

github-actions · 2024-07-20T00:13:08Z

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.

github-actions · 2024-07-28T00:14:43Z

This pull request has been closed due to lack of activity. This is not a judgement on the merit of the PR in any way. It is just a way of keeping the PR queue manageable. If you think that is incorrect, or the pull request requires review, you can revive the PR at any time.

github-actionsBot added API core data flink labels Sep 10, 2021

Reo-LEI mentioned this pull request Oct 21, 2021
Flink1.12.1 +Iceberg0.12.0 has problems with real-time reading and writing in upsert mode #3277
Closed

hameizi added 2 commits November 15, 2021 12:45

flink read iceberg upsert data use streaming mode
afba994

fix changelogmode
9889199

hameizi force-pushed the flink-read-upsert-streaming branch from 4391208 to 9889199Compare November 15, 2021 06:55

hameizi added 3 commits November 15, 2021 16:01

fix check code style
5613ccf

fix check code style
d201f70

fix
04f9965

hameizi mentioned this pull request Dec 9, 2021
Add watermark in flink table #3681
Closed

This was referenced Apr 18, 2022
iceberg v2 table changelog streaming read feature support #4528
Closed
API: Introduce a new IncrementalAppendScan interface #4580
Merged

stevenzwu reviewed Apr 19, 2022
View reviewed changes

hameizi closed this May 16, 2022

hameizi reopened this May 16, 2022

VidakM mentioned this pull request Mar 14, 2024
Flink SQL with Iceberg snapshots doesn't react if table has upsert #9948
Closed

github-actionsBot added the stale label Jul 20, 2024

github-actionsBot closed this Jul 28, 2024

Uh oh!

Conversation

hameizi commented Sep 10, 2021

Uh oh!

hameizi commented Sep 10, 2021

Uh oh!

openinx commented Sep 13, 2021

Uh oh!

hameizi commented Oct 18, 2021

Uh oh!

openinx commented Oct 18, 2021

Uh oh!

MOBIN-F commented Oct 25, 2021• edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

hameizi commented Oct 25, 2021

Uh oh!

MOBIN-F commented Oct 25, 2021

Uh oh!

hameizi commented Oct 25, 2021

Uh oh!

xinbinhuang commented Nov 3, 2021• edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

hameizi commented Nov 4, 2021

Uh oh!

Initial-neko commented Nov 10, 2021

Uh oh!

MOBIN-F commented Nov 12, 2021• edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

hameizi commented Nov 12, 2021

Uh oh!

openinx commented Nov 12, 2021

Uh oh!

hameizi commented Nov 12, 2021

Uh oh!

xianyouQ commented Jan 19, 2022

Uh oh!

hameizi commented Jan 20, 2022

Uh oh!

xianyouQ commented Jan 24, 2022

Uh oh!

kingeasternsun commented Jan 27, 2022

Uh oh!

liubo1022126 commented Feb 18, 2022

Uh oh!

xianyouQ commented Mar 16, 2022

Uh oh!

stevenzwuApr 19, 2022

Choose a reason for hiding this comment

Uh oh!

stevenzwuApr 19, 2022

Choose a reason for hiding this comment

Uh oh!

stevenzwuApr 19, 2022

Choose a reason for hiding this comment

Uh oh!

hameizi commented Apr 19, 2022

Uh oh!

GavinH1984 commented May 16, 2022

Uh oh!

hameizi commented May 16, 2022

Uh oh!

iflytek-hmwang5 commented Aug 2, 2022

Uh oh!

Xiangakun commented Jun 9, 2023

Uh oh!

github-actionsBot commented Jul 20, 2024

Uh oh!

github-actionsBot commented Jul 28, 2024

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

13 participants

MOBIN-F commented Oct 25, 2021•
edited
Loading

xinbinhuang commented Nov 3, 2021•
edited
Loading

MOBIN-F commented Nov 12, 2021•
edited
Loading