sqlmesh.core.node

  1from __future__ import annotations
  2
  3import typing as t
  4import zoneinfo
  5from datetime import datetime
  6from enum import Enum
  7from pathlib import Path
  8
  9from pydantic import Field
 10from sqlglot import exp
 11
 12from sqlmesh.utils.cron import CroniterCache
 13from sqlmesh.utils.date import TimeLike, to_datetime, validate_date_range
 14from sqlmesh.utils.errors import ConfigError
 15from sqlmesh.utils.pydantic import (
 16    PydanticModel,
 17    SQLGlotCron,
 18    field_validator,
 19    model_validator,
 20    PRIVATE_FIELDS,
 21)
 22
 23if t.TYPE_CHECKING:
 24    from sqlmesh.core._typing import Self
 25    from sqlmesh.core.snapshot import Node
 26
 27
 28class IntervalUnit(str, Enum):
 29    """IntervalUnit is the inferred granularity of an incremental node.
 30
 31    IntervalUnit can be one of 5 types, YEAR, MONTH, DAY, HOUR, MINUTE. The unit is inferred
 32    based on the cron schedule of a node. The minimum time delta between a sample set of dates
 33    is used to determine which unit a node's schedule is.
 34
 35    It's designed to align with common partitioning schemes, hence why there is no WEEK unit
 36    because generally tables are not partitioned by week
 37    """
 38
 39    YEAR = "year"
 40    MONTH = "month"
 41    DAY = "day"
 42    HOUR = "hour"
 43    HALF_HOUR = "half_hour"
 44    QUARTER_HOUR = "quarter_hour"
 45    FIVE_MINUTE = "five_minute"
 46
 47    @classmethod
 48    def from_cron(klass, cron: str) -> IntervalUnit:
 49        croniter = CroniterCache(cron)
 50        interval_seconds = croniter.interval_seconds
 51
 52        if not interval_seconds:
 53            samples = [croniter.get_next() for _ in range(5)]
 54            interval_seconds = int(min(b - a for a, b in zip(samples, samples[1:])).total_seconds())
 55
 56        for unit, seconds in INTERVAL_SECONDS.items():
 57            if seconds <= interval_seconds:
 58                return unit
 59        raise ConfigError(f"Invalid cron '{cron}': must run at a frequency of 5 minutes or slower.")
 60
 61    @property
 62    def is_date_granularity(self) -> bool:
 63        return self in (IntervalUnit.YEAR, IntervalUnit.MONTH, IntervalUnit.DAY)
 64
 65    @property
 66    def is_year(self) -> bool:
 67        return self == IntervalUnit.YEAR
 68
 69    @property
 70    def is_month(self) -> bool:
 71        return self == IntervalUnit.MONTH
 72
 73    @property
 74    def is_day(self) -> bool:
 75        return self == IntervalUnit.DAY
 76
 77    @property
 78    def is_hour(self) -> bool:
 79        return self == IntervalUnit.HOUR
 80
 81    @property
 82    def is_minute(self) -> bool:
 83        return self in (IntervalUnit.FIVE_MINUTE, IntervalUnit.QUARTER_HOUR, IntervalUnit.HALF_HOUR)
 84
 85    @property
 86    def cron_expr(self) -> str:
 87        if self == IntervalUnit.FIVE_MINUTE:
 88            return "*/5 * * * *"
 89        if self == IntervalUnit.QUARTER_HOUR:
 90            return "*/15 * * * *"
 91        if self == IntervalUnit.HALF_HOUR:
 92            return "*/30 * * * *"
 93        if self == IntervalUnit.HOUR:
 94            return "0 * * * *"
 95        if self == IntervalUnit.DAY:
 96            return "0 0 * * *"
 97        if self == IntervalUnit.MONTH:
 98            return "0 0 1 * *"
 99        if self == IntervalUnit.YEAR:
100            return "0 0 1 1 *"
101        return ""
102
103    def croniter(self, value: TimeLike) -> CroniterCache:
104        return CroniterCache(self.cron_expr, value)
105
106    def cron_next(self, value: TimeLike, estimate: bool = False) -> datetime:
107        """
108        Get the next timestamp given a time-like value for this interval unit.
109
110        Args:
111            value: A variety of date formats.
112            estimate: Whether or not to estimate, only use this if the value is floored.
113
114        Returns:
115            The timestamp for the next run.
116        """
117        return self.croniter(value).get_next(estimate=estimate)
118
119    def cron_prev(self, value: TimeLike, estimate: bool = False) -> datetime:
120        """
121        Get the previous timestamp given a time-like value for this interval unit.
122
123        Args:
124            value: A variety of date formats.
125            estimate: Whether or not to estimate, only use this if the value is floored.
126
127        Returns:
128            The timestamp for the previous run.
129        """
130        return self.croniter(value).get_prev(estimate=estimate)
131
132    def cron_floor(self, value: TimeLike, estimate: bool = False) -> datetime:
133        """
134        Get the floor timestamp given a time-like value for this interval unit.
135
136        Args:
137            value: A variety of date formats.
138            estimate: Whether or not to estimate, only use this if the value is floored.
139
140        Returns:
141            The timestamp floor.
142        """
143        croniter = self.croniter(value)
144        croniter.get_next(estimate=estimate)
145        return croniter.get_prev(estimate=True)
146
147    @property
148    def seconds(self) -> int:
149        return INTERVAL_SECONDS[self]
150
151    @property
152    def milliseconds(self) -> int:
153        return self.seconds * 1000
154
155
156class DbtNodeInfo(PydanticModel):
157    """
158    Represents dbt-specific model information set by the dbt loader and intended to be made available at the Snapshot level
159    (as opposed to hidden within the individual model jinja macro registries).
160
161    This allows for things like injecting implementations of variables / functions into the Jinja context that are compatible with
162    their dbt equivalents but are backed by the sqlmesh snapshots in any given plan / environment
163    """
164
165    unique_id: str
166    """This is the node/resource name/unique_id that's used as the node key in the dbt manifest.
167    It's prefixed by the resource type and is exposed in context variables like {{ selected_resources }}.
168
169    Examples:
170        - test.jaffle_shop.unique_stg_orders_order_id.e3b841c71a
171        - seed.jaffle_shop.raw_payments
172        - model.jaffle_shop.stg_orders
173    """
174
175    name: str
176    """Name of this object in the dbt global namespace, used by things like {{ ref() }} calls.    
177    
178    Examples:
179        - unique_stg_orders_order_id
180        - raw_payments
181        - stg_orders
182    """
183
184    fqn: str
185    """Used for selectors in --select/--exclude.
186    Takes the filesystem into account so may be structured differently to :unique_id.
187    
188    Examples:
189        - jaffle_shop.staging.unique_stg_orders_order_id
190        - jaffle_shop.raw_payments
191        - jaffle_shop.staging.stg_orders
192    """
193
194    alias: t.Optional[str] = None
195    """This is dbt's way of overriding the _physical table_ a model is written to.
196
197    It's used in the following situation:
198     - Say you have two models, "stg_customers" and "customers"
199     - You want "stg_customers" to be written to the "staging" schema as eg "staging.customers" - NOT "staging.stg_customers"
200     - But you cant rename the file to "customers" because it will conflict with your other model file "customers"
201     - Even if you put it in a different folder, eg "staging/customers.sql" - dbt still has a global namespace so it will conflict
202        when you try to do something like "{{ ref('customers') }}"
203     - So dbt's solution to this problem is to keep calling it "stg_customers" at the dbt project/model level,
204        but allow overriding the physical table to "customers" via something like "{{ config(alias='customers', schema='staging') }}"
205
206    Note that if :alias is set, it does *not* replace :name at the model level and cannot be used interchangably with :name.
207    It also does not affect the :fqn or :unique_id. It's just used to override :name when it comes time to generate the physical table name.
208    """
209
210    @model_validator(mode="after")
211    def post_init(self) -> Self:
212        # by default, dbt sets alias to the same as :name
213        # however, we only want to include :alias if it is actually different / actually providing an override
214        if self.alias == self.name:
215            self.alias = None
216        return self
217
218    def to_expression(self) -> exp.Expr:
219        """Produce a SQLGlot expression representing this object, for use in things like the model/audit definition renderers"""
220        return exp.tuple_(
221            *(
222                exp.PropertyEQ(this=exp.var(k), expression=exp.Literal.string(v))
223                for k, v in sorted(self.model_dump(exclude_none=True).items())
224            )
225        )
226
227
228class DbtInfoMixin:
229    """This mixin encapsulates properties that only exist for dbt compatibility and are otherwise not required
230    for native projects"""
231
232    @property
233    def dbt_node_info(self) -> t.Optional[DbtNodeInfo]:
234        raise NotImplementedError()
235
236    @property
237    def dbt_unique_id(self) -> t.Optional[str]:
238        """Used for compatibility with jinja context variables such as {{ selected_resources }}"""
239        if self.dbt_node_info:
240            return self.dbt_node_info.unique_id
241        return None
242
243    @property
244    def dbt_fqn(self) -> t.Optional[str]:
245        """Used in the selector engine for compatibility with selectors that select models by dbt fqn"""
246        if self.dbt_node_info:
247            return self.dbt_node_info.fqn
248        return None
249
250
251# this must be sorted in descending order
252INTERVAL_SECONDS = {
253    IntervalUnit.YEAR: 60 * 60 * 24 * 365,
254    IntervalUnit.MONTH: 60 * 60 * 24 * 28,
255    IntervalUnit.DAY: 60 * 60 * 24,
256    IntervalUnit.HOUR: 60 * 60,
257    IntervalUnit.HALF_HOUR: 60 * 30,
258    IntervalUnit.QUARTER_HOUR: 60 * 15,
259    IntervalUnit.FIVE_MINUTE: 60 * 5,
260}
261
262
263def _cron_tz_validator(cls: t.Type, v: t.Any) -> t.Optional[zoneinfo.ZoneInfo]:
264    if not v or v == "UTC":
265        return None
266
267    v = str_or_exp_to_str(v)
268
269    try:
270        return zoneinfo.ZoneInfo(v)
271    except Exception as e:
272        available_timezones = zoneinfo.available_timezones()
273
274        if available_timezones:
275            raise ConfigError(f"{e}. {v} must be in {available_timezones}.")
276        else:
277            raise ConfigError(
278                f"{e}. IANA time zone data is not available on your system. `pip install tzdata` to leverage cron time zones or remove this field which will default to UTC."
279            )
280
281    return None
282
283
284cron_tz_validator = field_validator("cron_tz", mode="before")(_cron_tz_validator)
285
286
287class _Node(DbtInfoMixin, PydanticModel):
288    """
289    Node is the core abstraction for entity that can be executed within the scheduler.
290
291    Args:
292        name: The name of the node.
293        project: The name of the project this node belongs to, used in multi-repo deployments.
294        description: The optional node description.
295        owner: The owner of the node.
296        start: The earliest date that the node will be executed for. If this is None,
297            then the date is inferred by taking the most recent start date of its ancestors.
298            The start date can be a static datetime or a relative datetime like "1 year ago"
299        end: The latest date that the model will be executed for. If this is None,
300            the date from the scheduler will be used
301        cron: A cron string specifying how often the node should be run, leveraging the
302            [croniter](https://github.com/kiorky/croniter) library.
303        cron_tz: Time zone for the cron, defaults to utc, [IANA time zones](https://docs.python.org/3/library/zoneinfo.html).
304        interval_unit: The duration of an interval for the node. By default, it is computed from the cron expression.
305        tags: A list of tags that can be used to filter nodes.
306        stamp: An optional arbitrary string sequence used to create new node versions without making
307            changes to any of the functional components of the definition.
308    """
309
310    name: str
311    project: str = ""
312    description: t.Optional[str] = None
313    owner: t.Optional[str] = None
314    start: t.Optional[TimeLike] = None
315    end: t.Optional[TimeLike] = None
316    cron: SQLGlotCron = "@daily"
317    cron_tz: t.Optional[zoneinfo.ZoneInfo] = None
318    interval_unit_: t.Optional[IntervalUnit] = Field(alias="interval_unit", default=None)
319    tags: t.List[str] = []
320    stamp: t.Optional[str] = None
321    dbt_node_info_: t.Optional[DbtNodeInfo] = Field(alias="dbt_node_info", default=None)
322    _path: t.Optional[Path] = None
323    _data_hash: t.Optional[str] = None
324    _metadata_hash: t.Optional[str] = None
325
326    _croniter: t.Optional[CroniterCache] = None
327    __inferred_interval_unit: t.Optional[IntervalUnit] = None
328
329    _cron_tz_validator = cron_tz_validator
330
331    def __str__(self) -> str:
332        path = f": {self._path.name}" if self._path else ""
333        return f"{self.__class__.__name__}<{self.name}{path}>"
334
335    def __getstate__(self) -> t.Dict[t.Any, t.Any]:
336        state = super().__getstate__()
337        private = state[PRIVATE_FIELDS]
338        private["_data_hash"] = None
339        private["_metadata_hash"] = None
340        return state
341
342    def copy(self, **kwargs: t.Any) -> Self:
343        node = super().copy(**kwargs)
344        node._data_hash = None
345        node._metadata_hash = None
346        return node
347
348    @field_validator("name", mode="before")
349    @classmethod
350    def _name_validator(cls, v: t.Any) -> t.Optional[str]:
351        if v is None:
352            return None
353        if isinstance(v, exp.Expr):
354            return v.meta["sql"]
355        return str(v)
356
357    @field_validator("start", "end", mode="before")
358    @classmethod
359    def _date_validator(cls, v: t.Any) -> t.Optional[TimeLike]:
360        if isinstance(v, exp.Expr):
361            v = v.name
362        if v and not to_datetime(v):
363            raise ConfigError(f"'{v}' needs to be time-like: https://pypi.org/project/dateparser")
364        return v
365
366    @field_validator("owner", "description", "stamp", mode="before")
367    @classmethod
368    def _string_expr_validator(cls, v: t.Any) -> t.Optional[str]:
369        return str_or_exp_to_str(v)
370
371    @field_validator("interval_unit_", mode="before")
372    @classmethod
373    def _interval_unit_validator(cls, v: t.Any) -> t.Optional[t.Union[IntervalUnit, str]]:
374        if isinstance(v, IntervalUnit):
375            return v
376        v = str_or_exp_to_str(v)
377        if v:
378            v = v.lower()
379        return v
380
381    @model_validator(mode="after")
382    def _node_root_validator(self) -> Self:
383        interval_unit = self.interval_unit_
384        if interval_unit and not getattr(self, "allow_partials", None):
385            cron = self.cron
386            max_interval_unit = IntervalUnit.from_cron(cron)
387            if interval_unit.seconds > max_interval_unit.seconds:
388                raise ConfigError(
389                    f"Cron '{cron}' cannot be more frequent than interval unit '{interval_unit.value}'. "
390                    "If this is intentional, set allow_partials to True."
391                )
392
393        start = self.start
394        end = self.end
395
396        if end is not None and start is None:
397            raise ConfigError("Must define a start date if an end date is defined.")
398        validate_date_range(start, end)
399        return self
400
401    @property
402    def batch_size(self) -> t.Optional[int]:
403        """The maximal number of units in a single task for a backfill."""
404        return None
405
406    @property
407    def batch_concurrency(self) -> t.Optional[int]:
408        """The maximal number of batches that can run concurrently for a backfill."""
409        return None
410
411    @property
412    def interval_unit(self) -> IntervalUnit:
413        """Returns the interval unit using which data intervals are computed for this node."""
414        if self.interval_unit_ is not None:
415            return self.interval_unit_
416        return self._inferred_interval_unit()
417
418    @property
419    def depends_on(self) -> t.Set[str]:
420        return set()
421
422    @property
423    def fqn(self) -> str:
424        return self.name
425
426    @property
427    def data_hash(self) -> str:
428        """
429        Computes the data hash for the node.
430
431        Returns:
432            The data hash for the node.
433        """
434        raise NotImplementedError
435
436    @property
437    def metadata_hash(self) -> str:
438        """
439        Computes the metadata hash for the node.
440
441        Returns:
442            The metadata hash for the node.
443        """
444        raise NotImplementedError
445
446    def is_metadata_only_change(self, previous: _Node) -> bool:
447        """Determines if this node is a metadata only change in relation to the `previous` node.
448
449        Args:
450            previous: The previous node to compare against.
451
452        Returns:
453            True if this node is a metadata only change, False otherwise.
454        """
455        return self.data_hash == previous.data_hash and self.metadata_hash != previous.metadata_hash
456
457    def is_data_change(self, previous: _Node) -> bool:
458        """Determines if this node is a data change in relation to the `previous` node.
459
460        Args:
461            previous: The previous node to compare against.
462
463        Returns:
464            True if this node is a data change, False otherwise.
465        """
466        return (
467            self.data_hash != previous.data_hash or self.metadata_hash != previous.metadata_hash
468        ) and not self.is_metadata_only_change(previous)
469
470    def croniter(self, value: TimeLike) -> CroniterCache:
471        if self._croniter is None:
472            self._croniter = CroniterCache(self.cron, value, tz=self.cron_tz)
473        else:
474            self._croniter.curr = to_datetime(value, tz=self.cron_tz)
475        return self._croniter
476
477    def cron_next(self, value: TimeLike, estimate: bool = False) -> datetime:
478        """
479        Get the next timestamp given a time-like value and the node's cron.
480
481        Args:
482            value: A variety of date formats.
483            estimate: Whether or not to estimate, only use this if the value is floored.
484
485        Returns:
486            The timestamp for the next run.
487        """
488        return self.croniter(value).get_next(estimate=estimate)
489
490    def cron_prev(self, value: TimeLike, estimate: bool = False) -> datetime:
491        """
492        Get the previous timestamp given a time-like value and the node's cron.
493
494        Args:
495            value: A variety of date formats.
496            estimate: Whether or not to estimate, only use this if the value is floored.
497
498        Returns:
499            The timestamp for the previous run.
500        """
501        return self.croniter(value).get_prev(estimate=estimate)
502
503    def cron_floor(self, value: TimeLike, estimate: bool = False) -> datetime:
504        """
505        Get the floor timestamp given a time-like value and the node's cron.
506
507        Args:
508            value: A variety of date formats.
509            estimate: Whether or not to estimate, only use this if the value is floored.
510
511        Returns:
512            The timestamp floor.
513        """
514        return self.croniter(self.cron_next(value, estimate=estimate)).get_prev(estimate=True)
515
516    def text_diff(self, other: Node, rendered: bool = False) -> str:
517        """Produce a text diff against another node.
518
519        Args:
520            other: The node to diff against. Must be of the same type.
521
522        Returns:
523            A unified text diff showing additions and deletions.
524        """
525        raise NotImplementedError
526
527    def _inferred_interval_unit(self) -> IntervalUnit:
528        """Infers the interval unit from the cron expression.
529
530        The interval unit is used to determine the lag applied to start_date and end_date for node rendering and intervals.
531
532        Returns:
533            The IntervalUnit enum.
534        """
535        if not self.__inferred_interval_unit:
536            self.__inferred_interval_unit = IntervalUnit.from_cron(self.cron)
537        return self.__inferred_interval_unit
538
539    @property
540    def is_model(self) -> bool:
541        """Return True if this is a model node"""
542        return False
543
544    @property
545    def is_audit(self) -> bool:
546        """Return True if this is an audit node"""
547        return False
548
549    @property
550    def dbt_node_info(self) -> t.Optional[DbtNodeInfo]:
551        return self.dbt_node_info_
552
553
554class NodeType(str, Enum):
555    MODEL = "model"
556    AUDIT = "audit"
557
558    def __str__(self) -> str:
559        return self.name
560
561
562def str_or_exp_to_str(v: t.Any) -> t.Optional[str]:
563    if isinstance(v, exp.Expr):
564        return v.name
565    return str(v) if v is not None else None