Merge remote-tracking branch 'origin/main' into feature/DRM/revert-to-shared-connection-for-databricks-access-token-connections

Author: davem-bis

docs/concepts/overview.md

@@ -68,4 +68,4 @@ SQLMesh automatically runs audits when you apply a `plan` to an environment, or
 ## Infrastructure and orchestration
 Every company's data infrastructure is different. SQLMesh is flexible with regard to which engines and orchestration frameworks you use — its only requirement is access to the target SQL/analytics engine.
 
-SQLMesh keeps track of model versions and processed data intervals using your existing infrastructure. SQLMesh it automatically creates a `sqlmesh` schema in your data warehouse for its internal metadata.
+SQLMesh keeps track of model versions and processed data intervals using your existing infrastructure. It automatically creates a `sqlmesh` schema in your data warehouse for its internal metadata.

docs/guides/configuration.md

@@ -170,6 +170,16 @@ The examples specify a Snowflake connection whose password is stored in an envir
           account: <account>
     ```
 
+    !!! tip "Base64-encoded secrets"
+
+        If a secret is distributed base64-encoded in a single environment variable (for example a BigQuery service-account key), pipe the variable through the built-in `b64decode` filter to decode it to text inline:
+
+        ```yaml
+        keyfile_json: {{ env_var('BIGQUERY_KEY_B64') | b64decode }}
+        ```
+
+        A matching `b64encode` filter is also available. Both return UTF-8 text, so they are intended for string/JSON secrets rather than arbitrary binary data.
+
 === "Python"
 
     Python accesses environment variables via the `os` library's `environ` dictionary.

docs/integrations/engines/databricks.md

@@ -310,3 +310,25 @@ MODEL (
 
 If you attempt to alter without having this property set, you will get an error similar to `databricks.sql.exc.ServerOperationError: [DELTA_UNSUPPORTED_DROP_COLUMN] DROP COLUMN is not supported for your Delta table.`.
 [Databricks Documentation for more details](https://docs.databricks.com/en/delta/column-mapping.html#requirements).
+
+## Liquid Clustering
+
+SQLMesh supports the liquid clustering keywords AUTO and NONE
+
+```sql
+MODEL (
+    name sqlmesh_example.new_model,
+    ...
+    clustered_by AUTO
+)
+```
+
+To cluster by a column called `auto` or `none`, use parentheses and backticks
+
+```sql
+MODEL (
+    name sqlmesh_example.new_model,
+    ...
+    clustered_by (`auto`)
+)
+```

sqlmesh/core/constants.py

@@ -96,3 +96,5 @@
 HYBRID = "hybrid"
 
 DISABLE_SQLMESH_STATE_MIGRATION = "SQLMESH__AIRFLOW__DISABLE_STATE_MIGRATION"
+
+LIQUID_CLUSTERING_KEYWORDS: frozenset = frozenset({"AUTO", "NONE"})

sqlmesh/core/dialect.py

@@ -24,7 +24,7 @@
 from sqlglot.schema import MappingSchema
 from sqlglot.tokens import Token
 
-from sqlmesh.core.constants import MAX_MODEL_DEFINITION_SIZE
+from sqlmesh.core.constants import LIQUID_CLUSTERING_KEYWORDS, MAX_MODEL_DEFINITION_SIZE
 from sqlmesh.utils import get_source_columns_to_types
 from sqlmesh.utils.errors import SQLMeshError, ConfigError
 from sqlmesh.utils.pandas import columns_to_types_from_df
@@ -663,6 +663,27 @@ def parse(self: Parser) -> t.Optional[exp.Expr]:
                     value = exp.tuple_(*partitioned_by.this.expressions)
                 else:
                     value = partitioned_by.this
+            elif key == "clustered_by":
+                # Bare AUTO / NONE are Databricks liquid clustering keywords, not column refs.
+                # Detect keywords by token type: unquoted bare identifiers arrive as VAR tokens.
+                # Backtick-quoted identifiers (e.g. `auto`) have IDENTIFIER token type and are
+                # treated as real column names.
+                if (
+                    self._curr is not None
+                    and self._curr.token_type == TokenType.VAR
+                    and self._curr.text.upper() in LIQUID_CLUSTERING_KEYWORDS
+                ):
+                    value = exp.Var(this=self._curr.text.upper())
+                    self._advance()
+                else:
+                    parsed = self._parse_bracket(self._parse_field(any_token=True))
+                    # Unwrap Paren wrapping a bare column to match partitioned_by normalisation:
+                    # clustered_by (a) → stored as Column(a), not Paren(Column(a)).
+                    # Preserve parens around function expressions: (TO_DATE(col)) stays as-is.
+                    if isinstance(parsed, exp.Paren) and isinstance(parsed.this, exp.Column):
+                        value = parsed.unnest()
+                    else:
+                        value = parsed
             else:
                 value = self._parse_bracket(self._parse_field(any_token=True))
 
@@ -783,7 +804,10 @@ def format_model_expressions(
         A string representing the formatted model.
     """
     if len(expressions) == 1 and is_meta_expression(expressions[0]):
-        return expressions[0].sql(pretty=True, dialect=dialect)
+        # Meta expressions (MODEL/AUDIT/METRIC) are SQLMesh DDL, not standard SQL,
+        # so they must never be transpiled to the target dialect (e.g. tsql would
+        # rewrite a boolean property like `allow_partials TRUE` to `(1 = 1)`).
+        return expressions[0].sql(pretty=True, dialect=None)
 
     if rewrite_casts:
 
@@ -815,7 +839,14 @@ def cast_to_colon(node: exp.Expr) -> exp.Expr:
         expressions = new_expressions
 
     return ";\n\n".join(
-        expression.sql(pretty=True, dialect=dialect, **kwargs) for expression in expressions
+        # Meta expressions (MODEL/AUDIT/METRIC) are SQLMesh DDL and must stay
+        # dialect-agnostic; only the actual query/statement expressions transpile.
+        expression.sql(
+            pretty=True,
+            dialect=None if is_meta_expression(expression) else dialect,
+            **kwargs,
+        )
+        for expression in expressions
     ).strip()
 
 

sqlmesh/core/engine_adapter/databricks.py

@@ -6,6 +6,7 @@
 
 from sqlglot import exp
 
+from sqlmesh.core.constants import LIQUID_CLUSTERING_KEYWORDS
 from sqlmesh.core.dialect import to_schema
 from sqlmesh.core.engine_adapter.mixins import GrantsFromInfoSchemaMixin
 from sqlmesh.core.engine_adapter.shared import (
@@ -442,10 +443,16 @@ def _build_table_properties_exp(
             table_kind=table_kind,
         )
         if clustered_by:
-            # Databricks expects wrapped CLUSTER BY expressions
-            clustered_by_exp = exp.Cluster(
-                expressions=[exp.Tuple(expressions=[c.copy() for c in clustered_by])]
-            )
+            if len(clustered_by) == 1 and isinstance(clustered_by[0], exp.Var):
+                if clustered_by[0].name.upper() not in LIQUID_CLUSTERING_KEYWORDS:
+                    raise ValueError(f"Unexpected bare Var in clustered_by: {clustered_by[0]!r}")
+                # exp.Cluster with a bare Var generates: CLUSTER BY AUTO (no parens)
+                clustered_by_exp = exp.Cluster(expressions=[clustered_by[0].copy()])
+            else:
+                # Databricks expects column expressions wrapped in a tuple
+                clustered_by_exp = exp.Cluster(
+                    expressions=[exp.Tuple(expressions=[c.copy() for c in clustered_by])]
+                )
             expressions = properties.expressions if properties else []
             expressions.append(clustered_by_exp)
             properties = exp.Properties(expressions=expressions)

sqlmesh/core/engine_adapter/fabric.py

@@ -9,6 +9,8 @@
 from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_result
 from sqlmesh.core.engine_adapter.mssql import MSSQLEngineAdapter
 from sqlmesh.core.engine_adapter.shared import (
+    CommentCreationTable,
+    CommentCreationView,
     InsertOverwriteStrategy,
 )
 from sqlmesh.utils.errors import SQLMeshError
@@ -30,6 +32,10 @@ class FabricEngineAdapter(MSSQLEngineAdapter):
     SUPPORTS_TRANSACTIONS = False
     SUPPORTS_CREATE_DROP_CATALOG = True
     INSERT_OVERWRITE_STRATEGY = InsertOverwriteStrategy.DELETE_INSERT
+    # There is no standard method to handle comments in Fabric for now, so we disable it.
+    # Otherwise, it would be inherited from MSSQL and would not work.
+    COMMENT_CREATION_TABLE = CommentCreationTable.UNSUPPORTED
+    COMMENT_CREATION_VIEW = CommentCreationView.UNSUPPORTED
 
     def __init__(
         self, connection_factory_or_pool: t.Union[t.Callable, t.Any], *args: t.Any, **kwargs: t.Any

sqlmesh/core/model/definition.py

@@ -992,6 +992,12 @@ def validate_definition(self) -> None:
                 values = [
                     col.name
                     for expr in values
+                    if not (
+                        field == "clustered_by"
+                        and (self.dialect or "").lower() == "databricks"
+                        and isinstance(expr, exp.Var)
+                        and expr.name.upper() in c.LIQUID_CLUSTERING_KEYWORDS
+                    )
                     for col in t.cast(
                         exp.Expr, exp.maybe_parse(expr, dialect=self.dialect)
                     ).find_all(exp.Column)
@@ -1588,7 +1594,7 @@ def is_breaking_change(self, previous: Model) -> t.Optional[bool]:
 
         for edit in edits:
             if not isinstance(edit, Insert):
-                return None
+                return _additive_projection_change(previous_query, this_query, self.dialect)
 
             expr = edit.expression
             if isinstance(expr, exp.UDTF):
@@ -1602,7 +1608,7 @@ def is_breaking_change(self, previous: Model) -> t.Optional[bool]:
                 expr = parent
 
             if not _is_projection(expr) and expr.parent not in inserted_expressions:
-                return None
+                return _additive_projection_change(previous_query, this_query, self.dialect)
 
         return False
 
@@ -2907,6 +2913,98 @@ def _is_projection(expr: exp.Expr) -> bool:
     return isinstance(parent, exp.Select) and expr.arg_key == "expressions"
 
 
+def _has_ordinal_references(query: exp.Select) -> bool:
+    order = query.args.get("order")
+    if order and any(
+        isinstance(ob.this, exp.Literal) and ob.this.is_number for ob in order.expressions
+    ):
+        return True
+    group = query.args.get("group")
+    return bool(
+        group and any(isinstance(gb, exp.Literal) and gb.is_number for gb in group.expressions)
+    )
+
+
+def _additive_projection_change(
+    previous_query: exp.Query,
+    this_query: exp.Query,
+    dialect: DialectType,
+) -> t.Optional[bool]:
+    """Fallback for when SQLGlot's tree diff can't express an additive projection change.
+
+    SQLGlot's diff matches nodes by structural similarity, so interchangeable leaves (e.g. two
+    identical ``CAST(... AS T)`` target types) can be cross-matched. Inserting a same-type cast
+    above an existing one therefore yields spurious ``Move`` / ``Update`` edits even though a
+    column was simply added to the SELECT list. In that case the edit-based check above is
+    inconclusive, so we verify additivity directly against the output projections.
+
+    Returns ``False`` (non-breaking) only when the change is provably additive:
+      * both queries are simple ``SELECT`` statements,
+      * everything other than the projection list is structurally identical,
+      * no added projection is a (potentially cardinality-changing) ``UDTF``,
+      * every previous projection is preserved, in order, within the new projection list, and
+      * no mid-list insert shifts ordinal ``ORDER BY`` / ``GROUP BY`` references.
+
+    Otherwise returns ``None`` (undetermined), preserving the conservative default.
+    """
+    # UNIONs or other query expressions, are left to the caller's conservative diff result.
+    if not isinstance(previous_query, exp.Select) or not isinstance(this_query, exp.Select):
+        return None
+
+    previous_projections = previous_query.expressions
+    this_projections = this_query.expressions
+    # If the new query has not gained any projections, this cannot be an additive projection-only
+    # change, so there is nothing for this fallback to prove.
+    if len(this_projections) <= len(previous_projections):
+        return None
+
+    # Adding a UDTF projection (e.g. EXPLODE / UNNEST) can change row cardinality, so such a
+    # change is not safely non-breaking even when it appears as an extra SELECT item.
+    for projection in this_projections:
+        bare = projection.this if isinstance(projection, exp.Alias) else projection
+        if isinstance(bare, exp.UDTF):
+            return None
+
+    # Everything other than the projection list must be structurally identical. Replacing each
+    # SELECT list with the same dummy literal lets the expression equality check focus on the
+    # FROM / WHERE / GROUP BY / ORDER BY / etc. parts of the query.
+    previous_skeleton = previous_query.copy()
+    this_skeleton = this_query.copy()
+    previous_skeleton.set("expressions", [exp.Literal.number(1)])
+    this_skeleton.set("expressions", [exp.Literal.number(1)])
+    if previous_skeleton != this_skeleton:
+        return None
+
+    # Every previous projection must appear, in order, within the new projection list. Comparing
+    # dialect-normalized SQL makes semantically equivalent projection nodes match even when the
+    # parser built distinct object identities.
+    this_projection_sql = [p.sql(dialect=dialect, comments=False) for p in this_projections]
+    search_start = 0
+    matched_at: list[int] = []
+    for projection in previous_projections:
+        target_sql = projection.sql(dialect=dialect, comments=False)
+        # Continue after the previous match so added columns can appear before, between, or after
+        # the original projections, but existing projections cannot be reordered or rewritten.
+        for index in range(search_start, len(this_projection_sql)):
+            if this_projection_sql[index] == target_sql:
+                matched_at.append(index)
+                search_start = index + 1
+                break
+        else:
+            return None
+
+    # Mid-list inserts shift ordinal references in ORDER BY / GROUP BY clauses.
+    if _has_ordinal_references(this_query):
+        matched_set = set(matched_at)
+        last_matched = matched_at[-1]
+        if any(i < last_matched for i in range(len(this_projections)) if i not in matched_set):
+            return None
+
+    # At this point the query shape is unchanged and all prior outputs are preserved, so the only
+    # remaining difference is one or more additional, non-UDTF projections.
+    return False
+
+
 def _single_expr_or_tuple(values: t.Sequence[exp.Expr]) -> exp.Expr | exp.Tuple:
     return values[0] if len(values) == 1 else exp.Tuple(expressions=values)
 

sqlmesh/core/model/meta.py

@@ -12,6 +12,7 @@
 
 from sqlmesh.core import dialect as d
 from sqlmesh.core.config.common import VirtualEnvironmentMode
+from sqlmesh.core.constants import LIQUID_CLUSTERING_KEYWORDS
 from sqlmesh.core.config.linter import LinterConfig
 from sqlmesh.core.dialect import normalize_model_name
 from sqlmesh.utils import classproperty
@@ -190,10 +191,13 @@ def _gateway_validator(cls, v: t.Any) -> t.Optional[str]:
 
     @field_validator("partitioned_by_", "clustered_by", mode="before")
     def _partition_and_cluster_validator(cls, v: t.Any, info: ValidationInfo) -> t.List[exp.Expr]:
+        field = info.field_name or ""
+        dialect = (get_dialect(info) or "").lower()
+
         if (
             isinstance(v, list)
             and all(isinstance(i, str) for i in v)
-            and (info.field_name or "") == "partitioned_by_"
+            and field == "partitioned_by_"
         ):
             # this branch gets hit when we are deserializing from json because `partitioned_by` is stored as a List[str]
             # however, we should only invoke this if the list contains strings because this validator is also
@@ -206,9 +210,33 @@ def _partition_and_cluster_validator(cls, v: t.Any, info: ValidationInfo) -> t.L
             )
             v = parsed.this.expressions if isinstance(parsed.this, exp.Schema) else v
 
+        if isinstance(v, str) and field == "clustered_by":
+            v = [v]
+
+        if isinstance(v, list) and field == "clustered_by" and dialect == "databricks":
+            # When deserializing from JSON, clustered_by is stored as List[str].
+            # Restore keyword sentinels (AUTO/NONE) before list_of_fields_validator normalises
+            # them into quoted columns.
+            v = [
+                exp.Var(this=item.upper())
+                if isinstance(item, str) and item.upper() in LIQUID_CLUSTERING_KEYWORDS
+                else item
+                for item in v
+            ]
+
         expressions = list_of_fields_validator(v, validation_data(info))
 
         for expression in expressions:
+            # AUTO and NONE are Databricks liquid clustering keywords, not column references.
+            # Only skip for clustered_by with the Databricks dialect — meaningless elsewhere.
+            if (
+                field == "clustered_by"
+                and dialect == "databricks"
+                and isinstance(expression, exp.Var)
+                and expression.name.upper() in LIQUID_CLUSTERING_KEYWORDS
+            ):
+                continue
+
             num_cols = len(list(expression.find_all(exp.Column)))
 
             error_msg: t.Optional[str] = None

sqlmesh/integrations/dlt.py

@@ -208,7 +208,7 @@ def generate_incremental_model(
 FROM
   {from_clause}
 WHERE
-  {time_column} BETWEEN @start_ds AND @end_ds
+  {time_column} BETWEEN @start_ts AND @end_ts
 """