@@ -1,6 +1,7 @@
 # Release History
 
 # Unreleased
+- Add the kernel-only `enable_geospatial_support` connection option (default `False`). GEOMETRY / GEOGRAPHY results are exposed as `{"srid": int, "wkb": bytes}` values when true or WKT / EWKT strings when false.
 - Add the `escape_catalog_name` connection parameter (default `False`) to opt in to escaping unescaped `_` in metadata catalog names on Thrift and kernel connections. Existing escapes and `%` wildcards are preserved.
 
 # 4.6.0 (2026-09-24)
 
@@ -153,6 +153,7 @@ to change without notice.
 | `max_download_threads`                | `int`  |   ✅   |   ❌   | `10`          | Worker threads for cloud-fetch downloads. Not forwarded to the kernel.                                        |
 | `enable_query_result_lz4_compression` | `bool` |   ✅   |   ❌   | `True`        | LZ4-compress result payloads. Not forwarded; the kernel handles compression internally.                       |
 | `_disable_pandas`                     | `bool` |   ✅   |   ✅   | `False`       | Skip the pandas-based Arrow→row deserialization and materialize rows directly with PyArrow. This is a **Python-side** result-conversion toggle, not a wire option: the kernel returns results as Arrow (`RecordBatch`es) and the connector runs the *same* `_convert_arrow_table` for both backends, so the flag is honored on the kernel path too. Affects only row fetches (`fetchone`/`fetchmany`/`fetchall`); the `fetch*_arrow` methods return the Arrow table unchanged regardless of this flag. |
+| `enable_geospatial_support`           | `bool` |   ❌   |   ✅   | `False` | Return GEOMETRY / GEOGRAPHY as `{"srid": int, "wkb": bytes}` values when `True`, or as WKT / EWKT strings when `False`. This is a local result conversion and is never forwarded to SEA. |
 | `_use_arrow_native_complex_types`     | `bool` |   ✅   |   ✅   | `True`        | Return `ARRAY`/`MAP`/`STRUCT` as native Arrow types instead of JSON strings. Forwarded to the kernel.         |
 | `_use_arrow_native_decimals`          | `bool` |   ✅   |   ❌   | `True`        | Thrift wire encoding for `DECIMAL`: `True` → native Arrow `decimal128`, `False` → Arrow string. **No value-level effect**, though: the connector unconditionally re-casts the column back to `decimal128` (`convert_decimals_in_arrow_table`, `thrift_backend.py`), so both `fetchall()` and `fetchall_arrow()` yield `Decimal` / `decimal128(p,s)` either way (verified live). Not forwarded to the kernel, which always returns native Arrow decimals. |
 | `_use_arrow_native_timestamps`        | `bool` |   ✅   |   ❌   | `True`        | Thrift wire encoding for `TIMESTAMP`: `True` → native Arrow timestamp (→ Python `datetime`), `False` → Arrow string (→ Python **`str`**). **Unlike decimals there is no re-cast**, so `False` genuinely surfaces strings — and `cursor.description` still reports the type code as `'timestamp'`, a mismatch to watch for (verified live). Note the connector always also sends the `spark.thriftserver.arrowBasedRowSet.timestampAsString=false` conf, but the `timestampAsArrow=False` flag wins. Not forwarded to the kernel, which always returns native Arrow timestamps. |
 
@@ -1 +1 @@
-ab1e6a707e4a15a3a7a8c78d8fdd12c752c51102
+9c2e2e6182fc7285bcfff39b2177941bc804897c
@@ -170,6 +170,26 @@ def _kernel_session_accepts_kwarg(name: str) -> bool:
     return name in params
 
 
+def _kernel_geospatial_kwargs(value: bool) -> Dict[str, bool]:
+    """Build the explicit native geospatial-support kwarg.
+
+    The value is always passed explicitly so the driver and kernel contracts
+    cannot drift. Older kernel wheels do not declare
+    ``enable_geospatial_support`` and must fail clearly rather than silently
+    return a different public value shape.
+    """
+    if not isinstance(value, bool):
+        raise ValueError(
+            "enable_geospatial_support must be a bool; " f"got {type(value).__name__}"
+        )
+    if not _kernel_session_accepts_kwarg("enable_geospatial_support"):
+        raise NotSupportedError(
+            "enable_geospatial_support requires a newer databricks-sql-kernel "
+            "wheel that exposes geospatial result representation support."
+        )
+    return {"enable_geospatial_support": value}
+
+
 def _kernel_telemetry_kwargs(options: Dict[str, Any]) -> Dict[str, Any]:
     """Build phase-7 telemetry/system kwargs for ``databricks_sql_kernel.Session``.
 
@@ -263,6 +283,12 @@ def __init__(
         # The kernel binding owns type and range validation.
         self._request_timeout_secs = kwargs.get("request_timeout_secs")
         self._max_connections = kwargs.get("max_connections")
+        # Opt-in native GEOMETRY / GEOGRAPHY support. True requests
+        # the canonical Arrow struct and surfaces as ``{"srid": int, "wkb":
+        # bytes}`` through pyarrow; False requests WKT / EWKT strings. This is
+        # intentionally separate from ``session_configuration``: it is never
+        # forwarded to SEA.
+        self._enable_geospatial_support = kwargs.get("enable_geospatial_support", False)
         # Kernel telemetry phase 7 adds binding/runtime identity and
         # telemetry config kwargs directly to ``databricks_sql_kernel.Session``.
         self._telemetry_options = kwargs.get("telemetry_options") or {}
@@ -381,6 +407,9 @@ def open_session(
             # kernel's ``retry_*`` kwargs. Empty when at defaults.
             retry_kwargs = _kernel_retry_kwargs(self._retry_options)
             telemetry_kwargs = _kernel_telemetry_kwargs(self._telemetry_options)
+            geospatial_kwargs = _kernel_geospatial_kwargs(
+                self._enable_geospatial_support
+            )
             max_connections_kwargs: Dict[str, Any] = {}
             if _kernel_session_accepts_kwarg("max_connections"):
                 max_connections_kwargs["max_connections"] = self._max_connections
@@ -428,6 +457,7 @@ def open_session(
                 **tls_kwargs,
                 **retry_kwargs,
                 **telemetry_kwargs,
+                **geospatial_kwargs,
                 **max_connections_kwargs,
                 **http_headers_kwargs,
             )
@@ -1015,11 +1045,9 @@ def get_tables(
         if self._kernel_session is None:
             raise InterfaceError("get_tables requires an open session.")
         try:
-            # ``table_types`` is filtered kernel-side (the kernel applies
-            # it to the reshaped result, case-insensitively as of the
-            # batch-3 kernel change), so we forward it and let the kernel
-            # do the work — no connector-side drain + refilter. Passing it
-            # through preserves streaming for large schemas.
+            # The kernel filters ``table_types`` with case-sensitive
+            # matching for Thrift parity. Forward the values unchanged
+            # to preserve those semantics and streaming for large schemas.
             stream = self._kernel_session.metadata().list_tables(
                 catalog=escape_metadata_catalog_name(
                     _catalog_or_none(catalog_name), self._escape_catalog_name
 
@@ -3,6 +3,7 @@
 import errno
 import logging
 import math
+import re
 import time
 import threading
 from typing import Dict, List, Optional, Union, Any, TYPE_CHECKING
@@ -807,38 +808,50 @@ def _col_to_description(col, field=None, host_url=None):
         else:
             precision, scale = None, None
 
-        # Extract variant type from field if available
+        # Extract logical types that Thrift surfaces as strings from Arrow metadata.
         if field is not None:
             try:
-                # Check for variant type in metadata
                 if field.metadata and b"Spark:DataType:SqlName" in field.metadata:
                     sql_type = field.metadata.get(b"Spark:DataType:SqlName")
-                    if sql_type == b"VARIANT":
-                        cleaned_type = "variant"
+                    logical_type = re.fullmatch(
+                        rb"(variant|geometry|geography)(?:\([^()]*\))?",
+                        sql_type.strip(),
+                        re.IGNORECASE,
+                    )
+                    if logical_type:
+                        cleaned_type = logical_type.group(1).decode("ascii").lower()
             except Exception as e:
-                logger.debug(f"Could not extract variant type from field: {e}")
+                logger.debug(f"Could not extract logical type from field: {e}")
 
         return col.columnName, cleaned_type, None, None, precision, scale, None
 
     @staticmethod
     def _hive_schema_to_description(t_table_schema, schema_bytes=None, host_url=None):
+        arrow_fields = []
         field_dict = {}
         if pyarrow and schema_bytes:
             try:
                 arrow_schema = pyarrow.ipc.read_schema(pyarrow.py_buffer(schema_bytes))
-                # Build a dictionary mapping column names to fields
-                for field in arrow_schema:
+                arrow_fields = list(arrow_schema)
+                # Retain a name fallback for malformed/partial schemas, but use
+                # result-column position whenever it is available. SQL permits
+                # duplicate aliases, for which a name lookup is ambiguous.
+                for field in arrow_fields:
                     field_dict[field.name] = field
             except Exception as e:
                 logger.debug(f"Could not parse arrow schema: {e}")
 
         return [
             ThriftDatabricksClient._col_to_description(
                 col,
-                field_dict.get(col.columnName) if field_dict else None,
+                (
+                    arrow_fields[index]
+                    if index < len(arrow_fields)
+                    else field_dict.get(col.columnName)
+                ),
                 host_url,
             )
-            for col in t_table_schema.columns
+            for index, col in enumerate(t_table_schema.columns)
         ]
 
     def _results_message_to_execute_response(
 
@@ -194,6 +194,13 @@ def __init__(
                 decision. This is an intentional divergence from the
                 Thrift/SEA paths, where an explicit ``True`` can still
                 be suppressed by the feature flag.
+            :param enable_geospatial_support: `bool`, optional (default is False)
+                Kernel backend only. Controls the public representation of
+                ``GEOMETRY`` and ``GEOGRAPHY`` result values. ``True`` returns
+                ``{"srid": int, "wkb": bytes}``; ``False`` returns WKT / EWKT
+                strings (for example ``"SRID=4326;POINT(1 2)"``). The
+                conversion is local to the kernel/driver and this option is
+                never sent to the SQL Execution API.
             :param use_hybrid_disposition: `bool`, optional (default is False)
                 Use the hybrid disposition instead of the inline disposition.
             :param server_hostname: Databricks instance host name.
 
@@ -332,6 +332,9 @@ def _create_backend(
                 retry_options=kernel_retry_options,
                 request_timeout_secs=kwargs.get("_socket_timeout"),
                 max_connections=kwargs.get("_pool_maxsize") or None,
+                enable_geospatial_support=kwargs.get(
+                    "enable_geospatial_support", False
+                ),
                 telemetry_options=kernel_telemetry_options,
                 escape_catalog_name=kwargs.get("escape_catalog_name", False),
             )
 
@@ -383,12 +383,12 @@ def test_columns_with_empty_string_filter_matches_nothing(conn, empty_filter):
         assert cur.fetchall() == []
 
 
-def test_tables_table_types_filter_is_case_insensitive(conn):
-    """Lowercase ``table_types=['view']`` / uppercase ``['TABLE']``
-    each match the right object regardless of case — locks the
-    kernel-side case-insensitive ``table_types`` match (batch-3 kernel
-    B2) end-to-end plus the connector drain removal (the filter now
-    runs kernel-side, not client-side).
+def test_tables_table_types_filter_is_case_sensitive(conn):
+    """Table types match exact case, consistent with Thrift.
+
+    Uppercase ``['VIEW']`` / ``['TABLE']`` match the corresponding
+    object; lowercase and mixed-case values match no objects. The
+    kernel applies the filter to values forwarded by the connector.
 
     Self-contained: creates a table + a view over it in the session's
     default (writable) schema, scopes the lookup to a unique name
@@ -411,12 +411,12 @@ def _names_and_types():
                 ni, ti = cols.index("TABLE_NAME"), cols.index("TABLE_TYPE")
                 return {(r[ni], r[ti]) for r in rows}
 
-            # Lowercase 'view' must match the VIEW (and only it).
+            # Uppercase 'VIEW' must match the VIEW (and only it).
             cur.tables(
                 catalog_name=cat,
                 schema_name=sch,
                 table_name=name_pat,
-                table_types=["view"],
+                table_types=["VIEW"],
             )
             assert _names_and_types() == {(vw, "VIEW")}
 
@@ -428,6 +428,16 @@ def _names_and_types():
                 table_types=["TABLE"],
             )
             assert _names_and_types() == {(tbl, "TABLE")}
+
+            # Lowercase and mixed-case requests must not match either type.
+            for table_type in ("view", "View", "table", "Table"):
+                cur.tables(
+                    catalog_name=cat,
+                    schema_name=sch,
+                    table_name=name_pat,
+                    table_types=[table_type],
+                )
+                assert _names_and_types() == set(), table_type
         finally:
             cur.execute(f"DROP VIEW IF EXISTS {vw}")
             cur.execute(f"DROP TABLE IF EXISTS {tbl}")
 
@@ -392,6 +392,94 @@ def fake_session(**kw):
     assert captured["max_connections"] == max_connections
 
 
+@pytest.mark.parametrize("enabled", [True, False])
+def test_open_session_passes_geospatial_representation_to_kernel(
+    monkeypatch, enabled
+):
+    captured = {}
+
+    def fake_session(*, enable_geospatial_support=True, **kw):
+        captured["enable_geospatial_support"] = enable_geospatial_support
+        sess = MagicMock()
+        sess.session_id = "sess-id"
+        return sess
+
+    monkeypatch.setattr(kernel_client._kernel, "Session", fake_session)
+    c = kernel_client.KernelDatabricksClient(
+        server_hostname="example.cloud.databricks.com",
+        http_path="/sql/1.0/warehouses/abc",
+        auth_provider=AccessTokenAuthProvider("dapi-test"),
+        ssl_options=None,
+        enable_geospatial_support=enabled,
+    )
+
+    c.open_session(session_configuration=None, catalog=None, schema=None)
+
+    assert captured["enable_geospatial_support"] is enabled
+
+
+def test_open_session_disables_native_geospatial_support_by_default(monkeypatch):
+    captured = {}
+
+    def fake_session(**kw):
+        captured.update(kw)
+        sess = MagicMock()
+        sess.session_id = "sess-id"
+        return sess
+
+    monkeypatch.setattr(kernel_client._kernel, "Session", fake_session)
+    c = kernel_client.KernelDatabricksClient(
+        server_hostname="example.cloud.databricks.com",
+        http_path="/sql/1.0/warehouses/abc",
+        auth_provider=AccessTokenAuthProvider("dapi-test"),
+        ssl_options=None,
+    )
+
+    c.open_session(session_configuration=None, catalog=None, schema=None)
+
+    assert captured["enable_geospatial_support"] is False
+
+
+def test_open_session_rejects_explicit_geospatial_option_with_old_kernel(
+    monkeypatch,
+):
+    def fake_session_without_geospatial(
+        host,
+        http_path,
+        *,
+        catalog=None,
+        schema=None,
+        session_conf=None,
+        complex_types_as_json=False,
+        intervals_as_string=False,
+        request_timeout_secs=None,
+        auth_type=None,
+        access_token=None,
+    ):
+        sess = MagicMock()
+        sess.session_id = "sess-id"
+        return sess
+
+    monkeypatch.setattr(
+        kernel_client._kernel, "Session", fake_session_without_geospatial
+    )
+    c = kernel_client.KernelDatabricksClient(
+        server_hostname="example.cloud.databricks.com",
+        http_path="/sql/1.0/warehouses/abc",
+        auth_provider=AccessTokenAuthProvider("dapi-test"),
+        ssl_options=None,
+        enable_geospatial_support=False,
+    )
+
+    with pytest.raises(NotSupportedError, match="newer databricks-sql-kernel"):
+        c.open_session(session_configuration=None, catalog=None, schema=None)
+
+
+def test_geospatial_option_rejects_non_bool():
+    with pytest.raises(ValueError, match="must be a bool"):
+        kernel_client._kernel_geospatial_kwargs("false")
+
+
 def test_open_session_passes_phase_7_telemetry_kwargs_to_kernel(monkeypatch):
     """Kernel telemetry phase 7 added binding/runtime identity and
     telemetry config kwargs to ``databricks_sql_kernel.Session``."""
@@ -490,6 +578,7 @@ def fake_session_without_optional_kwargs(
         catalog=None,
         schema=None,
         session_conf=None,
+        enable_geospatial_support=True,
         complex_types_as_json=False,
         intervals_as_string=False,
         request_timeout_secs=None,
@@ -589,7 +678,9 @@ def raise_value_error(_obj):
     kwargs = kernel_client._kernel_telemetry_kwargs(
         {"enable_telemetry": True, "telemetry_batch_size": 17}
     )
-    assert kwargs == {}, f"expected no phase-7 kwargs when signature unreadable, got {kwargs}"
+    assert (
+        kwargs == {}
+    ), f"expected no phase-7 kwargs when signature unreadable, got {kwargs}"
 
 
 def test_execute_command_forwards_parameters_to_bind_param():
@@ -1535,7 +1626,7 @@ def _make_tables_stream() -> MagicMock:
 
 def test_get_tables_forwards_table_types_to_kernel():
     """``table_types`` is forwarded verbatim to the kernel's
-    ``list_tables`` (which filters case-insensitively) — the
+    ``list_tables`` (which filters case-sensitively) — the
     connector no longer drains + refilters client-side. The stream
     flows back through the normal ``KernelResultSet`` path unchanged."""
     c = _make_client()
Original file line number	Diff line number	Diff line change
`@@ -1 +1 @@`
`1`		`-ab1e6a707e4a15a3a7a8c78d8fdd12c752c51102`
	`1`	`+9c2e2e6182fc7285bcfff39b2177941bc804897c`