Databricks finetuning

CHANGES.md

@@ -1,3 +1,8 @@
+# 2026-07-30 (9.12.3)
+
+* Improve Databricks schema generation by setting identifier and display from the first field and inferring `mainGeometry` from the first geo field.
+* Represent Databricks date, time, and timestamp columns as strings with the matching schema formats.
+
 # 2026-07-30 (9.12.2)
 
 * Bufix table filter on dataset instead of name for expired schemas.

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "amsterdam-schema-tools"
-version = "9.12.2"
+version = "9.12.3"
 description = "Tools to work with Amsterdam Schema."
 readme = "README.md"
 requires-python = ">=3.10"

src/schematools/contrib/databricks/client.py

@@ -99,6 +99,7 @@ def get_databricks_info(catalog: str, schema: str, table_name: str) -> Databrick
     client = WorkspaceClient()
     table_rows = _execute_sql(client, TABLE_DATA_SQL, parameters=parameters)
     column_rows = _execute_sql(client, COLUMN_DATA_SQL, parameters=parameters)
+    first_field_name = column_rows[0][0] if column_rows and column_rows[0] else None
     table_data = TableData.from_row(table_rows[0]) if table_rows else None
     column_data = {
         column.name: column for column in (ColumnData.from_row(row) for row in column_rows or [])
@@ -109,4 +110,5 @@ def get_databricks_info(catalog: str, schema: str, table_name: str) -> Databrick
         table_name=table_name,
         table_data=table_data,
         column_data=column_data,
+        first_field_name=first_field_name,
     )

src/schematools/contrib/databricks/types.py

@@ -215,13 +215,16 @@ def semver(val: str):
     "int": "integer",
     "bigint": "integer",
     "smallint": "integer",
-    "timestamp": "datetime",
-    "date": "date",
+    "timestamp": "string",
+    "date": "string",
+    "time": "string",
     "boolean": "boolean",
     "double": "number",
     "float": "number",
 }
 
+SCHEMA_FORMAT = {"timestamp": "date-time", "date": "date", "time": "time"}
+
 BASE_TABLE_SCHEMA: dict = {
     "type": "table",
     "version": "1.0.0",
@@ -332,6 +335,8 @@ class DatabricksInfo:
     table_name: str
     table_data: TableData | None
     column_data: dict[str, ColumnData]
+    first_field_name: str | None = None
+    geo_field: str | None = None
     errors: list[str] = field(default_factory=list)
 
     def _collect_spec_errors(self, tags: Tags, specs: dict[str, AttributeSpec]) -> list[str]:
@@ -353,6 +358,9 @@ def _apply_tag_specs(self, target: dict, tags: Tags, specs: dict[str, AttributeS
                 continue
             if attr == "$ref":
                 target.pop("type", None)  # Remove 'type' if '$ref' is present
+                if self.geo_field is None:
+                    # set geo_field to first geo field encountered
+                    self.geo_field = target.get("title")
             target[attr] = spec.transform(value)
 
     def _validate_table_tags(self) -> list[str]:
@@ -389,8 +397,11 @@ def _build_column_schema(self, column: ColumnData) -> dict:
         column_schema = {
             "title": column.name,
             "type": SCHEMA_TYPES.get(column.type_name),
-            "description": column.comment,
         }
+        if column.comment:
+            column_schema["description"] = column.comment
+        if format := SCHEMA_FORMAT.get(column.type_name):
+            column_schema["format"] = format
         self._apply_tag_specs(column_schema, column.tags, COLUMN_ATTRIBUTES)
         return column_schema
 
@@ -407,8 +418,11 @@ def table_id(self) -> str:
         return toCamelCase(table_tags["id"] or self.table_name)
 
     def get_base_schema(self) -> dict:
-        schema = {"id": self.table_id}
+        schema: dict[str, Any] = {"id": self.table_id}
         schema.update(deepcopy(BASE_TABLE_SCHEMA))
+        if self.first_field_name is not None:
+            schema["schema"]["identifier"] = toCamelCase(self.first_field_name)
+            schema["schema"]["display"] = toCamelCase(self.first_field_name)
         return schema
 
     @cached_property
@@ -421,6 +435,8 @@ def dict(self) -> dict:
             schema["schema"]["properties"][toCamelCase(column.name)] = self._build_column_schema(
                 column
             )
+        if self.geo_field is not None and "mainGeometry" not in schema["schema"]:
+            schema["schema"]["mainGeometry"] = toCamelCase(self.geo_field)
         return schema
 
     @cached_property

tests/test_databricks_types.py

@@ -7,6 +7,7 @@
 import pytest
 from databricks.sdk.service.catalog import EntityTagAssignment
 
+from schematools.contrib.databricks import client as databricks_client
 from schematools.contrib.databricks.types import (
     ColumnData,
     DatabricksInfo,
@@ -96,6 +97,102 @@ def test_databricks_info_validates_and_renders_table_json() -> None:
     assert payload["schema"]["properties"]["name"]["minLength"] == 2
 
 
+def test_databricks_info_sets_temporal_formats_and_defaults_from_first_field() -> None:
+    info = DatabricksInfo(
+        catalog="main",
+        schema="default",
+        table_name="events_table",
+        table_data=TableData(comment=None, tags=Tags(_tags=[])),
+        column_data={
+            "event_id": ColumnData("event_id", "string", None, None, "", Tags(_tags=[])),
+            "created_at": ColumnData("created_at", "timestamp", None, None, "", Tags(_tags=[])),
+            "event_date": ColumnData("event_date", "date", None, None, "", Tags(_tags=[])),
+            "event_time": ColumnData("event_time", "time", None, None, "", Tags(_tags=[])),
+        },
+        first_field_name="event_id",
+    )
+
+    payload = json.loads(info.json)
+
+    assert payload["schema"]["identifier"] == "eventId"
+    assert payload["schema"]["display"] == "eventId"
+    assert payload["schema"]["properties"]["createdAt"] == {
+        "title": "created_at",
+        "type": "string",
+        "format": "date-time",
+    }
+    assert payload["schema"]["properties"]["eventDate"] == {
+        "title": "event_date",
+        "type": "string",
+        "format": "date",
+    }
+    assert payload["schema"]["properties"]["eventTime"] == {
+        "title": "event_time",
+        "type": "string",
+        "format": "time",
+    }
+
+
+def test_databricks_info_sets_main_geometry_from_first_geo_ref_column() -> None:
+    info = DatabricksInfo(
+        catalog="main",
+        schema="default",
+        table_name="buildings_table",
+        table_data=TableData(comment=None, tags=Tags(_tags=[])),
+        column_data={
+            "shape": ColumnData(
+                "shape",
+                "string",
+                None,
+                None,
+                "",
+                Tags(_tags=[Tag(key="$ref", value="Point", type="columns")]),
+            ),
+            "secondary_shape": ColumnData(
+                "secondary_shape",
+                "string",
+                None,
+                None,
+                "",
+                Tags(_tags=[Tag(key="$ref", value="Polygon", type="columns")]),
+            ),
+        },
+    )
+
+    payload = json.loads(info.json)
+
+    assert payload["schema"]["mainGeometry"] == "shape"
+    assert (
+        payload["schema"]["properties"]["shape"]["$ref"] == "https://geojson.org/schema/Point.json"
+    )
+    assert "type" not in payload["schema"]["properties"]["shape"]
+
+
+def test_get_databricks_info_uses_first_column_name_for_defaults(monkeypatch) -> None:
+    table_rows = [("Table comment", json.dumps([]))]
+    column_rows = [
+        ("event_id", "string", True, None, None, json.dumps([])),
+        ("created_at", "timestamp", True, None, None, json.dumps([])),
+    ]
+
+    def fake_execute_sql(_client, sql_statement: str, parameters=None):
+        if sql_statement == databricks_client.TABLE_DATA_SQL:
+            return table_rows
+        if sql_statement == databricks_client.COLUMN_DATA_SQL:
+            return column_rows
+        raise AssertionError(f"Unexpected SQL statement: {sql_statement}")
+
+    monkeypatch.setattr(databricks_client, "DATABRICKS_WAREHOUSE_ID", "warehouse-id")
+    monkeypatch.setattr(databricks_client, "WorkspaceClient", lambda: object())
+    monkeypatch.setattr(databricks_client, "_execute_sql", fake_execute_sql)
+
+    info = databricks_client.get_databricks_info("main", "default", "events_table")
+
+    assert info.first_field_name == "event_id"
+    assert info.get_base_schema()["schema"]["identifier"] == "eventId"
+    assert info.get_base_schema()["schema"]["display"] == "eventId"
+
+
 def test_databricks_info_reports_explicit_none_values() -> None:
     table_tags = Tags(_tags=[Tag(key="auth", value=None, type="tables")])