Merge branch 'main' into feat/clustered-by-auto-none

EhabEasee · web-flow · commit fb8c119e5300 · 2026-06-30T20:37:30.000+02:00
diff --git a/docs/concepts/overview.md b/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 &mdash; 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.
diff --git a/docs/guides/configuration.md b/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.
diff --git a/sqlmesh/core/dialect.py b/sqlmesh/core/dialect.py
@@ -804,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:
 
@@ -836,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()
 
 
diff --git a/sqlmesh/core/engine_adapter/fabric.py b/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
diff --git a/sqlmesh/integrations/dlt.py b/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
 """
 
 
diff --git a/sqlmesh/utils/jinja.py b/sqlmesh/utils/jinja.py
@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import base64
 import importlib
 import json
 import re
@@ -28,11 +29,40 @@
 SQLMESH_JINJA_PACKAGE = "sqlmesh.utils.jinja"
 
 
+def b64decode(value: t.Union[str, bytes]) -> str:
+    """Decode a base64-encoded value and return it as UTF-8 text.
+
+    Intended for base64-encoded string/JSON secrets (for example a service-account
+    key stored in an environment variable), not arbitrary binary payloads.
+    """
+    decoded = value.encode("utf-8") if isinstance(value, str) else value
+    return base64.b64decode(decoded).decode("utf-8")
+
+
+def b64encode(value: t.Union[str, bytes]) -> str:
+    """Base64-encode a value and return the encoding as UTF-8 text.
+
+    The input is treated as UTF-8 text, mirroring ``b64decode``; it is intended for
+    string/JSON secrets rather than arbitrary binary payloads.
+    """
+    encoded = value.encode("utf-8") if isinstance(value, str) else value
+    return base64.b64encode(encoded).decode("utf-8")
+
+
+def create_builtin_filters() -> t.Dict[str, t.Callable]:
+    return {
+        "b64decode": b64decode,
+        "b64encode": b64encode,
+    }
+
+
 def environment(**kwargs: t.Any) -> Environment:
     extensions = kwargs.pop("extensions", [])
     extensions.append("jinja2.ext.do")
     extensions.append("jinja2.ext.loopcontrols")
-    return Environment(extensions=extensions, **kwargs)
+    env = Environment(extensions=extensions, **kwargs)
+    env.filters.update(create_builtin_filters())
+    return env
 
 
 ENVIRONMENT = environment()
diff --git a/tests/cli/test_cli.py b/tests/cli/test_cli.py
@@ -921,7 +921,7 @@ def test_dlt_filesystem_pipeline(tmp_path):
 FROM
   filesystem_pipeline_dataset.equipment as c
 WHERE
-  TO_TIMESTAMP(CAST(c._dlt_load_id AS DOUBLE)) BETWEEN @start_ds AND @end_ds
+  TO_TIMESTAMP(CAST(c._dlt_load_id AS DOUBLE)) BETWEEN @start_ts AND @end_ts
 """
 
     with open(equipment_model_path) as file:
@@ -1064,7 +1064,7 @@ def test_dlt_pipeline(runner, tmp_path):
 FROM
   sushi_dataset.sushi_types as c
 WHERE
-  TO_TIMESTAMP(CAST(c._dlt_load_id AS DOUBLE)) BETWEEN @start_ds AND @end_ds
+  TO_TIMESTAMP(CAST(c._dlt_load_id AS DOUBLE)) BETWEEN @start_ts AND @end_ts
 """
 
     dlt_sushi_types_model_path = tmp_path / "models/incremental_sushi_types.sql"
@@ -1095,7 +1095,7 @@ def test_dlt_pipeline(runner, tmp_path):
 FROM
   sushi_dataset._dlt_loads as c
 WHERE
-  TO_TIMESTAMP(CAST(c.load_id AS DOUBLE)) BETWEEN @start_ds AND @end_ds
+  TO_TIMESTAMP(CAST(c.load_id AS DOUBLE)) BETWEEN @start_ts AND @end_ts
 """
 
     with open(dlt_loads_model_path) as file:
@@ -1122,7 +1122,7 @@ def test_dlt_pipeline(runner, tmp_path):
 ON
   c._dlt_parent_id = p._dlt_id
 WHERE
-  TO_TIMESTAMP(CAST(p._dlt_load_id AS DOUBLE)) BETWEEN @start_ds AND @end_ds
+  TO_TIMESTAMP(CAST(p._dlt_load_id AS DOUBLE)) BETWEEN @start_ts AND @end_ts
 """
 
     with open(dlt_sushi_fillings_model_path) as file:
diff --git a/tests/core/engine_adapter/test_fabric.py b/tests/core/engine_adapter/test_fabric.py
@@ -286,3 +286,23 @@ def test_merge_exists(
         f"MERGE INTO [target] AS [__MERGE_TARGET__] USING (SELECT CAST([id] AS INT) AS [id], CAST([ts] AS DATETIME2(6)) AS [ts] FROM [__temp_target_{temp_table_id}]) AS [__MERGE_SOURCE__] ON [__MERGE_TARGET__].[id] = [__MERGE_SOURCE__].[id] AND [__MERGE_TARGET__].[ts] = [__MERGE_SOURCE__].[ts] WHEN NOT MATCHED THEN INSERT ([id], [ts]) VALUES ([__MERGE_SOURCE__].[id], [__MERGE_SOURCE__].[ts]);",
         f"DROP TABLE IF EXISTS [__temp_target_{temp_table_id}];",
     ]
+
+
+def test_comments(make_mocked_engine_adapter: t.Callable, mocker: MockerFixture):
+    adapter = make_mocked_engine_adapter(FabricEngineAdapter)
+    comment = "\\"
+
+    create_table_comment_mock = mocker.patch.object(adapter, "_create_table_comment")
+    create_column_comments_mock = mocker.patch.object(adapter, "_create_column_comments")
+    mocker.patch.object(adapter, "_create_table")
+
+    adapter.create_table(
+        "test_table",
+        {"a": exp.DataType.build("INT"), "b": exp.DataType.build("INT")},
+        table_description=comment,
+        column_descriptions={"a": comment},
+    )
+
+    create_table_comment_mock.assert_not_called()
+    create_column_comments_mock.assert_not_called()
+    assert to_sql_calls(adapter) == []
diff --git a/tests/core/engine_adapter/test_mssql.py b/tests/core/engine_adapter/test_mssql.py
@@ -1002,3 +1002,24 @@ def python_scd2_model(context, **kwargs):
     snapshot: Snapshot = make_snapshot(m)
     assert snapshot.node.physical_properties == m.physical_properties
     assert snapshot.node.physical_properties.get("mssql_merge_exists")
+
+
+def test_comments(make_mocked_engine_adapter: t.Callable, mocker: MockerFixture):
+    adapter = make_mocked_engine_adapter(MSSQLEngineAdapter)
+    table = exp.to_table("test_table")
+    comment = "\\"
+
+    mocker.patch.object(adapter, "_create_table")
+
+    adapter.create_table(
+        "test_table",
+        {"a": exp.DataType.build("INT"), "b": exp.DataType.build("INT")},
+        table_description=comment,
+        column_descriptions={"a": comment},
+    )
+
+    sql_calls = to_sql_calls(adapter)
+    assert sql_calls == [
+        adapter._build_create_comment_table_exp(table, comment),
+        adapter._build_create_comment_column_exp(table, "a", comment),
+    ]
diff --git a/tests/core/test_dialect.py b/tests/core/test_dialect.py
@@ -230,6 +230,60 @@ def test_format_model_expressions():
   1::Int32 AS bar"""
     )
 
+    x = format_model_expressions(
+        parse(
+            """
+            MODEL(name a.b, kind FULL, dialect tsql, allow_partials true);
+            SELECT TRUE AS col, CAST(x AS INT) AS y FROM t
+            """
+        ),
+        dialect="tsql",
+    )
+    # The MODEL header is SQLMesh DDL and must not be transpiled: a boolean property
+    # such as `allow_partials true` must stay `TRUE`, not become tsql's `(1 = 1)`.
+    # The query body must still transpile to the target dialect.
+    assert (
+        x
+        == """MODEL (
+  name a.b,
+  kind FULL,
+  dialect tsql,
+  allow_partials TRUE
+);
+
+SELECT
+  1 AS col,
+  x::INTEGER AS y
+FROM t"""
+    )
+
+    x = format_model_expressions(
+        parse(
+            """
+            AUDIT(name my_audit, dialect tsql, blocking false);
+            SELECT TRUE AS col, CAST(x AS INT) AS y FROM t WHERE x > 0
+            """
+        ),
+        dialect="tsql",
+    )
+    # AUDIT headers are SQLMesh DDL too: a `false` boolean property must stay
+    # `FALSE`, not become tsql's `(1 = 0)`, while the query body still transpiles.
+    assert (
+        x
+        == """AUDIT (
+  name my_audit,
+  dialect tsql,
+  blocking FALSE
+);
+
+SELECT
+  1 AS col,
+  x::INTEGER AS y
+FROM t
+WHERE
+  x > 0"""
+    )
+
     x = format_model_expressions(
         parse(
             """
diff --git a/tests/integrations/test_dlt.py b/tests/integrations/test_dlt.py
@@ -0,0 +1,21 @@
+from sqlmesh.integrations.dlt import generate_incremental_model
+
+
+def test_generate_incremental_model_filters_on_timestamp_macros() -> None:
+    # The DLT-generated model's time column is a timestamp
+    # (TO_TIMESTAMP(...)). It must therefore be filtered with the inclusive
+    # timestamp macros @start_ts/@end_ts, not the categorical date macros
+    # @start_ds/@end_ds, which both render midnight and exclude any rows past
+    # 00:00:00 on a single-day run.
+    model = generate_incremental_model(
+        "dataset_sqlmesh.incremental_equipment",
+        "  CAST(c.item_id AS BIGINT) AS item_id",
+        "",
+        "dataset.equipment",
+        "duckdb",
+        "c._dlt_load_id",
+    )
+
+    assert "BETWEEN @start_ts AND @end_ts" in model
+    assert "@start_ds" not in model
+    assert "@end_ds" not in model
diff --git a/tests/utils/test_jinja.py b/tests/utils/test_jinja.py
@@ -1,6 +1,8 @@
 from __future__ import annotations
 
-from sqlmesh.utils import AttributeDict
+from base64 import b64encode
+
+from sqlmesh.utils import AttributeDict, yaml
 from sqlmesh.utils.jinja import (
     ENVIRONMENT,
     JinjaMacroRegistry,
@@ -329,3 +331,25 @@ def test_macro_registry_to_expressions_sorted():
         == "refs = {'orders': {'database': 'jaffle_shop', 'nested_list': ['a', 'b', 'c'], 'schema': 'main'}, 'payments': {'database': 'jaffle_shop', 'nested': {'baz': 'bing', 'foo': 'bar'}, 'schema': 'main'}}\n"
         "sources = {}"
     )
+
+
+def test_builtin_base64_filters():
+    encoded = b64encode(b"secret").decode("utf-8")
+
+    env = JinjaMacroRegistry().build_environment()
+    assert env.from_string("{{ value | b64decode }}").render(value=encoded) == "secret"
+    assert env.from_string("{{ 'secret' | b64encode }}").render() == encoded
+    assert env.from_string("{{ 'secret' | b64encode | b64decode }}").render() == "secret"
+
+    # The same filters are available when rendering Jinja in config YAML files.
+    config = yaml.load(f'env_vars:\n  TOKEN: "{{{{ "{encoded}" | b64decode }}}}"')
+    assert config == {"env_vars": {"TOKEN": "secret"}}
+
+
+def test_builtin_b64decode_with_env_var(monkeypatch):
+    # Real-world use case: a base64-encoded secret stored in an environment variable
+    # is decoded inline in config YAML via env_var(...) piped through b64decode.
+    monkeypatch.setenv("SNOWFLAKE_PW_B64", b64encode(b"super-secret-pw").decode("utf-8"))
+
+    config = yaml.load("password: \"{{ env_var('SNOWFLAKE_PW_B64') | b64decode }}\"")
+    assert config == {"password": "super-secret-pw"}
