Refactor: Update Doris engine adapter

Author: xinge-ji

sqlmesh/cli/project_init.py

@@ -207,7 +207,7 @@ def _gen_example_objects(schema_name: str, dialect: str) -> ExampleObjects:
   name {incremental_model_name},
   kind INCREMENTAL_BY_TIME_RANGE (
     time_column event_date
-    {"partition_by_time_column false" if dialect == "doris" else ""}
+    {",partition_by_time_column false" if dialect == "doris" else ""}
   ),
   start '2020-01-01',
   cron '@daily',

sqlmesh/core/config/connection.py

@@ -2304,7 +2304,7 @@ class DorisConnectionConfig(ConnectionConfig):
     host: str
     user: str
     password: str
-    port: t.Optional[int] = 9030  # Default Doris FE port
+    port: t.Optional[int] = 9030
     database: t.Optional[str] = None
     charset: t.Optional[str] = None
     collation: t.Optional[str] = None
@@ -2317,7 +2317,7 @@ class DorisConnectionConfig(ConnectionConfig):
     type_: t.Literal["doris"] = Field(alias="type", default="doris")
     DIALECT: t.ClassVar[t.Literal["doris"]] = "doris"
     DISPLAY_NAME: t.ClassVar[t.Literal["Apache Doris"]] = "Apache Doris"
-    DISPLAY_ORDER: t.ClassVar[t.Literal[17]] = 17
+    DISPLAY_ORDER: t.ClassVar[t.Literal[18]] = 18
 
     _engine_import_validator = _get_engine_import_validator("pymysql", "doris")
 

sqlmesh/core/engine_adapter/base.py

@@ -2658,7 +2658,7 @@ def _get_temp_table_name(self, table: TableName) -> str:
         """
         table_obj = exp.to_table(table)
         return f"__temp_{table_obj.name}_{random_id(short=True)}"
-    
+
     def _get_data_objects(
         self, schema_name: SchemaName, object_names: t.Optional[t.Set[str]] = None
     ) -> t.List[DataObject]:

sqlmesh/core/engine_adapter/doris.py

@@ -1768,7 +1768,9 @@ def create(
         ):
             physical_properties = dict(physical_properties)
             physical_properties["unique_key"] = (
-                model.unique_key[0] if len(model.unique_key) == 1 else exp.Tuple(expressions=model.unique_key)
+                model.unique_key[0]
+                if len(model.unique_key) == 1
+                else exp.Tuple(expressions=model.unique_key)
             )
 
         logger.info("Creating table '%s'", table_name)

sqlmesh/core/snapshot/evaluator.py

@@ -824,7 +824,7 @@ def test_insert_overwrite_by_time_partition(ctx_query_and_df: TestContext):
         date_3 = (current_date + timedelta(days=2)).strftime("%Y-%m-%d")
         date_4 = (current_date + timedelta(days=3)).strftime("%Y-%m-%d")
         date_5 = (current_date + timedelta(days=4)).strftime("%Y-%m-%d")
-        
+
     ctx.columns_to_types = {"id": "int", "ds": ds_type}
     table = ctx.table("test_table")
     if ctx.dialect == "bigquery":
@@ -933,7 +933,7 @@ def test_insert_overwrite_by_time_partition_source_columns(ctx_query_and_df: Tes
         date_3 = (current_date + timedelta(days=2)).strftime("%Y-%m-%d")
         date_4 = (current_date + timedelta(days=3)).strftime("%Y-%m-%d")
         date_5 = (current_date + timedelta(days=4)).strftime("%Y-%m-%d")
-        
+
     ctx.columns_to_types = {"id": "int", "ds": ds_type}
     columns_to_types = {
         "id": exp.DataType.build("int"),
@@ -1114,7 +1114,9 @@ def test_merge_source_columns(ctx_query_and_df: TestContext):
     columns_to_types = ctx.columns_to_types.copy()
     columns_to_types["ignored_column"] = exp.DataType.build("int")
 
-    ctx.engine_adapter.create_table(table, columns_to_types, table_format=table_format, table_properties=table_properties)
+    ctx.engine_adapter.create_table(
+        table, columns_to_types, table_format=table_format, table_properties=table_properties
+    )
     input_data = pd.DataFrame(
         [
             {"id": 1, "ds": "2022-01-01", "ignored_source": "ignored_value"},
@@ -3843,6 +3845,10 @@ def test_materialized_view_evaluation(ctx: TestContext, mocker: MockerFixture):
         pytest.skip(f"Skipping engine {dialect} as it does not support materialized views")
     elif dialect in ("snowflake", "databricks"):
         pytest.skip(f"Skipping {dialect} as they're not enabled on standard accounts")
+    elif dialect == "doris":
+        pytest.skip(
+            f"Skipping doris as synchronous materialized views do not support specifying schema"
+        )
 
     model_name = ctx.table("test_tbl")
     mview_name = ctx.table("test_mview")

tests/core/engine_adapter/integration/test_integration.py

@@ -81,7 +81,9 @@ def test_create_materialized_view(make_mocked_engine_adapter: t.Callable[..., Do
         target_columns_to_types={"a": exp.DataType.build("INT"), "b": exp.DataType.build("INT")},
         column_descriptions={"a": "test_column_description", "b": "test_column_description"},
     )
-    adapter.create_view("test_view", parse_one("SELECT a, b FROM tbl"), replace=False, materialized=True)
+    adapter.create_view(
+        "test_view", parse_one("SELECT a, b FROM tbl"), replace=False, materialized=True
+    )
 
     assert to_sql_calls(adapter) == [
         "CREATE MATERIALIZED VIEW `test_view` (`a` COMMENT 'test_column_description', `b` COMMENT 'test_column_description') AS SELECT `a`, `b` FROM `tbl`",
@@ -160,7 +162,9 @@ def test_rename_table(make_mocked_engine_adapter: t.Callable[..., DorisEngineAda
     adapter.cursor.execute.assert_called_once_with("ALTER TABLE `old_table` RENAME `new_table`")
 
 
-def test_replace_by_key(make_mocked_engine_adapter: t.Callable[..., DorisEngineAdapter], mocker: MockerFixture):
+def test_replace_by_key(
+    make_mocked_engine_adapter: t.Callable[..., DorisEngineAdapter], mocker: MockerFixture
+):
     adapter = make_mocked_engine_adapter(DorisEngineAdapter)
     temp_table = parse_one("temp_table")
     mocker.patch.object(adapter, "_get_temp_table", return_value=temp_table)
@@ -174,8 +178,12 @@ def test_replace_by_key(make_mocked_engine_adapter: t.Callable[..., DorisEngineA
 
     adapter._replace_by_key(target_table, source_query, columns_to_types, [exp.column("a")], True)
     adapter._replace_by_key(target_table, source_query, columns_to_types, [exp.column("a")], False)
-    adapter._replace_by_key(target_table, source_query, columns_to_types, [exp.column("a"), exp.column("b")], True)
-    adapter._replace_by_key(target_table, source_query, columns_to_types, [exp.column("a"), exp.column("b")], False)
+    adapter._replace_by_key(
+        target_table, source_query, columns_to_types, [exp.column("a"), exp.column("b")], True
+    )
+    adapter._replace_by_key(
+        target_table, source_query, columns_to_types, [exp.column("a"), exp.column("b")], False
+    )
 
     assert to_sql_calls(adapter, identify=True) == [
         "INSERT INTO `target_table` (`a`, `b`) SELECT `a`, `b` FROM (SELECT `a` AS `a`, `b` AS `b`, ROW_NUMBER() OVER (PARTITION BY `a` ORDER BY `a`) AS _row_number FROM `temp_table`) AS _t WHERE _row_number = 1",
@@ -189,7 +197,9 @@ def test_create_index(make_mocked_engine_adapter: t.Callable[..., DorisEngineAda
     adapter = make_mocked_engine_adapter(DorisEngineAdapter)
 
     adapter.create_index("test_table", "test_index", ("cola",))
-    adapter.cursor.execute.assert_called_once_with("CREATE INDEX IF NOT EXISTS `test_index` ON `test_table`(`cola`)")
+    adapter.cursor.execute.assert_called_once_with(
+        "CREATE INDEX IF NOT EXISTS `test_index` ON `test_table`(`cola`)"
+    )
 
 
 def test_create_table_with_distributed_by(
@@ -271,7 +281,11 @@ def test_create_table_with_partitioned_by(
         "test_table",
         target_columns_to_types={"a": exp.DataType.build("INT"), "b": exp.DataType.build("DATE")},
         partitioned_by=[exp.to_column("b")],
-        table_properties={"partitions": exp.Literal.string("FROM ('2000-11-14') TO ('2021-11-14') INTERVAL 2 YEAR")},
+        table_properties={
+            "partitions": exp.Literal.string(
+                "FROM ('2000-11-14') TO ('2021-11-14') INTERVAL 2 YEAR"
+            )
+        },
     )
 
     assert to_sql_calls(adapter) == [

tests/core/engine_adapter/test_doris.py

@@ -1797,6 +1797,7 @@ def test_schema_differ_overrides(make_config) -> None:
     assert adapter._schema_differ_overrides == override
     assert adapter.schema_differ.parameterized_type_defaults == {}
 
+
 def test_doris(make_config):
     """Test DorisConnectionConfig basic functionality"""
     # Basic configuration
@@ -1818,7 +1819,7 @@ def test_doris(make_config):
     assert config.database == "demo"
     assert config.DIALECT == "doris"
     assert config.DISPLAY_NAME == "Apache Doris"
-    assert config.DISPLAY_ORDER == 17
+    assert config.DISPLAY_ORDER == 18
     assert config.is_recommended_for_state_sync is False
 
     # Test with minimal configuration (using default port)

tests/core/test_connection_config.py

@@ -1185,7 +1185,6 @@ def test_create_view_with_properties(mocker: MockerFixture, adapter_mock, make_s
             "clustered_by": [],
             "partition_interval_unit": None,
             "partitioned_by": [],
-            "key": exp.convert("value"),  # Physical properties should also be in materialized_properties
         },
         table_description=None,
         replace=False,