This is an automated email from the ASF dual-hosted git repository.
jrgemignani pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/age.git
The following commit(s) were added to refs/heads/master by this push:
new 92e48e9b Add reduce() list folding function (#2435)
92e48e9b is described below
commit 92e48e9b907f491c92db487790c1624dba23b812
Author: John Gemignani <[email protected]>
AuthorDate: Fri Jun 26 15:54:54 2026 -0700
Add reduce() list folding function (#2435)
Implement the openCypher reduce(acc = init, var IN list | body) expression,
which folds an arbitrary expression over a list, threading an accumulator
across the elements in list order. This closes a long-standing gap (reduce()
was previously unsupported) and works both at the SQL top level and inside a
cypher() RETURN/WHERE.
Implementation
--------------
reduce() is desugared, at transform time, into a correlated scalar subquery
over a new ordered aggregate rather than a new executor node, so no
PostgreSQL
core changes are required:
CASE WHEN list IS NULL THEN NULL
ELSE COALESCE((SELECT ag_catalog.age_reduce(
<init>, '<serialized-body>'::text,
r.elem ORDER BY r.ord)
FROM unnest(<list>) WITH ORDINALITY AS r(elem,
ord)),
<init>)
END
- A new cypher_reduce extensible node carries the accumulator/element
names
and the init/list/body expressions (grammar production, keyword, and the
copy/out/read serialization plumbing).
- The fold body is transformed against a throwaway two-column agtype
namespace, its accumulator and element Var references are rewritten to
PARAM_EXEC params 0 and 1, and it is serialized with nodeToString()
into a
text argument.
- age_reduce_transfn (a custom agtype aggregate transition function)
deserializes and compiles the body once per group with ExecInitExpr,
then
evaluates it per element with ExecEvalExpr, rebinding the two params.
The
body is normalized to agtype at transform time so a boolean or other
non-agtype result cannot be misread as a by-reference Datum.
Semantics
---------
- List order is preserved (unnest WITH ORDINALITY + aggregate ORDER BY).
- An empty list yields the initial value; a NULL list yields NULL.
- The list and initial value may reference outer-query variables (e.g.
reduce(total = 0, n IN nodes(p) | total + n.age)); the body may
reference
only the accumulator and element.
- Arithmetic, string, list-building, boolean/comparison (AND/OR/=/>),
CASE,
and element property-access bodies are all supported.
- Outer-variable, query-parameter, nested-reduce, and aggregate references
inside the body raise a clean ERRCODE_FEATURE_NOT_SUPPORTED error.
- reduce is registered as a safe keyword so it remains usable as a
property
or map key, preserving backward compatibility.
Tests
-----
Adds the age_reduce regression test (registered in the install SQL and the
upgrade template so age_upgrade passes), covering: arithmetic/product/string
folds; order sensitivity; empty/NULL list; NULL element and NULL init;
list-building and CASE bodies; boolean and comparison bodies; element
property
access; multiple and nested (in list/init) reduce(); reduce() in boolean
expressions, WHERE, and list comprehensions; folds over collected nodes and
node list properties; the not-supported rejections; and reduce as a map key.
Following reviewer feedback, three further semantics-coverage gaps are
pinned directly so the mechanisms that make the aggregate desugaring
correct are exercised by tests rather than only correct by inspection:
- A fold body that produces null mid-fold and then recovers: the
agtype 'null' running state is a readable value, so a later element
folds back out of it (distinct from "null propagates to a null
result", which was already covered).
- An empty list with a NULL initial value: COALESCE(<no rows>, init)
yields NULL, kept distinct from a body that legitimately folds to
agtype 'null', which must not be resurrected to the initial value.
- A type error and a runtime division-by-zero error in the body: both
abort cleanly out of the standalone per-element evaluator rather
than corrupting the running aggregate state.
All multi-row results are ordered. 42/42 installcheck pass.
Future work
-----------
The body restriction (accumulator and element only) is a property of the
standalone expression evaluation and can be relaxed without core changes:
- Allow loop-invariant outer-variable and cypher $parameter references in
the body by capturing them as additional eager aggregate arguments bound
to extra param slots.
- Support a nested reduce() inside the body via an SPI-based evaluation
fallback for subquery-bearing bodies.
Aggregates inside the body remain intentionally unsupported, matching the
openCypher specification.
Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
modified: Makefile
modified: age--1.7.0--y.y.y.sql
new file: regress/expected/age_reduce.out
new file: regress/sql/age_reduce.sql
modified: sql/age_aggregate.sql
modified: src/backend/nodes/ag_nodes.c
modified: src/backend/nodes/cypher_copyfuncs.c
modified: src/backend/nodes/cypher_outfuncs.c
modified: src/backend/nodes/cypher_readfuncs.c
modified: src/backend/parser/cypher_analyze.c
modified: src/backend/parser/cypher_clause.c
modified: src/backend/parser/cypher_gram.y
modified: src/backend/utils/adt/agtype.c
modified: src/include/nodes/ag_nodes.h
modified: src/include/nodes/cypher_copyfuncs.h
modified: src/include/nodes/cypher_nodes.h
modified: src/include/nodes/cypher_outfuncs.h
modified: src/include/nodes/cypher_readfuncs.h
modified: src/include/parser/cypher_kwlist.h
---
Makefile | 1 +
age--1.7.0--y.y.y.sql | 22 ++
regress/expected/age_reduce.out | 538 +++++++++++++++++++++++++++++++++++
regress/sql/age_reduce.sql | 350 +++++++++++++++++++++++
sql/age_aggregate.sql | 22 ++
src/backend/nodes/ag_nodes.c | 6 +-
src/backend/nodes/cypher_copyfuncs.c | 12 +
src/backend/nodes/cypher_outfuncs.c | 12 +
src/backend/nodes/cypher_readfuncs.c | 14 +
src/backend/parser/cypher_analyze.c | 19 ++
src/backend/parser/cypher_clause.c | 421 +++++++++++++++++++++++++++
src/backend/parser/cypher_gram.y | 92 +++++-
src/backend/utils/adt/agtype.c | 165 +++++++++++
src/include/nodes/ag_nodes.h | 4 +-
src/include/nodes/cypher_copyfuncs.h | 4 +
src/include/nodes/cypher_nodes.h | 18 ++
src/include/nodes/cypher_outfuncs.h | 1 +
src/include/nodes/cypher_readfuncs.h | 3 +
src/include/parser/cypher_kwlist.h | 1 +
19 files changed, 1701 insertions(+), 4 deletions(-)
diff --git a/Makefile b/Makefile
index b0059213..21c7f81f 100644
--- a/Makefile
+++ b/Makefile
@@ -216,6 +216,7 @@ REGRESS = scan \
list_comprehension \
predicate_functions \
pattern_expression \
+ age_reduce \
map_projection \
direct_field_access \
security \
diff --git a/age--1.7.0--y.y.y.sql b/age--1.7.0--y.y.y.sql
index b40cde09..ec909d2b 100644
--- a/age--1.7.0--y.y.y.sql
+++ b/age--1.7.0--y.y.y.sql
@@ -1100,3 +1100,25 @@ $function$;
COMMENT ON FUNCTION ag_catalog.create_subgraph(name, name, text, text) IS
'Materializes a new persistent graph as the induced subgraph of from_graph
selected by a Cypher node predicate (on n) and relationship predicate (on r);
''*'' keeps all. An edge is kept only if its predicate holds and both endpoints
are kept. Returns (node_count, relationship_count).';
+
+--
+-- reduce(acc = init, var IN list | body) fold support
+--
+-- Transition function for the age_reduce aggregate. The fold body is compiled
+-- by transform_cypher_reduce() with the accumulator and element rewritten to
+-- PARAM_EXEC params 0 and 1 and serialized into the text argument; the
+-- transition evaluates it for each element in list order. It must be callable
+-- with a NULL transition state (no initcond), so it is intentionally not
STRICT.
+CREATE FUNCTION ag_catalog.age_reduce_transfn(agtype, agtype, text, agtype)
+ RETURNS agtype
+ LANGUAGE c
+PARALLEL UNSAFE
+AS 'MODULE_PATHNAME';
+
+-- aggregate definition for reduce(); direct arguments are
+-- (init, serialized-body, element), with the element fed ORDER BY ordinality.
+CREATE AGGREGATE ag_catalog.age_reduce(agtype, text, agtype)
+(
+ stype = agtype,
+ sfunc = ag_catalog.age_reduce_transfn
+);
diff --git a/regress/expected/age_reduce.out b/regress/expected/age_reduce.out
new file mode 100644
index 00000000..8a198965
--- /dev/null
+++ b/regress/expected/age_reduce.out
@@ -0,0 +1,538 @@
+/*
+ * 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.
+ */
+LOAD 'age';
+SET search_path TO ag_catalog;
+SELECT create_graph('reduce');
+NOTICE: graph "reduce" has been created
+ create_graph
+--------------
+
+(1 row)
+
+--
+-- Basic folds
+--
+-- sum
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 0, x IN [1, 2, 3] | s + x)
+$$) AS (result agtype);
+ result
+--------
+ 6
+(1 row)
+
+-- sum of a longer list
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 0, x IN [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] | s + x)
+$$) AS (result agtype);
+ result
+--------
+ 55
+(1 row)
+
+-- product (factorial)
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(p = 1, x IN [1, 2, 3, 4, 5] | p * x)
+$$) AS (result agtype);
+ result
+--------
+ 120
+(1 row)
+
+-- non-zero initial accumulator
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 100, x IN [1, 2, 3] | s + x)
+$$) AS (result agtype);
+ result
+--------
+ 106
+(1 row)
+
+-- single element
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 0, x IN [42] | s + x)
+$$) AS (result agtype);
+ result
+--------
+ 42
+(1 row)
+
+--
+-- List order is significant
+--
+-- left-associative subtraction: ((((0-1)-2)-3)-4) = -10
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 0, x IN [1, 2, 3, 4] | s - x)
+$$) AS (result agtype);
+ result
+--------
+ -10
+(1 row)
+
+-- forward string concatenation
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = '', x IN ['a', 'b', 'c'] | s + x)
+$$) AS (result agtype);
+ result
+--------
+ "abc"
+(1 row)
+
+-- reverse string concatenation (element before accumulator)
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = '', x IN ['a', 'b', 'c'] | x + s)
+$$) AS (result agtype);
+ result
+--------
+ "cba"
+(1 row)
+
+--
+-- Empty and NULL list semantics
+--
+-- empty list returns the initial value
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 0, x IN [] | s + x)
+$$) AS (result agtype);
+ result
+--------
+ 0
+(1 row)
+
+-- empty list returns the initial value (non-zero)
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 999, x IN [] | s + x)
+$$) AS (result agtype);
+ result
+--------
+ 999
+(1 row)
+
+-- NULL list returns NULL
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 0, x IN null | s + x)
+$$) AS (result agtype);
+ result
+--------
+
+(1 row)
+
+-- empty list with a NULL initial value yields NULL: the list is empty (not
+-- null) so the fold runs over zero rows, and COALESCE(<no rows>, init) is
+-- COALESCE(NULL, NULL) -> NULL. (Distinct from a NULL *list*, which the outer
+-- CASE short-circuits to NULL, and from a non-empty list with a NULL init,
+-- which seeds the accumulator with agtype 'null'.)
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = null, x IN [] | s + x)
+$$) AS (result agtype);
+ result
+--------
+
+(1 row)
+
+--
+-- NULL handling within the fold
+--
+-- a NULL element propagates through arithmetic to NULL
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 0, x IN [1, null, 3] | s + x)
+$$) AS (result agtype);
+ result
+--------
+ null
+(1 row)
+
+-- NULL initial value
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = null, x IN [1, 2, 3] | s)
+$$) AS (result agtype);
+ result
+--------
+ null
+(1 row)
+
+-- a body that always evaluates to null yields null, NOT the initial value:
+-- every step stores agtype 'null' as the running state, so the final state is
+-- a real agtype 'null' and the empty-list COALESCE(..., init) guard must not
+-- resurrect the initial value here (the load-bearing fold-to-null vs
empty-list
+-- distinction)
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 7, x IN [1, 2, 3] | null)
+$$) AS (result agtype);
+ result
+--------
+ null
+(1 row)
+
+-- the accumulator legitimately becomes null mid-fold and the body climbs back
+-- out of it: element 2 sets the accumulator to null, element 3 produces a
fresh
+-- non-null value, and element 4 reads that recovered state (999 + 4), proving
a
+-- null intermediate state does not poison the rest of the fold
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 0, x IN [1, 2, 3, 4] |
+ CASE WHEN x = 2 THEN null
+ WHEN x = 3 THEN 999
+ ELSE s + x END)
+$$) AS (result agtype);
+ result
+--------
+ 1003
+(1 row)
+
+--
+-- Errors raised from the fold body propagate cleanly
+--
+-- a type error in the body (agtype number + map) aborts the statement rather
+-- than corrupting the running aggregate state or crashing the backend
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 0, x IN [1, 2] | s + {a: 1})
+$$) AS (result agtype);
+ERROR: invalid left operand for agtype concatenation
+-- a runtime arithmetic error in the body (division by zero) likewise aborts
+-- the fold; the error surfaces from the standalone per-element evaluator
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 1, x IN [1, 0, 2] | s / x)
+$$) AS (result agtype);
+ERROR: division by zero
+--
+-- Building a list with the accumulator
+--
+-- collect squares
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(acc = [], x IN [1, 2, 3] | acc + [x * x])
+$$) AS (result agtype);
+ result
+-----------
+ [1, 4, 9]
+(1 row)
+
+--
+-- A conditional body (CASE)
+--
+-- sum of even elements only
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 0, x IN [1, 2, 3, 4, 5, 6] | CASE WHEN x % 2 = 0 THEN s
+ x ELSE s END)
+$$) AS (result agtype);
+ result
+--------
+ 12
+(1 row)
+
+--
+-- Boolean and comparison fold bodies
+--
+-- the body evaluates to a boolean, which is normalized to an agtype boolean
+-- (a boolean accumulator is a real Cypher use case for "all"/"any" style
folds)
+-- logical AND fold: all true?
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = true, x IN [true, true, false] | s AND x)
+$$) AS (result agtype);
+ result
+--------
+ false
+(1 row)
+
+-- logical OR fold: any true?
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = false, x IN [false, true, false] | s OR x)
+$$) AS (result agtype);
+ result
+--------
+ true
+(1 row)
+
+-- a comparison body
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = false, x IN [1, 2, 3] | x = 2)
+$$) AS (result agtype);
+ result
+--------
+ false
+(1 row)
+
+-- "does any element equal 2?" (search fold)
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(found = false, x IN [1, 2, 3] | found OR x = 2)
+$$) AS (result agtype);
+ result
+--------
+ true
+(1 row)
+
+-- "are all elements positive?" (using a comparison inside the fold)
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = true, x IN [1, 2, 3] | s AND x > 0)
+$$) AS (result agtype);
+ result
+--------
+ true
+(1 row)
+
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = true, x IN [1, -2, 3] | s AND x > 0)
+$$) AS (result agtype);
+ result
+--------
+ false
+(1 row)
+
+--
+-- Property access on the element variable
+--
+-- sum a field across a list of maps
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 0, x IN [{n: 1}, {n: 2}, {n: 3}] | s + x.n)
+$$) AS (result agtype);
+ result
+--------
+ 6
+(1 row)
+
+-- concatenate a string field across a list of maps
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = '', x IN [{w: 'a'}, {w: 'b'}, {w: 'c'}] | s + x.w)
+$$) AS (result agtype);
+ result
+--------
+ "abc"
+(1 row)
+
+--
+-- Multiple reduce() in one expression
+--
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 0, x IN [1, 2, 3] | s + x) + reduce(p = 1, y IN [2, 3] |
p * y)
+$$) AS (result agtype);
+ result
+--------
+ 12
+(1 row)
+
+--
+-- reduce() in a boolean expression
+--
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 0, x IN [1, 2, 3] | s + x) > 5
+ AND reduce(p = 1, y IN [2, 3] | p * y) < 10
+$$) AS (result agtype);
+ result
+--------
+ true
+(1 row)
+
+--
+-- reduce() nested in the list or initial value of another reduce()
+--
+-- nesting is allowed in the list and the initial value (both evaluated in the
+-- outer context) even though it is rejected inside the fold body.
+-- nested reduce() in the list
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 0, x IN [reduce(a = 0, y IN [1, 2, 3] | a + y), 10] | s
+ x)
+$$) AS (result agtype);
+ result
+--------
+ 16
+(1 row)
+
+-- nested reduce() in the initial value
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = reduce(a = 0, y IN [1, 2, 3] | a + y), x IN [10, 20] | s
+ x)
+$$) AS (result agtype);
+ result
+--------
+ 36
+(1 row)
+
+--
+-- reduce() over a correlated (per-row) list
+--
+SELECT * FROM cypher('reduce', $$
+ UNWIND [[1, 2], [3, 4, 5], []] AS arr
+ RETURN reduce(s = 0, x IN arr | s + x) AS total
+ ORDER BY total
+$$) AS (result agtype);
+ result
+--------
+ 0
+ 3
+ 12
+(3 rows)
+
+--
+-- reduce() with the list and initial value bound in an outer clause
+--
+SELECT * FROM cypher('reduce', $$
+ WITH [10, 20, 30] AS ns
+ RETURN reduce(t = 0, n IN ns | t + n)
+$$) AS (result agtype);
+ result
+--------
+ 60
+(1 row)
+
+-- the initial value may reference an outer variable (correlation is allowed
+-- in the init and the list, only not in the body)
+SELECT * FROM cypher('reduce', $$
+ WITH 5 AS base
+ RETURN reduce(s = base, x IN [1, 2, 3] | s + x)
+$$) AS (result agtype);
+ result
+--------
+ 11
+(1 row)
+
+--
+-- reduce() nested inside a list comprehension
+--
+SELECT * FROM cypher('reduce', $$
+ RETURN [v IN [1, 2, 3] | reduce(s = 0, x IN [v, v, v] | s + x)]
+$$) AS (result agtype);
+ result
+-----------
+ [3, 6, 9]
+(1 row)
+
+--
+-- reduce() in a WHERE clause
+--
+SELECT * FROM cypher('reduce', $$
+ UNWIND [[1, 2, 3], [1, 1], [10]] AS l
+ WITH l WHERE reduce(s = 0, x IN l | s + x) > 3
+ RETURN l
+ ORDER BY l
+$$) AS (result agtype);
+ result
+-----------
+ [1, 2, 3]
+ [10]
+(2 rows)
+
+--
+-- reduce() over graph data (the canonical Cypher example)
+--
+SELECT * FROM cypher('reduce', $$
+ CREATE (:person {name: 'Alice', age: 38}),
+ (:person {name: 'Bob', age: 25}),
+ (:person {name: 'Daniel', age: 54})
+$$) AS (result agtype);
+ result
+--------
+(0 rows)
+
+-- sum the ages of all person nodes
+SELECT * FROM cypher('reduce', $$
+ MATCH (p:person)
+ WITH collect(p) AS people
+ RETURN reduce(total = 0, n IN people | total + n.age)
+$$) AS (result agtype);
+ result
+--------
+ 117
+(1 row)
+
+--
+-- reduce() over a graph node's list property
+--
+SELECT * FROM cypher('reduce', $$
+ CREATE (:bag {name: 'low', vals: [1, 2, 3]}),
+ (:bag {name: 'mid', vals: [5, 5, 5]}),
+ (:bag {name: 'high', vals: [10, 20, 30]})
+$$) AS (result agtype);
+ result
+--------
+(0 rows)
+
+-- filter nodes by a reduce() over their list property
+SELECT * FROM cypher('reduce', $$
+ MATCH (u:bag) WHERE reduce(s = 0, x IN u.vals | s + x) > 10
+ RETURN u.name
+ ORDER BY u.name
+$$) AS (result agtype);
+ result
+--------
+ "high"
+ "mid"
+(2 rows)
+
+-- compute a reduce() value per node and order by it
+SELECT * FROM cypher('reduce', $$
+ MATCH (u:bag)
+ RETURN u.name AS name, reduce(s = 0, x IN u.vals | s + x) AS total
+ ORDER BY total
+$$) AS (name agtype, total agtype);
+ name | total
+--------+-------
+ "low" | 6
+ "mid" | 15
+ "high" | 60
+(3 rows)
+
+--
+-- Not-yet-supported constructs raise a clean feature error
+--
+-- an outer variable referenced in the body
+SELECT * FROM cypher('reduce', $$
+ WITH 5 AS w
+ RETURN reduce(s = 0, x IN [1, 2] | s + x + w)
+$$) AS (result agtype);
+ERROR: a reduce() expression may only reference its accumulator and element
variables
+LINE 1: SELECT * FROM cypher('reduce', $$
+ ^
+-- a nested reduce() in the body
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 0, x IN [1, 2] | s + reduce(t = 0, y IN [x] | t + y))
+$$) AS (result agtype);
+ERROR: subqueries (including a nested reduce()) are not supported in a
reduce() expression
+LINE 1: SELECT * FROM cypher('reduce', $$
+ ^
+-- an aggregate function in the body
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 0, x IN [1, 2] | s + count(x))
+$$) AS (result agtype);
+ERROR: aggregate functions are not supported in a reduce() expression
+LINE 1: SELECT * FROM cypher('reduce', $$
+ ^
+--
+-- "reduce" as a property key name (safe_keywords backward compatibility):
+-- because reduce() introduced a reserved keyword, confirm the word is still
+-- usable as a map key, the same way any/none/single are.
+--
+SELECT * FROM cypher('reduce', $$
+ RETURN {reduce: 1, any: 2, none: 3}
+$$) AS (result agtype);
+ result
+------------------------------------
+ {"any": 2, "none": 3, "reduce": 1}
+(1 row)
+
+--
+-- Cleanup
+--
+SELECT * FROM drop_graph('reduce', true);
+NOTICE: drop cascades to 4 other objects
+DETAIL: drop cascades to table reduce._ag_label_vertex
+drop cascades to table reduce._ag_label_edge
+drop cascades to table reduce.person
+drop cascades to table reduce.bag
+NOTICE: graph "reduce" has been dropped
+ drop_graph
+------------
+
+(1 row)
+
diff --git a/regress/sql/age_reduce.sql b/regress/sql/age_reduce.sql
new file mode 100644
index 00000000..cf126101
--- /dev/null
+++ b/regress/sql/age_reduce.sql
@@ -0,0 +1,350 @@
+/*
+ * 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.
+ */
+
+LOAD 'age';
+SET search_path TO ag_catalog;
+
+SELECT create_graph('reduce');
+
+--
+-- Basic folds
+--
+-- sum
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 0, x IN [1, 2, 3] | s + x)
+$$) AS (result agtype);
+
+-- sum of a longer list
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 0, x IN [1, 2, 3, 4, 5, 6, 7, 8, 9, 10] | s + x)
+$$) AS (result agtype);
+
+-- product (factorial)
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(p = 1, x IN [1, 2, 3, 4, 5] | p * x)
+$$) AS (result agtype);
+
+-- non-zero initial accumulator
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 100, x IN [1, 2, 3] | s + x)
+$$) AS (result agtype);
+
+-- single element
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 0, x IN [42] | s + x)
+$$) AS (result agtype);
+
+--
+-- List order is significant
+--
+-- left-associative subtraction: ((((0-1)-2)-3)-4) = -10
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 0, x IN [1, 2, 3, 4] | s - x)
+$$) AS (result agtype);
+
+-- forward string concatenation
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = '', x IN ['a', 'b', 'c'] | s + x)
+$$) AS (result agtype);
+
+-- reverse string concatenation (element before accumulator)
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = '', x IN ['a', 'b', 'c'] | x + s)
+$$) AS (result agtype);
+
+--
+-- Empty and NULL list semantics
+--
+-- empty list returns the initial value
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 0, x IN [] | s + x)
+$$) AS (result agtype);
+
+-- empty list returns the initial value (non-zero)
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 999, x IN [] | s + x)
+$$) AS (result agtype);
+
+-- NULL list returns NULL
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 0, x IN null | s + x)
+$$) AS (result agtype);
+
+-- empty list with a NULL initial value yields NULL: the list is empty (not
+-- null) so the fold runs over zero rows, and COALESCE(<no rows>, init) is
+-- COALESCE(NULL, NULL) -> NULL. (Distinct from a NULL *list*, which the outer
+-- CASE short-circuits to NULL, and from a non-empty list with a NULL init,
+-- which seeds the accumulator with agtype 'null'.)
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = null, x IN [] | s + x)
+$$) AS (result agtype);
+
+--
+-- NULL handling within the fold
+--
+-- a NULL element propagates through arithmetic to NULL
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 0, x IN [1, null, 3] | s + x)
+$$) AS (result agtype);
+
+-- NULL initial value
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = null, x IN [1, 2, 3] | s)
+$$) AS (result agtype);
+
+-- a body that always evaluates to null yields null, NOT the initial value:
+-- every step stores agtype 'null' as the running state, so the final state is
+-- a real agtype 'null' and the empty-list COALESCE(..., init) guard must not
+-- resurrect the initial value here (the load-bearing fold-to-null vs
empty-list
+-- distinction)
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 7, x IN [1, 2, 3] | null)
+$$) AS (result agtype);
+
+-- the accumulator legitimately becomes null mid-fold and the body climbs back
+-- out of it: element 2 sets the accumulator to null, element 3 produces a
fresh
+-- non-null value, and element 4 reads that recovered state (999 + 4), proving
a
+-- null intermediate state does not poison the rest of the fold
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 0, x IN [1, 2, 3, 4] |
+ CASE WHEN x = 2 THEN null
+ WHEN x = 3 THEN 999
+ ELSE s + x END)
+$$) AS (result agtype);
+
+--
+-- Errors raised from the fold body propagate cleanly
+--
+-- a type error in the body (agtype number + map) aborts the statement rather
+-- than corrupting the running aggregate state or crashing the backend
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 0, x IN [1, 2] | s + {a: 1})
+$$) AS (result agtype);
+
+-- a runtime arithmetic error in the body (division by zero) likewise aborts
+-- the fold; the error surfaces from the standalone per-element evaluator
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 1, x IN [1, 0, 2] | s / x)
+$$) AS (result agtype);
+
+--
+-- Building a list with the accumulator
+--
+-- collect squares
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(acc = [], x IN [1, 2, 3] | acc + [x * x])
+$$) AS (result agtype);
+
+--
+-- A conditional body (CASE)
+--
+-- sum of even elements only
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 0, x IN [1, 2, 3, 4, 5, 6] | CASE WHEN x % 2 = 0 THEN s
+ x ELSE s END)
+$$) AS (result agtype);
+
+--
+-- Boolean and comparison fold bodies
+--
+-- the body evaluates to a boolean, which is normalized to an agtype boolean
+-- (a boolean accumulator is a real Cypher use case for "all"/"any" style
folds)
+-- logical AND fold: all true?
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = true, x IN [true, true, false] | s AND x)
+$$) AS (result agtype);
+
+-- logical OR fold: any true?
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = false, x IN [false, true, false] | s OR x)
+$$) AS (result agtype);
+
+-- a comparison body
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = false, x IN [1, 2, 3] | x = 2)
+$$) AS (result agtype);
+
+-- "does any element equal 2?" (search fold)
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(found = false, x IN [1, 2, 3] | found OR x = 2)
+$$) AS (result agtype);
+
+-- "are all elements positive?" (using a comparison inside the fold)
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = true, x IN [1, 2, 3] | s AND x > 0)
+$$) AS (result agtype);
+
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = true, x IN [1, -2, 3] | s AND x > 0)
+$$) AS (result agtype);
+
+--
+-- Property access on the element variable
+--
+-- sum a field across a list of maps
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 0, x IN [{n: 1}, {n: 2}, {n: 3}] | s + x.n)
+$$) AS (result agtype);
+
+-- concatenate a string field across a list of maps
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = '', x IN [{w: 'a'}, {w: 'b'}, {w: 'c'}] | s + x.w)
+$$) AS (result agtype);
+
+--
+-- Multiple reduce() in one expression
+--
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 0, x IN [1, 2, 3] | s + x) + reduce(p = 1, y IN [2, 3] |
p * y)
+$$) AS (result agtype);
+
+--
+-- reduce() in a boolean expression
+--
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 0, x IN [1, 2, 3] | s + x) > 5
+ AND reduce(p = 1, y IN [2, 3] | p * y) < 10
+$$) AS (result agtype);
+
+--
+-- reduce() nested in the list or initial value of another reduce()
+--
+-- nesting is allowed in the list and the initial value (both evaluated in the
+-- outer context) even though it is rejected inside the fold body.
+-- nested reduce() in the list
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 0, x IN [reduce(a = 0, y IN [1, 2, 3] | a + y), 10] | s
+ x)
+$$) AS (result agtype);
+
+-- nested reduce() in the initial value
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = reduce(a = 0, y IN [1, 2, 3] | a + y), x IN [10, 20] | s
+ x)
+$$) AS (result agtype);
+
+--
+-- reduce() over a correlated (per-row) list
+--
+SELECT * FROM cypher('reduce', $$
+ UNWIND [[1, 2], [3, 4, 5], []] AS arr
+ RETURN reduce(s = 0, x IN arr | s + x) AS total
+ ORDER BY total
+$$) AS (result agtype);
+
+--
+-- reduce() with the list and initial value bound in an outer clause
+--
+SELECT * FROM cypher('reduce', $$
+ WITH [10, 20, 30] AS ns
+ RETURN reduce(t = 0, n IN ns | t + n)
+$$) AS (result agtype);
+
+-- the initial value may reference an outer variable (correlation is allowed
+-- in the init and the list, only not in the body)
+SELECT * FROM cypher('reduce', $$
+ WITH 5 AS base
+ RETURN reduce(s = base, x IN [1, 2, 3] | s + x)
+$$) AS (result agtype);
+
+--
+-- reduce() nested inside a list comprehension
+--
+SELECT * FROM cypher('reduce', $$
+ RETURN [v IN [1, 2, 3] | reduce(s = 0, x IN [v, v, v] | s + x)]
+$$) AS (result agtype);
+
+--
+-- reduce() in a WHERE clause
+--
+SELECT * FROM cypher('reduce', $$
+ UNWIND [[1, 2, 3], [1, 1], [10]] AS l
+ WITH l WHERE reduce(s = 0, x IN l | s + x) > 3
+ RETURN l
+ ORDER BY l
+$$) AS (result agtype);
+
+--
+-- reduce() over graph data (the canonical Cypher example)
+--
+SELECT * FROM cypher('reduce', $$
+ CREATE (:person {name: 'Alice', age: 38}),
+ (:person {name: 'Bob', age: 25}),
+ (:person {name: 'Daniel', age: 54})
+$$) AS (result agtype);
+
+-- sum the ages of all person nodes
+SELECT * FROM cypher('reduce', $$
+ MATCH (p:person)
+ WITH collect(p) AS people
+ RETURN reduce(total = 0, n IN people | total + n.age)
+$$) AS (result agtype);
+
+--
+-- reduce() over a graph node's list property
+--
+SELECT * FROM cypher('reduce', $$
+ CREATE (:bag {name: 'low', vals: [1, 2, 3]}),
+ (:bag {name: 'mid', vals: [5, 5, 5]}),
+ (:bag {name: 'high', vals: [10, 20, 30]})
+$$) AS (result agtype);
+
+-- filter nodes by a reduce() over their list property
+SELECT * FROM cypher('reduce', $$
+ MATCH (u:bag) WHERE reduce(s = 0, x IN u.vals | s + x) > 10
+ RETURN u.name
+ ORDER BY u.name
+$$) AS (result agtype);
+
+-- compute a reduce() value per node and order by it
+SELECT * FROM cypher('reduce', $$
+ MATCH (u:bag)
+ RETURN u.name AS name, reduce(s = 0, x IN u.vals | s + x) AS total
+ ORDER BY total
+$$) AS (name agtype, total agtype);
+
+--
+-- Not-yet-supported constructs raise a clean feature error
+--
+-- an outer variable referenced in the body
+SELECT * FROM cypher('reduce', $$
+ WITH 5 AS w
+ RETURN reduce(s = 0, x IN [1, 2] | s + x + w)
+$$) AS (result agtype);
+
+-- a nested reduce() in the body
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 0, x IN [1, 2] | s + reduce(t = 0, y IN [x] | t + y))
+$$) AS (result agtype);
+
+-- an aggregate function in the body
+SELECT * FROM cypher('reduce', $$
+ RETURN reduce(s = 0, x IN [1, 2] | s + count(x))
+$$) AS (result agtype);
+
+--
+-- "reduce" as a property key name (safe_keywords backward compatibility):
+-- because reduce() introduced a reserved keyword, confirm the word is still
+-- usable as a map key, the same way any/none/single are.
+--
+SELECT * FROM cypher('reduce', $$
+ RETURN {reduce: 1, any: 2, none: 3}
+$$) AS (result agtype);
+
+--
+-- Cleanup
+--
+SELECT * FROM drop_graph('reduce', true);
diff --git a/sql/age_aggregate.sql b/sql/age_aggregate.sql
index a8ea425c..fb258e5c 100644
--- a/sql/age_aggregate.sql
+++ b/sql/age_aggregate.sql
@@ -216,3 +216,25 @@ CREATE AGGREGATE ag_catalog.age_collect(variadic "any")
finalfunc = ag_catalog.age_collect_aggfinalfn,
parallel = safe
);
+
+--
+-- reduce(acc = init, var IN list | body) fold support
+--
+-- Transition function for the age_reduce aggregate. The fold body is compiled
+-- by transform_cypher_reduce() with the accumulator and element rewritten to
+-- PARAM_EXEC params 0 and 1 and serialized into the text argument; the
+-- transition evaluates it for each element in list order. It must be callable
+-- with a NULL transition state (no initcond), so it is intentionally not
STRICT.
+CREATE FUNCTION ag_catalog.age_reduce_transfn(agtype, agtype, text, agtype)
+ RETURNS agtype
+ LANGUAGE c
+PARALLEL UNSAFE
+AS 'MODULE_PATHNAME';
+
+-- aggregate definition for reduce(); direct arguments are
+-- (init, serialized-body, element), with the element fed ORDER BY ordinality.
+CREATE AGGREGATE ag_catalog.age_reduce(agtype, text, agtype)
+(
+ stype = agtype,
+ sfunc = ag_catalog.age_reduce_transfn
+);
diff --git a/src/backend/nodes/ag_nodes.c b/src/backend/nodes/ag_nodes.c
index bd78549c..ac659e1b 100644
--- a/src/backend/nodes/ag_nodes.c
+++ b/src/backend/nodes/ag_nodes.c
@@ -65,7 +65,8 @@ const char *node_names[] = {
"cypher_delete_information",
"cypher_delete_item",
"cypher_merge_information",
- "cypher_predicate_function"
+ "cypher_predicate_function",
+ "cypher_reduce"
};
/*
@@ -134,7 +135,8 @@ const ExtensibleNodeMethods node_methods[] = {
DEFINE_NODE_METHODS_EXTENDED(cypher_delete_information),
DEFINE_NODE_METHODS_EXTENDED(cypher_delete_item),
DEFINE_NODE_METHODS_EXTENDED(cypher_merge_information),
- DEFINE_NODE_METHODS_EXTENDED(cypher_predicate_function)
+ DEFINE_NODE_METHODS_EXTENDED(cypher_predicate_function),
+ DEFINE_NODE_METHODS_EXTENDED(cypher_reduce)
};
static bool equal_ag_node(const ExtensibleNode *a, const ExtensibleNode *b)
diff --git a/src/backend/nodes/cypher_copyfuncs.c
b/src/backend/nodes/cypher_copyfuncs.c
index 283096ca..54921875 100644
--- a/src/backend/nodes/cypher_copyfuncs.c
+++ b/src/backend/nodes/cypher_copyfuncs.c
@@ -185,3 +185,15 @@ void copy_cypher_predicate_function(ExtensibleNode
*newnode,
COPY_NODE_FIELD(expr);
COPY_NODE_FIELD(where);
}
+
+/* copy function for cypher_reduce */
+void copy_cypher_reduce(ExtensibleNode *newnode, const ExtensibleNode *from)
+{
+ COPY_LOCALS(cypher_reduce);
+
+ COPY_STRING_FIELD(acc_varname);
+ COPY_NODE_FIELD(init_expr);
+ COPY_STRING_FIELD(elem_varname);
+ COPY_NODE_FIELD(list_expr);
+ COPY_NODE_FIELD(body_expr);
+}
diff --git a/src/backend/nodes/cypher_outfuncs.c
b/src/backend/nodes/cypher_outfuncs.c
index 84d32a8f..4a35be02 100644
--- a/src/backend/nodes/cypher_outfuncs.c
+++ b/src/backend/nodes/cypher_outfuncs.c
@@ -200,6 +200,18 @@ void out_cypher_predicate_function(StringInfo str, const
ExtensibleNode *node)
WRITE_NODE_FIELD(where);
}
+/* serialization function for the cypher_reduce ExtensibleNode. */
+void out_cypher_reduce(StringInfo str, const ExtensibleNode *node)
+{
+ DEFINE_AG_NODE(cypher_reduce);
+
+ WRITE_STRING_FIELD(acc_varname);
+ WRITE_NODE_FIELD(init_expr);
+ WRITE_STRING_FIELD(elem_varname);
+ WRITE_NODE_FIELD(list_expr);
+ WRITE_NODE_FIELD(body_expr);
+}
+
/* serialization function for the cypher_merge ExtensibleNode. */
void out_cypher_merge(StringInfo str, const ExtensibleNode *node)
{
diff --git a/src/backend/nodes/cypher_readfuncs.c
b/src/backend/nodes/cypher_readfuncs.c
index 1e7e0ef8..a9a2ffab 100644
--- a/src/backend/nodes/cypher_readfuncs.c
+++ b/src/backend/nodes/cypher_readfuncs.c
@@ -329,3 +329,17 @@ void read_cypher_predicate_function(struct ExtensibleNode
*node)
READ_NODE_FIELD(expr);
READ_NODE_FIELD(where);
}
+
+/*
+ * Deserialize a string representing the cypher_reduce data structure.
+ */
+void read_cypher_reduce(struct ExtensibleNode *node)
+{
+ READ_LOCALS(cypher_reduce);
+
+ READ_STRING_FIELD(acc_varname);
+ READ_NODE_FIELD(init_expr);
+ READ_STRING_FIELD(elem_varname);
+ READ_NODE_FIELD(list_expr);
+ READ_NODE_FIELD(body_expr);
+}
diff --git a/src/backend/parser/cypher_analyze.c
b/src/backend/parser/cypher_analyze.c
index 5b72f332..5dd53dcd 100644
--- a/src/backend/parser/cypher_analyze.c
+++ b/src/backend/parser/cypher_analyze.c
@@ -865,6 +865,25 @@ bool cypher_raw_expr_tree_walker_impl(Node *node,
return true;
}
}
+ else if (is_ag_node(node, cypher_reduce))
+ {
+ cypher_reduce *rd = (cypher_reduce *)node;
+
+ if (WALK(rd->init_expr))
+ {
+ return true;
+ }
+
+ if (WALK(rd->list_expr))
+ {
+ return true;
+ }
+
+ if (WALK(rd->body_expr))
+ {
+ return true;
+ }
+ }
/* Add more node types here as needed */
else
{
diff --git a/src/backend/parser/cypher_clause.c
b/src/backend/parser/cypher_clause.c
index 5ac9dea6..a3a1a504 100644
--- a/src/backend/parser/cypher_clause.c
+++ b/src/backend/parser/cypher_clause.c
@@ -50,6 +50,7 @@
#include "parser/cypher_expr.h"
#include "parser/cypher_item.h"
#include "parser/cypher_parse_agg.h"
+#include "utils/agtype.h"
#include "parser/cypher_transform_entity.h"
#include "utils/ag_cache.h"
#include "utils/ag_func.h"
@@ -315,6 +316,10 @@ static Query
*transform_cypher_list_comprehension(cypher_parsestate *cpstate,
static Query *transform_cypher_predicate_function(cypher_parsestate *cpstate,
cypher_clause *clause);
+/* reduce */
+static Query *transform_cypher_reduce(cypher_parsestate *cpstate,
+ cypher_clause *clause);
+
/* merge */
static Query *transform_cypher_merge(cypher_parsestate *cpstate,
cypher_clause *clause);
@@ -579,6 +584,10 @@ Query *transform_cypher_clause(cypher_parsestate *cpstate,
{
result = transform_cypher_predicate_function(cpstate, clause);
}
+ else if (is_ag_node(self, cypher_reduce))
+ {
+ result = transform_cypher_reduce(cpstate, clause);
+ }
else
{
ereport(ERROR, (errmsg_internal("unexpected Node for cypher_clause")));
@@ -2016,6 +2025,418 @@ static Query
*transform_cypher_predicate_function(cypher_parsestate *cpstate,
}
}
+/*
+ * Mutator context for rewriting the fold body's accumulator/element Vars
+ * (columns 1 and 2 of the throwaway namespace RTE) into PARAM_EXEC params.
+ */
+typedef struct reduce_var_param_context
+{
+ int varno; /* rangetable index of the dummy (acc, elem) RTE */
+} reduce_var_param_context;
+
+/*
+ * Rewrite Var(varno, 1) -> Param(PARAM_EXEC, 0) [accumulator] and
+ * Var(varno, 2) -> Param(PARAM_EXEC, 1) [element] in the transformed fold
+ * body, so the body can be evaluated standalone inside age_reduce_transfn
+ * with the two params rebound for every element.
+ */
+static Node *reduce_var_to_param_mutator(Node *node, void *context)
+{
+ reduce_var_param_context *ctx = (reduce_var_param_context *) context;
+
+ if (node == NULL)
+ {
+ return NULL;
+ }
+
+ if (IsA(node, Var))
+ {
+ Var *var = (Var *) node;
+
+ /*
+ * Only the dummy (acc, elem) RTE at this level is rewritten. The
+ * varlevelsup == 0 check is essential: an outer-query RTE can share
+ * the same varno (each parse state's range table is numbered from 1),
+ * so without it a correlated outer reference at attno 1/2 would be
+ * silently rewritten into the accumulator/element param. Outer Vars
+ * are instead left in place and rejected by reduce_body_check_walker.
+ */
+ if (var->varno == ctx->varno && var->varlevelsup == 0 &&
+ (var->varattno == 1 || var->varattno == 2))
+ {
+ Param *param = makeNode(Param);
+
+ param->paramkind = PARAM_EXEC;
+ param->paramid = var->varattno - 1;
+ param->paramtype = AGTYPEOID;
+ param->paramtypmod = -1;
+ param->paramcollid = InvalidOid;
+ param->location = -1;
+
+ return (Node *) param;
+ }
+ }
+
+ return expression_tree_mutator(node, reduce_var_to_param_mutator, context);
+}
+
+/*
+ * Build a throwaway subquery "SELECT NULL::agtype AS <acc>, NULL::agtype AS
+ * <elem>" used only to give the fold body a namespace in which the accumulator
+ * and element variables resolve to agtype columns. Those references are later
+ * rewritten to PARAM_EXEC params and the subquery is discarded.
+ */
+static Query *make_reduce_var_subquery(char *acc_name, char *elem_name)
+{
+ Query *subquery = makeNode(Query);
+ Const *acc_const;
+ Const *elem_const;
+ TargetEntry *acc_te;
+ TargetEntry *elem_te;
+
+ acc_const = makeConst(AGTYPEOID, -1, InvalidOid, -1, (Datum) 0, true,
false);
+ elem_const = makeConst(AGTYPEOID, -1, InvalidOid, -1, (Datum) 0, true,
false);
+
+ acc_te = makeTargetEntry((Expr *) acc_const, 1, acc_name, false);
+ elem_te = makeTargetEntry((Expr *) elem_const, 2, elem_name, false);
+
+ subquery->commandType = CMD_SELECT;
+ subquery->targetList = list_make2(acc_te, elem_te);
+ subquery->jointree = makeFromExpr(NIL, NULL);
+ subquery->rtable = NIL;
+ subquery->rteperminfos = NIL;
+
+ return subquery;
+}
+
+/*
+ * Validate a transformed-and-mutated reduce() fold body. After
+ * reduce_var_to_param_mutator() has replaced the accumulator and element with
+ * PARAM_EXEC params 0 and 1, a valid body is a pure expression over those two
+ * params: it must contain no other Vars (outer-query references), no other
+ * params, and no aggregates or subqueries, because the body is evaluated
+ * standalone (ExecEvalExpr) inside age_reduce_transfn with only those two
+ * param slots bound.
+ */
+static bool reduce_body_check_walker(Node *node, void *context)
+{
+ if (node == NULL)
+ {
+ return false;
+ }
+
+ if (IsA(node, Var))
+ {
+ ereport(ERROR,
+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("a reduce() expression may only reference its
accumulator and element variables")));
+ }
+
+ if (IsA(node, Param))
+ {
+ Param *param = (Param *) node;
+
+ if (param->paramkind != PARAM_EXEC ||
+ (param->paramid != 0 && param->paramid != 1))
+ {
+ ereport(ERROR,
+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("a reduce() expression may not reference query
parameters")));
+ }
+ }
+
+ if (IsA(node, Aggref) || IsA(node, GroupingFunc) || IsA(node, WindowFunc))
+ {
+ ereport(ERROR,
+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("aggregate functions are not supported in a reduce()
expression")));
+ }
+
+ if (IsA(node, SubLink))
+ {
+ ereport(ERROR,
+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("subqueries (including a nested reduce()) are not
supported in a reduce() expression")));
+ }
+
+ return expression_tree_walker(node, reduce_body_check_walker, context);
+}
+
+/*
+ * Transform a cypher_reduce node into a query tree.
+ *
+ * reduce(acc = init, var IN list | body) is rewritten into a scalar subquery
+ * over the age_reduce aggregate, with the list unnested WITH ORDINALITY and
the
+ * aggregate ordered by that ordinality so the fold runs in list order:
+ *
+ * SELECT ag_catalog.age_reduce(<init>, '<serialized-body>'::text,
+ * r.elem ORDER BY r.ord)
+ * FROM unnest(<list>) WITH ORDINALITY AS r(elem, ord)
+ *
+ * The fold body is transformed separately with the accumulator and element
+ * rewritten to PARAM_EXEC params 0 and 1, serialized into the text argument,
+ * and evaluated per element inside age_reduce_transfn.
+ *
+ * The null/empty-list guard
+ * (CASE WHEN list IS NULL THEN NULL ELSE COALESCE(<agg>, init) END) is built
+ * at the grammar level in build_reduce_node().
+ */
+static Query *transform_cypher_reduce(cypher_parsestate *cpstate,
+ cypher_clause *clause)
+{
+ cypher_reduce *reduce = (cypher_reduce *) clause->self;
+ Query *query;
+ Query *var_subquery;
+ cypher_parsestate *body_cpstate;
+ ParseState *body_pstate;
+ ParseNamespaceItem *body_pnsi;
+ Node *body_node;
+ char *body_serialized;
+ reduce_var_param_context mutator_ctx;
+ cypher_parsestate *child_cpstate;
+ ParseState *child_pstate;
+ FuncCall *unnest_fc;
+ RangeFunction *rf;
+ RangeTblEntry *rte = NULL;
+ int rtindex = 0;
+ List *namespace = NULL;
+ Node *from_item;
+ Node *init_node;
+ Node *elem_var;
+ Var *ord_var;
+ TargetEntry *ord_te;
+ SortGroupClause *sortcl;
+ Oid sort_ltop;
+ Oid sort_eqop;
+ bool sort_hashable;
+ Const *body_const;
+ Aggref *agg;
+ Oid agg_oid;
+ Oid agg_argtypes[3];
+ TargetEntry *result_te;
+
+ /*
+ * 1. Resolve the fold body's accumulator and element variables against a
+ * throwaway 2-column agtype subquery, rewrite those Vars to PARAM_EXEC
+ * params, validate it is a pure expression over those params, and
+ * serialize the body for age_reduce_transfn.
+ */
+ body_cpstate = make_cypher_parsestate(cpstate);
+ body_pstate = (ParseState *) body_cpstate;
+
+ var_subquery = make_reduce_var_subquery(reduce->acc_varname,
+ reduce->elem_varname);
+ body_pnsi = addRangeTableEntryForSubquery(body_pstate, var_subquery,
+ makeAlias("reduce_vars", NIL),
+ false, true);
+ addNSItemToQuery(body_pstate, body_pnsi, false, true, true);
+
+ body_node = transform_cypher_expr(body_cpstate, reduce->body_expr,
+ EXPR_KIND_SELECT_TARGET);
+
+ /*
+ * The accumulator is always an agtype value (the aggregate's stype is
+ * agtype). A fold body can legitimately produce a non-agtype scalar -- for
+ * example "s AND x" or "x = 2" yield a boolean -- so normalize the body to
+ * agtype here. Without this the transition function would treat a by-value
+ * Datum (e.g. bool) as a by-reference varlena and crash. A boolean is
+ * wrapped in ag_catalog.bool_to_agtype() (AGE registers no implicit
+ * boolean-to-agtype cast); any other non-agtype type is coerced through
+ * the normal cast machinery, which raises a clean error if impossible.
+ */
+ if (exprType(body_node) != AGTYPEOID)
+ {
+ if (exprType(body_node) == BOOLOID)
+ {
+ Oid bool_to_agtype_oid = get_ag_func_oid("bool_to_agtype", 1,
+ BOOLOID);
+
+ body_node = (Node *) makeFuncExpr(bool_to_agtype_oid, AGTYPEOID,
+ list_make1(body_node),
+ InvalidOid, InvalidOid,
+ COERCE_EXPLICIT_CALL);
+ }
+ else
+ {
+ body_node = coerce_to_common_type(body_pstate, body_node,
+ AGTYPEOID, "reduce");
+ }
+ }
+
+ mutator_ctx.varno = body_pnsi->p_rtindex;
+ body_node = reduce_var_to_param_mutator(body_node, &mutator_ctx);
+
+ reduce_body_check_walker(body_node, NULL);
+
+ body_serialized = nodeToString(body_node);
+
+ free_cypher_parsestate(body_cpstate);
+
+ /*
+ * 2. Build the outer aggregate query:
+ * SELECT age_reduce(<init>, '<body>'::text, r.elem ORDER BY r.ord)
+ * FROM unnest(<list>) WITH ORDINALITY AS r(elem, ord)
+ */
+ query = makeNode(Query);
+ query->commandType = CMD_SELECT;
+
+ child_cpstate = make_cypher_parsestate(cpstate);
+ child_pstate = (ParseState *) child_cpstate;
+
+ unnest_fc = makeFuncCall(list_make1(makeString("unnest")),
+ list_make1(reduce->list_expr),
+ COERCE_SQL_SYNTAX, -1);
+ rf = makeNode(RangeFunction);
+ rf->lateral = false;
+ rf->ordinality = true;
+ rf->is_rowsfrom = false;
+ rf->functions = list_make1(list_make2((Node *) unnest_fc, NIL));
+ rf->alias = makeAlias("reduce_src",
+ list_make2(makeString(reduce->elem_varname),
+ makeString("reduce_ordinality")));
+ rf->coldeflist = NIL;
+
+ from_item = transform_from_clause_item(child_cpstate, (Node *) rf,
+ &rte, &rtindex, &namespace);
+ checkNameSpaceConflicts(child_pstate, child_pstate->p_namespace,
namespace);
+ child_pstate->p_joinlist = lappend(child_pstate->p_joinlist, from_item);
+ child_pstate->p_namespace = list_concat(child_pstate->p_namespace,
+ namespace);
+ setNamespaceLateralState(child_pstate->p_namespace, false, true);
+
+ /* arguments to age_reduce: init, serialized body text, element column */
+ init_node = transform_cypher_expr(child_cpstate, reduce->init_expr,
+ EXPR_KIND_SELECT_TARGET);
+ elem_var = colNameToVar(child_pstate, reduce->elem_varname, false, -1);
+ body_const = makeConst(TEXTOID, -1, InvalidOid, -1,
+ CStringGetTextDatum(body_serialized), false, false);
+
+ /* the WITH ORDINALITY column (bigint), used only to order the fold */
+ ord_var = makeVar(rtindex, 2, INT8OID, -1, InvalidOid, 0);
+ get_sort_group_operators(INT8OID, true, true, false,
+ &sort_ltop, &sort_eqop, NULL, &sort_hashable);
+
+ ord_te = makeTargetEntry((Expr *) ord_var, 4, NULL, true);
+ ord_te->ressortgroupref = 1;
+
+ sortcl = makeNode(SortGroupClause);
+ sortcl->tleSortGroupRef = 1;
+ sortcl->eqop = sort_eqop;
+ sortcl->sortop = sort_ltop;
+ sortcl->reverse_sort = false;
+ sortcl->nulls_first = false;
+ sortcl->hashable = sort_hashable;
+
+ /*
+ * Evaluate <init> exactly once per reduce() instead of once per element.
+ * A regular aggregate argument is evaluated by the executor for every
+ * input row, but age_reduce_transfn only reads the init argument on the
+ * first transition (when the running state is still NULL). Re-evaluating
+ * an expensive init wastes work, and a volatile init would fire its side
+ * effects once per element.
+ *
+ * Rows are fed to the aggregate in ascending ordinality order, so the
+ * first transition is always the row with ordinality 1. Wrapping init in
+ * CASE WHEN reduce_ordinality = 1 THEN <init> ELSE NULL::agtype END
+ * computes <init> on exactly that row (CASE only evaluates the matching
+ * branch's result) and passes a NULL init -- which the transition
+ * function ignores -- on every other row. The empty-list case is handled
+ * separately by the COALESCE(..., init) guard in build_reduce_node().
+ */
+ {
+ OpExpr *ord_is_first;
+ Const *one_const;
+ CaseWhen *init_when;
+ CaseExpr *init_case;
+ Const *null_init;
+
+ one_const = makeConst(INT8OID, -1, InvalidOid, sizeof(int64),
+ Int64GetDatum(1), false, true);
+
+ ord_is_first = makeNode(OpExpr);
+ ord_is_first->opno = sort_eqop; /* int8 equality */
+ ord_is_first->opfuncid = get_opcode(sort_eqop);
+ ord_is_first->opresulttype = BOOLOID;
+ ord_is_first->opretset = false;
+ ord_is_first->opcollid = InvalidOid;
+ ord_is_first->inputcollid = InvalidOid;
+ ord_is_first->args = list_make2(copyObject(ord_var), one_const);
+ ord_is_first->location = -1;
+
+ null_init = makeConst(AGTYPEOID, -1, InvalidOid, -1, (Datum) 0,
+ true, false);
+
+ init_when = makeNode(CaseWhen);
+ init_when->expr = (Expr *) ord_is_first;
+ init_when->result = (Expr *) init_node;
+ init_when->location = -1;
+
+ init_case = makeNode(CaseExpr);
+ init_case->casetype = AGTYPEOID;
+ init_case->casecollid = InvalidOid;
+ init_case->arg = NULL;
+ init_case->args = list_make1(init_when);
+ init_case->defresult = (Expr *) null_init;
+ init_case->location = -1;
+
+ init_node = (Node *) init_case;
+ }
+
+ /* look up the age_reduce(agtype, text, agtype) aggregate */
+ agg_argtypes[0] = AGTYPEOID;
+ agg_argtypes[1] = TEXTOID;
+ agg_argtypes[2] = AGTYPEOID;
+ agg_oid = LookupFuncName(list_make2(makeString("ag_catalog"),
+ makeString("age_reduce")),
+ 3, agg_argtypes, false);
+
+ agg = makeNode(Aggref);
+ agg->aggfnoid = agg_oid;
+ agg->aggtype = AGTYPEOID;
+ agg->aggcollid = InvalidOid;
+ agg->inputcollid = InvalidOid;
+ agg->aggtranstype = InvalidOid; /* filled by the planner */
+ agg->aggargtypes = list_make3_oid(AGTYPEOID, TEXTOID, AGTYPEOID);
+ agg->aggdirectargs = NIL;
+ agg->args = list_make4(makeTargetEntry((Expr *) init_node, 1, NULL, false),
+ makeTargetEntry((Expr *) body_const, 2, NULL,
false),
+ makeTargetEntry((Expr *) elem_var, 3, NULL, false),
+ ord_te);
+ agg->aggorder = list_make1(sortcl);
+ agg->aggdistinct = NIL;
+ agg->aggfilter = NULL;
+ agg->aggstar = false;
+ agg->aggvariadic = false;
+ agg->aggkind = AGGKIND_NORMAL;
+ agg->aggpresorted = false;
+ agg->agglevelsup = 0;
+ agg->aggsplit = AGGSPLIT_SIMPLE;
+ agg->aggno = -1;
+ agg->aggtransno = -1;
+ agg->location = -1;
+
+ child_pstate->p_hasAggs = true;
+
+ result_te = makeTargetEntry((Expr *) agg,
+ (AttrNumber) child_pstate->p_next_resno++,
+ "reduce", false);
+
+ query->targetList = list_make1(result_te);
+ query->jointree = makeFromExpr(child_pstate->p_joinlist, NULL);
+ query->rtable = child_pstate->p_rtable;
+ query->rteperminfos = child_pstate->p_rteperminfos;
+ query->hasAggs = true;
+ query->hasSubLinks = child_pstate->p_hasSubLinks;
+ query->hasTargetSRFs = child_pstate->p_hasTargetSRFs;
+
+ assign_query_collations(child_pstate, query);
+ parse_check_aggregates(child_pstate, query);
+
+ free_cypher_parsestate(child_cpstate);
+
+ return query;
+}
+
/*
* Iterate through the list of items to delete and extract the variable name.
* Then find the resno that the variable name belongs to.
diff --git a/src/backend/parser/cypher_gram.y b/src/backend/parser/cypher_gram.y
index b23fa705..83d69c83 100644
--- a/src/backend/parser/cypher_gram.y
+++ b/src/backend/parser/cypher_gram.y
@@ -131,7 +131,7 @@
MATCH MERGE
NONE NOT NULL_P
ON OPERATOR OPTIONAL OR ORDER
- REMOVE RETURN
+ REDUCE REMOVE RETURN
SET SINGLE SKIP STARTS
THEN TRUE_P
UNION UNWIND
@@ -332,6 +332,11 @@ static Node
*build_predicate_function_node(cypher_predicate_function_kind kind,
/* pattern expression helper */
static Node *make_exists_pattern_sublink(Node *pattern, int location);
+/* reduce(acc = init, var IN list | body) */
+static Node *build_reduce_node(char *acc_varname, Node *init_expr,
+ char *elem_varname, Node *list_expr,
+ Node *body_expr, int location);
+
/* helper functions */
static ExplainStmt *make_explain_stmt(List *options);
static void validate_return_item_aliases(List *items, ag_scanner_t scanner);
@@ -1963,6 +1968,10 @@ expr_func_subexpr:
{
$$ = build_predicate_function_node(CPFK_SINGLE, $3, $5, $7, @1);
}
+ | REDUCE '(' var_name '=' expr ',' var_name IN expr '|' expr ')'
+ {
+ $$ = build_reduce_node($3, $5, $7, $9, $11, @1);
+ }
;
expr_subquery:
@@ -2561,6 +2570,7 @@ safe_keywords:
| OPTIONAL { $$ = KEYWORD_STRDUP($1); }
| OR { $$ = KEYWORD_STRDUP($1); }
| ORDER { $$ = KEYWORD_STRDUP($1); }
+ | REDUCE { $$ = KEYWORD_STRDUP($1); }
| REMOVE { $$ = KEYWORD_STRDUP($1); }
| RETURN { $$ = KEYWORD_STRDUP($1); }
| SET { $$ = KEYWORD_STRDUP($1); }
@@ -3646,6 +3656,86 @@ static Node *make_exists_pattern_sublink(Node *pattern,
int location)
return (Node *)node_to_agtype((Node *)n, "boolean", location);
}
+/*
+ * Helper function to build a reduce() grammar node.
+ *
+ * Follows the openCypher syntax:
+ * reduce(acc = init, var IN list | body)
+ *
+ * The accumulator `acc` is seeded with `init` and threaded across the
+ * elements of `list` (bound to `var`) in list order, with `body` producing
+ * the next accumulator value at each step. The result is the final
+ * accumulator value, or `init` when the list is empty.
+ *
+ * The reduce node is wrapped in an EXPR_SUBLINK (scalar subquery) whose
+ * subselect is the cypher_reduce node; transform_cypher_reduce() in
+ * cypher_clause.c rewrites it into a correlated scalar subquery over an
+ * ordered aggregate.
+ *
+ * The whole thing is then wrapped so the openCypher null/empty-list semantics
+ * hold without the transform layer having to special-case them:
+ *
+ * CASE WHEN list IS NULL THEN NULL
+ * ELSE COALESCE((reduce subquery), init) END
+ *
+ * A NULL list yields NULL; an empty list yields `init` (the aggregate runs
+ * over zero rows and returns SQL NULL, which COALESCE replaces with init);
+ * a non-empty list yields the fold result. The list and init grammar nodes
+ * are shared between the reduce node and this guard, which is safe because
+ * AGE's expression transformer builds new nodes rather than mutating in place.
+ */
+static Node *build_reduce_node(char *acc_varname, Node *init_expr,
+ char *elem_varname, Node *list_expr,
+ Node *body_expr, int location)
+{
+ SubLink *sub;
+ cypher_reduce *reduce_node = NULL;
+ CoalesceExpr *coalesce;
+ NullTest *null_test;
+ CaseWhen *case_when;
+ CaseExpr *guard;
+
+ reduce_node = make_ag_node(cypher_reduce);
+ reduce_node->acc_varname = acc_varname;
+ reduce_node->init_expr = init_expr;
+ reduce_node->elem_varname = elem_varname;
+ reduce_node->list_expr = list_expr;
+ reduce_node->body_expr = body_expr;
+
+ sub = makeNode(SubLink);
+ sub->subLinkId = 0;
+ sub->testexpr = NULL;
+ sub->operName = NIL;
+ sub->subselect = (Node *) reduce_node;
+ sub->location = location;
+ sub->subLinkType = EXPR_SUBLINK;
+
+ /* COALESCE((reduce subquery), init) -- empty list falls back to init */
+ coalesce = makeNode(CoalesceExpr);
+ coalesce->args = list_make2((Node *) sub, init_expr);
+ coalesce->location = location;
+
+ /* CASE WHEN list IS NULL THEN NULL ELSE <coalesce> END */
+ null_test = makeNode(NullTest);
+ null_test->arg = (Expr *) list_expr;
+ null_test->nulltesttype = IS_NULL;
+ null_test->argisrow = false;
+ null_test->location = location;
+
+ case_when = makeNode(CaseWhen);
+ case_when->expr = (Expr *) null_test;
+ case_when->result = (Expr *) make_null_const(location);
+ case_when->location = location;
+
+ guard = makeNode(CaseExpr);
+ guard->arg = NULL;
+ guard->args = list_make1(case_when);
+ guard->defresult = (Expr *) coalesce;
+ guard->location = location;
+
+ return (Node *) guard;
+}
+
/* Helper function to create an ExplainStmt node */
static ExplainStmt *make_explain_stmt(List *options)
{
diff --git a/src/backend/utils/adt/agtype.c b/src/backend/utils/adt/agtype.c
index cfff8138..bf69bf1f 100644
--- a/src/backend/utils/adt/agtype.c
+++ b/src/backend/utils/adt/agtype.c
@@ -47,8 +47,11 @@
#include "miscadmin.h"
#include "parser/parse_coerce.h"
#include "nodes/nodes.h"
+#include "nodes/nodeFuncs.h"
+#include "executor/executor.h"
#include "utils/acl.h"
#include "utils/builtins.h"
+#include "utils/datum.h"
#include "executor/cypher_utils.h"
#include "utils/float.h"
#include "utils/lsyscache.h"
@@ -11575,6 +11578,168 @@ Datum
age_float8_stddev_pop_aggfinalfn(PG_FUNCTION_ARGS)
PG_RETURN_POINTER(agtype_value_to_agtype(&agtv_float));
}
+/*
+ * Per-aggregate-group evaluation state for reduce(). Caches the compiled
+ * fold-body expression and a standalone ExprContext whose PARAM_EXEC slots
+ * (0 = accumulator, 1 = current element) are rebound on every element.
+ */
+typedef struct reduce_eval_ctx
+{
+ ExprState *body_state; /* compiled fold-body expression */
+ ExprContext *econtext; /* eval context carrying the param slots */
+ ParamExecData *params; /* [0] = accumulator, [1] = current element */
+} reduce_eval_ctx;
+
+/* Build an agtype 'null' Datum (a real agtype value, not a SQL NULL). */
+static Datum reduce_agtype_null(void)
+{
+ agtype_value agtv;
+
+ agtv.type = AGTV_NULL;
+ return AGTYPE_P_GET_DATUM(agtype_value_to_agtype(&agtv));
+}
+
+/*
+ * age_reduce_transfn(state agtype, init agtype, body text, element agtype)
+ *
+ * Transition function for the age_reduce aggregate that implements the Cypher
+ * reduce(acc = init, var IN list | body) fold. The fold body is compiled by
+ * transform_cypher_reduce() with the accumulator and element rewritten to
+ * PARAM_EXEC params 0 and 1, then serialized into the `body` text argument.
+ *
+ * On the first element of a group the accumulator is seeded from `init`
+ * (the running state is NULL because the aggregate uses no initcond); on
+ * every element the body is evaluated with the params rebound, and the result
+ * becomes the next accumulator state.
+ *
+ * The accumulator and element are normalized to a non-NULL agtype 'null'
+ * before evaluation so that (a) the fold body sees agtype values and Cypher
+ * null semantics apply, and (b) the running state is never a SQL NULL, which
+ * keeps PG_ARGISNULL(0) a reliable "first element of the group" signal even
+ * when the fold legitimately produces null.
+ */
+PG_FUNCTION_INFO_V1(age_reduce_transfn);
+
+Datum age_reduce_transfn(PG_FUNCTION_ARGS)
+{
+ MemoryContext aggcontext;
+ MemoryContext oldctx;
+ reduce_eval_ctx *rc;
+ Datum acc;
+ Datum element;
+ Datum result;
+ bool result_isnull;
+
+ if (!AggCheckCallContext(fcinfo, &aggcontext))
+ {
+ ereport(ERROR,
+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("age_reduce_transfn called in a non-aggregate
context")));
+ }
+
+ /* the fold can run over a large list; stay responsive to cancellation */
+ CHECK_FOR_INTERRUPTS();
+
+ /*
+ * One-time per-FmgrInfo setup: deserialize and compile the fold body, and
+ * build the standalone ExprContext plus its two PARAM_EXEC slots. The body
+ * text is a query constant, so caching the compiled state across groups is
+ * correct.
+ */
+ rc = (reduce_eval_ctx *) fcinfo->flinfo->fn_extra;
+ if (rc == NULL)
+ {
+ text *body_txt;
+ char *body_str;
+ Node *body_node;
+
+ if (PG_ARGISNULL(2))
+ {
+ ereport(ERROR,
+ (errcode(ERRCODE_INTERNAL_ERROR),
+ errmsg("age_reduce: missing fold expression")));
+ }
+
+ oldctx = MemoryContextSwitchTo(fcinfo->flinfo->fn_mcxt);
+ rc = (reduce_eval_ctx *) palloc0(sizeof(reduce_eval_ctx));
+ body_txt = PG_GETARG_TEXT_PP(2);
+ body_str = text_to_cstring(body_txt);
+ body_node = (Node *) stringToNode(body_str);
+
+ /*
+ * age_reduce() is SQL-callable, so the serialized body argument is
+ * not guaranteed to have come from transform_cypher_reduce(). The
+ * running state is stored as an agtype varlena (the datumCopy() below
+ * uses typbyval=false, typlen=-1), so a body that evaluates to a
+ * by-value type (e.g. a bare boolean or integer) would have its Datum
+ * misread as a pointer and could crash the backend. Reject any body
+ * whose result type is not agtype. transform_cypher_reduce() always
+ * normalizes the fold body to agtype, so a planner-generated reduce()
+ * is never rejected here.
+ */
+ if (exprType(body_node) != AGTYPEOID)
+ {
+ ereport(ERROR,
+ (errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
+ errmsg("age_reduce: fold expression must return
agtype")));
+ }
+
+ rc->body_state = ExecInitExpr((Expr *) body_node, NULL);
+ rc->econtext = CreateStandaloneExprContext();
+ rc->params = (ParamExecData *) palloc0(sizeof(ParamExecData) * 2);
+ rc->econtext->ecxt_param_exec_vals = rc->params;
+ fcinfo->flinfo->fn_extra = rc;
+ MemoryContextSwitchTo(oldctx);
+ }
+
+ /*
+ * Seed the accumulator. The aggregate declares no initcond, so on the
+ * first element the running state (arg 0) is NULL and we use `init`
+ * (arg 1); thereafter the accumulator is the prior state. A NULL init is
+ * normalized to agtype 'null'.
+ */
+ if (PG_ARGISNULL(0))
+ {
+ acc = PG_ARGISNULL(1) ? reduce_agtype_null() : PG_GETARG_DATUM(1);
+ }
+ else
+ {
+ acc = PG_GETARG_DATUM(0);
+ }
+
+ /* a NULL element is likewise normalized to agtype 'null' */
+ element = PG_ARGISNULL(3) ? reduce_agtype_null() : PG_GETARG_DATUM(3);
+
+ /* bind PARAM_EXEC 0 = accumulator, 1 = current element */
+ rc->params[0].value = acc;
+ rc->params[0].isnull = false;
+ rc->params[0].execPlan = NULL;
+ rc->params[1].value = element;
+ rc->params[1].isnull = false;
+ rc->params[1].execPlan = NULL;
+
+ /* evaluate the fold body for this element */
+ ResetExprContext(rc->econtext);
+ result = ExecEvalExpr(rc->body_state, rc->econtext, &result_isnull);
+
+ /*
+ * Never let the running state become a SQL NULL: a null fold result is
+ * stored as agtype 'null' so the next element is not mistaken for the
+ * first one (see PG_ARGISNULL(0) above).
+ */
+ if (result_isnull)
+ {
+ result = reduce_agtype_null();
+ }
+
+ /* the new state must survive in the aggregate context across elements */
+ oldctx = MemoryContextSwitchTo(aggcontext);
+ result = datumCopy(result, false, -1);
+ MemoryContextSwitchTo(oldctx);
+
+ PG_RETURN_DATUM(result);
+}
+
PG_FUNCTION_INFO_V1(age_agtype_larger_aggtransfn);
Datum age_agtype_larger_aggtransfn(PG_FUNCTION_ARGS)
diff --git a/src/include/nodes/ag_nodes.h b/src/include/nodes/ag_nodes.h
index 47c55041..57b2deab 100644
--- a/src/include/nodes/ag_nodes.h
+++ b/src/include/nodes/ag_nodes.h
@@ -77,7 +77,9 @@ typedef enum ag_node_tag
cypher_delete_item_t,
cypher_merge_information_t,
/* predicate functions */
- cypher_predicate_function_t
+ cypher_predicate_function_t,
+ /* reduce */
+ cypher_reduce_t
} ag_node_tag;
extern const char *node_names[];
diff --git a/src/include/nodes/cypher_copyfuncs.h
b/src/include/nodes/cypher_copyfuncs.h
index e770cebe..fcad598b 100644
--- a/src/include/nodes/cypher_copyfuncs.h
+++ b/src/include/nodes/cypher_copyfuncs.h
@@ -56,4 +56,8 @@ void copy_cypher_merge_information(ExtensibleNode *newnode,
/* predicate function data structure */
void copy_cypher_predicate_function(ExtensibleNode *newnode,
const ExtensibleNode *from);
+
+/* reduce data structure */
+void copy_cypher_reduce(ExtensibleNode *newnode,
+ const ExtensibleNode *from);
#endif
diff --git a/src/include/nodes/cypher_nodes.h b/src/include/nodes/cypher_nodes.h
index 3433bebb..5efbe95f 100644
--- a/src/include/nodes/cypher_nodes.h
+++ b/src/include/nodes/cypher_nodes.h
@@ -247,6 +247,24 @@ typedef struct cypher_predicate_function
Node *where; /* the predicate to test */
} cypher_predicate_function;
+/*
+ * reduce(acc = init, var IN list | body)
+ *
+ * Folds `body` over `list`, threading an accumulator `acc` (seeded with
+ * `init`) across the elements in list order, binding each element to `var`.
+ * Transformed into a correlated scalar subquery over an ordered aggregate
+ * by transform_cypher_reduce() in cypher_clause.c.
+ */
+typedef struct cypher_reduce
+{
+ ExtensibleNode extensible;
+ char *acc_varname; /* accumulator variable name */
+ Node *init_expr; /* initial accumulator value */
+ char *elem_varname; /* per-element variable name */
+ Node *list_expr; /* the list to fold over */
+ Node *body_expr; /* the fold expression evaluated per element */
+} cypher_reduce;
+
typedef enum cypher_map_projection_element_type
{
PROPERTY_SELECTOR = 0, /* map_var { .key } */
diff --git a/src/include/nodes/cypher_outfuncs.h
b/src/include/nodes/cypher_outfuncs.h
index 55285bdb..fc2a830b 100644
--- a/src/include/nodes/cypher_outfuncs.h
+++ b/src/include/nodes/cypher_outfuncs.h
@@ -50,6 +50,7 @@ void out_cypher_map_projection(StringInfo str, const
ExtensibleNode *node);
void out_cypher_list(StringInfo str, const ExtensibleNode *node);
void out_cypher_list_comprehension(StringInfo str, const ExtensibleNode *node);
void out_cypher_predicate_function(StringInfo str, const ExtensibleNode *node);
+void out_cypher_reduce(StringInfo str, const ExtensibleNode *node);
/* comparison expression */
void out_cypher_comparison_aexpr(StringInfo str, const ExtensibleNode *node);
diff --git a/src/include/nodes/cypher_readfuncs.h
b/src/include/nodes/cypher_readfuncs.h
index 9202ba51..2f27f91c 100644
--- a/src/include/nodes/cypher_readfuncs.h
+++ b/src/include/nodes/cypher_readfuncs.h
@@ -54,4 +54,7 @@ void read_cypher_merge_information(struct ExtensibleNode
*node);
/* predicate function data structure */
void read_cypher_predicate_function(struct ExtensibleNode *node);
+/* reduce data structure */
+void read_cypher_reduce(struct ExtensibleNode *node);
+
#endif
diff --git a/src/include/parser/cypher_kwlist.h
b/src/include/parser/cypher_kwlist.h
index 44ac0945..909b5d27 100644
--- a/src/include/parser/cypher_kwlist.h
+++ b/src/include/parser/cypher_kwlist.h
@@ -36,6 +36,7 @@ PG_KEYWORD("operator", OPERATOR, RESERVED_KEYWORD)
PG_KEYWORD("optional", OPTIONAL, RESERVED_KEYWORD)
PG_KEYWORD("or", OR, RESERVED_KEYWORD)
PG_KEYWORD("order", ORDER, RESERVED_KEYWORD)
+PG_KEYWORD("reduce", REDUCE, RESERVED_KEYWORD)
PG_KEYWORD("remove", REMOVE, RESERVED_KEYWORD)
PG_KEYWORD("return", RETURN, RESERVED_KEYWORD)
PG_KEYWORD("set", SET, RESERVED_KEYWORD)
