trino icon indicating copy to clipboard operation
trino copied to clipboard

Add partitioning push down

Open dain opened this issue 1 year ago • 0 comments

Description

Add partitioning push down to table scan which a connector can use to activate optional partitioning, or choose between multiple partitioning strategies. This replaces the existing Metadata makeCompatiblePartitioning method used exclusively by Hive with a more generic applyPartitioning method.

Hive has been updated to the new system, and now only applies bucketed execution when it is actually used in the coordinator. This can improve performance when parallelism is limited by the bucketing and the bucketing isn't necessary for the query. Additionally, mismatched bucket execution (support join between tables where bucket count differes by a power of two) in Hive is activated by default. I believe this was disabled before, because we did not have the system to automatically disable bucket execution when the bucket count is small compred to the number of nodes.

Iceberg has been updated to support bucketed execution also. This applies the same optimizations available to Hive which allows the engine to eliminate unnecessary redistribution of tables. Additionally, since Iceberg supports multiple independent partitioning functions, a table can effectively have multiple distributions, which makes the optimization even more effective. Iceberg bucket execution can be controlled with the iceberg.bucket-execution configuration property and the bucket_execution_enabled session property.

Finally, for bucketed tables without a fixed node assignment, the connector can request a stable node distribution across queries. This implemented in Hive and Iceberg and improves cache hit rate for file system caching. The implementation is a simple Rendezvous Hashing (Highest Random Weight) algorithm.

Follup Work

  • Iceberg support for mismatched buckets

Release notes

( ) This is not user-visible or is docs only, and no release notes are required. ( ) Release notes are required. Please propose a release note for me. (X) Release notes are required, with the following suggested text:

# SPI
* Add partitioning push down, which a connector can use to activate optional partitioning, or choose between multiple partitioning strategies. ({issue}`23432`)

# Iceberg
* Add bucketed execution which can improve performance when running a join or aggregation on a bucketed table. This can be disabled with `iceberg.bucket-execution` configuration property, and the `bucket_execution_enabled` session property. ({issue}`23432`)

# Hive
* Bucket execution is now only enabled when actually useful in the query. ({issue}`23432`)
* Enable mismatched bucket execution optimization by default.  This can be disabled with `hive.optimize-mismatched-bucket-count` configuration property, and the `optimize_mismatched_bucket_count` session property. ({issue}`23432`)

dain avatar Sep 16 '24 04:09 dain