Add readiness check for Doris backends

Author: xinge-ji

.circleci/wait-for-db.sh

@@ -36,6 +36,23 @@ clickhouse_ready() {
 
 doris_ready() {
     probe_port 9030
+    
+    # Check that we have 3 alive backends
+    echo "Checking for 3 alive Doris backends..."
+    
+    while true; do
+        echo "Checking Doris backends..."
+        # Use docker compose exec to run mysql command inside the fe-01 container
+        # Use timeout to prevent hanging, and handle connection errors gracefully
+        ALIVE_BACKENDS=$(timeout 10 docker compose -f tests/core/engine_adapter/integration/docker/compose.doris.yaml exec -T fe-01 mysql -uroot -e "show backends \G" 2>/dev/null | grep -c "Alive: true" || echo "0")
+        echo "Found $ALIVE_BACKENDS alive backends"
+        if [ "$ALIVE_BACKENDS" -ge 3 ]; then
+            echo "Doris has 3 or more alive backends"
+            break
+        fi
+        echo "Waiting for more backends to become alive..."
+        sleep 5
+    done
 }
 
 postgres_ready() {

sqlmesh/core/model/meta.py

@@ -470,7 +470,11 @@ def physical_properties(self) -> t.Dict[str, exp.Expression]:
 
         # For INCREMENTAL_BY_UNIQUE_KEY models for doris dialect, 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 and self.dialect == "doris":
+        if (
+            isinstance(self.kind, IncrementalByUniqueKeyKind)
+            and self.unique_key
+            and self.dialect == "doris"
+        ):
             # Convert unique_key expressions to a format suitable for table_properties
             if len(self.unique_key) == 1:
                 # Single column key

tests/core/test_context.py

@@ -64,8 +64,6 @@
 from tests.utils.test_helpers import use_terminal_console
 from tests.utils.test_filesystem import create_temp_file
 
-import logging
-logger = logging.getLogger(__name__)
 
 def test_global_config(copy_to_temp_path: t.Callable):
     context = Context(paths=copy_to_temp_path("examples/sushi"))
@@ -2527,7 +2525,10 @@ def _get_missing_intervals(plan: Plan, name: str) -> t.List[t.Tuple[datetime, da
     assert len(plan.missing_intervals) == 4
 
     assert _get_missing_intervals(plan, "sqlmesh_example.daily_model") == [
-        (to_datetime(f"{current_year}-01-31 00:00:00"), to_datetime(f"{current_year}-02-01 00:00:00"))
+        (
+            to_datetime(f"{current_year}-01-31 00:00:00"),
+            to_datetime(f"{current_year}-02-01 00:00:00"),
+        )
     ]
     assert _get_missing_intervals(plan, "sqlmesh_example.weekly_model") == [
         (
@@ -2554,21 +2555,35 @@ def _get_missing_intervals(plan: Plan, name: str) -> t.List[t.Tuple[datetime, da
     # show that the data was created (which shows that when the Plan became an EvaluatablePlan and eventually evaluated, the start date overrides didnt get dropped)
     assert context.engine_adapter.fetchall(
         "select start_dt, end_dt from sqlmesh_example__pr_env.daily_model"
-    ) == [(to_datetime(f"{current_year}-01-31 00:00:00"), to_datetime(f"{current_year}-01-31 23:59:59.999999"))]
+    ) == [
+        (
+            to_datetime(f"{current_year}-01-31 00:00:00"),
+            to_datetime(f"{current_year}-01-31 23:59:59.999999"),
+        )
+    ]
     assert context.engine_adapter.fetchall(
         "select start_dt, end_dt from sqlmesh_example__pr_env.weekly_model"
     ) == [
-        (to_datetime(f"{current_year}-01-19 00:00:00"), to_datetime(f"{current_year}-01-25 23:59:59.999999")),
+        (
+            to_datetime(f"{current_year}-01-19 00:00:00"),
+            to_datetime(f"{current_year}-01-25 23:59:59.999999"),
+        ),
     ]
     assert context.engine_adapter.fetchall(
         "select start_dt, end_dt from sqlmesh_example__pr_env.monthly_model"
     ) == [
-        (to_datetime(f"{current_year}-01-01 00:00:00"), to_datetime(f"{current_year}-01-31 23:59:59.999999")),
+        (
+            to_datetime(f"{current_year}-01-01 00:00:00"),
+            to_datetime(f"{current_year}-01-31 23:59:59.999999"),
+        ),
     ]
     assert context.engine_adapter.fetchall(
         "select start_dt, end_dt from sqlmesh_example__pr_env.ended_daily_model"
     ) == [
-        (to_datetime(f"{current_year}-01-18 00:00:00"), to_datetime(f"{current_year}-01-18 23:59:59.999999")),
+        (
+            to_datetime(f"{current_year}-01-18 00:00:00"),
+            to_datetime(f"{current_year}-01-18 23:59:59.999999"),
+        ),
     ]
 
 
@@ -2688,53 +2703,93 @@ def _get_missing_intervals(name: str) -> t.List[t.Tuple[datetime, datetime]]:
 
     # We only operate on completed intervals, so given the current_time this is the range of the last completed week
     assert _get_missing_intervals("sqlmesh_example.weekly_model") == [
-        (to_datetime(f"{current_year}-01-19 00:00:00"), to_datetime(f"{current_year}-01-26 00:00:00"))
+        (
+            to_datetime(f"{current_year}-01-19 00:00:00"),
+            to_datetime(f"{current_year}-01-26 00:00:00"),
+        )
     ]
 
     # The daily model needs to cover the week, so it gets its start date moved back to line up
     _get_missing_intervals("sqlmesh_example.daily_model") == [
-        (to_datetime(f"{current_year}-01-19 00:00:00"), to_datetime(f"{current_year}-02-01 00:00:00"))
+        (
+            to_datetime(f"{current_year}-01-19 00:00:00"),
+            to_datetime(f"{current_year}-02-01 00:00:00"),
+        )
     ]
 
     # The hourly model needs to cover both the daily model and the weekly model, so it also gets its start date moved back to line up with the weekly model
     assert _get_missing_intervals("sqlmesh_example.hourly_model") == [
-        (to_datetime(f"{current_year}-01-19 00:00:00"), to_datetime(f"{current_year}-02-01 00:00:00"))
+        (
+            to_datetime(f"{current_year}-01-19 00:00:00"),
+            to_datetime(f"{current_year}-02-01 00:00:00"),
+        )
     ]
 
     # The two-hourly model only needs to cover 2 hours and should be unaffected by the fact its sibling node has a weekly child node
     # However it still gets backfilled for 24 hours because the plan start is 1 day and this satisfies min_intervals: 1
     assert _get_missing_intervals("sqlmesh_example.two_hourly_model") == [
-        (to_datetime(f"{current_year}-01-31 00:00:00"), to_datetime(f"{current_year}-02-01 00:00:00"))
+        (
+            to_datetime(f"{current_year}-01-31 00:00:00"),
+            to_datetime(f"{current_year}-02-01 00:00:00"),
+        )
     ]
 
     # The unrelated model has no upstream constraints, so its start date doesnt get moved to line up with the weekly model
     # However it still gets backfilled for 24 hours because the plan start is 1 day and this satisfies min_intervals: 1
     _get_missing_intervals("sqlmesh_example.unrelated_monthly_model") == [
-        (to_datetime(f"{current_year}-01-01 00:00:00"), to_datetime(f"{current_year}-02-01 00:00:00"))
+        (
+            to_datetime(f"{current_year}-01-01 00:00:00"),
+            to_datetime(f"{current_year}-02-01 00:00:00"),
+        )
     ]
 
     # Check that actually running the plan produces the correct result, since missing intervals are re-calculated in the evaluator
     context.apply(plan)
 
     assert context.engine_adapter.fetchall(
         "select min(start_dt), max(end_dt) from sqlmesh_example__pr_env.weekly_model"
-    ) == [(to_datetime(f"{current_year}-01-19 00:00:00"), to_datetime(f"{current_year}-01-25 23:59:59.999999"))]
+    ) == [
+        (
+            to_datetime(f"{current_year}-01-19 00:00:00"),
+            to_datetime(f"{current_year}-01-25 23:59:59.999999"),
+        )
+    ]
 
     assert context.engine_adapter.fetchall(
         "select min(start_dt), max(end_dt) from sqlmesh_example__pr_env.daily_model"
-    ) == [(to_datetime(f"{current_year}-01-19 00:00:00"), to_datetime(f"{current_year}-01-31 23:59:59.999999"))]
+    ) == [
+        (
+            to_datetime(f"{current_year}-01-19 00:00:00"),
+            to_datetime(f"{current_year}-01-31 23:59:59.999999"),
+        )
+    ]
 
     assert context.engine_adapter.fetchall(
         "select min(start_dt), max(end_dt) from sqlmesh_example__pr_env.hourly_model"
-    ) == [(to_datetime(f"{current_year}-01-19 00:00:00"), to_datetime(f"{current_year}-01-31 23:59:59.999999"))]
+    ) == [
+        (
+            to_datetime(f"{current_year}-01-19 00:00:00"),
+            to_datetime(f"{current_year}-01-31 23:59:59.999999"),
+        )
+    ]
 
     assert context.engine_adapter.fetchall(
         "select min(start_dt), max(end_dt) from sqlmesh_example__pr_env.two_hourly_model"
-    ) == [(to_datetime(f"{current_year}-01-31 00:00:00"), to_datetime(f"{current_year}-01-31 23:59:59.999999"))]
+    ) == [
+        (
+            to_datetime(f"{current_year}-01-31 00:00:00"),
+            to_datetime(f"{current_year}-01-31 23:59:59.999999"),
+        )
+    ]
 
     assert context.engine_adapter.fetchall(
         "select min(start_dt), max(end_dt) from sqlmesh_example__pr_env.unrelated_monthly_model"
-    ) == [(to_datetime(f"{current_year}-01-01 00:00:00"), to_datetime(f"{current_year}-01-31 23:59:59.999999"))]
+    ) == [
+        (
+            to_datetime(f"{current_year}-01-01 00:00:00"),
+            to_datetime(f"{current_year}-01-31 23:59:59.999999"),
+        )
+    ]
 
 
 def test_defaults_pre_post_statements(tmp_path: Path):