fix(sqlalchemy-spanner): fix get_multi_indexes crash on SEARCH indexes (#17907)

Thank you for opening a Pull Request! Before submitting your PR, there
are a few things you can do to make sure it goes smoothly:
- [ ] Make sure to open an issue as a
[bug/issue](https://github.com/googleapis/google-cloud-python/issues)
before writing your code! That way we can discuss the change, evaluate
designs, and agree on the general idea
- [ ] Ensure the tests and linter pass
- [ ] Code coverage does not decrease (if any source code was changed)
- [ ] Appropriate docs were updated (if necessary)

Fixes #<issue_number_goes_here> 🦕

Author: sakthivelmanii

packages/sqlalchemy-spanner/.coveragerc

@@ -0,0 +1,6 @@
+[run]
+branch = True
+source = google/cloud/sqlalchemy_spanner
+
+[report]
+fail_under = 30

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

@@ -426,8 +426,10 @@ def returning_clause(self, stmt, returning_cols, **kw):
             )
             for c in expression._select_iterables(
                 filter(
-                    lambda col: not col.dialect_options.get("spanner", {}).get(
-                        "exclude_from_returning", False
+                    lambda col: (
+                        not col.dialect_options.get("spanner", {}).get(
+                            "exclude_from_returning", False
+                        )
                     ),
                     returning_cols,
                 )
@@ -1300,6 +1302,7 @@ def get_multi_indexes(
                 {table_type_query}
                 {schema_filter_query}
                 i.index_type != 'PRIMARY_KEY'
+                AND i.index_type != 'SEARCH'
                 AND i.spanner_is_managed = FALSE
             GROUP BY i.table_catalog, i.table_schema, i.table_name,
                      i.index_name, i.is_unique
@@ -1324,7 +1327,9 @@ def get_multi_indexes(
                     "column_names": row[3],
                     "unique": row[4],
                     "column_sorting": {
-                        col: order.lower() for col, order in zip(row[3], row[5])
+                        col: order.lower()
+                        for col, order in zip(row[3], row[5] or [])
+                        if order
                     },
                     "include_columns": include_columns if include_columns else [],
                     "dialect_options": dialect_options,

packages/sqlalchemy-spanner/noxfile.py

@@ -85,6 +85,7 @@ class = StreamHandler
 UNIT_TEST_STANDARD_DEPENDENCIES = [
     "mock",
     "pytest",
+    "pytest-cov",
 ]
 
 UNIT_TEST_EXTERNAL_DEPENDENCIES = [
@@ -361,14 +362,22 @@ def unit(session, test_type):
         return
 
     if test_type == "unit":
-        # Run SQLAlchemy dialect compliance test suite with OpenTelemetry.
+        # Run SQLAlchemy dialect unit tests with pytest-cov if COVERAGE_FILE is set.
         session.install(
             *UNIT_TEST_STANDARD_DEPENDENCIES,
             *UNIT_TEST_EXTERNAL_DEPENDENCIES,
             *UNIT_TEST_DEPENDENCIES,
         )
         session.install(".")
-        session.run("py.test", "--quiet", os.path.join("tests/unit"), *session.posargs)
+        pytest_args = ["--quiet", os.path.join("tests/unit")]
+        if "COVERAGE_FILE" in os.environ:
+            pytest_args.extend(
+                [
+                    "--cov=google.cloud.sqlalchemy_spanner",
+                    "--cov-config=.coveragerc",
+                ]
+            )
+        session.run("py.test", *pytest_args, *session.posargs)
         return
 
 

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

@@ -0,0 +1,86 @@
+# 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.
+
+from unittest.mock import MagicMock
+from sqlalchemy.testing import eq_
+from sqlalchemy.testing.plugin.plugin_base import fixtures
+from google.cloud.sqlalchemy_spanner.sqlalchemy_spanner import SpannerDialect
+
+
+class TestSpannerDialect(fixtures.TestBase):
+    def test_get_multi_indexes_excludes_search_indexes_sql(self):
+        """Test that get_multi_indexes SQL query excludes SEARCH indexes."""
+        dialect = SpannerDialect()
+        connection = MagicMock()
+        mock_snapshot = MagicMock()
+        mock_snapshot.execute_sql.return_value = []
+        connection.connection.database.snapshot.return_value.__enter__.return_value = (
+            mock_snapshot
+        )
+
+        dialect.get_multi_indexes(connection)
+
+        # Retrieve the SQL executed by snapshot
+        executed_sql = mock_snapshot.execute_sql.call_args[0][0]
+        assert "i.index_type != 'SEARCH'" in executed_sql
+
+    def test_get_multi_indexes_handles_none_column_ordering(self):
+        """Test get_multi_indexes with None column ordering."""
+        dialect = SpannerDialect()
+        connection = MagicMock()
+        mock_snapshot = MagicMock()
+        # Mock row: schema, table, index_name, columns,
+        # is_unique, column_orderings, storing_columns
+        mock_row = [
+            "public",
+            "my_table",
+            "idx_search",
+            ["col1"],
+            False,
+            [None],  # column_ordering is None
+            [],
+        ]
+        mock_snapshot.execute_sql.return_value = [mock_row]
+        connection.connection.database.snapshot.return_value.__enter__.return_value = (
+            mock_snapshot
+        )
+
+        res = dialect.get_multi_indexes(connection)
+        assert ("public", "my_table") in res
+        index_info = res[("public", "my_table")][0]
+        eq_(index_info["column_sorting"], {})
+
+    def test_get_multi_indexes_handles_null_column_orderings_array(self):
+        """Test get_multi_indexes when column_orderings array is None."""
+        dialect = SpannerDialect()
+        connection = MagicMock()
+        mock_snapshot = MagicMock()
+        mock_row = [
+            "public",
+            "my_table",
+            "idx_test",
+            ["col1"],
+            False,
+            None,  # row[5] is None
+            [],
+        ]
+        mock_snapshot.execute_sql.return_value = [mock_row]
+        connection.connection.database.snapshot.return_value.__enter__.return_value = (
+            mock_snapshot
+        )
+
+        res = dialect.get_multi_indexes(connection)
+        assert ("public", "my_table") in res
+        index_info = res[("public", "my_table")][0]
+        eq_(index_info["column_sorting"], {})