Merge branch 'main' into fix_databricks_column_types

MisterWheatley · web-flow · commit 55b4acbc4f0e · 2026-05-26T14:58:50.000+02:00
diff --git a/.github/workflows/release.yaml b/.github/workflows/release.yaml
@@ -12,12 +12,12 @@ jobs:
       - uses: actions/checkout@v5
       - uses: actions/setup-node@v6
         with:
-          node-version: '20'
+          node-version: '22'
       - uses: pnpm/action-setup@v4
         with:
           version: latest
       - name: Install dependencies
-        run: pnpm install
+        run: pnpm install --frozen-lockfile
       - name: Build UI
         run: pnpm --prefix web/client run build
       - name: Upload UI build artifact
diff --git a/.github/workflows/release_extension.yaml b/.github/workflows/release_extension.yaml
@@ -30,7 +30,7 @@ jobs:
       - name: Setup Node.js
         uses: actions/setup-node@v6
         with:
-          node-version: '20'
+          node-version: '22'
       - name: Install pnpm
         uses: pnpm/action-setup@v4
         with:
diff --git a/.github/workflows/release_shared_js.yaml b/.github/workflows/release_shared_js.yaml
@@ -33,7 +33,7 @@ jobs:
       - name: Setup Node.js
         uses: actions/setup-node@v6
         with:
-          node-version: '20'
+          node-version: '22'
           registry-url: 'https://registry.npmjs.org'
       - name: Update npm
         run: npm install -g npm@latest
diff --git a/.nvmrc b/.nvmrc
@@ -1 +1 @@
-20
+22
diff --git a/docs/integrations/engines/duckdb.md b/docs/integrations/engines/duckdb.md
@@ -79,6 +79,7 @@ SQLMesh will place models with the explicit catalog "ephemeral", such as `epheme
               type: ducklake
               path: 'catalog.ducklake'
               data_path: data/ducklake
+              override_data_path: true
               encrypted: True
               data_inlining_row_limit: 10
               metadata_schema: main
@@ -105,6 +106,7 @@ SQLMesh will place models with the explicit catalog "ephemeral", such as `epheme
                             type="ducklake",
                             path="catalog.ducklake",
                             data_path="data/ducklake",
+                            override_data_path=False,
                             encrypted=True,
                             data_inlining_row_limit=10,
                             metadata_schema="main",
@@ -120,6 +122,7 @@ SQLMesh will place models with the explicit catalog "ephemeral", such as `epheme
 
 - `path`: Path to the DuckLake catalog file
 - `data_path`: Path where DuckLake data files are stored
+- `override_data_path`: Whether data_override_path option is set
 - `encrypted`: Whether to enable encryption for the catalog (default: `False`)
 - `data_inlining_row_limit`: Maximum number of rows to inline in the catalog (default: `0`)
 - `metadata_schema`: The schema in the catalog server in which to store the DuckLake metadata tables (default: `main`)
@@ -364,6 +367,7 @@ The `filesystems` accepts a list of file systems to register in the DuckDB conne
               type: ducklake
               path: myducklakecatalog.duckdb
               data_path: abfs://MyFabricWorkspace/MyFabricLakehouse.Lakehouse/Files/DuckLake.Files
+              override_data_path: False
         extensions:
           - ducklake
         filesystems:
diff --git a/package.json b/package.json
@@ -1,6 +1,6 @@
 {
   "engines": {
-    "node": ">=20.0.0",
+    "node": ">=22.0.0",
     "pnpm": ">=10.0.0"
   },
   "scripts": {
diff --git a/pnpm-workspace.yaml b/pnpm-workspace.yaml
@@ -4,3 +4,9 @@ packages:
   - vscode/react
   - web/client
   - web/common
+allowBuilds:
+  '@swc/core': true
+  '@tailwindcss/oxide': true
+  '@vscode/vsce-sign': true
+  esbuild: true
+  keytar: true
diff --git a/sqlmesh/cli/main.py b/sqlmesh/cli/main.py
@@ -631,16 +631,22 @@ def invalidate(ctx: click.Context, environment: str, **kwargs: t.Any) -> None:
     is_flag=True,
     help="Cleanup snapshots that are not referenced in any environment, regardless of when they're set to expire",
 )
+@click.option(
+    "--force-delete",
+    is_flag=True,
+    help="Delete expired environment and snapshot state records even when the physical table or view drops fail. "
+    "Any objects that could not be dropped become orphaned and must be removed manually.",
+)
 @click.pass_context
 @error_handler
 @cli_analytics
-def janitor(ctx: click.Context, ignore_ttl: bool, **kwargs: t.Any) -> None:
+def janitor(ctx: click.Context, ignore_ttl: bool, force_delete: bool, **kwargs: t.Any) -> None:
     """
     Run the janitor process on-demand.
 
     The janitor cleans up old environments and expired snapshots.
     """
-    ctx.obj.run_janitor(ignore_ttl, **kwargs)
+    ctx.obj.run_janitor(ignore_ttl, force_delete=force_delete, **kwargs)
 
 
 @cli.command("destroy")
diff --git a/sqlmesh/core/config/connection.py b/sqlmesh/core/config/connection.py
@@ -238,6 +238,7 @@ class DuckDBAttachOptions(BaseConfig):
 
     # DuckLake specific options
     data_path: t.Optional[str] = None
+    override_data_path: t.Optional[bool] = False
     encrypted: bool = False
     data_inlining_row_limit: t.Optional[int] = None
     metadata_schema: t.Optional[str] = None
@@ -258,6 +259,8 @@ def to_sql(self, alias: str) -> str:
                 path = f"ducklake:{path}"
             if self.data_path is not None:
                 options.append(f"DATA_PATH '{self.data_path}'")
+                if self.override_data_path:
+                    options.append("OVERRIDE_DATA_PATH true")
             if self.encrypted:
                 options.append("ENCRYPTED")
             if self.data_inlining_row_limit is not None:
@@ -2097,6 +2100,7 @@ class ClickhouseConnectionConfig(ConnectionConfig):
     https_proxy: t.Optional[str] = None
     server_host_name: t.Optional[str] = None
     tls_mode: t.Optional[str] = None
+    secure: bool = False
 
     concurrent_tasks: int = 1
     register_comments: bool = True
@@ -2133,6 +2137,7 @@ def _connection_kwargs_keys(self) -> t.Set[str]:
             "https_proxy",
             "server_host_name",
             "tls_mode",
+            "secure",
         }
         return kwargs
 
diff --git a/sqlmesh/core/context.py b/sqlmesh/core/context.py
@@ -887,12 +887,12 @@ def _has_environment_changed() -> bool:
         return completion_status
 
     @python_api_analytics
-    def run_janitor(self, ignore_ttl: bool) -> bool:
+    def run_janitor(self, ignore_ttl: bool, force_delete: bool = False) -> bool:
         success = False
 
         if self.console.start_cleanup(ignore_ttl):
             try:
-                self._run_janitor(ignore_ttl)
+                self._run_janitor(ignore_ttl, force_delete=force_delete)
                 success = True
             finally:
                 self.console.stop_cleanup(success=success)
@@ -2896,24 +2896,43 @@ def _destroy(self) -> bool:
 
         return True
 
-    def _run_janitor(self, ignore_ttl: bool = False) -> None:
+    def _run_janitor(self, ignore_ttl: bool = False, force_delete: bool = False) -> None:
         current_ts = now_timestamp()
+        failures: t.List[str] = []
 
         # Clean up expired environments by removing their views and schemas
-        self._cleanup_environments(current_ts=current_ts)
+        failures.extend(
+            self._cleanup_environments(current_ts=current_ts, force_delete=force_delete)
+        )
 
-        delete_expired_snapshots(
-            self.state_sync,
-            self.snapshot_evaluator,
-            current_ts=current_ts,
-            ignore_ttl=ignore_ttl,
-            console=self.console,
-            batch_size=self.config.janitor.expired_snapshots_batch_size,
+        failures.extend(
+            delete_expired_snapshots(
+                self.state_sync,
+                self.snapshot_evaluator,
+                current_ts=current_ts,
+                ignore_ttl=ignore_ttl,
+                force_delete=force_delete,
+                console=self.console,
+                batch_size=self.config.janitor.expired_snapshots_batch_size,
+            )
         )
         self.state_sync.compact_intervals()
 
-    def _cleanup_environments(self, current_ts: t.Optional[int] = None) -> None:
+        if failures:
+            failure_string = "\n  - ".join(failures)
+            summary = f"Janitor completed with failures:\n  {failure_string}"
+            if force_delete:
+                summary += "\nState records have been deleted, but the underlying objects may still exist in the database.\nPlease investigate and clean up manually the above if necessary."
+            if self.config.janitor.warn_on_delete_failure:
+                self.console.log_warning(summary)
+            else:
+                raise SQLMeshError(summary)
+
+    def _cleanup_environments(
+        self, current_ts: t.Optional[int] = None, force_delete: bool = False
+    ) -> t.List[str]:
         current_ts = current_ts or now_timestamp()
+        failures: t.List[str] = []
 
         expired_environments_summaries = self.state_sync.get_expired_environments(
             current_ts=current_ts
@@ -2923,15 +2942,20 @@ def _cleanup_environments(self, current_ts: t.Optional[int] = None) -> None:
             expired_env = self.state_reader.get_environment(expired_env_summary.name)
 
             if expired_env:
-                cleanup_expired_views(
-                    default_adapter=self.engine_adapter,
-                    engine_adapters=self.engine_adapters,
-                    environments=[expired_env],
-                    warn_on_delete_failure=self.config.janitor.warn_on_delete_failure,
-                    console=self.console,
+                failures.extend(
+                    cleanup_expired_views(
+                        default_adapter=self.engine_adapter,
+                        engine_adapters=self.engine_adapters,
+                        environments=[expired_env],
+                        console=self.console,
+                    )
                 )
 
-        self.state_sync.delete_expired_environments(current_ts=current_ts)
+        # we want to retry on the next janitor pass if drops failed, unless
+        # force_delete is set in which case we purge state records regardless
+        if not failures or force_delete:
+            self.state_sync.delete_expired_environments(current_ts=current_ts)
+        return failures
 
     def _try_connection(self, connection_name: str, validator: t.Callable[[], None]) -> None:
         connection_name = connection_name.capitalize()
diff --git a/sqlmesh/core/dialect.py b/sqlmesh/core/dialect.py
@@ -774,7 +774,8 @@ def format_model_expressions(
     if rewrite_casts:
 
         def cast_to_colon(node: exp.Expr) -> exp.Expr:
-            if isinstance(node, exp.Cast) and not any(
+            # Directly check type instead of isinstance to avoid rewriting subclasses of CAST, e.g. JSONCast
+            if type(node) is exp.Cast and not any(
                 # Only convert CAST into :: if it doesn't have additional args set, otherwise this
                 # conversion could alter the semantics (eg. changing SAFE_CAST in BigQuery to CAST)
                 arg
diff --git a/sqlmesh/core/janitor.py b/sqlmesh/core/janitor.py
@@ -16,16 +16,16 @@
     RowBoundary,
     ExpiredBatchRange,
 )
-from sqlmesh.utils.errors import SQLMeshError
 
 
 def cleanup_expired_views(
     default_adapter: EngineAdapter,
     engine_adapters: t.Dict[str, EngineAdapter],
     environments: t.List[Environment],
-    warn_on_delete_failure: bool = False,
     console: t.Optional[Console] = None,
-) -> None:
+) -> t.List[str]:
+    failures: t.List[str] = []
+
     expired_schema_or_catalog_environments = [
         environment
         for environment in environments
@@ -85,10 +85,8 @@ def get_adapter(gateway_managed: bool, gateway: t.Optional[str] = None) -> Engin
                 console.update_cleanup_progress(expired_view)
         except Exception as e:
             message = f"Failed to drop the expired environment view '{expired_view}': {e}"
-            if warn_on_delete_failure:
-                logger.warning(message)
-            else:
-                raise SQLMeshError(message) from e
+            logger.warning(message)
+            failures.append(message)
 
     # Drop the schemas for the expired environments
     for engine_adapter, schema in schemas_to_drop:
@@ -102,10 +100,8 @@ def get_adapter(gateway_managed: bool, gateway: t.Optional[str] = None) -> Engin
                 console.update_cleanup_progress(schema.sql(dialect=engine_adapter.dialect))
         except Exception as e:
             message = f"Failed to drop the expired environment schema '{schema}': {e}"
-            if warn_on_delete_failure:
-                logger.warning(message)
-            else:
-                raise SQLMeshError(message) from e
+            logger.warning(message)
+            failures.append(message)
 
     # Drop any catalogs that were associated with a snapshot where the engine adapter supports dropping catalogs
     # catalogs_to_drop is only populated when environment_suffix_target is set to 'catalog'
@@ -117,10 +113,10 @@ def get_adapter(gateway_managed: bool, gateway: t.Optional[str] = None) -> Engin
                     console.update_cleanup_progress(catalog)
             except Exception as e:
                 message = f"Failed to drop the expired environment catalog '{catalog}': {e}"
-                if warn_on_delete_failure:
-                    logger.warning(message)
-                else:
-                    raise SQLMeshError(message) from e
+                logger.warning(message)
+                failures.append(message)
+
+    return failures
 
 
 def delete_expired_snapshots(
@@ -129,9 +125,10 @@ def delete_expired_snapshots(
     *,
     current_ts: int,
     ignore_ttl: bool = False,
+    force_delete: bool = False,
     batch_size: t.Optional[int] = None,
     console: t.Optional[Console] = None,
-) -> None:
+) -> t.List[str]:
     """Delete all expired snapshots in batches.
 
     This helper function encapsulates the logic for deleting expired snapshots in batches,
@@ -142,12 +139,14 @@ def delete_expired_snapshots(
         snapshot_evaluator: SnapshotEvaluator instance to clean up tables associated with snapshots.
         current_ts: Timestamp used to evaluate expiration.
         ignore_ttl: If True, include snapshots regardless of TTL (only checks if unreferenced).
+        force_delete: If True, delete snapshot state records even when physical table cleanup fails.
         batch_size: Maximum number of snapshots to fetch per batch.
         console: Optional console for reporting progress.
 
     Returns:
-        The total number of deleted expired snapshots.
+        List of failure messages so callers can surface them at the end of the janitor run.
     """
+    failures: t.List[str] = []
     num_expired_snapshots = 0
     for batch in iter_expired_snapshot_batches(
         state_reader=state_sync,
@@ -165,17 +164,32 @@ def delete_expired_snapshots(
             len(batch.expired_snapshot_ids),
             end_info,
         )
-        snapshot_evaluator.cleanup(
-            target_snapshots=batch.cleanup_tasks,
-            on_complete=console.update_cleanup_progress if console else None,
-        )
-        state_sync.delete_expired_snapshots(
-            batch_range=ExpiredBatchRange(
-                start=RowBoundary.lowest_boundary(),
-                end=batch.batch_range.end,
-            ),
-            ignore_ttl=ignore_ttl,
-        )
-        logger.info("Cleaned up expired snapshots batch")
-        num_expired_snapshots += len(batch.expired_snapshot_ids)
+        cleanup_succeeded = True
+        try:
+            snapshot_evaluator.cleanup(
+                target_snapshots=batch.cleanup_tasks,
+                on_complete=console.update_cleanup_progress if console else None,
+            )
+        except Exception as failed_drops:
+            message = f"Failed to clean up: {failed_drops}"
+            logger.warning(message)
+            failures.append(message)
+            cleanup_succeeded = False
+
+        if cleanup_succeeded or force_delete:
+            try:
+                state_sync.delete_expired_snapshots(
+                    batch_range=ExpiredBatchRange(
+                        start=RowBoundary.lowest_boundary(),
+                        end=batch.batch_range.end,
+                    ),
+                    ignore_ttl=ignore_ttl,
+                )
+                logger.info("Cleaned up expired snapshots batch")
+                num_expired_snapshots += len(batch.expired_snapshot_ids)
+            except Exception as e:
+                message = f"Failed to delete expired snapshot state records: {e}"
+                logger.warning(message)
+                failures.append(message)
     logger.info("Cleaned up %s expired snapshots", num_expired_snapshots)
+    return failures
diff --git a/sqlmesh/core/snapshot/evaluator.py b/sqlmesh/core/snapshot/evaluator.py
diff --git a/tests/core/integration/test_aux_commands.py b/tests/core/integration/test_aux_commands.py
diff --git a/tests/core/test_connection_config.py b/tests/core/test_connection_config.py
diff --git a/tests/core/test_dialect.py b/tests/core/test_dialect.py
diff --git a/tests/core/test_janitor.py b/tests/core/test_janitor.py
diff --git a/tests/core/test_snapshot_evaluator.py b/tests/core/test_snapshot_evaluator.py

Original file line number	Diff line number	Diff line change
`@@ -1,6 +1,6 @@`
`1`	`1`	`{`
`2`	`2`	`"engines": {`
`3`		`- "node": ">=20.0.0",`
	`3`	`+ "node": ">=22.0.0",`
`4`	`4`	`"pnpm": ">=10.0.0"`
`5`	`5`	`},`
`6`	`6`	`"scripts": {`