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/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. 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" )