Skip to content

[fix](cost model) scale group-by-less global agg cost by BE number when partitioned #66354

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
foxtail463 wants to merge 2 commits into
base: master
Choose a base branch
from
Open

[fix](cost model) scale group-by-less global agg cost by BE number when partitioned #66354

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
19f7f33
[fix](nereids) scale group-by-less global agg cost by BE number when …
Aug 1, 2026
f9bba83
Fix regression
Aug 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion fe/fe-core/src/main/java/org/apache/doris/nereids/cost/CostModel.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -360,7 +360,9 @@ public Cost visitPhysicalHashAggregate(
exprCost / 100 + inputStatistics.getRowCount() / beNumber,
inputStatistics.getRowCount() / beNumber, 0);
} else {
int factor = aggregate.getGroupByExpressions().isEmpty() ? 1 : beNumber;
boolean isPartitioned = !aggregate.getGroupByExpressions().isEmpty()
|| aggregate.getPartitionExpressions().filter(expressions -> !expressions.isEmpty()).isPresent();
int factor = isPartitioned ? beNumber : 1;
// global
return Cost.of(context.getSessionVariable(),
exprCost / 100 + inputStatistics.getRowCount() / factor,
Expand Down
38 changes: 38 additions & 0 deletions fe/fe-core/src/test/java/org/apache/doris/nereids/cost/CostModelV1Test.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,12 +17,26 @@

package org.apache.doris.nereids.cost;

import org.apache.doris.nereids.PlanContext;
import org.apache.doris.nereids.sqltest.SqlTestBase;
import org.apache.doris.nereids.trees.expressions.Slot;
import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateParam;
import org.apache.doris.nereids.trees.plans.AggMode;
import org.apache.doris.nereids.trees.plans.AggPhase;
import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.physical.PhysicalHashAggregate;
import org.apache.doris.nereids.trees.plans.physical.PhysicalHashJoin;
import org.apache.doris.nereids.util.PlanChecker;
import org.apache.doris.nereids.util.PlanConstructor;
import org.apache.doris.statistics.Statistics;
import org.apache.doris.statistics.StatisticsBuilder;

import com.google.common.collect.ImmutableList;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.mockito.Mockito;

import java.util.Optional;

class CostModelV1Test extends SqlTestBase {

Expand All @@ -39,4 +53,28 @@ void testMaterializingCost() {
.getBestPlanTree();
p.anyMatch(j -> j instanceof PhysicalHashJoin && ((PhysicalHashJoin<?, ?>) j).getJoinType().isRightJoin());
}

@Test
void testPartitionedScalarAggregateCostUsesClusterScale() {
int originBeNumberForTest = connectContext.getSessionVariable().getBeNumberForTest();
connectContext.getSessionVariable().setBeNumberForTest(4);
try {
Plan child = PlanConstructor.newLogicalOlapScan(101, "partitioned_scalar_agg_t", 0);
Slot partitionKey = child.getOutput().get(0);
PhysicalHashAggregate<Plan> aggregate = new PhysicalHashAggregate<Plan>(
ImmutableList.of(), ImmutableList.of(partitionKey), Optional.of(ImmutableList.of(partitionKey)),
new AggregateParam(AggPhase.GLOBAL, AggMode.INPUT_TO_RESULT), false, null, false, child);
Statistics childStats = new StatisticsBuilder().setRowCount(1000).build();
PlanContext context = Mockito.mock(PlanContext.class);
Mockito.when(context.getChildStatistics(0)).thenReturn(childStats);
Mockito.when(context.getSessionVariable()).thenReturn(connectContext.getSessionVariable());

Cost cost = new CostModel(connectContext).visitPhysicalHashAggregate(aggregate, context);

Assertions.assertEquals(250, cost.getCpuCost(), 1e-9);
Assertions.assertEquals(250, cost.getMemoryCost(), 1e-9);
} finally {
connectContext.getSessionVariable().setBeNumberForTest(originBeNumberForTest);
}
}
}
42 changes: 20 additions & 22 deletions regression-test/data/shape_check/tpcds_sf100/noStatsRfPrune/query16.out
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,33 +3,31 @@
PhysicalResultSink
--PhysicalLimit[GLOBAL]
----PhysicalLimit[LOCAL]
------hashAgg[DISTINCT_GLOBAL]
------hashAgg[GLOBAL]
--------PhysicalDistribute[DistributionSpecGather]
----------hashAgg[DISTINCT_LOCAL]
------------hashAgg[GLOBAL]
--------------hashAgg[LOCAL]
----------hashAgg[GLOBAL]
------------PhysicalProject
--------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_call_center_sk = call_center.cc_call_center_sk)) otherCondition=() build RFs:RF3 cc_call_center_sk->cs_call_center_sk
----------------PhysicalProject
------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_call_center_sk = call_center.cc_call_center_sk)) otherCondition=() build RFs:RF3 cc_call_center_sk->cs_call_center_sk
------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF2 ca_address_sk->cs_ship_addr_sk
--------------------PhysicalProject
----------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF2 ca_address_sk->cs_ship_addr_sk
------------------------PhysicalProject
--------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->cs_ship_date_sk
----------------------------hashJoin[LEFT_ANTI_JOIN bucketShuffle] hashCondition=((cs1.cs_order_number = cr1.cr_order_number)) otherCondition=()
----------------------hashJoin[INNER_JOIN broadcast] hashCondition=((cs1.cs_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->cs_ship_date_sk
------------------------hashJoin[LEFT_ANTI_JOIN bucketShuffle] hashCondition=((cs1.cs_order_number = cr1.cr_order_number)) otherCondition=()
--------------------------PhysicalProject
----------------------------hashJoin[RIGHT_SEMI_JOIN shuffle] hashCondition=((cs1.cs_order_number = cs2.cs_order_number)) otherCondition=(( not (cs1.cs_warehouse_sk = cs2.cs_warehouse_sk))) build RFs:RF0 cs_order_number->cs_order_number
------------------------------PhysicalProject
--------------------------------hashJoin[RIGHT_SEMI_JOIN shuffle] hashCondition=((cs1.cs_order_number = cs2.cs_order_number)) otherCondition=(( not (cs1.cs_warehouse_sk = cs2.cs_warehouse_sk))) build RFs:RF0 cs_order_number->cs_order_number
----------------------------------PhysicalProject
------------------------------------PhysicalOlapScan[catalog_sales(cs2)] apply RFs: RF0
----------------------------------PhysicalProject
------------------------------------PhysicalOlapScan[catalog_sales(cs1)] apply RFs: RF1 RF2 RF3
--------------------------------PhysicalOlapScan[catalog_sales(cs2)] apply RFs: RF0
------------------------------PhysicalProject
--------------------------------PhysicalOlapScan[catalog_returns(cr1)]
----------------------------PhysicalProject
------------------------------filter((date_dim.d_date <= '2002-05-31') and (date_dim.d_date >= '2002-04-01'))
--------------------------------PhysicalOlapScan[date_dim]
--------------------------------PhysicalOlapScan[catalog_sales(cs1)] apply RFs: RF1 RF2 RF3
--------------------------PhysicalProject
----------------------------PhysicalOlapScan[catalog_returns(cr1)]
------------------------PhysicalProject
--------------------------filter((customer_address.ca_state = 'WV'))
----------------------------PhysicalOlapScan[customer_address]
--------------------------filter((date_dim.d_date <= '2002-05-31') and (date_dim.d_date >= '2002-04-01'))
----------------------------PhysicalOlapScan[date_dim]
--------------------PhysicalProject
----------------------filter(call_center.cc_county IN ('Barrow County', 'Daviess County', 'Luce County', 'Richland County', 'Ziebach County'))
------------------------PhysicalOlapScan[call_center]
----------------------filter((customer_address.ca_state = 'WV'))
------------------------PhysicalOlapScan[customer_address]
----------------PhysicalProject
------------------filter(call_center.cc_county IN ('Barrow County', 'Daviess County', 'Luce County', 'Richland County', 'Ziebach County'))
--------------------PhysicalOlapScan[call_center]

42 changes: 20 additions & 22 deletions regression-test/data/shape_check/tpcds_sf100/noStatsRfPrune/query94.out
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,33 +3,31 @@
PhysicalResultSink
--PhysicalLimit[GLOBAL]
----PhysicalLimit[LOCAL]
------hashAgg[DISTINCT_GLOBAL]
------hashAgg[GLOBAL]
--------PhysicalDistribute[DistributionSpecGather]
----------hashAgg[DISTINCT_LOCAL]
------------hashAgg[GLOBAL]
--------------hashAgg[LOCAL]
----------hashAgg[GLOBAL]
------------PhysicalProject
--------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_web_site_sk = web_site.web_site_sk)) otherCondition=() build RFs:RF3 web_site_sk->ws_web_site_sk
----------------PhysicalProject
------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_web_site_sk = web_site.web_site_sk)) otherCondition=() build RFs:RF3 web_site_sk->ws_web_site_sk
------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF2 ca_address_sk->ws_ship_addr_sk
--------------------PhysicalProject
----------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF2 ca_address_sk->ws_ship_addr_sk
------------------------PhysicalProject
--------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->ws_ship_date_sk
----------------------------hashJoin[LEFT_ANTI_JOIN bucketShuffle] hashCondition=((ws1.ws_order_number = wr1.wr_order_number)) otherCondition=()
----------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF1 d_date_sk->ws_ship_date_sk
------------------------hashJoin[LEFT_ANTI_JOIN bucketShuffle] hashCondition=((ws1.ws_order_number = wr1.wr_order_number)) otherCondition=()
--------------------------PhysicalProject
----------------------------hashJoin[RIGHT_SEMI_JOIN shuffle] hashCondition=((ws1.ws_order_number = ws2.ws_order_number)) otherCondition=(( not (ws1.ws_warehouse_sk = ws2.ws_warehouse_sk))) build RFs:RF0 ws_order_number->ws_order_number
------------------------------PhysicalProject
--------------------------------hashJoin[RIGHT_SEMI_JOIN shuffle] hashCondition=((ws1.ws_order_number = ws2.ws_order_number)) otherCondition=(( not (ws1.ws_warehouse_sk = ws2.ws_warehouse_sk))) build RFs:RF0 ws_order_number->ws_order_number
----------------------------------PhysicalProject
------------------------------------PhysicalOlapScan[web_sales(ws2)] apply RFs: RF0
----------------------------------PhysicalProject
------------------------------------PhysicalOlapScan[web_sales(ws1)] apply RFs: RF1 RF2 RF3
--------------------------------PhysicalOlapScan[web_sales(ws2)] apply RFs: RF0
------------------------------PhysicalProject
--------------------------------PhysicalOlapScan[web_returns(wr1)]
----------------------------PhysicalProject
------------------------------filter((date_dim.d_date <= '2000-04-01') and (date_dim.d_date >= '2000-02-01'))
--------------------------------PhysicalOlapScan[date_dim]
--------------------------------PhysicalOlapScan[web_sales(ws1)] apply RFs: RF1 RF2 RF3
--------------------------PhysicalProject
----------------------------PhysicalOlapScan[web_returns(wr1)]
------------------------PhysicalProject
--------------------------filter((customer_address.ca_state = 'OK'))
----------------------------PhysicalOlapScan[customer_address]
--------------------------filter((date_dim.d_date <= '2000-04-01') and (date_dim.d_date >= '2000-02-01'))
----------------------------PhysicalOlapScan[date_dim]
--------------------PhysicalProject
----------------------filter((web_site.web_company_name = 'pri'))
------------------------PhysicalOlapScan[web_site]
----------------------filter((customer_address.ca_state = 'OK'))
------------------------PhysicalOlapScan[customer_address]
----------------PhysicalProject
------------------filter((web_site.web_company_name = 'pri'))
--------------------PhysicalOlapScan[web_site]

44 changes: 21 additions & 23 deletions regression-test/data/shape_check/tpcds_sf100/noStatsRfPrune/query95.out
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,34 +11,32 @@ PhysicalCteAnchor ( cteId=CTEId#0 )
--PhysicalResultSink
----PhysicalLimit[GLOBAL]
------PhysicalLimit[LOCAL]
--------hashAgg[DISTINCT_GLOBAL]
--------hashAgg[GLOBAL]
----------PhysicalDistribute[DistributionSpecGather]
------------hashAgg[DISTINCT_LOCAL]
--------------hashAgg[GLOBAL]
----------------hashAgg[LOCAL]
------------hashAgg[GLOBAL]
--------------PhysicalProject
----------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_web_site_sk = web_site.web_site_sk)) otherCondition=() build RFs:RF7 web_site_sk->ws_web_site_sk
------------------PhysicalProject
--------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_web_site_sk = web_site.web_site_sk)) otherCondition=() build RFs:RF7 web_site_sk->ws_web_site_sk
--------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF6 ca_address_sk->ws_ship_addr_sk
----------------------PhysicalProject
------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_addr_sk = customer_address.ca_address_sk)) otherCondition=() build RFs:RF6 ca_address_sk->ws_ship_addr_sk
--------------------------PhysicalProject
----------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF5 d_date_sk->ws_ship_date_sk
------------------------------hashJoin[RIGHT_SEMI_JOIN shuffleBucket] hashCondition=((ws1.ws_order_number = ws_wh.ws_order_number)) otherCondition=() build RFs:RF4 ws_order_number->ws_order_number
--------------------------------PhysicalCteConsumer ( cteId=CTEId#0 ) apply RFs: RF4
--------------------------------hashJoin[RIGHT_SEMI_JOIN bucketShuffle] hashCondition=((ws1.ws_order_number = web_returns.wr_order_number)) otherCondition=() build RFs:RF2 ws_order_number->wr_order_number;RF8 ws_order_number->ws_order_number;RF9 ws_order_number->ws_order_number
----------------------------------PhysicalProject
------------------------------------hashJoin[INNER_JOIN shuffle] hashCondition=((web_returns.wr_order_number = ws_wh.ws_order_number)) otherCondition=() build RFs:RF1 wr_order_number->ws_order_number
--------------------------------------PhysicalCteConsumer ( cteId=CTEId#0 ) apply RFs: RF1
--------------------------------------PhysicalProject
----------------------------------------PhysicalOlapScan[web_returns] apply RFs: RF2
------------------------hashJoin[INNER_JOIN broadcast] hashCondition=((ws1.ws_ship_date_sk = date_dim.d_date_sk)) otherCondition=() build RFs:RF5 d_date_sk->ws_ship_date_sk
--------------------------hashJoin[RIGHT_SEMI_JOIN shuffleBucket] hashCondition=((ws1.ws_order_number = ws_wh.ws_order_number)) otherCondition=() build RFs:RF4 ws_order_number->ws_order_number
----------------------------PhysicalCteConsumer ( cteId=CTEId#0 ) apply RFs: RF4
----------------------------hashJoin[RIGHT_SEMI_JOIN bucketShuffle] hashCondition=((ws1.ws_order_number = web_returns.wr_order_number)) otherCondition=() build RFs:RF2 ws_order_number->wr_order_number;RF8 ws_order_number->ws_order_number;RF9 ws_order_number->ws_order_number
------------------------------PhysicalProject
--------------------------------hashJoin[INNER_JOIN shuffle] hashCondition=((web_returns.wr_order_number = ws_wh.ws_order_number)) otherCondition=() build RFs:RF1 wr_order_number->ws_order_number
----------------------------------PhysicalCteConsumer ( cteId=CTEId#0 ) apply RFs: RF1
----------------------------------PhysicalProject
------------------------------------PhysicalOlapScan[web_sales(ws1)] apply RFs: RF5 RF6 RF7
------------------------------------PhysicalOlapScan[web_returns] apply RFs: RF2
------------------------------PhysicalProject
--------------------------------filter((date_dim.d_date <= '1999-04-02') and (date_dim.d_date >= '1999-02-01'))
----------------------------------PhysicalOlapScan[date_dim]
--------------------------------PhysicalOlapScan[web_sales(ws1)] apply RFs: RF5 RF6 RF7
--------------------------PhysicalProject
----------------------------filter((customer_address.ca_state = 'NC'))
------------------------------PhysicalOlapScan[customer_address]
----------------------------filter((date_dim.d_date <= '1999-04-02') and (date_dim.d_date >= '1999-02-01'))
------------------------------PhysicalOlapScan[date_dim]
----------------------PhysicalProject
------------------------filter((web_site.web_company_name = 'pri'))
--------------------------PhysicalOlapScan[web_site]
------------------------filter((customer_address.ca_state = 'NC'))
--------------------------PhysicalOlapScan[customer_address]
------------------PhysicalProject
--------------------filter((web_site.web_company_name = 'pri'))
----------------------PhysicalOlapScan[web_site]

Loading
Loading