fix: verify-stac reports existing parquet columns as absent (nested columns, wide assets)

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
 
 

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