[HUDI-8631] Fix the bug where the Flink table config hoodie.populate.meta.fields is not effective and optimize write performance

[usberkeley]

Change Logs

1. Fix the bug

• Resolve missing hoodie.populate.meta.fields in Table Config (hoodie.properties)
• Enhance validity checks

2. Optimize write performance

• Reduce unnecessary decoding overhead when disabling meta fields

Impact

Improve write performance. After optimization, the write speed with hoodie.populate.meta.fields=false is 42.9% faster than with hoodie.populate.meta.fields=true.

Testing method
Consume from the earliest position in Kafka until all messages are consumed (Kafka Lag = 0), and compare the time taken for both.

1）populate meta fields
time taken: 21hours and 25mins

2）no meta fields
time taken: 12hours and 14mins

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

medium

Documentation Update

none

Contributor's checklist

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

[usberkeley]

Dear Danny, Could you please take a look at this PR? @danny0405

Summary

1. When enable populateMetaFields, why the number of record key fields must be equal to one？

the number of record key fields must be equal to one, also known as KeyGeneratorType.SIMPLE. The Log Scanner needs to regenerate the Record Key from HoodieRecord.

2. Why not support KeyGeneratorType.COMPLEX in this PR?

The implementation classes for KeyGenerator are distributed across the hudi-client-common and hudi-spark-client packages, while The Log Scanner is in the hudi-common package. This makes it impossible to directly introduce the KeyGenerator implementation classes into the Log Scanner.

Another approach would be to pass the KeyGenerator implementation class object when constructing the Log Scanner. However, this would require modifying all modules that use the Log Scanner, which are over 30 instances. Some of these instances (such as Hudi Reader, which does not depend on the hudi-client-common package) cannot be implemented, and it would also increase the complexity of using the Log Scanner.

3. When and how to support KeyGeneratorType.COMPLEX in the future?

The next PR will support it by consolidating the KeyGenerator implementation classes from the hudi-client-common package (Flink) and the hudi-spark-client package (Spark) into the hudi-common package. I think that the KeyGenerator implementation classes are fundamental and should belong to the hudi-common package. This will also allow the Log Scanner to support KeyGeneratorType.COMPLEX when meta fields are disabled.

4. Why are allowOperationMetadataField allowed only when populateMetaFields is enabled?

1. Disabling populateMetaFields can reduce the performance overhead of decoding HoodieRecords. However, if allowOperationMetadataField is enabled, decoding performance is still affected even if populateMetaFields is disabled. Therefore, the impact of these two settings on performance is interconnected.
2. Both are metadata fields. populateMetaFields is the main switch, while allowOperationMetadataField just controls the activation of specific metadata fields. When the main switch is off, the sub-switches should have no effect.

5. In #11028, we already fixed the unnecessary rewrite when the schemas are exactly the same, is your benchmark based on the fix then?

Yes

[hudi-bot]

CI report:

• 0167459 Azure: SUCCESS

Bot commands

@hudi-bot supports the following commands:

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

[geserdugarov]

@usberkeley, @yihua , @danny0405
I would like to enquire about why we need to optimize any performance, when for Flink first thing to do is support of hoodie.populate.meta.fields. Please, correct me if I'm wrong.

hoodie.populate.meta.fields option meant to be used only for append mode, so we need to look at:

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

Lines 98 to 101 in 3cb874f

	if (OptionsResolver.isAppendMode(conf)) {
	// close compaction for append mode
	conf.set(FlinkOptions.COMPACTION_SCHEDULE_ENABLED, false);
	DataStream<Object> pipeline = Pipelines.append(conf, rowType, dataStream);

which will lead to

hudi/hudi-client/hudi-flink-client/src/main/java/org/apache/hudi/io/storage/row/HoodieRowDataCreateHandle.java

Line 140 in 3cb874f

fileWriter.writeRow(recordKey, rowData);

But at this stage file writer is already initialized with row type:

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

Line 95 in 3cb874f

this.rowType = preserveHoodieMetadata ? rowType : addMetadataFields(rowType, writeConfig.allowOperationMetadataField()); // patch up with metadata fields

So we can use row type without metadata columns, and write in append mode only initial Flink row data.

[geserdugarov]

@usberkeley, @yihua , @danny0405
Sorry to be difficult, but if I understand correctly, and this MR proposes optimization of Flink write by hoodie.populate.meta.fields = false. Then, I suppose, we should do it only, and not mix it with other optimizations.

For support of hoodie.populate.meta.fields = false in Flink, it's enough to implement changes like #12516.

By any chance, could this MR be refactoring and optimization everything related to hoodie.populate.meta.fields everywhere, but not related to support of hoodie.populate.meta.fields in Flink?

[geserdugarov]

@usberkeley, if you don't mind, it would be better to reopen and use already created HUDI-8308 for optimizations.

[usberkeley]

@usberkeley, if you don't mind, it would be better to reopen and use already created HUDI-8308 for optimizations.

Dear geserdugarov, look amazing.
Sure, as long as it helps the community, we'll choose the right solution

I've been busy with a major feature recently, so I'll take a closer look at your PR a bit later