From a9c819421edbab1db68859f0eb4970c2759719a9 Mon Sep 17 00:00:00 2001 From: "Joshua (D) Drake" <136637981+ChronicallyJD@users.noreply.github.com> Date: Sun, 2 Aug 2026 21:11:23 -0600 Subject: [PATCH] feat: parallel int sum/avg partials (#289 phase 5/6 slice 2) Extend the parallel arm to sum/avg over int2/int4, completing the batch-eligible kinds (count/sum/avg over int + float). sum(int) emits its int8 partial (NULL until a value is seen, so the strict int8pl combine and its overflow check match core's parallel sum); avg(int) emits the int8[2] {N,sum} array int4_avg_accum builds, combined by int4_avg_combine and finalized by int8_avg -- built with construct_array + explicit INT8OID params (not construct_array_builtin, same cross-version reason as the float8 array). Integer sums and numeric avg have no float reassociation, so the parallel fold equals the serial oracle EXACTLY: parallel_vector_agg gains an exact sum(k)+avg(k) check and a plan premise. 18/18 on pg15a/16a/17a/18a/19a assert. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01UX1jrWiQsJJA1t4pkmkb4T --- src/columnar_vector.c | 50 +++++++++++++++++++++++++++++++------ test/parallel_vector_agg.sh | 10 ++++++++ 2 files changed, 53 insertions(+), 7 deletions(-) diff --git a/src/columnar_vector.c b/src/columnar_vector.c index 46ab664..4a9b991 100644 --- a/src/columnar_vector.c +++ b/src/columnar_vector.c @@ -744,13 +744,13 @@ columnar_agg_metadata_answerable(ColumnarAggKind kind) /* * columnar_parallel_agg_ok - * The first parallel slice (#289 phase 5/6): kinds whose transition state is - * a plain, non-internal value the batch fold already holds and a core - * Finalize can combine -- count(*), count(col), and sum/avg over float4/8. - * count -> int8, sum(float) -> the identity float, avg(float) -> the _float8 - * {N,Sx,Sxx} array. The int8 sum/avg-int kinds (int8[]/int8[2]) and every - * internal-transtype kind stay on the serial node or the ordinary core Agg - * for now; q6's count(*)+avg(float8) is covered. + * Kinds whose transition state is a plain, non-internal value the batch fold + * already holds and a core Finalize can combine (#289 phase 5/6): count(*), + * count(col), and sum/avg over int2/int4/float4/float8. The transition types: + * count/sum(int) -> int8, sum(float) -> the identity float, avg(int) -> the + * int8[2] {N,sum} array, avg(float) -> the _float8 {N,Sx,Sxx} array. The + * internal-transtype kinds (sum/avg over int8/numeric) are not batch-eligible + * and stay on the serial node or the ordinary core Agg. */ static bool columnar_parallel_agg_ok(ColumnarAggKind kind) @@ -759,7 +759,9 @@ columnar_parallel_agg_ok(ColumnarAggKind kind) { case COLUMNAR_AGG_COUNT_STAR: case COLUMNAR_AGG_COUNT_COL: + case COLUMNAR_AGG_SUM_INT: case COLUMNAR_AGG_SUM_FLOAT: + case COLUMNAR_AGG_AVG_INT: case COLUMNAR_AGG_AVG_FLOAT: return true; default: @@ -1836,6 +1838,21 @@ columnar_agg_emit_partial(ColumnarAggSpec *spec, bool *isnull) /* transtype int8: the running count is the partial state */ return Int64GetDatum(spec->count); + case COLUMNAR_AGG_SUM_INT: + + /* + * sum(int2/int4) -> int8 transtype. NULL until a value is seen: core's + * sum(int) carries a NULL state for an empty input and its strict int8pl + * combine then treats a worker that matched no rows as absent, so the + * cross-worker sum and its overflow check match an ordinary parallel sum. + */ + if (!spec->sawValue) + { + *isnull = true; + return (Datum) 0; + } + return Int64GetDatum(spec->sum); + case COLUMNAR_AGG_SUM_FLOAT: /* @@ -1878,6 +1895,25 @@ columnar_agg_emit_partial(ColumnarAggSpec *spec, bool *isnull) TYPALIGN_DOUBLE)); } + case COLUMNAR_AGG_AVG_INT: + { + /* + * avg(int2/int4) -> _int8 {N, sum}: the same array int4_avg_accum + * builds and int4_avg_combine merges, finalized by int8_avg to + * numeric. Always non-null -- an empty worker emits {0,0}, which + * int8_avg maps to NULL after the combine. int8's by-value/alignment + * follow FLOAT8PASSBYVAL / TYPALIGN_DOUBLE, exactly as core builds it. + */ + Datum elems[2]; + + elems[0] = Int64GetDatum(spec->count); + elems[1] = Int64GetDatum(spec->sum); + return PointerGetDatum(construct_array(elems, 2, INT8OID, + sizeof(int64), + FLOAT8PASSBYVAL, + TYPALIGN_DOUBLE)); + } + default: /* diff --git a/test/parallel_vector_agg.sh b/test/parallel_vector_agg.sh index 237369f..d4f741b 100644 --- a/test/parallel_vector_agg.sh +++ b/test/parallel_vector_agg.sh @@ -74,6 +74,16 @@ check "avg(v) parallel-vec ~= core parallel agg" "$(reldiff 'avg(v)')" ok check "sum(v) parallel-vec ~= core parallel agg" "$(reldiff 'sum(v)')" ok check "avg(w::float8) f4 ~= core parallel agg" "$(reldiff 'avg(w)::float8')" ok +# int sum/avg (#289 phase 5/6): sum(int)->int8, avg(int)->numeric. Integer sums +# and numeric division have no float reassociation, so the parallel fold must +# equal the serial oracle EXACTLY, and the plan must be the parallel fold. +PLAN_I="$(q -c "$PAR $UG $PP" -c "EXPLAIN (COSTS OFF) SELECT sum(k), avg(k) FROM t WHERE k < 700")" +check "premise: int sum/avg takes the parallel fold" \ + "$(printf '%s' "$PLAN_I" | grep -qiE 'Gather' && printf '%s' "$PLAN_I" | grep -qi 'Batch Fold: yes' && echo y || echo n)" y +IK_VEC="$(q -c "$PAR $UG $PP" -c "SELECT sum(k), avg(k) FROM t WHERE k < 700")" +IK_SER="$(q -c "SET max_parallel_workers_per_gather=0;" -c "SELECT sum(k), avg(k) FROM t WHERE k < 700")" +check "sum(k)+avg(k) int: parallel fold == serial (exact)" "$IK_VEC" "$IK_SER" + # ---- edge cases ------------------------------------------------------------ # a NULL test is not batch-foldable: the partial must still be correct on the # row path (shape ineligible from the start -> shared counter untouched).