fix(sqlalchemy-spanner): handle empty filter_names in multi-table reflection

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

@@ -993,7 +993,9 @@ def _get_table_filter_query(
                     table_filter_query = table_filter_query + " OR " + query
                 else:
                     table_filter_query = query
-            table_filter_query = "(" + table_filter_query + ") "
+            # An empty filter_names selects no tables. "()" is not valid SQL,
+            # so render a condition that is always false.
+            table_filter_query = "(" + (table_filter_query or "FALSE") + ") "
             if append_query:
                 table_filter_query = table_filter_query + " AND "
 

packages/sqlalchemy-spanner/tests/test_suite_20.py

@@ -1297,6 +1297,28 @@ def test_reflect_bytes_column_max_len(self, connection):
 
         Table("bytes_table", metadata, autoload=True)
 
+    def test_reflect_with_empty_filter_names(self, connection):
+        """
+        SPANNER SPECIFIC TEST:
+
+        An empty filter_names selects no tables. Checks that multi-table
+        reflection returns nothing for it instead of running invalid SQL, as
+        MetaData.reflect() does when it has no tables left to load.
+        """
+        insp = inspect(connection)
+        eq_(insp.get_multi_columns(filter_names=[]), {})
+        eq_(insp.get_multi_pk_constraint(filter_names=[]), {})
+        eq_(insp.get_multi_foreign_keys(filter_names=[]), {})
+        eq_(insp.get_multi_indexes(filter_names=[]), {})
+
+        metadata = MetaData()
+        metadata.reflect(connection, only=[])
+        eq_(list(metadata.tables), [])
+        metadata.reflect(connection, only=["users"])
+        # The table is already in the metadata, so this loads no tables.
+        metadata.reflect(connection, only=["users"])
+        eq_(list(metadata.tables), ["users"])
+
     @testing.requires.unique_constraint_reflection
     def test_get_unique_constraints(self, metadata, connection, use_schema=False):
         # SQLite dialect needs to parse the names of the constraints

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

@@ -21,6 +21,34 @@
 
 
 class TestSpannerDialect(fixtures.TestBase):
+    def test_get_table_filter_query(self):
+        """Test the WHERE condition rendered for filter_names."""
+        dialect = SpannerDialect()
+        eq_(dialect._get_table_filter_query(None, "col", True), "")
+        eq_(
+            dialect._get_table_filter_query(["t1", "t2"], "col", True),
+            "(col.table_name = 't1' OR col.table_name = 't2')  AND ",
+        )
+        # An empty filter_names matches no tables instead of rendering "()".
+        eq_(dialect._get_table_filter_query([], "col", True), "(FALSE)  AND ")
+        eq_(dialect._get_table_filter_query((), "col"), "(FALSE) ")
+
+    def test_get_multi_columns_with_empty_filter_names(self):
+        """Test get_multi_* with an empty filter_names returns no tables."""
+        dialect = SpannerDialect()
+        connection = MagicMock()
+        mock_snapshot = MagicMock()
+        mock_snapshot.execute_sql.return_value = []
+        connection.connection.database.snapshot.return_value.__enter__.return_value = (
+            mock_snapshot
+        )
+
+        eq_(dialect.get_multi_columns(connection, filter_names=[]), {})
+
+        executed_sql = mock_snapshot.execute_sql.call_args[0][0]
+        assert "(FALSE)" in executed_sql
+        assert "()" not in executed_sql
+
     def test_get_multi_indexes_excludes_search_indexes_sql(self):
         """Test that get_multi_indexes SQL query excludes SEARCH indexes."""
         dialect = SpannerDialect()