diff --git a/scripts/verify-stac.py b/scripts/verify-stac.py index 059ac03..a5ccce9 100755 --- a/scripts/verify-stac.py +++ b/scripts/verify-stac.py @@ -1450,10 +1450,12 @@ def check_polygon_row_dup(doc: dict, mcp: MCPClient) -> list[Finding]: # --------------------------------------------------------------------------- _PARTITION_KEY_RE = re.compile(r"/([^/=]+)=\*/") -# GeoParquet 1.1 bbox-covering struct leaves: a STAC declares a single `bbox` column but -# the file stores it as a struct with these leaves. Neither the `bbox` declaration nor the -# leaves should be treated as absent/undocumented. -_BBOX_COVER = {"xmin", "ymin", "zmin", "xmax", "ymax", "zmax"} +# A GeoParquet 1.1 bbox covering is spatial-index machinery, not a column an author writes +# up, so an undeclared one is not an undocumented column. Its DECLARED side needs no +# special case any more: the check reads a column's name from its leaf's path, so the +# covering struct lands on 'bbox' — the name the STAC declares — like any other column. +# The leaf names stay listed for a writer that emits them as real top-level columns. +_BBOX_COVER = {"bbox", "xmin", "ymin", "zmin", "xmax", "ymax", "zmax"} def _partition_keys(href: str) -> set[str]: @@ -1472,8 +1474,15 @@ def check_declared_schema_matches_data(doc: dict, mcp: MCPClient) -> list[Findin declared but absent from EVERY file -> HARD (stale STAC; a metadata fix) declared but absent from SOME files -> HARD (heterogeneous; a data rebuild) present in data, undocumented -> ADVISORY (self-describing contract; kept - advisory because nested/covering leaves make - a hard extra-column failure FP-prone) + advisory because a lookup/crosswalk table + may carry a column nobody wrote up yet) + + Both comparisons run IN SQL and return only the discrepancies, for the same reason + check_values_match_distinct does it: the MCP query tool caps a result at 50 rows, so + reading back a whole column list and diffing it in Python drops the tail of any asset + wider than that and reports the dropped columns as absent. gbif-hex-2026-06 has 61 + top-level columns, and which 11 vanished varied run to run with the (unordered) + GROUP BY. A discrepancy list is short by construction — usually empty. """ out = [] for key, asset in doc.get("assets", {}).items(): @@ -1486,57 +1495,88 @@ def check_declared_schema_matches_data(doc: dict, mcp: MCPClient) -> list[Findin if not declared: continue # parquet-no-table-columns already HARD-flags a missing schema part_keys = _partition_keys(asset.get("href", "")) + # A path-supplied partition key and the writer-named geometry column are exempt in + # both directions: neither has to appear in a footer, and neither is undocumented. + exempt = set(part_keys) + original = {} + for name in declared: + low = name.lower() + original.setdefault(low, name) # report the STAC's own spelling back + if _is_geom_col(name): + exempt.add(low) + compare = [low for low in original if low not in exempt] + if not compare: + continue + # `present` reads TOP-LEVEL column names from each leaf chunk's `path_in_schema`, + # whose FIRST segment is the column that leaf belongs to: a `string[]` column reads + # 'common_names_en, list, element', a struct 'bbox, xmin', a map + # 'tags, key_value, key', a plain column just its own name. + # + # `parquet_schema`'s flat name list cannot answer this. The parquet schema is a TREE + # flattened in pre-order, and a LIST / STRUCT / MAP column is a group node whose + # physical `type` is NULL — the same NULL that marks the schema root. So filtering on + # `type IS NOT NULL` to drop the root also dropped the column itself, and kept the + # group's machinery leaves ('element', 'key', 'value', struct fields) in its place: + # every declared nested column read as absent while its leaves read as phantom + # top-level columns (iucn-taxonomy-2025: 9 populated `string[]` columns reported + # absent, plus an 'element' undocumented-column advisory). Splitting the leaf path is + # exact, needs no ordering assumption, and subsumes the GeoParquet `bbox` covering + # struct, which lands on 'bbox' like any other column. + # + # `parquet_metadata` is the column-chunk footer, so this stays a footer read. A file + # with no row groups has no chunks and drops out of both `present` and `total`, which + # is what we want: an empty file has no data to disagree with the STAC, and counting + # it would report every column as missing from it. + # MATERIALIZED so the footer is read once per query rather than once per CTE that + # references it — two queries, two reads, the same as the pair this replaced. + ctes = (f"WITH meta AS MATERIALIZED (SELECT lower(split_part(path_in_schema, ', ', 1)) " + f"AS name, file_name FROM parquet_metadata('{s3}')), " + "present AS (SELECT name, COUNT(DISTINCT file_name) AS nf FROM meta GROUP BY 1), " + "total AS (SELECT COUNT(DISTINCT file_name) AS n FROM meta) ") + cmp_values = ", ".join("('" + n.replace("'", "''") + "')" for n in sorted(compare)) + known_values = ", ".join( + "('" + n.replace("'", "''") + "')" for n in sorted(set(original) | exempt)) try: - total = int(mcp.query( - f"SELECT COUNT(DISTINCT file_name) AS n FROM parquet_schema('{s3}')")[0]["n"]) - # `type IS NOT NULL` keeps only leaf columns, dropping the schema root and any - # struct group node (whose physical type is NULL) — so `parquet_schema`'s - # nested/struct entries do not read as phantom columns (#534 caveat). - rows = mcp.query( - "SELECT lower(name) AS name, COUNT(DISTINCT file_name) AS nf " - f"FROM parquet_schema('{s3}') WHERE type IS NOT NULL GROUP BY 1") + missing = mcp.query( + ctes + f"SELECT d.name AS name, COALESCE(p.nf, 0) AS nf, t.n AS total " + f"FROM (VALUES {cmp_values}) d(name) CROSS JOIN total t " + "LEFT JOIN present p ON p.name = d.name " + "WHERE t.n > 0 AND COALESCE(p.nf, 0) < t.n ORDER BY 1") + extra = mcp.query( + ctes + "SELECT p.name AS name, t.n AS total FROM present p CROSS JOIN total t " + f"WHERE t.n > 0 AND p.nf = t.n AND p.name NOT IN " + f"(SELECT * FROM (VALUES {known_values}) k(name)) ORDER BY 1") except (MCPError, ValueError, KeyError, IndexError) as e: out.append(Finding(ADVISORY, "schema-match-check-failed", f"asset '{key}': could not read parquet footers ({e}).")) continue - if not total: - continue - present = {} - for r in rows: + for r in missing: try: - present[str(r["name"]).lower()] = int(r["nf"]) + low, nf, total = str(r["name"]).lower(), int(r["nf"]), int(r["total"]) except (KeyError, ValueError, TypeError): continue - # a `bbox` declared as one column may be stored as a covering struct — treat it as - # present if its leaves are (in as many files as the leaves appear). - bbox_leaf_files = (min((present[l] for l in _BBOX_COVER if l in present), default=0)) - - declared_lower = set() - for name in declared: - low = name.lower() - declared_lower.add(low) - if low in part_keys or _is_geom_col(name): - continue # path-supplied partition key / writer-named geometry column - nf = present.get(low, 0) - if low == "bbox" and nf == 0: - nf = bbox_leaf_files # covering-struct case + name = original.get(low, low) if nf == 0: out.append(Finding(HARD, "declared-column-absent", f"asset '{key}': STAC declares column '{name}' but it is absent from " f"all {total} parquet file(s) — a stale/incorrect schema; fix the STAC " f"table:columns.")) - elif nf < total: + else: out.append(Finding(HARD, "declared-column-heterogeneous", f"asset '{key}': STAC declares column '{name}' but it is missing from " f"{total - nf} of {total} parquet file(s) — a partial/mixed-vintage " f"build; rebuild the asset. See data-workflows#534/#520.")) - for low, nf in sorted(present.items()): - if (nf == total and low not in declared_lower and low not in part_keys - and not _is_geom_col(low) and low not in _BBOX_COVER and low != "bbox"): - out.append(Finding(ADVISORY, "undocumented-column", - f"asset '{key}': parquet column '{low}' is present in all {total} " - f"file(s) but not declared in table:columns — add it (assets must be " - f"self-describing) or confirm it is intentional.")) + for r in extra: + try: + low, total = str(r["name"]).lower(), int(r["total"]) + except (KeyError, ValueError, TypeError): + continue + if _is_geom_col(low) or low in _BBOX_COVER or low.endswith("_bbox"): + continue + out.append(Finding(ADVISORY, "undocumented-column", + f"asset '{key}': parquet column '{low}' is present in all {total} " + f"file(s) but not declared in table:columns — add it (assets must be " + f"self-describing) or confirm it is intentional.")) return out diff --git a/tests/test_verify_stac.py b/tests/test_verify_stac.py index 137ae05..4321620 100644 --- a/tests/test_verify_stac.py +++ b/tests/test_verify_stac.py @@ -20,6 +20,7 @@ """ import importlib.util import pathlib +import re import unittest _SRC = pathlib.Path(__file__).resolve().parent.parent / "scripts" / "verify-stac.py" @@ -244,11 +245,41 @@ def _schema_doc(declared, href=HEX_HREF): "table:columns": [{"name": n} for n in declared]}}} +class _SchemaMCP: + """Emulate the two discrepancy queries the check runs, from a + {column: number-of-files-containing-it} map. + + Both comparisons happen in SQL and return only the columns that DISAGREE, so a stub + replaying a whole column list would not exercise what the check actually asks for. + The declared / known name lists are read back out of the SQL's VALUES clause. + """ + + def __init__(self, total, name_nf): + self.total, self.name_nf, self.sql = total, name_nf, [] + + @staticmethod + def _values(sql, marker): + head = sql.split(marker)[0] + return re.findall(r"\('([^']*)'\)", head[head.rfind("(VALUES"):]) + + def query(self, sql): + self.sql.append(sql) + if not self.total: + return [] + if "LEFT JOIN present" in sql: + return [{"name": n, "nf": self.name_nf.get(n, 0), "total": self.total} + for n in self._values(sql, ") d(name)") + if self.name_nf.get(n, 0) < self.total] + if "FROM present p CROSS JOIN total" in sql: + known = self._values(sql, ") k(name)") + return [{"name": n, "total": self.total} + for n, nf in sorted(self.name_nf.items()) + if nf == self.total and n not in known] + raise AssertionError(f"unscripted query: {sql}") + + def _schema_mcp(total, name_nf): - return ScriptedMCP([ - ("AS n FROM parquet_schema", [{"n": total}]), # total-files query - ("GROUP BY 1", [{"name": n, "nf": nf} for n, nf in name_nf.items()]), - ]) + return _SchemaMCP(total, name_nf) class DeclaredSchemaMatch(unittest.TestCase): @@ -294,6 +325,94 @@ def test_geometry_column_is_skipped(self): f = vs.check_declared_schema_matches_data(doc, _schema_mcp(1, {"_cng_fid": 1})) self.assertEqual(f, []) + def test_column_names_come_from_the_leaf_path_not_the_flat_name_list(self): + # Dependency-free guard on the generated SQL: a nested column is a group node with + # a NULL physical type, so any name list filtered on `type IS NOT NULL` loses it. + doc = _schema_doc(["_cng_fid"]) + mcp = _schema_mcp(1, {"_cng_fid": 1}) + vs.check_declared_schema_matches_data(doc, mcp) + grouped = [s for s in mcp.sql if "GROUP BY 1" in s][0] + self.assertIn("path_in_schema", grouped) + self.assertNotIn("type IS NOT NULL", grouped) + + +class DeclaredSchemaMatchAgainstRealParquet(unittest.TestCase): + """Run the generated SQL against a REAL parquet footer, under the MCP's own result-row + cap, rather than trusting that the SQL text looks right. + + Two live shapes this pins, both of which reported columns that exist as absent: + + * NESTED columns — iucn-taxonomy-2025's nine populated `string[]` columns + (`common_names_en`, `synonyms`, `threat_codes`, …) each read as "absent from all 1 + parquet file(s)" while the LIST's `element` leaf read as an undocumented top-level + column. + * WIDE assets — gbif-hex-2026-06 has 61 top-level columns and the MCP returns at + most 50 rows, so whichever 11 fell off the end read as absent. + + Acting on either would have deleted correct schema documentation. + """ + + MCP_ROW_CAP = 50 # the MCP query tool's result cap, reproduced here on purpose + ROWS = ("SELECT 1 AS _cng_fid, ['a','b'] AS names, {'xmin': 1, 'xmax': 2} AS bbox, " + "MAP{'k':'v'} AS tags, 'x' AS plain") + + def _run(self, declared, rows_sql=None): + try: + import duckdb + except ImportError: # pragma: no cover - CI installs it + self.skipTest("duckdb not installed") + import tempfile + cap = self.MCP_ROW_CAP + with tempfile.TemporaryDirectory() as tmp: + path = pathlib.Path(tmp) / "data_0.parquet" + con = duckdb.connect() + con.execute(f"COPY ({rows_sql or self.ROWS}) TO '{path}' (FORMAT PARQUET)") + + class Exec: + """Point the check's footer reads at the local file, and truncate the way + the MCP does. Both function names are rewritten so this is a true + red/green against either implementation rather than erroring on the s3 + path.""" + + def query(self, sql): + sql = re.sub(r"parquet_(metadata|schema)\('[^']*'\)", + lambda m: f"parquet_{m.group(1)}('{path}')", sql) + cur = con.execute(sql) + names = [d[0] for d in cur.description] + return [dict(zip(names, r)) for r in cur.fetchall()][:cap] + + return vs.check_declared_schema_matches_data(_schema_doc(declared), Exec()) + + def test_declared_nested_columns_are_not_reported_absent(self): + # THE iucn-taxonomy case: every one of these exists and holds data. `bbox` is the + # GeoParquet 1.1 covering struct, declared as one column and stored as leaves. + self.assertEqual(self._run(["_cng_fid", "names", "bbox", "tags", "plain"]), [], + "a LIST / STRUCT / MAP column must not read as absent") + + def test_nested_machinery_is_not_an_undocumented_column(self): + # 'element', 'key', 'value' and the struct's leaves are parts of a column, not + # columns; only the real undeclared columns may be reported. An undeclared bbox + # covering stays exempt — it is spatial-index machinery, not authored schema. + f = self._run(["_cng_fid", "plain"]) + self.assertEqual({x.code for x in f}, {"undocumented-column"}) + named = sorted(re.search(r"parquet column '([^']+)'", x.message).group(1) for x in f) + self.assertEqual(named, ["names", "tags"]) + + def test_a_wide_asset_survives_the_result_row_cap(self): + # THE gbif case: more columns than the MCP will return rows. Diffing a truncated + # column list in Python reports the tail as absent; diffing in SQL returns only + # the (here empty) discrepancy list. + wide = [f"c{i:02d}" for i in range(self.MCP_ROW_CAP + 11)] + f = self._run(wide, rows_sql="SELECT " + ", ".join(f"{i} AS {c}" for i, c in enumerate(wide))) + self.assertEqual(f, [], "a wide asset must not report its tail columns as absent") + + def test_a_genuinely_absent_column_still_hard_fails(self): + # Mutation guard: the fix must not blind the check it is fixing. + f = self._run(["_cng_fid", "names", "bbox", "tags", "plain", "ghost"]) + self.assertEqual([x.code for x in f], ["declared-column-absent"]) + self.assertEqual(f[0].severity, vs.HARD) + self.assertIn("ghost", f[0].message) + # --- #535: a vector hex must hold every feature of its flat GeoParquet -------