Merge branch 'main' into fix/databricks-seed-null-booleans

Author: nkwork9999

Makefile

@@ -63,7 +63,8 @@ install-dev-dbt-%:
 	fi; \
 	if [ "$$version" = "1.3.0" ]; then \
 		echo "Applying overrides for dbt $$version - upgrading google-cloud-bigquery"; \
-		$(PIP) install 'google-cloud-bigquery>=3.0.0' --upgrade; \
+		$(PIP) install 'google-cloud-bigquery>=3.0.0' \
+			'pyOpenSSL>=24.0.0' --upgrade; \
 	fi; \
 	mv pyproject.toml.backup pyproject.toml; \
 	echo "Restored original pyproject.toml"

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`)
+)
+```

docs/prerequisites.md

@@ -4,7 +4,7 @@ This page describes the system prerequisites needed to run SQLMesh and provides
 
 ## SQLMesh prerequisites
 
-You'll need Python 3.8 or higher to use SQLMesh. You can check your python version by running the following command:
+You'll need Python 3.9 or higher to use SQLMesh. You can check your python version by running the following command:
 ```bash
 python3 --version
 ```

pyproject.toml

@@ -7,7 +7,6 @@ authors = [{ name = "SQLMesh Contributors" }]
 license = { file = "LICENSE" }
 requires-python = ">= 3.9"
 dependencies = [
-    "astor",
     "click",
     "croniter",
     "duckdb>=0.10.0,!=0.10.3",
@@ -202,7 +201,6 @@ disable_error_code = "annotation-unchecked"
 [[tool.mypy.overrides]]
 module = [
     "api.*",
-    "astor.*",
     "IPython.*",
     "hyperscript.*",
     "py.*",

sqlmesh/core/config/connection.py

@@ -13,8 +13,8 @@
 from sys import version_info
 
 import pydantic
+from pydantic import Field, computed_field
 from packaging import version
-from pydantic import Field
 from pydantic_core import from_json
 from sqlglot import exp
 from sqlglot.errors import ParseError
@@ -110,7 +110,14 @@ class ConnectionConfig(abc.ABC, BaseConfig):
     catalog_type_overrides: t.Optional[t.Dict[str, str]] = None
 
     # Whether to share a  single connection across threads or create a new connection per thread.
-    shared_connection: t.ClassVar[bool] = False
+    #
+    # MyPy throws a "Decorators on top of @property are not supported" error despite this being a
+    # valid decoration, and Pydantic recommend disabling the MyPy hint for this reason - see:
+    # https://pydantic.dev/docs/validation/2.0/usage/computed_fields/
+    @computed_field  # type: ignore[prop-decorator]
+    @property
+    def shared_connection(self) -> bool:
+        return False
 
     @property
     @abc.abstractmethod
@@ -311,7 +318,10 @@ class BaseDuckDBConnectionConfig(ConnectionConfig):
 
     token: t.Optional[str] = None
 
-    shared_connection: t.ClassVar[bool] = True
+    @computed_field  # type: ignore[prop-decorator]
+    @property
+    def shared_connection(self) -> bool:
+        return True
 
     _data_file_to_adapter: t.ClassVar[t.Dict[str, EngineAdapter]] = {}
 
@@ -820,11 +830,15 @@ class DatabricksConnectionConfig(ConnectionConfig):
     DISPLAY_NAME: t.ClassVar[t.Literal["Databricks"]] = "Databricks"
     DISPLAY_ORDER: t.ClassVar[t.Literal[3]] = 3
 
-    shared_connection: t.ClassVar[bool] = True
-
     _concurrent_tasks_validator = concurrent_tasks_validator
     _http_headers_validator = http_headers_validator
 
+    @computed_field  # type: ignore[prop-decorator]
+    @property
+    def shared_connection(self) -> bool:
+        """The connection should only be shared if U2M OAuth is being used"""
+        return self.auth_type is not None and self.oauth_client_secret is None
+
     @model_validator(mode="before")
     def _databricks_connect_validator(cls, data: t.Any) -> t.Any:
         # SQLQueryContextLogger will output any error SQL queries even if they are in a try/except block.

sqlmesh/core/config/loader.py

@@ -272,4 +272,4 @@ def convert_config_type(
     config_obj: Config,
     config_type: t.Type[C],
 ) -> C:
-    return config_type.parse_obj(config_obj.dict())
+    return config_type.parse_obj(config_obj.dict(exclude_computed_fields=True))

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))
 

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 (
@@ -153,7 +154,7 @@ def _set_spark_engine_adapter_if_needed(self) -> None:
             host=self._extra_config["databricks_connect_server_hostname"],
             token=self._extra_config.get("databricks_connect_access_token"),
         )
-        if "databricks_connect_use_serverless" in self._extra_config:
+        if self._extra_config.get("databricks_connect_use_serverless"):
             connect_kwargs["serverless"] = True
         else:
             connect_kwargs["cluster_id"] = self._extra_config["databricks_connect_cluster_id"]
@@ -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)