State every SQL aggregate with the MEOS function behind each of its roles

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<T>` — 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.

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)

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

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()