change example project start year to fit in doris partition history constraint. change doris docker to 1fe3be

Author: xinge-ji

sqlmesh/cli/project_init.py

@@ -7,6 +7,7 @@
 from sqlmesh.integrations.dlt import generate_dlt_models_and_settings
 from sqlmesh.utils.date import yesterday_ds
 from sqlmesh.utils.errors import SQLMeshError
+from datetime import datetime
 
 from sqlmesh.core.config.connection import (
     CONNECTION_CONFIG_TO_TYPE,
@@ -160,7 +161,7 @@ class ExampleObjects:
     python_macros: t.Dict[str, str]
 
 
-def _gen_example_objects(schema_name: str) -> ExampleObjects:
+def _gen_example_objects(schema_name: str, start_year: int) -> ExampleObjects:
     sql_models: t.Dict[str, str] = {}
     python_models: t.Dict[str, str] = {}
     seeds: t.Dict[str, str] = {}
@@ -194,7 +195,7 @@ def _gen_example_objects(schema_name: str) -> ExampleObjects:
   kind INCREMENTAL_BY_TIME_RANGE (
     time_column event_date
   ),
-  start '2020-01-01',
+  start '{start_year}-01-01',
   cron '@daily',
   grain (id, event_date)
 );
@@ -223,14 +224,14 @@ def _gen_example_objects(schema_name: str) -> ExampleObjects:
 );
   """
 
-    seeds["seed_data"] = """id,item_id,event_date
-1,2,2020-01-01
-2,1,2020-01-01
-3,3,2020-01-03
-4,1,2020-01-04
-5,1,2020-01-05
-6,1,2020-01-06
-7,1,2020-01-07
+    seeds["seed_data"] = f"""id,item_id,event_date
+1,2,{start_year}-01-01
+2,1,{start_year}-01-01
+3,3,{start_year}-01-03
+4,1,{start_year}-01-04
+5,1,{start_year}-01-05
+6,1,{start_year}-01-06
+7,1,{start_year}-01-07
 """
 
     audits["assert_positive_order_ids"] = """AUDIT (
@@ -328,6 +329,9 @@ def init_example_project(
                 "Please provide a DLT pipeline with the `--dlt-pipeline` flag to generate a SQLMesh project from DLT."
             )
 
+        if engine_type == "doris":
+            start = datetime(datetime.now().year, 1, 1).strftime("%Y-%m-%d")
+
     _create_config(config_path, engine_type, dialect, settings, start, template, cli_mode)
     if template == ProjectTemplate.DBT:
         return config_path
@@ -340,7 +344,7 @@ def init_example_project(
         )
         return config_path
 
-    example_objects = _gen_example_objects(schema_name=schema_name)
+    example_objects = _gen_example_objects(schema_name=schema_name, start_year=datetime.now().year)
 
     if template != ProjectTemplate.EMPTY:
         _create_object_files(models_path, example_objects.sql_models, "sql")

sqlmesh/core/config/connection.py

@@ -50,6 +50,8 @@
     "trino",
     # Nullable types are problematic
     "clickhouse",
+    # Do not support table name starts with "_"
+    "doris",
 }
 MOTHERDUCK_TOKEN_REGEX = re.compile(r"(\?|\&)(motherduck_token=)(\S*)")
 

sqlmesh/core/engine_adapter/doris.py

@@ -649,16 +649,6 @@ def _create_table_from_columns(
         """
         table_properties = kwargs.get("table_properties", {})
 
-        # Set default replication_num to 1 for testing environments if not specified
-        if (
-            "replication_num" not in table_properties
-            and "replication_allocation" not in table_properties
-        ):
-            table_properties["replication_num"] = "1"
-            logger.info(
-                f"[Doris] Added default replication_num for testing: {table_properties['replication_num']}"
-            )
-
         # Convert primary_key to unique_key for Doris (Doris doesn't support primary keys)
         if primary_key and "unique_key" not in table_properties:
             table_properties["unique_key"] = primary_key
@@ -688,7 +678,6 @@ def _build_partitioned_by_exp(
     ) -> t.Optional[t.Union[exp.PartitionedByProperty, exp.PartitionByRangeProperty, exp.Property]]:
         """Doris supports range and list partition, but sqlglot only supports range partition, so we use PartitionByRangeProperty."""
         # Handle partitioned_by_expr from kwargs
-        logging.info(f"[Doris] _build_partitioned_by_exp called with kwargs: {kwargs}")
         partitioned_by_expr = kwargs.get("partitioned_by_expr")
         create_expressions = None
 
@@ -861,18 +850,20 @@ def _build_table_properties_exp(
                     dynamic_partition_props = {
                         "dynamic_partition.enable": "true",
                         "dynamic_partition.time_unit": "DAY",
-                        "dynamic_partition.history_partition_num": "400",
-                        "dynamic_partition.end": "7",
+                        "dynamic_partition.start": "-490",
+                        "dynamic_partition.end": "10",
                         "dynamic_partition.prefix": "p",
                         "dynamic_partition.buckets": "32",
                         "dynamic_partition.create_history_partition": "true",
                     }
 
                     # Use partition_interval_unit if provided to set the time_unit
                     if partition_interval_unit:
-                        dynamic_partition_props["dynamic_partition.time_unit"] = (
-                            partition_interval_unit.upper()
-                        )
+                        if hasattr(partition_interval_unit, "value"):
+                            time_unit = partition_interval_unit.value.upper()
+                        else:
+                            time_unit = str(partition_interval_unit).upper()
+                        dynamic_partition_props["dynamic_partition.time_unit"] = time_unit
 
                     # Add missing dynamic partition properties to table_properties_copy
                     for key, value in dynamic_partition_props.items():

sqlmesh/core/model/meta.py

@@ -464,9 +464,22 @@ def batch_concurrency(self) -> t.Optional[int]:
     @cached_property
     def physical_properties(self) -> t.Dict[str, exp.Expression]:
         """A dictionary of properties that will be applied to the physical layer. It replaces table_properties which is deprecated."""
+        properties = {}
         if self.physical_properties_:
-            return {e.this.name: e.expression for e in self.physical_properties_.expressions}
-        return {}
+            properties = {e.this.name: e.expression for e in self.physical_properties_.expressions}
+
+        # For INCREMENTAL_BY_UNIQUE_KEY models, add the unique_key to physical_properties
+        # so it gets passed to table_properties during table creation
+        if isinstance(self.kind, IncrementalByUniqueKeyKind) and self.unique_key:
+            # Convert unique_key expressions to a format suitable for table_properties
+            if len(self.unique_key) == 1:
+                # Single column key
+                properties["unique_key"] = self.unique_key[0]
+            else:
+                # Multiple column key - create a tuple expression
+                properties["unique_key"] = exp.Tuple(expressions=self.unique_key)
+
+        return properties
 
     @cached_property
     def virtual_properties(self) -> t.Dict[str, exp.Expression]:

tests/core/engine_adapter/integration/docker/compose.doris.yaml

@@ -1,17 +1,80 @@
 services:
-  fe:
+  fe-01:
     image: apache/doris:fe-2.1.10
-    hostname: fe
+    container_name: doris-fe-01
+    hostname: fe-01
     environment:
-     - FE_SERVERS=fe1:127.0.0.1:9010
-     - FE_ID=1
-    network_mode: host
-  be:
+      - FE_SERVERS=fe1:172.20.80.2:9030
+      - FE_ID=1
+    ports:
+      - "8030:8030"
+      - "9030:9030"
+    volumes:
+      - ./fe-meta:/opt/apache-doris/fe/doris-meta
+      - ./fe-log:/opt/apache-doris/fe/log
+    networks:
+      doris_net:
+        ipv4_address: 172.20.80.2
+
+  be-01:
     image: apache/doris:be-2.1.10
-    hostname: be
+    container_name: doris-be-01
+    hostname: be-01
+    depends_on:
+      - fe-01
+    environment:
+      - FE_SERVERS=fe1:172.20.80.2:9030
+      - BE_ADDR=172.20.80.3:9050
+    ports:
+      - "8040:8040"
+      - "9050:9050"
+    volumes:
+      - ./be01-storage:/opt/apache-doris/be/storage
+      - ./be01-log:/opt/apache-doris/be/log
+    networks:
+      doris_net:
+        ipv4_address: 172.20.80.3
+
+  be-02:
+    image: apache/doris:be-2.1.10
+    container_name: doris-be-02
+    hostname: be-02
+    depends_on:
+      - fe-01
     environment:
-     - FE_SERVERS=fe1:127.0.0.1:9010
-     - BE_ADDR=127.0.0.1:9050
+      - FE_SERVERS=fe1:172.20.80.2:9030
+      - BE_ADDR=172.20.80.4:9050
+    ports:
+      - "8041:8040"
+      - "9051:9050"
+    volumes:
+      - ./be02-storage:/opt/apache-doris/be/storage
+      - ./be02-log:/opt/apache-doris/be/log
+    networks:
+      doris_net:
+        ipv4_address: 172.20.80.4
+
+  be-03:
+    image: apache/doris:be-2.1.10
+    container_name: doris-be-03
+    hostname: be-03
     depends_on:
-      - fe
-    network_mode: host
+      - fe-01
+    environment:
+      - FE_SERVERS=fe1:172.20.80.2:9030
+      - BE_ADDR=172.20.80.5:9050
+    ports:
+      - "8042:8040"
+      - "9052:9050"
+    volumes:
+      - ./be03-storage:/opt/apache-doris/be/storage
+      - ./be03-log:/opt/apache-doris/be/log
+    networks:
+      doris_net:
+        ipv4_address: 172.20.80.5
+
+networks:
+  doris_net:
+    ipam:
+      config:
+        - subnet: 172.20.80.0/24