Merge branch 'TobikoData:main' into doris

Author: xinge-ji

sqlmesh/core/context.py

@@ -1677,6 +1677,11 @@ def plan_builder(
             end_override_per_model=max_interval_end_per_model,
             console=self.console,
             user_provided_flags=user_provided_flags,
+            selected_models={
+                dbt_name
+                for model in model_selector.expand_model_selections(select_models or "*")
+                if (dbt_name := snapshots[model].node.dbt_name)
+            },
             explain=explain or False,
             ignore_cron=ignore_cron or False,
         )

sqlmesh/core/environment.py

@@ -312,6 +312,7 @@ def execute_environment_statements(
     start: t.Optional[TimeLike] = None,
     end: t.Optional[TimeLike] = None,
     execution_time: t.Optional[TimeLike] = None,
+    selected_models: t.Optional[t.Set[str]] = None,
 ) -> None:
     try:
         rendered_expressions = [
@@ -327,6 +328,7 @@ def execute_environment_statements(
                 execution_time=execution_time,
                 environment_naming_info=environment_naming_info,
                 engine_adapter=adapter,
+                selected_models=selected_models,
             )
         ]
     except Exception as e:

sqlmesh/core/node.py

@@ -199,6 +199,7 @@ class _Node(PydanticModel):
     interval_unit_: t.Optional[IntervalUnit] = Field(alias="interval_unit", default=None)
     tags: t.List[str] = []
     stamp: t.Optional[str] = None
+    dbt_name: t.Optional[str] = None  # dbt node name
     _path: t.Optional[Path] = None
     _data_hash: t.Optional[str] = None
     _metadata_hash: t.Optional[str] = None

sqlmesh/core/plan/builder.py

@@ -129,6 +129,7 @@ def __init__(
         end_override_per_model: t.Optional[t.Dict[str, datetime]] = None,
         console: t.Optional[PlanBuilderConsole] = None,
         user_provided_flags: t.Optional[t.Dict[str, UserProvidedFlags]] = None,
+        selected_models: t.Optional[t.Set[str]] = None,
     ):
         self._context_diff = context_diff
         self._no_gaps = no_gaps
@@ -169,6 +170,7 @@ def __init__(
         self._console = console or get_console()
         self._choices: t.Dict[SnapshotId, SnapshotChangeCategory] = {}
         self._user_provided_flags = user_provided_flags
+        self._selected_models = selected_models
         self._explain = explain
 
         self._start = start
@@ -347,6 +349,7 @@ def build(self) -> Plan:
             ensure_finalized_snapshots=self._ensure_finalized_snapshots,
             ignore_cron=self._ignore_cron,
             user_provided_flags=self._user_provided_flags,
+            selected_models=self._selected_models,
         )
         self._latest_plan = plan
         return plan

sqlmesh/core/plan/definition.py

@@ -70,6 +70,8 @@ class Plan(PydanticModel, frozen=True):
     execution_time_: t.Optional[TimeLike] = Field(default=None, alias="execution_time")
 
     user_provided_flags: t.Optional[t.Dict[str, UserProvidedFlags]] = None
+    selected_models: t.Optional[t.Set[str]] = None
+    """Models that have been selected for this plan (used for dbt selected_resources)"""
 
     @cached_property
     def start(self) -> TimeLike:
@@ -282,6 +284,7 @@ def to_evaluatable(self) -> EvaluatablePlan:
             },
             environment_statements=self.context_diff.environment_statements,
             user_provided_flags=self.user_provided_flags,
+            selected_models=self.selected_models,
         )
 
     @cached_property
@@ -319,6 +322,7 @@ class EvaluatablePlan(PydanticModel):
     disabled_restatement_models: t.Set[str]
     environment_statements: t.Optional[t.List[EnvironmentStatements]] = None
     user_provided_flags: t.Optional[t.Dict[str, UserProvidedFlags]] = None
+    selected_models: t.Optional[t.Set[str]] = None
 
     def is_selected_for_backfill(self, model_fqn: str) -> bool:
         return self.models_to_backfill is None or model_fqn in self.models_to_backfill

sqlmesh/core/plan/evaluator.py

@@ -137,6 +137,7 @@ def visit_before_all_stage(self, stage: stages.BeforeAllStage, plan: Evaluatable
             start=plan.start,
             end=plan.end,
             execution_time=plan.execution_time,
+            selected_models=plan.selected_models,
         )
 
     def visit_after_all_stage(self, stage: stages.AfterAllStage, plan: EvaluatablePlan) -> None:
@@ -150,6 +151,7 @@ def visit_after_all_stage(self, stage: stages.AfterAllStage, plan: EvaluatablePl
             start=plan.start,
             end=plan.end,
             execution_time=plan.execution_time,
+            selected_models=plan.selected_models,
         )
 
     def visit_create_snapshot_records_stage(
@@ -257,6 +259,7 @@ def visit_backfill_stage(self, stage: stages.BackfillStage, plan: EvaluatablePla
             allow_destructive_snapshots=plan.allow_destructive_models,
             allow_additive_snapshots=plan.allow_additive_models,
             selected_snapshot_ids=stage.selected_snapshot_ids,
+            selected_models=plan.selected_models,
         )
         if errors:
             raise PlanError("Plan application failed.")

sqlmesh/core/renderer.py

@@ -221,17 +221,19 @@ def _resolve_table(table: str | exp.Table) -> str:
                         dialect=self._dialect, identify=True, comments=False
                     )
 
-                all_refs = list(
-                    self._jinja_macro_registry.global_objs.get("sources", {}).values()  # type: ignore
-                ) + list(
-                    self._jinja_macro_registry.global_objs.get("refs", {}).values()  # type: ignore
-                )
-                for ref in all_refs:
-                    if ref.event_time_filter:
-                        ref.event_time_filter["start"] = render_kwargs["start_tstz"]
-                        ref.event_time_filter["end"] = to_tstz(
-                            make_ts_exclusive(render_kwargs["end_tstz"], dialect=self._dialect)
-                        )
+                if self._model and self._model.kind.is_incremental_by_time_range:
+                    all_refs = list(
+                        self._jinja_macro_registry.global_objs.get("sources", {}).values()  # type: ignore
+                    ) + list(
+                        self._jinja_macro_registry.global_objs.get("refs", {}).values()  # type: ignore
+                    )
+                    for ref in all_refs:
+                        if ref.event_time_filter:
+                            ref.event_time_filter["start"] = render_kwargs["start_tstz"]
+                            ref.event_time_filter["end"] = to_tstz(
+                                make_ts_exclusive(render_kwargs["end_tstz"], dialect=self._dialect)
+                            )
+
                 jinja_env = self._jinja_macro_registry.build_environment(**jinja_env_kwargs)
 
                 expressions = []

sqlmesh/core/scheduler.py

@@ -416,6 +416,7 @@ def run_merged_intervals(
         start: t.Optional[TimeLike] = None,
         end: t.Optional[TimeLike] = None,
         allow_destructive_snapshots: t.Optional[t.Set[str]] = None,
+        selected_models: t.Optional[t.Set[str]] = None,
         allow_additive_snapshots: t.Optional[t.Set[str]] = None,
         selected_snapshot_ids: t.Optional[t.Set[SnapshotId]] = None,
         run_environment_statements: bool = False,
@@ -445,7 +446,13 @@ def run_merged_intervals(
         if not selected_snapshots:
             selected_snapshots = list(merged_intervals)
 
-        snapshot_dag = snapshots_to_dag(selected_snapshots)
+        # Build the full DAG from all snapshots to preserve transitive dependencies
+        full_dag = snapshots_to_dag(self.snapshots.values())
+
+        # Create a subdag that includes the selected snapshots and all their upstream dependencies
+        # This ensures that transitive dependencies are preserved even when intermediate nodes are not selected
+        selected_snapshot_ids_set = {s.snapshot_id for s in selected_snapshots}
+        snapshot_dag = full_dag.subdag(*selected_snapshot_ids_set)
 
         batched_intervals = self.batch_intervals(
             merged_intervals, deployability_index, environment_naming_info, dag=snapshot_dag
@@ -472,6 +479,7 @@ def run_merged_intervals(
                 start=start,
                 end=end,
                 execution_time=execution_time,
+                selected_models=selected_models,
             )
 
         # We only need to create physical tables if the snapshot is not representative or if it
@@ -533,6 +541,7 @@ def run_node(node: SchedulingUnit) -> None:
                             allow_destructive_snapshots=allow_destructive_snapshots,
                             allow_additive_snapshots=allow_additive_snapshots,
                             target_table_exists=snapshot.snapshot_id not in snapshots_to_create,
+                            selected_models=selected_models,
                         )
 
                     evaluation_duration_ms = now_timestamp() - execution_start_ts
@@ -602,6 +611,7 @@ def run_node(node: SchedulingUnit) -> None:
                     start=start,
                     end=end,
                     execution_time=execution_time,
+                    selected_models=selected_models,
                 )
 
             self.state_sync.recycle()
@@ -642,20 +652,11 @@ def _dag(
             upstream_dependencies: t.List[SchedulingUnit] = []
 
             for p_sid in snapshot.parents:
-                if p_sid in self.snapshots:
-                    p_intervals = intervals_per_snapshot.get(p_sid.name, [])
-
-                    if not p_intervals and p_sid in original_snapshots_to_create:
-                        upstream_dependencies.append(CreateNode(snapshot_name=p_sid.name))
-                    elif len(p_intervals) > 1:
-                        upstream_dependencies.append(DummyNode(snapshot_name=p_sid.name))
-                    else:
-                        for i, interval in enumerate(p_intervals):
-                            upstream_dependencies.append(
-                                EvaluateNode(
-                                    snapshot_name=p_sid.name, interval=interval, batch_index=i
-                                )
-                            )
+                upstream_dependencies.extend(
+                    self._find_upstream_dependencies(
+                        p_sid, intervals_per_snapshot, original_snapshots_to_create
+                    )
+                )
 
             batch_concurrency = snapshot.node.batch_concurrency
             batch_size = snapshot.node.batch_size
@@ -699,6 +700,36 @@ def _dag(
                     )
         return dag
 
+    def _find_upstream_dependencies(
+        self,
+        parent_sid: SnapshotId,
+        intervals_per_snapshot: t.Dict[str, Intervals],
+        snapshots_to_create: t.Set[SnapshotId],
+    ) -> t.List[SchedulingUnit]:
+        if parent_sid not in self.snapshots:
+            return []
+
+        p_intervals = intervals_per_snapshot.get(parent_sid.name, [])
+
+        if p_intervals:
+            if len(p_intervals) > 1:
+                return [DummyNode(snapshot_name=parent_sid.name)]
+            interval = p_intervals[0]
+            return [EvaluateNode(snapshot_name=parent_sid.name, interval=interval, batch_index=0)]
+        if parent_sid in snapshots_to_create:
+            return [CreateNode(snapshot_name=parent_sid.name)]
+        # This snapshot has no intervals and doesn't need creation which means
+        # that it can be a transitive dependency
+        transitive_deps: t.List[SchedulingUnit] = []
+        parent_snapshot = self.snapshots[parent_sid]
+        for grandparent_sid in parent_snapshot.parents:
+            transitive_deps.extend(
+                self._find_upstream_dependencies(
+                    grandparent_sid, intervals_per_snapshot, snapshots_to_create
+                )
+            )
+        return transitive_deps
+
     def _run_or_audit(
         self,
         environment: str | EnvironmentNamingInfo,
@@ -808,6 +839,7 @@ def _run_or_audit(
             run_environment_statements=run_environment_statements,
             audit_only=audit_only,
             auto_restatement_triggers=auto_restatement_triggers,
+            selected_models={s.node.dbt_name for s in merged_intervals if s.node.dbt_name},
         )
 
         return CompletionStatus.FAILURE if errors else CompletionStatus.SUCCESS

sqlmesh/core/snapshot/evaluator.py

@@ -50,7 +50,7 @@
     CustomKind,
 )
 from sqlmesh.core.model.kind import _Incremental
-from sqlmesh.utils import CompletionStatus
+from sqlmesh.utils import CompletionStatus, columns_to_types_all_known
 from sqlmesh.core.schema_diff import (
     has_drop_alteration,
     TableAlterOperation,
@@ -747,6 +747,11 @@ def _evaluate_snapshot(
             adapter.execute(model.render_pre_statements(**render_statements_kwargs))
 
             if not target_table_exists or (model.is_seed and not snapshot.intervals):
+                columns_to_types_provided = (
+                    model.kind.is_materialized
+                    and model.columns_to_types_
+                    and columns_to_types_all_known(model.columns_to_types_)
+                )
                 if self._can_clone(snapshot, deployability_index):
                     self._clone_snapshot_in_dev(
                         snapshot=snapshot,
@@ -759,7 +764,7 @@ def _evaluate_snapshot(
                     )
                     runtime_stage = RuntimeStage.EVALUATING
                     target_table_exists = True
-                elif model.annotated or model.is_seed or model.kind.is_scd_type_2:
+                elif columns_to_types_provided or model.is_seed or model.kind.is_scd_type_2:
                     self._execute_create(
                         snapshot=snapshot,
                         table_name=target_table_name,

sqlmesh/dbt/builtin.py

@@ -545,6 +545,7 @@ def create_builtin_globals(
             "run_query": sql_execution.run_query,
             "statement": sql_execution.statement,
             "graph": adapter.graph,
+            "selected_resources": list(jinja_globals.get("selected_models") or []),
         }
     )