You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cloudberry.apache.org by ma...@apache.org on 2025/02/11 08:19:24 UTC
(cloudberry) branch main updated: Get streaming hash aggregate back
This is an automated email from the ASF dual-hosted git repository.
maxyang pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/cloudberry.git
The following commit(s) were added to refs/heads/main by this push:
new 36868ee497 Get streaming hash aggregate back
36868ee497 is described below
commit 36868ee4972c840988b062a6b9a6d7fc8c7335d9
Author: Adam Lee <ad...@vmware.com>
AuthorDate: Tue Mar 14 14:16:52 2023 +0800
Get streaming hash aggregate back
In multi-phase aggregate, this feature streams entries of the bottom
phase when out of memory instead of spilling to disk, it avoids the disk
I/O and roughly deduplicates to save the network I/O.
The planner always supports streaming aggregate and there are lots of
cases, but the executor part was lost while merging PostgreSQL 12, this
commit gets it back.
---
src/backend/executor/nodeAgg.c | 100 ++++++++++++++++++++++++++++++++---------
src/include/nodes/execnodes.h | 2 +
src/include/nodes/pathnodes.h | 2 +-
3 files changed, 82 insertions(+), 22 deletions(-)
diff --git a/src/backend/executor/nodeAgg.c b/src/backend/executor/nodeAgg.c
index f44186d72e..20fc5c0edb 100644
--- a/src/backend/executor/nodeAgg.c
+++ b/src/backend/executor/nodeAgg.c
@@ -244,14 +244,6 @@
*-------------------------------------------------------------------------
*/
-/*
- * GPDB_12_MERGE_FIXME: we lost the "streaming bottom" feature in the merge.
- *
- * And the detailed cdb executor instruments to print by explain.
- *
- * They were in execHHashagg.c
- */
-
#include "postgres.h"
#include "access/htup_details.h"
@@ -1946,6 +1938,13 @@ hash_agg_check_limits(AggState *aggstate)
static void
hash_agg_enter_spill_mode(AggState *aggstate)
{
+ if (aggstate->streaming)
+ {
+ /* don't spill, only mark the hash table as filled. */
+ aggstate->table_filled = true;
+ return;
+ }
+
aggstate->hash_spill_mode = true;
hashagg_recompile_expressions(aggstate, aggstate->table_filled, true);
@@ -2202,7 +2201,7 @@ lookup_hash_entries(AggState *aggstate)
bool *p_isnew;
/* if hash table already spilled, don't create new entries */
- p_isnew = aggstate->hash_spill_mode ? NULL : &isnew;
+ p_isnew = (aggstate->hash_spill_mode || (aggstate->streaming && aggstate->table_filled)) ? NULL : &isnew;
select_current_set(aggstate, setno, true);
prepare_hash_slot(perhash,
@@ -2247,13 +2246,7 @@ lookup_hash_entries(AggState *aggstate)
* stored in the expression context to be used when ExecProject evaluates
* the result tuple.
*
- * Streaming bottom: forces end of passes when no tuple for underlying node.
- *
- * MPP-2614: Btree scan will return null tuple at end of scan. However,
- * if one calls ExecProNode again on a btree scan, it will restart from
- * beginning even though we did not call rescan. This is a bug on
- * btree scan, but mask it off here for v3.1. Really should fix Btree
- * code.
+ * Streaming: forces not spilling, roughly deduplicates to save network I/O.
*/
static TupleTableSlot *
ExecAgg(PlanState *pstate)
@@ -2649,9 +2642,28 @@ agg_fill_hash_table(AggState *aggstate)
*/
for (;;)
{
+ /* hash table in memory is filled, done with this round */
+ if (aggstate->streaming && aggstate->table_filled)
+ break;
+
outerslot = fetch_input_tuple(aggstate);
if (TupIsNull(outerslot))
- break;
+ {
+ /* outer plan is done */
+ aggstate->input_done = true;
+
+ if (aggstate->hash_ngroups_current == 0)
+ {
+ /* this round got nothing but NULL */
+ aggstate->agg_done = true;
+ return;
+ }
+ else
+ {
+ /* this round got things, break to initialize the hash table */
+ break;
+ }
+ }
/* set up for lookup_hash_entries and advance_aggregates */
tmpcontext->ecxt_outertuple = outerslot;
@@ -2853,7 +2865,44 @@ agg_retrieve_hash_table(AggState *aggstate)
result = agg_retrieve_hash_table_in_memory(aggstate);
if (result == NULL)
{
- if (!agg_refill_hash_table(aggstate))
+ if (aggstate->streaming)
+ {
+ /* outer plan is already done, halt the agg */
+ if (aggstate->input_done)
+ {
+ aggstate->agg_done = true;
+ break;
+ }
+
+ /*
+ * reset hash tables and related structures, codes are copied
+ * from agg_refill_hash_table()
+ */
+ {
+ /* there could be residual pergroup pointers; clear them */
+ for (int setoff = 0;
+ setoff < aggstate->maxsets + aggstate->num_hashes;
+ setoff++)
+ aggstate->all_pergroups[setoff] = NULL;
+
+ /* free memory and reset hash tables */
+ ReScanExprContext(aggstate->hashcontext);
+ for (int setno = 0; setno < aggstate->num_hashes; setno++)
+ ResetTupleHashTable(aggstate->perhash[setno].hashtable);
+
+ aggstate->hash_ngroups_current = 0;
+
+ aggstate->table_filled = false;
+ }
+
+ /* refill the hash table from outer, since streaming doesn't spill */
+ agg_fill_hash_table(aggstate);
+
+ /* halt if outer got nothing but NULL in the above agg_fill_hash_table() */
+ if (aggstate->agg_done)
+ break;
+ }
+ else if (!agg_refill_hash_table(aggstate))
{
aggstate->agg_done = true;
break;
@@ -3413,6 +3462,7 @@ ExecInitAgg(Agg *node, EState *estate, int eflags)
aggstate->sort_in = NULL;
aggstate->sort_out = NULL;
aggstate->aggs_used = NULL;
+ aggstate->streaming = node->streaming;
/*
* phases[0] always exists, but is dummy in sorted/plain mode
@@ -4604,7 +4654,11 @@ ExecReScanAgg(AggState *node)
node->agg_done = false;
- if (node->aggstrategy == AGG_HASHED)
+ /*
+ * Greenplum: on streaming mode, the hash table's status is complex, just
+ * don't reuse it at all.
+ */
+ if (node->aggstrategy == AGG_HASHED && !node->streaming)
{
/*
* In the hashed case, if we haven't yet built the hash table then we
@@ -4680,6 +4734,8 @@ ExecReScanAgg(AggState *node)
*/
if (node->aggstrategy == AGG_HASHED || node->aggstrategy == AGG_MIXED)
{
+ node->input_done = false;
+
hashagg_reset_spill_state(node);
node->hash_ever_spilled = false;
@@ -5022,7 +5078,6 @@ ExecEagerFreeAgg(AggState *node)
node->sort_out = NULL;
}
-
hashagg_reset_spill_state(node);
if (node->hash_metacxt != NULL)
@@ -5079,6 +5134,9 @@ ReuseHashTable(AggState *node)
{
PlanState *outerPlan = outerPlanState(node);
Agg *aggnode = (Agg *) node->ss.ps.plan;
- return (outerPlan->chgParam == NULL && !node->hash_ever_spilled &&
+
+ return (outerPlan->chgParam == NULL &&
+ !node->hash_ever_spilled &&
+ !node->streaming &&
!bms_overlap(node->ss.ps.chgParam, aggnode->aggParams));
}
diff --git a/src/include/nodes/execnodes.h b/src/include/nodes/execnodes.h
index 32ae92f64a..a02a5fb79d 100644
--- a/src/include/nodes/execnodes.h
+++ b/src/include/nodes/execnodes.h
@@ -2813,6 +2813,8 @@ typedef struct AggState
SharedAggInfo *shared_info; /* one entry per worker */
Bitmapset *aggs_used; /* which aggs are used in this query */
+ /* stream entries when out of memory instead of spilling to disk */
+ bool streaming;
} AggState;
typedef struct TupleSplitState
diff --git a/src/include/nodes/pathnodes.h b/src/include/nodes/pathnodes.h
index 0deb0e310a..e37f9d8983 100644
--- a/src/include/nodes/pathnodes.h
+++ b/src/include/nodes/pathnodes.h
@@ -2226,7 +2226,7 @@ typedef struct AggPath
uint64 transitionSpace; /* for pass-by-ref transition data */
List *groupClause; /* a list of SortGroupClause's */
List *qual; /* quals (HAVING quals), if any */
- bool streaming;
+ bool streaming; /* stream entries when out of memory instead of spilling to disk */
} AggPath;
/*
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cloudberry.apache.org
For additional commands, e-mail: commits-help@cloudberry.apache.org