[HUDI-8920] Optimized SerDe costs of Flink write, simple bucket and non bucket cases

[geserdugarov]

Change Logs

Changes in Flink stream write into Hudi table with simple bucket index corresponding to #12697.

Benchmark description

Lineitem table from TPC-H benchmark was used. 60 mln rows, from which 20 mln rows are unique.

Perfomance estimation results

	current with Kryo	HoodieFlinkRecord	Optimization
Non bucket
Data passed, GB	43.9	29.3	33.6%
Total time, s	570	358	37.2%
Simple bucket index
Data passed, GB	19.4	13.6	29.9%
Total time, s	344	237	31.1%

Flink operators

Current with Kryo:

After switch to HoodieFlinkRecord:

Impact

Flink write performance improvement.

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

Low

Documentation Update

After merge

Contributor's checklist

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

[wombatu-kun]

PR's title says "simple bucket and non bucket cases"

[geserdugarov]

Missed it. Thanks! Fixed in 5a81536.

[geserdugarov]

I've added IT tests in 6ac8c9f, and also manually checked restore from Flink checkpoint for non bucket case. Restoring from checkpoint was successful.

[hudi-bot]

CI report:

• 466c013 UNKNOWN
• 08c9f31 Azure: SUCCESS

Bot commands

@hudi-bot supports the following commands:

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

[geserdugarov]

This draft PR contains exactly the same commits as here. But I've added one extra, which contains write.fast.mode turned on by default. CI is almost successful with only failed integration tests in test-flink (flink1.20, 1.11.3).
I will look on fails, and check it they are related to hardcoded use of HoodieRecord in tests, or currently not supported cases like consistent hashing and bounded context.