jrgemignani commented on code in PR #2435:
URL: https://github.com/apache/age/pull/2435#discussion_r3392394252
##########
src/backend/parser/cypher_clause.c:
##########
@@ -1962,6 +1971,363 @@ 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;
+
+ /* 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);
Review Comment:
Added
##########
src/backend/utils/adt/agtype.c:
##########
@@ -11575,6 +11577,149 @@ 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);
+ rc->body_state = ExecInitExpr((Expr *) body_node, NULL);
+ rc->econtext = CreateStandaloneExprContext();
Review Comment:
added
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
