1278 recursive databricks and spark checks and custom queries

CHANGELOG.md

@@ -5,6 +5,12 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [Unreleased]
+
+### Added
+
+- Databricks backend now supports recursive array and struct checks via CTE-based virtual models, enabling read-only contract validation on warehouses with limited CREATE permissions (#1278).
+
 ## [1.0.1] - 2026-06-10
 
 ### Added

README.md

@@ -2523,6 +2523,20 @@ Ensure you have a JDK 17 or 21 installed. Java 25 causes issues.
 java --version
 ```
 
+#### Linux system package for postgres/psycopg-based tests
+
+**Ubuntu/Debian:**
+```bash
+sudo apt-get update
+sudo apt-get install -y libpq-dev
+```
+**Fedora/RHEL:**
+```bash
+# Fedora/RHEL:
+sudo dnf install -y postgresql-devel
+# Arch:
+sudo pacman -S postgresql-libs
+```
 
 ### Docker Build
 

datacontract/engines/checks/create_checks.py

@@ -23,10 +23,14 @@
 
 from datacontract.engines.checks.check_spec import CheckSpec, MetricType, Op, Threshold
 from datacontract.engines.checks.type_normalize import normalize_type_name
+from datacontract.model.server import get_server_type
 
 logger = logging.getLogger(__name__)
 
 _FILE_SERVER_TYPES = {"local", "s3", "gcs", "azure"}
+_VERIFIED_NESTED_SQL_SERVER_TYPES = {"dataframe", "databricks"}
+_SUPPORTED_NESTED_STRUCT_SERVER_TYPES = {"dataframe", "databricks"}
+_SUPPORTED_NESTED_ARRAY_SERVER_TYPES = {"dataframe", "databricks"}
 
 
 # ---------------------------------------------------------------------------
@@ -60,6 +64,39 @@ def expected_type_category(prop: SchemaProperty) -> tuple[str, str]:
     return normalize_type_name(label), (label or "")
 
 
+def _property_type(prop: SchemaProperty) -> str:
+    return normalize_type_name(prop.physicalType or prop.logicalType)
+
+
+def _iter_property_paths(
+    model: str,
+    properties: list[SchemaProperty] | None,
+    server_type: str | None,
+    prefix: str | None = None,
+    nested: bool = False,
+):
+    for prop in properties or []:
+        field = prop.physicalName or prop.name
+        field_path = f"{prefix}.{field}" if prefix else field
+        yield model, field_path, prop, nested
+
+        prop_type = _property_type(prop)
+        if (
+            server_type in _SUPPORTED_NESTED_STRUCT_SERVER_TYPES
+            and prop_type in {"object", "record", "struct"}
+            and prop.properties
+        ):
+            yield from _iter_property_paths(model, prop.properties, server_type, field_path, True)
+        elif (
+            server_type in _SUPPORTED_NESTED_ARRAY_SERVER_TYPES
+            and prop_type == "array"
+            and prop.items
+            and prop.items.properties
+        ):
+            nested_model = f"{model}__{field}"
+            yield from _iter_property_paths(nested_model, prop.items.properties, server_type, None, True)
+
+
 _PERCENT_UNITS = {"percent", "percentage", "%"}
 
 
@@ -146,25 +183,24 @@ def create_checks(
 
 def _to_schema_checks(schema_object: SchemaObject, server: Optional[Server]) -> List[CheckSpec]:
     checks: List[CheckSpec] = []
-    server_type = server.type if server and server.type else None
+    server_type = get_server_type(server) if server is not None else None
     model = to_schema_name(schema_object, server_type)
     properties = schema_object.properties or []
     check_types = is_check_types(server)
     uses_raw_view = (
-        server is not None and server.type in _FILE_SERVER_TYPES and server.format in ("csv", "parquet", "json")
+        server is not None and server_type in _FILE_SERVER_TYPES and server.format in ("csv", "parquet", "json")
     )
 
-    for prop in properties:
+    for item_model, field, prop, is_nested in _iter_property_paths(model, properties, server_type):
         # ODCS physicalName is the real column; mirror to_schema_name at field level.
-        field = prop.physicalName or prop.name
 
         checks.append(
             CheckSpec(
-                key=f"{model}__{field}__field_is_present",
+                key=f"{item_model}__{field}__field_is_present",
                 category="schema",
                 type="field_is_present",
                 name=f"Check that field '{field}' is present",
-                model=model,
+                model=item_model,
                 field=field,
                 metric=MetricType.FIELD_PRESENT,
                 uses_raw_view=uses_raw_view,
@@ -175,11 +211,11 @@ def _to_schema_checks(schema_object: SchemaObject, server: Optional[Server]) ->
             category, label = expected_type_category(prop)
             checks.append(
                 CheckSpec(
-                    key=f"{model}__{field}__field_type",
+                    key=f"{item_model}__{field}__field_type",
                     category="schema",
                     type="field_type",
                     name=f"Check that field {field} has type {label}",
-                    model=model,
+                    model=item_model,
                     field=field,
                     metric=MetricType.FIELD_TYPE,
                     expected_category=category,
@@ -190,7 +226,7 @@ def _to_schema_checks(schema_object: SchemaObject, server: Optional[Server]) ->
         if prop.required:
             checks.append(
                 _missing_count_check(
-                    model,
+                    item_model,
                     field,
                     "field_required",
                     Threshold(Op.EQ, 0),
@@ -201,7 +237,7 @@ def _to_schema_checks(schema_object: SchemaObject, server: Optional[Server]) ->
         if prop.unique:
             checks.append(
                 _duplicate_count_check(
-                    model,
+                    item_model,
                     field,
                     "field_unique",
                     Threshold(Op.EQ, 0),
@@ -214,7 +250,7 @@ def _to_schema_checks(schema_object: SchemaObject, server: Optional[Server]) ->
         if min_length is not None:
             checks.append(
                 _invalid_count_check(
-                    model,
+                    item_model,
                     field,
                     "field_min_length",
                     name=f"Check that field {field} has a min length of {min_length}",
@@ -226,7 +262,7 @@ def _to_schema_checks(schema_object: SchemaObject, server: Optional[Server]) ->
         if max_length is not None:
             checks.append(
                 _invalid_count_check(
-                    model,
+                    item_model,
                     field,
                     "field_max_length",
                     name=f"Check that field {field} has a max length of {max_length}",
@@ -238,7 +274,7 @@ def _to_schema_checks(schema_object: SchemaObject, server: Optional[Server]) ->
         if minimum is not None:
             checks.append(
                 _invalid_count_check(
-                    model,
+                    item_model,
                     field,
                     "field_minimum",
                     name=f"Check that field {field} has a minimum of {minimum}",
@@ -250,7 +286,7 @@ def _to_schema_checks(schema_object: SchemaObject, server: Optional[Server]) ->
         if maximum is not None:
             checks.append(
                 _invalid_count_check(
-                    model,
+                    item_model,
                     field,
                     "field_maximum",
                     name=f"Check that field {field} has a maximum of {maximum}",
@@ -262,7 +298,7 @@ def _to_schema_checks(schema_object: SchemaObject, server: Optional[Server]) ->
         if exclusive_minimum is not None:
             checks.append(
                 _invalid_count_check(
-                    model,
+                    item_model,
                     field,
                     "field_minimum",
                     name=f"Check that field {field} has a minimum of {exclusive_minimum}",
@@ -271,7 +307,7 @@ def _to_schema_checks(schema_object: SchemaObject, server: Optional[Server]) ->
             )
             checks.append(
                 _invalid_count_check(
-                    model,
+                    item_model,
                     field,
                     "field_not_equal",
                     name=f"Check that field {field} is not equal to {exclusive_minimum}",
@@ -283,7 +319,7 @@ def _to_schema_checks(schema_object: SchemaObject, server: Optional[Server]) ->
         if exclusive_maximum is not None:
             checks.append(
                 _invalid_count_check(
-                    model,
+                    item_model,
                     field,
                     "field_maximum",
                     name=f"Check that field {field} has a maximum of {exclusive_maximum}",
@@ -292,7 +328,7 @@ def _to_schema_checks(schema_object: SchemaObject, server: Optional[Server]) ->
             )
             checks.append(
                 _invalid_count_check(
-                    model,
+                    item_model,
                     field,
                     "field_not_equal",
                     name=f"Check that field {field} is not equal to {exclusive_maximum}",
@@ -304,7 +340,7 @@ def _to_schema_checks(schema_object: SchemaObject, server: Optional[Server]) ->
         if pattern is not None:
             checks.append(
                 _invalid_count_check(
-                    model,
+                    item_model,
                     field,
                     "field_regex",
                     name=f"Check that field {field} matches regex pattern {pattern}",
@@ -316,7 +352,7 @@ def _to_schema_checks(schema_object: SchemaObject, server: Optional[Server]) ->
         if enum_values:
             checks.append(
                 _invalid_count_check(
-                    model,
+                    item_model,
                     field,
                     "field_enum",
                     name=f"Check that field {field} only contains enum values {enum_values}",
@@ -325,7 +361,7 @@ def _to_schema_checks(schema_object: SchemaObject, server: Optional[Server]) ->
             )
 
         if prop.quality:
-            checks.extend(_quality_checks(model, field, prop.quality, server))
+            checks.extend(_quality_checks(item_model, field, prop.quality, server, is_nested=is_nested))
 
     if schema_object.quality:
         checks.extend(_quality_checks(model, None, schema_object.quality, server))
@@ -421,7 +457,7 @@ def _row_count_check(model, threshold: Threshold, severity=None) -> CheckSpec:
 # quality list
 # ---------------------------------------------------------------------------
 def _quality_checks(
-    model: str, field: Optional[str], quality_list: List[DataQuality], server: Optional[Server]
+    model: str, field: Optional[str], quality_list: List[DataQuality], server: Optional[Server], is_nested: bool = False
 ) -> List[CheckSpec]:
     checks: List[CheckSpec] = []
     count = 0
@@ -444,6 +480,31 @@ def _quality_checks(
                 )
             )
         elif quality.type == "sql":
+            server_type = get_server_type(server) if server is not None else None
+            if is_nested and server_type not in _VERIFIED_NESTED_SQL_SERVER_TYPES:
+                if field is None:
+                    check_key = f"{model}__quality_sql_{count}"
+                    check_type = "model_quality_sql"
+                else:
+                    check_key = f"{model}__{field}__quality_sql_{count}"
+                    check_type = "field_quality_sql"
+                checks.append(
+                    CheckSpec(
+                        key=check_key,
+                        category="quality",
+                        type=check_type,
+                        name=quality.description or "Quality Check",
+                        model=model,
+                        field=field,
+                        metric=MetricType.UNSUPPORTED,
+                        preset_result="warning",
+                        preset_reason=(
+                            "Nested SQL quality checks are only verified for Spark (dataframe) and Databricks."
+                        ),
+                    )
+                )
+                count += 1
+                continue
             if field is None:
                 check_key = f"{model}__quality_sql_{count}"
                 check_type = "model_quality_sql"