fix(sqlalchemy-spanner): add TOKENLIST type and unknown type fallback in introspection

Spanner full-text search uses TOKENLIST columns, but the dialect does
not recognize this type — _type_map has no entry and _designate_type()
crashes with KeyError on any unrecognized type string.

This change:
1. Adds a first-class TOKENLIST type (TypeEngine subclass) with
   forward mapping (_type_map), inverse mapping (_type_map_inv), and
   DDL compilation (SpannerTypeCompiler.visit_TOKENLIST). This ensures
   schema introspection roundtrips correctly through Alembic and
   other DDL tools, and allows TOKENLIST columns to be referenced
   in SEARCH()/SCORE()/SNIPPET() queries via SQLAlchemy.
2. Adds a fallback in _designate_type() so unrecognized types return
   NullType with a warning instead of crashing. This future-proofs
   against new Spanner types.

The Go client had the same gap and fixed it in
googleapis/google-cloud-go#11522 (released in spanner v1.78.0),
which added TOKENLIST as a first-class TypeBase.

Fixes #16621

Author: hochoy

packages/sqlalchemy-spanner/google/cloud/sqlalchemy_spanner/sqlalchemy_spanner.py

@@ -12,6 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 import base64
+import warnings
 
 import re
 
@@ -107,6 +108,21 @@ def process(value):
         return process
 
 
+class TOKENLIST(types.TypeEngine):
+    """Spanner TOKENLIST type for full-text search indexes.
+
+    TOKENLIST columns store tokenized text produced by functions like
+    TOKENIZE_FULLTEXT() and back SEARCH INDEX structures. They are always
+    generated, always HIDDEN, and cannot be read or written directly by
+    applications.
+
+    This type exists so that schema introspection can roundtrip correctly
+    (reflect → DDL generation) without losing type information.
+    """
+
+    __visit_name__ = "TOKENLIST"
+
+
 # Spanner-to-SQLAlchemy types map
 _type_map = {
     "BOOL": types.Boolean,
@@ -122,6 +138,7 @@ def process(value):
     "TIMESTAMP": types.TIMESTAMP,
     "ARRAY": types.ARRAY,
     "JSON": types.JSON,
+    "TOKENLIST": TOKENLIST,
 }
 
 
@@ -140,6 +157,7 @@ def process(value):
     types.TIMESTAMP: "TIMESTAMP",
     types.Integer: "INT64",
     types.NullType: "INT64",
+    TOKENLIST: "TOKENLIST",
 }
 
 _compound_keywords = {
@@ -819,6 +837,9 @@ def visit_BIGINT(self, type_, **kw):
     def visit_JSON(self, type_, **kw):
         return "JSON"
 
+    def visit_TOKENLIST(self, type_, **kw):
+        return "TOKENLIST"
+
 
 class SpannerDialect(DefaultDialect):
     """Cloud Spanner dialect.
@@ -1226,7 +1247,14 @@ def _designate_type(self, str_repr):
             inner_type = self._designate_type(inner_type_str)
             return _type_map["ARRAY"](inner_type)
         else:
-            return _type_map[str_repr]
+            try:
+                return _type_map[str_repr]
+            except KeyError:
+                warnings.warn(
+                    "Did not recognize Spanner type '%s', "
+                    "mapping it to NullType" % str_repr
+                )
+                return types.NullType()
 
     @engine_to_connection
     def get_multi_indexes(

packages/sqlalchemy-spanner/tests/unit/test_types.py

@@ -0,0 +1,82 @@
+# Copyright 2026 Google LLC
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+import warnings
+
+from sqlalchemy import types
+from sqlalchemy.testing.plugin.plugin_base import fixtures
+
+from google.cloud.sqlalchemy_spanner import sqlalchemy_spanner
+
+
+class TestDesignateType(fixtures.TestBase):
+    """Unit tests for SpannerDialect._designate_type."""
+
+    def setup_method(self):
+        self.dialect = sqlalchemy_spanner.SpannerDialect()
+
+    def test_known_types(self):
+        assert isinstance(self.dialect._designate_type("BOOL"), types.Boolean)
+        assert isinstance(self.dialect._designate_type("INT64"), types.BIGINT)
+        assert isinstance(self.dialect._designate_type("FLOAT64"), types.Float)
+        assert isinstance(self.dialect._designate_type("DATE"), types.DATE)
+        assert isinstance(self.dialect._designate_type("TIMESTAMP"), types.TIMESTAMP)
+        assert isinstance(self.dialect._designate_type("JSON"), types.JSON)
+
+    def test_string_with_length(self):
+        result = self.dialect._designate_type("STRING(255)")
+        assert isinstance(result, types.String)
+        assert result.length == 255
+
+    def test_bytes_with_length(self):
+        result = self.dialect._designate_type("BYTES(1024)")
+        assert isinstance(result, types.LargeBinary)
+        assert result.length == 1024
+
+    def test_tokenlist_returns_tokenlist_type(self):
+        result = self.dialect._designate_type("TOKENLIST")
+        assert isinstance(result, sqlalchemy_spanner.TOKENLIST)
+
+    def test_unknown_type_returns_nulltype_with_warning(self):
+        with warnings.catch_warnings(record=True) as caught:
+            warnings.simplefilter("always")
+            result = self.dialect._designate_type("SOME_FUTURE_TYPE")
+        assert isinstance(result, types.NullType)
+        assert len(caught) == 1
+        assert "SOME_FUTURE_TYPE" in str(caught[0].message)
+
+    def test_array_of_known_type(self):
+        result = self.dialect._designate_type("ARRAY<INT64>")
+        assert isinstance(result, types.ARRAY)
+
+
+class TestTokenlistType(fixtures.TestBase):
+    """Verify TOKENLIST is a proper first-class type."""
+
+    def test_in_type_map(self):
+        assert "TOKENLIST" in sqlalchemy_spanner._type_map
+        assert sqlalchemy_spanner._type_map["TOKENLIST"] is sqlalchemy_spanner.TOKENLIST
+
+    def test_in_inverse_type_map(self):
+        assert sqlalchemy_spanner.TOKENLIST in sqlalchemy_spanner._type_map_inv
+        assert sqlalchemy_spanner._type_map_inv[sqlalchemy_spanner.TOKENLIST] == "TOKENLIST"
+
+    def test_type_compiler_roundtrip(self):
+        compiler = sqlalchemy_spanner.SpannerTypeCompiler(
+            sqlalchemy_spanner.SpannerDialect()
+        )
+        assert compiler.process(sqlalchemy_spanner.TOKENLIST()) == "TOKENLIST"
+
+    def test_is_type_engine(self):
+        assert issubclass(sqlalchemy_spanner.TOKENLIST, types.TypeEngine)