fix: use catalog name on generated queries

Author: betodealmeida

docker/pythonpath_dev/superset_config_docker_light.py

@@ -19,6 +19,7 @@
 
 # Import all settings from the main config first
 from flask_caching.backends.filesystemcache import FileSystemCache
+
 from superset_config import *  # noqa: F403
 
 # Override caching to use simple in-memory cache instead of Redis

superset/connectors/sqla/models.py

@@ -1369,8 +1369,24 @@ def get_template_processor(self, **kwargs: Any) -> BaseTemplateProcessor:
 
     def get_sqla_table(self) -> TableClause:
         tbl = table(self.table_name)
-        if self.schema:
+
+        # For databases that support cross-catalog queries (like BigQuery),
+        # include the catalog in the table identifier to generate
+        # project.dataset.table format
+        if self.catalog and self.database.db_engine_spec.supports_cross_catalog_queries:
+            # SQLAlchemy doesn't have built-in catalog support for TableClause,
+            # so we need to construct the full identifier manually
+            if self.schema:
+                full_name = f"{self.catalog}.{self.schema}.{self.table_name}"
+            else:
+                full_name = f"{self.catalog}.{self.table_name}"
+
+            tbl = table(full_name)
+
+        # Otherwise include only the schema, if any
+        elif self.schema:
             tbl.schema = self.schema
+
         return tbl
 
     def get_from_clause(

tests/unit_tests/connectors/sqla/models_test.py

@@ -287,3 +287,65 @@ def test_normalize_prequery_result_type_custom_sql() -> None:
         sqla_table._normalize_prequery_result_type(row, dimension, columns_by_name)
         == "Car"
     )
+
+
+def test_get_sqla_table_with_catalog(mocker: MockerFixture) -> None:
+    """Test that get_sqla_table includes catalog in table identifier
+    for databases with cross-catalog support
+    """
+    # Test case 1: Database supports cross-catalog queries (like BigQuery)
+    database_with_catalog = mocker.MagicMock()
+    database_with_catalog.db_engine_spec.supports_cross_catalog_queries = True
+
+    table_with_catalog = SqlaTable(
+        table_name="test_table",
+        database=database_with_catalog,
+        schema="test_dataset",
+        catalog="test_project",
+    )
+
+    sqla_table = table_with_catalog.get_sqla_table()
+    # Should generate "test_project.test_dataset.test_table"
+    assert sqla_table.name == "test_project.test_dataset.test_table"
+    # Schema should not be set as it's now part of the table name
+    assert sqla_table.schema is None
+
+    # Test case 2: Database doesn't support cross-catalog queries
+    database_no_catalog = mocker.MagicMock()
+    database_no_catalog.db_engine_spec.supports_cross_catalog_queries = False
+
+    table_no_catalog_support = SqlaTable(
+        table_name="test_table",
+        database=database_no_catalog,
+        schema="test_schema",
+        catalog="test_catalog",
+    )
+
+    sqla_table_no_catalog = table_no_catalog_support.get_sqla_table()
+    # Should ignore catalog and use traditional schema.table format
+    assert sqla_table_no_catalog.name == "test_table"
+    assert sqla_table_no_catalog.schema == "test_schema"
+
+    # Test case 3: Table with catalog but no schema (BigQuery style)
+    table_catalog_only = SqlaTable(
+        table_name="test_table",
+        database=database_with_catalog,
+        catalog="test_project",
+    )
+
+    sqla_table_catalog_only = table_catalog_only.get_sqla_table()
+    # Should generate "test_project.test_table"
+    assert sqla_table_catalog_only.name == "test_project.test_table"
+    assert sqla_table_catalog_only.schema is None
+
+    # Test case 4: Table with schema but no catalog (traditional behavior)
+    table_schema_only = SqlaTable(
+        table_name="test_table",
+        database=database_with_catalog,
+        schema="test_schema",
+    )
+
+    sqla_table_schema_only = table_schema_only.get_sqla_table()
+    # Should generate just "test_table" with schema set separately
+    assert sqla_table_schema_only.name == "test_table"
+    assert sqla_table_schema_only.schema == "test_schema"