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