[HIVE-25975] Optimize ClusteredWriter for bucketed Iceberg tables

• Type: Improvement
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 4.0.0-alpha-1
• Labels: pull-request-available
• Assignee: Ádám Szita
• Reporter: Ádám Szita
• Created: 23/Feb/22 12:25
• Updated: 17/Nov/22 08:48
• Resolved: 04/Mar/22 11:28

Description

The first version of the ClusteredWriter in Hive-Iceberg will be lenient for bucketed tables: i.e. the records do not need to be ordered by the bucket values, the writer will just close its current file and open a new one for out-of-order records. 

This is suboptimal for the long-term due to creating many small files. Spark uses a UDF to compute the bucket value for each record and therefore it is able to order the records by bucket values, achieving optimal clustering.

The proposed change adds a new UDF that uses Iceberg's bucket transformation function to produce bucket values from constants or any column input. All types that Iceberg buckets support are supported in this UDF too, except for UUID.

This UDF is then used in SortedDynPartitionOptimizer to sort data during write if the target Iceberg target has bucket transform partitioning.

To enable this, Hive has been extended with the feature that allows storage handlers to define custom sorting expressions, to be passed to FileSink operator's DynPartContext during dynamic partitioning write scenarios.

The lenient version of ClusteredWriter in patched-iceberg-core has been disposed of as it is not needed anymore with this feature in.

Issue Links

• links to: GitHub Pull Request #3060