Restrict size of automatically broadcasted tables #11667

sopel39 · 2018-10-09T10:17:12Z

Add join-max-broadcast-table-size property

Property defines maximum size of a table
that can be broadcast for JOIN (0 disables the limit).

Supersedes #11570

sopel39 · 2018-10-09T10:21:44Z

I've done benchmarks with:

red <- current stats calculator with broadcast table size limit = 100MB
green <- broadcast limit disabled
blue <- upper estimate stats calculator with broadcast table size limit = 100MB

sopel39 · 2018-10-09T11:33:01Z

Observations:

upper estimate calculator does bring more problems than benefits
upper estimate calculator improves performance (vs disabled and current "probabilistic" stats calculator) for queries: tpcds/q22, tpcds/q72, tpcds/q95, tpcds/q76. This suggest that for those queries we are overestimating number of rows for build side.
using current "probabilistic" calculator with broadcast table size limit improves performance (vs no limit) in queries: tpch/q09, tpcds/q80, tpcds/q81, tpcds/q64, tpcds/q40, tpcds/q24_1, tpcds/q24_2
using current "probabilistic" calculator with broadcast table size limit degrades performance for some queries: tpch/q09, tpcds/q82, tpcds/q72, tpcds/q85, tpcds/q43, tpcds/q50, tpcds/q37.
Overall averaged improvement is 0.029x and weighted by query time is 0.17x (for individual query the value is from 1 to inf [if perf improved] or from -inf to -1 [if perf degraded]).
We greatly underestimate join build side construction, see query tpch/09. Will create separate issue for that

Conclusions:

I would land toggle for limiting broadcast build side (using "normal" stats calculator) as it slightly improves overall performance (based on TPCH/TPCDS benchmarks) but provides better cluster resource utilization (memory and concurrency).
It can however greatly improve performance for individual queries and prevent them from failing on memory limit. Such toggle is available in other engines (e.g: spark, Netizza) too.

Additional steps for the future:

Fix cost calculator to better account for join build side construction (related to: Quantify stats/cost calculator quality. Probabilistically adjust models as a longer term goal #11615)
Improve (if possible) stats calculator so that it doesn't under-account number of rows (see 2 from "observations")
Consider improving upper stats calculator as it helped with some queries
Take into account query memory limits holistically (https://github.com/prestodb/presto/pulls/findepi)

mbasmanova · 2018-10-10T19:21:32Z

presto-main/src/main/java/com/facebook/presto/SystemSessionProperties.java

@@ -149,6 +151,15 @@ public SystemSessionProperties(
                        false,
                        value -> JoinDistributionType.valueOf(((String) value).toUpperCase()),
                        JoinDistributionType::name),
+                new PropertyMetadata<>(
+                        JOIN_MAX_BROADCAST_TABLE_SIZE,
+                        "Maximum size of a table that can be broadcast for JOIN (0 disables the limit). The most pessimistic size estimate is taken when doing comparison.",


The most pessimistic size estimate is taken when doing comparison. - This doesn't seem to be the case, is it?

Can we make it nullable? So if the property is not set the limit is disabled. 0 rather implies that the broadcast should never be chosen.

0 disables the limit

It should be rather +INF to disable limit

The most pessimistic size estimate is taken when doing comparison.

I would prefer to not to go to such details.

Maximum size

Please update to Maximum estimated size

mbasmanova

@sopel39 Karol, I like this change. It is quite straightforward. When you refer to a similar option in Spark, are you thinking about spark.sql.autoBroadcastJoinThreshold ?

https://spark.apache.org/docs/latest/sql-programming-guide.html
Configures the maximum size in bytes for a table that will be broadcast to all worker nodes when performing a join. By setting this value to -1 broadcasting can be disabled. Note that currently statistics are only supported for Hive Metastore tables where the command ANALYZE TABLE COMPUTE STATISTICS noscan has been run.

arhimondr · 2018-10-10T21:55:27Z

presto-main/src/main/java/com/facebook/presto/SystemSessionProperties.java

@@ -149,6 +151,15 @@ public SystemSessionProperties(
                        false,
                        value -> JoinDistributionType.valueOf(((String) value).toUpperCase()),
                        JoinDistributionType::name),
+                new PropertyMetadata<>(
+                        JOIN_MAX_BROADCAST_TABLE_SIZE,
+                        "Maximum size of a table that can be broadcast for JOIN (0 disables the limit). The most pessimistic size estimate is taken when doing comparison.",


Can we make it nullable? So if the property is not set the limit is disabled. 0 rather implies that the broadcast should never be chosen.

arhimondr · 2018-10-10T21:56:38Z

presto-main/src/main/java/com/facebook/presto/sql/analyzer/FeaturesConfig.java

+    }
+
+    @Config("join-max-broadcast-table-size")
+    @MinDataSize("0MB")


It is extra. DataSize cannot be negative.

arhimondr · 2018-10-10T21:57:55Z

.../main/java/com/facebook/presto/sql/planner/iterative/rule/DetermineJoinDistributionType.java

+        double buildSideSizeInBytes = buildSideStatsEstimate.getOutputSizeInBytes(buildSide.getOutputSymbols(), context.getSymbolAllocator().getTypes());
+        JoinDistributionType joinDistributionType = getJoinDistributionType(context.getSession());
+        Optional<DataSize> joinMaxBroadcastTableSize = getJoinMaxBroadcastTableSize(context.getSession());
+        return joinDistributionType.canReplicate()


shortcircuit this check

arhimondr · 2018-10-10T21:58:10Z

.../main/java/com/facebook/presto/sql/planner/iterative/rule/DetermineJoinDistributionType.java

+        JoinDistributionType joinDistributionType = getJoinDistributionType(context.getSession());
+        Optional<DataSize> joinMaxBroadcastTableSize = getJoinMaxBroadcastTableSize(context.getSession());
+        return joinDistributionType.canReplicate()
+                && (!joinMaxBroadcastTableSize.isPresent() || buildSideSizeInBytes <= joinMaxBroadcastTableSize.get().toBytes());


Shortcircuit joinMaxBroadcastTableSize.isPresent() before doing anything else

arhimondr · 2018-10-10T22:00:45Z

...t/java/com/facebook/presto/sql/planner/iterative/rule/TestDetermineJoinDistributionType.java

+                        p.join(
+                                INNER,
+                                p.values(new PlanNodeId("valuesA"), aRows, p.symbol("A1", BIGINT)),
+                                p.filter(TRUE_LITERAL, p.values(new PlanNodeId("valuesB"), bRows, p.symbol("B1", BIGINT))),


Why not just p.values?

It's leftover from upper boundary stats calculator.

arhimondr · 2018-10-10T22:02:07Z

...t/java/com/facebook/presto/sql/planner/iterative/rule/TestDetermineJoinDistributionType.java

+                        p.join(
+                                INNER,
+                                p.values(new PlanNodeId("valuesA"), aRows, p.symbol("A1", BIGINT)),
+                                p.filter(TRUE_LITERAL, p.values(new PlanNodeId("valuesB"), bRows, p.symbol("B1", BIGINT))),


Ditto about filter

arhimondr · 2018-10-10T22:03:18Z

presto-main/src/test/java/com/facebook/presto/sql/planner/iterative/rule/TestReorderJoins.java

+                        p.join(
+                                INNER,
+                                p.values(new PlanNodeId("valuesA"), aRows, p.symbol("A1")),
+                                p.filter(TRUE_LITERAL, p.values(new PlanNodeId("valuesB"), bRows, p.symbol("B1"))),


arhimondr · 2018-10-10T22:03:24Z

presto-main/src/test/java/com/facebook/presto/sql/planner/iterative/rule/TestReorderJoins.java

+                        p.join(
+                                INNER,
+                                p.values(new PlanNodeId("valuesA"), aRows, p.symbol("A1")),
+                                p.filter(TRUE_LITERAL, p.values(new PlanNodeId("valuesB"), bRows, p.symbol("B1"))),


kokosing

% comments

kokosing · 2018-10-11T04:42:18Z

presto-main/src/main/java/com/facebook/presto/SystemSessionProperties.java

@@ -149,6 +151,15 @@ public SystemSessionProperties(
                        false,
                        value -> JoinDistributionType.valueOf(((String) value).toUpperCase()),
                        JoinDistributionType::name),
+                new PropertyMetadata<>(
+                        JOIN_MAX_BROADCAST_TABLE_SIZE,
+                        "Maximum size of a table that can be broadcast for JOIN (0 disables the limit). The most pessimistic size estimate is taken when doing comparison.",


0 disables the limit

It should be rather +INF to disable limit

presto-main/src/main/java/com/facebook/presto/SystemSessionProperties.java

kokosing · 2018-10-11T04:45:11Z

presto-main/src/main/java/com/facebook/presto/SystemSessionProperties.java

@@ -149,6 +151,15 @@ public SystemSessionProperties(
                        false,
                        value -> JoinDistributionType.valueOf(((String) value).toUpperCase()),
                        JoinDistributionType::name),
+                new PropertyMetadata<>(
+                        JOIN_MAX_BROADCAST_TABLE_SIZE,
+                        "Maximum size of a table that can be broadcast for JOIN (0 disables the limit). The most pessimistic size estimate is taken when doing comparison.",


The most pessimistic size estimate is taken when doing comparison.

I would prefer to not to go to such details.

Maximum size

Please update to Maximum estimated size

kokosing · 2018-10-11T04:52:15Z

.../main/java/com/facebook/presto/sql/planner/iterative/rule/DetermineJoinDistributionType.java

    private PlanNode getCostBasedJoin(JoinNode joinNode, Context context)
    {
        CostProvider costProvider = context.getCostProvider();
        List<PlanNodeWithCost> possibleJoinNodes = new ArrayList<>();

-        if (!mustPartition(joinNode)) {
+        if (!mustPartition(joinNode) && canReplicate(joinNode, context)) {
            possibleJoinNodes.add(getJoinNodeWithCost(costProvider, joinNode.withDistributionType(REPLICATED)));
        }
        if (!mustReplicate(joinNode, context)) {
            possibleJoinNodes.add(getJoinNodeWithCost(costProvider, joinNode.withDistributionType(PARTITIONED)));
        }


above two ifs can be extracted as method and reused below

Property defines maximum estimated size of a table that can be broadcast for JOIN. This is used to prevent CBO from broadcasting large tables which could saturate the cluster. Additionally, performance penality is small if CBO incorrectly chooses to broadcast small table. Related PR: prestodb#11667

sopel39 · 2018-10-11T14:42:57Z

@mbasmanova that's the one (autoBroadcastJoinThreshold)

Property defines maximum estimated size of a table that can be broadcast for JOIN. This is used to prevent CBO from broadcasting large tables which could saturate the cluster. Additionally, performance penality is small if CBO incorrectly chooses to broadcast small table. Related PR: #11667

facebook-github-bot added the CLA Signed label Oct 9, 2018

sopel39 mentioned this pull request Oct 9, 2018

Restrict size of automatically broadcasted tables #11570

Closed

sopel39 requested review from arhimondr, rschlussel-zz, findepi and kokosing October 9, 2018 11:32

sopel39 mentioned this pull request Oct 9, 2018

Join lookup source building cost is greatly underaccounted by CBO #11669

Closed

mbasmanova added correctness cbo and removed correctness labels Oct 9, 2018

mbasmanova mentioned this pull request Oct 9, 2018

CBO: Do not consider plans that exceed memory limits #11495

Closed

rschlussel requested review from rschlussel and removed request for rschlussel-zz October 9, 2018 17:52

sopel39 force-pushed the ks/join_max_broadcast_size branch from 044c6c1 to 60582a6 Compare October 10, 2018 07:57

mbasmanova reviewed Oct 10, 2018

View reviewed changes

mbasmanova approved these changes Oct 10, 2018

View reviewed changes

arhimondr reviewed Oct 10, 2018

View reviewed changes

kokosing approved these changes Oct 11, 2018

View reviewed changes

findepi self-assigned this Oct 11, 2018

Introduce assertDetermineJoinDistributionType

f0679c4

sopel39 force-pushed the ks/join_max_broadcast_size branch from 60582a6 to 634bfb9 Compare October 11, 2018 14:33

sopel39 added 2 commits October 11, 2018 16:34

Extract addJoinsWithDifferentDistributions method

0cd8d78

sopel39 force-pushed the ks/join_max_broadcast_size branch from 634bfb9 to 0cd8d78 Compare October 11, 2018 14:40

sopel39 merged commit b45c5a8 into prestodb:master Oct 11, 2018

sopel39 deleted the ks/join_max_broadcast_size branch October 12, 2018 10:25

sopel39 mentioned this pull request Oct 15, 2018

Avoid making wild guesses in FILTER and JOIN estimates #11625

Merged

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Restrict size of automatically broadcasted tables #11667

Restrict size of automatically broadcasted tables #11667

sopel39 commented Oct 9, 2018 •

edited

Loading

sopel39 commented Oct 9, 2018 •

edited

Loading

sopel39 commented Oct 9, 2018 •

edited

Loading

mbasmanova Oct 10, 2018

arhimondr Oct 10, 2018

kokosing Oct 11, 2018

kokosing Oct 11, 2018

mbasmanova left a comment

arhimondr Oct 10, 2018

arhimondr Oct 10, 2018

arhimondr Oct 10, 2018

arhimondr Oct 10, 2018

arhimondr Oct 10, 2018

sopel39 Oct 11, 2018

arhimondr Oct 10, 2018

arhimondr Oct 10, 2018

arhimondr Oct 10, 2018

kokosing left a comment

kokosing Oct 11, 2018

kokosing Oct 11, 2018

kokosing Oct 11, 2018

sopel39 commented Oct 11, 2018

Restrict size of automatically broadcasted tables #11667

Restrict size of automatically broadcasted tables #11667

Conversation

sopel39 commented Oct 9, 2018 • edited Loading

sopel39 commented Oct 9, 2018 • edited Loading

sopel39 commented Oct 9, 2018 • edited Loading

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

mbasmanova 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

kokosing 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

sopel39 commented Oct 11, 2018

sopel39 commented Oct 9, 2018 •

edited

Loading

sopel39 commented Oct 9, 2018 •

edited

Loading

sopel39 commented Oct 9, 2018 •

edited

Loading