feat(cloud): emit dataset schema metadata when a schema check runs

m1n0 · claude · m1n0 · commit 5a351febc90c · 2026-07-15T15:58:35.000+02:00
Add a top-level metadata[] block to the sodaCoreInsertScanResults payload
carrying each dataset's discovered columns ({columnName, sourceDataType})
whenever a schema check produced a non-empty column set, so Soda Cloud can
refresh the dataset schema from a v4 contract or data standard.

Matches the v3 discover payload shape: raw un-parameterized data type
(SqlDataType.name), primary-key flag omitted. Emitted once per distinct
dataset that ran a schema check, and omitted entirely when no schema check
discovered columns so an empty schema never reaches the wire (the backend
full-reconciles and would otherwise soft-delete every column).

DTL-1807

Co-Authored-By: Claude Opus 4.8 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/soda-core/src/soda_core/common/soda_cloud.py b/soda-core/src/soda_core/common/soda_cloud.py
@@ -1588,6 +1588,56 @@ def _build_token_usage_dicts(contract_verification_result: ContractVerificationR
     return []
 
 
+def _build_dataset_metadata_json_dicts(results: list[ContractVerificationResult]) -> list[dict]:
+    """Top-level ``metadata`` array that refreshes each dataset's column schema
+    in Soda Cloud.
+
+    One entry per distinct dataset that ran a schema check which discovered a
+    non-empty column set. Reuses v3 discover's ``{columnName, sourceDataType}``
+    shape so the backend runs the same full column reconciliation (columns
+    absent from ``schema`` are soft-deleted) — so we send every discovered
+    column, in discovery order, and never an empty schema.
+
+    ``datasetQualifiedName`` is the collection's ``soda_qualified_dataset_name``
+    (the raw ``yaml.dataset`` dqn), identical to the identifier the backend
+    builds from each check, so the block resolves to the same dataset the checks
+    registered under. ``sourceDataType`` is the raw, un-parameterized DB type
+    (``SqlDataType.name``, e.g. ``character varying``) — not the parameterized
+    ``character varying(255)`` form — matching what v3 discover uploads.
+    ``isPrimaryKey`` is deliberately omitted (the v4 schema check doesn't
+    discover PKs); omitting it preserves the backend's existing PK flags.
+    """
+    from soda_core.contracts.impl.check_types.schema_check import SchemaCheckResult
+
+    dataset_metadata: list[dict] = []
+    seen_dataset_qualified_names: set[str] = set()
+    for r in results:
+        dataset_qualified_name: str = r.check_collection.soda_qualified_dataset_name
+        if dataset_qualified_name in seen_dataset_qualified_names:
+            continue
+        for check_result in r.check_results:
+            # Only a schema check with a discovered (non-empty) column set
+            # contributes a metadata entry — an empty schema would soft-delete
+            # every column on the backend, so we skip it.
+            if not isinstance(check_result, SchemaCheckResult) or not check_result.actual_columns:
+                continue
+            seen_dataset_qualified_names.add(dataset_qualified_name)
+            dataset_metadata.append(
+                {
+                    "datasetQualifiedName": dataset_qualified_name,
+                    "schema": [
+                        {
+                            "columnName": column.column_name,
+                            "sourceDataType": (column.sql_data_type.name if column.sql_data_type else None),
+                        }
+                        for column in check_result.actual_columns
+                    ],
+                }
+            )
+            break
+    return dataset_metadata
+
+
 def _build_check_collection_results_json_dict(
     results: list[ContractVerificationResult],
     wire_source: str = "soda-contract",
@@ -1690,6 +1740,11 @@ def _build_check_collection_results_json_dict(
             "postProcessingStages": post_processing_stages,
             "resultsIngestionMode": ingestion_mode.value,
             "tokenUsage": token_usage,
+            # Per-dataset column schema for datasets that ran a schema check,
+            # so the backend can refresh the dataset schema in Cloud. ``None``
+            # when no schema check produced columns, so ``to_jsonnable`` strips
+            # the key entirely (no empty ``metadata`` on the wire).
+            "metadata": _build_dataset_metadata_json_dicts(results) or None,
         }
     )
 
diff --git a/soda-tests/tests/unit/test_dataset_metadata_wire.py b/soda-tests/tests/unit/test_dataset_metadata_wire.py
@@ -0,0 +1,213 @@
+"""Unit tests for the top-level ``metadata`` block in the wire payload.
+
+Locks the shape of the dataset ``metadata`` array that
+``_build_check_collection_results_json_dict`` emits when a check collection
+runs a schema check. The block lets the backend refresh the dataset's column
+schema in Cloud, reusing the ``{columnName, sourceDataType}`` shape v3 discover
+emits (DTL-1807).
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+
+from soda_core.common.logs import Location
+from soda_core.common.metadata_types import ColumnMetadata, SqlDataType
+from soda_core.common.soda_cloud import _build_check_collection_results_json_dict
+from soda_core.contracts.contract_verification import (
+    Check,
+    CheckCollectionStatus,
+    CheckOutcome,
+    CheckResult,
+    Contract,
+    ContractVerificationResult,
+    DataSource,
+    YamlFileContentInfo,
+)
+from soda_core.contracts.impl.check_types.schema_check import SchemaCheckResult
+
+
+def _make_check(*, type: str, relative_path: str) -> Check:
+    return Check(
+        column_name=None,
+        type=type,
+        qualifier=None,
+        name=type,
+        relative_path=relative_path,
+        identity="abc",
+        definition=f"{type}: ...",
+        contract_file_line=1,
+        contract_file_column=1,
+        threshold=None,
+        attributes={},
+        location=Location(file_path="fake.yml", line=1, column=1),
+    )
+
+
+def _column(name: str, sql_data_type: SqlDataType) -> ColumnMetadata:
+    return ColumnMetadata(column_name=name, sql_data_type=sql_data_type)
+
+
+def _make_schema_check_result(actual_columns: list[ColumnMetadata]) -> SchemaCheckResult:
+    return SchemaCheckResult(
+        check=_make_check(type="schema", relative_path="checks.schema"),
+        outcome=CheckOutcome.PASSED,
+        expected_columns=[],
+        actual_columns=actual_columns,
+        expected_column_names_not_actual=[],
+        actual_column_names_not_expected=[],
+        column_data_type_mismatches=[],
+        are_columns_out_of_order=False,
+    )
+
+
+def _make_row_count_check_result() -> CheckResult:
+    return CheckResult(
+        check=_make_check(type="row_count", relative_path="checks.row_count"),
+        outcome=CheckOutcome.PASSED,
+        diagnostic_metric_values={"check_rows_tested": 0, "dataset_rows_tested": 0},
+    )
+
+
+def _make_result(
+    *,
+    soda_qualified_dataset_name: str,
+    dataset_prefix: list[str],
+    dataset_name: str,
+    check_results: list,
+) -> ContractVerificationResult:
+    ts = datetime(2026, 1, 1, 12, 0, 0, tzinfo=timezone.utc)
+    return ContractVerificationResult(
+        check_collection=Contract(
+            data_source_name="test_ds",
+            dataset_prefix=dataset_prefix,
+            dataset_name=dataset_name,
+            soda_qualified_dataset_name=soda_qualified_dataset_name,
+            source=YamlFileContentInfo(
+                source_content_str="# c",
+                local_file_path="/fake/c.yml",
+                soda_cloud_file_id="file-c",
+            ),
+        ),
+        data_source=DataSource(name="test_ds", type="postgres"),
+        data_timestamp=ts,
+        started_timestamp=ts,
+        ended_timestamp=ts,
+        status=CheckCollectionStatus.PASSED,
+        measurements=[],
+        check_results=check_results,
+        sending_results_to_soda_cloud_failed=False,
+        log_records=[],
+        post_processing_stages=[],
+    )
+
+
+def test_metadata_present_with_schema_of_column_name_and_source_data_type():
+    """A collection that ran a schema check → one ``metadata`` entry whose
+    ``schema`` mirrors the discovered columns as ``{columnName, sourceDataType}``."""
+    result = _make_result(
+        soda_qualified_dataset_name="postgres/public/customers",
+        dataset_prefix=["public"],
+        dataset_name="customers",
+        check_results=[
+            _make_schema_check_result(
+                actual_columns=[
+                    _column("id", SqlDataType(name="integer")),
+                    _column("name", SqlDataType(name="character varying", character_maximum_length=255)),
+                ]
+            )
+        ],
+    )
+
+    payload = _build_check_collection_results_json_dict([result], wire_source="soda-contract")
+
+    assert payload["metadata"] == [
+        {
+            "datasetQualifiedName": "postgres/public/customers",
+            "schema": [
+                {"columnName": "id", "sourceDataType": "integer"},
+                {"columnName": "name", "sourceDataType": "character varying"},
+            ],
+        }
+    ]
+
+
+def test_metadata_source_data_type_is_unparameterized():
+    """``sourceDataType`` is the raw un-parameterized type (``character
+    varying``), NOT the parameterized ``character varying(255)`` form."""
+    result = _make_result(
+        soda_qualified_dataset_name="postgres/public/customers",
+        dataset_prefix=["public"],
+        dataset_name="customers",
+        check_results=[
+            _make_schema_check_result(
+                actual_columns=[_column("name", SqlDataType(name="character varying", character_maximum_length=255))]
+            )
+        ],
+    )
+    payload = _build_check_collection_results_json_dict([result], wire_source="soda-contract")
+    assert payload["metadata"][0]["schema"][0]["sourceDataType"] == "character varying"
+
+
+def test_metadata_absent_when_no_schema_check():
+    """No schema check → no ``metadata`` key at all (``to_jsonnable`` strips it)."""
+    result = _make_result(
+        soda_qualified_dataset_name="postgres/public/customers",
+        dataset_prefix=["public"],
+        dataset_name="customers",
+        check_results=[_make_row_count_check_result()],
+    )
+    payload = _build_check_collection_results_json_dict([result], wire_source="soda-contract")
+    assert "metadata" not in payload
+
+
+def test_metadata_absent_when_schema_check_has_no_columns():
+    """A schema check that discovered no columns (missing dataset) → no
+    ``metadata`` entry. We never send an empty schema — that would soft-delete
+    every column on the backend."""
+    result = _make_result(
+        soda_qualified_dataset_name="postgres/public/customers",
+        dataset_prefix=["public"],
+        dataset_name="customers",
+        check_results=[_make_schema_check_result(actual_columns=[])],
+    )
+    payload = _build_check_collection_results_json_dict([result], wire_source="soda-contract")
+    assert "metadata" not in payload
+
+
+def test_metadata_one_entry_per_dataset_in_session():
+    """A combined multi-file session emits one ``metadata`` entry per distinct
+    dataset that ran a schema check — prefixed and non-prefixed dqns alike."""
+    prefixed = _make_result(
+        soda_qualified_dataset_name="postgres/public/customers",
+        dataset_prefix=["public"],
+        dataset_name="customers",
+        check_results=[_make_schema_check_result(actual_columns=[_column("id", SqlDataType(name="integer"))])],
+    )
+    non_prefixed = _make_result(
+        soda_qualified_dataset_name="postgres/orders",
+        dataset_prefix=[],
+        dataset_name="orders",
+        check_results=[_make_schema_check_result(actual_columns=[_column("total", SqlDataType(name="numeric"))])],
+    )
+    no_schema = _make_result(
+        soda_qualified_dataset_name="postgres/public/events",
+        dataset_prefix=["public"],
+        dataset_name="events",
+        check_results=[_make_row_count_check_result()],
+    )
+
+    payload = _build_check_collection_results_json_dict(
+        [prefixed, non_prefixed, no_schema], wire_source="data-standard"
+    )
+
+    assert payload["metadata"] == [
+        {
+            "datasetQualifiedName": "postgres/public/customers",
+            "schema": [{"columnName": "id", "sourceDataType": "integer"}],
+        },
+        {
+            "datasetQualifiedName": "postgres/orders",
+            "schema": [{"columnName": "total", "sourceDataType": "numeric"}],
+        },
+    ]
