Enhance Doris integration

Author: xinge-ji

Makefile

@@ -210,6 +210,7 @@ risingwave-test: engine-risingwave-up
 
 doris-test: engine-doris-up
 	pytest -n auto -m "doris" --reruns 3 --junitxml=test-results/junit-doris.xml
+
 #################
 # Cloud Engines #
 #################

docs/guides/configuration.md

@@ -911,6 +911,7 @@ These pages describe the connection configuration options for each execution eng
 * [Athena](../integrations/engines/athena.md)
 * [BigQuery](../integrations/engines/bigquery.md)
 * [Databricks](../integrations/engines/databricks.md)
+* [Doris](../integrations/engines/doris.md)
 * [DuckDB](../integrations/engines/duckdb.md)
 * [Fabric](../integrations/engines/fabric.md)
 * [MotherDuck](../integrations/engines/motherduck.md)
@@ -953,6 +954,7 @@ Unsupported state engines, even for development:
 * [ClickHouse](../integrations/engines/clickhouse.md)
 * [Spark](../integrations/engines/spark.md)
 * [Trino](../integrations/engines/trino.md)
+* [Doris](../integrations/engines/doris.md)
 
 This example gateway configuration uses Snowflake for the data warehouse connection and Postgres for the state backend connection:
 

docs/guides/connections.md

@@ -81,6 +81,7 @@ default_gateway: local_db
 
 * [BigQuery](../integrations/engines/bigquery.md)
 * [Databricks](../integrations/engines/databricks.md)
+* [Doris](../integrations/engines/doris.md)
 * [DuckDB](../integrations/engines/duckdb.md)
 * [MotherDuck](../integrations/engines/motherduck.md)
 * [MySQL](../integrations/engines/mysql.md)

docs/integrations/engines/doris.md

@@ -139,7 +139,7 @@ MODEL (
   kind FULL,
   partitioned_by RANGE(event_date),
   physical_properties (
-    partitioned_by_expr = (
+    partitions = (
       'PARTITION `p2023` VALUES [("2023-01-01"), ("2024-01-01"))', 
       'PARTITION `p2024` VALUES [("2024-01-01"), ("2025-01-01"))', 
       'PARTITION `p2025` VALUES [("2025-01-01"), ("2026-01-01"))', 

docs/integrations/overview.md

@@ -16,6 +16,7 @@ SQLMesh supports the following execution engines for running SQLMesh projects (e
 * [BigQuery](./engines/bigquery.md) (bigquery)
 * [ClickHouse](./engines/clickhouse.md) (clickhouse)
 * [Databricks](./engines/databricks.md) (databricks)
+* [Doris](./engines/doris.md) (doris)
 * [DuckDB](./engines/duckdb.md) (duckdb)
 * [Fabric](./engines/fabric.md) (fabric)
 * [MotherDuck](./engines/motherduck.md) (motherduck)

mkdocs.yml

@@ -82,6 +82,7 @@ nav:
       - integrations/engines/bigquery.md
       - integrations/engines/clickhouse.md
       - integrations/engines/databricks.md
+      - integrations/engines/doris.md
       - integrations/engines/duckdb.md
       - integrations/engines/fabric.md
       - integrations/engines/motherduck.md

pyproject.toml

@@ -272,7 +272,7 @@ markers = [
     "pyspark: test for PySpark that need to run separately from the other spark tests",
     "trino: test for Trino (all connectors)",
     "risingwave: test for Risingwave",
-    "doris: test for Doris",,
+    "doris: test for Doris",
 
     # Other
     "set_default_connection",

sqlmesh/core/engine_adapter/doris.py

@@ -6,7 +6,7 @@
 
 from sqlglot import exp, parse_one
 
-from sqlmesh.core.dialect import to_schema, transform_values
+from sqlmesh.core.dialect import to_schema
 
 from sqlmesh.core.engine_adapter.base import (
     InsertOverwriteStrategy,
@@ -17,21 +17,21 @@
     PandasNativeFetchDFSupportMixin,
 )
 from sqlmesh.core.engine_adapter.shared import (
+    CatalogSupport,
     CommentCreationTable,
     CommentCreationView,
     DataObject,
     DataObjectType,
     set_catalog,
 )
-from sqlmesh.core.schema_diff import SchemaDiffer
-from sqlmesh.utils import random_id, get_source_columns_to_types
+from sqlmesh.utils import random_id
 from sqlmesh.utils.errors import (
     SQLMeshError,
 )
 
 if t.TYPE_CHECKING:
     from sqlmesh.core._typing import SchemaName, TableName
-    from sqlmesh.core.engine_adapter._typing import QueryOrDF, Query
+    from sqlmesh.core.engine_adapter._typing import QueryOrDF
     from sqlmesh.core.node import IntervalUnit
 
 logger = logging.getLogger(__name__)
@@ -48,13 +48,18 @@ class DorisEngineAdapter(
     COMMENT_CREATION_VIEW = CommentCreationView.IN_SCHEMA_DEF_NO_COMMANDS
     MAX_TABLE_COMMENT_LENGTH = 2048
     MAX_COLUMN_COMMENT_LENGTH = 255
-    SUPPORTS_REPLACE_TABLE = False 
+    SUPPORTS_INDEXES = True
+    SUPPORTS_REPLACE_TABLE = False
     MAX_IDENTIFIER_LENGTH = 64
     SUPPORTS_MATERIALIZED_VIEWS = True
     SUPPORTS_MATERIALIZED_VIEW_SCHEMA = True
     SUPPORTS_CREATE_DROP_CATALOG = False
     INSERT_OVERWRITE_STRATEGY = InsertOverwriteStrategy.DELETE_INSERT
 
+    @property
+    def catalog_support(self) -> CatalogSupport:
+        return CatalogSupport.FULL_SUPPORT
+
     def create_schema(
         self,
         schema_name: SchemaName,
@@ -289,7 +294,6 @@ def drop_view(
         """
         Drop view in Doris.
         """
-        # Remove cascade from kwargs as Doris doesn't support it
         if materialized and kwargs.get("view_properties"):
             view_properties = kwargs.pop("view_properties")
             if view_properties.get("materialized_type") == "SYNC" and view_properties.get(
@@ -1116,4 +1120,4 @@ def _properties_to_expressions(self, properties: t.Dict[str, t.Any]) -> t.List[e
             if not isinstance(value, exp.Expression):
                 value = exp.Literal.string(str(value))
             expressions.append(exp.Property(this=exp.Literal.string(str(key)), value=value))
-        return expressions
+        return expressions

tests/core/engine_adapter/integration/__init__.py

@@ -77,6 +77,7 @@ def pytest_marks(self) -> t.List[MarkDecorator]:
     IntegrationTestEngine("spark", native_dataframe_type="pyspark"),
     IntegrationTestEngine("clickhouse", catalog_types=["standalone", "cluster"]),
     IntegrationTestEngine("risingwave"),
+    IntegrationTestEngine("doris"),
     # Cloud engines that need paid accounts / special credentials
     IntegrationTestEngine("clickhouse_cloud", cloud=True),
     IntegrationTestEngine("redshift", cloud=True),
@@ -448,7 +449,7 @@ def get_table_comment(
                     AND pgc.relkind = '{"v" if table_kind == "VIEW" else "r"}'
                 ;
             """
-        elif self.dialect in ["mysql", "snowflake"]:
+        elif self.dialect in ["mysql", "snowflake", "doris"]:
             # Snowflake treats all identifiers as uppercase unless they are lowercase and quoted.
             # They are lowercase and quoted in sushi but not in the inline tests.
             if self.dialect == "snowflake" and snowflake_capitalize_ids:
@@ -458,6 +459,7 @@ def get_table_comment(
             comment_field_name = {
                 "mysql": "table_comment",
                 "snowflake": "comment",
+                "doris": "table_comment",
             }
 
             query = f"""
@@ -563,7 +565,7 @@ def get_column_comments(
                     AND pgc.relkind = '{"v" if table_kind == "VIEW" else "r"}'
                 ;
             """
-        elif self.dialect in ["mysql", "snowflake", "trino"]:
+        elif self.dialect in ["mysql", "snowflake", "trino", "doris"]:
             # Snowflake treats all identifiers as uppercase unless they are lowercase and quoted.
             # They are lowercase and quoted in sushi but not in the inline tests.
             if self.dialect == "snowflake" and snowflake_capitalize_ids:
@@ -574,6 +576,7 @@ def get_column_comments(
                 "mysql": "column_comment",
                 "snowflake": "comment",
                 "trino": "comment",
+                "doris": "column_comment",
             }
 
             query = f"""

tests/core/engine_adapter/integration/config.yaml

@@ -118,6 +118,16 @@ gateways:
       host: {{ env_var('DOCKER_HOSTNAME', 'localhost') }}
       port: 4566
       check_import: false
+  inttest_doris:
+    connection:
+      type: doris
+      host: {{ env_var('DOCKER_HOSTNAME', 'localhost') }}
+      port: 9030
+      user: root
+      password: ""
+      check_import: false
+    state_connection:
+      type: duckdb
 
 
   # Cloud databases