[HUDI-5503] Optimize flink table factory option check #7608

hbgstc123 · 2023-01-05T03:56:15Z

Change Logs

1.remove pk and pre-combine field check for source and append mode sink

2.fallback to table config if pk or pre-combine field not set.

Impact

1.In flink sql, no need to set pk and pre-combine field when reading hudi or writing in append mode
2.If pk or pre-combine field not set in option, fallback to table config if possible

Risk level (write none, low medium or high below)

low

Contributor's checklist

Read through contributor's guide
Change Logs and Impact were stated clearly
Adequate tests were added if applicable
CI passed

hbgstc123 · 2023-01-05T10:44:51Z

@hudi-bot run azure

danny0405 · 2023-01-06T04:00:26Z

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/RowDataKeyGen.java

-    if (this.simpleRecordKey) {
+    if (!hasRecordKey) {
+      return DEFAULT_RECORD_KEY;
+    } else if (this.simpleRecordKey) {


Not sure whether we should use the empty string for the pk-less scenario, because all the records would have the same value primary key, which breaks the pk-less semantics, for pk-less, we actually mean all the records are unique, there is no need to define the primary key.

Another solution is to use the UUID as the primary key, WDYT ?

danny0405 · 2023-01-06T04:00:48Z

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/RowDataKeyGen.java

-    if (this.simpleRecordKey) {
+    if (!hasRecordKey) {
+      return DEFAULT_RECORD_KEY;
+    } else if (this.simpleRecordKey) {


Not sure whether we should use the empty string for the pk-less scenario, because all the records would have the same value primary key, which breaks the pk-less semantics, for pk-less, we actually mean all the records are unique, there is no need to define the primary key.

Another solution is to use the UUID as the primary key, WDYT ?

Not sure if remove the pk field will cause error somewhere, and write a identical value should use very low storage in columnar file format like parquet, and UUID will use much more space since its uniq so cannot compress well, and i don't know where we can use uuid, so i think maybe store a identical value for pk is better.

I change default key value to RowDataKeyGen.EMPTY_RECORDKEY_PLACEHOLDER since empty row key will report error.

In #7622, empty string is also used for keyless primary keys, so it's okey here if we reach an agreement and never uses the primary key.

danny0405 · 2023-01-06T04:02:07Z

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableFactory.java

    Configuration conf = FlinkOptions.fromMap(context.getCatalogTable().getOptions());
    ResolvedSchema schema = context.getCatalogTable().getResolvedSchema();
-    sanityCheck(conf, schema);
    setupConfOptions(conf, context.getObjectIdentifier(), context.getCatalogTable(), schema);


We can not remove the sanity check for MOR table source, the MOR table relies on the pk and preCombine key for merging payloads.

Oh i miss that pk field is used to emit delete data.
I add sanity check for stream read mor table.

Yeah, mering happens for mor table no matter whether the consuming is streaming or batch.

But as in the code of MergeOnReadInputFormat.MergeIterator.hasNext(), record key is read from meta field, and in constructor of AbstractHoodieLogRecordReader the preCombineField is read from table config.
So it seems that the passed in recordKey is only used to emit delete record when stream read mor log and preCombine field is not used at all when reading hudi table?

You are right

danny0405 · 2023-01-06T04:04:26Z

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableFactory.java

+    if (!conf.contains(FlinkOptions.RECORD_KEY_FIELD) && !schema.getPrimaryKey().isPresent()
+        && propsMap.containsKey(HoodieTableConfig.RECORDKEY_FIELDS.key())
+        && writeColumnNames.contains(propsMap.get(HoodieTableConfig.RECORDKEY_FIELDS.key()))) {
+      conf.set(FlinkOptions.RECORD_KEY_FIELD, propsMap.get(HoodieTableConfig.RECORDKEY_FIELDS.key()));


The writeColumnNames.contains does not work when there are multiple pk fields.

right, will fix this. thanks

hbgstc123 · 2023-01-09T02:28:24Z

@hudi-bot run azure

danny0405

I see there is a key-less support for Spark here: #7622, we can do similar changes in Flink side, so that in append mode, there is no need to define primary keys.

You can do similar work in this patch actually, but i guess the change in SimpleKeyGenerator is a blocker, we can merge this patch after #7622 is merged.

danny0405 · 2023-01-09T03:07:02Z

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/sink/bulk/RowDataKeyGen.java

-    if (this.simpleRecordKey) {
+    if (!hasRecordKey) {
+      return DEFAULT_RECORD_KEY;
+    } else if (this.simpleRecordKey) {


In #7622, empty string is also used for keyless primary keys, so it's okey here if we reach an agreement and never uses the primary key.

danny0405 · 2023-01-09T03:08:06Z

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableFactory.java

    Configuration conf = FlinkOptions.fromMap(context.getCatalogTable().getOptions());
    ResolvedSchema schema = context.getCatalogTable().getResolvedSchema();
-    sanityCheck(conf, schema);
    setupConfOptions(conf, context.getObjectIdentifier(), context.getCatalogTable(), schema);


Yeah, mering happens for mor table no matter whether the consuming is streaming or batch.

hbgstc123 · 2023-01-10T02:35:27Z

I see there is a key-less support for Spark here: #7622, we can do similar changes in Flink side, so that in append mode, there is no need to define primary keys.

You can do similar work in this patch actually, but i guess the change in SimpleKeyGenerator is a blocker, we can merge this patch after #7622 is merged.

ok, so lets wait for #7622 and do similar change for append only write.

danny0405 · 2023-01-11T09:37:34Z

In #7295 , we decide to fix the hoodie hive table by merging the table options for flink read/write path, which has some overlap with this PR, just a kind reminder.

danny0405 · 2023-01-28T05:48:52Z

Thanks for the contribution, I have reviewed and applied a patch:
HUDI-5503.patch.zip, please rebase with latest master code and then apply the path with cmd:
git apply xxx.patch

hbgstc123 · 2023-01-29T01:58:06Z

Thanks for the contribution, I have reviewed and applied a patch: HUDI-5503.patch.zip, please rebase with latest master code and then apply the path with cmd: git apply xxx.patch

Thanks for the patch, I applied with small changes: separate sanity check for source and sink(when create source only check pk if stream read mor table) and made corresponding change to TestHoodieTableFactory.java

danny0405 · 2023-01-29T03:28:33Z

hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/table/HoodieTableFactory.java

+    setupTableOptions(conf.getString(FlinkOptions.PATH), conf);
+    ResolvedSchema schema = context.getCatalogTable().getResolvedSchema();
+    sourceSanityCheck(conf, schema);
+    setupConfOptions(conf, context.getObjectIdentifier(), context.getCatalogTable(), schema);


The sanity check for source can be removed. When the primary key definition is missing, the streaming source for MOR table would distinguish the case as pk-less, so no deletes are emitted.

danny0405

+1, LGTM.

hudi-bot · 2023-01-29T12:36:27Z

CI report:

530e8a4 Azure: FAILURE

Bot commands

@hudi-bot supports the following commands:

@hudi-bot run azure re-run the last Azure build

Co-authored-by: hbg <[email protected]>

hbgstc123 force-pushed the optimize_flink_option_check branch from 0024bdd to 04a9f9c Compare January 5, 2023 08:02

hbgstc123 force-pushed the optimize_flink_option_check branch from 04a9f9c to 0765120 Compare January 5, 2023 12:24

danny0405 reviewed Jan 6, 2023

View reviewed changes

hbgstc123 force-pushed the optimize_flink_option_check branch 3 times, most recently from 158f8a9 to 3ca99dc Compare January 8, 2023 11:16

danny0405 requested changes Jan 9, 2023

View reviewed changes

danny0405 self-assigned this Jan 9, 2023

danny0405 added engine:flink Flink integration writer-core labels Jan 9, 2023

danny0405 changed the title ~~[HUDI-5503]Optimize flink table factory option check~~ [HUDI-5503] Optimize flink table factory option check Jan 28, 2023

hbgstc123 force-pushed the optimize_flink_option_check branch from 3ca99dc to 6feac73 Compare January 28, 2023 16:19

danny0405 reviewed Jan 29, 2023

View reviewed changes

danny0405 added the priority:critical Production degraded; pipelines stalled label Jan 29, 2023

[HUDI-5503]Optimize flink table factory option check

530e8a4

hbgstc123 force-pushed the optimize_flink_option_check branch from 6feac73 to 530e8a4 Compare January 29, 2023 08:46

danny0405 approved these changes Jan 29, 2023

View reviewed changes

danny0405 merged commit b00dac5 into apache:master Jan 29, 2023

voonhous mentioned this pull request Jan 30, 2023

[HUDI-5004] Allow nested field as primary key and preCombineField in flink sql #6915

Closed

4 tasks

yihua pushed a commit that referenced this pull request Jan 30, 2023

[HUDI-5503] Optimize flink table factory option check (#7608)

48977a6

Co-authored-by: hbg <[email protected]>

fengjian428 pushed a commit to fengjian428/hudi that referenced this pull request Jan 31, 2023

[HUDI-5503] Optimize flink table factory option check (apache#7608)

f17ee08

Co-authored-by: hbg <[email protected]>

danny0405 added priority:blocker Production down; release blocker and removed priority:critical Production degraded; pipelines stalled labels Feb 5, 2023

fengjian428 pushed a commit to fengjian428/hudi that referenced this pull request Apr 5, 2023

[HUDI-5503] Optimize flink table factory option check (apache#7608)

30db177

Co-authored-by: hbg <[email protected]>

[HUDI-5503] Optimize flink table factory option check #7608

[HUDI-5503] Optimize flink table factory option check #7608

Uh oh!

Conversation

hbgstc123 commented Jan 5, 2023

Change Logs

Impact

Risk level (write none, low medium or high below)

Contributor's checklist

Uh oh!

hbgstc123 commented Jan 5, 2023

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!

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!

hbgstc123 commented Jan 9, 2023

Uh oh!

danny0405 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!

hbgstc123 commented Jan 10, 2023

Uh oh!

danny0405 commented Jan 11, 2023

Uh oh!

danny0405 commented Jan 28, 2023

Uh oh!

hbgstc123 commented Jan 29, 2023 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

danny0405 Jan 29, 2023 • 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!

danny0405 left a comment

Choose a reason for hiding this comment

Uh oh!

hudi-bot commented Jan 29, 2023

CI report:

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

hbgstc123 commented Jan 29, 2023 •

edited

Loading

danny0405 Jan 29, 2023 •

edited

Loading