You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by "englefly (via GitHub)" <gi...@apache.org> on 2026/02/14 02:40:55 UTC
[PR] [feat](eager-agg)Eager agg 0214 [doris]
englefly opened a new pull request, #60757:
URL: https://github.com/apache/doris/pull/60757
### What problem does this PR solve?
Issue Number: close #xxx
Related PR: #xxx
Problem Summary:
### Release note
None
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [ ] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [ ] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [ ] No.
- [ ] Yes. <!-- Add document PR link here. eg: https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR should merge into -->
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](eager-agg)Eager agg 0214 [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #60757:
URL: https://github.com/apache/doris/pull/60757#issuecomment-3983060686
<details>
<summary>TPC-H: <b>Total hot run time: 27530 ms</b></summary>
```
machine: 'aliyun_ecs.c7a.8xlarge_32C64G'
scripts: https://github.com/apache/doris/tree/master/tools/tpch-tools
Tpch sf100 test result on commit 4fd41b6bde3a7c72efcd889b0042d38412de1f9b, data reload: false
------ Round 1 ----------------------------------
============================================
q1 17645 4563 4312 4312
q2 q3 10643 807 532 532
q4 4711 357 255 255
q5 7926 1222 1047 1047
q6 215 174 145 145
q7 834 850 660 660
q8 10430 1479 1357 1357
q9 6384 4772 4680 4680
q10 6484 1937 1652 1652
q11 488 252 236 236
q12 764 576 469 469
q13 18133 2941 2182 2182
q14 244 234 208 208
q15 957 785 807 785
q16 804 735 703 703
q17 763 915 403 403
q18 6014 5353 5262 5262
q19 1232 980 626 626
q20 512 492 391 391
q21 4475 1860 1375 1375
q22 348 287 250 250
Total cold run time: 100006 ms
Total hot run time: 27530 ms
----- Round 2, with runtime_filter_mode=off -----
============================================
q1 4497 4324 4347 4324
q2 q3 3855 4301 3770 3770
q4 848 1179 755 755
q5 4022 4365 4341 4341
q6 175 166 140 140
q7 1716 1611 1502 1502
q8 2416 2650 2691 2650
q9 7767 7386 7470 7386
q10 3712 3950 3571 3571
q11 524 434 411 411
q12 506 615 449 449
q13 2750 3366 2380 2380
q14 286 319 284 284
q15 897 840 788 788
q16 728 764 729 729
q17 1204 1397 1339 1339
q18 7101 6735 6660 6660
q19 953 909 956 909
q20 2106 2185 2029 2029
q21 4224 3928 3547 3547
q22 507 434 378 378
Total cold run time: 50794 ms
Total hot run time: 48342 ms
```
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](eager-agg)Eager agg 0214 [doris]
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #60757:
URL: https://github.com/apache/doris/pull/60757#issuecomment-3942436586
# FE UT Coverage Report
Increment line coverage ` 63.48% (412/649)` :tada:
[Increment coverage report](http://coverage.selectdb-in.cc/coverage/60757_742e5dc1a961547a7c51c084b61e93ee833a0d02/fe_increment_report/index.html)
[Complete coverage report](http://coverage.selectdb-in.cc/coverage/60757_742e5dc1a961547a7c51c084b61e93ee833a0d02/fe_report/index.html)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](eager-agg)stabilize and extend eager aggregation pushdown (join/outer join/case-if/count-if) with broad regression baseline refresh [doris]
Posted by "englefly (via GitHub)" <gi...@apache.org>.
englefly commented on code in PR #60757:
URL: https://github.com/apache/doris/pull/60757#discussion_r3393372966
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/PushDownAggregation.java:
##########
@@ -0,0 +1,319 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.rules.rewrite.eageraggregation;
+
+import org.apache.doris.nereids.jobs.JobContext;
+import org.apache.doris.nereids.rules.analysis.NormalizeAggregate;
+import org.apache.doris.nereids.rules.rewrite.AdjustNullable;
+import org.apache.doris.nereids.trees.expressions.CaseWhen;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.functions.Function;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Max;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Min;
+import org.apache.doris.nereids.trees.expressions.functions.agg.RollUpTrait;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.If;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.Nvl;
+import org.apache.doris.nereids.trees.expressions.literal.BigIntLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalUnion;
+import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.qe.SessionVariable;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * push down aggregation
+ */
+public class PushDownAggregation extends DefaultPlanRewriter<JobContext> implements CustomRewriter {
+ private static final Logger LOG = LoggerFactory.getLogger(PushDownAggregation.class);
+
+ public final EagerAggRewriter writer = new EagerAggRewriter();
+
+ private final Set<Class> pushDownAggFunctionSet = Sets.newHashSet(
+ Count.class,
+ Sum.class,
+ Max.class,
+ Min.class);
+
+ private final Set<Class> acceptNodeType = Sets.newHashSet(
+ LogicalUnion.class,
+ LogicalProject.class,
+ LogicalFilter.class,
+ LogicalRelation.class,
+ LogicalJoin.class);
+
+ @Override
+ public Plan rewriteRoot(Plan plan, JobContext jobContext) {
+ if (SessionVariable.isFeDebug()) {
+ try {
+ new AdjustNullable(false).rewriteRoot(plan, null);
+ } catch (Exception e) {
+ LOG.warn("(PushDownAggregation) input plan has nullable problem", e);
+ return plan;
+ }
+ }
+ int mode = SessionVariable.getEagerAggregationMode();
+ if (mode < 0) {
+ return plan;
+ } else {
+ Plan result = plan.accept(this, jobContext);
+ if (SessionVariable.isFeDebug()) {
+ result = new AdjustNullable(true).rewriteRoot(result, null);
+ }
+ return result;
+ }
+ }
+
+ @Override
+ public Plan visitLogicalAggregate(LogicalAggregate<? extends Plan> agg, JobContext context) {
+ Plan newChild = agg.child().accept(this, context);
+ if (newChild != agg.child()) {
+ return agg.withChildren(newChild);
+ }
+
+ if (agg.getSourceRepeat().isPresent()) {
+ return agg;
+ }
+
+ List<SlotReference> groupKeys = new ArrayList<>();
+ for (Expression groupKey : agg.getGroupByExpressions()) {
+ if (groupKey instanceof SlotReference) {
+ groupKeys.add((SlotReference) groupKey);
+ } else {
+ SessionVariable.throwAnalysisExceptionWhenFeDebug(
+ "PushDownAggregation failed: agg is not normalized\n "
+ + agg.treeString());
+ return agg;
+ }
+ }
+
+ Set<AggregateFunction> aggFunctions = Sets.newHashSet();
+ boolean hasDecomposedAggIf = false;
+ boolean hasCaseWhen = false;
+ Map<NamedExpression, List<AggregateFunction>> aggFunctionsForOutputExpressions = Maps.newHashMap();
+ for (NamedExpression aggOutput : agg.getOutputExpressions()) {
+ List<AggregateFunction> funcs = Lists.newArrayList();
+ aggFunctionsForOutputExpressions.put(aggOutput, funcs);
+ for (Object obj : aggOutput.collect(AggregateFunction.class::isInstance)) {
+ AggregateFunction aggFunction = (AggregateFunction) obj;
+ if (aggFunction.isDistinct()) {
+ return agg;
+ }
+ if (pushDownAggFunctionSet.contains(aggFunction.getClass())) {
+ // CaseWhen and If (which CASE WHEN is normalized into) must both be checked.
+ // When an agg function contains an If/CaseWhen whose condition tests IS NULL
+ // (e.g. count(if(col IS NULL, value, NULL))), pushing it to the nullable side
+ // of an outer join produces wrong results: null-extended rows make "col IS NULL"
+ // TRUE at the top level, but the pre-aggregated count slot becomes NULL after
+ // null-extension, and ifnull(sum(NULL), 0) = 0 instead of the correct 1.
+ if (!hasCaseWhen && aggFunction.anyMatch(e -> e instanceof CaseWhen || e instanceof If)) {
+ hasCaseWhen = true;
+ }
+ if (aggFunction.arity() > 0 && aggFunction.child(0) instanceof If
+ && !(aggFunction instanceof Count)) {
+ // Decompose Sum/Max/Min(If(cond, a, b)) into separate agg functions.
+ // Count(If(...)) is NOT decomposed here because the top-level
+ // replacement (Count->Sum rollup) cannot match the decomposed
+ // Count(a)/Count(b) as sub-expressions of the original Count(If(cond,a,b)).
+ // Count(If(...)) is pushed down as-is and rolled up normally.
+ If body = (If) (aggFunction).child(0);
+ Set<Slot> valueSlots = Sets.newHashSet(body.getTrueValue().getInputSlots());
+ valueSlots.addAll(body.getFalseValue().getInputSlots());
+ if (body.getCondition().getInputSlots().stream().anyMatch(s -> valueSlots.contains(s))) {
+ // do not push down sum(if a then a else b)
+ return agg;
+ }
Review Comment:
这个注释 的意思是如果 condition中的slot和value中的slot有交集就不下推
sum(if(x, x, y)) 为例，如果下推，就会产生 sum(x) group by x 的聚合，这个聚合没有意义
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](eager-agg)Eager agg 0214 [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #60757:
URL: https://github.com/apache/doris/pull/60757#issuecomment-3948404643
<details>
<summary>TPC-DS: <b>Total hot run time: 152941 ms</b></summary>
```
machine: 'aliyun_ecs.c7a.8xlarge_32C64G'
scripts: https://github.com/apache/doris/tree/master/tools/tpcds-tools
TPC-DS sf100 test result on commit d8703bf984891db2830e567b16976638d0b6c9b4, data reload: false
query5 4332 635 506 506
query6 314 233 219 219
query7 4213 474 270 270
query8 348 261 234 234
query9 8726 2754 2747 2747
query10 518 385 341 341
query11 7431 5864 5548 5548
query12 178 125 127 125
query13 1265 455 355 355
query14 6194 3794 3571 3571
query14_1 2862 2837 2877 2837
query15 204 194 175 175
query16 992 471 427 427
query17 1113 726 623 623
query18 2476 457 352 352
query19 219 209 185 185
query20 137 129 130 129
query21 224 178 120 120
query22 4791 5091 4882 4882
query23 16713 16010 15670 15670
query23_1 15828 15890 15867 15867
query24 7874 1707 1306 1306
query24_1 1290 1287 1308 1287
query25 580 473 416 416
query26 1237 255 142 142
query27 2780 471 285 285
query28 4500 1867 1858 1858
query29 793 566 477 477
query30 309 237 208 208
query31 1340 1299 1225 1225
query32 77 68 74 68
query33 533 353 271 271
query34 915 914 548 548
query35 644 665 611 611
query36 1062 1130 946 946
query37 137 92 83 83
query38 2994 2898 2904 2898
query39 894 905 853 853
query39_1 844 822 813 813
query40 227 153 134 134
query41 61 59 57 57
query42 300 297 301 297
query43 237 252 227 227
query44
query45 192 188 180 180
query46 876 970 600 600
query47 2095 2129 2033 2033
query48 312 314 231 231
query49 665 456 379 379
query50 664 279 218 218
query51 4093 4044 4111 4044
query52 286 296 278 278
query53 291 341 280 280
query54 286 262 253 253
query55 87 84 80 80
query56 310 306 311 306
query57 1348 1336 1274 1274
query58 286 273 266 266
query59 1330 1487 1316 1316
query60 344 343 315 315
query61 177 144 148 144
query62 620 584 531 531
query63 304 279 268 268
query64 4979 1279 983 983
query65
query66 1450 471 358 358
query67 16381 16336 16291 16291
query68
query69 382 311 284 284
query70 983 927 967 927
query71 333 300 290 290
query72 2829 2664 2502 2502
query73 535 545 331 331
query74 9976 9926 9829 9829
query75 2842 2762 2475 2475
query76 2325 1036 686 686
query77 373 405 318 318
query78 11239 11401 10714 10714
query79 1129 805 592 592
query80 717 649 562 562
query81 516 283 242 242
query82 1339 149 119 119
query83 379 282 253 253
query84 259 127 102 102
query85 933 572 522 522
query86 376 309 286 286
query87 3133 3076 2978 2978
query88 3515 2655 2634 2634
query89 423 374 342 342
query90 1933 195 170 170
query91 170 153 129 129
query92 83 73 69 69
query93 899 821 494 494
query94 453 319 292 292
query95 584 342 330 330
query96 636 510 226 226
query97 2490 2501 2424 2424
query98 232 222 213 213
query99 1019 996 865 865
Total cold run time: 232655 ms
Total hot run time: 152941 ms
```
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](eager-agg)Eager agg 0214 [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #60757:
URL: https://github.com/apache/doris/pull/60757#issuecomment-3900698464
<details>
<summary>TPC-DS: <b>Total hot run time: 152972 ms</b></summary>
```
machine: 'aliyun_ecs.c7a.8xlarge_32C64G'
scripts: https://github.com/apache/doris/tree/master/tools/tpcds-tools
TPC-DS sf100 test result on commit f90def55f7e6340e69de14ccf648f940ba0378c3, data reload: false
query5 4329 637 531 531
query6 317 225 203 203
query7 4227 478 270 270
query8 343 241 221 221
query9 8748 2771 2768 2768
query10 533 378 377 377
query11 7388 5866 5543 5543
query12 183 127 126 126
query13 1294 461 354 354
query14 5720 3818 3527 3527
query14_1 2773 2771 2820 2771
query15 206 197 179 179
query16 1025 481 471 471
query17 1110 729 624 624
query18 2443 442 358 358
query19 238 211 183 183
query20 136 134 131 131
query21 228 151 126 126
query22 4743 5049 4848 4848
query23 16875 15925 15889 15889
query23_1 15876 15848 15839 15839
query24 7710 1689 1329 1329
query24_1 1359 1302 1260 1260
query25 585 526 468 468
query26 1018 269 165 165
query27 2721 466 288 288
query28 4499 1890 1862 1862
query29 864 554 465 465
query30 321 247 215 215
query31 1360 1300 1208 1208
query32 82 74 69 69
query33 528 340 275 275
query34 924 897 576 576
query35 648 682 593 593
query36 1067 1146 917 917
query37 136 90 79 79
query38 2965 2871 2868 2868
query39 893 863 852 852
query39_1 829 828 836 828
query40 228 148 133 133
query41 63 59 58 58
query42 304 301 290 290
query43 240 246 220 220
query44
query45 195 194 177 177
query46 873 969 609 609
query47 2112 2166 2077 2077
query48 330 312 232 232
query49 632 461 378 378
query50 675 272 220 220
query51 4106 4078 4062 4062
query52 286 302 282 282
query53 290 334 294 294
query54 310 272 255 255
query55 136 90 82 82
query56 310 310 302 302
query57 1375 1345 1284 1284
query58 295 294 270 270
query59 1378 1436 1255 1255
query60 344 341 336 336
query61 148 145 146 145
query62 642 582 534 534
query63 302 277 280 277
query64 4742 1276 1015 1015
query65
query66 1410 456 358 358
query67 16535 16302 16314 16302
query68
query69 382 323 296 296
query70 939 980 987 980
query71 346 311 289 289
query72 2742 2649 2332 2332
query73 542 539 318 318
query74 10079 9967 9769 9769
query75 2844 2753 2433 2433
query76 2301 1042 667 667
query77 357 366 294 294
query78 11301 11406 10646 10646
query79 2570 813 600 600
query80 1726 601 534 534
query81 571 284 244 244
query82 1009 150 118 118
query83 330 262 253 253
query84 250 117 102 102
query85 917 471 425 425
query86 412 297 295 295
query87 3135 3093 2963 2963
query88 3539 2685 2670 2670
query89 417 362 337 337
query90 2023 182 165 165
query91 166 155 132 132
query92 78 74 74 74
query93 1132 813 516 516
query94 633 334 287 287
query95 599 391 303 303
query96 641 511 228 228
query97 2449 2476 2435 2435
query98 238 214 218 214
query99 996 986 926 926
Total cold run time: 235608 ms
Total hot run time: 152972 ms
```
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](eager-agg)stabilize and extend eager aggregation pushdown (join/outer join/case-if/count-if) with broad regression baseline refresh [doris]
Posted by "englefly (via GitHub)" <gi...@apache.org>.
englefly commented on code in PR #60757:
URL: https://github.com/apache/doris/pull/60757#discussion_r3392779919
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/PushDownAggregation.java:
##########
@@ -0,0 +1,319 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.rules.rewrite.eageraggregation;
+
+import org.apache.doris.nereids.jobs.JobContext;
+import org.apache.doris.nereids.rules.analysis.NormalizeAggregate;
+import org.apache.doris.nereids.rules.rewrite.AdjustNullable;
+import org.apache.doris.nereids.trees.expressions.CaseWhen;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.functions.Function;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Max;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Min;
+import org.apache.doris.nereids.trees.expressions.functions.agg.RollUpTrait;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.If;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.Nvl;
+import org.apache.doris.nereids.trees.expressions.literal.BigIntLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalUnion;
+import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.qe.SessionVariable;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * push down aggregation
+ */
+public class PushDownAggregation extends DefaultPlanRewriter<JobContext> implements CustomRewriter {
+ private static final Logger LOG = LoggerFactory.getLogger(PushDownAggregation.class);
+
+ public final EagerAggRewriter writer = new EagerAggRewriter();
+
+ private final Set<Class> pushDownAggFunctionSet = Sets.newHashSet(
+ Count.class,
+ Sum.class,
+ Max.class,
+ Min.class);
+
+ private final Set<Class> acceptNodeType = Sets.newHashSet(
+ LogicalUnion.class,
+ LogicalProject.class,
+ LogicalFilter.class,
+ LogicalRelation.class,
+ LogicalJoin.class);
+
+ @Override
+ public Plan rewriteRoot(Plan plan, JobContext jobContext) {
+ if (SessionVariable.isFeDebug()) {
+ try {
+ new AdjustNullable(false).rewriteRoot(plan, null);
+ } catch (Exception e) {
+ LOG.warn("(PushDownAggregation) input plan has nullable problem", e);
+ return plan;
+ }
+ }
+ int mode = SessionVariable.getEagerAggregationMode();
+ if (mode < 0) {
+ return plan;
+ } else {
+ Plan result = plan.accept(this, jobContext);
+ if (SessionVariable.isFeDebug()) {
Review Comment:
可以方便定位，通过日志可以确定是pushDownAggregation 之前的rule 有问题
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](eager-agg)Eager agg 0214 [doris]
Posted by "englefly (via GitHub)" <gi...@apache.org>.
englefly commented on PR #60757:
URL: https://github.com/apache/doris/pull/60757#issuecomment-3982522327
run buildall
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](eager-agg)stabilize and extend eager aggregation pushdown (join/outer join/case-if/count-if) with broad regression baseline refresh [doris]
Posted by "englefly (via GitHub)" <gi...@apache.org>.
englefly merged PR #60757:
URL: https://github.com/apache/doris/pull/60757
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](eager-agg)stabilize and extend eager aggregation pushdown (join/outer join/case-if/count-if) with broad regression baseline refresh [doris]
Posted by "englefly (via GitHub)" <gi...@apache.org>.
englefly commented on code in PR #60757:
URL: https://github.com/apache/doris/pull/60757#discussion_r3393791410
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/EagerAggRewriter.java:
##########
@@ -0,0 +1,625 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.rules.rewrite.eageraggregation;
+
+import org.apache.doris.nereids.rules.analysis.NormalizeAggregate;
+import org.apache.doris.nereids.rules.rewrite.StatsDerive;
+import org.apache.doris.nereids.stats.ExpressionEstimation;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.CaseWhen;
+import org.apache.doris.nereids.trees.expressions.Cast;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.If;
+import org.apache.doris.nereids.trees.plans.JoinType;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCatalogRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalUnion;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.types.DataType;
+import org.apache.doris.qe.SessionVariable;
+import org.apache.doris.statistics.ColumnStatistic;
+import org.apache.doris.statistics.Statistics;
+
+import com.google.common.collect.Lists;
+
+import java.util.ArrayList;
+import java.util.IdentityHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * eager aggregation
+ * agg[sum(t1.A) group by t1.B]
+ * ->join(t1.C=t2.D)
+ * ->T1(A, B, C)
+ * ->T2(D)
+ *
+ * =>
+ * agg[sum(x) group by t1.B]
+ * ->join(t1.C=t2.D)
+ * ->agg[sum(A) as x, group by B]
+ * ->T1(A, B, C)
+ * ->T2(D)
+ */
+public class EagerAggRewriter extends DefaultPlanRewriter<PushDownAggContext> {
+ private static final double LOWER_AGGREGATE_EFFECT_COEFFICIENT = 10000;
+ private static final double LOW_AGGREGATE_EFFECT_COEFFICIENT = 1000;
+ private static final double MEDIUM_AGGREGATE_EFFECT_COEFFICIENT = 100;
+ private final StatsDerive derive = new StatsDerive(false);
+
+ @Override
+ public Plan visitLogicalJoin(LogicalJoin<? extends Plan, ? extends Plan> join, PushDownAggContext context) {
+ boolean toLeft = false;
+ boolean toRight = false;
+ boolean pushHere = false;
+ if (context.getAggFunctions().isEmpty()) {
+ // select t1.v from t1 join t2 on t1.id = t2.id group by t1.v, t2.v
+ // if no agg function, try to push agg to the child which contains all group keys
+ // TODO: consider t1.rows/(t1.id, t1.v).ndv and t2.rows/(t2.id, t2.v).ndv to determine push target
+ if (join.left().getOutputSet().containsAll(context.getGroupKeys())) {
+ toLeft = true;
+ } else if (join.right().getOutputSet().containsAll(context.getGroupKeys())) {
+ toRight = true;
+ } else {
+ pushHere = true;
+ }
+ } else {
+ for (AggregateFunction aggFunc : context.getAggFunctions()) {
+ if (join.left().getOutputSet().containsAll(aggFunc.getInputSlots())) {
+ toLeft = true;
+ } else if (join.right().getOutputSet().containsAll(aggFunc.getInputSlots())) {
+ toRight = true;
+ } else {
+ pushHere = true;
+ }
+ }
+ }
+
+ if (pushHere || (toLeft && toRight)) {
Review Comment:
这里永远是false，应该去掉。
这是因为上面的判断逻辑改了，这里没有跟着改。最开始考虑一部分推左边，一部分推右边的场景，所以会出现toLeft = toRight = true.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](eager-agg)stabilize and extend eager aggregation pushdown (join/outer join/case-if/count-if) with broad regression baseline refresh [doris]
Posted by "englefly (via GitHub)" <gi...@apache.org>.
englefly commented on code in PR #60757:
URL: https://github.com/apache/doris/pull/60757#discussion_r3392779919
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/PushDownAggregation.java:
##########
@@ -0,0 +1,319 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.rules.rewrite.eageraggregation;
+
+import org.apache.doris.nereids.jobs.JobContext;
+import org.apache.doris.nereids.rules.analysis.NormalizeAggregate;
+import org.apache.doris.nereids.rules.rewrite.AdjustNullable;
+import org.apache.doris.nereids.trees.expressions.CaseWhen;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.functions.Function;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Max;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Min;
+import org.apache.doris.nereids.trees.expressions.functions.agg.RollUpTrait;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.If;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.Nvl;
+import org.apache.doris.nereids.trees.expressions.literal.BigIntLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalUnion;
+import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.qe.SessionVariable;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * push down aggregation
+ */
+public class PushDownAggregation extends DefaultPlanRewriter<JobContext> implements CustomRewriter {
+ private static final Logger LOG = LoggerFactory.getLogger(PushDownAggregation.class);
+
+ public final EagerAggRewriter writer = new EagerAggRewriter();
+
+ private final Set<Class> pushDownAggFunctionSet = Sets.newHashSet(
+ Count.class,
+ Sum.class,
+ Max.class,
+ Min.class);
+
+ private final Set<Class> acceptNodeType = Sets.newHashSet(
+ LogicalUnion.class,
+ LogicalProject.class,
+ LogicalFilter.class,
+ LogicalRelation.class,
+ LogicalJoin.class);
+
+ @Override
+ public Plan rewriteRoot(Plan plan, JobContext jobContext) {
+ if (SessionVariable.isFeDebug()) {
+ try {
+ new AdjustNullable(false).rewriteRoot(plan, null);
+ } catch (Exception e) {
+ LOG.warn("(PushDownAggregation) input plan has nullable problem", e);
+ return plan;
+ }
+ }
+ int mode = SessionVariable.getEagerAggregationMode();
+ if (mode < 0) {
+ return plan;
+ } else {
+ Plan result = plan.accept(this, jobContext);
+ if (SessionVariable.isFeDebug()) {
Review Comment:
可以方便定位，通过日志可以确定是pushDownAggregation 之前的rule 有问题
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/PushDownAggregation.java:
##########
@@ -0,0 +1,319 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.rules.rewrite.eageraggregation;
+
+import org.apache.doris.nereids.jobs.JobContext;
+import org.apache.doris.nereids.rules.analysis.NormalizeAggregate;
+import org.apache.doris.nereids.rules.rewrite.AdjustNullable;
+import org.apache.doris.nereids.trees.expressions.CaseWhen;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.functions.Function;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Max;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Min;
+import org.apache.doris.nereids.trees.expressions.functions.agg.RollUpTrait;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.If;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.Nvl;
+import org.apache.doris.nereids.trees.expressions.literal.BigIntLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalUnion;
+import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.qe.SessionVariable;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * push down aggregation
+ */
+public class PushDownAggregation extends DefaultPlanRewriter<JobContext> implements CustomRewriter {
+ private static final Logger LOG = LoggerFactory.getLogger(PushDownAggregation.class);
+
+ public final EagerAggRewriter writer = new EagerAggRewriter();
+
+ private final Set<Class> pushDownAggFunctionSet = Sets.newHashSet(
+ Count.class,
+ Sum.class,
+ Max.class,
+ Min.class);
+
+ private final Set<Class> acceptNodeType = Sets.newHashSet(
+ LogicalUnion.class,
+ LogicalProject.class,
+ LogicalFilter.class,
+ LogicalRelation.class,
+ LogicalJoin.class);
+
+ @Override
+ public Plan rewriteRoot(Plan plan, JobContext jobContext) {
+ if (SessionVariable.isFeDebug()) {
+ try {
+ new AdjustNullable(false).rewriteRoot(plan, null);
+ } catch (Exception e) {
+ LOG.warn("(PushDownAggregation) input plan has nullable problem", e);
+ return plan;
+ }
+ }
+ int mode = SessionVariable.getEagerAggregationMode();
+ if (mode < 0) {
+ return plan;
+ } else {
+ Plan result = plan.accept(this, jobContext);
+ if (SessionVariable.isFeDebug()) {
+ result = new AdjustNullable(true).rewriteRoot(result, null);
+ }
+ return result;
+ }
+ }
+
+ @Override
+ public Plan visitLogicalAggregate(LogicalAggregate<? extends Plan> agg, JobContext context) {
+ Plan newChild = agg.child().accept(this, context);
+ if (newChild != agg.child()) {
+ return agg.withChildren(newChild);
+ }
+
+ if (agg.getSourceRepeat().isPresent()) {
+ return agg;
+ }
+
+ List<SlotReference> groupKeys = new ArrayList<>();
+ for (Expression groupKey : agg.getGroupByExpressions()) {
+ if (groupKey instanceof SlotReference) {
+ groupKeys.add((SlotReference) groupKey);
+ } else {
+ SessionVariable.throwAnalysisExceptionWhenFeDebug(
+ "PushDownAggregation failed: agg is not normalized\n "
+ + agg.treeString());
+ return agg;
+ }
+ }
Review Comment:
可以方便定位，通过日志可以确定是pushDownAggregation 之前的rule 有问题
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](eager-agg)Eager agg 0214 [doris]
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #60757:
URL: https://github.com/apache/doris/pull/60757#issuecomment-3901294575
# FE Regression Coverage Report
Increment line coverage ` 74.58% (484/649)` :tada:
[Increment coverage report](http://coverage.selectdb-in.cc/coverage/60757_f90def55f7e6340e69de14ccf648f940ba0378c3_merge_fe/increment_report/index.html)
[Complete coverage report](http://coverage.selectdb-in.cc/coverage/60757_f90def55f7e6340e69de14ccf648f940ba0378c3_merge_fe/report/index.html)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](eager-agg)stabilize and extend eager aggregation pushdown (join/outer join/case-if/count-if) with broad regression baseline refresh [doris]
Posted by "englefly (via GitHub)" <gi...@apache.org>.
englefly commented on code in PR #60757:
URL: https://github.com/apache/doris/pull/60757#discussion_r3393791410
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/EagerAggRewriter.java:
##########
@@ -0,0 +1,625 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.rules.rewrite.eageraggregation;
+
+import org.apache.doris.nereids.rules.analysis.NormalizeAggregate;
+import org.apache.doris.nereids.rules.rewrite.StatsDerive;
+import org.apache.doris.nereids.stats.ExpressionEstimation;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.CaseWhen;
+import org.apache.doris.nereids.trees.expressions.Cast;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.If;
+import org.apache.doris.nereids.trees.plans.JoinType;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCatalogRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalUnion;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.types.DataType;
+import org.apache.doris.qe.SessionVariable;
+import org.apache.doris.statistics.ColumnStatistic;
+import org.apache.doris.statistics.Statistics;
+
+import com.google.common.collect.Lists;
+
+import java.util.ArrayList;
+import java.util.IdentityHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * eager aggregation
+ * agg[sum(t1.A) group by t1.B]
+ * ->join(t1.C=t2.D)
+ * ->T1(A, B, C)
+ * ->T2(D)
+ *
+ * =>
+ * agg[sum(x) group by t1.B]
+ * ->join(t1.C=t2.D)
+ * ->agg[sum(A) as x, group by B]
+ * ->T1(A, B, C)
+ * ->T2(D)
+ */
+public class EagerAggRewriter extends DefaultPlanRewriter<PushDownAggContext> {
+ private static final double LOWER_AGGREGATE_EFFECT_COEFFICIENT = 10000;
+ private static final double LOW_AGGREGATE_EFFECT_COEFFICIENT = 1000;
+ private static final double MEDIUM_AGGREGATE_EFFECT_COEFFICIENT = 100;
+ private final StatsDerive derive = new StatsDerive(false);
+
+ @Override
+ public Plan visitLogicalJoin(LogicalJoin<? extends Plan, ? extends Plan> join, PushDownAggContext context) {
+ boolean toLeft = false;
+ boolean toRight = false;
+ boolean pushHere = false;
+ if (context.getAggFunctions().isEmpty()) {
+ // select t1.v from t1 join t2 on t1.id = t2.id group by t1.v, t2.v
+ // if no agg function, try to push agg to the child which contains all group keys
+ // TODO: consider t1.rows/(t1.id, t1.v).ndv and t2.rows/(t2.id, t2.v).ndv to determine push target
+ if (join.left().getOutputSet().containsAll(context.getGroupKeys())) {
+ toLeft = true;
+ } else if (join.right().getOutputSet().containsAll(context.getGroupKeys())) {
+ toRight = true;
+ } else {
+ pushHere = true;
+ }
+ } else {
+ for (AggregateFunction aggFunc : context.getAggFunctions()) {
+ if (join.left().getOutputSet().containsAll(aggFunc.getInputSlots())) {
+ toLeft = true;
+ } else if (join.right().getOutputSet().containsAll(aggFunc.getInputSlots())) {
+ toRight = true;
+ } else {
+ pushHere = true;
+ }
+ }
+ }
+
+ if (pushHere || (toLeft && toRight)) {
Review Comment:
toLeft 和 toRight 的设置
```
for (AggregateFunction aggFunc : context.getAggFunctions()) {
if (join.left().getOutputSet().containsAll(aggFunc.getInputSlots())) {
toLeft = true;
} else if (join.right().getOutputSet().containsAll(aggFunc.getInputSlots())) {
toRight = true;
} else {
pushHere = true;
}
}
```
存在这种场景：context.getAggFunctions()= { sum(left_col), sum(right_col)}
于是 toLeft 和 toRight 都是true
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](eager-agg)Eager agg [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #60757:
URL: https://github.com/apache/doris/pull/60757#issuecomment-4002470488
PR approved by at least one committer and no changes requested.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](eager-agg)Eager agg [doris]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #60757:
URL: https://github.com/apache/doris/pull/60757#issuecomment-4002470725
PR approved by anyone and no changes requested.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](eager-agg)Eager agg 0214 [doris]
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #60757:
URL: https://github.com/apache/doris/pull/60757#issuecomment-3948856126
# FE Regression Coverage Report
Increment line coverage ` 74.58% (484/649)` :tada:
[Increment coverage report](http://coverage.selectdb-in.cc/coverage/60757_d8703bf984891db2830e567b16976638d0b6c9b4_merge_fe/increment_report/index.html)
[Complete coverage report](http://coverage.selectdb-in.cc/coverage/60757_d8703bf984891db2830e567b16976638d0b6c9b4_merge_fe/report/index.html)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](eager-agg)Eager agg 0214 [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #60757:
URL: https://github.com/apache/doris/pull/60757#issuecomment-3942248930
<details>
<summary>TPC-H: <b>Total hot run time: 27615 ms</b></summary>
```
machine: 'aliyun_ecs.c7a.8xlarge_32C64G'
scripts: https://github.com/apache/doris/tree/master/tools/tpch-tools
Tpch sf100 test result on commit 742e5dc1a961547a7c51c084b61e93ee833a0d02, data reload: false
------ Round 1 ----------------------------------
============================================
q1 17674 4467 4292 4292
q2 q3 10663 795 510 510
q4 4678 354 255 255
q5 7565 1188 1027 1027
q6 176 181 142 142
q7 787 845 662 662
q8 9302 1443 1298 1298
q9 4760 4714 4661 4661
q10 6298 1902 1670 1670
q11 490 267 241 241
q12 715 563 475 475
q13 18062 2902 2149 2149
q14 229 231 213 213
q15 934 793 801 793
q16 743 724 674 674
q17 700 844 415 415
q18 5919 5499 5309 5309
q19 1203 989 596 596
q20 512 488 380 380
q21 4436 1935 1581 1581
q22 352 336 272 272
Total cold run time: 96198 ms
Total hot run time: 27615 ms
----- Round 2, with runtime_filter_mode=off -----
============================================
q1 4639 4529 4581 4529
q2 q3 4008 4364 3824 3824
q4 862 1164 744 744
q5 4024 4387 4448 4387
q6 184 172 139 139
q7 1798 1604 1481 1481
q8 2443 2711 2539 2539
q9 7574 7446 7268 7268
q10 3760 3926 3589 3589
q11 530 456 429 429
q12 520 604 485 485
q13 2957 3205 2335 2335
q14 280 302 271 271
q15 882 812 795 795
q16 690 757 758 757
q17 1132 1440 1413 1413
q18 7288 6837 6699 6699
q19 860 877 953 877
q20 2053 2124 2029 2029
q21 3949 3482 3264 3264
q22 453 437 404 404
Total cold run time: 50886 ms
Total hot run time: 48258 ms
```
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](eager-agg)Eager agg 0214 [doris]
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #60757:
URL: https://github.com/apache/doris/pull/60757#issuecomment-3942765168
# FE Regression Coverage Report
Increment line coverage ` 74.58% (484/649)` :tada:
[Increment coverage report](http://coverage.selectdb-in.cc/coverage/60757_742e5dc1a961547a7c51c084b61e93ee833a0d02_merge_fe/increment_report/index.html)
[Complete coverage report](http://coverage.selectdb-in.cc/coverage/60757_742e5dc1a961547a7c51c084b61e93ee833a0d02_merge_fe/report/index.html)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](eager-agg)Eager agg 0214 [doris]
Posted by "englefly (via GitHub)" <gi...@apache.org>.
englefly commented on PR #60757:
URL: https://github.com/apache/doris/pull/60757#issuecomment-3948255941
run buildall
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](eager-agg)stabilize and extend eager aggregation pushdown (join/outer join/case-if/count-if) with broad regression baseline refresh [doris]
Posted by "englefly (via GitHub)" <gi...@apache.org>.
englefly commented on code in PR #60757:
URL: https://github.com/apache/doris/pull/60757#discussion_r3393679623
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/EagerAggRewriter.java:
##########
@@ -0,0 +1,625 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.rules.rewrite.eageraggregation;
+
+import org.apache.doris.nereids.rules.analysis.NormalizeAggregate;
+import org.apache.doris.nereids.rules.rewrite.StatsDerive;
+import org.apache.doris.nereids.stats.ExpressionEstimation;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.CaseWhen;
+import org.apache.doris.nereids.trees.expressions.Cast;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.If;
+import org.apache.doris.nereids.trees.plans.JoinType;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCatalogRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalUnion;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.types.DataType;
+import org.apache.doris.qe.SessionVariable;
+import org.apache.doris.statistics.ColumnStatistic;
+import org.apache.doris.statistics.Statistics;
+
+import com.google.common.collect.Lists;
+
+import java.util.ArrayList;
+import java.util.IdentityHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * eager aggregation
+ * agg[sum(t1.A) group by t1.B]
+ * ->join(t1.C=t2.D)
+ * ->T1(A, B, C)
+ * ->T2(D)
+ *
+ * =>
+ * agg[sum(x) group by t1.B]
+ * ->join(t1.C=t2.D)
+ * ->agg[sum(A) as x, group by B]
+ * ->T1(A, B, C)
+ * ->T2(D)
+ */
+public class EagerAggRewriter extends DefaultPlanRewriter<PushDownAggContext> {
+ private static final double LOWER_AGGREGATE_EFFECT_COEFFICIENT = 10000;
+ private static final double LOW_AGGREGATE_EFFECT_COEFFICIENT = 1000;
+ private static final double MEDIUM_AGGREGATE_EFFECT_COEFFICIENT = 100;
+ private final StatsDerive derive = new StatsDerive(false);
+
+ @Override
+ public Plan visitLogicalJoin(LogicalJoin<? extends Plan, ? extends Plan> join, PushDownAggContext context) {
+ boolean toLeft = false;
+ boolean toRight = false;
+ boolean pushHere = false;
+ if (context.getAggFunctions().isEmpty()) {
+ // select t1.v from t1 join t2 on t1.id = t2.id group by t1.v, t2.v
+ // if no agg function, try to push agg to the child which contains all group keys
+ // TODO: consider t1.rows/(t1.id, t1.v).ndv and t2.rows/(t2.id, t2.v).ndv to determine push target
+ if (join.left().getOutputSet().containsAll(context.getGroupKeys())) {
+ toLeft = true;
+ } else if (join.right().getOutputSet().containsAll(context.getGroupKeys())) {
+ toRight = true;
+ } else {
+ pushHere = true;
+ }
+ } else {
+ for (AggregateFunction aggFunc : context.getAggFunctions()) {
+ if (join.left().getOutputSet().containsAll(aggFunc.getInputSlots())) {
+ toLeft = true;
+ } else if (join.right().getOutputSet().containsAll(aggFunc.getInputSlots())) {
+ toRight = true;
+ } else {
+ pushHere = true;
+ }
Review Comment:
如果没有agg function，就按group key 来决定下推，这相当于agg function就是求 group key的distinct。
这个分支有agg function，就只需要下推agg function，下推后的group key 是join key
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](eager-agg)stabilize and extend eager aggregation pushdown (join/outer join/case-if/count-if) with broad regression baseline refresh [doris]
Posted by "englefly (via GitHub)" <gi...@apache.org>.
englefly commented on code in PR #60757:
URL: https://github.com/apache/doris/pull/60757#discussion_r3392837049
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/PushDownAggregation.java:
##########
@@ -0,0 +1,319 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.rules.rewrite.eageraggregation;
+
+import org.apache.doris.nereids.jobs.JobContext;
+import org.apache.doris.nereids.rules.analysis.NormalizeAggregate;
+import org.apache.doris.nereids.rules.rewrite.AdjustNullable;
+import org.apache.doris.nereids.trees.expressions.CaseWhen;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.functions.Function;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Max;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Min;
+import org.apache.doris.nereids.trees.expressions.functions.agg.RollUpTrait;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.If;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.Nvl;
+import org.apache.doris.nereids.trees.expressions.literal.BigIntLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalUnion;
+import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.qe.SessionVariable;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * push down aggregation
+ */
+public class PushDownAggregation extends DefaultPlanRewriter<JobContext> implements CustomRewriter {
+ private static final Logger LOG = LoggerFactory.getLogger(PushDownAggregation.class);
+
+ public final EagerAggRewriter writer = new EagerAggRewriter();
+
+ private final Set<Class> pushDownAggFunctionSet = Sets.newHashSet(
+ Count.class,
+ Sum.class,
+ Max.class,
+ Min.class);
+
+ private final Set<Class> acceptNodeType = Sets.newHashSet(
+ LogicalUnion.class,
+ LogicalProject.class,
+ LogicalFilter.class,
+ LogicalRelation.class,
+ LogicalJoin.class);
+
+ @Override
+ public Plan rewriteRoot(Plan plan, JobContext jobContext) {
+ if (SessionVariable.isFeDebug()) {
+ try {
+ new AdjustNullable(false).rewriteRoot(plan, null);
+ } catch (Exception e) {
+ LOG.warn("(PushDownAggregation) input plan has nullable problem", e);
+ return plan;
+ }
+ }
+ int mode = SessionVariable.getEagerAggregationMode();
+ if (mode < 0) {
+ return plan;
+ } else {
+ Plan result = plan.accept(this, jobContext);
+ if (SessionVariable.isFeDebug()) {
+ result = new AdjustNullable(true).rewriteRoot(result, null);
+ }
+ return result;
+ }
+ }
+
+ @Override
+ public Plan visitLogicalAggregate(LogicalAggregate<? extends Plan> agg, JobContext context) {
+ Plan newChild = agg.child().accept(this, context);
+ if (newChild != agg.child()) {
+ return agg.withChildren(newChild);
+ }
+
+ if (agg.getSourceRepeat().isPresent()) {
+ return agg;
+ }
+
+ List<SlotReference> groupKeys = new ArrayList<>();
+ for (Expression groupKey : agg.getGroupByExpressions()) {
+ if (groupKey instanceof SlotReference) {
+ groupKeys.add((SlotReference) groupKey);
+ } else {
+ SessionVariable.throwAnalysisExceptionWhenFeDebug(
+ "PushDownAggregation failed: agg is not normalized\n "
+ + agg.treeString());
+ return agg;
+ }
+ }
+
+ Set<AggregateFunction> aggFunctions = Sets.newHashSet();
+ boolean hasDecomposedAggIf = false;
+ boolean hasCaseWhen = false;
+ Map<NamedExpression, List<AggregateFunction>> aggFunctionsForOutputExpressions = Maps.newHashMap();
+ for (NamedExpression aggOutput : agg.getOutputExpressions()) {
+ List<AggregateFunction> funcs = Lists.newArrayList();
+ aggFunctionsForOutputExpressions.put(aggOutput, funcs);
+ for (Object obj : aggOutput.collect(AggregateFunction.class::isInstance)) {
+ AggregateFunction aggFunction = (AggregateFunction) obj;
+ if (aggFunction.isDistinct()) {
+ return agg;
+ }
+ if (pushDownAggFunctionSet.contains(aggFunction.getClass())) {
+ // CaseWhen and If (which CASE WHEN is normalized into) must both be checked.
+ // When an agg function contains an If/CaseWhen whose condition tests IS NULL
+ // (e.g. count(if(col IS NULL, value, NULL))), pushing it to the nullable side
+ // of an outer join produces wrong results: null-extended rows make "col IS NULL"
+ // TRUE at the top level, but the pre-aggregated count slot becomes NULL after
+ // null-extension, and ifnull(sum(NULL), 0) = 0 instead of the correct 1.
+ if (!hasCaseWhen && aggFunction.anyMatch(e -> e instanceof CaseWhen || e instanceof If)) {
+ hasCaseWhen = true;
+ }
Review Comment:
后面补充这些函数的支持
IF(cond, a, b) | 条件为真返回 a，否则返回 b
IFNULL(a, b) | a 不为 NULL 返回 a，否则返回 b（别名：NVL）
CASE WHEN | 多分支条件逻辑，类似 switch-case
COALESCE(a, b, ...) | 返回参数列表中第一个非 NULL 的值
NULLIF(a, b) | 若 a = b 返回 NULL，否则返回 a
LEAST(a, b, ...) | 返回多个参数中的最小值，任一为 NULL 则返回 NULL
GREATEST(a, b, ...) | 返回多个参数中的最大值，任一为 NULL 则返回 NULL
NULL_OR_EMPTY(str) | 字符串为 NULL 或空串时返回 true
NOT_NULL_OR_EMPTY(str) | 字符串既不为 NULL 也不为空串时返回 true
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](eager-agg)Eager agg 0214 [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #60757:
URL: https://github.com/apache/doris/pull/60757#issuecomment-3900663739
<details>
<summary>TPC-H: <b>Total hot run time: 27608 ms</b></summary>
```
machine: 'aliyun_ecs.c7a.8xlarge_32C64G'
scripts: https://github.com/apache/doris/tree/master/tools/tpch-tools
Tpch sf100 test result on commit f90def55f7e6340e69de14ccf648f940ba0378c3, data reload: false
------ Round 1 ----------------------------------
============================================
q1 17676 4511 4329 4329
q2 q3 10643 785 510 510
q4 4682 359 249 249
q5 7557 1242 1014 1014
q6 177 175 146 146
q7 780 840 669 669
q8 9752 1500 1332 1332
q9 5280 4783 4753 4753
q10 6324 1922 1668 1668
q11 471 268 232 232
q12 815 576 468 468
q13 18066 2941 2172 2172
q14 235 230 217 217
q15 954 802 795 795
q16 791 727 674 674
q17 711 861 413 413
q18 6009 5490 5268 5268
q19 1508 998 637 637
q20 491 485 401 401
q21 4475 1837 1419 1419
q22 337 296 242 242
Total cold run time: 97734 ms
Total hot run time: 27608 ms
----- Round 2, with runtime_filter_mode=off -----
============================================
q1 4558 4341 4551 4341
q2 q3 3866 4268 3753 3753
q4 850 1152 768 768
q5 4008 4331 4335 4331
q6 182 173 147 147
q7 1721 1619 1491 1491
q8 2440 2655 2528 2528
q9 7988 7518 7376 7376
q10 3764 3989 3571 3571
q11 519 442 410 410
q12 478 612 491 491
q13 2740 3334 2324 2324
q14 289 333 290 290
q15 906 803 793 793
q16 742 791 703 703
q17 1169 1438 1330 1330
q18 7195 6885 6866 6866
q19 886 918 945 918
q20 2057 2155 2131 2131
q21 3967 3516 3302 3302
q22 462 459 416 416
Total cold run time: 50787 ms
Total hot run time: 48280 ms
```
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](eager-agg)Eager agg 0214 [doris]
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #60757:
URL: https://github.com/apache/doris/pull/60757#issuecomment-3900701928
# FE UT Coverage Report
Increment line coverage ` 63.48% (412/649)` :tada:
[Increment coverage report](http://coverage.selectdb-in.cc/coverage/60757_f90def55f7e6340e69de14ccf648f940ba0378c3/fe_increment_report/index.html)
[Complete coverage report](http://coverage.selectdb-in.cc/coverage/60757_f90def55f7e6340e69de14ccf648f940ba0378c3/fe_report/index.html)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](eager-agg)Eager agg 0214 [doris]
Posted by "englefly (via GitHub)" <gi...@apache.org>.
englefly commented on PR #60757:
URL: https://github.com/apache/doris/pull/60757#issuecomment-3942179475
run buildall
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](eager-agg)Eager agg 0214 [doris]
Posted by "Thearas (via GitHub)" <gi...@apache.org>.
Thearas commented on PR #60757:
URL: https://github.com/apache/doris/pull/60757#issuecomment-3900510399
Thank you for your contribution to Apache Doris.
Don't know what should be done next? See [How to process your PR](https://cwiki.apache.org/confluence/display/DORIS/How+to+process+your+PR).
Please clearly describe your PR:
1. What problem was fixed (it's best to include specific error reporting information). How it was fixed.
2. Which behaviors were modified. What was the previous behavior, what is it now, why was it modified, and what possible impacts might there be.
3. What features were added. Why was this function added?
4. Which code was refactored and why was this part of the code refactored?
5. Which functions were optimized and what is the difference before and after the optimization?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](eager-agg)Eager agg 0214 [doris]
Posted by "englefly (via GitHub)" <gi...@apache.org>.
englefly commented on PR #60757:
URL: https://github.com/apache/doris/pull/60757#issuecomment-3900510863
run buildall
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](eager-agg)stabilize and extend eager aggregation pushdown (join/outer join/case-if/count-if) with broad regression baseline refresh [doris]
Posted by "englefly (via GitHub)" <gi...@apache.org>.
englefly commented on code in PR #60757:
URL: https://github.com/apache/doris/pull/60757#discussion_r3393372966
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/PushDownAggregation.java:
##########
@@ -0,0 +1,319 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.rules.rewrite.eageraggregation;
+
+import org.apache.doris.nereids.jobs.JobContext;
+import org.apache.doris.nereids.rules.analysis.NormalizeAggregate;
+import org.apache.doris.nereids.rules.rewrite.AdjustNullable;
+import org.apache.doris.nereids.trees.expressions.CaseWhen;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.functions.Function;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Max;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Min;
+import org.apache.doris.nereids.trees.expressions.functions.agg.RollUpTrait;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.If;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.Nvl;
+import org.apache.doris.nereids.trees.expressions.literal.BigIntLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalUnion;
+import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.qe.SessionVariable;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * push down aggregation
+ */
+public class PushDownAggregation extends DefaultPlanRewriter<JobContext> implements CustomRewriter {
+ private static final Logger LOG = LoggerFactory.getLogger(PushDownAggregation.class);
+
+ public final EagerAggRewriter writer = new EagerAggRewriter();
+
+ private final Set<Class> pushDownAggFunctionSet = Sets.newHashSet(
+ Count.class,
+ Sum.class,
+ Max.class,
+ Min.class);
+
+ private final Set<Class> acceptNodeType = Sets.newHashSet(
+ LogicalUnion.class,
+ LogicalProject.class,
+ LogicalFilter.class,
+ LogicalRelation.class,
+ LogicalJoin.class);
+
+ @Override
+ public Plan rewriteRoot(Plan plan, JobContext jobContext) {
+ if (SessionVariable.isFeDebug()) {
+ try {
+ new AdjustNullable(false).rewriteRoot(plan, null);
+ } catch (Exception e) {
+ LOG.warn("(PushDownAggregation) input plan has nullable problem", e);
+ return plan;
+ }
+ }
+ int mode = SessionVariable.getEagerAggregationMode();
+ if (mode < 0) {
+ return plan;
+ } else {
+ Plan result = plan.accept(this, jobContext);
+ if (SessionVariable.isFeDebug()) {
+ result = new AdjustNullable(true).rewriteRoot(result, null);
+ }
+ return result;
+ }
+ }
+
+ @Override
+ public Plan visitLogicalAggregate(LogicalAggregate<? extends Plan> agg, JobContext context) {
+ Plan newChild = agg.child().accept(this, context);
+ if (newChild != agg.child()) {
+ return agg.withChildren(newChild);
+ }
+
+ if (agg.getSourceRepeat().isPresent()) {
+ return agg;
+ }
+
+ List<SlotReference> groupKeys = new ArrayList<>();
+ for (Expression groupKey : agg.getGroupByExpressions()) {
+ if (groupKey instanceof SlotReference) {
+ groupKeys.add((SlotReference) groupKey);
+ } else {
+ SessionVariable.throwAnalysisExceptionWhenFeDebug(
+ "PushDownAggregation failed: agg is not normalized\n "
+ + agg.treeString());
+ return agg;
+ }
+ }
+
+ Set<AggregateFunction> aggFunctions = Sets.newHashSet();
+ boolean hasDecomposedAggIf = false;
+ boolean hasCaseWhen = false;
+ Map<NamedExpression, List<AggregateFunction>> aggFunctionsForOutputExpressions = Maps.newHashMap();
+ for (NamedExpression aggOutput : agg.getOutputExpressions()) {
+ List<AggregateFunction> funcs = Lists.newArrayList();
+ aggFunctionsForOutputExpressions.put(aggOutput, funcs);
+ for (Object obj : aggOutput.collect(AggregateFunction.class::isInstance)) {
+ AggregateFunction aggFunction = (AggregateFunction) obj;
+ if (aggFunction.isDistinct()) {
+ return agg;
+ }
+ if (pushDownAggFunctionSet.contains(aggFunction.getClass())) {
+ // CaseWhen and If (which CASE WHEN is normalized into) must both be checked.
+ // When an agg function contains an If/CaseWhen whose condition tests IS NULL
+ // (e.g. count(if(col IS NULL, value, NULL))), pushing it to the nullable side
+ // of an outer join produces wrong results: null-extended rows make "col IS NULL"
+ // TRUE at the top level, but the pre-aggregated count slot becomes NULL after
+ // null-extension, and ifnull(sum(NULL), 0) = 0 instead of the correct 1.
+ if (!hasCaseWhen && aggFunction.anyMatch(e -> e instanceof CaseWhen || e instanceof If)) {
+ hasCaseWhen = true;
+ }
+ if (aggFunction.arity() > 0 && aggFunction.child(0) instanceof If
+ && !(aggFunction instanceof Count)) {
+ // Decompose Sum/Max/Min(If(cond, a, b)) into separate agg functions.
+ // Count(If(...)) is NOT decomposed here because the top-level
+ // replacement (Count->Sum rollup) cannot match the decomposed
+ // Count(a)/Count(b) as sub-expressions of the original Count(If(cond,a,b)).
+ // Count(If(...)) is pushed down as-is and rolled up normally.
+ If body = (If) (aggFunction).child(0);
+ Set<Slot> valueSlots = Sets.newHashSet(body.getTrueValue().getInputSlots());
+ valueSlots.addAll(body.getFalseValue().getInputSlots());
+ if (body.getCondition().getInputSlots().stream().anyMatch(s -> valueSlots.contains(s))) {
+ // do not push down sum(if a then a else b)
+ return agg;
+ }
Review Comment:
这个注释 的意思是如果 condition中的slot和value中的slot有交集就不下推
sum(if(x, x, y)) 为例，如果下推，就会产生 sum(x) group by x 的聚合，这个聚合没有意义
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](eager-agg)stabilize and extend eager aggregation pushdown (join/outer join/case-if/count-if) with broad regression baseline refresh [doris]
Posted by "englefly (via GitHub)" <gi...@apache.org>.
englefly commented on code in PR #60757:
URL: https://github.com/apache/doris/pull/60757#discussion_r3393791410
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/EagerAggRewriter.java:
##########
@@ -0,0 +1,625 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.rules.rewrite.eageraggregation;
+
+import org.apache.doris.nereids.rules.analysis.NormalizeAggregate;
+import org.apache.doris.nereids.rules.rewrite.StatsDerive;
+import org.apache.doris.nereids.stats.ExpressionEstimation;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.CaseWhen;
+import org.apache.doris.nereids.trees.expressions.Cast;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.If;
+import org.apache.doris.nereids.trees.plans.JoinType;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCatalogRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalUnion;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.types.DataType;
+import org.apache.doris.qe.SessionVariable;
+import org.apache.doris.statistics.ColumnStatistic;
+import org.apache.doris.statistics.Statistics;
+
+import com.google.common.collect.Lists;
+
+import java.util.ArrayList;
+import java.util.IdentityHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * eager aggregation
+ * agg[sum(t1.A) group by t1.B]
+ * ->join(t1.C=t2.D)
+ * ->T1(A, B, C)
+ * ->T2(D)
+ *
+ * =>
+ * agg[sum(x) group by t1.B]
+ * ->join(t1.C=t2.D)
+ * ->agg[sum(A) as x, group by B]
+ * ->T1(A, B, C)
+ * ->T2(D)
+ */
+public class EagerAggRewriter extends DefaultPlanRewriter<PushDownAggContext> {
+ private static final double LOWER_AGGREGATE_EFFECT_COEFFICIENT = 10000;
+ private static final double LOW_AGGREGATE_EFFECT_COEFFICIENT = 1000;
+ private static final double MEDIUM_AGGREGATE_EFFECT_COEFFICIENT = 100;
+ private final StatsDerive derive = new StatsDerive(false);
+
+ @Override
+ public Plan visitLogicalJoin(LogicalJoin<? extends Plan, ? extends Plan> join, PushDownAggContext context) {
+ boolean toLeft = false;
+ boolean toRight = false;
+ boolean pushHere = false;
+ if (context.getAggFunctions().isEmpty()) {
+ // select t1.v from t1 join t2 on t1.id = t2.id group by t1.v, t2.v
+ // if no agg function, try to push agg to the child which contains all group keys
+ // TODO: consider t1.rows/(t1.id, t1.v).ndv and t2.rows/(t2.id, t2.v).ndv to determine push target
+ if (join.left().getOutputSet().containsAll(context.getGroupKeys())) {
+ toLeft = true;
+ } else if (join.right().getOutputSet().containsAll(context.getGroupKeys())) {
+ toRight = true;
+ } else {
+ pushHere = true;
+ }
+ } else {
+ for (AggregateFunction aggFunc : context.getAggFunctions()) {
+ if (join.left().getOutputSet().containsAll(aggFunc.getInputSlots())) {
+ toLeft = true;
+ } else if (join.right().getOutputSet().containsAll(aggFunc.getInputSlots())) {
+ toRight = true;
+ } else {
+ pushHere = true;
+ }
+ }
+ }
+
+ if (pushHere || (toLeft && toRight)) {
Review Comment:
toLeft 和 toRight 的设置
```
for (AggregateFunction aggFunc : context.getAggFunctions()) {
if (join.left().getOutputSet().containsAll(aggFunc.getInputSlots())) {
toLeft = true;
} else if (join.right().getOutputSet().containsAll(aggFunc.getInputSlots())) {
toRight = true;
} else {
pushHere = true;
}
}
```
存在这种场景：context.getAggFunctions()= { sum(left_col), sum(right_col)}
于是 toLeft 和 toRight 都是true
这里有冗余的比较。如果toLeft是true，再遇到toRight时就可以停止了。
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](eager-agg)Eager agg 0214 [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #60757:
URL: https://github.com/apache/doris/pull/60757#issuecomment-3942286702
<details>
<summary>TPC-DS: <b>Total hot run time: 152691 ms</b></summary>
```
machine: 'aliyun_ecs.c7a.8xlarge_32C64G'
scripts: https://github.com/apache/doris/tree/master/tools/tpcds-tools
TPC-DS sf100 test result on commit 742e5dc1a961547a7c51c084b61e93ee833a0d02, data reload: false
query5 4334 629 523 523
query6 326 217 218 217
query7 4218 477 265 265
query8 321 236 223 223
query9 8703 2739 2727 2727
query10 518 373 334 334
query11 7390 5856 5638 5638
query12 175 122 125 122
query13 1258 465 327 327
query14 5662 3807 3591 3591
query14_1 2831 2752 2762 2752
query15 195 191 172 172
query16 978 475 435 435
query17 1074 689 593 593
query18 2434 447 337 337
query19 205 207 172 172
query20 133 122 123 122
query21 218 145 119 119
query22 4773 4861 4722 4722
query23 16676 15900 15654 15654
query23_1 15678 15921 15741 15741
query24 7404 1694 1293 1293
query24_1 1281 1247 1268 1247
query25 560 488 433 433
query26 1299 272 161 161
query27 2918 479 295 295
query28 4799 1904 1983 1904
query29 870 574 463 463
query30 310 243 207 207
query31 1355 1276 1216 1216
query32 78 72 72 72
query33 502 324 277 277
query34 903 925 572 572
query35 635 671 635 635
query36 1076 1144 942 942
query37 133 97 80 80
query38 2896 2940 2844 2844
query39 880 872 859 859
query39_1 806 847 852 847
query40 234 153 131 131
query41 61 60 61 60
query42 300 298 297 297
query43 249 240 221 221
query44
query45 199 192 179 179
query46 886 988 604 604
query47 2105 2143 2052 2052
query48 328 323 240 240
query49 646 470 398 398
query50 707 287 220 220
query51 4089 4066 4035 4035
query52 286 291 281 281
query53 294 351 284 284
query54 303 284 283 283
query55 100 85 92 85
query56 324 342 317 317
query57 1342 1322 1269 1269
query58 299 289 280 280
query59 1332 1415 1283 1283
query60 355 340 337 337
query61 166 182 170 170
query62 635 596 543 543
query63 311 276 278 276
query64 5080 1342 1108 1108
query65
query66 1476 445 368 368
query67 16372 16570 16285 16285
query68
query69 392 300 279 279
query70 893 1016 944 944
query71 351 318 304 304
query72 2790 2677 2388 2388
query73 542 549 318 318
query74 10022 9950 9781 9781
query75 2845 2783 2456 2456
query76 2299 1038 698 698
query77 355 393 311 311
query78 11221 11293 10643 10643
query79 3089 816 589 589
query80 1718 621 522 522
query81 596 278 252 252
query82 1016 147 118 118
query83 336 266 245 245
query84 251 130 98 98
query85 928 490 440 440
query86 448 299 292 292
query87 3141 3062 2943 2943
query88 3522 2669 2649 2649
query89 433 362 344 344
query90 2002 178 177 177
query91 171 158 133 133
query92 75 72 68 68
query93 1359 815 494 494
query94 637 330 286 286
query95 584 391 309 309
query96 628 511 226 226
query97 2461 2513 2411 2411
query98 237 218 217 217
query99 989 998 920 920
Total cold run time: 236738 ms
Total hot run time: 152691 ms
```
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](eager-agg)Eager agg 0214 [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #60757:
URL: https://github.com/apache/doris/pull/60757#issuecomment-3983106266
<details>
<summary>TPC-DS: <b>Total hot run time: 153438 ms</b></summary>
```
machine: 'aliyun_ecs.c7a.8xlarge_32C64G'
scripts: https://github.com/apache/doris/tree/master/tools/tpcds-tools
TPC-DS sf100 test result on commit 4fd41b6bde3a7c72efcd889b0042d38412de1f9b, data reload: false
query5 4330 637 518 518
query6 337 227 214 214
query7 4211 467 263 263
query8 327 240 234 234
query9 8700 2744 2734 2734
query10 541 405 344 344
query11 7324 5836 5674 5674
query12 197 131 126 126
query13 1280 479 364 364
query14 6196 3868 3591 3591
query14_1 2815 2807 2839 2807
query15 205 197 178 178
query16 994 493 469 469
query17 1122 727 629 629
query18 2580 447 354 354
query19 221 217 182 182
query20 136 138 134 134
query21 227 151 125 125
query22 4716 4869 4706 4706
query23 16890 15932 15717 15717
query23_1 15765 15975 15999 15975
query24 7335 1720 1252 1252
query24_1 1255 1268 1282 1268
query25 582 487 423 423
query26 1275 273 161 161
query27 3040 482 292 292
query28 4503 1906 1911 1906
query29 811 578 484 484
query30 310 246 213 213
query31 1367 1289 1203 1203
query32 81 73 71 71
query33 510 337 271 271
query34 911 919 564 564
query35 670 679 606 606
query36 1113 1064 976 976
query37 130 96 84 84
query38 2967 2986 2823 2823
query39 874 865 843 843
query39_1 833 832 826 826
query40 233 153 138 138
query41 64 60 60 60
query42 313 310 299 299
query43 247 251 222 222
query44
query45 201 188 181 181
query46 873 993 606 606
query47 2120 2126 2055 2055
query48 310 321 227 227
query49 619 462 393 393
query50 678 286 217 217
query51 4280 4185 4098 4098
query52 287 291 281 281
query53 290 333 291 291
query54 303 277 265 265
query55 94 86 88 86
query56 318 332 299 299
query57 1370 1353 1270 1270
query58 292 277 305 277
query59 1339 1463 1316 1316
query60 341 338 325 325
query61 165 139 146 139
query62 649 586 553 553
query63 318 278 273 273
query64 4973 1268 990 990
query65
query66 1425 460 353 353
query67 16383 16424 16299 16299
query68
query69 381 303 281 281
query70 986 957 937 937
query71 347 313 303 303
query72 2859 2720 2437 2437
query73 542 558 345 345
query74 10009 9911 9810 9810
query75 2844 2755 2485 2485
query76 2297 1037 701 701
query77 370 380 324 324
query78 11304 11348 10691 10691
query79 1165 829 600 600
query80 1359 657 584 584
query81 575 284 259 259
query82 1277 158 119 119
query83 378 275 247 247
query84 253 118 102 102
query85 1093 553 433 433
query86 438 325 297 297
query87 3157 3132 2992 2992
query88 3568 2669 2647 2647
query89 435 373 345 345
query90 1877 177 168 168
query91 163 162 133 133
query92 82 77 71 71
query93 911 820 507 507
query94 652 332 297 297
query95 600 342 322 322
query96 641 519 224 224
query97 2469 2514 2481 2481
query98 230 220 223 220
query99 1010 1002 927 927
Total cold run time: 234638 ms
Total hot run time: 153438 ms
```
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](eager-agg)Eager agg 0214 [doris]
Posted by "hello-stephen (via GitHub)" <gi...@apache.org>.
hello-stephen commented on PR #60757:
URL: https://github.com/apache/doris/pull/60757#issuecomment-3970254500
# FE UT Coverage Report
Increment line coverage ` 63.48% (412/649)` :tada:
[Increment coverage report](http://coverage.selectdb-in.cc/coverage/60757_295bd29e62c92509a9b96dd18e2ee72013855af9/fe_increment_report/index.html)
[Complete coverage report](http://coverage.selectdb-in.cc/coverage/60757_295bd29e62c92509a9b96dd18e2ee72013855af9/fe_report/index.html)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](eager-agg)Eager agg 0214 [doris]
Posted by "englefly (via GitHub)" <gi...@apache.org>.
englefly commented on PR #60757:
URL: https://github.com/apache/doris/pull/60757#issuecomment-4002441738
@copilot 帮我写一下这个pr的描述
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](eager-agg)Eager agg 0214 [doris]
Posted by "doris-robot (via GitHub)" <gi...@apache.org>.
doris-robot commented on PR #60757:
URL: https://github.com/apache/doris/pull/60757#issuecomment-3948370160
<details>
<summary>TPC-H: <b>Total hot run time: 27628 ms</b></summary>
```
machine: 'aliyun_ecs.c7a.8xlarge_32C64G'
scripts: https://github.com/apache/doris/tree/master/tools/tpch-tools
Tpch sf100 test result on commit d8703bf984891db2830e567b16976638d0b6c9b4, data reload: false
------ Round 1 ----------------------------------
============================================
q1 17629 4560 4300 4300
q2 q3 10656 766 520 520
q4 4671 343 250 250
q5 7556 1201 1018 1018
q6 184 179 158 158
q7 783 847 679 679
q8 9311 1455 1360 1360
q9 4820 4790 4693 4693
q10 6303 1893 1632 1632
q11 469 265 253 253
q12 716 575 475 475
q13 18061 2914 2209 2209
q14 231 231 214 214
q15 958 789 790 789
q16 781 713 679 679
q17 705 876 433 433
q18 5896 5413 5225 5225
q19 1387 977 628 628
q20 503 512 392 392
q21 4490 1935 1461 1461
q22 344 291 260 260
Total cold run time: 96454 ms
Total hot run time: 27628 ms
----- Round 2, with runtime_filter_mode=off -----
============================================
q1 4449 4355 4381 4355
q2 q3 3879 4388 3818 3818
q4 880 1201 804 804
q5 4078 4385 4305 4305
q6 191 177 145 145
q7 1743 1629 1506 1506
q8 2459 2686 2593 2593
q9 7828 7368 7310 7310
q10 3779 3983 3684 3684
q11 510 432 455 432
q12 543 627 466 466
q13 2779 3399 2405 2405
q14 295 303 279 279
q15 845 825 838 825
q16 740 779 713 713
q17 1162 1456 1380 1380
q18 7085 6796 6669 6669
q19 886 879 920 879
q20 2143 2232 2014 2014
q21 4168 3587 3411 3411
q22 458 445 418 418
Total cold run time: 50900 ms
Total hot run time: 48411 ms
```
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](eager-agg)stabilize and extend eager aggregation pushdown (join/outer join/case-if/count-if) with broad regression baseline refresh [doris]
Posted by "morrySnow (via GitHub)" <gi...@apache.org>.
morrySnow commented on code in PR #60757:
URL: https://github.com/apache/doris/pull/60757#discussion_r3387234408
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/PushDownAggregation.java:
##########
@@ -0,0 +1,319 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.rules.rewrite.eageraggregation;
+
+import org.apache.doris.nereids.jobs.JobContext;
+import org.apache.doris.nereids.rules.analysis.NormalizeAggregate;
+import org.apache.doris.nereids.rules.rewrite.AdjustNullable;
+import org.apache.doris.nereids.trees.expressions.CaseWhen;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.functions.Function;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Max;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Min;
+import org.apache.doris.nereids.trees.expressions.functions.agg.RollUpTrait;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.If;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.Nvl;
+import org.apache.doris.nereids.trees.expressions.literal.BigIntLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalUnion;
+import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.qe.SessionVariable;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * push down aggregation
+ */
+public class PushDownAggregation extends DefaultPlanRewriter<JobContext> implements CustomRewriter {
+ private static final Logger LOG = LoggerFactory.getLogger(PushDownAggregation.class);
+
+ public final EagerAggRewriter writer = new EagerAggRewriter();
+
+ private final Set<Class> pushDownAggFunctionSet = Sets.newHashSet(
+ Count.class,
+ Sum.class,
+ Max.class,
+ Min.class);
+
+ private final Set<Class> acceptNodeType = Sets.newHashSet(
+ LogicalUnion.class,
+ LogicalProject.class,
+ LogicalFilter.class,
+ LogicalRelation.class,
+ LogicalJoin.class);
+
+ @Override
+ public Plan rewriteRoot(Plan plan, JobContext jobContext) {
+ if (SessionVariable.isFeDebug()) {
+ try {
+ new AdjustNullable(false).rewriteRoot(plan, null);
+ } catch (Exception e) {
+ LOG.warn("(PushDownAggregation) input plan has nullable problem", e);
+ return plan;
+ }
+ }
+ int mode = SessionVariable.getEagerAggregationMode();
+ if (mode < 0) {
+ return plan;
+ } else {
+ Plan result = plan.accept(this, jobContext);
+ if (SessionVariable.isFeDebug()) {
+ result = new AdjustNullable(true).rewriteRoot(result, null);
+ }
+ return result;
+ }
+ }
+
+ @Override
+ public Plan visitLogicalAggregate(LogicalAggregate<? extends Plan> agg, JobContext context) {
+ Plan newChild = agg.child().accept(this, context);
+ if (newChild != agg.child()) {
+ return agg.withChildren(newChild);
+ }
+
+ if (agg.getSourceRepeat().isPresent()) {
+ return agg;
+ }
+
+ List<SlotReference> groupKeys = new ArrayList<>();
+ for (Expression groupKey : agg.getGroupByExpressions()) {
+ if (groupKey instanceof SlotReference) {
+ groupKeys.add((SlotReference) groupKey);
+ } else {
+ SessionVariable.throwAnalysisExceptionWhenFeDebug(
+ "PushDownAggregation failed: agg is not normalized\n "
+ + agg.treeString());
+ return agg;
+ }
+ }
Review Comment:
这个检查可能必要性不大，这里即使绕过去了。后面的 cascades也会报错
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/PushDownAggregation.java:
##########
@@ -0,0 +1,319 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.rules.rewrite.eageraggregation;
+
+import org.apache.doris.nereids.jobs.JobContext;
+import org.apache.doris.nereids.rules.analysis.NormalizeAggregate;
+import org.apache.doris.nereids.rules.rewrite.AdjustNullable;
+import org.apache.doris.nereids.trees.expressions.CaseWhen;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.functions.Function;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Max;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Min;
+import org.apache.doris.nereids.trees.expressions.functions.agg.RollUpTrait;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.If;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.Nvl;
+import org.apache.doris.nereids.trees.expressions.literal.BigIntLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalUnion;
+import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.qe.SessionVariable;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * push down aggregation
+ */
+public class PushDownAggregation extends DefaultPlanRewriter<JobContext> implements CustomRewriter {
+ private static final Logger LOG = LoggerFactory.getLogger(PushDownAggregation.class);
+
+ public final EagerAggRewriter writer = new EagerAggRewriter();
+
+ private final Set<Class> pushDownAggFunctionSet = Sets.newHashSet(
+ Count.class,
+ Sum.class,
+ Max.class,
+ Min.class);
+
+ private final Set<Class> acceptNodeType = Sets.newHashSet(
+ LogicalUnion.class,
+ LogicalProject.class,
+ LogicalFilter.class,
+ LogicalRelation.class,
+ LogicalJoin.class);
+
+ @Override
+ public Plan rewriteRoot(Plan plan, JobContext jobContext) {
+ if (SessionVariable.isFeDebug()) {
+ try {
+ new AdjustNullable(false).rewriteRoot(plan, null);
+ } catch (Exception e) {
+ LOG.warn("(PushDownAggregation) input plan has nullable problem", e);
+ return plan;
+ }
+ }
+ int mode = SessionVariable.getEagerAggregationMode();
+ if (mode < 0) {
+ return plan;
+ } else {
+ Plan result = plan.accept(this, jobContext);
+ if (SessionVariable.isFeDebug()) {
Review Comment:
所有的 fedebug，可以换用更精细的 org.apache.doris.common.util.DebugPointUtil
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/PushDownAggregation.java:
##########
@@ -0,0 +1,319 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.rules.rewrite.eageraggregation;
+
+import org.apache.doris.nereids.jobs.JobContext;
+import org.apache.doris.nereids.rules.analysis.NormalizeAggregate;
+import org.apache.doris.nereids.rules.rewrite.AdjustNullable;
+import org.apache.doris.nereids.trees.expressions.CaseWhen;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.functions.Function;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Max;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Min;
+import org.apache.doris.nereids.trees.expressions.functions.agg.RollUpTrait;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.If;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.Nvl;
+import org.apache.doris.nereids.trees.expressions.literal.BigIntLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalUnion;
+import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.qe.SessionVariable;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * push down aggregation
+ */
+public class PushDownAggregation extends DefaultPlanRewriter<JobContext> implements CustomRewriter {
+ private static final Logger LOG = LoggerFactory.getLogger(PushDownAggregation.class);
+
+ public final EagerAggRewriter writer = new EagerAggRewriter();
+
+ private final Set<Class> pushDownAggFunctionSet = Sets.newHashSet(
+ Count.class,
+ Sum.class,
+ Max.class,
+ Min.class);
+
+ private final Set<Class> acceptNodeType = Sets.newHashSet(
+ LogicalUnion.class,
+ LogicalProject.class,
+ LogicalFilter.class,
+ LogicalRelation.class,
+ LogicalJoin.class);
+
+ @Override
+ public Plan rewriteRoot(Plan plan, JobContext jobContext) {
+ if (SessionVariable.isFeDebug()) {
+ try {
+ new AdjustNullable(false).rewriteRoot(plan, null);
+ } catch (Exception e) {
+ LOG.warn("(PushDownAggregation) input plan has nullable problem", e);
+ return plan;
+ }
+ }
+ int mode = SessionVariable.getEagerAggregationMode();
+ if (mode < 0) {
+ return plan;
+ } else {
+ Plan result = plan.accept(this, jobContext);
+ if (SessionVariable.isFeDebug()) {
+ result = new AdjustNullable(true).rewriteRoot(result, null);
+ }
+ return result;
+ }
+ }
+
+ @Override
+ public Plan visitLogicalAggregate(LogicalAggregate<? extends Plan> agg, JobContext context) {
+ Plan newChild = agg.child().accept(this, context);
+ if (newChild != agg.child()) {
+ return agg.withChildren(newChild);
+ }
+
+ if (agg.getSourceRepeat().isPresent()) {
+ return agg;
+ }
+
+ List<SlotReference> groupKeys = new ArrayList<>();
+ for (Expression groupKey : agg.getGroupByExpressions()) {
+ if (groupKey instanceof SlotReference) {
+ groupKeys.add((SlotReference) groupKey);
+ } else {
+ SessionVariable.throwAnalysisExceptionWhenFeDebug(
+ "PushDownAggregation failed: agg is not normalized\n "
+ + agg.treeString());
+ return agg;
+ }
+ }
+
+ Set<AggregateFunction> aggFunctions = Sets.newHashSet();
+ boolean hasDecomposedAggIf = false;
+ boolean hasCaseWhen = false;
+ Map<NamedExpression, List<AggregateFunction>> aggFunctionsForOutputExpressions = Maps.newHashMap();
+ for (NamedExpression aggOutput : agg.getOutputExpressions()) {
+ List<AggregateFunction> funcs = Lists.newArrayList();
+ aggFunctionsForOutputExpressions.put(aggOutput, funcs);
+ for (Object obj : aggOutput.collect(AggregateFunction.class::isInstance)) {
+ AggregateFunction aggFunction = (AggregateFunction) obj;
+ if (aggFunction.isDistinct()) {
+ return agg;
+ }
+ if (pushDownAggFunctionSet.contains(aggFunction.getClass())) {
+ // CaseWhen and If (which CASE WHEN is normalized into) must both be checked.
+ // When an agg function contains an If/CaseWhen whose condition tests IS NULL
+ // (e.g. count(if(col IS NULL, value, NULL))), pushing it to the nullable side
+ // of an outer join produces wrong results: null-extended rows make "col IS NULL"
+ // TRUE at the top level, but the pre-aggregated count slot becomes NULL after
+ // null-extension, and ifnull(sum(NULL), 0) = 0 instead of the correct 1.
+ if (!hasCaseWhen && aggFunction.anyMatch(e -> e instanceof CaseWhen || e instanceof If)) {
+ hasCaseWhen = true;
+ }
Review Comment:
可能遇到 ifnull(x, y) 等函数依然有问题
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalSetOperation.java:
##########
@@ -92,6 +97,38 @@ public LogicalSetOperation(PlanType planType, Qualifier qualifier, List<NamedExp
this.qualifier = qualifier;
this.outputs = ImmutableList.copyOf(outputs);
this.regularChildrenOutputs = ImmutableList.copyOf(regularChildrenOutputs);
+ //if (SessionVariable.isFeDebug()) {
+ // checkOutputs(outputs, regularChildrenOutputs)
+ // .ifPresent(msg -> SessionVariable.throwAnalysisExceptionWhenFeDebug(msg));
+ //}
+ }
+
+ // check every slot in outputs has its counterpart in regularChildrenOutputs, and they have the same data type.
+ private Optional<String> checkOutputs(List<NamedExpression> outputs,
Review Comment:
看起来这个函数只在注释里面用了
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/OrExpansion.java:
##########
@@ -176,10 +176,13 @@ public Plan visitLogicalJoin(LogicalJoin<? extends Plan, ? extends Plan> join, O
}
//4. union all joins and put producers to context
List<List<SlotReference>> childrenOutputs = joins.stream()
- .map(j -> j.getOutput().stream()
+ .map(j -> j.getOutput().stream() //.map(j -> j.getOutput().stream().distinct()
Review Comment:
多余的注释
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalUnion.java:
##########
@@ -424,4 +424,12 @@ private static List<NamedExpression> castToCommonType(List<NamedExpression> row,
}
return changed ? castedRow.build() : row;
}
+
+ public LogicalSetOperation withChildrenAndOutputs(List<Plan> children, List<NamedExpression> newOuptuts,
Review Comment:
为什么不是返回 LogicalUnion？
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/ColumnPruning.java:
##########
@@ -221,28 +219,21 @@ public Plan visitLogicalUnion(LogicalUnion union, PruneContext context) {
}
LogicalUnion prunedOutputUnion = pruneUnionOutput(union, context);
// start prune children of union
- List<Slot> originOutput = union.getOutput();
Review Comment:
这个修改是因为 union 的列裁剪有bug？
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/EagerAggRewriter.java:
##########
@@ -0,0 +1,625 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.rules.rewrite.eageraggregation;
+
+import org.apache.doris.nereids.rules.analysis.NormalizeAggregate;
+import org.apache.doris.nereids.rules.rewrite.StatsDerive;
+import org.apache.doris.nereids.stats.ExpressionEstimation;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.CaseWhen;
+import org.apache.doris.nereids.trees.expressions.Cast;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.If;
+import org.apache.doris.nereids.trees.plans.JoinType;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCatalogRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalUnion;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.types.DataType;
+import org.apache.doris.qe.SessionVariable;
+import org.apache.doris.statistics.ColumnStatistic;
+import org.apache.doris.statistics.Statistics;
+
+import com.google.common.collect.Lists;
+
+import java.util.ArrayList;
+import java.util.IdentityHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * eager aggregation
+ * agg[sum(t1.A) group by t1.B]
+ * ->join(t1.C=t2.D)
+ * ->T1(A, B, C)
+ * ->T2(D)
+ *
+ * =>
+ * agg[sum(x) group by t1.B]
+ * ->join(t1.C=t2.D)
+ * ->agg[sum(A) as x, group by B]
+ * ->T1(A, B, C)
+ * ->T2(D)
+ */
+public class EagerAggRewriter extends DefaultPlanRewriter<PushDownAggContext> {
+ private static final double LOWER_AGGREGATE_EFFECT_COEFFICIENT = 10000;
+ private static final double LOW_AGGREGATE_EFFECT_COEFFICIENT = 1000;
+ private static final double MEDIUM_AGGREGATE_EFFECT_COEFFICIENT = 100;
+ private final StatsDerive derive = new StatsDerive(false);
+
+ @Override
+ public Plan visitLogicalJoin(LogicalJoin<? extends Plan, ? extends Plan> join, PushDownAggContext context) {
+ boolean toLeft = false;
+ boolean toRight = false;
+ boolean pushHere = false;
+ if (context.getAggFunctions().isEmpty()) {
+ // select t1.v from t1 join t2 on t1.id = t2.id group by t1.v, t2.v
+ // if no agg function, try to push agg to the child which contains all group keys
+ // TODO: consider t1.rows/(t1.id, t1.v).ndv and t2.rows/(t2.id, t2.v).ndv to determine push target
+ if (join.left().getOutputSet().containsAll(context.getGroupKeys())) {
+ toLeft = true;
+ } else if (join.right().getOutputSet().containsAll(context.getGroupKeys())) {
+ toRight = true;
+ } else {
+ pushHere = true;
+ }
+ } else {
+ for (AggregateFunction aggFunc : context.getAggFunctions()) {
+ if (join.left().getOutputSet().containsAll(aggFunc.getInputSlots())) {
+ toLeft = true;
+ } else if (join.right().getOutputSet().containsAll(aggFunc.getInputSlots())) {
+ toRight = true;
+ } else {
+ pushHere = true;
+ }
+ }
+ }
+
+ if (pushHere || (toLeft && toRight)) {
Review Comment:
toLeft and toRight 永远是 false？
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/EagerAggRewriter.java:
##########
@@ -0,0 +1,625 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.rules.rewrite.eageraggregation;
+
+import org.apache.doris.nereids.rules.analysis.NormalizeAggregate;
+import org.apache.doris.nereids.rules.rewrite.StatsDerive;
+import org.apache.doris.nereids.stats.ExpressionEstimation;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.CaseWhen;
+import org.apache.doris.nereids.trees.expressions.Cast;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.If;
+import org.apache.doris.nereids.trees.plans.JoinType;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCatalogRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalUnion;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.types.DataType;
+import org.apache.doris.qe.SessionVariable;
+import org.apache.doris.statistics.ColumnStatistic;
+import org.apache.doris.statistics.Statistics;
+
+import com.google.common.collect.Lists;
+
+import java.util.ArrayList;
+import java.util.IdentityHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * eager aggregation
+ * agg[sum(t1.A) group by t1.B]
+ * ->join(t1.C=t2.D)
+ * ->T1(A, B, C)
+ * ->T2(D)
+ *
+ * =>
+ * agg[sum(x) group by t1.B]
+ * ->join(t1.C=t2.D)
+ * ->agg[sum(A) as x, group by B]
+ * ->T1(A, B, C)
+ * ->T2(D)
+ */
+public class EagerAggRewriter extends DefaultPlanRewriter<PushDownAggContext> {
+ private static final double LOWER_AGGREGATE_EFFECT_COEFFICIENT = 10000;
+ private static final double LOW_AGGREGATE_EFFECT_COEFFICIENT = 1000;
+ private static final double MEDIUM_AGGREGATE_EFFECT_COEFFICIENT = 100;
+ private final StatsDerive derive = new StatsDerive(false);
+
+ @Override
+ public Plan visitLogicalJoin(LogicalJoin<? extends Plan, ? extends Plan> join, PushDownAggContext context) {
+ boolean toLeft = false;
+ boolean toRight = false;
+ boolean pushHere = false;
+ if (context.getAggFunctions().isEmpty()) {
+ // select t1.v from t1 join t2 on t1.id = t2.id group by t1.v, t2.v
+ // if no agg function, try to push agg to the child which contains all group keys
+ // TODO: consider t1.rows/(t1.id, t1.v).ndv and t2.rows/(t2.id, t2.v).ndv to determine push target
+ if (join.left().getOutputSet().containsAll(context.getGroupKeys())) {
+ toLeft = true;
+ } else if (join.right().getOutputSet().containsAll(context.getGroupKeys())) {
+ toRight = true;
+ } else {
+ pushHere = true;
+ }
+ } else {
+ for (AggregateFunction aggFunc : context.getAggFunctions()) {
+ if (join.left().getOutputSet().containsAll(aggFunc.getInputSlots())) {
+ toLeft = true;
+ } else if (join.right().getOutputSet().containsAll(aggFunc.getInputSlots())) {
+ toRight = true;
+ } else {
+ pushHere = true;
+ }
Review Comment:
这里为什么不检查 group by key 了？
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/PushDownAggContext.java:
##########
@@ -0,0 +1,144 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.rules.rewrite.eageraggregation;
+
+import org.apache.doris.nereids.CascadesContext;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableSet;
+
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * PushDownAggContext
+ */
+public class PushDownAggContext {
+ public static final int BIG_JOIN_BUILD_SIZE = 400_000;
+ // count(if(...)): if(...) push down as a whole
+ // sum/min/max(if(truePart, elsePart)): if(...) can be split to sum(truePart) and sum(elsePart)
+ public final boolean hasDecomposedAggIf;
+ // When aggFunc(if(...)) is present, pushing down the null-supplemented side of the outer join is avoided.
+ // This is because null values are highly error-prone,
+ // so the push-down operation is not performed during hashCaseWhen.
+ public final boolean hasCaseWhen;
+ private final List<AggregateFunction> aggFunctions;
+ private final List<SlotReference> groupKeys;
+ private final HashMap<AggregateFunction, Alias> aliasMap;
+ private final Set<Slot> aggFunctionsInputSlots;
+
+ // cascadesContext is used for normalizeAgg
+ private final CascadesContext cascadesContext;
+
+ private final boolean passThroughBigJoin;
+
+ /**
+ * constructor
+ */
+ public PushDownAggContext(List<AggregateFunction> aggFunctions,
+ List<SlotReference> groupKeys, Map<AggregateFunction, Alias> aliasMap, CascadesContext cascadesContext,
+ boolean passThroughBigJoin, boolean hasDecomposedAggIf, boolean hasCaseWhen) {
+ this.groupKeys = groupKeys.stream().distinct().collect(Collectors.toList());
+ this.aggFunctions = ImmutableList.copyOf(aggFunctions);
+ this.cascadesContext = cascadesContext;
+
+ HashMap<AggregateFunction, Alias> builtAliasMap = new HashMap<>();
+ if (aliasMap == null) {
+ for (AggregateFunction aggFunction : this.aggFunctions) {
+ builtAliasMap.put(aggFunction, new Alias(aggFunction, aggFunction.getName()));
+ }
+ } else {
+ for (AggregateFunction aggFunction : this.aggFunctions) {
+ Alias alias = aliasMap.get(aggFunction);
+ if (alias == null) {
+ alias = new Alias(aggFunction, aggFunction.getName());
+ }
+ builtAliasMap.put(aggFunction, alias);
+ }
+ }
+ this.aliasMap = builtAliasMap;
+
+ this.aggFunctionsInputSlots = aggFunctions.stream()
+ .flatMap(aggFunction -> aggFunction.getInputSlots().stream())
+ .filter(Slot.class::isInstance)
+ .collect(ImmutableSet.toImmutableSet());
+ this.passThroughBigJoin = passThroughBigJoin;
+ this.hasDecomposedAggIf = hasDecomposedAggIf;
+ this.hasCaseWhen = hasCaseWhen;
+ }
+
+ /**
+ * check validation
+ * @return true, if groupKeys is not empty and no group by key is in aggFunctionsInputSlots
+ */
+ public boolean isValid() {
+ return !groupKeys.isEmpty()
+ && !groupKeys.stream().anyMatch(s -> aggFunctionsInputSlots.contains(s));
Review Comment:
使用 nonMatch。另外这里是为什么呢？sum(a) group by a 好像也没啥问题？
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalSetOperation.java:
##########
@@ -92,6 +97,38 @@ public LogicalSetOperation(PlanType planType, Qualifier qualifier, List<NamedExp
this.qualifier = qualifier;
this.outputs = ImmutableList.copyOf(outputs);
this.regularChildrenOutputs = ImmutableList.copyOf(regularChildrenOutputs);
+ //if (SessionVariable.isFeDebug()) {
+ // checkOutputs(outputs, regularChildrenOutputs)
+ // .ifPresent(msg -> SessionVariable.throwAnalysisExceptionWhenFeDebug(msg));
+ //}
Review Comment:
多余的注释需要删掉
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/PushDownAggregation.java:
##########
@@ -0,0 +1,319 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.rules.rewrite.eageraggregation;
+
+import org.apache.doris.nereids.jobs.JobContext;
+import org.apache.doris.nereids.rules.analysis.NormalizeAggregate;
+import org.apache.doris.nereids.rules.rewrite.AdjustNullable;
+import org.apache.doris.nereids.trees.expressions.CaseWhen;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.functions.Function;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Max;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Min;
+import org.apache.doris.nereids.trees.expressions.functions.agg.RollUpTrait;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.If;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.Nvl;
+import org.apache.doris.nereids.trees.expressions.literal.BigIntLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalUnion;
+import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.qe.SessionVariable;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * push down aggregation
+ */
+public class PushDownAggregation extends DefaultPlanRewriter<JobContext> implements CustomRewriter {
+ private static final Logger LOG = LoggerFactory.getLogger(PushDownAggregation.class);
+
+ public final EagerAggRewriter writer = new EagerAggRewriter();
+
+ private final Set<Class> pushDownAggFunctionSet = Sets.newHashSet(
+ Count.class,
+ Sum.class,
+ Max.class,
+ Min.class);
+
+ private final Set<Class> acceptNodeType = Sets.newHashSet(
+ LogicalUnion.class,
+ LogicalProject.class,
+ LogicalFilter.class,
+ LogicalRelation.class,
+ LogicalJoin.class);
+
+ @Override
+ public Plan rewriteRoot(Plan plan, JobContext jobContext) {
+ if (SessionVariable.isFeDebug()) {
+ try {
+ new AdjustNullable(false).rewriteRoot(plan, null);
+ } catch (Exception e) {
+ LOG.warn("(PushDownAggregation) input plan has nullable problem", e);
+ return plan;
+ }
+ }
+ int mode = SessionVariable.getEagerAggregationMode();
+ if (mode < 0) {
+ return plan;
+ } else {
+ Plan result = plan.accept(this, jobContext);
+ if (SessionVariable.isFeDebug()) {
+ result = new AdjustNullable(true).rewriteRoot(result, null);
+ }
+ return result;
+ }
+ }
+
+ @Override
+ public Plan visitLogicalAggregate(LogicalAggregate<? extends Plan> agg, JobContext context) {
+ Plan newChild = agg.child().accept(this, context);
+ if (newChild != agg.child()) {
+ return agg.withChildren(newChild);
+ }
+
+ if (agg.getSourceRepeat().isPresent()) {
+ return agg;
+ }
+
+ List<SlotReference> groupKeys = new ArrayList<>();
+ for (Expression groupKey : agg.getGroupByExpressions()) {
+ if (groupKey instanceof SlotReference) {
+ groupKeys.add((SlotReference) groupKey);
+ } else {
+ SessionVariable.throwAnalysisExceptionWhenFeDebug(
+ "PushDownAggregation failed: agg is not normalized\n "
+ + agg.treeString());
+ return agg;
+ }
+ }
+
+ Set<AggregateFunction> aggFunctions = Sets.newHashSet();
+ boolean hasDecomposedAggIf = false;
+ boolean hasCaseWhen = false;
+ Map<NamedExpression, List<AggregateFunction>> aggFunctionsForOutputExpressions = Maps.newHashMap();
+ for (NamedExpression aggOutput : agg.getOutputExpressions()) {
+ List<AggregateFunction> funcs = Lists.newArrayList();
+ aggFunctionsForOutputExpressions.put(aggOutput, funcs);
+ for (Object obj : aggOutput.collect(AggregateFunction.class::isInstance)) {
+ AggregateFunction aggFunction = (AggregateFunction) obj;
+ if (aggFunction.isDistinct()) {
+ return agg;
+ }
+ if (pushDownAggFunctionSet.contains(aggFunction.getClass())) {
+ // CaseWhen and If (which CASE WHEN is normalized into) must both be checked.
+ // When an agg function contains an If/CaseWhen whose condition tests IS NULL
+ // (e.g. count(if(col IS NULL, value, NULL))), pushing it to the nullable side
+ // of an outer join produces wrong results: null-extended rows make "col IS NULL"
+ // TRUE at the top level, but the pre-aggregated count slot becomes NULL after
+ // null-extension, and ifnull(sum(NULL), 0) = 0 instead of the correct 1.
+ if (!hasCaseWhen && aggFunction.anyMatch(e -> e instanceof CaseWhen || e instanceof If)) {
+ hasCaseWhen = true;
+ }
+ if (aggFunction.arity() > 0 && aggFunction.child(0) instanceof If
+ && !(aggFunction instanceof Count)) {
+ // Decompose Sum/Max/Min(If(cond, a, b)) into separate agg functions.
+ // Count(If(...)) is NOT decomposed here because the top-level
+ // replacement (Count->Sum rollup) cannot match the decomposed
+ // Count(a)/Count(b) as sub-expressions of the original Count(If(cond,a,b)).
+ // Count(If(...)) is pushed down as-is and rolled up normally.
+ If body = (If) (aggFunction).child(0);
+ Set<Slot> valueSlots = Sets.newHashSet(body.getTrueValue().getInputSlots());
+ valueSlots.addAll(body.getFalseValue().getInputSlots());
+ if (body.getCondition().getInputSlots().stream().anyMatch(s -> valueSlots.contains(s))) {
+ // do not push down sum(if a then a else b)
+ return agg;
+ }
+ AggregateFunction aggTrue = (AggregateFunction) aggFunction.withChildren(body.getTrueValue());
+ aggFunctions.add(aggTrue);
+ funcs.add(aggTrue);
+ if (!(body.getFalseValue() instanceof NullLiteral)) {
+ AggregateFunction aggFalse =
+ (AggregateFunction) aggFunction.withChildren(body.getFalseValue());
+ aggFunctions.add(aggFalse);
+ funcs.add(aggFalse);
+ }
+ groupKeys.addAll(body.getCondition().getInputSlots()
+ .stream().map(slot -> (SlotReference) slot).collect(Collectors.toList()));
+ hasDecomposedAggIf = true;
+ } else {
+ aggFunctions.add(aggFunction);
+ funcs.add(aggFunction);
+ }
+
+ } else {
+ return agg;
+ }
+ }
+ }
+
+ groupKeys = groupKeys.stream().distinct().collect(Collectors.toList());
+ if (!checkSubTreePattern(agg.child())) {
+ return agg;
+ }
+
+ PushDownAggContext pushDownContext = new PushDownAggContext(new ArrayList<>(aggFunctions),
+ groupKeys, null, context.getCascadesContext(), false, hasDecomposedAggIf, hasCaseWhen);
+ if (!pushDownContext.isValid()) {
+ return agg;
+ }
+ try {
+ Plan child = agg.child().accept(writer, pushDownContext);
+ if (child != agg.child()) {
+ // agg has been pushed down, rewrite agg output expressions
+ // before: agg[sum(A), by (B)]
+ // ->join(C=D)
+ // ->scan(T1[A...])
+ // ->scan(T2)
+ // after: agg[sum(x), by(B)]
+ // ->join(C=D)
+ // ->agg[sum(A) as x, by(B,C)]
+ // ->scan(T1[A...])
+ // ->scan(T2)
+ List<NamedExpression> newOutputExpressions = new ArrayList<>();
+ //Map<AggregateFunction, AggregateFunction> replaceMap = new HashMap<>();
+ //for (AggregateFunction aggFunc : pushDownContext.getAliasMap().keySet()) {
+ // Alias alias = pushDownContext.getAliasMap().get(aggFunc);
+ // replaceMap.put(aggFunc, (AggregateFunction) aggFunc.withChildren((Expression) alias.toSlot()));
+ //}
+
+ for (NamedExpression ne : agg.getOutputExpressions()) {
+ if (ne instanceof SlotReference) {
+ newOutputExpressions.add(ne);
+ } else {
+ // every expression has its own replaceMap
+ // aggregation(output=[min(A), sum(A)])
+ // --> join
+ // -> T1 [A ...]
+ // -> T2 [...]
+ // =>
+ // aggregation(output=[min(minA), sum(sumA)])
+ // --> join
+ // -> agg(output=[min(A) as minA, sum(A) as sumA])
+ // -> T1 [A ...]
+ // -> T2 [...]
+ // for min(A), replaceMap: A->minA
+ // for sum(A), replaceMap: A->sumA
+ // for count(A), replaceMap: count(A)->sum(countA), because count needs rollup to sum
+ Map<Expression, Expression> replaceMap = new HashMap<>();
+ List<AggregateFunction> relatedAggFunc = aggFunctionsForOutputExpressions.get(ne);
+ for (AggregateFunction func : relatedAggFunc) {
+ Slot pushedDownSlot = pushDownContext.getAliasMap().get(func).toSlot();
+ if (func instanceof Count) {
+ // For count(A), after pushdown we have count(A) as x,
+ // and the top agg should use sum(x) instead of count(x).
+ // Wrap with ifnull(..., 0) because COUNT never returns NULL,
+ // but after pushdown across an outer join, the intermediate count
+ // slot can be NULL (null-extended), making sum(NULL) = NULL.
+ Function rollUpFunc = ((RollUpTrait) func).constructRollUp(pushedDownSlot);
+ replaceMap.put(func, new Nvl(rollUpFunc, new BigIntLiteral(0)));
+ } else if (func.arity() > 0) {
+ // For sum/max/min, replace the child expression with the pushed down slot
+ replaceMap.put(func.child(0), pushedDownSlot);
+ }
+ }
+ NamedExpression replaceAliasExpr = (NamedExpression) ExpressionUtils.replace(ne, replaceMap);
+ replaceAliasExpr = (NamedExpression) ExpressionUtils.rebuildSignature(replaceAliasExpr);
+ newOutputExpressions.add(replaceAliasExpr);
+ }
+ }
+ LogicalAggregate<Plan> eagerAgg =
+ agg.withAggOutputChild(newOutputExpressions, child);
+ NormalizeAggregate normalizeAggregate = new NormalizeAggregate();
+ return normalizeAggregate.normalizeAgg(eagerAgg, Optional.empty(),
+ context.getCascadesContext());
+ }
+ } catch (RuntimeException e) {
+ String msg = "PushDownAggregation failed: " + e.getMessage() + "\n" + agg.treeString();
+ LOG.info(msg, e);
+ SessionVariable.throwAnalysisExceptionWhenFeDebug(msg);
+ }
+ return agg;
+ }
+
+ private boolean checkSubTreePattern(Plan root) {
+ return containsPushDownJoin(root)
+ && checkPlanNodeType(root);
+ }
+
+ private boolean containsPushDownJoin(Plan root) {
+ if (root instanceof LogicalJoin && !((LogicalJoin) root).isMarkJoin()) {
+ return true;
+ }
Review Comment:
这里可以对mark join 和 asof join 等做短路
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/PushDownAggregation.java:
##########
@@ -0,0 +1,319 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.rules.rewrite.eageraggregation;
+
+import org.apache.doris.nereids.jobs.JobContext;
+import org.apache.doris.nereids.rules.analysis.NormalizeAggregate;
+import org.apache.doris.nereids.rules.rewrite.AdjustNullable;
+import org.apache.doris.nereids.trees.expressions.CaseWhen;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.functions.Function;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Max;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Min;
+import org.apache.doris.nereids.trees.expressions.functions.agg.RollUpTrait;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.If;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.Nvl;
+import org.apache.doris.nereids.trees.expressions.literal.BigIntLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalUnion;
+import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.qe.SessionVariable;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * push down aggregation
+ */
+public class PushDownAggregation extends DefaultPlanRewriter<JobContext> implements CustomRewriter {
+ private static final Logger LOG = LoggerFactory.getLogger(PushDownAggregation.class);
+
+ public final EagerAggRewriter writer = new EagerAggRewriter();
+
+ private final Set<Class> pushDownAggFunctionSet = Sets.newHashSet(
+ Count.class,
+ Sum.class,
+ Max.class,
+ Min.class);
+
+ private final Set<Class> acceptNodeType = Sets.newHashSet(
+ LogicalUnion.class,
+ LogicalProject.class,
+ LogicalFilter.class,
+ LogicalRelation.class,
+ LogicalJoin.class);
+
+ @Override
+ public Plan rewriteRoot(Plan plan, JobContext jobContext) {
+ if (SessionVariable.isFeDebug()) {
+ try {
+ new AdjustNullable(false).rewriteRoot(plan, null);
+ } catch (Exception e) {
+ LOG.warn("(PushDownAggregation) input plan has nullable problem", e);
+ return plan;
+ }
+ }
+ int mode = SessionVariable.getEagerAggregationMode();
+ if (mode < 0) {
+ return plan;
+ } else {
+ Plan result = plan.accept(this, jobContext);
+ if (SessionVariable.isFeDebug()) {
+ result = new AdjustNullable(true).rewriteRoot(result, null);
+ }
+ return result;
+ }
+ }
+
+ @Override
+ public Plan visitLogicalAggregate(LogicalAggregate<? extends Plan> agg, JobContext context) {
+ Plan newChild = agg.child().accept(this, context);
+ if (newChild != agg.child()) {
+ return agg.withChildren(newChild);
+ }
+
+ if (agg.getSourceRepeat().isPresent()) {
+ return agg;
+ }
+
+ List<SlotReference> groupKeys = new ArrayList<>();
+ for (Expression groupKey : agg.getGroupByExpressions()) {
+ if (groupKey instanceof SlotReference) {
+ groupKeys.add((SlotReference) groupKey);
+ } else {
+ SessionVariable.throwAnalysisExceptionWhenFeDebug(
+ "PushDownAggregation failed: agg is not normalized\n "
+ + agg.treeString());
+ return agg;
+ }
+ }
+
+ Set<AggregateFunction> aggFunctions = Sets.newHashSet();
+ boolean hasDecomposedAggIf = false;
+ boolean hasCaseWhen = false;
+ Map<NamedExpression, List<AggregateFunction>> aggFunctionsForOutputExpressions = Maps.newHashMap();
+ for (NamedExpression aggOutput : agg.getOutputExpressions()) {
+ List<AggregateFunction> funcs = Lists.newArrayList();
+ aggFunctionsForOutputExpressions.put(aggOutput, funcs);
+ for (Object obj : aggOutput.collect(AggregateFunction.class::isInstance)) {
+ AggregateFunction aggFunction = (AggregateFunction) obj;
+ if (aggFunction.isDistinct()) {
+ return agg;
+ }
+ if (pushDownAggFunctionSet.contains(aggFunction.getClass())) {
+ // CaseWhen and If (which CASE WHEN is normalized into) must both be checked.
+ // When an agg function contains an If/CaseWhen whose condition tests IS NULL
+ // (e.g. count(if(col IS NULL, value, NULL))), pushing it to the nullable side
+ // of an outer join produces wrong results: null-extended rows make "col IS NULL"
+ // TRUE at the top level, but the pre-aggregated count slot becomes NULL after
+ // null-extension, and ifnull(sum(NULL), 0) = 0 instead of the correct 1.
+ if (!hasCaseWhen && aggFunction.anyMatch(e -> e instanceof CaseWhen || e instanceof If)) {
+ hasCaseWhen = true;
+ }
+ if (aggFunction.arity() > 0 && aggFunction.child(0) instanceof If
+ && !(aggFunction instanceof Count)) {
+ // Decompose Sum/Max/Min(If(cond, a, b)) into separate agg functions.
+ // Count(If(...)) is NOT decomposed here because the top-level
+ // replacement (Count->Sum rollup) cannot match the decomposed
+ // Count(a)/Count(b) as sub-expressions of the original Count(If(cond,a,b)).
+ // Count(If(...)) is pushed down as-is and rolled up normally.
+ If body = (If) (aggFunction).child(0);
+ Set<Slot> valueSlots = Sets.newHashSet(body.getTrueValue().getInputSlots());
+ valueSlots.addAll(body.getFalseValue().getInputSlots());
+ if (body.getCondition().getInputSlots().stream().anyMatch(s -> valueSlots.contains(s))) {
+ // do not push down sum(if a then a else b)
+ return agg;
+ }
Review Comment:
这块整段没看懂，为什么 if(x,y,z) 要拆开，以及为什么 if(x, x, y) 不能work？需要更清晰的注释
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/BoundFunction.java:
##########
@@ -155,4 +155,8 @@ private Supplier<FunctionSignature> buildSignatureCache(Supplier<FunctionSignatu
});
}
}
+
+ //public void rebuildSignature() {
+ // this.signatureCache = buildSignatureCache(null);
+ //}
Review Comment:
多余的注释
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/PushDownAggregation.java:
##########
@@ -0,0 +1,319 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.rules.rewrite.eageraggregation;
+
+import org.apache.doris.nereids.jobs.JobContext;
+import org.apache.doris.nereids.rules.analysis.NormalizeAggregate;
+import org.apache.doris.nereids.rules.rewrite.AdjustNullable;
+import org.apache.doris.nereids.trees.expressions.CaseWhen;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.functions.Function;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Max;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Min;
+import org.apache.doris.nereids.trees.expressions.functions.agg.RollUpTrait;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.If;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.Nvl;
+import org.apache.doris.nereids.trees.expressions.literal.BigIntLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalUnion;
+import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.qe.SessionVariable;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * push down aggregation
+ */
+public class PushDownAggregation extends DefaultPlanRewriter<JobContext> implements CustomRewriter {
+ private static final Logger LOG = LoggerFactory.getLogger(PushDownAggregation.class);
+
+ public final EagerAggRewriter writer = new EagerAggRewriter();
+
+ private final Set<Class> pushDownAggFunctionSet = Sets.newHashSet(
+ Count.class,
+ Sum.class,
+ Max.class,
+ Min.class);
+
+ private final Set<Class> acceptNodeType = Sets.newHashSet(
+ LogicalUnion.class,
+ LogicalProject.class,
+ LogicalFilter.class,
+ LogicalRelation.class,
+ LogicalJoin.class);
+
+ @Override
+ public Plan rewriteRoot(Plan plan, JobContext jobContext) {
+ if (SessionVariable.isFeDebug()) {
+ try {
+ new AdjustNullable(false).rewriteRoot(plan, null);
+ } catch (Exception e) {
+ LOG.warn("(PushDownAggregation) input plan has nullable problem", e);
+ return plan;
+ }
+ }
+ int mode = SessionVariable.getEagerAggregationMode();
+ if (mode < 0) {
+ return plan;
+ } else {
+ Plan result = plan.accept(this, jobContext);
+ if (SessionVariable.isFeDebug()) {
+ result = new AdjustNullable(true).rewriteRoot(result, null);
+ }
+ return result;
+ }
+ }
+
+ @Override
+ public Plan visitLogicalAggregate(LogicalAggregate<? extends Plan> agg, JobContext context) {
+ Plan newChild = agg.child().accept(this, context);
+ if (newChild != agg.child()) {
+ return agg.withChildren(newChild);
+ }
+
+ if (agg.getSourceRepeat().isPresent()) {
+ return agg;
+ }
+
+ List<SlotReference> groupKeys = new ArrayList<>();
+ for (Expression groupKey : agg.getGroupByExpressions()) {
+ if (groupKey instanceof SlotReference) {
+ groupKeys.add((SlotReference) groupKey);
+ } else {
+ SessionVariable.throwAnalysisExceptionWhenFeDebug(
+ "PushDownAggregation failed: agg is not normalized\n "
+ + agg.treeString());
+ return agg;
+ }
+ }
+
+ Set<AggregateFunction> aggFunctions = Sets.newHashSet();
+ boolean hasDecomposedAggIf = false;
+ boolean hasCaseWhen = false;
+ Map<NamedExpression, List<AggregateFunction>> aggFunctionsForOutputExpressions = Maps.newHashMap();
+ for (NamedExpression aggOutput : agg.getOutputExpressions()) {
+ List<AggregateFunction> funcs = Lists.newArrayList();
+ aggFunctionsForOutputExpressions.put(aggOutput, funcs);
+ for (Object obj : aggOutput.collect(AggregateFunction.class::isInstance)) {
+ AggregateFunction aggFunction = (AggregateFunction) obj;
+ if (aggFunction.isDistinct()) {
+ return agg;
+ }
+ if (pushDownAggFunctionSet.contains(aggFunction.getClass())) {
+ // CaseWhen and If (which CASE WHEN is normalized into) must both be checked.
+ // When an agg function contains an If/CaseWhen whose condition tests IS NULL
+ // (e.g. count(if(col IS NULL, value, NULL))), pushing it to the nullable side
+ // of an outer join produces wrong results: null-extended rows make "col IS NULL"
+ // TRUE at the top level, but the pre-aggregated count slot becomes NULL after
+ // null-extension, and ifnull(sum(NULL), 0) = 0 instead of the correct 1.
+ if (!hasCaseWhen && aggFunction.anyMatch(e -> e instanceof CaseWhen || e instanceof If)) {
+ hasCaseWhen = true;
+ }
+ if (aggFunction.arity() > 0 && aggFunction.child(0) instanceof If
+ && !(aggFunction instanceof Count)) {
+ // Decompose Sum/Max/Min(If(cond, a, b)) into separate agg functions.
+ // Count(If(...)) is NOT decomposed here because the top-level
+ // replacement (Count->Sum rollup) cannot match the decomposed
+ // Count(a)/Count(b) as sub-expressions of the original Count(If(cond,a,b)).
+ // Count(If(...)) is pushed down as-is and rolled up normally.
+ If body = (If) (aggFunction).child(0);
+ Set<Slot> valueSlots = Sets.newHashSet(body.getTrueValue().getInputSlots());
+ valueSlots.addAll(body.getFalseValue().getInputSlots());
+ if (body.getCondition().getInputSlots().stream().anyMatch(s -> valueSlots.contains(s))) {
+ // do not push down sum(if a then a else b)
+ return agg;
+ }
+ AggregateFunction aggTrue = (AggregateFunction) aggFunction.withChildren(body.getTrueValue());
+ aggFunctions.add(aggTrue);
+ funcs.add(aggTrue);
+ if (!(body.getFalseValue() instanceof NullLiteral)) {
+ AggregateFunction aggFalse =
+ (AggregateFunction) aggFunction.withChildren(body.getFalseValue());
+ aggFunctions.add(aggFalse);
+ funcs.add(aggFalse);
+ }
+ groupKeys.addAll(body.getCondition().getInputSlots()
+ .stream().map(slot -> (SlotReference) slot).collect(Collectors.toList()));
+ hasDecomposedAggIf = true;
+ } else {
+ aggFunctions.add(aggFunction);
+ funcs.add(aggFunction);
+ }
+
+ } else {
+ return agg;
+ }
+ }
+ }
+
+ groupKeys = groupKeys.stream().distinct().collect(Collectors.toList());
Review Comment:
这个应该移动到check 之后执行
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/ColumnPruning.java:
##########
@@ -221,28 +219,21 @@ public Plan visitLogicalUnion(LogicalUnion union, PruneContext context) {
}
LogicalUnion prunedOutputUnion = pruneUnionOutput(union, context);
// start prune children of union
- List<Slot> originOutput = union.getOutput();
- Set<Slot> prunedOutput = prunedOutputUnion.getOutputSet();
- List<Integer> prunedOutputIndexes = IntStream.range(0, originOutput.size())
- .filter(index -> prunedOutput.contains(originOutput.get(index)))
- .boxed()
- .collect(ImmutableList.toImmutableList());
-
ImmutableList.Builder<Plan> prunedChildren = ImmutableList.builder();
ImmutableList.Builder<List<SlotReference>> prunedChildrenOutputs = ImmutableList.builder();
for (int i = 0; i < prunedOutputUnion.arity(); i++) {
List<SlotReference> regularChildOutputs = prunedOutputUnion.getRegularChildOutput(i);
RoaringBitmap prunedChildOutputExprIds = new RoaringBitmap();
- Builder<SlotReference> prunedChildOutputBuilder
- = ImmutableList.builderWithExpectedSize(regularChildOutputs.size());
- for (Integer index : prunedOutputIndexes) {
- SlotReference slot = regularChildOutputs.get(index);
- prunedChildOutputBuilder.add(slot);
- prunedChildOutputExprIds.add(slot.getExprId().asInt());
- }
-
- List<SlotReference> prunedChildOutput = prunedChildOutputBuilder.build();
+ //Builder<SlotReference> prunedChildOutputBuilder
+ // = ImmutableList.builderWithExpectedSize(regularChildOutputs.size());
+ //for (Integer index : prunedOutputIndexes) {
+ // SlotReference slot = regularChildOutputs.get(index);
+ // prunedChildOutputBuilder.add(slot);
+ // prunedChildOutputExprIds.add(slot.getExprId().asInt());
+ //}
+ regularChildOutputs.forEach(col -> prunedChildOutputExprIds.add(col.getExprId().asInt()));
+ List<SlotReference> prunedChildOutput = regularChildOutputs; //prunedChildOutputBuilder.build();
Review Comment:
多余的注释
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/OrExpansion.java:
##########
@@ -176,10 +176,13 @@ public Plan visitLogicalJoin(LogicalJoin<? extends Plan, ? extends Plan> join, O
}
//4. union all joins and put producers to context
List<List<SlotReference>> childrenOutputs = joins.stream()
- .map(j -> j.getOutput().stream()
+ .map(j -> j.getOutput().stream() //.map(j -> j.getOutput().stream().distinct()
.map(SlotReference.class::cast)
.collect(ImmutableList.toImmutableList()))
.collect(ImmutableList.toImmutableList());
+ //LogicalUnion union = new LogicalUnion(Qualifier.ALL,
+ // new ArrayList<>(join.getOutput().stream().distinct().collect(Collectors.toList())),
+ // childrenOutputs, ImmutableList.of(), false, joins);
Review Comment:
多余的注释
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/OrExpansion.java:
##########
@@ -319,8 +322,26 @@ private List<Plan> expandInnerJoin(CascadesContext ctx, Pair<List<Expression>,
LogicalCTEConsumer left = new LogicalCTEConsumer(ctx.getStatementContext().getNextRelationId(),
leftProducer.getCteId(), "", leftProducer);
+ List<NamedExpression> leftOutput = new ArrayList<>();
Review Comment:
这里是因为 or expansion 有bug？
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/ColumnPruning.java:
##########
@@ -221,28 +219,21 @@ public Plan visitLogicalUnion(LogicalUnion union, PruneContext context) {
}
LogicalUnion prunedOutputUnion = pruneUnionOutput(union, context);
// start prune children of union
- List<Slot> originOutput = union.getOutput();
- Set<Slot> prunedOutput = prunedOutputUnion.getOutputSet();
- List<Integer> prunedOutputIndexes = IntStream.range(0, originOutput.size())
- .filter(index -> prunedOutput.contains(originOutput.get(index)))
- .boxed()
- .collect(ImmutableList.toImmutableList());
-
ImmutableList.Builder<Plan> prunedChildren = ImmutableList.builder();
ImmutableList.Builder<List<SlotReference>> prunedChildrenOutputs = ImmutableList.builder();
for (int i = 0; i < prunedOutputUnion.arity(); i++) {
List<SlotReference> regularChildOutputs = prunedOutputUnion.getRegularChildOutput(i);
RoaringBitmap prunedChildOutputExprIds = new RoaringBitmap();
- Builder<SlotReference> prunedChildOutputBuilder
- = ImmutableList.builderWithExpectedSize(regularChildOutputs.size());
- for (Integer index : prunedOutputIndexes) {
- SlotReference slot = regularChildOutputs.get(index);
- prunedChildOutputBuilder.add(slot);
- prunedChildOutputExprIds.add(slot.getExprId().asInt());
- }
-
- List<SlotReference> prunedChildOutput = prunedChildOutputBuilder.build();
+ //Builder<SlotReference> prunedChildOutputBuilder
+ // = ImmutableList.builderWithExpectedSize(regularChildOutputs.size());
+ //for (Integer index : prunedOutputIndexes) {
+ // SlotReference slot = regularChildOutputs.get(index);
+ // prunedChildOutputBuilder.add(slot);
+ // prunedChildOutputExprIds.add(slot.getExprId().asInt());
+ //}
Review Comment:
多余的注释
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/PushDownAggregation.java:
##########
@@ -0,0 +1,319 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.rules.rewrite.eageraggregation;
+
+import org.apache.doris.nereids.jobs.JobContext;
+import org.apache.doris.nereids.rules.analysis.NormalizeAggregate;
+import org.apache.doris.nereids.rules.rewrite.AdjustNullable;
+import org.apache.doris.nereids.trees.expressions.CaseWhen;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.functions.Function;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Max;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Min;
+import org.apache.doris.nereids.trees.expressions.functions.agg.RollUpTrait;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.If;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.Nvl;
+import org.apache.doris.nereids.trees.expressions.literal.BigIntLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalUnion;
+import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.qe.SessionVariable;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * push down aggregation
+ */
+public class PushDownAggregation extends DefaultPlanRewriter<JobContext> implements CustomRewriter {
+ private static final Logger LOG = LoggerFactory.getLogger(PushDownAggregation.class);
+
+ public final EagerAggRewriter writer = new EagerAggRewriter();
+
+ private final Set<Class> pushDownAggFunctionSet = Sets.newHashSet(
+ Count.class,
+ Sum.class,
+ Max.class,
+ Min.class);
+
+ private final Set<Class> acceptNodeType = Sets.newHashSet(
+ LogicalUnion.class,
+ LogicalProject.class,
+ LogicalFilter.class,
+ LogicalRelation.class,
+ LogicalJoin.class);
+
+ @Override
+ public Plan rewriteRoot(Plan plan, JobContext jobContext) {
+ if (SessionVariable.isFeDebug()) {
+ try {
+ new AdjustNullable(false).rewriteRoot(plan, null);
+ } catch (Exception e) {
+ LOG.warn("(PushDownAggregation) input plan has nullable problem", e);
+ return plan;
+ }
+ }
+ int mode = SessionVariable.getEagerAggregationMode();
+ if (mode < 0) {
+ return plan;
+ } else {
+ Plan result = plan.accept(this, jobContext);
+ if (SessionVariable.isFeDebug()) {
+ result = new AdjustNullable(true).rewriteRoot(result, null);
+ }
+ return result;
+ }
+ }
+
+ @Override
+ public Plan visitLogicalAggregate(LogicalAggregate<? extends Plan> agg, JobContext context) {
+ Plan newChild = agg.child().accept(this, context);
+ if (newChild != agg.child()) {
+ return agg.withChildren(newChild);
+ }
+
+ if (agg.getSourceRepeat().isPresent()) {
+ return agg;
+ }
+
+ List<SlotReference> groupKeys = new ArrayList<>();
+ for (Expression groupKey : agg.getGroupByExpressions()) {
+ if (groupKey instanceof SlotReference) {
+ groupKeys.add((SlotReference) groupKey);
+ } else {
+ SessionVariable.throwAnalysisExceptionWhenFeDebug(
+ "PushDownAggregation failed: agg is not normalized\n "
+ + agg.treeString());
+ return agg;
+ }
+ }
+
+ Set<AggregateFunction> aggFunctions = Sets.newHashSet();
+ boolean hasDecomposedAggIf = false;
+ boolean hasCaseWhen = false;
+ Map<NamedExpression, List<AggregateFunction>> aggFunctionsForOutputExpressions = Maps.newHashMap();
+ for (NamedExpression aggOutput : agg.getOutputExpressions()) {
+ List<AggregateFunction> funcs = Lists.newArrayList();
+ aggFunctionsForOutputExpressions.put(aggOutput, funcs);
+ for (Object obj : aggOutput.collect(AggregateFunction.class::isInstance)) {
+ AggregateFunction aggFunction = (AggregateFunction) obj;
+ if (aggFunction.isDistinct()) {
+ return agg;
+ }
+ if (pushDownAggFunctionSet.contains(aggFunction.getClass())) {
+ // CaseWhen and If (which CASE WHEN is normalized into) must both be checked.
+ // When an agg function contains an If/CaseWhen whose condition tests IS NULL
+ // (e.g. count(if(col IS NULL, value, NULL))), pushing it to the nullable side
+ // of an outer join produces wrong results: null-extended rows make "col IS NULL"
+ // TRUE at the top level, but the pre-aggregated count slot becomes NULL after
+ // null-extension, and ifnull(sum(NULL), 0) = 0 instead of the correct 1.
+ if (!hasCaseWhen && aggFunction.anyMatch(e -> e instanceof CaseWhen || e instanceof If)) {
+ hasCaseWhen = true;
+ }
+ if (aggFunction.arity() > 0 && aggFunction.child(0) instanceof If
+ && !(aggFunction instanceof Count)) {
+ // Decompose Sum/Max/Min(If(cond, a, b)) into separate agg functions.
+ // Count(If(...)) is NOT decomposed here because the top-level
+ // replacement (Count->Sum rollup) cannot match the decomposed
+ // Count(a)/Count(b) as sub-expressions of the original Count(If(cond,a,b)).
+ // Count(If(...)) is pushed down as-is and rolled up normally.
+ If body = (If) (aggFunction).child(0);
+ Set<Slot> valueSlots = Sets.newHashSet(body.getTrueValue().getInputSlots());
+ valueSlots.addAll(body.getFalseValue().getInputSlots());
+ if (body.getCondition().getInputSlots().stream().anyMatch(s -> valueSlots.contains(s))) {
+ // do not push down sum(if a then a else b)
+ return agg;
+ }
+ AggregateFunction aggTrue = (AggregateFunction) aggFunction.withChildren(body.getTrueValue());
+ aggFunctions.add(aggTrue);
+ funcs.add(aggTrue);
+ if (!(body.getFalseValue() instanceof NullLiteral)) {
+ AggregateFunction aggFalse =
+ (AggregateFunction) aggFunction.withChildren(body.getFalseValue());
+ aggFunctions.add(aggFalse);
+ funcs.add(aggFalse);
+ }
+ groupKeys.addAll(body.getCondition().getInputSlots()
+ .stream().map(slot -> (SlotReference) slot).collect(Collectors.toList()));
+ hasDecomposedAggIf = true;
+ } else {
+ aggFunctions.add(aggFunction);
+ funcs.add(aggFunction);
+ }
+
+ } else {
+ return agg;
+ }
+ }
+ }
+
+ groupKeys = groupKeys.stream().distinct().collect(Collectors.toList());
+ if (!checkSubTreePattern(agg.child())) {
+ return agg;
+ }
+
+ PushDownAggContext pushDownContext = new PushDownAggContext(new ArrayList<>(aggFunctions),
+ groupKeys, null, context.getCascadesContext(), false, hasDecomposedAggIf, hasCaseWhen);
+ if (!pushDownContext.isValid()) {
+ return agg;
+ }
+ try {
+ Plan child = agg.child().accept(writer, pushDownContext);
+ if (child != agg.child()) {
+ // agg has been pushed down, rewrite agg output expressions
+ // before: agg[sum(A), by (B)]
+ // ->join(C=D)
+ // ->scan(T1[A...])
+ // ->scan(T2)
+ // after: agg[sum(x), by(B)]
+ // ->join(C=D)
+ // ->agg[sum(A) as x, by(B,C)]
+ // ->scan(T1[A...])
+ // ->scan(T2)
+ List<NamedExpression> newOutputExpressions = new ArrayList<>();
+ //Map<AggregateFunction, AggregateFunction> replaceMap = new HashMap<>();
+ //for (AggregateFunction aggFunc : pushDownContext.getAliasMap().keySet()) {
+ // Alias alias = pushDownContext.getAliasMap().get(aggFunc);
+ // replaceMap.put(aggFunc, (AggregateFunction) aggFunc.withChildren((Expression) alias.toSlot()));
+ //}
Review Comment:
多余的注释
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/EagerAggRewriter.java:
##########
@@ -0,0 +1,625 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.rules.rewrite.eageraggregation;
+
+import org.apache.doris.nereids.rules.analysis.NormalizeAggregate;
+import org.apache.doris.nereids.rules.rewrite.StatsDerive;
+import org.apache.doris.nereids.stats.ExpressionEstimation;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.CaseWhen;
+import org.apache.doris.nereids.trees.expressions.Cast;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.If;
+import org.apache.doris.nereids.trees.plans.JoinType;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCatalogRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalUnion;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.types.DataType;
+import org.apache.doris.qe.SessionVariable;
+import org.apache.doris.statistics.ColumnStatistic;
+import org.apache.doris.statistics.Statistics;
+
+import com.google.common.collect.Lists;
+
+import java.util.ArrayList;
+import java.util.IdentityHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * eager aggregation
+ * agg[sum(t1.A) group by t1.B]
+ * ->join(t1.C=t2.D)
+ * ->T1(A, B, C)
+ * ->T2(D)
+ *
+ * =>
+ * agg[sum(x) group by t1.B]
+ * ->join(t1.C=t2.D)
+ * ->agg[sum(A) as x, group by B]
+ * ->T1(A, B, C)
+ * ->T2(D)
+ */
+public class EagerAggRewriter extends DefaultPlanRewriter<PushDownAggContext> {
+ private static final double LOWER_AGGREGATE_EFFECT_COEFFICIENT = 10000;
+ private static final double LOW_AGGREGATE_EFFECT_COEFFICIENT = 1000;
+ private static final double MEDIUM_AGGREGATE_EFFECT_COEFFICIENT = 100;
+ private final StatsDerive derive = new StatsDerive(false);
+
+ @Override
+ public Plan visitLogicalJoin(LogicalJoin<? extends Plan, ? extends Plan> join, PushDownAggContext context) {
+ boolean toLeft = false;
+ boolean toRight = false;
+ boolean pushHere = false;
+ if (context.getAggFunctions().isEmpty()) {
+ // select t1.v from t1 join t2 on t1.id = t2.id group by t1.v, t2.v
+ // if no agg function, try to push agg to the child which contains all group keys
+ // TODO: consider t1.rows/(t1.id, t1.v).ndv and t2.rows/(t2.id, t2.v).ndv to determine push target
+ if (join.left().getOutputSet().containsAll(context.getGroupKeys())) {
+ toLeft = true;
+ } else if (join.right().getOutputSet().containsAll(context.getGroupKeys())) {
+ toRight = true;
+ } else {
+ pushHere = true;
+ }
+ } else {
+ for (AggregateFunction aggFunc : context.getAggFunctions()) {
+ if (join.left().getOutputSet().containsAll(aggFunc.getInputSlots())) {
+ toLeft = true;
+ } else if (join.right().getOutputSet().containsAll(aggFunc.getInputSlots())) {
+ toRight = true;
+ } else {
+ pushHere = true;
+ }
+ }
+ }
+
+ if (pushHere || (toLeft && toRight)) {
+ if (SessionVariable.isEagerAggregationOnJoin()) {
+ return genAggregate(join, context);
+ } else {
+ return join;
+ }
+ }
+ // Do not push aggregation to the nullable side of outer joins when agg function contains case-when.
+ // CaseWhen expressions may produce non-null values from null-padded rows (e.g., WHEN col IS NULL THEN -54),
+ // so pre-aggregation before the join loses those contributions.
+ if (context.hasDecomposedAggIf || context.hasCaseWhen) {
+ JoinType joinType = join.getJoinType();
+ if (joinType.isFullOuterJoin()) {
+ return join;
+ }
+ if (joinType.isRightOuterJoin()) {
+ toLeft = false;
+ }
+ if (joinType.isLeftOuterJoin()) {
+ toRight = false;
+ }
+ if (!toLeft && !toRight && !pushHere) {
+ return join;
+ }
+ }
+
+ // Do not push count(*)/count(literal)/count(preserved_side_col) to the nullable side of outer joins.
Review Comment:
为什么 count(preserved_side_col) 能推给另一边？
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/PushDownAggregation.java:
##########
@@ -0,0 +1,319 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.rules.rewrite.eageraggregation;
+
+import org.apache.doris.nereids.jobs.JobContext;
+import org.apache.doris.nereids.rules.analysis.NormalizeAggregate;
+import org.apache.doris.nereids.rules.rewrite.AdjustNullable;
+import org.apache.doris.nereids.trees.expressions.CaseWhen;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.functions.Function;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Max;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Min;
+import org.apache.doris.nereids.trees.expressions.functions.agg.RollUpTrait;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.If;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.Nvl;
+import org.apache.doris.nereids.trees.expressions.literal.BigIntLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalUnion;
+import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.qe.SessionVariable;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * push down aggregation
+ */
+public class PushDownAggregation extends DefaultPlanRewriter<JobContext> implements CustomRewriter {
+ private static final Logger LOG = LoggerFactory.getLogger(PushDownAggregation.class);
+
+ public final EagerAggRewriter writer = new EagerAggRewriter();
+
+ private final Set<Class> pushDownAggFunctionSet = Sets.newHashSet(
+ Count.class,
+ Sum.class,
+ Max.class,
+ Min.class);
+
+ private final Set<Class> acceptNodeType = Sets.newHashSet(
+ LogicalUnion.class,
+ LogicalProject.class,
+ LogicalFilter.class,
+ LogicalRelation.class,
+ LogicalJoin.class);
+
+ @Override
+ public Plan rewriteRoot(Plan plan, JobContext jobContext) {
+ if (SessionVariable.isFeDebug()) {
+ try {
+ new AdjustNullable(false).rewriteRoot(plan, null);
+ } catch (Exception e) {
+ LOG.warn("(PushDownAggregation) input plan has nullable problem", e);
+ return plan;
+ }
+ }
+ int mode = SessionVariable.getEagerAggregationMode();
+ if (mode < 0) {
+ return plan;
+ } else {
+ Plan result = plan.accept(this, jobContext);
+ if (SessionVariable.isFeDebug()) {
+ result = new AdjustNullable(true).rewriteRoot(result, null);
+ }
+ return result;
+ }
+ }
+
+ @Override
+ public Plan visitLogicalAggregate(LogicalAggregate<? extends Plan> agg, JobContext context) {
+ Plan newChild = agg.child().accept(this, context);
+ if (newChild != agg.child()) {
+ return agg.withChildren(newChild);
+ }
+
+ if (agg.getSourceRepeat().isPresent()) {
+ return agg;
+ }
+
+ List<SlotReference> groupKeys = new ArrayList<>();
+ for (Expression groupKey : agg.getGroupByExpressions()) {
+ if (groupKey instanceof SlotReference) {
+ groupKeys.add((SlotReference) groupKey);
+ } else {
+ SessionVariable.throwAnalysisExceptionWhenFeDebug(
+ "PushDownAggregation failed: agg is not normalized\n "
+ + agg.treeString());
+ return agg;
+ }
+ }
+
+ Set<AggregateFunction> aggFunctions = Sets.newHashSet();
+ boolean hasDecomposedAggIf = false;
+ boolean hasCaseWhen = false;
+ Map<NamedExpression, List<AggregateFunction>> aggFunctionsForOutputExpressions = Maps.newHashMap();
+ for (NamedExpression aggOutput : agg.getOutputExpressions()) {
+ List<AggregateFunction> funcs = Lists.newArrayList();
+ aggFunctionsForOutputExpressions.put(aggOutput, funcs);
+ for (Object obj : aggOutput.collect(AggregateFunction.class::isInstance)) {
+ AggregateFunction aggFunction = (AggregateFunction) obj;
+ if (aggFunction.isDistinct()) {
+ return agg;
+ }
+ if (pushDownAggFunctionSet.contains(aggFunction.getClass())) {
+ // CaseWhen and If (which CASE WHEN is normalized into) must both be checked.
+ // When an agg function contains an If/CaseWhen whose condition tests IS NULL
+ // (e.g. count(if(col IS NULL, value, NULL))), pushing it to the nullable side
+ // of an outer join produces wrong results: null-extended rows make "col IS NULL"
+ // TRUE at the top level, but the pre-aggregated count slot becomes NULL after
+ // null-extension, and ifnull(sum(NULL), 0) = 0 instead of the correct 1.
+ if (!hasCaseWhen && aggFunction.anyMatch(e -> e instanceof CaseWhen || e instanceof If)) {
+ hasCaseWhen = true;
+ }
+ if (aggFunction.arity() > 0 && aggFunction.child(0) instanceof If
+ && !(aggFunction instanceof Count)) {
+ // Decompose Sum/Max/Min(If(cond, a, b)) into separate agg functions.
+ // Count(If(...)) is NOT decomposed here because the top-level
+ // replacement (Count->Sum rollup) cannot match the decomposed
+ // Count(a)/Count(b) as sub-expressions of the original Count(If(cond,a,b)).
+ // Count(If(...)) is pushed down as-is and rolled up normally.
+ If body = (If) (aggFunction).child(0);
+ Set<Slot> valueSlots = Sets.newHashSet(body.getTrueValue().getInputSlots());
+ valueSlots.addAll(body.getFalseValue().getInputSlots());
+ if (body.getCondition().getInputSlots().stream().anyMatch(s -> valueSlots.contains(s))) {
+ // do not push down sum(if a then a else b)
+ return agg;
+ }
+ AggregateFunction aggTrue = (AggregateFunction) aggFunction.withChildren(body.getTrueValue());
+ aggFunctions.add(aggTrue);
+ funcs.add(aggTrue);
+ if (!(body.getFalseValue() instanceof NullLiteral)) {
+ AggregateFunction aggFalse =
+ (AggregateFunction) aggFunction.withChildren(body.getFalseValue());
+ aggFunctions.add(aggFalse);
+ funcs.add(aggFalse);
+ }
+ groupKeys.addAll(body.getCondition().getInputSlots()
+ .stream().map(slot -> (SlotReference) slot).collect(Collectors.toList()));
+ hasDecomposedAggIf = true;
+ } else {
+ aggFunctions.add(aggFunction);
+ funcs.add(aggFunction);
+ }
+
+ } else {
+ return agg;
+ }
+ }
+ }
+
+ groupKeys = groupKeys.stream().distinct().collect(Collectors.toList());
+ if (!checkSubTreePattern(agg.child())) {
+ return agg;
+ }
+
+ PushDownAggContext pushDownContext = new PushDownAggContext(new ArrayList<>(aggFunctions),
+ groupKeys, null, context.getCascadesContext(), false, hasDecomposedAggIf, hasCaseWhen);
+ if (!pushDownContext.isValid()) {
+ return agg;
+ }
+ try {
+ Plan child = agg.child().accept(writer, pushDownContext);
+ if (child != agg.child()) {
+ // agg has been pushed down, rewrite agg output expressions
+ // before: agg[sum(A), by (B)]
+ // ->join(C=D)
+ // ->scan(T1[A...])
+ // ->scan(T2)
+ // after: agg[sum(x), by(B)]
+ // ->join(C=D)
+ // ->agg[sum(A) as x, by(B,C)]
+ // ->scan(T1[A...])
+ // ->scan(T2)
+ List<NamedExpression> newOutputExpressions = new ArrayList<>();
+ //Map<AggregateFunction, AggregateFunction> replaceMap = new HashMap<>();
+ //for (AggregateFunction aggFunc : pushDownContext.getAliasMap().keySet()) {
+ // Alias alias = pushDownContext.getAliasMap().get(aggFunc);
+ // replaceMap.put(aggFunc, (AggregateFunction) aggFunc.withChildren((Expression) alias.toSlot()));
+ //}
+
+ for (NamedExpression ne : agg.getOutputExpressions()) {
+ if (ne instanceof SlotReference) {
+ newOutputExpressions.add(ne);
+ } else {
+ // every expression has its own replaceMap
+ // aggregation(output=[min(A), sum(A)])
+ // --> join
+ // -> T1 [A ...]
+ // -> T2 [...]
+ // =>
+ // aggregation(output=[min(minA), sum(sumA)])
+ // --> join
+ // -> agg(output=[min(A) as minA, sum(A) as sumA])
+ // -> T1 [A ...]
+ // -> T2 [...]
+ // for min(A), replaceMap: A->minA
+ // for sum(A), replaceMap: A->sumA
+ // for count(A), replaceMap: count(A)->sum(countA), because count needs rollup to sum
+ Map<Expression, Expression> replaceMap = new HashMap<>();
+ List<AggregateFunction> relatedAggFunc = aggFunctionsForOutputExpressions.get(ne);
+ for (AggregateFunction func : relatedAggFunc) {
+ Slot pushedDownSlot = pushDownContext.getAliasMap().get(func).toSlot();
+ if (func instanceof Count) {
+ // For count(A), after pushdown we have count(A) as x,
+ // and the top agg should use sum(x) instead of count(x).
+ // Wrap with ifnull(..., 0) because COUNT never returns NULL,
+ // but after pushdown across an outer join, the intermediate count
+ // slot can be NULL (null-extended), making sum(NULL) = NULL.
+ Function rollUpFunc = ((RollUpTrait) func).constructRollUp(pushedDownSlot);
+ replaceMap.put(func, new Nvl(rollUpFunc, new BigIntLiteral(0)));
+ } else if (func.arity() > 0) {
+ // For sum/max/min, replace the child expression with the pushed down slot
+ replaceMap.put(func.child(0), pushedDownSlot);
+ }
+ }
+ NamedExpression replaceAliasExpr = (NamedExpression) ExpressionUtils.replace(ne, replaceMap);
+ replaceAliasExpr = (NamedExpression) ExpressionUtils.rebuildSignature(replaceAliasExpr);
+ newOutputExpressions.add(replaceAliasExpr);
+ }
+ }
+ LogicalAggregate<Plan> eagerAgg =
+ agg.withAggOutputChild(newOutputExpressions, child);
+ NormalizeAggregate normalizeAggregate = new NormalizeAggregate();
+ return normalizeAggregate.normalizeAgg(eagerAgg, Optional.empty(),
+ context.getCascadesContext());
Review Comment:
为什么需要重新normalize？
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](eager-agg)stabilize and extend eager aggregation pushdown (join/outer join/case-if/count-if) with broad regression baseline refresh [doris]
Posted by "englefly (via GitHub)" <gi...@apache.org>.
englefly commented on code in PR #60757:
URL: https://github.com/apache/doris/pull/60757#discussion_r3393679623
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/EagerAggRewriter.java:
##########
@@ -0,0 +1,625 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.rules.rewrite.eageraggregation;
+
+import org.apache.doris.nereids.rules.analysis.NormalizeAggregate;
+import org.apache.doris.nereids.rules.rewrite.StatsDerive;
+import org.apache.doris.nereids.stats.ExpressionEstimation;
+import org.apache.doris.nereids.trees.expressions.Alias;
+import org.apache.doris.nereids.trees.expressions.CaseWhen;
+import org.apache.doris.nereids.trees.expressions.Cast;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.If;
+import org.apache.doris.nereids.trees.plans.JoinType;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCatalogRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalUnion;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.types.DataType;
+import org.apache.doris.qe.SessionVariable;
+import org.apache.doris.statistics.ColumnStatistic;
+import org.apache.doris.statistics.Statistics;
+
+import com.google.common.collect.Lists;
+
+import java.util.ArrayList;
+import java.util.IdentityHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * eager aggregation
+ * agg[sum(t1.A) group by t1.B]
+ * ->join(t1.C=t2.D)
+ * ->T1(A, B, C)
+ * ->T2(D)
+ *
+ * =>
+ * agg[sum(x) group by t1.B]
+ * ->join(t1.C=t2.D)
+ * ->agg[sum(A) as x, group by B]
+ * ->T1(A, B, C)
+ * ->T2(D)
+ */
+public class EagerAggRewriter extends DefaultPlanRewriter<PushDownAggContext> {
+ private static final double LOWER_AGGREGATE_EFFECT_COEFFICIENT = 10000;
+ private static final double LOW_AGGREGATE_EFFECT_COEFFICIENT = 1000;
+ private static final double MEDIUM_AGGREGATE_EFFECT_COEFFICIENT = 100;
+ private final StatsDerive derive = new StatsDerive(false);
+
+ @Override
+ public Plan visitLogicalJoin(LogicalJoin<? extends Plan, ? extends Plan> join, PushDownAggContext context) {
+ boolean toLeft = false;
+ boolean toRight = false;
+ boolean pushHere = false;
+ if (context.getAggFunctions().isEmpty()) {
+ // select t1.v from t1 join t2 on t1.id = t2.id group by t1.v, t2.v
+ // if no agg function, try to push agg to the child which contains all group keys
+ // TODO: consider t1.rows/(t1.id, t1.v).ndv and t2.rows/(t2.id, t2.v).ndv to determine push target
+ if (join.left().getOutputSet().containsAll(context.getGroupKeys())) {
+ toLeft = true;
+ } else if (join.right().getOutputSet().containsAll(context.getGroupKeys())) {
+ toRight = true;
+ } else {
+ pushHere = true;
+ }
+ } else {
+ for (AggregateFunction aggFunc : context.getAggFunctions()) {
+ if (join.left().getOutputSet().containsAll(aggFunc.getInputSlots())) {
+ toLeft = true;
+ } else if (join.right().getOutputSet().containsAll(aggFunc.getInputSlots())) {
+ toRight = true;
+ } else {
+ pushHere = true;
+ }
Review Comment:
如果没有agg function，就按group key 来决定下推，这相当于agg function就是求 group key的distinct。
这个分支有agg function，就只需要下推agg function，下推后的group key 是join key 和 group key 与被推侧output的交集
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
Re: [PR] [feat](eager-agg)stabilize and extend eager aggregation pushdown (join/outer join/case-if/count-if) with broad regression baseline refresh [doris]
Posted by "englefly (via GitHub)" <gi...@apache.org>.
englefly commented on code in PR #60757:
URL: https://github.com/apache/doris/pull/60757#discussion_r3392837049
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/eageraggregation/PushDownAggregation.java:
##########
@@ -0,0 +1,319 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.rules.rewrite.eageraggregation;
+
+import org.apache.doris.nereids.jobs.JobContext;
+import org.apache.doris.nereids.rules.analysis.NormalizeAggregate;
+import org.apache.doris.nereids.rules.rewrite.AdjustNullable;
+import org.apache.doris.nereids.trees.expressions.CaseWhen;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.expressions.functions.Function;
+import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Count;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Max;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Min;
+import org.apache.doris.nereids.trees.expressions.functions.agg.RollUpTrait;
+import org.apache.doris.nereids.trees.expressions.functions.agg.Sum;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.If;
+import org.apache.doris.nereids.trees.expressions.functions.scalar.Nvl;
+import org.apache.doris.nereids.trees.expressions.literal.BigIntLiteral;
+import org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalUnion;
+import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.nereids.util.ExpressionUtils;
+import org.apache.doris.qe.SessionVariable;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * push down aggregation
+ */
+public class PushDownAggregation extends DefaultPlanRewriter<JobContext> implements CustomRewriter {
+ private static final Logger LOG = LoggerFactory.getLogger(PushDownAggregation.class);
+
+ public final EagerAggRewriter writer = new EagerAggRewriter();
+
+ private final Set<Class> pushDownAggFunctionSet = Sets.newHashSet(
+ Count.class,
+ Sum.class,
+ Max.class,
+ Min.class);
+
+ private final Set<Class> acceptNodeType = Sets.newHashSet(
+ LogicalUnion.class,
+ LogicalProject.class,
+ LogicalFilter.class,
+ LogicalRelation.class,
+ LogicalJoin.class);
+
+ @Override
+ public Plan rewriteRoot(Plan plan, JobContext jobContext) {
+ if (SessionVariable.isFeDebug()) {
+ try {
+ new AdjustNullable(false).rewriteRoot(plan, null);
+ } catch (Exception e) {
+ LOG.warn("(PushDownAggregation) input plan has nullable problem", e);
+ return plan;
+ }
+ }
+ int mode = SessionVariable.getEagerAggregationMode();
+ if (mode < 0) {
+ return plan;
+ } else {
+ Plan result = plan.accept(this, jobContext);
+ if (SessionVariable.isFeDebug()) {
+ result = new AdjustNullable(true).rewriteRoot(result, null);
+ }
+ return result;
+ }
+ }
+
+ @Override
+ public Plan visitLogicalAggregate(LogicalAggregate<? extends Plan> agg, JobContext context) {
+ Plan newChild = agg.child().accept(this, context);
+ if (newChild != agg.child()) {
+ return agg.withChildren(newChild);
+ }
+
+ if (agg.getSourceRepeat().isPresent()) {
+ return agg;
+ }
+
+ List<SlotReference> groupKeys = new ArrayList<>();
+ for (Expression groupKey : agg.getGroupByExpressions()) {
+ if (groupKey instanceof SlotReference) {
+ groupKeys.add((SlotReference) groupKey);
+ } else {
+ SessionVariable.throwAnalysisExceptionWhenFeDebug(
+ "PushDownAggregation failed: agg is not normalized\n "
+ + agg.treeString());
+ return agg;
+ }
+ }
+
+ Set<AggregateFunction> aggFunctions = Sets.newHashSet();
+ boolean hasDecomposedAggIf = false;
+ boolean hasCaseWhen = false;
+ Map<NamedExpression, List<AggregateFunction>> aggFunctionsForOutputExpressions = Maps.newHashMap();
+ for (NamedExpression aggOutput : agg.getOutputExpressions()) {
+ List<AggregateFunction> funcs = Lists.newArrayList();
+ aggFunctionsForOutputExpressions.put(aggOutput, funcs);
+ for (Object obj : aggOutput.collect(AggregateFunction.class::isInstance)) {
+ AggregateFunction aggFunction = (AggregateFunction) obj;
+ if (aggFunction.isDistinct()) {
+ return agg;
+ }
+ if (pushDownAggFunctionSet.contains(aggFunction.getClass())) {
+ // CaseWhen and If (which CASE WHEN is normalized into) must both be checked.
+ // When an agg function contains an If/CaseWhen whose condition tests IS NULL
+ // (e.g. count(if(col IS NULL, value, NULL))), pushing it to the nullable side
+ // of an outer join produces wrong results: null-extended rows make "col IS NULL"
+ // TRUE at the top level, but the pre-aggregated count slot becomes NULL after
+ // null-extension, and ifnull(sum(NULL), 0) = 0 instead of the correct 1.
+ if (!hasCaseWhen && aggFunction.anyMatch(e -> e instanceof CaseWhen || e instanceof If)) {
+ hasCaseWhen = true;
+ }
Review Comment:
不会有问题. ifnull(x, y) 会作为一个整体下推.
if(a>0, b, c) 拆开推的原因是 a 可以来自左表, b/c 可以来自右表,所以拆开后可以推的范围更大.
ifnull(x, y) 相当于 if(x is null, x, y), 如果条件和 value 来自同一张表,那么也就没有拆开的必要了,只能当做整体下推.
类似的函数也都整体下推
IF(cond, a, b) | 条件为真返回 a，否则返回 b
IFNULL(a, b) | a 不为 NULL 返回 a，否则返回 b（别名：NVL）
CASE WHEN | 多分支条件逻辑，类似 switch-case
COALESCE(a, b, ...) | 返回参数列表中第一个非 NULL 的值
NULLIF(a, b) | 若 a = b 返回 NULL，否则返回 a
LEAST(a, b, ...) | 返回多个参数中的最小值，任一为 NULL 则返回 NULL
GREATEST(a, b, ...) | 返回多个参数中的最大值，任一为 NULL 则返回 NULL
NULL_OR_EMPTY(str) | 字符串为 NULL 或空串时返回 true
NOT_NULL_OR_EMPTY(str) | 字符串既不为 NULL 也不为空串时返回 true
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org