Add the temporal-covering projection generator

generate_covering.py

@@ -0,0 +1,39 @@
+"""Generate the temporal-covering projection from the enriched MEOS catalog.
+
+Usage:
+    python run.py <path-to-MobilityDB-meos-include>   # produce output/meos-idl.json
+    python generate_covering.py                       # read output/meos-idl.json
+    python generate_covering.py path.json out.json
+"""
+
+import json
+import sys
+from pathlib import Path
+
+from generator.covering import build_covering_projection
+
+IN_PATH = Path(sys.argv[1]) if len(sys.argv) > 1 else Path("output/meos-idl.json")
+OUT_PATH = (Path(sys.argv[2]) if len(sys.argv) > 2
+            else Path("output/meos-covering-projection.json"))
+
+
+def main() -> None:
+    if not IN_PATH.exists():
+        sys.exit(f"Catalog not found: {IN_PATH} — run `python run.py` first.")
+
+    catalog = json.loads(IN_PATH.read_text())
+    if "temporalCovering" not in catalog:
+        sys.exit(f"{IN_PATH} has no `temporalCovering` — it is attached by "
+                 "run.py; regenerate the catalog.")
+
+    proj = build_covering_projection(catalog)
+
+    OUT_PATH.parent.mkdir(parents=True, exist_ok=True)
+    OUT_PATH.write_text(json.dumps(proj, indent=2) + "\n")
+    print(f"      → {OUT_PATH} written", file=sys.stderr)
+    print(f"[covering-projection] {proj['count']} temporal types projected "
+          f"to covering columns", file=sys.stderr)
+
+
+if __name__ == "__main__":
+    main()

generator/covering.py

@@ -0,0 +1,63 @@
+"""Temporal-covering projection generator.
+
+Projects the ``temporalCovering`` block of the MEOS catalog
+(``meos-idl.json``, produced by ``parser/covering.py``) onto the canonical,
+language-agnostic covering-column contract: per temporal type, the ordered
+covering columns with the fully-composed MEOS expression that derives each
+from the value.
+
+Every binding generator (PyMEOS, JMEOS, MobilityDuck, MobilitySpark, …)
+renders this same contract in its own idiom — a DuckDB ``GENERATED`` column,
+a Spark UDF projection, a PyMEOS writer — so a temporal table prunes the
+same way on every platform (Iceberg manifest + Parquet row-group min/max).
+The ``VALUE`` placeholder is the temporal column reference the binding
+substitutes.
+
+Pure ``dict`` → ``dict``; no libclang and no MEOS runtime.
+"""
+
+from __future__ import annotations
+
+
+def _column_expr(column: dict, box_from: str) -> str:
+    """Compose the MEOS expression that derives one covering column from the
+    temporal value (``VALUE``). A ``box`` column is read off the value's box;
+    a ``value`` column is read off the value directly."""
+    if column["source"] == "value":
+        return f"{column['accessor']}(VALUE)"
+    return f"{column['accessor']}({box_from}(VALUE))"
+
+
+def build_covering_projection(catalog: dict) -> dict:
+    """Project ``temporalCovering`` onto the canonical covering-column contract."""
+    cov = catalog.get("temporalCovering")
+    if not cov:
+        raise ValueError("catalog has no `temporalCovering` — run run.py")
+
+    types = {}
+    for tname, spec in cov["byType"].items():
+        box = spec.get("box")
+        box_from = box["from"] if box else None
+        columns = []
+        for col in spec["columns"]:
+            entry = {
+                "name": col["name"],
+                "sqlType": col["sqlType"],
+                "expr": _column_expr(col, box_from),
+            }
+            if col.get("when"):
+                entry["when"] = col["when"]
+            columns.append(entry)
+        types[tname] = {
+            "class": spec["class"],
+            "boxType": box["type"] if box else None,
+            "columns": columns,
+        }
+
+    return {
+        "version": cov["version"],
+        "valueCodec": cov["valueCodec"],
+        "metadataKeys": cov["metadataKeys"],
+        "types": types,
+        "count": len(types),
+    }

tests/test_covering_projection.py

@@ -0,0 +1,77 @@
+"""Unit tests for generator/covering.py.
+python3 tests/test_covering_projection.py
+
+Also the CI gate: when the enriched catalog with `temporalCovering` is
+present, every covered type projects to a well-formed covering expression
+composed against the value.
+"""
+
+import json
+import sys
+import unittest
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+from parser.covering import attach_temporal_covering
+from generator.covering import build_covering_projection
+
+MAP = ROOT / "meta" / "temporal-covering.json"
+_CATALOG = ROOT / "output" / "meos-idl.json"
+
+
+def _projected():
+    return build_covering_projection(attach_temporal_covering({}, MAP))
+
+
+class ProjectionTests(unittest.TestCase):
+    def test_spatial_box_composition(self):
+        cols = {c["name"]: c
+                for c in _projected()["types"]["tgeompoint"]["columns"]}
+        # box columns compose accessor(box_from(VALUE))
+        self.assertEqual(cols["xmin"]["expr"],
+                         "stbox_xmin(tspatial_to_stbox(VALUE))")
+        self.assertEqual(cols["xmin"]["sqlType"], "double")
+        # srid is read off the value, not the box
+        self.assertEqual(cols["srid"]["expr"], "tspatial_srid(VALUE)")
+        # zmin is conditional on 3D
+        self.assertEqual(cols["zmin"]["when"], "hasZ")
+
+    def test_number_box_composition(self):
+        t = _projected()["types"]["tfloat"]
+        cols = {c["name"]: c for c in t["columns"]}
+        self.assertEqual(t["boxType"], "TBOX")
+        self.assertEqual(cols["vmin"]["expr"], "tbox_xmin(tnumber_to_tbox(VALUE))")
+        self.assertEqual(cols["tmax"]["expr"], "tbox_tmax(tnumber_to_tbox(VALUE))")
+
+    def test_count_and_codec(self):
+        p = _projected()
+        self.assertEqual(p["count"], len(p["types"]))
+        self.assertEqual(p["valueCodec"]["asHexWkb"], "temporal_as_hexwkb")
+
+    def test_requires_temporal_covering(self):
+        with self.assertRaises(ValueError):
+            build_covering_projection({"functions": []})
+
+
+@unittest.skipUnless(_CATALOG.exists(), "run `python run.py` first")
+class LiveProjectionGate(unittest.TestCase):
+    def test_every_type_projects_wellformed(self):
+        cat = attach_temporal_covering(json.loads(_CATALOG.read_text()), MAP)
+        p = build_covering_projection(cat)
+        self.assertEqual(p["count"], 13)
+        # time-only types (tbool/ttext) project to tmin/tmax via the value, no box
+        self.assertEqual(p["types"]["tbool"]["boxType"], None)
+        self.assertEqual(
+            {c["name"] for c in p["types"]["tbool"]["columns"]}, {"tmin", "tmax"})
+        for spec in p["types"].values():
+            self.assertTrue(spec["columns"])
+            for c in spec["columns"]:
+                # composed against the value, balanced parentheses
+                self.assertIn("(VALUE)", c["expr"])
+                self.assertEqual(c["expr"].count("("), c["expr"].count(")"))
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)