Add partitioning push down #23432

dain · 2024-09-16T04:02:12Z

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`)

sopel39 · 2024-09-26T08:56:28Z

plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveNodePartitioningProvider.java

@@ -99,7 +99,8 @@ public Optional<ConnectorBucketNodeMap> getBucketNodeMapping(ConnectorTransactio
    public ToIntFunction<ConnectorSplit> getSplitBucketFunction(
            ConnectorTransactionHandle transactionHandle,
            ConnectorSession session,
-            ConnectorPartitioningHandle partitioningHandle)
+            ConnectorPartitioningHandle partitioningHandle,
+            int bucketCount)
    {
        return value -> ((HiveSplit) value).getReadBucketNumber()


shouldn't this validate bucketCount that it matches table bucketing?

No, the ConnectorPartitioningHandle represents a logical partitioning, and is not specific to any table. This is a common misconception. Additionally, Hive supports mismatched bucket execution if the counts differ by a power of two.

Maybe we can add some of the above to the code comment in ConnectorPartitioningHandle ?

sopel39 · 2024-09-26T09:08:04Z

core/trino-main/src/main/java/io/trino/sql/planner/NodePartitioningManager.java

@@ -162,35 +164,40 @@ private NodePartitionMap getNodePartitioningMap(
        requireNonNull(partitioningHandle, "partitioningHandle is null");

        if (partitioningHandle.getConnectorHandle() instanceof SystemPartitioningHandle) {
-            return systemNodePartitionMap(session, partitioningHandle, systemPartitioningCache, partitionCount);
+            return new NodePartitionMap(systemBucketToNode(session, partitioningHandle, systemPartitioningCache, partitionCount), _ -> {


nit: getNodePartitioningMap is used in both scan and insert path, but currently NodePartitioningManager is making assumptions around partitionCount when it's absent. cc @marton-bod

I don't follow. This is no different than the behavior of the code we have today. Also all existing tests pass.

core/trino-main/src/main/java/io/trino/sql/planner/NodePartitioningManager.java

sopel39 · 2024-09-26T09:31:49Z

plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveNodePartitioningProvider.java

@@ -81,18 +75,7 @@ public Optional<ConnectorBucketNodeMap> getBucketNodeMapping(ConnectorTransactio
        if (!handle.isUsePartitionedBucketing()) {
            return Optional.of(createBucketNodeMap(handle.getBucketCount()));
        }
-
-        // Allocate a fixed number of buckets. Trino will assign consecutive buckets


@raunaqmorarka could you run insert benchmarks on this change? I thin kit should be fine as https://github.com/trinodb/trino/blob/master/core/trino-main/src/main/java/io/trino/sql/planner/NodePartitioningManager.java#L264 provides enough buckets for global and local parallelism.

The new code is just moving this decission to from the Hive connector to the core engine.

core/trino-main/src/main/java/io/trino/sql/planner/optimizations/AddExchanges.java

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergTableHandle.java

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergConfig.java

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergMetadata.java

sopel39 · 2024-10-07T10:00:01Z

core/trino-main/src/main/java/io/trino/sql/planner/optimizations/AddExchanges.java

@@ -677,9 +684,54 @@ public PlanWithProperties visitFilter(FilterNode node, PreferredProperties prefe
        @Override
        public PlanWithProperties visitTableScan(TableScanNode node, PreferredProperties preferredProperties)
        {
+            // attempt tp push preferred partitioning into the table scan


nit: tp -> to

sopel39 · 2024-10-07T10:10:06Z

core/trino-main/src/main/java/io/trino/sql/planner/optimizations/AddExchanges.java

+            return metadata.applyPartitioning(session, node.getTable(), partitioningProperties.getPartitioning().map(Partitioning::getHandle), partitionColumns)
+                    .map(node::withTableHandle)
+                    // Activate the partitioning if it passes the rules defined in DetermineTableScanNodePartitioning
+                    .map(newNode -> DetermineTableScanNodePartitioning.setUseConnectorNodePartitioning(


Why it doesn't return node in case setUseConnectorNodePartitioning doesn't set withUseConnectorNodePartitioning(true)?

I think that would be more complex, but also it does end up capturing the information that we could have used the partitioning, but the optimizer decided not to use it. I don't think this shows up in the output today, but it could be added.

sopel39 · 2024-10-08T09:22:02Z

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergConfig.java

+
+    @Config("iceberg.bucket-execution")
+    @ConfigDescription("Enable bucket-aware execution: use physical bucketing information to optimize queries")
+    public IcebergConfig setBucketExecutionEnabled(boolean bucketExecutionEnabled)


note that it won't work anyway because DetermineTableScanNodePartitioning uses:

int numberOfBuckets = bucketNodeMap.map(ConnectorBucketNodeMap::getBucketCount) .orElseGet(() -> nodePartitioningManager.getNodeCount(context.getSession(), partitioning.partitioningHandle()));

It's essential for connector to be able to provide bucket count to the engine.

If it's just scanning one partition, then it doesn't make sense to funnel that though single node (unless the partition is super small itself).

The iceberg and hive code both only handle the pushdown if hash bucketing is used. This is not used for value based.

I think you miss the point. CBO (DetermineTableScanNodePartitioning) decides whether to use connector provided bucketing or not. DetermineTableScanNodePartitioning won't use connector bucketing if it doesn't know how many "buckets" there are

Yes, but in AddExchanges we perform pushdown and if it works, we then directly call DetermineTableScanNodePartitioning so it can decide if we should actually use the bucketing from the connector. So both systems must decide that the partitioning should be used for execution.

@dain DetermineTableScanNodePartitioning will never choose to use Iceberg partitioning in current code, because it doesn't know partition count.

I assure it it does. When you have have a hash bucket function it uses that number. You can pull the branch and run it yourself.

When you have have a hash bucket function it uses that number

What is the number? Is it artificial or it actually represents how many partitions there are? What happens if there is only single partition scanned?

The code in io.trino.sql.planner.iterative.rule.DetermineTableScanNodePartitioning#apply is:

int numberOfBuckets = bucketNodeMap.map(ConnectorBucketNodeMap::getBucketCount) .orElseGet(() -> nodePartitioningManager.getNodeCount(context.getSession(), partitioning.getPartitioningHandle())); int numberOfTasks = max(taskCountEstimator.estimateSourceDistributedTaskCount(context.getSession()), 1); return Result.ofPlanNode(node .withUseConnectorNodePartitioning((double) numberOfBuckets / numberOfTasks >= getTableScanNodePartitioningMinBucketToTaskRatio(context.getSession())));

If the number of buckets is some arbitrary big number, but only single partition is scanned, then such query won't utilize the cluster properly (it will essentially utilize single node).

github-actions · 2024-11-08T17:02:50Z

This pull request has gone a while without any activity. Tagging the Trino developer relations team: @bitsondatadev @colebow @mosabua

mosabua · 2024-11-08T18:45:33Z

Adding stale-ignore and performance labels.. also cc @martint

Check boolean nullsAndAnyReplicated field before more complex fields

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

Add support for pushing plan partitioning into Iceberg when Iceberg tables use hash bucked partitioning. This enables co-located joins which can be significantly more efficient. Additionally, since Iceberg supports multiple independent partitioning functions, a table can effectively have multiple distributions, which makes the optimization more effective. This feature can be controlled with the iceberg.bucket-execution configuration property and the bucket_execution_enabled session property.

This improves cache hit rate for file system caching

findinpath · 2024-11-11T12:21:15Z

plugin/trino-iceberg/src/main/java/io/trino/plugin/iceberg/IcebergConfig.java

+        return bucketExecutionEnabled;
+    }
+
+    @Config("iceberg.bucket-execution")


I'd like to see a test in this commit highlighting actually what the 🥩 of the PR is actually about.

raunaqmorarka

Can we do the last 2 commits as a separate PR ? I think the rest can land already and it will potentially allow more ppl to review the iceberg bits

raunaqmorarka · 2024-11-12T15:59:50Z

core/trino-spi/src/main/java/io/trino/spi/connector/ConnectorNodePartitioningProvider.java

+     * the bucket will be assigned to a node chosen by the system.  The ConnectorPartitionHandle is defined
+     * declared in ConnectorTablePartitioning property of ConnectorTableProperties.


defined declared one of them is redundant

raunaqmorarka · 2024-11-19T05:28:49Z

.../java/io/trino/execution/scheduler/faulttolerant/FaultTolerantPartitioningSchemeFactory.java

@@ -86,7 +86,7 @@ private FaultTolerantPartitioningScheme create(PartitioningHandle partitioningHa
            if (connectorBucketNodeMap.isEmpty()) {
                return createSystemSchema(partitionCount.orElse(maxPartitionCount));
            }


buckedCount typo in commit message

raunaqmorarka · 2024-11-19T05:32:35Z

core/trino-spi/src/main/java/io/trino/spi/connector/ConnectorNodePartitioningProvider.java

+    default Optional<ConnectorBucketNodeMap> getBucketNodeMapping(
+            ConnectorTransactionHandle transactionHandle,
+            ConnectorSession session,
+            ConnectorPartitioningHandle partitioningHandle)


formatting change in separate commit ?

raunaqmorarka · 2024-11-19T05:38:24Z

core/trino-spi/src/main/java/io/trino/spi/connector/ConnectorPartitioningHandle.java

+ * <p>
+ * If the ConnectorPartitioningHandle of two tables are equal, the tables are guaranteed
+ * to have the same partitioning scheme across nodes, and the engine may use a colocated join.
+ */
 public interface ConnectorPartitioningHandle
 {
    default boolean isSingleNode()


Can we also add comments for isSingleNode and isCoordinatorOnly ?

raunaqmorarka · 2024-11-19T05:42:18Z

plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveNodePartitioningProvider.java

@@ -99,7 +99,8 @@ public Optional<ConnectorBucketNodeMap> getBucketNodeMapping(ConnectorTransactio
    public ToIntFunction<ConnectorSplit> getSplitBucketFunction(
            ConnectorTransactionHandle transactionHandle,
            ConnectorSession session,
-            ConnectorPartitioningHandle partitioningHandle)
+            ConnectorPartitioningHandle partitioningHandle,
+            int bucketCount)
    {
        return value -> ((HiveSplit) value).getReadBucketNumber()


Maybe we can add some of the above to the code comment in ConnectorPartitioningHandle ?

raunaqmorarka · 2024-11-19T06:18:56Z

core/trino-main/src/main/java/io/trino/sql/planner/optimizations/AddExchanges.java

+            List<ColumnHandle> partitionColumns;
+            if (partitioningProperties.getPartitioning().isPresent()) {
+                Partitioning partitioning = partitioningProperties.getPartitioning().get();
+                // constant partitioning values cannot be pushed into the table scan


When we have a combination of constants and partitioning columns, are we not able to take advantage of partitioning ?

raunaqmorarka · 2024-11-19T06:30:11Z

plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveTablePartitioning.java

 import static java.lang.String.format;
 import static java.util.Objects.requireNonNull;
 import static java.util.stream.Collectors.toList;

 /**
+ * @param active if not set partitioning is disabled for reading


comment seems a bit unclear, maybe If not set, partitioning is disabled for reading

raunaqmorarka · 2024-11-19T06:30:32Z

plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveTablePartitioning.java

 * @param tableBucketCount Number of buckets in the table, as specified in table metadata
- * @param readBucketCount Number of buckets the table will appear to have when the Hive connector presents the table to the engine for read.
+ * @param forWrite if true, the partitioning is being used for writing and cannot be changed


if -> If

raunaqmorarka · 2024-11-19T06:44:15Z

plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveConfig.java

@@ -904,7 +904,7 @@ public boolean isBucketExecutionEnabled()
    }

    @Config("hive.bucket-execution")
-    @ConfigDescription("Enable bucket-aware execution: only use a single worker per bucket")
+    @ConfigDescription("Enable bucket-aware execution: use physical bucketing information to optimize queries")


Should not be in this commit

raunaqmorarka · 2024-11-19T06:44:23Z

plugin/trino-hive/src/main/java/io/trino/plugin/hive/HiveSessionProperties.java

@@ -161,7 +161,7 @@ public HiveSessionProperties(
        sessionProperties = ImmutableList.of(
                booleanProperty(
                        BUCKET_EXECUTION_ENABLED,
-                        "Enable bucket-aware execution: only use a single worker per bucket",
+                        "Enable bucket-aware execution: use physical bucketing information to optimize queries",


Should not be in this commit

cla-bot bot added the cla-signed label Sep 16, 2024

github-actions bot added iceberg Iceberg connector hive Hive connector labels Sep 16, 2024

dain force-pushed the apply-partitioning branch 2 times, most recently from 2177896 to 1ebd691 Compare September 21, 2024 16:37

This was referenced Sep 25, 2024

[WIP] Add bucketed join support for iceberg connector (#79) #22206

Closed

Pass Iceberg partitioning information through getTableProperties #19062

Open

sopel39 reviewed Sep 26, 2024

View reviewed changes

dain force-pushed the apply-partitioning branch 8 times, most recently from 0ca7a63 to 5957f6f Compare October 2, 2024 23:05

dain force-pushed the apply-partitioning branch from 5957f6f to e7cb58b Compare October 6, 2024 01:22

dain changed the title ~~WIP Add partitioning push down~~ Add partitioning push down Oct 6, 2024

dain requested a review from electrum October 6, 2024 01:22

dain force-pushed the apply-partitioning branch from e7cb58b to 8c1a6d9 Compare October 6, 2024 01:27

sopel39 reviewed Oct 7, 2024

View reviewed changes

sopel39 reviewed Oct 8, 2024

View reviewed changes

github-actions bot added the stale label Nov 8, 2024

mosabua added performance stale-ignore Use this label on PRs that should be ignored by the stale bot so they are not flagged or closed. and removed stale labels Nov 8, 2024

dain added 2 commits November 10, 2024 16:41

Fix plan rendering in Hive test failure messages

e3dcc9f

Rename HiveBucketHandle to HiveTablePartitioning

23cc39e

dain added 10 commits November 10, 2024 16:52

Optimize ActualProperties compatible checks

1b9149e

Check boolean nullsAndAnyReplicated field before more complex fields

Improve connector partitioning JavaDocs

58ec143

Add buckedCount to getSplitBucketFunction

88d4578

Support partition functions with no bucket count preference

de4c6e6

Use automatic system bucket assignment in Hive node partitioning

5779539

Simplify FaultTolerantPartitioningSchemeFactory

931de87

Enable Hive mismatched bucked count optimization by default

e559520

Use stable node to bucket mapping for Hive and Iceberg

6904c4a

This improves cache hit rate for file system caching

dain force-pushed the apply-partitioning branch from 8c1a6d9 to 6904c4a Compare November 11, 2024 00:56

findinpath reviewed Nov 11, 2024

View reviewed changes

raunaqmorarka reviewed Nov 19, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add partitioning push down #23432

Add partitioning push down #23432

dain commented Sep 16, 2024 •

edited

Loading

sopel39 Sep 26, 2024

dain Oct 6, 2024

raunaqmorarka Nov 19, 2024

sopel39 Sep 26, 2024

dain Oct 6, 2024

sopel39 Sep 26, 2024

dain Oct 6, 2024

sopel39 Oct 7, 2024

raunaqmorarka Nov 19, 2024

sopel39 Oct 7, 2024

dain Oct 8, 2024

sopel39 Oct 8, 2024

dain Oct 11, 2024

sopel39 Oct 11, 2024

dain Oct 17, 2024

sopel39 Oct 18, 2024

dain Nov 11, 2024

sopel39 Nov 11, 2024

github-actions bot commented Nov 8, 2024

mosabua commented Nov 8, 2024

findinpath Nov 11, 2024

raunaqmorarka left a comment

raunaqmorarka Nov 12, 2024

raunaqmorarka Nov 19, 2024

raunaqmorarka Nov 19, 2024

raunaqmorarka Nov 19, 2024

raunaqmorarka Nov 19, 2024

raunaqmorarka Nov 19, 2024

raunaqmorarka Nov 19, 2024

raunaqmorarka Nov 19, 2024

raunaqmorarka Nov 19, 2024

raunaqmorarka Nov 19, 2024

		* the bucket will be assigned to a node chosen by the system. The ConnectorPartitionHandle is defined
		* declared in ConnectorTablePartitioning property of ConnectorTableProperties.

Add partitioning push down #23432

Are you sure you want to change the base?

Add partitioning push down #23432

Conversation

dain commented Sep 16, 2024 • edited Loading

Description

Follup Work

Release notes

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

github-actions bot commented Nov 8, 2024

mosabua commented Nov 8, 2024

Choose a reason for hiding this comment

raunaqmorarka left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dain commented Sep 16, 2024 •

edited

Loading