From 60c59a231e0c7b0b721aab022c54ecff3cd459e0 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 5 Aug 2026 18:28:42 +0200 Subject: [PATCH 1/2] Fix Databricks tests for GEOGRAPHY and GEOMETRY columns ibis reads the first GEOGRAPHY/GEOMETRY type parameter as a geometry subtype (PostGIS's GEOGRAPHY(POINT, 4326)), but Databricks declares only the SRID, GEOGRAPHY(4326), so reflecting the table raised KeyError: '4326' and every check of the model failed - including checks on unrelated columns. Patch ibis's Databricks type mapper to read a lone numeric parameter as the SRID, and type any column ibis still cannot represent as Unknown so the rest of the model is still checked, as the pyspark path already does. Also drop the databricksType custom property from `import unity`: it always duplicated physicalType, and the export it fed already falls back to physicalType. So that the export stays lossless, keep varchar(n)/char(n) lengths and pass through parameterized types with no databricks mapping. Fixes #1483 --- CHANGELOG.md | 7 + .../engines/ibis/connections/connect.py | 6 + .../ibis/connections/databricks_patch.py | 142 ++++++++++++++++++ datacontract/export/sql_type_converter.py | 11 +- datacontract/imports/unity_importer.py | 1 - docs/docs/reference/databricks.md | 2 +- .../databricks-unity/import/datacontract.yaml | 21 --- .../import/datacontract_complex_types.yaml | 15 -- tests/test_sql_type_converter_physicaltype.py | 20 ++- tests/test_test_databricks.py | 91 +++++++++++ 10 files changed, 275 insertions(+), 41 deletions(-) create mode 100644 datacontract/engines/ibis/connections/databricks_patch.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 089dd5160..78a0c6c38 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,13 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Changed +- `datacontract import unity` no longer writes the `databricksType` custom property, which duplicated `physicalType` +- `datacontract export sql --server databricks` keeps the declared length of `varchar(n)` and `char(n)` instead of exporting `STRING` + +### Fixed +- `datacontract test` for Databricks no longer fails all checks of a model with a `GEOGRAPHY` or `GEOMETRY` column (#1483) + ## [1.1.0] - 2026-08-04 This release drops the pyspark compile-time dependency. The server types `dataframe` and `databricks` still work with a provided Spark session. diff --git a/datacontract/engines/ibis/connections/connect.py b/datacontract/engines/ibis/connections/connect.py index 591c37662..811f98d58 100644 --- a/datacontract/engines/ibis/connections/connect.py +++ b/datacontract/engines/ibis/connections/connect.py @@ -276,6 +276,12 @@ def _databricks_connect(ibis, **kwargs): """ from ibis.backends.databricks import Backend + from datacontract.engines.ibis.connections.databricks_patch import apply_databricks_compatibility_patch + + # Databricks-only column types (GEOGRAPHY(4326), …) otherwise fail the whole + # model when ibis reflects the table. + apply_databricks_compatibility_patch() + original_post_connect = Backend._post_connect Backend._post_connect = lambda self, *, memtable_volume=None: None try: diff --git a/datacontract/engines/ibis/connections/databricks_patch.py b/datacontract/engines/ibis/connections/databricks_patch.py new file mode 100644 index 000000000..874f61462 --- /dev/null +++ b/datacontract/engines/ibis/connections/databricks_patch.py @@ -0,0 +1,142 @@ +"""Make ibis's Databricks schema introspection survive Databricks-only column types. + +ibis reflects a Databricks table by reading ``DESCRIBE ... AS JSON`` and mapping +every column through ``DatabricksType.from_string`` (see +``ibis.backends.databricks._databricks_schema_to_ibis``). The whole model is +reflected in one pass, so a single column ibis cannot represent raises out of +``con.table(...)`` and every check of that model fails with the same error — +including checks on unrelated columns. + +Two things are patched here: + +1. ``GEOGRAPHY`` / ``GEOMETRY`` with an SRID. ibis reads the first type + parameter as a geometry *subtype* (the PostGIS spelling + ``GEOGRAPHY(POINT, 4326)``), but Databricks declares only the SRID, + ``GEOGRAPHY(4326)``, so the subtype lookup fails with ``KeyError: '4326'`` + (https://github.com/datacontract/datacontract-cli/issues/1483). A lone + numeric parameter is handed to ibis as the SRID instead. + +2. Anything else ibis still cannot convert becomes ``Unknown`` for that column + alone, with a warning naming it, so the rest of the model is still checked. + This mirrors what the pyspark path already does for Spark types ibis has no + mapping for (``_pyspark_table_unconvertible_as_unknown``). + +Both patches are process-global (ibis looks these up as class/module attributes) +and idempotent. Point 1 can go once a fixed ibis release is available. +""" + +from __future__ import annotations + +import logging + +logger = logging.getLogger(__name__) + +_PATCHED_FLAG = "__datacontract_patched__" + + +def apply_databricks_compatibility_patch() -> None: + """Patch ibis's Databricks type conversion. Safe to call more than once. + + Best-effort: if the backend internals differ from what we expect (e.g. a + future ibis refactor), the patch is skipped and the originals stay in place + rather than breaking the connection. + """ + try: + _patch_geospatial_srid() + _patch_unconvertible_columns() + except Exception: # pragma: no cover - defensive, never block a connection + logger.debug("Could not apply Databricks compatibility patch", exc_info=True) + + +def _patch_geospatial_srid() -> None: + """Read a lone numeric ``GEOGRAPHY`` / ``GEOMETRY`` parameter as the SRID.""" + from ibis.backends.sql.datatypes import DatabricksType + + for geotype in ("GEOGRAPHY", "GEOMETRY"): + original = getattr(DatabricksType, f"_from_sqlglot_{geotype}") + if getattr(original, _PATCHED_FLAG, False): + continue + setattr(DatabricksType, f"_from_sqlglot_{geotype}", _srid_aware(original)) + + +def _srid_aware(original): + """Wrap ibis's ``_from_sqlglot_GEO*`` to accept ``GEO*()``. + + ibis's signature is ``(subtype, srid)``; Databricks passes the SRID alone, so + a first parameter that is a plain number is moved into the ``srid`` slot. A + named subtype (``GEOMETRY(POINT, 4326)``) is left where it is. + """ + # __func__ unwraps the classmethod so the wrapper can pass its own cls. + unbound = original.__func__ + + def from_sqlglot_geo(cls, arg=None, srid=None, nullable=None): + if srid is None and _is_numeric_param(arg): + arg, srid = None, arg + return unbound(cls, arg, srid, nullable=nullable) + + from_sqlglot_geo.__name__ = unbound.__name__ + setattr(from_sqlglot_geo, _PATCHED_FLAG, True) + return classmethod(from_sqlglot_geo) + + +def _is_numeric_param(param) -> bool: + """True for a sqlglot type parameter that is a bare integer, e.g. ``4326``.""" + if param is None: + return False + try: + int(param.this.this) + except (AttributeError, TypeError, ValueError): + return False + return True + + +def _patch_unconvertible_columns() -> None: + """Type a column ibis cannot convert as ``Unknown`` instead of failing the model.""" + from ibis.backends import databricks as databricks_backend + + original = databricks_backend._databricks_schema_to_ibis + if getattr(original, _PATCHED_FLAG, False): + return + databricks_backend._databricks_schema_to_ibis = _tolerant_schema_reader(original) + + +def _tolerant_schema_reader(original): + """Wrap ibis's ``_databricks_schema_to_ibis`` with a per-column fallback.""" + + def databricks_schema_to_ibis(schema): + try: + return original(schema) + except Exception: + logger.debug("Databricks schema conversion failed, retrying column by column", exc_info=True) + return _schema_column_by_column(original, schema) + + setattr(databricks_schema_to_ibis, _PATCHED_FLAG, True) + return databricks_schema_to_ibis + + +def _schema_column_by_column(original, schema): + """Convert each column on its own, typing the unconvertible ones as ``Unknown``.""" + import ibis.expr.datatypes as dt + import ibis.expr.schema as sch + + fields, unknown_columns = {}, [] + for item in schema: + name = item["name"] + try: + fields[name] = original([item])[name] + except Exception: + fields[name] = dt.unknown + unknown_columns.append(f"{name} ({_type_name(item)})") + if unknown_columns: + logger.warning( + f"Column(s) {', '.join(unknown_columns)} have a type ibis cannot represent. " + f"Type checks for these columns will fail." + ) + return sch.Schema(fields) + + +def _type_name(item) -> str: + try: + return str(item["type"]["name"]) + except Exception: + return "unknown type" diff --git a/datacontract/export/sql_type_converter.py b/datacontract/export/sql_type_converter.py index 599f0890e..22747c7b2 100644 --- a/datacontract/export/sql_type_converter.py +++ b/datacontract/export/sql_type_converter.py @@ -442,6 +442,10 @@ def convert_to_databricks(field: Union[SchemaProperty, FieldLike]) -> None | str if base_type is None: return None + if base_type in ["varchar", "char"] and _get_params(field): + # Databricks has VARCHAR(n) / CHAR(n); collapsing them to STRING would + # drop the declared length. + return _attach_params_if_present(base_type.upper(), field) if base_type in ["string", "varchar", "text"]: return "STRING" if base_type in ["timestamp", "timestamp_tz"]: @@ -488,8 +492,11 @@ def convert_to_databricks(field: Union[SchemaProperty, FieldLike]) -> None | str return "ARRAY" if base_type in ["variant"]: return "VARIANT" - if _get_params(field): - return _get_type(field) + # A parameterized type with no mapping (map, geography(4326)) is + # already spelled the way Databricks declares it, so pass it through. + field_type = _get_type(field) + if field_type and ("(" in field_type or "<" in field_type): + return field_type return _warn_cannot_map_type(field, "databricks") diff --git a/datacontract/imports/unity_importer.py b/datacontract/imports/unity_importer.py index 849f075b0..bae8982f4 100644 --- a/datacontract/imports/unity_importer.py +++ b/datacontract/imports/unity_importer.py @@ -175,7 +175,6 @@ def _to_property(column: ColumnInfo) -> SchemaProperty: required=required if required else None, properties=nested_properties, items=items, - custom_properties={"databricksType": sql_type} if sql_type else None, ) diff --git a/docs/docs/reference/databricks.md b/docs/docs/reference/databricks.md index 47977c617..9fd590c82 100644 --- a/docs/docs/reference/databricks.md +++ b/docs/docs/reference/databricks.md @@ -39,7 +39,7 @@ The authentication method is selected from the variables you set, in this order: ### Importing -`datacontract import unity` keeps the full Unity Catalog type text as `physicalType` (e.g. `decimal(10,2)`, `struct`, `map`) — also duplicated in the `databricksType` custom property — and maps to `logicalType`: +`datacontract import unity` keeps the full Unity Catalog type text as `physicalType` (e.g. `decimal(10,2)`, `struct`, `map`) and maps to `logicalType`: | Databricks type | `logicalType` | |---|---| diff --git a/tests/fixtures/databricks-unity/import/datacontract.yaml b/tests/fixtures/databricks-unity/import/datacontract.yaml index ae894b326..7159823fe 100644 --- a/tests/fixtures/databricks-unity/import/datacontract.yaml +++ b/tests/fixtures/databricks-unity/import/datacontract.yaml @@ -19,44 +19,23 @@ schema: properties: - name: id physicalType: int - customProperties: - - property: databricksType - value: int logicalType: integer required: true - name: name physicalType: varchar(255) - customProperties: - - property: databricksType - value: varchar(255) logicalType: string - name: age physicalType: smallint - customProperties: - - property: databricksType - value: smallint logicalType: integer - name: salary physicalType: decimal(10,2) - customProperties: - - property: databricksType - value: decimal(10,2) logicalType: number - name: join_date physicalType: date - customProperties: - - property: databricksType - value: date logicalType: date - name: updated_at physicalType: timestamp - customProperties: - - property: databricksType - value: timestamp logicalType: timestamp - name: is_active physicalType: boolean - customProperties: - - property: databricksType - value: boolean logicalType: boolean \ No newline at end of file diff --git a/tests/fixtures/databricks-unity/import/datacontract_complex_types.yaml b/tests/fixtures/databricks-unity/import/datacontract_complex_types.yaml index 6ba16c9fb..bf5bafc48 100644 --- a/tests/fixtures/databricks-unity/import/datacontract_complex_types.yaml +++ b/tests/fixtures/databricks-unity/import/datacontract_complex_types.yaml @@ -18,15 +18,9 @@ schema: properties: - name: id physicalType: bigint - customProperties: - - property: databricksType - value: bigint logicalType: integer - name: id_list physicalType: array - customProperties: - - property: databricksType - value: array logicalType: array items: name: items @@ -34,9 +28,6 @@ schema: logicalType: integer - name: id_struct physicalType: struct - customProperties: - - property: databricksType - value: struct logicalType: object properties: - name: value @@ -44,9 +35,6 @@ schema: logicalType: integer - name: struct_list physicalType: array> - customProperties: - - property: databricksType - value: array> logicalType: array items: name: items @@ -61,6 +49,3 @@ schema: logicalType: integer - name: attributes physicalType: map - customProperties: - - property: databricksType - value: map diff --git a/tests/test_sql_type_converter_physicaltype.py b/tests/test_sql_type_converter_physicaltype.py index f424861dd..29c5075a1 100644 --- a/tests/test_sql_type_converter_physicaltype.py +++ b/tests/test_sql_type_converter_physicaltype.py @@ -205,12 +205,30 @@ def test_decimal_10_2_bigquery(): def test_varchar_100_databricks(): - """VARCHAR(100) on databricks: 'varchar' -> 'STRING'.""" + """VARCHAR(100) on databricks: 'varchar' accepts params -> 'VARCHAR(100)'. + + Databricks has VARCHAR(n) and enforces the length; collapsing it to STRING + would drop the length the contract declares. + """ field = SchemaProperty(name="col", physicalType="VARCHAR(100)") result = convert_to_sql_type(field, "databricks") + assert result == "VARCHAR(100)" + + +def test_varchar_without_length_databricks(): + """A VARCHAR with no length has nothing to preserve -> 'STRING'.""" + field = SchemaProperty(name="col", physicalType="VARCHAR") + result = convert_to_sql_type(field, "databricks") assert result == "STRING" +def test_map_databricks(): + """MAP has no databricks mapping, but the declared type is already Databricks SQL.""" + field = SchemaProperty(name="col", physicalType="map") + result = convert_to_sql_type(field, "databricks") + assert result == "map" + + def test_decimal_18_4_trino(): """DECIMAL(18,4) on trino: 'decimal' -> 'decimal', accepts params -> 'decimal(18,4)'.""" field = SchemaProperty(name="col", physicalType="DECIMAL(18,4)") diff --git a/tests/test_test_databricks.py b/tests/test_test_databricks.py index 18a06bd89..01a303105 100644 --- a/tests/test_test_databricks.py +++ b/tests/test_test_databricks.py @@ -38,6 +38,97 @@ class StubIbis: assert Backend._post_connect is original +def test_connect_applies_the_type_compatibility_patch(monkeypatch): + # Databricks-only column types (GEOGRAPHY(4326)) break ibis's schema + # reflection, which fails every check of the model, so the patch must be in + # place before the first table is read. + from datacontract.engines.ibis.connections.connect import _databricks_connect + + applied = [] + monkeypatch.setattr( + "datacontract.engines.ibis.connections.databricks_patch.apply_databricks_compatibility_patch", + lambda: applied.append(True), + ) + + class StubIbis: + class databricks: + @staticmethod + def connect(**kwargs): + return "connection" + + assert _databricks_connect(StubIbis(), server_hostname="example") == "connection" + assert applied == [True] + + +@pytest.fixture +def databricks_type_patch(): + """Apply the Databricks type patch, restoring ibis's originals afterwards.""" + from ibis.backends import databricks as databricks_backend + from ibis.backends.sql.datatypes import DatabricksType + + from datacontract.engines.ibis.connections.databricks_patch import apply_databricks_compatibility_patch + + geo_methods = ("_from_sqlglot_GEOGRAPHY", "_from_sqlglot_GEOMETRY") + # not defined on DatabricksType itself, so None means "restore by removing" + originals = {name: DatabricksType.__dict__.get(name) for name in geo_methods} + original_schema_reader = databricks_backend._databricks_schema_to_ibis + + apply_databricks_compatibility_patch() + try: + yield + finally: + for name, method in originals.items(): + if method is None: + delattr(DatabricksType, name) + else: + setattr(DatabricksType, name, method) + databricks_backend._databricks_schema_to_ibis = original_schema_reader + + +def test_geospatial_type_with_srid(databricks_type_patch): + # Databricks declares a geospatial column as GEOGRAPHY(), while ibis + # reads the first type parameter as a geometry subtype (PostGIS's + # GEOGRAPHY(POINT, 4326)) and fails with KeyError: '4326'. + from ibis.backends.sql.datatypes import DatabricksType + + geography = DatabricksType.from_string("geography(4326)") + assert geography.geotype == "geography" + assert geography.srid == 4326 + + geometry = DatabricksType.from_string("geometry(4326)") + assert geometry.geotype == "geometry" + assert geometry.srid == 4326 + + +def test_geospatial_type_without_srid_and_with_subtype(databricks_type_patch): + # The subtype spellings ibis already understood must keep working. + import ibis.expr.datatypes as dt + from ibis.backends.sql.datatypes import DatabricksType + + assert DatabricksType.from_string("geography").srid is None + assert DatabricksType.from_string("geometry(point,4326)") == dt.Point(geotype="geometry", srid=4326) + assert DatabricksType.from_string("int") == dt.int32 + + +def test_unconvertible_column_does_not_affect_the_other_columns(databricks_type_patch): + # One column ibis cannot represent must not fail the whole model: it becomes + # unknown (its type checks fail), every other column keeps its real type. + import ibis.expr.datatypes as dt + from ibis.backends import databricks as databricks_backend + + schema = databricks_backend._databricks_schema_to_ibis( + [ + {"name": "id", "type": {"name": "int"}, "nullable": False}, + {"name": "geo", "type": {"name": "geography(4326)"}, "nullable": True}, + {"name": "mystery", "type": {"name": "geography(OGC:CRS84)"}, "nullable": True}, + ] + ) + + assert schema["id"] == dt.Int32(nullable=False) + assert schema["geo"].srid == 4326 + assert schema["mystery"] == dt.unknown + + @pytest.mark.skipif( os.environ.get("DATACONTRACT_DATABRICKS_TOKEN") is None, reason="Requires DATACONTRACT_DATABRICKS_TOKEN to be set" ) From 88d8b5138513e0e2d0b06899ffd6520a1f21576b Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 5 Aug 2026 18:30:47 +0200 Subject: [PATCH 2/2] Regenerate the docs release notes from CHANGELOG.md --- docs/docs/release-notes.md | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/docs/docs/release-notes.md b/docs/docs/release-notes.md index c786cce7e..4f4646002 100644 --- a/docs/docs/release-notes.md +++ b/docs/docs/release-notes.md @@ -26,6 +26,13 @@ marked as such in the entry. ## Unreleased {#unreleased} +### Changed +- `datacontract import unity` no longer writes the `databricksType` custom property, which duplicated `physicalType` +- `datacontract export sql --server databricks` keeps the declared length of `varchar(n)` and `char(n)` instead of exporting `STRING` + +### Fixed +- `datacontract test` for Databricks no longer fails all checks of a model with a `GEOGRAPHY` or `GEOMETRY` column ([#1483](https://github.com/datacontract/datacontract-cli/issues/1483)) + ## 1.1.0 — 2026-08-04 {#v1-1-0} This release drops the pyspark compile-time dependency. The server types `dataframe` and `databricks` still work with a provided Spark session.