diff --git a/Cargo.lock b/Cargo.lock index e22801b7ea..b66c5c84ad 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -148,6 +148,18 @@ dependencies = [ "pin-project-lite", ] +[[package]] +name = "async-compression" +version = "0.4.52" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6ad47a740fb906fffdd891e8b6d5fde2511600f7bfaebd68939fa08a14999459" +dependencies = [ + "compression-codecs", + "compression-core", + "pin-project-lite", + "tokio", +] + [[package]] name = "async-io" version = "2.6.0" @@ -693,6 +705,23 @@ dependencies = [ "static_assertions", ] +[[package]] +name = "compression-codecs" +version = "0.4.47" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "84f840cff84d10085260e460023934d0af699e71212b09b2a41452e8bd9cda43" +dependencies = [ + "compression-core", + "flate2", + "memchr", +] + +[[package]] +name = "compression-core" +version = "0.4.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2ebdda89edc534bf8d5dd2857562633b6495557e5846552ade994b99ed13c36a" + [[package]] name = "concurrent-queue" version = "2.5.0" @@ -4313,6 +4342,7 @@ dependencies = [ "thiserror 2.0.21", "tokio", "toml", + "tower-http", "utoipa", "uuid", ] @@ -5133,12 +5163,16 @@ version = "0.6.11" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4cfcf7e2740e6fc6d4d688b4ef00650406bb94adf4731e43c096c3a19fe40840" dependencies = [ + "async-compression", "bitflags 2.13.1", "bytes", + "futures-core", "futures-util", "http", "http-body", "pin-project-lite", + "tokio", + "tokio-util", "tower", "tower-layer", "tower-service", diff --git a/Cargo.toml b/Cargo.toml index 5e951f06d6..af397e0d79 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -72,6 +72,8 @@ rust-version = "1.88" argon2 = "0.5.3" axum = { version = "0.8", default-features = false, features = ["http1", "json", "query", "tokio"] } axum-server = { version = "0.8", default-features = false, features = ["tls-rustls-no-provider"] } +# Compresses the server's larger JSON reads and UI assets for browsers. +tower-http = { version = "0.6", default-features = false, features = ["compression-gzip"] } fensu-policy = "=0.22.0" futures-util = { version = "0.3", default-features = false, features = ["std"] } # `sqb` names compute log captures with random ids, as `uuid.uuid4().hex`. diff --git a/benchmarks/duckshop/README.md b/benchmarks/duckshop/README.md index f4386bd6cc..ac5fd5faa5 100644 --- a/benchmarks/duckshop/README.md +++ b/benchmarks/duckshop/README.md @@ -155,6 +155,12 @@ Benchmarks and core measurements leave the area out: with `--exclude edge_cases`; - the construct census counts the core project unless given `--area edge` or `--area all`. +`uv run --script benchmarks/duckshop/scripts/duckdb_project.py ` writes a DuckDB copy +of the core project for demonstrations: it compiles the core copy, translates each compiled model +with Polyglot and the rewrites in `duckdb_rewrites.py`, and leaves out tests, scenarios, custom +Rules and Snowflake-only hooks. `sqb build` in the copy loads the synthetic data and builds every +model on DuckDB, which an E2E test checks. It is not a benchmark workload. + Changes to the core project's compiled SQL, after which benchmark results are not comparable with earlier ones: diff --git a/benchmarks/duckshop/edge_cases/orchestration/upstream_source_orders.py b/benchmarks/duckshop/edge_cases/orchestration/upstream_source_orders.py new file mode 100644 index 0000000000..7fc6d7150f --- /dev/null +++ b/benchmarks/duckshop/edge_cases/orchestration/upstream_source_orders.py @@ -0,0 +1,10 @@ +"""Schedule a job whose selection reaches an upstream source that a build never executes.""" + +from sqlbuild.orchestration import job, schedule + +annual_orders = job( + "annual_orders", + description="Build orders together with the source it reads.", + select="+orders", + schedule=schedule(cron="0 0 * * *", timezone="UTC", starts_at="2000-01-01T00:00:00Z"), +) diff --git a/benchmarks/duckshop/scripts/duckdb_project.py b/benchmarks/duckshop/scripts/duckdb_project.py new file mode 100644 index 0000000000..0b9e17d92f --- /dev/null +++ b/benchmarks/duckshop/scripts/duckdb_project.py @@ -0,0 +1,267 @@ +# /// script +# requires-python = ">=3.12" +# dependencies = ["polyglot-sql-chio==0.13.10", "pyyaml"] +# /// +"""Write a DuckDB copy of core DuckShop, translated from its compiled Snowflake models. + + uv run --script benchmarks/duckshop/scripts/duckdb_project.py + +Needs `sqb` on PATH. The script copies the core project (`core_project.py`), compiles it, and +writes a DuckDB project to DESTINATION: + +- Each model keeps its header, with Snowflake types mapped to DuckDB's, and gets its compiled SQL + translated by Polyglot and finished by `duckdb_rewrites.py`. Macros, constants and private + declarations are already inlined there, so the copy drops them; it keeps the enums a column is + typed with. +- Snowflake functions DuckDB lacks become project SQL macros, written where their callers are. +- Sources, seeds, schemas, audits and the synthetic data loaders are adapted for DuckDB. +- Tests, scenarios, custom Rules and Snowflake-only hooks are left out. + +The copy is for demonstrations: build it with `sqb load` and `sqb build` on DuckDB. It is not a +benchmark workload, and the Snowflake project stays the source of truth. +""" + +from __future__ import annotations + +import argparse +import json +import pathlib +import re +import shutil +import subprocess +import sys +import tempfile +from typing import Any + +import polyglot_sql +import yaml +from core_project import copy_core_project +from duckdb_rewrites import balanced, convert, shim_calls, write_shims + +TYPE_MAP = [ + (r"\bNUMBER\(38, ?0\)|\b(INT|INTEGER|SMALLINT|TINYINT)\b", "BIGINT"), + (r"\b(VARIANT|OBJECT|ARRAY)\b", "JSON"), + (r"\bTIMESTAMP_NTZ\b", "TIMESTAMP"), + (r"\bTIMESTAMP_(TZ|LTZ)\b", "TIMESTAMPTZ"), + (r"\bNUMBER\b", "DECIMAL"), +] +CONNECTION = """[connections.warehouse] +database = "duckshop.duckdb" + +[targets.dev] +connection = "warehouse" +schema = "dev" + +[targets.prod] +connection = "warehouse" +schema = "analytics" +""" +WRITER_QUOTE = ( + ' return "\'" + text.replace("\\\\", "\\\\\\\\").replace("\'", "\'\'") + "\'"', + ' return "\'" + text.replace("\'", "\'\'") + "\'"', +) +WRITER_INSERT = '''def insert_sql( + destination: str, columns: tuple[tuple[str, str], ...], rows: list[dict[str, object]] +) -> str: + """One INSERT for a batch: typed casts happen in the select list.""" + + names = ", ".join(name for name, _ in columns) + aliases = ", ".join(f"column{index}" for index in range(1, len(columns) + 1)) + select = ", ".join( + f"CAST(column{index} AS {sql_type})" for index, (_, sql_type) in enumerate(columns, 1) + ) + values = ",\\n".join( + "(" + ", ".join(literal(row[name]) for name, _ in columns) + ")" for row in rows + ) + rows_sql = f"FROM (VALUES\\n{values}) AS v({aliases})" + return f"INSERT INTO {destination} ({names})\\nSELECT {select}\\n{rows_sql}" + + +def statements''' + + +def types(text: str) -> str: + """Snowflake types as DuckDB's. Snowflake's integers are all NUMBER(38, 0): BIGINT here.""" + + for pattern, new in TYPE_MAP: + text = re.sub(pattern, new, text) + return text + + +def translate(sql: str, model: str | None = None) -> str: + # GET with a computed key and REGEXP_SUBSTR, which returns NULL rather than '' when nothing + # matches, go to their shims untranslated: Polyglot's DuckDB forms behave differently. + sql = re.sub(r"(? str: + m = re.match(rf"\s*{keyword} \(.*?^\);\n", text, re.S | re.M) + if m is None: + raise ValueError(f"no {keyword} header") + return m.group(0) + + +def audit_sql(text: str) -> str: + """Audit SQL in headers and audit files, with Snowflake's IFF as DuckDB's IF.""" + + return re.sub(r"(? str: + """A model header without private constants and enums, which its compiled SQL inlines, and + without Snowflake-only hooks.""" + + head = header(text, "MODEL") + head = re.sub(r'(?m)^ +inline_sql\("[^"]*CHANGE_TRACKING[^"]*"\),\n', "", head) + head = re.sub(r"(?m)^ post_hooks \[\n \],\n", "", head) + for block in ("constants", "enums"): + m = re.search(rf"^ {block} \(", head, re.M) + if m: + end = balanced(head, m.end()) + head = head[: m.start()] + head[end:].lstrip(",").lstrip("\n") + return audit_sql(types(head)) + + +def compile_snowflake(project: pathlib.Path) -> dict[str, Any]: + result = subprocess.run( + ["sqb", "compile", "--json"], cwd=project, capture_output=True, text=True, check=False + ) + if result.returncode != 0: + sys.stderr.write(result.stderr) + raise SystemExit("compiling the Snowflake project failed") + return json.loads(result.stdout) + + +def write_models(snow: pathlib.Path, out: pathlib.Path, report: dict[str, Any]) -> None: + for model in report["resources"]["models"]: + path = out / model["relative_path"] + body = translate(model["query_sql"], model["name"]) + path.write_text(model_header(path.read_text()) + "\n" + body + "\n") + for path in (out / "functions" / "sql").glob("*.sql"): + text = path.read_text() + head = header(text, "FUNCTION") + compiled = snow / "target" / "compiled" / "functions" / path.name + body = compiled.read_text() if compiled.exists() else text[len(head) :] + path.write_text(types(head) + "\n" + translate(body) + "\n") + + +def write_sources(out: pathlib.Path) -> None: + for path in [*(out / "sources").glob("*.yml"), out / "seeds" / "seeds.yml"]: + doc = yaml.safe_load(path.read_text()) + for entry in doc.get("sources", doc.get("seeds", [])): + for column in entry.get("columns", []): + column["type"] = types(column["type"]) + settings = entry.get("csv_settings") or {} + if isinstance(settings.get("na_values"), dict): + # DuckDB seeds take one null list for every column. + values = {str(v) for vs in settings["na_values"].values() for v in vs} + settings["na_values"] = sorted(values) + if entry.get("expression"): + entry["expression"] = translate(entry["expression"]) + "\n" + if (entry.get("freshness") or {}).get("query"): + # The query reads the Snowflake raw table, which the copy does not have. + del entry["freshness"] + path.write_text(yaml.safe_dump(doc, sort_keys=False, width=100)) + for path in out.rglob("*.py"): + if "functions" not in path.parts: + path.write_text(types(path.read_text())) + writer = out / "python" / "loaders" / "_writer.py" + text = writer.read_text() + if WRITER_QUOTE[0] not in text: + raise SystemExit("python/loaders/_writer.py changed: update WRITER_QUOTE") + text = text.replace(*WRITER_QUOTE) + text, count = re.subn( + r"def insert_sql\(.*?\n\n\ndef statements", lambda _: WRITER_INSERT, text, flags=re.S + ) + if count != 1: + raise SystemExit("python/loaders/_writer.py changed: update WRITER_INSERT") + writer.write_text(text) + + +def write_declarations(out: pathlib.Path) -> None: + """Keep enums that type a column, owned by their columns' folder; drop other declarations, + which compiled model SQL inlines. Integer enums become BIGINT columns: DuckDB types their + members INTEGER.""" + + for path in [p for p in out.rglob("*.sql") if "audits" in str(p.relative_to(out))]: + path.write_text(audit_sql(path.read_text())) + for path in out.glob("models/**/_sqlbuild/_schemas/*.sql"): + path.write_text(types(path.read_text())) + models = [p for p in out.glob("models/**/*.sql") if "_sqlbuild" not in p.parts] + enums = {path.stem: path for path in out.glob("**/enums/*.sql")} + for name in [n for n, p in enums.items() if re.search(r"members \(\w+ \d", p.read_text())]: + del enums[name] + for path in models: + path.write_text(re.sub(rf"\(type {name}\b", "(type BIGINT", path.read_text())) + owners: dict[str, set[pathlib.Path]] = {} + for path in models: + for name in re.findall(r"\(type (\w+)\b", header(path.read_text(), "MODEL")): + if name in enums: + owners.setdefault(name, set()).add(path.parent) + keep: dict[pathlib.Path, str] = {} + for name, folders in owners.items(): + folder = folders.pop() / "_sqlbuild" / "_enums" if len(folders) == 1 else out / "enums" + keep[folder / f"{name}.sql"] = types(enums[name].read_text()) + for gone in ("tests", "macros", "rules", "constants", "enums"): + shutil.rmtree(out / gone, ignore_errors=True) + for folder in ("_macros", "macros", "constants", "_constants", "enums", "_enums"): + for path in list(out.glob(f"models/**/_sqlbuild/{folder}")): + shutil.rmtree(path) + for path, text in keep.items(): + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text) + write_shims(out) + + +def write_project_file(out: pathlib.Path) -> None: + path = out / "sqlbuild_project.toml" + text = path.read_text().replace('adapter = "snowflake"', 'adapter = "duckdb"') + start = text.index("[connections.warehouse]") + end = text.index("\n\n", text.index("[targets.prod]")) + text = text[:start] + CONNECTION + text[end:] + text = re.sub(r"select = \[[^\]]*\]", 'select = ["SQBRGRAPH", "SQBRCONTRACT"]', text, count=1) + text = re.sub(r"(?ms)^\[scenario\.local_type_overrides\.snowflake\]\n.*?(?=^\[)", "", text) + # Rule exceptions for files the copy leaves out, such as macros, would fail compilation. + head, *blocks = re.split(r"(?m)^(?=\[\[rules\.rule_exceptions\]\])", text) + kept = [] + for block in blocks: + m = re.search(r'^path = "([^"]+)"', block, re.M) + if not m or (out / m.group(1)).exists(): + kept.append(block) + path.write_text(head + "".join(kept)) + (out / "sqlbuild_local.toml.example").unlink(missing_ok=True) + + +def main() -> int: + parser = argparse.ArgumentParser(description="Write a DuckDB copy of core DuckShop.") + parser.add_argument("destination", type=pathlib.Path, help="new folder for the DuckDB copy") + args = parser.parse_args() + out: pathlib.Path = args.destination + if out.exists(): + print(f"{out} already exists", file=sys.stderr) + return 1 + with tempfile.TemporaryDirectory() as scratch: + snow = pathlib.Path(scratch) / "snowflake" + copy_core_project(snow) + report = compile_snowflake(snow) + shutil.copytree(snow, out, ignore=shutil.ignore_patterns("target", ".sqlbuild", "logs")) + write_models(snow, out, report) + write_sources(out) + write_declarations(out) + write_project_file(out) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/benchmarks/duckshop/scripts/duckdb_rewrites.py b/benchmarks/duckshop/scripts/duckdb_rewrites.py new file mode 100644 index 0000000000..4715310c03 --- /dev/null +++ b/benchmarks/duckshop/scripts/duckdb_rewrites.py @@ -0,0 +1,623 @@ +"""SQL rewrites that finish Polyglot's Snowflake-to-DuckDB translation of DuckShop's models. + +Used by `duckdb_project.py`. Polyglot translates most Snowflake SQL; these rules cover what it +leaves out or translates with different semantics. Semi-structured values are JSON throughout: +Snowflake's VARIANT, OBJECT and ARRAY all become DuckDB JSON. Snowflake functions DuckDB lacks +become project SQL macros named `sf_`, one per argument count. +""" + +from __future__ import annotations + +import os +import pathlib +import re +from collections.abc import Callable + +_UNQUOTED = "TRIM(CAST(x AS VARCHAR), '\"')" +_SNOWFLAKE_DATE_PARTS = ( + "replace(replace(replace(replace(replace(replace(replace(replace(f, 'TZH:TZM', '%z'), " + "'TZHTZM', '%z'), 'YYYY', '%Y'), 'MM', '%m'), 'DD', '%d'), 'HH24', '%H'), 'MI', '%M'), " + "'SS', '%S')" +) +# Snowflake's regular expression flags as RE2 inline flags; `e` (extract a group) is handled by +# the callers. +_REGEX_FLAGS = ( + "CASE WHEN contains(f, 'i') THEN '(?i)' ELSE '' END || " + "CASE WHEN contains(f, 's') THEN '(?s)' ELSE '' END || " + "CASE WHEN contains(f, 'm') THEN '(?m)' ELSE '' END || p" +) + + +def _parse(function: str, cast: str) -> str: + """Parses `x` with the Snowflake format `f`.""" + return f"CAST({function}({_UNQUOTED}, {_SNOWFLAKE_DATE_PARTS}) AS {cast})" + + +# Each shim's overloads as (parameters, body). A shim with one overload keeps its own name; others +# get the argument count as a suffix, such as sf_to_number_3. +SHIMS: dict[str, list[tuple[str, str]]] = { + "sf_strip_null_value": [("x", "CASE WHEN CAST(x AS VARCHAR) = 'null' THEN NULL ELSE x END")], + "sf_try_to_number": [ + ("x", f"CAST(TRY_CAST({_UNQUOTED} AS DECIMAL(38, 0)) AS BIGINT)"), + ("x, p", f"CAST(TRY_CAST({_UNQUOTED} AS DECIMAL(38, 0)) AS BIGINT)"), + ("x, p, s", f"TRY_CAST({_UNQUOTED} AS DECIMAL(38, 9))"), + ], + "sf_get_path": [("j, p", "CAST(j AS JSON) -> ('$.' || p)")], + "sf_to_timestamp_ntz": [("x", f"TRY_CAST({_UNQUOTED} AS TIMESTAMP)")], + "sf_try_to_timestamp_tz": [ + ("x", f"TRY_CAST({_UNQUOTED} AS TIMESTAMPTZ)"), + ("x, f", _parse("try_strptime", "TIMESTAMPTZ")), + ], + "sf_object_keys": [("j", "json_keys(CAST(j AS JSON))")], + "sf_is_object": [("j", "json_type(CAST(j AS JSON)) = 'OBJECT'")], + "sf_initcap": [ + ( + "x", + "array_to_string(list_transform(string_split(lower(x), ' '), " + "lambda w: upper(w[1]) || w[2:]), ' ')", + ) + ], + "sf_array_size": [("x", "CAST(json_array_length(to_json(x)) AS BIGINT)")], + "sf_regexp_substr": [ + ("s, p", "NULLIF(regexp_extract(s, p), '')"), + ("s, p, pos", "NULLIF(regexp_extract(s, p), '')"), + ("s, p, pos, occ", "NULLIF(regexp_extract(s, p), '')"), + ( + "s, p, pos, occ, f", + f"NULLIF(CASE WHEN contains(f, 'e') THEN regexp_extract(s, {_REGEX_FLAGS}, 1) " + f"ELSE regexp_extract(s, {_REGEX_FLAGS}) END, '')", + ), + ("s, p, pos, occ, f, grp", f"NULLIF(regexp_extract(s, {_REGEX_FLAGS}, grp), '')"), + ], + "sf_to_varchar": [ + ("x", _UNQUOTED), + ("x, f", f"strftime(x, {_SNOWFLAKE_DATE_PARTS})"), + ], + "sf_to_number": [ + ("x", f"CAST(CAST({_UNQUOTED} AS DECIMAL(38, 0)) AS BIGINT)"), + ("x, p", f"CAST(CAST({_UNQUOTED} AS DECIMAL(38, 0)) AS BIGINT)"), + ("x, p, s", f"CAST({_UNQUOTED} AS DECIMAL(38, 9))"), + ], + "sf_parse_url": [ + ( + "u, p", + "{'host': regexp_extract(u, '^[a-zA-Z]+://([^/:?#]+)', 1), " + "'path': regexp_extract(u, '^[a-zA-Z]+://[^/?#]*(/[^?#]*)?', 1), " + "'parameters': map_from_entries(list_transform(list_filter(" + "string_split(regexp_extract(u, '[?]([^#]*)', 1), '&'), lambda kv: kv <> ''), " + "lambda kv: {'key': split_part(kv, '=', 1), 'value': split_part(kv, '=', 2)})), " + "'query': regexp_extract(u, '\\?([^#]*)', 1), " + "'scheme': regexp_extract(u, '^([a-zA-Z]+)://', 1)}", + ) + ], + "sf_is_decimal": [("x", "json_type(to_json(x)) IN ('DOUBLE', 'UBIGINT', 'BIGINT')")], + "sf_is_integer": [("x", "json_type(to_json(x)) IN ('UBIGINT', 'BIGINT')")], + "sf_typeof": [ + ( + "x", + "CASE json_type(to_json(x)) WHEN 'VARCHAR' THEN 'VARCHAR' WHEN 'BIGINT' THEN 'INTEGER' " + "WHEN 'UBIGINT' THEN 'INTEGER' WHEN 'DOUBLE' THEN 'DECIMAL' " + "WHEN 'BOOLEAN' THEN 'BOOLEAN' WHEN 'OBJECT' THEN 'OBJECT' " + "WHEN 'ARRAY' THEN 'ARRAY' ELSE 'NULL_VALUE' END", + ) + ], + "sf_array_contains": [("a, v", "list_contains(CAST(to_json(a) AS JSON[]), to_json(v))")], + "sf_array_flatten": [("x", "flatten(CAST(to_json(x) AS JSON[][]))")], + "sf_try_to_timestamp_ntz": [ + ("x", f"TRY_CAST({_UNQUOTED} AS TIMESTAMP)"), + ("x, f", _parse("try_strptime", "TIMESTAMP")), + ], + "sf_to_timestamp_tz": [ + ("x", f"CAST({_UNQUOTED} AS TIMESTAMPTZ)"), + ("x, f", _parse("strptime", "TIMESTAMPTZ")), + ], + "sf_regexp_like": [ + ("s, p", "regexp_full_match(s, p)"), + ("s, p, f", f"regexp_full_match(s, {_REGEX_FLAGS})"), + ], + "sf_array_distinct": [("x", "list_distinct(CAST(to_json(x) AS JSON[]))")], + "sf_sha2": [("x", "sha256(CAST(x AS VARCHAR))"), ("x, n", "sha256(CAST(x AS VARCHAR))")], + "sf_dateadd": [("u, n, d", "d + CAST(CAST(n AS VARCHAR) || ' ' || u AS INTERVAL)")], + "sf_get": [ + ( + "j, k", + "CASE WHEN json_type(CAST(j AS JSON)) = 'ARRAY' THEN CAST(j AS JSON) -> " + "TRY_CAST(k AS BIGINT) ELSE CAST(j AS JSON) -> CAST(k AS VARCHAR) END", + ) + ], + "sf_object_agg": [ + ( + "k, v", + "to_json(map_from_entries(ARRAY_AGG({'key': CAST(k AS VARCHAR), 'value': to_json(v)}) " + "FILTER (WHERE k IS NOT NULL AND v IS NOT NULL)))", + ) + ], + "sf_hash": [("x", "CAST(hash(x) >> 1 AS BIGINT)")], + "sf_list_count": [ + ( + "a", + "len(list_filter(CAST(to_json(a) AS JSON[]), lambda e: CAST(e AS VARCHAR) <> 'null'))", + ) + ], +} +# Declared result types, by shim or by shim and argument count. +RETURNS: dict[str, str] = { + "sf_strip_null_value": "JSON", + "sf_try_to_number": "BIGINT", + "sf_try_to_number_3": "DECIMAL(38, 9)", + "sf_to_number": "BIGINT", + "sf_to_number_3": "DECIMAL(38, 9)", + "sf_get_path": "JSON", + "sf_get": "JSON", + "sf_to_timestamp_ntz": "TIMESTAMP", + "sf_try_to_timestamp_ntz": "TIMESTAMP", + "sf_try_to_timestamp_tz": "TIMESTAMPTZ", + "sf_to_timestamp_tz": "TIMESTAMPTZ", + "sf_object_keys": "VARCHAR[]", + "sf_is_object": "BOOLEAN", + "sf_initcap": "VARCHAR", + "sf_array_size": "BIGINT", + "sf_regexp_substr": "VARCHAR", + "sf_to_varchar": "VARCHAR", + "sf_is_decimal": "BOOLEAN", + "sf_is_integer": "BOOLEAN", + "sf_typeof": "VARCHAR", + "sf_array_contains": "BOOLEAN", + "sf_array_flatten": "JSON[]", + "sf_regexp_like": "BOOLEAN", + "sf_array_distinct": "JSON[]", + "sf_sha2": "VARCHAR", + "sf_list_count": "BIGINT", + "sf_object_agg": "JSON", + "sf_hash": "BIGINT", +} +# Snowflake functions as Polyglot leaves them, and their shims or DuckDB equivalents. +RENAMES: dict[str, str] = { + "STRIP_NULL_VALUE": "sf_strip_null_value", + "TRY_TO_NUMBER": "sf_try_to_number", + "GET_PATH": "sf_get_path", + "TO_TIMESTAMP_NTZ": "sf_to_timestamp_ntz", + "TRY_TO_TIMESTAMP_TZ": "sf_try_to_timestamp_tz", + "OBJECT_KEYS": "sf_object_keys", + "IS_OBJECT": "sf_is_object", + "INITCAP": "sf_initcap", + "ARRAY_LENGTH": "sf_array_size", + "ARRAY_SIZE": "sf_array_size", + "ARRAY_CONSTRUCT": "json_array", + "OBJECT_CONSTRUCT": "json_object", + "REGEXP_SUBSTR": "sf_regexp_substr", + "TO_VARCHAR": "sf_to_varchar", + "TO_NUMBER": "sf_to_number", + "PARSE_URL": "sf_parse_url", + "OBJECT_AGG": "sf_object_agg", + "HASH": "sf_hash", + "JSON_GROUP_OBJECT": "sf_object_agg", + "IS_DECIMAL": "sf_is_decimal", + "IS_INTEGER": "sf_is_integer", + "TYPEOF": "sf_typeof", + "ARRAY_CONTAINS": "sf_array_contains", + "LIST_COUNT": "sf_list_count", + "ARRAY_DISTINCT": "sf_array_distinct", + "LIST_DISTINCT": "sf_array_distinct", + "SHA2": "sf_sha2", + "DATEADD": "sf_dateadd", + "FLATTEN": "sf_array_flatten", + "TRY_TO_TIMESTAMP_NTZ": "sf_try_to_timestamp_ntz", + "TO_TIMESTAMP_TZ": "sf_to_timestamp_tz", + "REGEXP_LIKE": "sf_regexp_like", +} +LIST_FUNCTIONS = { + "LIST_TRANSFORM": "list_transform", + "TRANSFORM": "list_transform", + "LIST_FILTER": "list_filter", + "FILTER": "list_filter", + "LIST_MIN": "list_min", + "LIST_MAX": "list_max", +} +FLATTEN_COLUMNS = re.compile( + r'\)\s+AS\s+(\w+)\(\s*SEQ,\s*KEY,\s*PATH,\s*"INDEX",\s*VALUE,\s*THIS\s*\)' +) +SNOWFLAKE_FORMATS = [ + ("YYYY", "%Y"), + ("HH24", "%H"), + ("MON", "%b"), + ("MM", "%m"), + ("DD", "%d"), + ("DY", "%a"), + ("MI", "%M"), + ("SS", "%S"), +] +# Columns that carry JSON values through a CTE, where a text cast must unquote them, by model. +JSON_COLUMNS = { + "int_product_specs_flat": {"pv.material", "pv.colour_family", "pv.dishwasher_safe"}, + "int_survey_answers_flat": {"t.answer_value"}, +} +JSON_FUNCTIONS = ("sf_strip_null_value", "sf_get", "sf_get_path", "to_json") + + +def balanced(text: str, start: int) -> int: + """Index just past the parenthesis that closes the one opened before `start`.""" + + depth, i, quote = 1, start, None + while depth: + c = text[i] + if quote: + if c == quote: + quote = None + elif c in "'\"": + quote = c + elif c == "(": + depth += 1 + elif c == ")": + depth -= 1 + i += 1 + return i + + +def split_args(text: str) -> list[str]: + """Top-level comma-separated arguments, keeping a lambda's parameters together.""" + + args, depth, quote, start = [], 0, None, 0 + for i, c in enumerate(text): + if quote: + if c == quote: + quote = None + elif c in "'\"": + quote = c + elif c in "([{": + depth += 1 + elif c in ")]}": + depth -= 1 + elif c == "," and depth == 0: + args.append(text[start:i]) + start = i + 1 + args.append(text[start:]) + merged: list[str] = [] + for arg in args: + if merged and re.match(r"\s*LAMBDA\b", merged[-1], re.I) and ":" not in merged[-1]: + merged[-1] += "," + arg + else: + merged.append(arg) + return merged + + +def calls(sql: str, name: str, replace: Callable[[list[str]], str]) -> str: + """Rewrite every call of `name`, innermost first, handing its argument list to `replace`.""" + + pattern = re.compile(rf"(? str: + """Polyglot's LATERAL UNNEST of a FLATTEN call, as a subquery with FLATTEN's columns.""" + + out, i = [], 0 + for m in re.finditer(r"LATERAL UNNEST\(\s*input => ", sql): + if m.start() < i: + continue + end = balanced(sql, m.end()) + args = sql[m.end() : end - 1] + outer = re.search(r',\s*"outer" => TRUE', args) is not None + expr = re.sub(r',\s*"outer" => TRUE', "", args) + alias = FLATTEN_COLUMNS.match(sql, end - 1) + if not alias: + continue + name = alias.group(1) + nulls = "CAST(NULL AS BIGINT) AS seq" + if re.search(rf"(? f.k AS value, ' + f"CAST(NULL AS JSON) AS this " + f"FROM (SELECT UNNEST(json_keys({obj})) AS k) AS f) AS {name}" + ) + else: + # An array's elements: INDEX and VALUE per element; OUTER keeps one null row. + items = f"CAST(to_json({expr}) AS JSON[])" + if outer: + items = ( + f"CASE WHEN COALESCE(len({items}), 0) = 0 THEN [CAST(NULL AS JSON)] " + f"ELSE {items} END" + ) + body = ( + f"(SELECT {nulls}, CAST(NULL AS VARCHAR) AS key, CAST(NULL AS VARCHAR) AS path, " + f'CAST(f.i - 1 AS BIGINT) AS "index", f.e AS value, CAST(NULL AS JSON) AS this ' + f"FROM (SELECT UNNEST({items}) AS e, generate_subscripts({items}, 1) AS i) AS f) " + f"AS {name}" + ) + out.append(sql[i : m.start()]) + out.append("LATERAL " + body) + i = alias.end() + out.append(sql[i:]) + return "".join(out) + + +def lambda_arg(arg: str) -> str: + """Snowflake's `x -> body` lambda as DuckDB's `lambda x: body`. Its list holds JSON, so a text + cast of the element unquotes it, as Snowflake's ::VARCHAR does.""" + + m = re.match(r"^\s*(?:LAMBDA\s+)?(\w+)\s*(?:->|:)", arg, re.I) + if m: + element = m.group(1) + arg = re.sub( + rf"CAST\({element} AS (TEXT|VARCHAR)\)", f"json_extract_string({element}, '$')", arg + ) + return re.sub(r"^(\s*)(\w+) -> ", r"\1lambda \2: ", arg) + + +def json_list(arg: str) -> str: + arg = arg.strip() + if arg.startswith(("CAST(to_json(", "list_")): + return arg + return f"CAST(to_json({arg}) AS JSON[])" + + +def casts(sql: str) -> str: + """CAST(x AS VARIANT|OBJECT|ARRAY) makes a JSON value, as Snowflake makes a variant.""" + + i = 0 + pattern = re.compile(r"(? str: + """Snowflake's ASOF JOIN ... MATCH_CONDITION (c) ON k as DuckDB's ASOF JOIN ... ON k AND c.""" + + while m := re.search(r"\s*MATCH_CONDITION \(", sql): + end = balanced(sql, m.end()) + condition = sql[m.end() : end - 1] + rest = sql[end:] + on = re.match(r"(\s*ON [^\n]+)", rest) + if on is None: + raise ValueError("MATCH_CONDITION without an ON clause on the next line") + sql = sql[: m.start()] + on.group(1) + f" AND {condition}" + rest[on.end() :] + return sql + + +def strftime_format(m: re.Match[str]) -> str: + fmt = m.group(2) + if "%" not in fmt: + for old, new in SNOWFLAKE_FORMATS: + fmt = fmt.replace(old, new) + return f"{m.group(1)}'{fmt}')" + + +def json_text(sql: str, model: str | None) -> str: + """CAST(json AS TEXT) keeps a JSON string's quotes in DuckDB; Snowflake's ::VARCHAR drops + them, so a text cast of a JSON value reads it as a string.""" + + pattern = re.compile(r"(? '[^']*'", bare) + or re.fullmatch(r"\w+\.value", bare) + or (head and head.group(1) == "COALESCE" and "to_json(" in bare) + ) + if t and jsonish: + text = f"json_extract_string({value}, '$') AS {t.group(1)}" + sql = sql[: m.end()] + text + sql[end - 1 :] + i = m.end() + return sql + + +def list_call(name: str, target: str) -> Callable[[list[str]], str]: + def replace(args: list[str]) -> str: + if args[0].strip().upper().startswith("WHERE"): + return f"{name}({','.join(args)})" + return f"{target}({', '.join([json_list(args[0]), *map(lambda_arg, args[1:])])})" + + return replace + + +def list_functions(sql: str) -> str: + """Snowflake's array functions on JSON lists, as DuckDB's list functions.""" + + for name, target in LIST_FUNCTIONS.items(): + sql = calls(sql, name, list_call(name, target)) + sql = calls( + sql, "LIST_APPEND", lambda a: f"list_append({json_list(a[0])}, to_json({a[1].strip()}))" + ) + sql = calls( + sql, "LIST_PREPEND", lambda a: f"list_prepend(to_json({a[0].strip()}), {json_list(a[1])})" + ) + sql = calls(sql, "LIST_SORT", lambda a: f"list_sort({', '.join([json_list(a[0]), *a[1:]])})") + sql = calls(sql, "LIST_CONCAT", lambda a: f"list_concat({', '.join(map(json_list, a))})") + return calls( + sql, + "REDUCE", + lambda a: f"list_reduce({json_list(a[0])}, {a[2].strip()}, {a[1].strip()})", + ) + + +def rewrite(sql: str, model: str | None = None) -> str: + """Finish Polyglot's DuckDB translation of one model, function or expression.""" + + sql = flatten(sql) + # A flattened JSON string read as text loses its quotes, as Snowflake's ::VARCHAR does. + sql = re.sub(r"CAST\((\w+)\.value AS (TEXT|VARCHAR)\)", r"(\1.value ->> '$')", sql) + sql = re.sub( + r"(STRFTIME\((?:[^()]|\([^()]*\))*?, )'([^']*)'\)", strftime_format, sql, flags=re.I + ) + sql = re.sub(r"(? '$.\2'", sql, flags=re.I) + sql = re.sub(r"CAST\((JSON\('[^']*'\))\[([\w.]+)\] AS TEXT\)", r"(\1 ->> \2)", sql) + sql = re.sub( + r"(? ('[^']*') AS (TEXT|VARCHAR)\)", r"(\1 ->> \2)", sql) + sql = re.sub( + r"CAST\(sf_strip_null_value\(([\w.]+) -> ('[^']*')\) AS (TEXT|VARCHAR)\)", + r"(\1 ->> \2)", + sql, + ) + sql = json_text(sql, model) + arrow = r"[\w.]+ -> '[^']*'" + return re.sub( + rf"CAST\(COALESCE\(((?:{arrow}, )+{arrow})\) AS (TEXT|VARCHAR)\)", + lambda m: "COALESCE(" + m.group(1).replace(" -> ", " ->> ") + ")", + sql, + ) + + +def replace_cte(sql: str, name: str, body: str) -> str: + m = re.search(rf"(? str: + return replace_cte( + sql, + "generated", + " SELECT CAST(d AS DATE) AS calendar_date\n" + " FROM range(DATE '2025-01-01', DATE '2025-01-01' + INTERVAL 1095 DAY, INTERVAL 1 DAY)" + " AS t(d)", + ) + + +def _category_paths(sql: str) -> str: + tree = ( + " SELECT c.category_id, c.parent_category_id, c.slug, c.display_name, c.sort_order,\n" + " 1 AS depth, '/' || c.slug AS slug_path, c.category_id AS root_category_id\n" + " FROM categories AS c\n WHERE c.is_root\n UNION ALL\n" + " SELECT c.category_id, c.parent_category_id, c.slug, c.display_name, c.sort_order,\n" + " t.depth + 1, t.slug_path || '/' || c.slug, t.root_category_id\n" + " FROM categories AS c\n INNER JOIN tree AS t ON c.parent_category_id = t.category_id" + ) + return re.sub(r"^WITH ", "WITH RECURSIVE ", replace_cte(sql, "tree", tree), count=1) + + +# Snowflake constructs with no mechanical DuckDB form (GENERATOR, CONNECT BY), rewritten by hand. +OVERRIDES: dict[str, Callable[[str], str]] = { + "dim_date": _dim_date, + "int_category_paths": _category_paths, +} + + +def convert(sql: str, model: str | None = None) -> str: + sql = rewrite(sql, model) + return OVERRIDES[model](sql) if model in OVERRIDES else sql + + +def shim_function_name(name: str, arity: int) -> str: + return f"{name}_{arity}" if len(SHIMS[name]) > 1 else name + + +def shim_calls(sql: str) -> str: + """Calls of shims as calls of their project SQL macros.""" + + longest_first = sorted(SHIMS, key=lambda name: len(name), reverse=True) + for name in longest_first: + sql = calls(sql, name, shim_call(name)) + return sql + + +def shim_call(name: str) -> Callable[[list[str]], str]: + def replace(args: list[str]) -> str: + return f"@sql.{shim_function_name(name, len(args))}({','.join(args)})" + + return replace + + +def parameters(body: str, names: list[str]) -> str: + """Each parameter as `@name`, outside string literals and lambda parameters.""" + + parts = re.split(r"('(?:[^']|'')*')", body) + for index in range(0, len(parts), 2): + for name in names: + parts[index] = re.sub(rf"(? pathlib.Path: + """The declaration folder that owns a macro called from `folders`, as SQLBuild requires: + private to one folder, or public at the folders' closest common folder.""" + + if len(folders) == 1: + return folders[0] / "_sqlbuild" / "_macros" / "sql" + common = pathlib.Path(os.path.commonpath(folders)) + if common in (root, root / "models"): + return root / "macros" / "sql" + return common / "_sqlbuild" / "macros" / "sql" + + +def write_shims(root: pathlib.Path) -> None: + """Write each shim overload a model calls, in the folder that owns its callers.""" + + callers: dict[str, set[pathlib.Path]] = {} + for path in root.glob("models/**/*.sql"): + for name in re.findall(r"@sql\.(\w+)\(", path.read_text()): + callers.setdefault(name, set()).add(path.parent) + for name, overloads in SHIMS.items(): + for params, body in overloads: + names = [p.strip() for p in params.split(",")] + macro = shim_function_name(name, len(names)) + if macro not in callers: + continue + directory = macro_folder(root, sorted(callers[macro])) + directory.mkdir(parents=True, exist_ok=True) + returns = RETURNS.get(macro, RETURNS.get(name)) + function = name.removeprefix("sf_").upper() + arguments = ", ".join(f"{p} ANY" for p in names) + (directory / f"{macro}.sql").write_text( + "MACRO (\n" + f" description \"Snowflake's {function} for DuckDB, " + f'taking {len(names)} arguments",\n' + f" arguments ({arguments}),\n" + + (f" returns {returns},\n" if returns else "") + + ");\n\n" + f"{parameters(body, names)}\n" + ) diff --git a/clients/typescript/src/schema.d.ts b/clients/typescript/src/schema.d.ts index 37ba35a5fa..0a2c10dcbb 100644 --- a/clients/typescript/src/schema.d.ts +++ b/clients/typescript/src/schema.d.ts @@ -258,6 +258,22 @@ export interface paths { patch?: never; trace?: never; }; + "/api/control/locations/{location_id}/targets/{target}/acknowledgements": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + get: operations["listRunAcknowledgements"]; + put?: never; + post: operations["acknowledgeRun"]; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; "/api/control/locations/{location_id}/targets/{target}/catalog": { parameters: { query?: never; @@ -1270,6 +1286,14 @@ export interface paths { export type webhooks = Record; export interface components { schemas: { + /** @description Marks a run as handled for everyone who can see it, or withdraws the mark; retried with the same event ID it applies once. */ + AcknowledgementRequest: { + acknowledged: boolean; + /** Format: uuid */ + event_id: string; + /** Format: uuid */ + run_id: string; + }; /** * @description Closed permission vocabulary; unknown actions fail deserialization. * @enum {string} @@ -2387,6 +2411,15 @@ export interface components { nodes: string[]; permissions: components["schemas"]["Action"][]; }; + /** @description A run whose latest acknowledgement event marks it handled. */ + RunAcknowledgement: { + /** Format: date-time */ + acknowledged_at: string; + /** @description The person's name, or the audit actor when the person no longer exists. */ + acknowledged_by: string; + /** Format: uuid */ + run_id: string; + }; /** @description This is a server-owned instruction, not an arbitrary status supplied by a worker. */ RunChange: { container_id?: string | null; @@ -3681,6 +3714,95 @@ export interface operations { }; }; }; + listRunAcknowledgements: { + parameters: { + query?: never; + header?: never; + path: { + location_id: string; + target: string; + }; + cookie?: never; + }; + requestBody?: never; + responses: { + /** @description Runs marked handled in the last week */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["RunAcknowledgement"][]; + }; + }; + /** @description Session rejected */ + 401: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + 403: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["ApiErrorBody"]; + }; + }; + }; + }; + acknowledgeRun: { + parameters: { + query?: never; + header: { + "x-sqb-csrf": string; + }; + path: { + location_id: string; + target: string; + }; + cookie?: never; + }; + requestBody: { + content: { + "application/json": components["schemas"]["AcknowledgementRequest"]; + }; + }; + responses: { + /** @description Whether the run's acknowledgement changed */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["MutationResponse"]; + }; + }; + /** @description The run is not in this location and target, or the event ID was reused */ + 400: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + /** @description Session rejected */ + 401: { + headers: { + [name: string]: unknown; + }; + content?: never; + }; + 403: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["ApiErrorBody"]; + }; + }; + }; + }; readPersonCatalog: { parameters: { query: { diff --git a/crates/sqlbuild-orchestration/Cargo.toml b/crates/sqlbuild-orchestration/Cargo.toml index bd64823fdf..edbf035ab1 100644 --- a/crates/sqlbuild-orchestration/Cargo.toml +++ b/crates/sqlbuild-orchestration/Cargo.toml @@ -35,6 +35,7 @@ subtle.workspace = true thiserror.workspace = true tokio.workspace = true toml.workspace = true +tower-http.workspace = true utoipa.workspace = true uuid.workspace = true diff --git a/crates/sqlbuild-orchestration/src/acknowledgements/main/acknowledge_person_run.rs b/crates/sqlbuild-orchestration/src/acknowledgements/main/acknowledge_person_run.rs new file mode 100644 index 0000000000..cbd42eca59 --- /dev/null +++ b/crates/sqlbuild-orchestration/src/acknowledgements/main/acknowledge_person_run.rs @@ -0,0 +1,94 @@ +//! Anyone who can see a location's runs marks one handled as an audit event; refusals precede the run lookup. + +use serde_json::json; +use uuid::Uuid; + +use crate::acknowledgements::models::AcknowledgementRequest; +use crate::authorization::main::authorize_listing::authorize_listing; +use crate::credentials::main::audit::append_audit::append_audit; +use crate::credentials::models::{AuditContext, DefaultRole}; +use crate::store::errors::ControlError; +use crate::store::main::{begin::begin, commit::commit, query::query}; +use crate::store::models::{Change, ControlStore, Parameter, TransactionMode}; + +/// Returns whether the run's acknowledgement changed; each change signals the run, so open overviews reload. +pub async fn acknowledge_person_run( + store: &ControlStore, + (context, credentials): (&AuditContext, Uuid), + (location_id, target, request): (Uuid, &str, &AcknowledgementRequest), + default_role: DefaultRole, +) -> Result { + let (transaction, context) = authorize_listing( + begin(store, TransactionMode::Write).await?, + (context, credentials), + (location_id, target), + default_role, + ) + .await?; + let (transaction, runs) = query( + transaction, + "SELECT run_id FROM sqb_runs WHERE run_id=$1 AND location_id=$2 AND target=$3", + vec![ + Parameter::Id(request.run_id), + Parameter::Id(location_id), + Parameter::Text(target.into()), + ], + ) + .await?; + if runs.is_empty() { + return Err(ControlError::Invalid( + "run does not exist in this location and target".into(), + )); + } + let body = + serde_json::to_value(request).map_err(|error| ControlError::Invalid(error.to_string()))?; + let (transaction, retried) = query( + transaction, + "SELECT scope_json FROM sqb_security_audit WHERE event_id=$1", + vec![Parameter::Id(request.event_id)], + ) + .await?; + if let Some(previous) = retried.first() { + if previous["scope_json"].get("request") != Some(&body) { + return Err(ControlError::Invalid( + "acknowledgement event ID already has different content".into(), + )); + } + commit(transaction).await?; + return Ok(false); + } + let (transaction, latest) = query( + transaction, + "SELECT action FROM sqb_security_audit \ + WHERE action IN ('run.acknowledged','run.acknowledgement_withdrawn') AND scope_json ->> 'run_id'=$1 \ + ORDER BY storage_id DESC LIMIT 1", + vec![Parameter::Text(request.run_id.to_string())], + ) + .await?; + let acknowledged = latest + .first() + .is_some_and(|row| row["action"].as_str() == Some("run.acknowledged")); + if acknowledged == request.acknowledged { + commit(transaction).await?; + return Ok(false); + } + let action = if request.acknowledged { + "run.acknowledged" + } else { + "run.acknowledgement_withdrawn" + }; + let mut transaction = append_audit( + transaction, + &context, + (request.event_id, action), + ( + json!({"location_id": location_id, "target": target, "run_id": request.run_id, "request": body}), + json!({"acknowledged": acknowledged}), + json!({"acknowledged": request.acknowledged}), + ), + ) + .await?; + transaction.changes.insert(Change::Run(request.run_id)); + commit(transaction).await?; + Ok(true) +} diff --git a/crates/sqlbuild-orchestration/src/acknowledgements/main/list_person_acknowledgements.rs b/crates/sqlbuild-orchestration/src/acknowledgements/main/list_person_acknowledgements.rs new file mode 100644 index 0000000000..2e51fb07b3 --- /dev/null +++ b/crates/sqlbuild-orchestration/src/acknowledgements/main/list_person_acknowledgements.rs @@ -0,0 +1,59 @@ +//! A run's acknowledgement is its latest acknowledgement event, projected from the audit log. + +use std::collections::BTreeMap; + +use chrono::{DateTime, Duration, Utc}; +use uuid::Uuid; + +use crate::acknowledgements::models::RunAcknowledgement; +use crate::authorization::main::authorize_listing::authorize_listing; +use crate::credentials::models::{AuditContext, DefaultRole}; +use crate::store::errors::ControlError; +use crate::store::main::{begin::begin, commit::commit, query::query}; +use crate::store::models::{ControlStore, Parameter, TransactionMode}; + +/// Runs marked handled in the last week; the overview lists only a day of problems. +pub async fn list_person_acknowledgements( + store: &ControlStore, + (context, credentials): (&AuditContext, Uuid), + (location_id, target): (Uuid, &str), + default_role: DefaultRole, +) -> Result, ControlError> { + let (transaction, _) = authorize_listing( + begin(store, TransactionMode::Write).await?, + (context, credentials), + (location_id, target), + default_role, + ) + .await?; + let since = transaction.now - Duration::days(7); + let (transaction, rows) = query( + transaction, + "SELECT a.action, COALESCE(p.name, a.actor) AS actor, a.occurred_at, a.scope_json ->> 'run_id' AS run_id \ + FROM sqb_security_audit a LEFT JOIN sqb_principals p ON 'principal:' || CAST(p.principal_id AS TEXT)=a.actor \ + WHERE a.occurred_at >= $1 AND a.action IN ('run.acknowledged','run.acknowledgement_withdrawn') \ + AND a.scope_json ->> 'location_id'=$2 AND a.scope_json ->> 'target'=$3 ORDER BY a.storage_id", + vec![ + Parameter::Time(since), + Parameter::Text(location_id.to_string()), + Parameter::Text(target.into()), + ], + ) + .await?; + commit(transaction).await?; + let mut latest: BTreeMap> = BTreeMap::new(); + for row in rows { + let run_id: Uuid = serde_json::from_value(row["run_id"].clone()) + .map_err(|error| ControlError::Schema(format!("acknowledgement run: {error}")))?; + let acknowledged_at: DateTime = serde_json::from_value(row["occurred_at"].clone()) + .map_err(|error| ControlError::Schema(format!("acknowledgement time: {error}")))?; + let marked = + (row["action"].as_str() == Some("run.acknowledged")).then(|| RunAcknowledgement { + run_id, + acknowledged_by: row["actor"].as_str().unwrap_or_default().into(), + acknowledged_at, + }); + latest.insert(run_id, marked); + } + Ok(latest.into_values().flatten().collect()) +} diff --git a/crates/sqlbuild-orchestration/src/acknowledgements/main/mod.rs b/crates/sqlbuild-orchestration/src/acknowledgements/main/mod.rs new file mode 100644 index 0000000000..7078a48559 --- /dev/null +++ b/crates/sqlbuild-orchestration/src/acknowledgements/main/mod.rs @@ -0,0 +1,2 @@ +pub mod acknowledge_person_run; +pub mod list_person_acknowledgements; diff --git a/crates/sqlbuild-orchestration/src/acknowledgements/mod.rs b/crates/sqlbuild-orchestration/src/acknowledgements/mod.rs new file mode 100644 index 0000000000..4fe9f2045f --- /dev/null +++ b/crates/sqlbuild-orchestration/src/acknowledgements/mod.rs @@ -0,0 +1,4 @@ +//! People mark problem runs handled for everyone who can see them; the audit log holds every change. + +pub mod main; +pub mod models; diff --git a/crates/sqlbuild-orchestration/src/acknowledgements/models.rs b/crates/sqlbuild-orchestration/src/acknowledgements/models.rs new file mode 100644 index 0000000000..3f6c3b108f --- /dev/null +++ b/crates/sqlbuild-orchestration/src/acknowledgements/models.rs @@ -0,0 +1,23 @@ +//! Acknowledgements of problem runs, kept as audit events. + +use chrono::{DateTime, Utc}; +use serde::{Deserialize, Serialize}; +use uuid::Uuid; + +/// Marks a run as handled for everyone who can see it, or withdraws the mark; retried with the same event ID it applies once. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)] +#[serde(deny_unknown_fields)] +pub struct AcknowledgementRequest { + pub event_id: Uuid, + pub run_id: Uuid, + pub acknowledged: bool, +} + +/// A run whose latest acknowledgement event marks it handled. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize, utoipa::ToSchema)] +pub struct RunAcknowledgement { + pub run_id: Uuid, + /// The person's name, or the audit actor when the person no longer exists. + pub acknowledged_by: String, + pub acknowledged_at: DateTime, +} diff --git a/crates/sqlbuild-orchestration/src/api/_helpers/authentication/password_sessions.rs b/crates/sqlbuild-orchestration/src/api/_helpers/authentication/password_sessions.rs index 916a5810d0..7788726caf 100644 --- a/crates/sqlbuild-orchestration/src/api/_helpers/authentication/password_sessions.rs +++ b/crates/sqlbuild-orchestration/src/api/_helpers/authentication/password_sessions.rs @@ -126,6 +126,11 @@ pub(crate) fn person_routes() -> Router { "/api/control/locations/{location_id}/targets/{target}/runs", get(crate::api::_helpers::resources::person_operations::recent_runs), ) + .route( + "/api/control/locations/{location_id}/targets/{target}/acknowledgements", + get(crate::api::_helpers::resources::person_operations::acknowledgements) + .post(crate::api::_helpers::resources::person_operations::acknowledge), + ) .route( "/api/control/locations/{location_id}/targets/{target}/runs/{run_id}/nodes", get(crate::api::_helpers::resources::person_reads::run_nodes), diff --git a/crates/sqlbuild-orchestration/src/api/_helpers/resources/compression.rs b/crates/sqlbuild-orchestration/src/api/_helpers/resources/compression.rs new file mode 100644 index 0000000000..23745598ff --- /dev/null +++ b/crates/sqlbuild-orchestration/src/api/_helpers/resources/compression.rs @@ -0,0 +1,17 @@ +//! Only people's reads and UI assets are compressed, never a response holding the CSRF token. + +use axum::extract::Request; +use axum::http::{Method, header}; +use axum::middleware::Next; +use axum::response::Response; + +pub(crate) async fn compress_reads_only(mut request: Request, next: Next) -> Response { + let path = request.uri().path(); + let read = path.starts_with("/api/control/") + || path.starts_with("/api/admin/") + || !path.starts_with("/api/"); + if request.method() != Method::GET || !read { + request.headers_mut().remove(header::ACCEPT_ENCODING); + } + next.run(request).await +} diff --git a/crates/sqlbuild-orchestration/src/api/_helpers/resources/mod.rs b/crates/sqlbuild-orchestration/src/api/_helpers/resources/mod.rs index 043cd0ad73..376e0414f6 100644 --- a/crates/sqlbuild-orchestration/src/api/_helpers/resources/mod.rs +++ b/crates/sqlbuild-orchestration/src/api/_helpers/resources/mod.rs @@ -2,6 +2,7 @@ pub(crate) mod administrators; pub(crate) mod audit; pub(crate) mod changes; pub(crate) mod client; +pub(crate) mod compression; pub(crate) mod docker_image; pub(crate) mod location_health; pub(crate) mod location_work; diff --git a/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_operations.rs b/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_operations.rs index 97eb53cf64..602c4e4621 100644 --- a/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_operations.rs +++ b/crates/sqlbuild-orchestration/src/api/_helpers/resources/person_operations.rs @@ -6,6 +6,9 @@ use axum::extract::{Path, Query, State}; use axum::{Extension, Json}; use uuid::Uuid; +use crate::acknowledgements::main::acknowledge_person_run::acknowledge_person_run; +use crate::acknowledgements::main::list_person_acknowledgements::list_person_acknowledgements; +use crate::acknowledgements::models::{AcknowledgementRequest, RunAcknowledgement}; use crate::api::errors::ApiError; use crate::api::models::{MutationResponse, RunsQuery}; use crate::credentials::models::AuthenticatedPerson; @@ -157,6 +160,55 @@ pub(crate) async fn rerun( })) } +#[utoipa::path(post, path = "/api/control/locations/{location_id}/targets/{target}/acknowledgements", operation_id = "acknowledgeRun", + params(("location_id" = Uuid, Path), ("target" = String, Path), ("x-sqb-csrf" = String, Header)), + request_body = AcknowledgementRequest, + responses((status = 200, body = MutationResponse, description = "Whether the run's acknowledgement changed"), + (status = 400, description = "The run is not in this location and target, or the event ID was reused"), + (status = 401, description = "Session rejected"), + (status = 403, body = crate::api::models::ApiErrorBody)), security(("session" = []), ("cliSession" = [])))] +pub(crate) async fn acknowledge( + (State(store), Extension(context)): (State, Extension), + (Path((location_id, target)), Extension(default_role)): ( + Path<(Uuid, String)>, + Extension, + ), + Extension(AuthenticatedPerson(principal, _)): Extension, + Json(request): Json, +) -> Result, ApiError> { + Ok(Json(MutationResponse { + applied: acknowledge_person_run( + &store, + (&context, principal), + (location_id, &target, &request), + default_role, + ) + .await?, + })) +} + +#[utoipa::path(get, path = "/api/control/locations/{location_id}/targets/{target}/acknowledgements", operation_id = "listRunAcknowledgements", + params(("location_id" = Uuid, Path), ("target" = String, Path)), + responses((status = 200, body = Vec, description = "Runs marked handled in the last week"), + (status = 401, description = "Session rejected"), + (status = 403, body = crate::api::models::ApiErrorBody)), security(("session" = []), ("cliSession" = [])))] +pub(crate) async fn acknowledgements( + (State(store), Extension(context)): (State, Extension), + Extension(default_role): Extension, + Path((location_id, target)): Path<(Uuid, String)>, + Extension(AuthenticatedPerson(principal, _)): Extension, +) -> Result>, ApiError> { + Ok(Json( + list_person_acknowledgements( + &store, + (&context, principal), + (location_id, &target), + default_role, + ) + .await?, + )) +} + #[utoipa::path(post, path = "/api/control/triggers", operation_id = "changeTrigger", params(("x-sqb-csrf" = String, Header)), request_body = TriggerChangeRequest, responses((status = 200, body = MutationResponse), (status = 401, description = "Session rejected"), diff --git a/crates/sqlbuild-orchestration/src/api/main/serve_api.rs b/crates/sqlbuild-orchestration/src/api/main/serve_api.rs index 87447e8f93..587f838c6f 100644 --- a/crates/sqlbuild-orchestration/src/api/main/serve_api.rs +++ b/crates/sqlbuild-orchestration/src/api/main/serve_api.rs @@ -2,8 +2,10 @@ use axum::routing::get; use axum::{Extension, Json, Router, middleware}; +use tower_http::compression::CompressionLayer; use crate::api::_helpers::authentication::{bearer::local_request, configure::configure}; +use crate::api::_helpers::resources::compression::compress_reads_only; use crate::api::_helpers::resources::server_errors::generic_server_errors; use crate::api::_helpers::resources::ui_assets::{UI_ASSETS, serve_ui}; use crate::api::_helpers::resources::{changes, router::router}; @@ -82,5 +84,8 @@ fn development_application( .layer(middleware::from_fn_with_state(address, local_request)), None => application, }; - application.layer(middleware::map_response(generic_server_errors)) + application + .layer(middleware::map_response(generic_server_errors)) + .layer(CompressionLayer::new()) + .layer(middleware::from_fn(compress_reads_only)) } diff --git a/crates/sqlbuild-orchestration/src/api/models.rs b/crates/sqlbuild-orchestration/src/api/models.rs index dd7cd5a91b..c52635f5df 100644 --- a/crates/sqlbuild-orchestration/src/api/models.rs +++ b/crates/sqlbuild-orchestration/src/api/models.rs @@ -26,6 +26,8 @@ use crate::api::_helpers::resources::{ person_operations::rerun, person_operations::change_trigger, person_operations::recent_runs, + person_operations::acknowledge, + person_operations::acknowledgements, person_reads::run_nodes, person_reads::nodes, person_reads::node_runs, diff --git a/crates/sqlbuild-orchestration/src/lib.rs b/crates/sqlbuild-orchestration/src/lib.rs index 213b03ccff..8a75d6bef1 100644 --- a/crates/sqlbuild-orchestration/src/lib.rs +++ b/crates/sqlbuild-orchestration/src/lib.rs @@ -2,6 +2,7 @@ #![forbid(unsafe_code)] +pub mod acknowledgements; pub mod api; pub mod authorization; pub mod credentials; diff --git a/crates/sqlbuild-orchestration/src/runs/_helpers/groups/lifecycle.rs b/crates/sqlbuild-orchestration/src/runs/_helpers/groups/lifecycle.rs index 44894e2600..66dce5299e 100644 --- a/crates/sqlbuild-orchestration/src/runs/_helpers/groups/lifecycle.rs +++ b/crates/sqlbuild-orchestration/src/runs/_helpers/groups/lifecycle.rs @@ -149,11 +149,15 @@ async fn finish_group( } } let mut outcomes: BTreeSet = BTreeSet::new(); + let mut good = false; + let build = parent.request_json.command.is_build(); + let source_needs_no_run = |key: &str| build && key.starts_with("source:"); for key in &parent.request_json.selection { let node = nodes .get(key) .ok_or_else(|| ControlError::Invalid("group node has no child evidence".into()))?; outcomes.insert(node.status.clone()); + good |= matches!(node.status.as_str(), "success" | "warn") && !source_needs_no_run(key); let outcome: NodeOutcome = serde_json::from_value(serde_json::json!(node.status)) .map_err(|error| ControlError::Invalid(error.to_string()))?; transaction = group_change( @@ -170,7 +174,6 @@ async fn finish_group( ) .await?; } - let good = outcomes.contains("success") || outcomes.contains("warn"); let bad = group .children .iter() diff --git a/crates/sqlbuild-orchestration/tests/api/test_acknowledgements.rs b/crates/sqlbuild-orchestration/tests/api/test_acknowledgements.rs new file mode 100644 index 0000000000..cfb5cbf894 --- /dev/null +++ b/crates/sqlbuild-orchestration/tests/api/test_acknowledgements.rs @@ -0,0 +1,147 @@ +//! A person who can see a location's runs marks one handled, everyone sees who did, and the mark can be withdrawn. + +use serde_json::{Value, json}; +use uuid::Uuid; + +use crate::api::models::{ApiAuthentication, HumanLogin}; +use crate::api::tests::denials_helpers::person_send; +use crate::api::tests::helpers::{start_api_with_authentication, stop_api}; +use crate::api::tests::test_types::AcknowledgementTestCase; +use crate::credentials::tests::location_helpers::create_location; +use crate::credentials::tests::role_helpers::role_fixture; +use crate::runs::main::request_job_run::request_job_run; +use crate::runs::models::JobRunRequest; +use crate::runs::tests::helpers::snapshot; +use crate::snapshots::main::publish_snapshot::publish_snapshot; +use crate::store::main::migrate_store::migrate_store; +use crate::store::tests::store_helpers::{close_test_store, open_test_store, run}; +use crate::store::tests::store_test_types::Backend; + +#[test] +fn given_a_failed_run_when_a_person_acknowledges_it_then_everyone_reads_who_did_until_it_is_withdrawn() + { + let test_cases = [ + AcknowledgementTestCase { + description: "SQLite", + backend: Backend::Sqlite, + expected_statuses: [200, 200, 400, 200], + expected_applied: [true, false, true], + expected_acknowledged_by: "orders-admin", + }, + AcknowledgementTestCase { + description: "PostgreSQL", + backend: Backend::Postgres, + expected_statuses: [200, 200, 400, 200], + expected_applied: [true, false, true], + expected_acknowledged_by: "orders-admin", + }, + ]; + for test_case in test_cases { + run(async { + let Some(fixture) = open_test_store(test_case.backend).await else { + return; + }; + let store = &fixture.store; + migrate_store(store).await.expect("migrate"); + let roles = role_fixture(store).await; + let inventory = create_location(store, "inventory", "orders-admin") + .await + .expect("location"); + let mut published = snapshot(inventory); + published.documents.insert( + "jobs".into(), + json!({"orders": {"description": "Build orders", "selection": ["orders"]}}), + ); + publish_snapshot(store, &published).await.expect("snapshot"); + let run_id = Uuid::now_v7(); + let queued = JobRunRequest { + run_id, + location_id: inventory, + target: "development".into(), + job_key: "orders".into(), + reason: "Refresh orders".into(), + priority: 0, + window: None, + full_refresh: false, + select: Vec::new(), + exclude: Vec::new(), + }; + assert!( + request_job_run(store, (&roles.context, None), &queued) + .await + .expect("queued run") + ); + let api = start_api_with_authentication( + store, + ApiAuthentication { + login: HumanLogin::Password("https://control.example.com".into()), + ..ApiAuthentication::default() + }, + ) + .await; + let client = reqwest::Client::new(); + let administrator = ( + roles.administrator.bearer.as_str(), + roles.administrator.csrf.as_str(), + ); + let url = |location: Uuid| { + format!( + "{}/api/control/locations/{location}/targets/development/acknowledgements", + api.url + ) + }; + let send = |location: Uuid, body: Option| { + let url = url(location); + let client = client.clone(); + async move { + let response = + person_send(&client, (&url, administrator.0, administrator.1), body).await; + let status = response.status().as_u16(); + ( + status, + response.json::().await.unwrap_or(Value::Null), + ) + } + }; + let acknowledge = + json!({"event_id": Uuid::now_v7(), "run_id": run_id, "acknowledged": true}); + let first = send(inventory, Some(acknowledge.clone())).await; + let retried = send(inventory, Some(acknowledge)).await; + let listed = send(inventory, None).await.1; + // The run exists, but not in this location, which the administrator can also see. + let elsewhere = send( + roles.location_id, + Some(json!({"event_id": Uuid::now_v7(), "run_id": run_id, "acknowledged": true})), + ) + .await; + let withdrawn = send( + inventory, + Some(json!({"event_id": Uuid::now_v7(), "run_id": run_id, "acknowledged": false})), + ) + .await; + let after = send(inventory, None).await.1; + assert_eq!( + ( + [first.0, retried.0, elsewhere.0, withdrawn.0], + [ + &first.1["applied"], + &retried.1["applied"], + &withdrawn.1["applied"] + ], + (&listed[0]["run_id"], &listed[0]["acknowledged_by"]), + after, + ), + ( + test_case.expected_statuses, + test_case.expected_applied.map(Value::Bool).each_ref(), + (&json!(run_id), &json!(test_case.expected_acknowledged_by)), + json!([]), + ), + "{}", + test_case.description + ); + stop_api(api).await; + close_test_store(fixture).await; + }); + } +} diff --git a/crates/sqlbuild-orchestration/tests/api/test_compression.rs b/crates/sqlbuild-orchestration/tests/api/test_compression.rs new file mode 100644 index 0000000000..5bdafe2d4d --- /dev/null +++ b/crates/sqlbuild-orchestration/tests/api/test_compression.rs @@ -0,0 +1,109 @@ +//! People's reads are compressed; sign-in, writes, the change feed and the location API are not. + +use reqwest::header; + +use crate::api::models::{ApiAuthentication, HumanLogin}; +use crate::api::tests::helpers::{start_api_with_authentication, stop_api}; +use crate::api::tests::test_types::CompressionTestCase; +use crate::credentials::tests::role_helpers::role_fixture; +use crate::store::main::migrate_store::migrate_store; +use crate::store::tests::store_helpers::{close_test_store, open_test_store, run}; +use crate::store::tests::store_test_types::Backend; + +#[test] +fn given_a_signed_in_browser_when_requesting_then_only_people_reads_are_compressed() { + let test_cases = [ + CompressionTestCase { + description: "a person's read", + method: reqwest::Method::GET, + path: "/api/control/locations/{location}/targets/development/nodes", + accept_encoding: "gzip", + expected_status: 200, + expected_gzip: true, + }, + CompressionTestCase { + description: "a client that does not accept gzip", + method: reqwest::Method::GET, + path: "/api/control/locations/{location}/targets/development/nodes", + accept_encoding: "identity", + expected_status: 200, + expected_gzip: false, + }, + CompressionTestCase { + description: "the sign-in session, which carries the CSRF token", + method: reqwest::Method::GET, + path: "/api/auth/session", + accept_encoding: "gzip", + expected_status: 200, + expected_gzip: false, + }, + CompressionTestCase { + description: "the change feed, whose events must not wait in an encoder", + method: reqwest::Method::GET, + path: "/api/control/locations/{location}/targets/development/changes", + accept_encoding: "gzip", + expected_status: 200, + expected_gzip: false, + }, + CompressionTestCase { + description: "a write", + method: reqwest::Method::POST, + path: "/api/control/locations/{location}/targets/development/plans", + accept_encoding: "gzip", + expected_status: 401, + expected_gzip: false, + }, + CompressionTestCase { + description: "the location API", + method: reqwest::Method::GET, + path: "/api/locations/{location}/targets/development/metadata/watermark", + accept_encoding: "gzip", + expected_status: 401, + expected_gzip: false, + }, + ]; + run(async { + let Some(fixture) = open_test_store(Backend::Sqlite).await else { + return; + }; + let store = &fixture.store; + migrate_store(store).await.expect("migrate"); + let roles = role_fixture(store).await; + let api = start_api_with_authentication( + store, + ApiAuthentication { + login: HumanLogin::Password("https://control.example.com".into()), + ..ApiAuthentication::default() + }, + ) + .await; + let client = reqwest::Client::new(); + let session = format!("__Host-sqb-session={}", roles.administrator.bearer); + for test_case in test_cases { + let path = test_case + .path + .replace("{location}", &roles.location_id.to_string()); + // Only headers are read: the change feed's body stays open. + let response = client + .request(test_case.method, format!("{}{path}", api.url)) + .header(header::HOST, "control.example.com") + .header(header::COOKIE, &session) + .header(header::ACCEPT_ENCODING, test_case.accept_encoding) + .send() + .await + .expect("request"); + let encoding = response.headers().get(header::CONTENT_ENCODING).cloned(); + assert_eq!( + ( + response.status().as_u16(), + encoding.as_ref().is_some_and(|value| value == "gzip") + ), + (test_case.expected_status, test_case.expected_gzip), + "{}: content-encoding {encoding:?}", + test_case.description + ); + } + stop_api(api).await; + close_test_store(fixture).await; + }); +} diff --git a/crates/sqlbuild-orchestration/tests/api/test_contracts.rs b/crates/sqlbuild-orchestration/tests/api/test_contracts.rs index 025a9bb80b..24b70b4ab0 100644 --- a/crates/sqlbuild-orchestration/tests/api/test_contracts.rs +++ b/crates/sqlbuild-orchestration/tests/api/test_contracts.rs @@ -32,6 +32,8 @@ mod statistics_eligibility; mod statistics_helpers; #[path = "statistics/retention/helpers.rs"] mod statistics_retention; +#[path = "test_acknowledgements.rs"] +mod test_acknowledgements; #[path = "test_catalog_reads.rs"] mod test_catalog_reads; #[path = "test_changes.rs"] @@ -42,6 +44,8 @@ mod test_checks; mod test_cli_login; #[path = "test_column_lineage.rs"] mod test_column_lineage; +#[path = "test_compression.rs"] +mod test_compression; #[path = "test_contract.rs"] mod test_contract; #[path = "test_custom_roles.rs"] diff --git a/crates/sqlbuild-orchestration/tests/api/test_denials.rs b/crates/sqlbuild-orchestration/tests/api/test_denials.rs index 5aa903401b..3f16d3c708 100644 --- a/crates/sqlbuild-orchestration/tests/api/test_denials.rs +++ b/crates/sqlbuild-orchestration/tests/api/test_denials.rs @@ -20,8 +20,9 @@ use crate::store::tests::store_helpers::{close_test_store, open_test_store, run} use crate::store::tests::store_test_types::Backend; /// Location-scoped reads and the action each needs. -const READS: [(&str, &str); 13] = [ +const READS: [(&str, &str); 14] = [ ("runs", "view"), + ("acknowledgements", "view"), ("nodes", "view"), ("nodes/orders/runs", "view"), ("jobs", "view"), @@ -239,6 +240,27 @@ fn given_refused_requests_when_people_send_them_then_no_node_is_named_and_unknow ), ], ), + ( + "acknowledge".to_owned(), + "view", + vec![ + ( + read(inventory, "acknowledgements"), + Some(json!({"event_id": Uuid::now_v7(), "run_id": queued.run_id, + "acknowledged": true})), + ), + ( + read(inventory, "acknowledgements"), + Some(json!({"event_id": Uuid::now_v7(), "run_id": unknown_run, + "acknowledged": true})), + ), + ( + read(Uuid::now_v7(), "acknowledgements"), + Some(json!({"event_id": Uuid::now_v7(), "run_id": queued.run_id, + "acknowledged": true})), + ), + ], + ), ( "rerun".to_owned(), "run.rerun", diff --git a/crates/sqlbuild-orchestration/tests/api/test_types.rs b/crates/sqlbuild-orchestration/tests/api/test_types.rs index f5e8e9df1a..e7c650905a 100644 --- a/crates/sqlbuild-orchestration/tests/api/test_types.rs +++ b/crates/sqlbuild-orchestration/tests/api/test_types.rs @@ -184,6 +184,16 @@ pub(crate) struct UiPathTestCase { pub(crate) expected_body: &'static str, } +pub(crate) struct CompressionTestCase { + pub(crate) description: &'static str, + pub(crate) method: reqwest::Method, + /// `{location}` stands for the fixture's location ID. + pub(crate) path: &'static str, + pub(crate) accept_encoding: &'static str, + pub(crate) expected_status: u16, + pub(crate) expected_gzip: bool, +} + pub(crate) struct ServerErrorTestCase { pub(crate) description: &'static str, pub(crate) path: &'static str, @@ -227,3 +237,13 @@ pub(crate) struct UpcomingTestCase { pub(crate) expected_hourly: [&'static str; 5], pub(crate) expected_calendar: [&'static str; 3], } + +pub(crate) struct AcknowledgementTestCase { + pub(crate) description: &'static str, + pub(crate) backend: Backend, + /// Statuses of: acknowledge, the same request retried, a run of another target, withdraw. + pub(crate) expected_statuses: [u16; 4], + /// Whether acknowledging, retrying and withdrawing each changed anything. + pub(crate) expected_applied: [bool; 3], + pub(crate) expected_acknowledged_by: &'static str, +} diff --git a/crates/sqlbuild-python/src/bindings/_helpers/compiled_project/snapshot_execution.rs b/crates/sqlbuild-python/src/bindings/_helpers/compiled_project/snapshot_execution.rs index a5fe394f0d..e117942e2d 100644 --- a/crates/sqlbuild-python/src/bindings/_helpers/compiled_project/snapshot_execution.rs +++ b/crates/sqlbuild-python/src/bindings/_helpers/compiled_project/snapshot_execution.rs @@ -441,27 +441,26 @@ impl WorkerBuild { } } } + let returned = self.returned.load(Ordering::SeqCst); + let build = self.command().is_build(); + let source_needs_no_run = |key: &str| build && key.starts_with("source:"); for key in &self.execution.claim.request.selection { - self.complete( - py, - key, - ( - if cancelled { - NodeOutcome::Cancelled - } else { - NodeOutcome::Skipped - }, - None, - (None, None), - ), - )?; + let outcome = if cancelled { + NodeOutcome::Cancelled + } else if returned && source_needs_no_run(key) { + NodeOutcome::Success + } else { + NodeOutcome::Skipped + }; + self.complete(py, key, (outcome, None, (None, None)))?; } let outcome = { let completed = self.completed.lock().map_err(|_| poisoned())?; let outcomes: Vec = completed.values().copied().collect(); - let good = outcomes - .iter() - .any(|value| matches!(value, NodeOutcome::Success | NodeOutcome::Warn)); + let good = completed.iter().any(|(key, value)| { + matches!(value, NodeOutcome::Success | NodeOutcome::Warn) + && !source_needs_no_run(key) + }); let bad = outcomes.iter().any(|value| { matches!( value, diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/_test_types.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/_test_types.py index 3aef15546a..6771ac246c 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/compile/_test_types.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/_test_types.py @@ -1281,6 +1281,16 @@ class BenchmarkCoreCopyCompileTestCase: expected_compiled_checksum: str +@dataclass(frozen=True) +class BenchmarkDuckDBPortTestCase: + """A script that ports a Snowflake benchmark project to DuckDB, and the port's build summary.""" + + description: str + port_script: Path + expected_final_line_prefix: str + expected_final_line_fragment: str + + @dataclass(frozen=True) class GeneratedBenchmarkCompileTestCase: """A generated benchmark project, its recorded checksum and its cold compile budget.""" diff --git a/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_duckshop_benchmark_compile.py b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_duckshop_benchmark_compile.py index ed6b2309d4..d6ba97e067 100644 --- a/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_duckshop_benchmark_compile.py +++ b/tests/e2e/src/sqlbuild/cli/commands/main/compile/test_duckshop_benchmark_compile.py @@ -13,6 +13,7 @@ from tests.e2e.src.sqlbuild.cli.commands.main.compile._test_types import ( BenchmarkCoreCopyCompileTestCase, + BenchmarkDuckDBPortTestCase, BenchmarkProjectCompileTestCase, BenchmarkProjectTestPlanTestCase, ) @@ -146,3 +147,47 @@ def test_given_benchmark_core_copy_when_compiling_then_leaves_out_the_excluded_a ) assert checksum.returncode == 0, checksum.stdout + checksum.stderr assert checksum.stdout.strip() == test_case.expected_compiled_checksum + + +@pytest.mark.parametrize( + "test_case", + [ + BenchmarkDuckDBPortTestCase( + description="duckshop_duckdb", + port_script=REPO_ROOT / "benchmarks" / "duckshop" / "scripts" / "duckdb_project.py", + expected_final_line_prefix="\u2713 Completed", + expected_final_line_fragment=" FAIL=0 ", + ) + ], + ids=lambda case: case.description, +) +def test_given_duckdb_port_when_building_then_every_model_loads_builds_and_passes( + test_case: BenchmarkDuckDBPortTestCase, tmp_path: Path +) -> None: + """The DuckDB copy is the demonstration project, so it must compile and build end to end.""" + project_dir: Path = tmp_path / test_case.description + ported: subprocess.CompletedProcess[str] = subprocess.run( + ["uv", "run", "-q", "--script", str(test_case.port_script), str(project_dir)], + capture_output=True, + text=True, + check=False, + ) + assert ported.returncode == 0, ported.stdout + ported.stderr + + compiled: subprocess.CompletedProcess[str] = run_sqb( + command=("--no-color", "compile", "--json"), + project_dir=project_dir, + ) + assert compiled.returncode == 0, compiled.stdout + compiled.stderr + report: dict[str, Any] = json.loads(compiled.stdout) + assert report["summary"]["errors"] == 0, report["diagnostics"] + assert report["summary"]["models"] > 0 + + built: subprocess.CompletedProcess[str] = run_sqb( + command=("--no-color", "build"), + project_dir=project_dir, + ) + assert built.returncode == 0, built.stdout + built.stderr + final_line: str = built.stdout.strip().splitlines()[-1] + assert final_line.startswith(test_case.expected_final_line_prefix), built.stdout[-4000:] + assert test_case.expected_final_line_fragment in final_line, built.stdout[-4000:] diff --git a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/_test_types.py b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/_test_types.py index 0692f9df09..e542ca01a9 100644 --- a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/_test_types.py +++ b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/_test_types.py @@ -105,12 +105,15 @@ class WorkerExecutionTestCase: expected_status: str expected_rows: list[tuple[int]] expected_node_count: int = 1 + # Node counts by status when the nodes don't all share the run's status. + expected_node_counts: dict[str, int] | None = None expected_origin: Literal["schedule", "sensor"] = "schedule" files: dict[str, str] = field(default_factory=dict) files_after_publication: dict[str, str] = field(default_factory=dict) store_setup: str = "" concurrency_limits: dict[str, int] = field(default_factory=dict) launcher: Literal["native", "uncooperative", "docker"] = "native" + warehouse_setup: Callable[[Path], None] = lambda _: None @dataclass(frozen=True) diff --git a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/helpers.py b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/helpers.py index 83d2828953..dff015df40 100644 --- a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/helpers.py +++ b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/helpers.py @@ -644,6 +644,7 @@ def scheduled_worker( } | test_case.files, ) + test_case.warehouse_setup(project / "warehouse.duckdb") executable: str = str(Path(sys.executable).with_name("sqb")) launchers: dict[str, Callable[[Path, str], AbstractContextManager[WorkerDeployment]]] = { "native": native_launcher, @@ -726,9 +727,9 @@ def worker_result(worker: ScheduledWorker, test_case: WorkerExecutionTestCase) - """Run/node outcomes, Recall, claims and the physical warehouse agree.""" api, project, control_store, process, log, _ = worker run: dict[str, Any] = wait_for_run(api, process, log) - assert run["node_counts_json"] == {test_case.expected_status: test_case.expected_node_count}, ( - log.read_text() - ) + assert run["node_counts_json"] == ( + test_case.expected_node_counts or {test_case.expected_status: test_case.expected_node_count} + ), log.read_text() assert run["job_key"] == "annual_orders" runs_url: str = f"{api.url}/api/locations/{api.location_id}/targets/development/runs" headers: dict[str, str] = {"Authorization": f"Bearer {api.bearer}"} diff --git a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_worker.py b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_worker.py index 0fb2cbabee..fc0a001287 100644 --- a/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_worker.py +++ b/tests/integration/src/sqlbuild/compiler/compiled_project/publication/test_worker.py @@ -428,6 +428,49 @@ def test_given_a_retained_baseline_when_starting_a_worker_then_verification_and_ ), }, ), + WorkerExecutionTestCase( + "a selected upstream source succeeds once the build returns", + 'SELECT order_id AS amount FROM __source("raw_orders")', + "SELECT amount FROM orders", + "success", + [(23,)], + expected_node_count=2, + files={ + "sources/raw_orders.yml": Path( + "benchmarks/duckshop/edge_cases/orchestration/source_orders.yml" + ).read_text(), + "orchestration/orders.py": Path( + "benchmarks/duckshop/edge_cases/orchestration/upstream_source_orders.py" + ).read_text(), + }, + warehouse_setup=lambda path: ( + duckdb.connect(str(path)) + .execute("CREATE TABLE raw_orders AS SELECT CAST(23 AS BIGINT) AS order_id") + .close() + ), + ), + WorkerExecutionTestCase( + "a failed build with a selected source fails rather than partly succeeds", + "WITH bad(value) AS (VALUES ('invalid')) " + """SELECT CAST(value AS BIGINT) + order_id AS amount FROM __source("raw_orders") CROSS JOIN bad""", + "SELECT count(*) FROM duckdb_tables() WHERE table_name = 'orders'", + "failed", + [(0,)], + expected_node_counts={"success": 1, "failed": 1}, + files={ + "sources/raw_orders.yml": Path( + "benchmarks/duckshop/edge_cases/orchestration/source_orders.yml" + ).read_text(), + "orchestration/orders.py": Path( + "benchmarks/duckshop/edge_cases/orchestration/upstream_source_orders.py" + ).read_text(), + }, + warehouse_setup=lambda path: ( + duckdb.connect(str(path)) + .execute("CREATE TABLE raw_orders AS SELECT CAST(23 AS BIGINT) AS order_id") + .close() + ), + ), WorkerExecutionTestCase( "warehouse error retains failed node evidence", "WITH source(value) AS (VALUES ('invalid')) SELECT CAST(value AS BIGINT) AS amount FROM source", diff --git a/ui/src/app.html b/ui/src/app.html index ed5c328f3d..97dffebe68 100644 --- a/ui/src/app.html +++ b/ui/src/app.html @@ -7,7 +7,7 @@ %sveltekit.head% - +
%sveltekit.body%
diff --git a/ui/src/lib/api/client.ts b/ui/src/lib/api/client.ts index a7df56b9e7..3bb3b5a18a 100644 --- a/ui/src/lib/api/client.ts +++ b/ui/src/lib/api/client.ts @@ -102,6 +102,8 @@ export interface ApiClient extends PlatformApi { /** Queued runs in claim order, with why each one waits. */ listQueue(scope: Scope): Promise; timeline(scope: Scope, window: { since: string; until: string }): Promise; + /** Marks a run handled for everyone who can see it, or withdraws the mark; recorded in the audit log. */ + acknowledgeRun(scope: Scope, runId: string, acknowledged: boolean): Promise; /** Writes a `cancel_requested` fact; the scheduler then signals the worker. */ cancelRun(scope: Scope, runId: string): Promise; /** `nodes` is required for `selected` and ignored otherwise. */ diff --git a/ui/src/lib/api/http-client.ts b/ui/src/lib/api/http-client.ts index 8e82b6ad32..9f3fa8d2de 100644 --- a/ui/src/lib/api/http-client.ts +++ b/ui/src/lib/api/http-client.ts @@ -19,7 +19,7 @@ import { RUN_WINDOW, groupRunNodes, parseRunCursor, planResultOf, queueEntriesOf import type { ChangeSubscription, ChangedChannel, ChangedSignal, ConnectionState } from './signals'; import type { AuthMode, Bootstrap, Capability, Location, NavCounts, PlatformEvent, Scope, WarehouseSync } from './types'; import type { Account, AuditEntry, AuditQuery, ConnectionTest, LocationDetail, ServerSettings, WarehouseConnection, LocationSummary, PlatformEventQuery, PlatformEventRecord, PlatformHealth, MaintenanceHistory, RoleAssignment, RoleAssignmentRequest, RoleDefinition, RoleDetail } from './types/platform'; -import type { CatalogNode, CatalogNodeDetail, Check, CheckDetail, CheckEvidence, CheckSql, Overview, SnapshotChange, ColumnLineage } from './types/build'; +import type { Acknowledgement, CatalogNode, CatalogNodeDetail, Check, CheckDetail, CheckEvidence, CheckSql, Overview, SnapshotChange, ColumnLineage } from './types/build'; import type { AdhocRunCheck, AdhocRunRequest, BulkRunResult, PlanResult, QueueEntry, Timeline, Job, JobDetail, JobLaunchOptions, RerunMode, SnapshotNode, Run, RunDetail, RunListQuery, RunRowPage, RunRowQuery, RunTabCounts, Trigger, TriggerDetail, TriggerState, RunEventPage, RunNodeError, RunOutputLine, RunStatement, StatementSql } from './types/operate'; type Wire = components['schemas']; @@ -43,6 +43,7 @@ const SUPPORTED: ReadonlySet = new Set([ 'getStatementSql', 'listRunOutput', 'getRunNodeError', + 'acknowledgeRun', 'cancelRun', 'cancelMatchingRuns', 'rerunRun', @@ -423,6 +424,15 @@ export function createHttpClient({ baseUrl = '', onUnauthorized }: HttpClientOpt if (!response.ok) throw new ApiError(errorText(error, response), response.status); return requireRun(scope, body.run_id); }, + async acknowledgeRun(scope: Scope, runId: string, acknowledged: boolean): Promise { + const { locationId, target } = parseLocationKey(scope.locationId); + const { error, response } = await http.POST('/api/control/locations/{location_id}/targets/{target}/acknowledgements', { + body: { event_id: crypto.randomUUID(), run_id: runId, acknowledged }, + params: { path: { location_id: locationId, target }, header: { 'x-sqb-csrf': csrfToken(document.cookie) } } + }); + if (response.status === 401) onUnauthorized(); + if (!response.ok) throw new ApiError(errorText(error, response), response.status); + }, async rerunRun(scope: Scope, runId: string, mode: Exclude): Promise { const source = await requireRun(scope, runId); const body = { run_id: crypto.randomUUID(), source_run_id: runId, from_failure: mode === 'from_failure', priority: source.priority, reason: 'Rerun from the web UI' }; @@ -689,11 +699,14 @@ export function createHttpClient({ baseUrl = '', onUnauthorized }: HttpClientOpt sqlLocations.has(scope.locationId) && !sqlRefused.has(`${scope.locationId}|${node.key}`) ? http.GET('/api/control/locations/{location_id}/targets/{target}/nodes/{node_key}/sql', { params: { path: { ...path, node_key: node.key } } }) : null, - checkResults(scope).catch(() => []), + keptRead(`checks ${scope.locationId}`, () => checkResults(scope)).catch(() => []), node.kind === 'source' ? null : http.GET('/api/control/locations/{location_id}/targets/{target}/nodes/{node_key}/runs', { params: { path: { ...path, node_key: node.key }, query: { limit: 100 } } }), - Promise.all([listedJobs(scope), listedTriggers(scope)]).then(([j, t]) => nodeJobsOf(node.key, j, t)).catch(() => null), + // Read afresh, not through a listing already in flight, which may predate the change that just cleared these. + Promise.all([keptRead(`jobs ${scope.locationId}`, () => fetchJobs(scope)), keptRead(`triggers ${scope.locationId}`, () => fetchTriggers(scope))]) + .then(([j, t]) => nodeJobsOf(node.key, j, t)) + .catch(() => null), node.relation ? relationStatistics(scope, node.relation) : [], node.materialization?.includes('microbatch') ? http.GET('/api/control/locations/{location_id}/targets/{target}/microbatches/{model}', { params: { path: { ...path, model: shortName(node.key) } } }) @@ -716,7 +729,8 @@ export function createHttpClient({ baseUrl = '', onUnauthorized }: HttpClientOpt implemented.listChecks(scope), implemented.listCatalogNodes(scope), implemented.listSnapshotChanges(scope, { limit: 100 }), - implemented.listPlatformEvents() + implemented.listPlatformEvents(), + acknowledgementsOf(scope) ]); // Every listing needs view on all of the target, so a partial viewer gets no counts at all. const read = await reads.catch((error: unknown) => { @@ -724,8 +738,8 @@ export function createHttpClient({ baseUrl = '', onUnauthorized }: HttpClientOpt throw error; }); if (!read) return null; - const [runs, checks, nodes, changes, events] = read; - return overviewOf({ runs, checks, nodes, changes, events, now: Date.now() }); + const [runs, checks, nodes, changes, events, acknowledgements] = read; + return overviewOf({ runs, checks, nodes, changes, events, acknowledgements, now: Date.now() }); }, listChecks: async (scope: Scope): Promise => (await checkResults(scope)).map(toCheck), getCheck: async (scope: Scope, id: string): Promise => checkDetailOf(await checkResults(scope, id)), @@ -800,6 +814,21 @@ export function createHttpClient({ baseUrl = '', onUnauthorized }: HttpClientOpt inFlight.clear(); olderWindows.clear(); } + /** + * Project-wide lists a node's detail needs only a slice of, kept between node clicks until the next + * change signal, and at most a minute in case a signal was missed while the feed reconnected. + */ + const kept = new Map }>(); + const KEEP_MS = 60_000; + function keptRead(key: string, load: () => Promise): Promise { + const held = kept.get(key); + if (held && Date.now() - held.at < KEEP_MS) return held.read as Promise; + const read = load(); + kept.set(key, { at: Date.now(), read }); + read.catch(() => kept.get(key)?.read === read && kept.delete(key)); + return read; + } + function shared(key: string, load: () => Promise): Promise { const pending = inFlight.get(key) as Promise | undefined; if (pending) return pending; @@ -945,6 +974,16 @@ export function createHttpClient({ baseUrl = '', onUnauthorized }: HttpClientOpt } const recentRuns = async (scope: Scope) => toRuns(await listedRuns(scope), scope.locationId); + /** Problem runs marked handled in the last week, by run ID. */ + async function acknowledgementsOf(scope: Scope): Promise> { + const { locationId, target } = parseLocationKey(scope.locationId); + const { data, error, response } = await http.GET('/api/control/locations/{location_id}/targets/{target}/acknowledgements', { + params: { path: { location_id: locationId, target } } + }); + if (response.status === 401) onUnauthorized(); + if (!data) throw new ApiError(errorText(error, response), response.status); + return new Map(data.map((a) => [a.run_id, { by: a.acknowledged_by, at: a.acknowledged_at }])); + } /** Triggers with their latest tick and the newest listed run each started. */ async function triggersOf(scope: Scope): Promise { @@ -1068,6 +1107,7 @@ export function createHttpClient({ baseUrl = '', onUnauthorized }: HttpClientOpt return data; } + implemented.changes.subscribe(() => kept.clear()); return new Proxy(implemented, { get(target, property, receiver) { if (property in target) return Reflect.get(target, property, receiver); diff --git a/ui/src/lib/api/http-overview.test.ts b/ui/src/lib/api/http-overview.test.ts index ec1b730f72..2c45f9624e 100644 --- a/ui/src/lib/api/http-overview.test.ts +++ b/ui/src/lib/api/http-overview.test.ts @@ -5,8 +5,8 @@ import type { CatalogNode, Check, NodeChange, PlatformEvent, Run, SnapshotChange const now = Date.parse('2026-11-09T12:00:00Z'); const hoursAgo = (h: number) => new Date(now - h * 3_600_000).toISOString(); -function run(id: string, status: Run['status'], queuedHoursAgo: number, endedHoursAgo: number | null, parentRunId: string | null = null): Run { - return { id, status, parentRunId, jobKey: 'orders', kind: 'job', statusReason: null, queuedAt: hoursAgo(queuedHoursAgo), startedAt: hoursAgo(queuedHoursAgo), endedAt: endedHoursAgo === null ? null : hoursAgo(endedHoursAgo) } as Run; +function run(id: string, status: Run['status'], queuedHoursAgo: number, endedHoursAgo: number | null, jobKey: string | null = 'orders', parentRunId: string | null = null, narrowing: Run['narrowing'] = null): Run { + return { id, status, parentRunId, jobKey, narrowing, window: null, kind: jobKey ? 'job' : 'adhoc', statusReason: null, queuedAt: hoursAgo(queuedHoursAgo), startedAt: hoursAgo(queuedHoursAgo), endedAt: endedHoursAgo === null ? null : hoursAgo(endedHoursAgo) } as Run; } const node = (key: string, kind: CatalogNode['kind']) => ({ key, kind }) as CatalogNode; @@ -19,15 +19,22 @@ describe('overview from live reads', () => { // Newest queued first, as the runs listing returns them. const runs = [ run('active', 'running', 1, null), - run('child-failed', 'failed', 2, 2, 'group'), - run('failed', 'failed', 3, 2.5), - run('presumed', 'presumed_failed', 4, 3.5), - run('partial', 'partial', 5, 4.5), - run('failed-4th', 'failed', 6, 5.5), - run('ok', 'success', 7, 0.5), - run('stuck', 'unresponsive', 8, null), - run('warn', 'warn', 23.5, 23), - run('yesterday-failed', 'failed', 25, 24.5) + run('child-failed', 'failed', 2, 2, 'orders', 'group'), + run('adhoc-new', 'failed', 2.2, 2.1, null), + run('orders-new', 'failed', 3, 2.5), + run('fulfilment-failed', 'failed', 4, 3.5, 'fulfilment'), + run('catalogue-partial', 'partial', 5, 4.5, 'catalogue'), + run('orders-old', 'failed', 6, 5.5), + // Neither a cancelled run nor a run of part of a job shows the job has recovered. + run('catalogue-cancelled', 'cancelled', 4.2, 1, 'catalogue'), + run('orders-part-ok', 'success', 2.4, 0.8, 'orders', null, { select: ['stg_orders'], exclude: [] }), + // Queued before the fulfilment failure but ended after it, so fulfilment has recovered. + run('fulfilment-ok', 'success', 7, 0.5, 'fulfilment'), + run('presumed', 'presumed_failed', 8, 1.5, 'inventory'), + run('stuck', 'unresponsive', 8, null, 'returns'), + run('adhoc-old', 'failed', 9, 8, null), + run('warn', 'warn', 23.5, 23, 'support'), + run('yesterday-failed', 'failed', 25, 24.5, 'archive') ]; const overview = overviewOf({ runs, @@ -35,19 +42,23 @@ describe('overview from live reads', () => { nodes: [node('model:orders', 'table'), node('model:totals', 'view'), node('model:events', 'incremental'), node('source:raw', 'source'), node('seed:regions', 'seed')], changes: [change(1, [['added', 'model:orders'], ['added', 'seed:regions'], ['removed', 'model:old']]), change(24 * 8, [['added', 'model:totals']])], events: [event('scheduler', true), event('cleared', false)], + acknowledgements: new Map([['orders-new', { by: 'ada', at: hoursAgo(1) }]]), now }); it('counts top-level runs queued in the last 24 hours', () => { - expect(overview.metrics.runs24h).toEqual({ total: 8, success: 1, warn: 1, partial: 1, failed: 3, active: 2 }); + expect(overview.metrics.runs24h).toEqual({ total: 13, success: 2, warn: 1, partial: 1, failed: 6, active: 2 }); }); - it('lists the three newest failed runs that ended today, then unresponsive runs and active platform events', () => { - expect(overview.attention.map((i) => (i.kind === 'platform_event' ? `event:${i.event.id}` : 'run' in i ? `${i.kind}:${i.run.id}` : i.kind))).toEqual([ - 'run_problem:failed', - 'run_problem:presumed', - 'run_problem:partial', - 'run_unresponsive:stuck', + it('lists each job’s latest failure once, and each ad-hoc failure, newest first, then unresponsive runs and active platform events', () => { + // Acknowledging the orders failure keeps its place and never brings the older orders failure back. + expect(overview.attention.map((i) => (i.kind === 'platform_event' ? `event:${i.event.id}` : 'run' in i ? `${i.kind}:${i.run.id}:${i.acknowledged?.by ?? '-'}` : i.kind))).toEqual([ + 'run_problem:presumed:-', + 'run_problem:adhoc-new:-', + 'run_problem:orders-new:ada', + 'run_problem:catalogue-partial:-', + 'run_problem:adhoc-old:-', + 'run_unresponsive:stuck:-', 'event:scheduler' ]); }); @@ -61,7 +72,7 @@ describe('overview from live reads', () => { }); it('names the run that ended last, not the newest queued', () => { - expect(overview.latestRun?.id).toBe('ok'); + expect(overview.latestRun?.id).toBe('fulfilment-ok'); }); it('leaves parts without a server read empty instead of zero', () => { diff --git a/ui/src/lib/api/http-overview.ts b/ui/src/lib/api/http-overview.ts index 9f2c47ce48..de5a7625b2 100644 --- a/ui/src/lib/api/http-overview.ts +++ b/ui/src/lib/api/http-overview.ts @@ -1,12 +1,10 @@ import { checkCountsOf } from './http-catalog'; -import type { AttentionItem, AttentionRun, CatalogNode, Check, Overview, PlatformEvent, Run, SnapshotChange } from './types'; +import type { Acknowledgement, AttentionItem, AttentionRun, CatalogNode, Check, Overview, PlatformEvent, Run, SnapshotChange } from './types'; const DAY = 24 * 60 * 60 * 1000; const MODEL_KINDS = new Set(['view', 'table', 'incremental']); const PROBLEM_STATUSES = new Set(['failed', 'partial', 'presumed_failed']); const ACTIVE_STATUSES = new Set(['queued', 'launched', 'running', 'unresponsive']); -/** The failed runs the health banner lists; the runs page has the rest. */ -const PROBLEM_RUNS = 3; const RECENT_CHANGES = 4; export type OverviewInputs = { @@ -17,25 +15,46 @@ export type OverviewInputs = { /** Publications, newest first. */ changes: SnapshotChange[]; events: PlatformEvent[]; + /** Problem runs marked handled, by run ID. */ + acknowledgements: Map; now: number; }; -const attentionRun = ({ id, jobKey, kind, status, statusReason, startedAt, endedAt }: Run): AttentionRun => ({ id, jobKey, kind, status, statusReason, startedAt, endedAt }); +/** + * One problem per job, newest first: the job's latest run that ended today, when that run did not succeed. + * Only a later failure or a good run of the whole job replaces it, so a cancelled run or a run of part of the + * job never hides a failure, and acknowledging the latest failure never brings an older one back. + * Each ad-hoc run is its own item. + */ +export function problemRunsOf(topLevel: Run[], now: number): Run[] { + const latest = new Map(); + for (const r of topLevel) { + if (!r.endedAt || Date.parse(r.endedAt) < now - DAY) continue; + const settles = PROBLEM_STATUSES.has(r.status) || ((r.status === 'success' || r.status === 'warn') && !r.narrowing && !r.window); + if (!settles) continue; + const job = r.jobKey ?? `adhoc:${r.id}`; + const seen = latest.get(job); + if (!seen || Date.parse(r.endedAt) > Date.parse(seen.endedAt!)) latest.set(job, r); + } + return [...latest.values()].filter((r) => PROBLEM_STATUSES.has(r.status)).sort((a, b) => Date.parse(b.endedAt!) - Date.parse(a.endedAt!)); +} + +const attentionRun = ({ id, jobKey, kind, status, statusReason, nodeCounts, startedAt, endedAt }: Run): AttentionRun => ({ id, jobKey, kind, status, statusReason, nodeCounts, startedAt, endedAt }); /** * The overview from reads the person already has. The server has no project-wide build state, cost * or group read yet, so stale counts and group status are null and cost is left out; the page hides them. */ -export function overviewOf({ runs, checks, nodes, changes, events, now }: OverviewInputs): Overview { +export function overviewOf({ runs, checks, nodes, changes, events, acknowledgements, now }: OverviewInputs): Overview { + const acknowledged = (r: Run) => acknowledgements.get(r.id) ?? null; const topLevel = runs.filter((r) => !r.parentRunId); const recent = topLevel.filter((r) => Date.parse(r.queuedAt) >= now - DAY); const count = (statuses: string[]) => recent.filter((r) => statuses.includes(r.status)).length; const attention: AttentionItem[] = [ + ...problemRunsOf(topLevel, now).map((r): AttentionItem => ({ kind: 'run_problem', run: attentionRun(r), acknowledged: acknowledged(r) })), ...topLevel - .filter((r) => r.endedAt && Date.parse(r.endedAt) >= now - DAY && PROBLEM_STATUSES.has(r.status)) - .slice(0, PROBLEM_RUNS) - .map((r): AttentionItem => ({ kind: 'run_problem', run: attentionRun(r) })), - ...topLevel.filter((r) => r.status === 'unresponsive').map((r): AttentionItem => ({ kind: 'run_unresponsive', run: attentionRun(r), lastHeartbeatAt: null })), + .filter((r) => r.status === 'unresponsive') + .map((r): AttentionItem => ({ kind: 'run_unresponsive', run: attentionRun(r), lastHeartbeatAt: null, acknowledged: acknowledged(r) })), ...events.filter((e) => e.active).map((event): AttentionItem => ({ kind: 'platform_event', event })) ]; const thisWeek = changes.filter((c) => Date.parse(c.publishedAt) >= now - 7 * DAY).flatMap((c) => c.nodes); diff --git a/ui/src/lib/api/http-runs.test.ts b/ui/src/lib/api/http-runs.test.ts index fbc616e519..cc1056afdb 100644 --- a/ui/src/lib/api/http-runs.test.ts +++ b/ui/src/lib/api/http-runs.test.ts @@ -288,6 +288,7 @@ describe('planResultOf', () => { it.each([ { description: 'a plan keeps its models and fills missing sections', outcome: { kind: 'plan', document: { selected_count: 1, models: [{ name: 'orders', reason: 'first_run' }] } }, expected: { plan: 'orders', error: null } }, { description: "sqb plan's own error is the result", outcome: { kind: 'plan', document: { error: { code: 'S007', message: "unknown selector name 'ordrs'" } } }, expected: { plan: null, error: "S007: unknown selector name 'ordrs'" } }, + { description: 'the help line follows the error', outcome: { kind: 'plan', document: { error: { code: 'S000', message: 'no cursor interval', help: 'pass --start-cursor-ts' } } }, expected: { plan: null, error: 'S000: no cursor interval\nhelp: pass --start-cursor-ts' } }, { description: 'a superseded plan asks to plan again', outcome: { kind: 'superseded' }, expected: { plan: null, error: 'The project changed on the location while it was planning; plan again' } } ])('$description', ({ outcome, expected }) => { const result = planResultOf(outcome as never); diff --git a/ui/src/lib/api/http-runs.ts b/ui/src/lib/api/http-runs.ts index c38991ec16..5930b21bc8 100644 --- a/ui/src/lib/api/http-runs.ts +++ b/ui/src/lib/api/http-runs.ts @@ -15,8 +15,12 @@ const PERSON = 'principal:'; export function planResultOf(outcome: Wire['LocationWorkOutcome']): PlanResult { if (outcome.kind === 'superseded') return { plan: null, error: 'The project changed on the location while it was planning; plan again' }; if (outcome.kind !== 'plan') return { plan: null, error: 'The location answered with something other than a plan' }; - const document = outcome.document as Partial & { error?: { code?: string; message?: string } }; - if (document.error) return { plan: null, error: [document.error.code, document.error.message].filter(Boolean).join(': ') }; + const document = outcome.document as Partial & { error?: { code?: string; message?: string; help?: string | null } }; + // The help line carries the fix, as `sqb plan` prints it under the error. + if (document.error) { + const { code, message, help } = document.error; + return { plan: null, error: [code, message].filter(Boolean).join(': ') + (help ? `\nhelp: ${help}` : '') }; + } const list = (value: T[] | undefined): T[] => value ?? []; return { plan: { diff --git a/ui/src/lib/api/mock/build.ts b/ui/src/lib/api/mock/build.ts index 099dd336bc..53e8714d98 100644 --- a/ui/src/lib/api/mock/build.ts +++ b/ui/src/lib/api/mock/build.ts @@ -4,6 +4,7 @@ // (synced fingerprints), compared against the snapshot change feed. import { ApiError } from '../client'; +import { problemRunsOf } from '../http-overview'; import type { AttentionItem, BuiltDefinition, @@ -235,12 +236,13 @@ export class MockBuild { const staleSources = nodes.filter((n) => n.sourceFreshness && n.sourceFreshness.status !== 'pass'); const attention: AttentionItem[] = []; - const pick = (r: Run) => ({ id: r.id, jobKey: r.jobKey, kind: r.kind, status: r.status, statusReason: r.statusReason, startedAt: r.startedAt, endedAt: r.endedAt }); - for (const r of runs.filter((r) => r.endedAt && Date.parse(r.endedAt) >= dayAgo && ['failed', 'partial', 'presumed_failed'].includes(r.status)).slice(0, 3)) { - attention.push({ kind: 'run_problem', run: pick(r) }); + const pick = (r: Run) => ({ id: r.id, jobKey: r.jobKey, kind: r.kind, status: r.status, statusReason: r.statusReason, nodeCounts: r.nodeCounts, startedAt: r.startedAt, endedAt: r.endedAt }); + const acknowledged = (r: Run) => this.#ctx.operate.acknowledgement(r.id); + for (const r of problemRunsOf(runs, now)) { + attention.push({ kind: 'run_problem', run: pick(r), acknowledged: acknowledged(r) }); } for (const r of runs.filter((r) => r.status === 'unresponsive')) { - attention.push({ kind: 'run_unresponsive', run: pick(r), lastHeartbeatAt: this.#ctx.operate.lastHeartbeat(r.id) }); + attention.push({ kind: 'run_unresponsive', run: pick(r), lastHeartbeatAt: this.#ctx.operate.lastHeartbeat(r.id), acknowledged: acknowledged(r) }); } if (changed.length || behind.length) attention.push({ kind: 'stale_models', changed, behind }); if (staleSources.length) attention.push({ kind: 'stale_sources', sources: staleSources.map((n) => ({ key: n.key, ...n.sourceFreshness! })) }); diff --git a/ui/src/lib/api/mock/mock-client.ts b/ui/src/lib/api/mock/mock-client.ts index 6aef706b00..b2f1f64d8f 100644 --- a/ui/src/lib/api/mock/mock-client.ts +++ b/ui/src/lib/api/mock/mock-client.ts @@ -237,6 +237,11 @@ export class MockApiClient implements ApiClient { return this.#operate.cancelRun(scope, runId); } + async acknowledgeRun(scope: Scope, runId: string, acknowledged: boolean): Promise { + this.#location(scope); + this.#operate.acknowledgeRun(runId, acknowledged, 'you'); + } + async rerunRun(scope: Scope, runId: string, mode: Exclude): Promise { this.#location(scope); return this.#operate.rerunRun(scope, runId, mode); diff --git a/ui/src/lib/api/mock/operate-store.ts b/ui/src/lib/api/mock/operate-store.ts index 8afd687b26..57d5ff7ead 100644 --- a/ui/src/lib/api/mock/operate-store.ts +++ b/ui/src/lib/api/mock/operate-store.ts @@ -7,6 +7,7 @@ import { flagSpec, VERBS } from '$lib/run/command-table'; import { ApiError } from '../client'; import type { LocalChangeBus } from '../signals'; import type { + Acknowledgement, AdhocPermission, AdhocRunCheck, PlanResult, @@ -652,6 +653,16 @@ export class MockOperate { })); } + /** Problem runs marked handled, by run ID. */ + #acknowledgements = new Map(); + acknowledgement(runId: string): Acknowledgement | null { + return this.#acknowledgements.get(runId) ?? null; + } + acknowledgeRun(runId: string, acknowledged: boolean, by: string): void { + if (acknowledged) this.#acknowledgements.set(runId, { by, at: new Date().toISOString() }); + else this.#acknowledgements.delete(runId); + } + /** Liveness of an active run, for the overview's unresponsive item. */ lastHeartbeat(runId: string): string | null { const rec = this.#byId(runId); diff --git a/ui/src/lib/api/types/build.ts b/ui/src/lib/api/types/build.ts index e86a845cf8..9536d87308 100644 --- a/ui/src/lib/api/types/build.ts +++ b/ui/src/lib/api/types/build.ts @@ -340,11 +340,13 @@ export type SnapshotChange = { // ---- overview ---- -export type AttentionRun = Pick; +export type AttentionRun = Pick; +/** Who marked a problem run handled; everyone who can see the run sees it. */ +export type Acknowledgement = { by: string; at: Iso }; export type AttentionItem = - | { kind: 'run_problem'; run: AttentionRun } - | { kind: 'run_unresponsive'; run: AttentionRun; lastHeartbeatAt: Iso | null } + | { kind: 'run_problem'; run: AttentionRun; acknowledged: Acknowledgement | null } + | { kind: 'run_unresponsive'; run: AttentionRun; lastHeartbeatAt: Iso | null; acknowledged: Acknowledgement | null } | { kind: 'stale_models'; changed: string[]; behind: string[] } | { kind: 'stale_sources'; sources: ({ key: string } & SourceFreshness)[] } | { kind: 'platform_event'; event: PlatformEvent } diff --git a/ui/src/lib/api/types/operate.ts b/ui/src/lib/api/types/operate.ts index 62970a7541..72bc25ccd7 100644 --- a/ui/src/lib/api/types/operate.ts +++ b/ui/src/lib/api/types/operate.ts @@ -505,6 +505,13 @@ export type PlanModel = { backfill: { action: string; duration: string | null }; qualified_name?: string; changed_functions?: string[]; + cursor_type?: string; + cursor?: { requested_bounds: { start: string | null; end: string | null }; resolved_bounds: { start: string | null; end: string | null } | null } | null; + /** Unified diff lines of the query since the model's last build, as `sqb plan` prints them. */ + query_diff?: string[]; + /** Unified diff lines of the model's config since its last build. */ + config_diff?: string[]; + schema_findings?: { kind: string; column: string; expected_type: string | null; actual_type: string | null }[]; }; export type PlanWarning = { severity: string; message: string; model_name?: string; code?: string }; diff --git a/ui/src/lib/build-area/attention.ts b/ui/src/lib/build-area/attention.ts index c9f35164d9..e18d05838d 100644 --- a/ui/src/lib/build-area/attention.ts +++ b/ui/src/lib/build-area/attention.ts @@ -1,7 +1,7 @@ // Presentation of the overview's attention items. Pure; no side effects. import type { AttentionItem } from '$lib/api'; -import { describeWait } from '$lib/operate/labels'; +import { describeWait, outcomeCounts } from '$lib/operate/labels'; import { eventKindLabel, problemHref } from '$lib/platform/labels'; import { formatAgo } from '$lib/format/time'; import { runStatusLabel } from '$lib/status'; @@ -28,7 +28,8 @@ export function attentionView(item: AttentionItem, now: number): AttentionView { key: `run:${item.run.id}`, tone: 'error', title: `${item.run.jobKey ?? 'Ad-hoc run'} ${runStatusLabel[item.run.status].toLowerCase()}`, - detail: item.run.statusReason ?? item.run.id, + // The node counts say what went wrong; the worker's reason only says how it stopped. + detail: outcomeCounts(item.run.nodeCounts) || item.run.statusReason || item.run.id, since: formatAgo(item.run.endedAt, now), href: `/runs/${item.run.id}`, action: 'View run' diff --git a/ui/src/lib/components/admin/audit-mini.svelte b/ui/src/lib/components/admin/audit-mini.svelte index 7f09ab4ab3..61615dfb02 100644 --- a/ui/src/lib/components/admin/audit-mini.svelte +++ b/ui/src/lib/components/admin/audit-mini.svelte @@ -1,4 +1,5 @@ - - {#if resource.error} +{#if resource.error} + Couldn't load: {resource.error} - {:else if resource.loading && resource.data === undefined} - {resource.slow ? 'Loading…' : ''} - {:else} + +{:else if resource.loading && resource.data === undefined} + + {#each widths as width, row (row)} + + + + {/each} +{:else} + {empty} - {/if} - + +{/if} diff --git a/ui/src/lib/components/platform/location-tokens.svelte b/ui/src/lib/components/platform/location-tokens.svelte index 5a5bc0ca71..3e0c64295a 100644 --- a/ui/src/lib/components/platform/location-tokens.svelte +++ b/ui/src/lib/components/platform/location-tokens.svelte @@ -1,4 +1,5 @@ -{text} + +{sync ? text : 'Warehouse last synced just now'} diff --git a/ui/src/lib/run/plan-report.svelte b/ui/src/lib/run/plan-report.svelte new file mode 100644 index 0000000000..044b2a33fe --- /dev/null +++ b/ui/src/lib/run/plan-report.svelte @@ -0,0 +1,180 @@ + + +
+ {#each sections as s (s.title)} + {@const shown = expanded[s.title] ? s.entries : s.entries.slice(0, limit)} +
+

{s.title} ({s.entries.length})

+ {#each shown as e (`${e.name}:${e.value}`)} + {@const link = href(e, s.title)} + {@const hasDiff = !!(e.queryDiff?.length || e.configDiff?.length || e.schema?.length)} +
+
+ {#if link}{e.name}{:else}{e.name}{/if} + {e.value} +
+ {#each e.detail as d, i (i)}
{d}
{/each} + {#if hasDiff} + {#if e.schema?.length} +
schema diff:
+ 
{#each e.schema as f (f.text)}{f.sign} {f.text}{'\n'}{/each}
+ {/if} + {#if e.configDiff?.length} +
config diff:
+ 
{#each e.configDiff as l, i (i)}{l}{'\n'}{/each}
+ {/if} + {#if e.queryDiff?.length} +
query diff:
+ 
{#each e.queryDiff as l, i (i)}{l}{'\n'}{/each}
+ {/if} + {/if} +
+ {/each} + {#if s.entries.length > shown.length} + + {/if} +
+ {/each} + {#if plan.warnings.length} +
+

Warnings ({plan.warnings.length})

+ {#each plan.warnings as w, i (i)} +
+ {#if w.model_name}{w.model_name}{/if} + {#if w.code}{w.code} {/if}{w.message} +
+ {/each} +
+ {/if} +
+ + diff --git a/ui/src/lib/run/plan-text.test.ts b/ui/src/lib/run/plan-text.test.ts new file mode 100644 index 0000000000..1ec8b747ba --- /dev/null +++ b/ui/src/lib/run/plan-text.test.ts @@ -0,0 +1,65 @@ +import { describe, expect, it } from 'vitest'; +import type { PlanDocument, PlanModel } from '$lib/api'; +import { actionText, planSections, readySummary } from './plan-text'; + +const model = (name: string, over: Partial = {}): PlanModel => ({ + name, + materialization_type: 'table', + action: 'create_table', + reason: 'no_change', + backfill: { action: 'forward', duration: null }, + ...over +}); +const plan = (models: PlanModel[], over: Partial = {}): PlanDocument => ({ + selected_count: models.length, + models, + seeds: [], + source_loads: [], + functions: [], + python_nodes: [], + warnings: [], + migrations: [], + column_migrations: [], + table_type_conversions: [], + retention: [], + ...over +}); + +describe('plan text, as sqb plan prints it', () => { + it('words a changed model by its backfill, as the CLI does', () => { + expect(actionText(model('a', { action: 'create_view', backfill: { action: 'full', duration: null } }))).toBe('recreate view'); + expect(actionText(model('a', { backfill: { action: 'bounded', duration: '7d' } }))).toBe('rebuild last 7d'); + expect(actionText(model('a', { reason: 'query_changed', backfill: { action: 'full', duration: null } }))).toBe('full rebuild'); + expect(actionText(model('a', { reason: 'first_run', backfill: { action: 'full', duration: null } }))).toBe('continue forward'); + }); + + it('puts each change in its own section in reason order, then the routine models, and leaves skipped ones out', () => { + const sections = planSections( + plan([ + model('fresh', { reason: 'first_run' }), + model('orders', { reason: 'query_changed', backfill: { action: 'full', duration: null }, query_diff: ['--- previous', '+++ current', '-SELECT 1', '+SELECT 2'] }), + model('lines', { materialization_type: 'incremental', incremental_strategy: 'merge', cursor_type: 'timestamp', cursor_column: 'placed_at', action: 'incremental_merge', reason: 'normal_incremental' }), + model('customers'), + model('retired', { action: 'skip', reason: 'disabled' }) + ]) + ); + expect(sections.map((s) => [s.title, s.entries.map((e) => e.name)])).toEqual([ + ['Query changed', ['orders']], + ['First run', ['fresh']], + ['Models', ['lines', 'customers']] + ]); + const [changed, , routine] = sections; + expect(changed.entries[0]).toMatchObject({ value: 'full rebuild', detail: ['policy: replay_on_change=full'], queryDiff: ['--- previous', '+++ current', '-SELECT 1', '+SELECT 2'] }); + // Unchanged tables still build: they are listed with their materialization, not as skipped. + expect(routine.entries).toEqual([ + { name: 'lines', value: 'merge (timestamp)', detail: ['cursor: placed_at (timestamp)'] }, + { name: 'customers', value: 'table', detail: [] } + ]); + }); + + it('collapses a plan that is all full refresh into counts by materialization', () => { + const full = plan([model('a', { reason: 'full_refresh' }), model('b', { reason: 'full_refresh' }), model('v', { reason: 'full_refresh', materialization_type: 'view' })]); + expect(planSections(full)).toEqual([{ title: 'Full refresh', changed: true, entries: [{ name: 'table', value: '2 models', detail: [] }, { name: 'view', value: '1 model', detail: [] }] }]); + expect(readySummary(full)).toBe('full refresh, 3 selected'); + }); +}); diff --git a/ui/src/lib/run/plan-text.ts b/ui/src/lib/run/plan-text.ts new file mode 100644 index 0000000000..f5c87c8cfd --- /dev/null +++ b/ui/src/lib/run/plan-text.ts @@ -0,0 +1,143 @@ +// The plan laid out as `sqb plan` prints it (crates/sqlbuild-core/src/cli_text), so the page and the +// terminal read the same: work that runs first, one section per change reason with each model's detail, +// then the routine models, functions, seeds and warnings. + +import type { PlanDocument, PlanModel } from '$lib/api'; + +export type PlanEntry = { + name: string; + /** What happens to it: an action for a changed model, its materialization otherwise. */ + value: string; + /** Detail lines under it, such as its cursor or the cause of the change. */ + detail: string[]; + queryDiff?: string[]; + configDiff?: string[]; + schema?: { sign: '+' | '-' | '~'; text: string }[]; +}; + +export type PlanSection = { title: string; entries: PlanEntry[]; changed: boolean }; + +/** The change reasons that get their own section, in the CLI's order. */ +const REASON_GROUPS: [string, string][] = [ + ['full_refresh', 'Full refresh'], + ['query_changed', 'Query changed'], + ['function_changed', 'Function changed'], + ['config_changed', 'Config changed'], + ['schema_changed', 'Schema changed'], + ['first_run', 'First run'], + ['renamed', 'Renamed'] +]; +const ROUTINE = new Set(['no_change', 'normal_incremental']); + +/** The model's materialization as the CLI labels it, such as `view` or `delete_insert (timestamp)`. */ +export function materializationLabel(m: PlanModel): string { + if (m.materialization_type !== 'incremental') return m.materialization_type; + const strategy = m.incremental_strategy ?? 'incremental'; + return m.cursor_type ? `${strategy} (${m.cursor_type})` : strategy; +} + +/** What a changed model's build does, as the CLI's `action_text` words it. */ +export function actionText(m: PlanModel): string { + if (m.action === 'create_view') return 'recreate view'; + if (m.action === 'custom') return 'run custom materialization'; + if (m.backfill.action === 'bounded' && m.backfill.duration) return `rebuild last ${m.backfill.duration}`; + if (m.backfill.action === 'full' && m.reason !== 'first_run') return 'full rebuild'; + return 'continue forward'; +} + +function cursorLines(m: PlanModel, showRange: boolean): string[] { + const column = m.cursor_column; + if (!column) return []; + const lines = [`cursor: ${column}${m.cursor_type ? ` (${m.cursor_type})` : ''}`]; + const requested = m.cursor?.requested_bounds; + if (requested && (requested.start || requested.end)) lines.push(`requested: ${requested.start ?? 'earliest available'} → ${requested.end ?? 'latest available'}`); + const range = m.cursor?.resolved_bounds ?? m.cursor_bounds; + if (showRange && range && (range.start || range.end)) lines.push(`range: ${range.start ?? '…'} → ${range.end ?? '…'}`); + return lines; +} + +function policyLine(m: PlanModel): string[] { + if (m.backfill.action === 'forward' || !(m.reason === 'query_changed' || m.reason === 'schema_changed')) return []; + const value = m.backfill.action === 'bounded' ? `bounded-${m.backfill.duration ?? 'full'}` : m.backfill.action; + return [`policy: replay_on_change=${value}`]; +} + +const SCHEMA_SIGNS: Record = { + column_added: { sign: '+', label: 'added' }, + column_removed: { sign: '-', label: 'removed' }, + column_type_changed: { sign: '~', label: 'type changed' } +}; + +function changedEntry(m: PlanModel, reason: string): PlanEntry { + if (reason === 'first_run') return { name: m.name, value: materializationLabel(m), detail: cursorLines(m, true) }; + return { + name: m.name, + value: actionText(m), + detail: [ + ...cursorLines(m, reason !== 'full_refresh'), + ...policyLine(m), + ...(m.changed_functions?.length ? [`cause: function ${m.changed_functions.join(', ')} changed`] : []) + ], + queryDiff: m.query_diff, + configDiff: m.config_diff, + schema: m.schema_findings?.map((f) => { + const kind = SCHEMA_SIGNS[f.kind] ?? { sign: '~' as const, label: f.kind }; + const types = f.kind === 'column_type_changed' ? ` ${f.expected_type ?? 'None'} → ${f.actual_type ?? 'None'}` : f.expected_type ? ` ${f.expected_type}` : ''; + return { sign: kind.sign, text: `${f.column}${types} (${kind.label})` }; + }) + }; +} + +/** Whether the plan reads as one full refresh: every model being built is one, as in the CLI. */ +export function isFullRefresh(plan: PlanDocument): boolean { + const active = plan.models.filter((m) => m.action !== 'skip'); + return active.length > 0 && active.every((m) => m.reason === 'full_refresh'); +} + +/** The plan's sections in the order `sqb plan` prints them. Skipped models are left out, as there. */ +export function planSections(plan: PlanDocument): PlanSection[] { + const active = plan.models.filter((m) => m.action !== 'skip'); + const sections: PlanSection[] = []; + const add = (title: string, entries: PlanEntry[], changed = false) => { + if (entries.length) sections.push({ title, entries, changed }); + }; + // Work that runs before any model builds comes first, as in the CLI. + add('Migrations', plan.migrations.map((r) => ({ name: r.model, value: [r.origin_model && `from ${r.origin_model}`, r.decision, r.completed_at ? 'completed' : 'pending'].filter(Boolean).join(' · '), detail: [] })), true); + add('Column migrations', plan.column_migrations.map((r) => ({ name: r.model, value: `${r.origin_column} → ${r.destination_column} · ${r.decision}`, detail: [] })), true); + add('Table type changes', plan.table_type_conversions.map((r) => ({ name: r.model, value: [r.actual_type, r.desired_type].filter(Boolean).join(' → '), detail: [] })), true); + add('Retention', plan.retention.map((r) => ({ name: r.model, value: r.action ?? '', detail: [] })), true); + for (const kind of ['loader', 'source'] as const) { + const loads = plan.source_loads.filter((s) => s.kind === kind); + const label = kind === 'loader' ? 'Loaders' : 'Sources'; + add(`${label} to reload`, loads.filter((s) => s.is_reload).map((s) => ({ name: s.name, value: s.write_strategy ?? '', detail: [] }))); + add(`${label} to load`, loads.filter((s) => !s.is_reload).map((s) => ({ name: s.name, value: s.write_strategy ?? '', detail: s.cursor_column ? [`cursor: ${s.cursor_column}`] : [] }))); + } + const functionEntry = (f: PlanDocument['functions'][number]) => ({ name: f.name, value: f.language, detail: [] }); + add('Changed functions', plan.functions.filter((f) => f.reason && f.reason !== 'no_change').map(functionEntry), true); + if (isFullRefresh(plan)) { + const counts = new Map(); + for (const m of active) counts.set(m.materialization_type, (counts.get(m.materialization_type) ?? 0) + 1); + add('Full refresh', [...counts].map(([kind, n]) => ({ name: kind, value: `${n} ${n === 1 ? 'model' : 'models'}`, detail: [] })), true); + } else { + for (const [reason, title] of REASON_GROUPS) add(title, active.filter((m) => m.reason === reason).map((m) => changedEntry(m, reason)), true); + add('Models', active.filter((m) => ROUTINE.has(m.reason)).map((m) => ({ name: m.name, value: materializationLabel(m), detail: cursorLines(m, true) }))); + } + add('Functions', plan.functions.filter((f) => !f.reason || f.reason === 'no_change').map(functionEntry)); + const seedLabel: Record = { first_run: 'first_run', config_changed: 'seed_changed', no_change: 'current' }; + add('Seeds', plan.seeds.map((s) => ({ name: s.name, value: seedLabel[s.reason] ?? '', detail: [] }))); + add('Python', plan.python_nodes.map((p) => ({ name: p.name, value: `${p.kind} · ${p.identity_status}`, detail: [] }))); + return sections; +} + +/** `N selected`, as the CLI's `Plan ready` line counts it. */ +export function readySummary(plan: PlanDocument): string { + const parts = [isFullRefresh(plan) ? 'full refresh' : '', `${plan.selected_count} selected`]; + const loads = plan.source_loads.length; + if (loads) parts.push(`${loads} source${loads === 1 ? '' : 's'} to load`); + return parts.filter(Boolean).join(', '); +} + +/** The models a plan changes, as catalog keys, for drawing just the part of a large selection that matters. */ +export function changedModelKeys(plan: PlanDocument): string[] { + return plan.models.filter((m) => m.action !== 'skip' && !ROUTINE.has(m.reason)).map((m) => `model:${m.name}`); +} diff --git a/ui/src/lib/run/range-calendar.svelte b/ui/src/lib/run/range-calendar.svelte new file mode 100644 index 0000000000..8c9b4c0321 --- /dev/null +++ b/ui/src/lib/run/range-calendar.svelte @@ -0,0 +1,166 @@ + + + + + + { + if (!v.start || !v.end) return; + const [first, last] = v.start.compare(v.end) <= 0 ? [v.start, v.end] : [v.end, v.start]; + onpick(first.toString(), last.toString()); + open = false; + }} + > + {#snippet children({ months, weekdays })} + + + + + + {#each months as month (month.value.toString())} + + + + {#each weekdays as weekday (weekday)}{weekday.slice(0, 2)}{/each} + + + + {#each month.weeks as week (week[0].toString())} + + {#each week as date (date.toString())} + + {/each} + + {/each} + + + {/each} +

{value.start && !value.end ? 'Pick the last day' : 'Pick the first day, then the last'}

+ {/snippet} + + + + + diff --git a/ui/src/lib/state/prefetch.ts b/ui/src/lib/state/prefetch.ts index b4ff520435..98e980d89c 100644 --- a/ui/src/lib/state/prefetch.ts +++ b/ui/src/lib/state/prefetch.ts @@ -1,11 +1,6 @@ import { browser } from '$app/environment'; import type { ApiClient, Scope } from '$lib/api'; -/** - * How long a click waits on the current page for the next page's data, so a quick answer shows - * whole instead of empty for a moment; longer than this, the page switches and shows its own loading state. - */ -const HOLD_MS = 200; /** A prefetched read older than this, such as from a hover with no click, is read again instead. */ const KEEP_MS = 5000; @@ -22,6 +17,8 @@ const keyOf = (method: PropertyKey, args: unknown[]) => `${String(method)}|${JSO */ export function withPrefetch(client: ApiClient): ApiClient { base = client; + // After a change, a read started earlier may predate it; the page reads again instead. + client.changes.subscribe(() => started.clear()); return new Proxy(client, { get(target, method, receiver) { const value = Reflect.get(target, method, receiver); @@ -40,13 +37,15 @@ export function withPrefetch(client: ApiClient): ApiClient { } /** - * Starts the reads the page at `pathname` makes first and waits for them briefly, so a fast answer - * arrives with the page. Errors are left for the page to show. + * Starts the reads the page at `pathname` makes first, from a hover or a click, without holding the + * switch: the page takes them over and shows its loading state until they answer. Errors are left + * for the page to show. */ -export async function prefetch(scope: Scope | null, pathname: string, search: URLSearchParams, reads: [RegExp, Read][]): Promise { +export function prefetch(scope: Scope | null, pathname: string, search: URLSearchParams, reads: [RegExp, Read][]): void { if (!browser || !base || !scope) return; const found = reads.map(([pattern, read]) => [pathname.match(pattern), read] as const).find(([match]) => match); if (!found) return; + for (const [key, kept] of started) if (Date.now() - kept.at >= KEEP_MS) started.delete(key); const client = base; const recorder = new Proxy(client, { get(target, method, receiver) { @@ -61,6 +60,5 @@ export async function prefetch(scope: Scope | null, pathname: string, search: UR }; } }); - const pending = Promise.allSettled(found[1](recorder, scope, found[0]!, search)); - await Promise.race([pending, new Promise((resolve) => setTimeout(resolve, HOLD_MS))]); + found[1](recorder, scope, found[0]!, search); } diff --git a/ui/src/lib/state/resource.svelte.ts b/ui/src/lib/state/resource.svelte.ts index d687c89941..e79faabe04 100644 --- a/ui/src/lib/state/resource.svelte.ts +++ b/ui/src/lib/state/resource.svelte.ts @@ -5,8 +5,6 @@ import { app } from './app.svelte'; /** How long a burst of change signals may last before one reload starts. */ const SIGNAL_PAUSE_MS = 300; -/** A first load faster than this shows no loading text, so it doesn't flash. */ -const SLOW_MS = 400; /** * The last answer of each read, kept in memory for this page load only, so returning to a page shows @@ -26,8 +24,6 @@ export type Resource = { readonly data: T | undefined; readonly error: string | null; readonly loading: boolean; - /** Loading with nothing to show for longer than a moment; pages show loading text only then. */ - readonly slow: boolean; refresh(): void; }; @@ -51,8 +47,6 @@ export function createResource( let data = $state(undefined); let error = $state(null); let loading = $state(true); - let slow = $state(false); - let slowTimer: ReturnType | null = null; let shownKey = ''; let latest = 0; let settled = 0; @@ -66,12 +60,7 @@ export function createResource( if (key !== shownKey && remembered.has(key)) data = remembered.get(key) as T; shownKey = key; loading = true; - if (slowTimer) clearTimeout(slowTimer); - slowTimer = setTimeout(() => (slow = data === undefined), SLOW_MS); const done = () => { - if (slowTimer) clearTimeout(slowTimer); - slowTimer = null; - slow = false; settled = request; if (reloadAgain && request === latest) { reloadAgain = false; @@ -126,7 +115,6 @@ export function createResource( }); return () => { unsubscribe(); - if (slowTimer) clearTimeout(slowTimer); if (pause) clearTimeout(pause); pause = null; }; @@ -142,9 +130,6 @@ export function createResource( get loading() { return loading; }, - get slow() { - return slow; - }, refresh }; } diff --git a/ui/src/routes/+layout.svelte b/ui/src/routes/+layout.svelte index 9e76a357b0..0d7921d3ba 100644 --- a/ui/src/routes/+layout.svelte +++ b/ui/src/routes/+layout.svelte @@ -1,4 +1,5 @@ @@ -113,12 +139,42 @@ {#if item.kind === 'stale_models'} goto(to(planHref(staleSelection)))}>Rebuild {staleSelection.length} {/if} + {#if 'acknowledged' in item && canAcknowledge} + + {/if} {view.action} → {/each} {/if} + {#if handled.length} +
+ + {#if showHandled} + {#each handled as { item, view } (view.key)} + {@const ack = acknowledgementOf(item)} +
+ + {view.title} + {ack ? `Acknowledged by ${ack.by} · ${formatAgo(ack.at, app.now)}` : ''} + {view.since} + + {#if 'acknowledged' in item && canAcknowledge} + + {/if} + {view.action} → + +
+ {/each} + {/if} +
+ {/if} + {#if acknowledgeError}

{acknowledgeError}

{/if}
@@ -187,7 +243,7 @@ {g.failed || '—'} {:else} -
{overview.loading ? (overview.slow ? 'Loading…' : '') : 'Nothing deployed yet. Deploy the project to see its models here.'}
+
{#if overview.data === undefined && !overview.error}{:else}Nothing deployed yet. Deploy the project to see its models here.{/if}
{/each}
{/if} @@ -207,7 +263,7 @@ {c.gitAuthor ? `${c.gitAuthor} · ` : ''}{formatAgo(c.publishedAt, app.now)} {:else} -
{overview.loading ? (overview.slow ? 'Loading…' : '') : 'Nothing deployed yet.'}
+
{#if overview.data === undefined && !overview.error}{:else}Nothing deployed yet.{/if}
{/each} @@ -230,6 +286,18 @@ .issue small { color: var(--sb-text-faint); font-size: 12px; white-space: nowrap; } .issue-actions { display: flex; min-width: 120px; align-items: center; justify-content: flex-end; gap: 10px; } .issue-actions a { color: var(--primary); font-weight: 500; white-space: nowrap; } + .ack { display: inline-flex; align-items: center; gap: 5px; height: 22px; padding: 0 8px; border: 1px solid var(--border); border-radius: 2px; color: var(--muted-foreground); font-size: 12px; white-space: nowrap; } + .ack:hover:not(:disabled) { border-color: var(--border-strong); color: var(--foreground); } + .ack:disabled, .undo:disabled { opacity: 0.5; } + .handled { border-top: 1px solid var(--inner-line); } + .handled-toggle { display: flex; align-items: center; gap: 6px; height: 34px; padding: 0 14px 0 17px; color: var(--muted-foreground); font-size: 12px; } + .handled-toggle:hover { color: var(--foreground); } + .handled-toggle :global(svg) { transition: transform 120ms; } + .handled-toggle :global(svg.open) { transform: rotate(90deg); } + .issue.muted .issue-title { color: var(--muted-foreground); font-weight: 500; } + .undo { color: var(--muted-foreground); font-size: 12px; } + .undo:hover:not(:disabled) { color: var(--foreground); } + .ack-error { margin: 0; padding: 8px 17px; border-top: 1px solid var(--inner-line); color: var(--sb-error); font-size: 12px; } .issue-icon { display: grid; width: 22px; height: 22px; place-items: center; border-radius: 2px; } .issue-icon.error { color: var(--sb-error); background: color-mix(in srgb, var(--sb-error) 12%, transparent); } .issue-icon.warning { color: var(--sb-warning); background: color-mix(in srgb, var(--sb-warning) 12%, transparent); } diff --git a/ui/src/routes/admin/people/+page.svelte b/ui/src/routes/admin/people/+page.svelte index 54a4eac7c4..109a5e3bcd 100644 --- a/ui/src/routes/admin/people/+page.svelte +++ b/ui/src/routes/admin/people/+page.svelte @@ -45,7 +45,7 @@ {#if !allowed} {:else} - + {#snippet filters()} ({ value: r, label: r }))} all="Any role" /> ({ value: l.id, label: l.name }))} all="All locations" /> diff --git a/ui/src/routes/admin/people/[name]/+page.svelte b/ui/src/routes/admin/people/[name]/+page.svelte index af8f8aaf98..de82e87467 100644 --- a/ui/src/routes/admin/people/[name]/+page.svelte +++ b/ui/src/routes/admin/people/[name]/+page.svelte @@ -1,4 +1,5 @@ - -
-

Run {job.kind === 'group' ? 'group' : 'job'} {job.key} now

- {#if canFullRefresh || canRewrite} -
- {#if canFullRefresh} - - {/if} - {#if canRewrite && cursorType} - - {#if useWindow} -
- - to - -
- {#if windowInvalid}

{windowInvalid}

{/if} - {/if} - {/if} - {#if !canRewrite}

Rebuilding or reprocessing a range needs extra permission (run.rewrite_history).

{:else if rewrites}

This replaces data that was already built. It's recorded in the audit log.

{/if} -
- {/if} - {#if error}

{error}

{/if} -
- - -
-
- - diff --git a/ui/src/routes/layout.css b/ui/src/routes/layout.css index 82da55e5f4..f76b7eee6d 100644 --- a/ui/src/routes/layout.css +++ b/ui/src/routes/layout.css @@ -245,6 +245,11 @@ text-align: left; white-space: nowrap; } +/* While the loading rows show, the header keeps its height but not its text: its columns are sized from + the header alone until the rows arrive, so it would visibly reflow. Header and rows then appear together. */ +.sb-list:has(tr.load-row[role='status']) thead th { + color: transparent; +} .sb-list tbody td { padding: 10px 16px; border-bottom: 1px solid var(--border); @@ -477,3 +482,31 @@ font-weight: 500; } +/* Placeholder bars a page shows while a read has nothing to show yet. */ +.sb-skeleton { + display: block; + height: 10px; + border-radius: 4px; + /* Mixed from the text colour, so the bar and its sheen stand out from any surface in both themes. */ + background: linear-gradient( + 90deg, + color-mix(in srgb, var(--foreground) 8%, transparent) 0%, + color-mix(in srgb, var(--foreground) 18%, transparent) 50%, + color-mix(in srgb, var(--foreground) 8%, transparent) 100% + ); + background-size: 200% 100%; + animation: sb-shimmer 1.2s ease-in-out infinite; +} +@keyframes sb-shimmer { + from { + background-position: 100% 0; + } + to { + background-position: -100% 0; + } +} +@media (prefers-reduced-motion: reduce) { + .sb-skeleton { + animation: none; + } +} diff --git a/ui/src/routes/lineage/+page.svelte b/ui/src/routes/lineage/+page.svelte index ff4087b94b..f873d077ce 100644 --- a/ui/src/routes/lineage/+page.svelte +++ b/ui/src/routes/lineage/+page.svelte @@ -1,4 +1,5 @@ + + + +
+
+ + + {#if single && first} + Rerun of{what(first)} · {shortRunId(first.run.id)} + {:else} + Rerun{remaining.length} runs + {/if} + +
+ {#if single} + {#each choices as c (c.mode)} + + {/each} + {:else} + + + {/if} +
+ Repeats each run's own version and settings, as frozen when it started +
+ + {#if loadError}

{loadError}

{/if} + +
+
+ {#if !details.data} +
Loading the run…
+ {:else if single && first} +
+ {nodes.length} node{nodes.length === 1 ? '' : 's'}·{runStatusLabel[first.run.status].toLowerCase()} {formatAgo(first.run.endedAt, app.now)} +
+ + +
+
+ {#if view === 'graph' && catalog.data} + {#if drawable.length}
{}} />
{/if} + {:else} + + + {#each nodes as key (key)} + {@const node = byKey.get(key)} + {@const outcome = first.nodes.find((n) => n.nodeKey === key)?.status} + + + + + + {/each} + +
{key.slice(key.indexOf(':') + 1)}{node ? kindLine(node) : ''}{outcome ? `was ${outcome.replace('_', ' ')}` : ''}
+ {/if} + {:else} +
{ready.length} of {rows.length} runs·{ready.reduce((n, r) => n + r.nodes.length, 0)} nodes
+ + + {#each rows as r (r.detail.run.id)} + + + + + + {/each} + +
{what(r.detail)} {shortRunId(r.detail.run.id)}{runStatusLabel[r.detail.run.status].toLowerCase()} {formatAgo(r.detail.run.endedAt, app.now)}{r.blocked ?? (r.nodes.length ? `${r.nodes.length} node${r.nodes.length === 1 ? '' : 's'}` : 'Nothing to rerun')}
+ {/if} +
+ +
+ +
+
+ {#if launchError}{launchError}{:else if blockedReason}{blockedReason}{/if} +
+ ▷ Run +
+
+ + diff --git a/ui/src/routes/providers/+page.svelte b/ui/src/routes/providers/+page.svelte index 7942fd5cbe..33f6eac232 100644 --- a/ui/src/routes/providers/+page.svelte +++ b/ui/src/routes/providers/+page.svelte @@ -21,7 +21,7 @@ - +
diff --git a/ui/src/routes/providers/[name]/+page.svelte b/ui/src/routes/providers/[name]/+page.svelte index 0221f216b4..daf5582a61 100644 --- a/ui/src/routes/providers/[name]/+page.svelte +++ b/ui/src/routes/providers/[name]/+page.svelte @@ -1,4 +1,5 @@
-
- {#each LEVELS as l (l.value)} - - {/each} -
-
- {#each present as c (c.value)} - - {/each} -
+ + {#if present.length > 1}{/if} {#if hasOutput}