HIVE-29322: Avoid TopNKeyOperator When ReduceSink TopNkey Filtering Provides Better Pruning for ORDER BY LIMIT Queries #6202
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -54,9 +54,11 @@ | |
| import org.apache.hadoop.hive.ql.exec.FunctionRegistry; | ||
| import org.apache.hadoop.hive.ql.exec.GroupByOperator; | ||
| import org.apache.hadoop.hive.ql.exec.JoinOperator; | ||
| import org.apache.hadoop.hive.ql.exec.LateralViewJoinOperator; | ||
| import org.apache.hadoop.hive.ql.exec.MapJoinOperator; | ||
| import org.apache.hadoop.hive.ql.exec.Operator; | ||
| import org.apache.hadoop.hive.ql.exec.OperatorUtils; | ||
| import org.apache.hadoop.hive.ql.exec.PTFOperator; | ||
| import org.apache.hadoop.hive.ql.exec.ReduceSinkOperator; | ||
| import org.apache.hadoop.hive.ql.exec.SelectOperator; | ||
| import org.apache.hadoop.hive.ql.exec.TableScanOperator; | ||
|
|
@@ -1300,9 +1302,10 @@ private static void runTopNKeyOptimization(OptimizeTezProcContext procCtx) | |
| return; | ||
| } | ||
|
|
||
| String topNKeyRegexPattern = buildTopNKeyRegexPattern(procCtx); | ||
| Map<SemanticRule, SemanticNodeProcessor> opRules = new LinkedHashMap<SemanticRule, SemanticNodeProcessor>(); | ||
| opRules.put( | ||
| new RuleRegExp("Top n key optimization", topNKeyRegexPattern), | ||
| new TopNKeyProcessor( | ||
| HiveConf.getIntVar(procCtx.conf, HiveConf.ConfVars.HIVE_MAX_TOPN_ALLOWED), | ||
| HiveConf.getFloatVar(procCtx.conf, ConfVars.HIVE_TOPN_EFFICIENCY_THRESHOLD), | ||
|
|
@@ -1322,6 +1325,49 @@ private static void runTopNKeyOptimization(OptimizeTezProcContext procCtx) | |
| ogw.startWalking(topNodes, null); | ||
| } | ||
|
|
||
| /* | ||
| * Build the ReduceSink matching pattern used by TopNKey optimization. | ||
| * | ||
| * For ORDER BY / LIMIT queries that do not involve GROUP BY or JOIN, | ||
| * applying TopNKey results in a performance regression. ReduceSink | ||
| * operators created only for ordering must therefore be excluded from | ||
| * TopNKey. | ||
| * | ||
| * When ORDER BY or LIMIT is present, restrict TopNKey to ReduceSink | ||
| * operators that originate from GROUP BY, JOIN, MAPJOIN, LATERAL VIEW | ||
| * JOIN or PTF query shapes | ||
| */ | ||
| private static String buildTopNKeyRegexPattern(OptimizeTezProcContext procCtx) { | ||
| String reduceSinkOp = ReduceSinkOperator.getOperatorName() + "%"; | ||
|
|
||
| boolean hasOrderOrLimit = | ||
| procCtx.parseContext.getQueryProperties().hasLimit() || | ||
| procCtx.parseContext.getQueryProperties().hasOrderBy(); | ||
|
Comment on lines
+1343
to
+1345
Member
There was a problem hiding this comment. Why do we need this? It is usually better if we can keep the optimization/transformation rules independent of the SQL syntax.
Contributor
Author
There was a problem hiding this comment. this is added for windowing queries to use topNkey Path - without group by / join in the query.
Member
There was a problem hiding this comment. select *
from ( select p_mfgr, rank() over(partition by p_mfgr order by p_name) r from part) a
where r < 4;Should such queries use the Plan A: With Top N Key OperatorPlan B: Without Top N Key OperatorThe plan structure is almost identical to the case of ORDER BY + LIMIT queries so from the discussion so far, I was under the impression that "Plan B" is better and more efficient in most cases.
Contributor
Author
There was a problem hiding this comment. I have debugged Windowing queries case and below is the observation. Testcase (unsorted data): INSERT INTO topnkey_windowing VALUES (NULL, NULL),(NULL, 109),('A', 109),('A', 104),('A', 110),('A', 120),('A', 103),('A', 109),('B', 105),('B', 106),('B', 106),('B', NULL),('B', 106),('A', 109); SELECT tw_code, ranking
I have tested with low-cardinality, monotonic-ordered windowing dataset and high-cardinality, multi-row-per-partition PTF test dataset. In this case, behaviour is similar to ORDER by.. Limit queries, where all the rows are forwarded. But query performance degradation is not observed for PTF operator case, comparing with disabling topnKeyoperator. From this experiments, with TopNkey enabled / disabled, performance is almost similar for Windowing queries.
Member
There was a problem hiding this comment. Many thanks for the additional experiments for the PTF case.
It's difficult to extract safe conclusions from the comparison between ORDER BY and windowing experiments cause dataset size and effective limit differ significantly. Dataset size:
Effective limit/top-n filter:
It would be great if you can run some experiments where the numbers are closer.
I still believe that this depends on the use-case. The example that you crafted for ORDER BY was clearly showing the downsides of the Top-N operator. The answer/benchmarks above imply that this will never happen for windowing functions but I don't understand why.
I don't fully understand the statement about the shuffle and reducer stages. Can you elaborate a bit more?
Contributor
Author
There was a problem hiding this comment.
Below are the number's with more data. With TopNkey enabled: With TopNkey disabled:
For PTF (windowing) queries, the shuffle and reducer stages are required to group rows by the PARTITION BY key before window functions can be evaluated. TopNKey operates per partition and, even when it forwards most or all rows, it does not introduce additional global shuffle or change the reducer fan-in. In contrast, for ORDER BY … LIMIT queries, TopNKey maintains a single global Top-N heap and can disable ReduceSink-level Top-N pruning; when input data is unsorted, this causes all rows to be shuffled globally, leading to severe performance degradation. |
||
|
|
||
| if (hasPTFReduceSink(procCtx) || !hasOrderOrLimit) { | ||
| return reduceSinkOp; | ||
| } | ||
|
|
||
| return "(" | ||
| + GroupByOperator.getOperatorName() + "|" | ||
| + PTFOperator.getOperatorName() + "|" | ||
| + JoinOperator.getOperatorName() + "|" | ||
| + MapJoinOperator.getOperatorName() + "|" | ||
| + LateralViewJoinOperator.getOperatorName() + "|" | ||
| + CommonMergeJoinOperator.getOperatorName() | ||
| + ").*%" | ||
| + reduceSinkOp; | ||
| } | ||
|
|
||
| private static boolean hasPTFReduceSink(OptimizeTezProcContext ctx) { | ||
| for (ReduceSinkOperator rs : ctx.visitedReduceSinks) { | ||
| if (rs.getConf().isPTFReduceSink()) { | ||
| return true; | ||
| } | ||
| } | ||
| return false; | ||
| } | ||
|
|
||
|
Comment on lines
+1362
to
+1370
Member
There was a problem hiding this comment. Not sure if this achieves the desired outcome. Basically, if there is a PTF RS anywhere in the plan we will apply the rule on every RS (no matter if it is PTF or not). Moreover, by relying on
Contributor
Author
There was a problem hiding this comment. For windowing queries, since there is not much performance issues with TopNKey enabled, currently making the queries to use TopNkey Path. But to match the plan, there is no sequence of PTF%RS% patterns for some queries. only RS% will work for this case.
Contributor
Author
There was a problem hiding this comment. is it a better solution to traverse the tree to find PTF ?
Member
There was a problem hiding this comment. I am not sure to what extend the claim that "there is not much performance issues with TopNKey enabled" holds. As mentioned in other comments first we have to confirm when/if TopNKey is efficient for PTF reducers and then decide to skip or not. |
||
| private boolean findParallelSemiJoinBranch(Operator<?> mapjoin, TableScanOperator bigTableTS, | ||
| ParseContext parseContext, | ||
| Map<ReduceSinkOperator, TableScanOperator> semijoins, | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I slightly suppose this level of complicated bailout should happen in TopNKeyProcessor. Most likely, can we skip adding a TopNKeyOperator when the RSO is not a PTFReduceSink and RSO's ancestors don't include RSO?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
i agree on this. @zabetak please provide your view on this. Thanks
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
As mentioned in https://github.com/apache/hive/pull/6202/changes#r2675925598 before talking about how to implement a change we need to understand what change we need to perform and if the change is needed.
The question if we should add or not a TopNKeyOperator below a windowing (PTF) ReduceSink remains open. Let's finalize the discussion in https://github.com/apache/hive/pull/6202/changes#r2668127719 and then we can can come back to this.