feat(doris): add catalog support for Apache Doris

superset/db_engine_specs/doris.py

@@ -22,11 +22,13 @@
 
 from flask_babel import gettext as __
 from sqlalchemy import Float, Integer, Numeric, String, TEXT, types
+from sqlalchemy.engine.reflection import Inspector
 from sqlalchemy.engine.url import URL
 from sqlalchemy.sql.type_api import TypeEngine
 
 from superset.db_engine_specs.mysql import MySQLEngineSpec
 from superset.errors import SupersetErrorType
+from superset.models.core import Database
 from superset.utils.core import GenericDataType
 
 # Regular expressions to catch custom errors
@@ -111,6 +113,7 @@
     )
     encryption_parameters = {"ssl": "0"}
     supports_dynamic_schema = True
+    supports_catalog = supports_dynamic_catalog = True
 
     column_type_mappings = (  # type: ignore
         (
@@ -245,17 +248,45 @@
         catalog: Optional[str] = None,
         schema: Optional[str] = None,
     ) -> tuple[URL, dict[str, Any]]:
-        database = uri.database
-        if schema and database:
-            schema = parse.quote(schema, safe="")
-            if "." in database:
-                database = database.split(".")[0] + "." + schema
-            else:
-                database = "internal." + schema
-            uri = uri.set(database=database)
-
+        if uri.database and "." in uri.database:
+            current_catalog, _ = uri.database.split(".", 1)
+        else:
+            current_catalog = uri.database
+
+        # In Apache Doris, each catalog has an information_schema for BI tool
+        # compatibility. See: https://github.com/apache/doris/pull/28919
+        adjusted_database = ".".join(
+            [catalog or current_catalog or "", "information_schema"]
+        ).rstrip(".")
+
+        uri = uri.set(database=adjusted_database)
         return uri, connect_args
 
+    @classmethod
+    def get_default_catalog(cls, database: Database) -> Optional[str]:
+        """
+        Return the default catalog.
+        """
+        if database.url_object.database is None:
+            return None
+
+        return database.url_object.database.split(".")[0]
+
+    @classmethod
+    def get_catalog_names(
+        cls,
+        database: Database,
+        inspector: Inspector,
+    ) -> set[str]:
+        """
+        Get all catalogs.
+        For Doris, the SHOW CATALOGS command returns multiple columns:
+        CatalogId, CatalogName, Type, IsCurrent, CreateTime, LastUpdateTime, Comment
+        We need to extract just the CatalogName column.
+        """
+        result = inspector.bind.execute("SHOW CATALOGS")
+        return {row.CatalogName for row in result}
+
     @classmethod
     def get_schema_from_engine_params(
         cls,

tests/unit_tests/db_engine_specs/test_doris.py

@@ -16,6 +16,7 @@
 # under the License.
 
 from typing import Any, Optional
+from unittest.mock import Mock
 
 import pytest
 from sqlalchemy import JSON, types
@@ -145,3 +146,52 @@ def test_get_schema_from_engine_params() -> None:
         )
         is None
     )
+
+
+def test_get_default_catalog() -> None:
+    """
+    Test the ``get_default_catalog`` method.
+    """
+    from superset.db_engine_specs.doris import DorisEngineSpec
+    from superset.models.core import Database
+
+    database = Mock(spec=Database)
+
+    # Test with catalog.schema format
+    database.url_object.database = "catalog1.schema1"
+    assert DorisEngineSpec.get_default_catalog(database) == "catalog1"
+
+    # Test with only catalog format
+    database.url_object.database = "catalog1"
+    assert DorisEngineSpec.get_default_catalog(database) == "catalog1"
+
+    # Test with None
+    database.url_object.database = None
+    assert DorisEngineSpec.get_default_catalog(database) is None
+
+
+def test_get_catalog_names() -> None:
+    """
+    Test the ``get_catalog_names`` method.
+    """
+    from superset.db_engine_specs.doris import DorisEngineSpec
+    from superset.models.core import Database
+
+    database = Mock(spec=Database)
+    inspector = Mock()
+
+    # Mock the execute result
+    mock_result = [
+        Mock(CatalogName="catalog1"),
+        Mock(CatalogName="catalog2"),
+        Mock(CatalogName="catalog3"),
+    ]
+    inspector.bind.execute.return_value = mock_result
+
+    catalogs = DorisEngineSpec.get_catalog_names(database, inspector)
+
+    # Verify the SQL query
+    inspector.bind.execute.assert_called_once_with("SHOW CATALOGS")
+
+    # Verify the returned catalog names
+    assert catalogs == {"catalog1", "catalog2", "catalog3"}