From 10e10025f7060af60d00fa4e41a597eaaad10780 Mon Sep 17 00:00:00 2001 From: Esteban Zimanyi Date: Sat, 3 Oct 2026 14:34:05 +0200 Subject: [PATCH] State every SQL aggregate with the MEOS function behind each of its roles The catalog gains the top-level list aggregates, read by parser/aggregates.py from the deployed CREATE AGGREGATE statements: per aggregate its arguments, its result type and, for each role it defines (transition, combine, final, serialize, deserialize, the roles chapter 18 of the MobilityDB manual names), the SQL function PostgreSQL calls and the public MEOS function carrying that function's SQL signature, None for a function of PostgreSQL itself (array_agg_transfn) or one no public MEOS function carries. Each role resolves as PostgreSQL resolves it, by name over the argument types it is called with: the transition function over the state and the arguments, the combine function over two states, the final function over the state, with FINALFUNC_EXTRA's arguments where stated. The result type is what the final function returns, else the state type. GENERATION.md lists the section. Witness: the catalog carries the aggregates' machinery functions alone, each tagged with the aggregate names it serves (sqlAgg), and no aggregate's own signature: nothing states that tCount(tgeompoint) answers a tint, that temporal_tagg_finalfn finishes tCount, tSum, tMin and merge alike, or which transition function an aggregate over a given type calls. Why: a binding generates an aggregate the way it generates a function, from the catalog; reading CREATE AGGREGATE or naming a result type itself is the hand-made code the bindings are rid of. JMEOS builds the aggregates of Spark and Flink from this list. Measured: over MobilityDB 985fdb26b7 the catalog states 349 aggregates, the count of the portable naming plan, 97 of them with a public MEOS function for every role they define; every other section of meos-idl.json is unchanged. pytest tests/ answers 468 passed and 11 skipped with MDB_SRC_ROOT set, test_aggregates.py's 9 tests among the passed. --- GENERATION.md | 5 +- parser/aggregates.py | 128 +++++++++++++++++++++++++++ run.py | 9 ++ tests/test_aggregates.py | 183 +++++++++++++++++++++++++++++++++++++++ 4 files changed, 324 insertions(+), 1 deletion(-) create mode 100644 parser/aggregates.py create mode 100644 tests/test_aggregates.py diff --git a/GENERATION.md b/GENERATION.md index 44c7374..041b941 100644 --- a/GENERATION.md +++ b/GENERATION.md @@ -19,7 +19,10 @@ that the preprocessor flattens to `int`), `@ingroup` groups, the `@sqlfn` SQL-na portable bare-name aliases, and the type registries: `typeRelations` for each base type's set, span, span set and temporal types, and `temporalTypes` for what MEOS states about each `Temporal` — its base, its bounding box, the MF-JSON type token `asMFJSON` writes for it, -and its number, spatial and linear classes. The `generator/` modules project the catalog onto the +and its number, spatial and linear classes. `aggregates` states every SQL aggregate: its +arguments, its result type and, for each of its roles (transition, combine, final, +serialize, deserialize), the SQL function PostgreSQL calls and the public MEOS function +carrying it. The `generator/` modules project the catalog onto the language-**agnostic** service contracts (OpenAPI, MCP, the runtime server, the OGC Moving Features projection) — the surfaces that need no foreign toolchain. Language bindings live in their own repos and generate from this catalog. diff --git a/parser/aggregates.py b/parser/aggregates.py new file mode 100644 index 0000000..46df043 --- /dev/null +++ b/parser/aggregates.py @@ -0,0 +1,128 @@ +"""State the SQL aggregates and the MEOS function behind each of their roles. + +A SQL aggregate is assembled from up to three functions, each a role chapter 18 of the +MobilityDB manual (doc/portable_sql.xml, "Aggregations") names: a transition function, +a combine function for parallel aggregation and a final function. PostgreSQL adds the +functions writing and reading a partial state, which travels between its workers. The +catalog states every `CREATE AGGREGATE` in the top-level list `aggregates`: + + {"sqlName": "tCount", "args": ["tgeompoint"], "ret": "tint", "stype": "internal", + "transition": {"sqlName": "tCountTransition", "meos": "temporal_tcount_transfn"}, + "combine": {"sqlName": "tcount_combinefn", "meos": "temporal_tcount_combinefn"}, + "final": {"sqlName": "tint_tagg_finalfn", "meos": "temporal_tagg_finalfn"}, + "serialize": {"sqlName": "taggstate_serialize", "meos": null}, + "deserialize": {"sqlName": "taggstate_deserialize", "meos": null}} + +`args` are the types of the arguments a caller passes, `ret` the type of the answer: +what the final function returns, or the state type for an aggregate without one +(`extent`). A role the aggregate does not define is absent. A role names the SQL +function PostgreSQL calls, `sqlName`, and the public MEOS function carrying that +function's SQL signature, `meos`, None for one no public MEOS function carries: a +function of PostgreSQL itself (`array_agg_transfn`, `array_agg_combine`), or one of +the extension over PostgreSQL's own memory (`taggstate_serialize`). PostgreSQL refuses +an aggregate naming a function it does not have, so every role names one. A binding +builds an aggregate from the MEOS functions of its roles and never reads SQL. + +Each role resolves as PostgreSQL resolves it, by name and argument types: the +transition function over the state type and the aggregate's arguments, the combine +function over two states, the final function over the state (and the aggregate's +arguments under `FINALFUNC_EXTRA`), the serialize function over the state and the +deserialize function over `bytea` and the state.""" +import re + +from parser.compositions import _type +from parser.sqlfn import _split_top_commas, _strip_sql_comments, sql_signature, sql_statements + +_CREATE_AGG = re.compile(r"CREATE\s+(?:OR\s+REPLACE\s+)?AGGREGATE\s+(\w+)\s*\(", re.I) +_OPTION = re.compile(r"^\s*(\w+)\s*(?:=\s*(.+?))?\s*$", re.S) + +# The options of a CREATE AGGREGATE naming a function, under the role each one plays. +_ROLES = {"sfunc": "transition", "combinefunc": "combine", "finalfunc": "final", + "serialfunc": "serialize", "deserialfunc": "deserialize"} + + +def _create_agg_stmts(text): + """Yield (sqlName, [raw arg decls], {option: value}) for every CREATE AGGREGATE in + `text`, its option names lower-cased; the argument list is read as #_create_fn_stmts + of parser/sqlfn.py reads a CREATE FUNCTION's.""" + for m in _CREATE_AGG.finditer(text): + i, depth, start = m.end(), 1, m.end() + while i < len(text) and depth: + depth += (text[i] == "(") - (text[i] == ")") + i += 1 + argdecls = [a for a in _split_top_commas(text[start:i - 1]) if a.strip()] + open_ = text.find("(", i) + j, depth = open_ + 1, 1 + while j < len(text) and depth: + depth += (text[j] == "(") - (text[j] == ")") + j += 1 + options = {} + for opt in _split_top_commas(text[open_ + 1:j - 1]): + om = _OPTION.match(opt) + if om: + options[om.group(1).lower()] = (om.group(2) or "").strip() + yield m.group(1), argdecls, options + + +def _agg_statements(sql_src): + """Every CREATE AGGREGATE under `sql_src`, as #_create_agg_stmts yields it.""" + out = [] + for sf in sorted(sql_src.rglob("*.sql")): + out.extend(_create_agg_stmts(_strip_sql_comments(sf.read_text(errors="ignore")))) + return out + + +def attach_aggregates(idl, sql_src): + """(idl with its top-level `aggregates`, count). Raises ValueError naming every + aggregate with no state type, or with a role two public MEOS functions carry.""" + stmts, vocab, composites = sql_statements(sql_src) + declared = {} + for sqlname, argdecls, ret, _, retset, _ in stmts: + sig = sql_signature(sqlname, argdecls, ret, retset, vocab, composites) + declared[(sqlname.lower(), tuple(_type(a) for a in sig["args"]))] = sig + carried = {} + for f in idl.get("functions", []): + if f.get("api") != "public": + continue + for s in f.get("sqlSignatures") or (): + name = (s.get("sqlName") or f.get("sqlfn") or "").lower() + carried.setdefault((name, tuple(_type(a) for a in s["args"])), []).append(f["name"]) + + out, errors = [], [] + for sqlname, argdecls, options in _agg_statements(sql_src): + args = sql_signature(sqlname, argdecls, None, False, vocab, composites)["args"] + stype = options.get("stype") + if not stype: + errors.append(f"{sqlname}({', '.join(args)}): no STYPE") + continue + extra = options.get("finalfunc_extra") is not None + called = {"transition": [stype] + args, "combine": [stype, stype], + "final": [stype] + (args if extra else []), + "serialize": [stype], "deserialize": ["bytea", stype]} + entry = {"sqlName": sqlname, "args": args, "ret": None, "stype": stype} + for option, role in _ROLES.items(): + fn = options.get(option) + if not fn: + continue + # A function may be written with its argument types, + # `SFUNC = appendInstantTransition(th3index, th3index)`; its name is the role's. + fn = re.sub(r"\s*\(.*\)\s*$", "", fn, flags=re.S).split(".")[-1] + key = (fn.lower(), tuple(_type(a) for a in called[role])) + meos = carried.get(key) or [] + if len(meos) > 1: + errors.append(f"{sqlname}({', '.join(args)}): {option} " + f"{fn}({', '.join(called[role])}) is carried by " + f"{', '.join(sorted(meos))}") + break + entry[role] = {"sqlName": fn, "meos": meos[0] if meos else None} + if role == "final": + entry["ret"] = declared[key]["ret"] if key in declared else None + else: + if "final" not in entry: + entry["ret"] = stype + out.append(entry) + if errors: + raise ValueError("SQL aggregates the catalog cannot state:\n " + + "\n ".join(errors)) + idl["aggregates"] = out + return idl, len(out) diff --git a/run.py b/run.py index 11610db..05558f5 100644 --- a/run.py +++ b/run.py @@ -17,6 +17,7 @@ from parser.outparam import extract_param_names, merge_outparams from parser.boundargs import (attach_call_literals, merge_boundargs, resolve_bound_names, strip_call_literals) +from parser.aggregates import attach_aggregates from parser.compositions import attach_compositions from parser.indexsearch import attach_index_search from parser.enrich import enrich_idl, restate_wire @@ -382,6 +383,14 @@ def main(): print(f" SQL compositions over another type's functions: {ncomp}", file=sys.stderr) + # State each SQL aggregate: its arguments, its result type and the SQL and MEOS + # function of each of its roles, read from the deployed CREATE AGGREGATE statements + # and resolved through the signatures above, so a binding builds an aggregate from + # MEOS functions as it builds a composition. + if SQL_SRC.exists(): + idl, nagg_sql = attach_aggregates(idl, SQL_SRC) + print(f" SQL aggregates: {nagg_sql}", file=sys.stderr) + # Stamp the MobilityDB source commit so the catalog is SELF-DESCRIBING about its freshness: # a consumer proves it is current by comparing sourceCommit to live upstream master, never by # inspecting whatever directory a vendored copy sits in. None when the source is not a git diff --git a/tests/test_aggregates.py b/tests/test_aggregates.py new file mode 100644 index 0000000..5cc55d0 --- /dev/null +++ b/tests/test_aggregates.py @@ -0,0 +1,183 @@ +"""The SQL aggregates, stated with the MEOS function behind each of their roles. + +Every `CREATE AGGREGATE` is stated in the catalog's top-level `aggregates` with its +arguments, its result type and, for each role it defines (transition, combine, final, +serialize, deserialize), the SQL function PostgreSQL calls and the public MEOS function +carrying that function's SQL signature. Each role resolves as PostgreSQL resolves it, by +name and the argument types it is called with. + +The catalog is synthetic and the SQL sits in a temp dir, as #CompositionTests of +tests/test_compositions.py arranges them. Plain unittest, no pytest dependency. +""" +import tempfile +import unittest +from pathlib import Path + +from parser.aggregates import attach_aggregates + + +def _fn(name, sqlfn, sigs, api="public"): + return {"name": name, "sqlfn": sqlfn, "api": api, + "sqlSignatures": [{"sqlName": sqlfn, **s} for s in sigs]} + + +FUNCTIONS = [ + _fn("temporal_tcount_transfn", "tCountTransition", + [{"args": ["internal", "tgeompoint"], "ret": "internal"}]), + _fn("temporal_tcount_combinefn", "tcount_combinefn", + [{"args": ["internal", "internal"], "ret": "internal"}]), + _fn("temporal_tagg_finalfn", "tCount", + [{"args": ["internal"], "ret": "tint", "sqlName": "tint_tagg_finalfn"}, + {"args": ["internal"], "ret": "tfloat", "sqlName": "tfloat_tagg_finalfn"}]), + _fn("tnumber_wavg_transfn", "wavg_transfn", + [{"args": ["internal", "tfloat", "interval"], "ret": "internal"}]), + _fn("tspatial_extent_transfn", "stbox_extent_transfn", + [{"args": ["stbox", "tgeompoint"], "ret": "stbox"}]), + _fn("set_union_finalfn", "floatset_union_finalfn", + [{"args": ["internal"], "ret": "floatset"}]), + # the transition of an aggregate whose MEOS function is internal + _fn("temporal_app_tinst_transfn", "appendInstantTransition", + [{"args": ["tfloat", "tfloat"], "ret": "tfloat"}], api="internal"), +] + +DECLARED = """ +CREATE FUNCTION tCountTransition(internal, tgeompoint) + RETURNS internal AS 'MODULE_PATHNAME', 'Temporal_tcount_transfn' LANGUAGE C; +CREATE FUNCTION tcount_combinefn(internal, internal) + RETURNS internal AS 'MODULE_PATHNAME', 'Temporal_tcount_combinefn' LANGUAGE C; +CREATE FUNCTION tint_tagg_finalfn(internal) + RETURNS tint AS 'MODULE_PATHNAME', 'Temporal_tagg_finalfn' LANGUAGE C; +CREATE FUNCTION tfloat_tagg_finalfn(internal) + RETURNS tfloat AS 'MODULE_PATHNAME', 'Temporal_tagg_finalfn' LANGUAGE C; +CREATE FUNCTION taggstate_serialize(internal) + RETURNS bytea AS 'MODULE_PATHNAME', 'Taggstate_serialize' LANGUAGE C; +CREATE FUNCTION taggstate_deserialize(bytea, internal) + RETURNS internal AS 'MODULE_PATHNAME', 'Taggstate_deserialize' LANGUAGE C; +CREATE FUNCTION wavg_transfn(internal, tfloat, interval) + RETURNS internal AS 'MODULE_PATHNAME', 'Tnumber_wavg_transfn' LANGUAGE C; +CREATE FUNCTION stbox_extent_transfn(stbox, tgeompoint) + RETURNS stbox AS 'MODULE_PATHNAME', 'Tspatial_extent_transfn' LANGUAGE C; +CREATE FUNCTION floatset_union_finalfn(internal) + RETURNS floatset AS 'MODULE_PATHNAME', 'Set_union_finalfn' LANGUAGE C; +CREATE FUNCTION appendInstantTransition(tfloat, tfloat) + RETURNS tfloat AS 'MODULE_PATHNAME', 'Temporal_app_tinst_transfn' LANGUAGE C; +CREATE FUNCTION temporal_append_finalfn(tfloat) + RETURNS tfloat AS 'MODULE_PATHNAME', 'Temporal_append_finalfn' LANGUAGE C; +""" + +AGGREGATES = DECLARED + """ +CREATE AGGREGATE tCount(tgeompoint) ( + SFUNC = tCountTransition, + STYPE = internal, + COMBINEFUNC = tcount_combinefn, + FINALFUNC = tint_tagg_finalfn, + SERIALFUNC = taggstate_serialize, + DESERIALFUNC = taggstate_deserialize, + PARALLEL = SAFE +); +-- a comment between statements +CREATE AGGREGATE wAvg(tfloat, interval) ( + SFUNC = wavg_transfn, + STYPE = internal, + FINALFUNC = tfloat_tagg_finalfn +); +CREATE AGGREGATE extent(tgeompoint) ( + SFUNC = stbox_extent_transfn, + STYPE = stbox, + PARALLEL = safe +); +CREATE AGGREGATE setUnion(float8) ( + SFUNC = array_agg_transfn, + STYPE = internal, + COMBINEFUNC = array_agg_combine, + FINALFUNC = floatset_union_finalfn +); +CREATE AGGREGATE appendInstantAgg(tfloat) ( + SFUNC = appendInstantTransition(tfloat, tfloat), + STYPE = tfloat, + FINALFUNC = temporal_append_finalfn, + PARALLEL = safe +); +""" + + +class AggregateTests(unittest.TestCase): + """Each aggregate the SQL states, with its roles and result type.""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.sql = Path(self.tmp.name) + + def tearDown(self): + self.tmp.cleanup() + + def _attach(self, text, functions=FUNCTIONS): + (self.sql / "aggregates.in.sql").write_text(text) + return attach_aggregates({"functions": [dict(f) for f in functions]}, self.sql) + + def _entries(self): + idl, n = self._attach(AGGREGATES) + self.assertEqual(n, len(idl["aggregates"])) + return {(e["sqlName"], tuple(e["args"])): e for e in idl["aggregates"]} + + def test_every_role_names_its_sql_and_meos_function(self): + e = self._entries()[("tCount", ("tgeompoint",))] + self.assertEqual(e["ret"], "tint") + self.assertEqual(e["stype"], "internal") + self.assertEqual(e["transition"], + {"sqlName": "tCountTransition", "meos": "temporal_tcount_transfn"}) + self.assertEqual(e["combine"], + {"sqlName": "tcount_combinefn", "meos": "temporal_tcount_combinefn"}) + self.assertEqual(e["final"], + {"sqlName": "tint_tagg_finalfn", "meos": "temporal_tagg_finalfn"}) + + def test_a_role_over_postgresql_memory_names_no_meos_function(self): + e = self._entries()[("tCount", ("tgeompoint",))] + self.assertEqual(e["serialize"], {"sqlName": "taggstate_serialize", "meos": None}) + self.assertEqual(e["deserialize"], {"sqlName": "taggstate_deserialize", "meos": None}) + + def test_the_transition_takes_the_state_and_every_argument(self): + e = self._entries()[("wAvg", ("tfloat", "interval"))] + self.assertEqual(e["transition"]["meos"], "tnumber_wavg_transfn") + self.assertEqual(e["ret"], "tfloat") + self.assertNotIn("combine", e) + + def test_without_a_final_function_the_answer_is_the_state(self): + e = self._entries()[("extent", ("tgeompoint",))] + self.assertEqual(e["ret"], "stbox") + self.assertEqual(e["transition"]["meos"], "tspatial_extent_transfn") + self.assertNotIn("final", e) + + def test_a_postgresql_function_names_no_meos_function(self): + # The argument stays as the SQL spells it, as #sql_signature keeps it in every + # sqlSignatures entry; the role resolves through #_type, as #_Resolver does. + e = self._entries()[("setUnion", ("float8",))] + self.assertEqual(e["transition"], {"sqlName": "array_agg_transfn", "meos": None}) + self.assertEqual(e["combine"], {"sqlName": "array_agg_combine", "meos": None}) + self.assertEqual(e["final"]["meos"], "set_union_finalfn") + self.assertEqual(e["ret"], "floatset") + + def test_a_function_written_with_its_argument_types(self): + e = self._entries()[("appendInstantAgg", ("tfloat",))] + self.assertEqual(e["transition"]["sqlName"], "appendInstantTransition") + self.assertEqual(e["ret"], "tfloat") + + def test_an_internal_meos_function_is_not_named(self): + e = self._entries()[("appendInstantAgg", ("tfloat",))] + self.assertIsNone(e["transition"]["meos"]) + + def test_a_role_two_public_functions_carry_stops_the_catalog(self): + twin = _fn("tgeompoint_tcount_transfn", "tCountTransition", + [{"args": ["internal", "tgeompoint"], "ret": "internal"}]) + with self.assertRaises(ValueError) as cm: + self._attach(AGGREGATES, FUNCTIONS + [twin]) + self.assertIn("tCount(tgeompoint)", str(cm.exception)) + + def test_an_aggregate_without_a_state_type_stops_the_catalog(self): + with self.assertRaises(ValueError) as cm: + self._attach("CREATE AGGREGATE bad(tint) (SFUNC = f);") + self.assertIn("bad(tint): no STYPE", str(cm.exception)) + + +if __name__ == "__main__": + unittest.main()