[HUDI-5338] Adjust coalesce behavior within NONE sort mode for bulk insert

[yihua]

Change Logs

Before this change, the NONE sort mode for bulk insert does coalesce for the input records or rows based on the shuffle parallelism of bulk insert (hoodie.bulkinsert.shuffle.parallelism) to reduce the parallelism. This could affect write latency if the cluster workers are not fully utilized due to reduced parallelism.

This PR adjusts NONE sort mode for bulk insert so that, by default, coalesce is not applied, matching the default parquet write behavior. The NONE sort mode still applies coalesce for clustering as the clustering operation relies on the bulk insert and the specified number of output Spark partitions to write a specific number of files.

New tests are added for the behavior change.

Impact

The removal of coalesce within NONE sort mode for bulk insert will reduce the write latency if the input parallelism is higher and the cluster workers are not fully utilized due to the lower shuffle parallelism of bulk insert.

For clustering, there is no behavior change, i.e., coalesce still happens in NONE sort mode for bulk insert in clustering.

Risk level

low

Documentation Update

HUDI-5339 for updating docs regarding the behavior change in NONE sort mode for bulk insert.

Contributor's checklist

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

[alexeykudinkin]

@yihua we also need to

• Review other impls that actually use parallelism hint (like GlobalSortPartitioner) and makes sure that we use max(config, input_parallelism)

[Zouxxyy]

@yihua @alexeykudinkin Can you have a review of this? #7372 (comment), I think they are somewhat related. Moreover, bulk insert is also used in cluster, and its parallelism is determined by cluser output files in one group.

[yihua]

@yihua we also need to

• Review other impls that actually use parallelism hint (like GlobalSortPartitioner) and makes sure that we use max(config, input_parallelism)

I created a ticket for the follow-up: HUDI-5360. Regarding the GlobalSortPartitioner and some other partitioners, we do want to reduce the parallelism instead of maximizing/keeping the input parallelism for file sizing, so that larger files can be created.

[yihua]

@yihua @alexeykudinkin Can you have a review of this? #7372 (comment), I think they are somewhat related.

Thanks for raising this. I'll check the PR.

Moreover, bulk insert is also used in cluster, and its parallelism is determined by cluser output files in one group.

Thanks for pointing this out. Yes, I also noticed this. I revised the PR so that the NONE sort mode can still respect the specified number of output partitions for clustering.

[yihua]

CI passes before rebasing. Affected tests also pass after rebasing. I'll merge this PR once Github actions pass.

[hudi-bot]

CI report:

• 637214c Azure: SUCCESS
• c5a9d2e Azure: PENDING

Bot commands

@hudi-bot supports the following commands:

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