Record estimation stats during query optimization #22769
Conversation
feilong-liu
force-pushed
the
log_est
branch
2 times, most recently
from
031972e
to
0f55520
Compare
| @@ -99,6 +102,8 @@ public final class Session | |||
| private final OptimizerInformationCollector optimizerInformationCollector = new OptimizerInformationCollector(); | |||
| private final OptimizerResultCollector optimizerResultCollector = new OptimizerResultCollector(); | |||
| private final CTEInformationCollector cteInformationCollector = new CTEInformationCollector(); | |||
| private final Map<PlanNodeId, PlanNodeStatsEstimate> planNodeEstimateMap = new HashMap<>(); | |||
There was a problem hiding this comment.
Shouldn't this cache go to CachingCostProvider and CachingStatsProviderRespectfully? Even normal providers should benefit from this cache
There was a problem hiding this comment.
This will assume that plan node with the same node ID has the same estimation among different optimizers. I think this is a strong assumption which may not hold true. Currently we are constructing CachingStatsProvider in each optimizer and cache the stats within each optimizer itself.
There was a problem hiding this comment.
I think the session is a bit of a funny place for this (like it seems mostly that it's there because that's already getting passed around). Maybe have some special stats info collector that we pass in to the optimizer and get the stats from.
There was a problem hiding this comment.
I tried to create an info collector and pass through optimizers. However I found it to be a huge change, as all optimizers, the CachingStatsProvider and CachingCostProvider, and code creating these instances need to change.
We already have a few information collectors in the Session class, I am wondering if we can keep these maps here to avoid the huge interface changes.
| @@ -160,7 +160,7 @@ private StatsAndCosts computeStats(PlanNode root, TypeProvider types) | |||
| (node instanceof JoinNode) || (node instanceof SemiJoinNode)).matches()) { | |||
| StatsProvider statsProvider = new CachingStatsProvider(statsCalculator, session, types); | |||
| CostProvider costProvider = new CachingCostProvider(costCalculator, statsProvider, Optional.empty(), session); | |||
| return StatsAndCosts.create(root, statsProvider, costProvider); | |||
| return StatsAndCosts.create(root, statsProvider, costProvider, session, true); | |||
There was a problem hiding this comment.
Should this be properly gated with a session property in case it goes wrong?
There was a problem hiding this comment.
Sure, let me add one.
There was a problem hiding this comment.
Added a session property for it
| @@ -99,6 +102,8 @@ public final class Session | |||
| private final OptimizerInformationCollector optimizerInformationCollector = new OptimizerInformationCollector(); | |||
| private final OptimizerResultCollector optimizerResultCollector = new OptimizerResultCollector(); | |||
| private final CTEInformationCollector cteInformationCollector = new CTEInformationCollector(); | |||
| private final Map<PlanNodeId, PlanNodeStatsEstimate> planNodeEstimateMap = new HashMap<>(); | |||
There was a problem hiding this comment.
I think the session is a bit of a funny place for this (like it seems mostly that it's there because that's already getting passed around). Maybe have some special stats info collector that we pass in to the optimizer and get the stats from.
| { | ||
| Iterable<PlanNode> planIterator = Traverser.forTree(PlanNode::getSources) | ||
| .depthFirstPreOrder(root); | ||
| ImmutableMap.Builder<PlanNodeId, PlanNodeStatsEstimate> stats = ImmutableMap.builder(); | ||
| ImmutableMap.Builder<PlanNodeId, PlanCostEstimate> costs = ImmutableMap.builder(); | ||
| for (PlanNode node : planIterator) { | ||
| stats.put(node.getId(), statsProvider.getStats(node)); | ||
| costs.put(node.getId(), costProvider.getCost(node)); | ||
| stats.put(node.getId(), useCache ? session.getPlanNodeEstimateMap().getOrDefault(node.getId(), PlanNodeStatsEstimate.unknown()) : statsProvider.getStats(node)); |
There was a problem hiding this comment.
why not calculate them fresh if the stats aren't there like a loading cache would. two benefits:
- we won't need the on and off code paths just to disable the feature for tests.
- would prevent confusion if a node happened to get a new id after any cost-based optimizations ran, but it's otherwise something we would know the stats for.
There was a problem hiding this comment.
why not calculate them fresh if the stats aren't there like a loading cache would. two benefits:
The reason is because the calculation of stats will invalidate the recorded stats of child nodes.
Take the example in the description:
presto:tpch> explain (type distributed) select name, price from (select custkey, sum(totalprice) price from orders group by custkey) t join customer using(custkey);
>
--------------------------------------------------------------------------------------------------------------------------------------------------->
Fragment 0 [SINGLE] >
Output layout: [name, sum] >
Output partitioning: SINGLE [] >
Stage Execution Strategy: UNGROUPED_EXECUTION >
- Output[PlanNodeId 17][name, price] => [name:varchar(25), sum:double] >
price := sum (1:41) >
- RemoteSource[1] => [name:varchar(25), sum:double] >
>
Fragment 1 [HASH] >
Output layout: [name, sum] >
Output partitioning: SINGLE [] >
Stage Execution Strategy: UNGROUPED_EXECUTION >
- InnerJoin[PlanNodeId 366][("custkey_9" = "custkey")][$hashvalue, $hashvalue_30] => [name:varchar(25), sum:double] >
Estimates: {source: CostBasedSourceInfo, rows: 1,001 (31.29kB), cpu: 1,309,503.16, memory: 35,640.00, network: 335,820.00} >
Distribution: PARTITIONED >
- RemoteSource[2] => [custkey_9:bigint, name:varchar(25), $hashvalue:bigint] >
- Project[PlanNodeId 489][projectLocality = LOCAL] => [custkey:bigint, sum:double, $hashvalue_30:bigint] >
$hashvalue_30 := combine_hash(BIGINT'0', COALESCE($operator$hash_code(custkey), BIGINT'0')) (1:143) >
- Aggregate(FINAL)[custkey][PlanNodeId 3] => [custkey:bigint, sum:double] >
Estimates: {source: CostBasedSourceInfo, rows: 990 (61.88kB), cpu: ?, memory: ?, network: ?} >
sum := "presto.default.sum"((sum_25)) (1:69) >
- LocalExchange[PlanNodeId 446][HASH][$hashvalue_27] (custkey) => [custkey:bigint, sum_25:double, $hashvalue_27:bigint] >
- RemoteSource[3] => [custkey:bigint, sum_25:double, $hashvalue_28:bigint] >
>
Fragment 2 [SOURCE] >
Output layout: [custkey_9, name, $hashvalue_26] >
Output partitioning: HASH [custkey_9][$hashvalue_26] >
Stage Execution Strategy: UNGROUPED_EXECUTION >
- ScanProject[PlanNodeId 8,487][table = TableHandle {connectorId='hive', connectorHandle='HiveTableHandle{schemaName=tpch, tableName=customer,>
Estimates: {source: CostBasedSourceInfo, rows: 1,500 (60.06kB), cpu: ?, memory: ?, network: ?}/{source: CostBasedSourceInfo, rows: ? (>
$hashvalue_26 := combine_hash(BIGINT'0', COALESCE($operator$hash_code(custkey_9), BIGINT'0')) (1:128) >
LAYOUT: tpch.customer{} >
custkey_9 := custkey:bigint:0:REGULAR (1:128) >
name := name:varchar(25):1:REGULAR (1:128) >
>
Fragment 3 [SOURCE] >
Output layout: [custkey, sum_25, $hashvalue_29] >
Output partitioning: HASH [custkey][$hashvalue_29] >
Stage Execution Strategy: UNGROUPED_EXECUTION >
- Project[PlanNodeId 488][projectLocality = LOCAL] => [custkey:bigint, sum_25:double, $hashvalue_29:bigint] >
$hashvalue_29 := combine_hash(BIGINT'0', COALESCE($operator$hash_code(custkey), BIGINT'0')) (1:112) >
- Aggregate(PARTIAL)[custkey][PlanNodeId 450] => [custkey:bigint, sum_25:double] >
sum_25 := "presto.default.sum"((totalprice)) (1:69) >
- TableScan[PlanNodeId 0][TableHandle {connectorId='hive', connectorHandle='HiveTableHandle{schemaName=tpch, tableName=orders, analyze>
Estimates: {source: CostBasedSourceInfo, rows: 15,000 (395.51kB), cpu: ?, memory: ?, network: ?} >
LAYOUT: tpch.orders{} >
custkey := custkey:bigint:1:REGULAR (1:96) >
totalprice := totalprice:double:3:REGULAR (1:96) >
>
>
(1 row)
At the end, when we estimate the stats of the output node, it will try to estimate the stats of the join node, which then go to the aggregation node. CBO return empty stats for the final aggregation node, which will be recorded and invalidate the stats previously recorded for both join and aggregation nodes. Because we are currently recording the latest estimation stats in our cache.
I think one solution is to not always record the latest estimation, but to skip if the new stats is unknown. However, my concern is that this may skip some legit changes, for example if this happens not in the end but during optimizations.
Currently I found the biggest and only usage of the printed estimation stats is to help me to debug optimization behaviour, so I think keeping the stats used during optimization is more important. Having the stats provider to calculate the stats in the end, which is not the stats used during query optimization can be misleading and not so useful to me.
Sure, let me change the code. |
feilong-liu
force-pushed
the
log_est
branch
2 times, most recently
from
0911789
to
6bc07f7
Compare
feilong-liu
force-pushed
the
log_est
branch
2 times, most recently
from
f0bcc1d
to
cfc8bd9
Compare
presto-main/src/main/java/com/facebook/presto/SystemSessionProperties.java
Outdated
Show resolved
Hide resolved
| @@ -1915,6 +1916,11 @@ public SystemSessionProperties( | |||
| "Rewrite left join with is null check to semi join", | |||
| featuresConfig.isRewriteExpressionWithConstantVariable(), | |||
| false), | |||
| booleanProperty( | |||
| PRINT_ESTIMATION_STATS_FROM_CACHE, | |||
| "In the end of query optimization, print the estimation stats from cache populated during optimization instead of calculating from ground", | |||
There was a problem hiding this comment.
When printing estimated plan stats after optimization is complete, such as in an EXPLAIN query or for logging in a QueryCompletedEvent, get stats from a cache that was populated during query optimization rather than recalculating the stats on the final plan.
| @@ -77,15 +79,16 @@ public StatsAndCosts getForSubplan(PlanNode root) | |||
| return new StatsAndCosts(filteredStats.build(), filteredCosts.build()); | |||
| } | |||
|
|
|||
| public static StatsAndCosts create(PlanNode root, StatsProvider statsProvider, CostProvider costProvider) | |||
| public static StatsAndCosts create(PlanNode root, StatsProvider statsProvider, CostProvider costProvider, Session session, boolean useCache) | |||
There was a problem hiding this comment.
maybe add a comment that useCache should only be false for tests
There was a problem hiding this comment.
Yeah, the argument was added for plan tests only. Since we decided to add a session property to control the behaviour, I decided to remove this argument here, as it will be turned off though session property for the tests now.
When optimizer returns a optimized plan, it will also return the estimation of stats for each node with the plan, However, instead of returning the exact stats which are used in optimization, it's actually recalculating the stats. This can be a problem. For example, currently CBO returns empty stats if the aggregation step is not single for an aggregation This means that, we will not get any CBO stats for partial and final aggregation, and all other node which are downstream of the aggregation. In this PR, it will record the stats during query optimization. For the same node, later stats will override previous ones.
Description
When optimizer returns a optimized plan, it will also return the estimation of stats for each node with the plan,
presto/presto-main/src/main/java/com/facebook/presto/sql/Optimizer.java
Line 139 in 5240412
However, instead of returning the exact stats which are used in optimization, it's actually recalculating the stats.
This can be a problem. For example, currently CBO returns empty stats if the aggregation step is not single for an aggregation
presto/presto-main/src/main/java/com/facebook/presto/cost/AggregationStatsRule.java
Line 56 in 5240412
This means that, we will not get any CBO stats for partial and final aggregation, and all other node which are downstream of the aggregation.
In this PR, it will record the stats during query optimization. For the same node, later stats will override previous ones.
With this change, we can see estimations for aggregations and join
Without this change, we cannot see these stats:
Motivation and Context
To make the stats recorded more accurate.
Impact
To make the stats recorded more accurate, and easier to debug query optimization problem.
Test Plan
Existing unit tests.
Contributor checklist
Release Notes
Please follow release notes guidelines and fill in the release notes below.