Merge remote-tracking branch 'upstream/main' into feat/add-fabric-engine

Author: mattiasthalen

.circleci/continue_config.yml

@@ -201,39 +201,6 @@ jobs:
           name: Run tests
           command: npm --prefix web/client run test
 
-  trigger_private_tests:
-    docker:
-      - image: cimg/python:3.12.0
-    resource_class: small
-    steps:
-      - checkout
-      - run:
-          name: Install setuptools scm
-          command: pip install setuptools_scm
-      - run:
-          name: Trigger private tests
-          command: |
-            export COMMIT_MESSAGE="$(git log --format=%s -n 1 $CIRCLE_SHA1)"
-            export FORMATTED_COMMIT_MESSAGE="${COMMIT_MESSAGE//\"/\\\"}"
-            # returns a version string like 0.1.0.dev11
-            export PACKAGE_VERSION="$(python ./.circleci/get_scm_version.py)"
-            curl --request POST \
-              --url $TOBIKO_PRIVATE_CIRCLECI_URL \
-              --header "Circle-Token: $TOBIKO_PRIVATE_CIRCLECI_KEY" \
-              --header "content-type: application/json" \
-              --data '{
-                "branch":"main",
-                "parameters":{
-                  "run_main_pr":false,
-                  "run_sqlmesh_commit":true,
-                  "sqlmesh_branch":"'$CIRCLE_BRANCH'",
-                  "sqlmesh_commit_author":"'$CIRCLE_USERNAME'",
-                  "sqlmesh_commit_hash":"'$CIRCLE_SHA1'",
-                  "sqlmesh_commit_message":"'"$FORMATTED_COMMIT_MESSAGE"'",
-                  "sqlmesh_package_version":"'$PACKAGE_VERSION'"
-                  }
-              }'
-
   engine_tests_docker:
     parameters:
       engine:
@@ -340,13 +307,6 @@ workflows:
             branches:
               only:
                 - main
-      - trigger_private_tests:
-          requires:
-            - style_and_cicd_tests
-          filters:
-            branches:
-              only:
-                - main
       - ui_style
       - ui_test
       - vscode_test

docs/integrations/github.md

@@ -300,6 +300,7 @@ Below is an example of how to define the default config for the bot in either YA
 | `run_on_deploy_to_prod`               | Indicates whether to run latest intervals when deploying to prod. If set to false, the deployment will backfill only the changed models up to the existing latest interval in production, ignoring any missing intervals beyond this point. Default: `False`                                                                                                                                             |  bool  |    N     |
 | `pr_environment_name`                 | The name of the PR environment to create for which a PR number will be appended to. Defaults to the repo name if not provided. Note: The name will be normalized to alphanumeric + underscore and lowercase.                                                                                                                                                                                             |  str   |    N     |
 | `prod_branch_name`                    | The name of the git branch associated with production. Ex: `prod`. Default: `main` or `master` is considered prod                                                                                                                                                                                                                                                                                        |  str   |    N     |
+| `forward_only_branch_suffix`          | If the git branch has this suffix, trigger a [forward-only](../concepts/plans.md#forward-only-plans) plan instead of a normal plan. Default: `-forward-only`                                                                                                                                                                                                                                             |  str   |    N     |
 
 Example with all properties defined:
 

docs/reference/cli.md

@@ -149,7 +149,7 @@ Options:
 ```
 Usage: sqlmesh destroy
 
-  Removes all project resources, including warehouse objects, state tables, the SQLMesh cache and any build artifacts.
+  Removes all state tables, the SQLMesh cache and all project resources, including warehouse objects. This includes all tables, views and schemas managed by SQLMesh, as well as any external resources that may have been created by other tools within those schemas.
 
 Options:
   --help               Show this message and exit.

docs/reference/notebook.md

@@ -250,7 +250,7 @@ options:
 ```
 %destroy
 
-Removes all project resources, including warehouse objects, state tables, the SQLMesh cache and any build artifacts.
+Removes all state tables, the SQLMesh cache, and other project resources, including warehouse objects. This includes all tables, views, and schemas managed by SQLMesh, as well as any external resources that may have been created by other tools within those schemas.
 ```
 
 #### dlt_refresh

sqlmesh/cli/options.py

@@ -63,6 +63,7 @@ def format_options(func: t.Callable) -> t.Callable:
         "--normalize",
         is_flag=True,
         help="Whether or not to normalize identifiers to lowercase.",
+        default=None,
     )(func)
     func = click.option(
         "--pad",
@@ -82,6 +83,7 @@ def format_options(func: t.Callable) -> t.Callable:
     func = click.option(
         "--leading-comma",
         is_flag=True,
+        default=None,
         help="Determines whether or not the comma is leading or trailing in select expressions. Default is trailing.",
     )(func)
     func = click.option(

sqlmesh/core/config/connection.py

@@ -354,11 +354,28 @@ def init(cursor: duckdb.DuckDBPyConnection) -> None:
                 except Exception as e:
                     raise ConfigError(f"Failed to load extension {extension['name']}: {e}")
 
-            for field, setting in self.connector_config.items():
-                try:
-                    cursor.execute(f"SET {field} = '{setting}'")
-                except Exception as e:
-                    raise ConfigError(f"Failed to set connector config {field} to {setting}: {e}")
+            if self.connector_config:
+                option_names = list(self.connector_config)
+                in_part = ",".join("?" for _ in range(len(option_names)))
+
+                cursor.execute(
+                    f"SELECT name, value FROM duckdb_settings() WHERE name IN ({in_part})",
+                    option_names,
+                )
+
+                existing_values = {field: setting for field, setting in cursor.fetchall()}
+
+                # only set connector_config items if the values differ from what is already set
+                # trying to set options like 'temp_directory' even to the same value can throw errors like:
+                # Not implemented Error: Cannot switch temporary directory after the current one has been used
+                for field, setting in self.connector_config.items():
+                    if existing_values.get(field) != setting:
+                        try:
+                            cursor.execute(f"SET {field} = '{setting}'")
+                        except Exception as e:
+                            raise ConfigError(
+                                f"Failed to set connector config {field} to {setting}: {e}"
+                            )
 
             if self.secrets:
                 duckdb_version = duckdb.__version__
@@ -386,7 +403,9 @@ def init(cursor: duckdb.DuckDBPyConnection) -> None:
                         if secret_settings:
                             secret_clause = ", ".join(secret_settings)
                             try:
-                                cursor.execute(f"CREATE SECRET {secret_name} ({secret_clause});")
+                                cursor.execute(
+                                    f"CREATE OR REPLACE SECRET {secret_name} ({secret_clause});"
+                                )
                             except Exception as e:
                                 raise ConfigError(f"Failed to create secret: {e}")
 

sqlmesh/core/config/loader.py

@@ -184,8 +184,14 @@ def load_config_from_python_module(
     module_path: Path,
     config_name: str = "config",
 ) -> C:
-    with sys_path(module_path.parent):
-        config_module = import_python_file(module_path, module_path.parent)
+    try:
+        with sys_path(module_path.parent):
+            config_module = import_python_file(module_path, module_path.parent)
+    except Exception as e:
+        raise ConfigError(
+            f"Failed to load config file: {e}",
+            location=module_path,
+        )
 
     try:
         config_obj = getattr(config_module, config_name)

sqlmesh/core/console.py

@@ -186,9 +186,19 @@ class DestroyConsole(abc.ABC):
     """Console for describing a destroy operation"""
 
     @abc.abstractmethod
-    def start_destroy(self) -> bool:
+    def start_destroy(
+        self,
+        schemas_to_delete: t.Optional[t.Set[str]] = None,
+        views_to_delete: t.Optional[t.Set[str]] = None,
+        tables_to_delete: t.Optional[t.Set[str]] = None,
+    ) -> bool:
         """Start a destroy operation.
 
+        Args:
+            schemas_to_delete: Set of schemas that will be deleted
+            views_to_delete: Set of views that will be deleted
+            tables_to_delete: Set of tables that will be deleted
+
         Returns:
             Whether or not the destroy operation should proceed
         """
@@ -289,11 +299,17 @@ def show_row_diff(
 
 class BaseConsole(abc.ABC):
     @abc.abstractmethod
-    def log_error(self, message: str) -> None:
+    def log_error(self, message: str, *args: t.Any, **kwargs: t.Any) -> None:
         """Display error info to the user."""
 
     @abc.abstractmethod
-    def log_warning(self, short_message: str, long_message: t.Optional[str] = None) -> None:
+    def log_warning(
+        self,
+        short_message: str,
+        long_message: t.Optional[str] = None,
+        *args: t.Any,
+        **kwargs: t.Any,
+    ) -> None:
         """Display warning info to the user.
 
         Args:
@@ -824,7 +840,12 @@ def print_connection_config(
     ) -> None:
         pass
 
-    def start_destroy(self) -> bool:
+    def start_destroy(
+        self,
+        schemas_to_delete: t.Optional[t.Set[str]] = None,
+        views_to_delete: t.Optional[t.Set[str]] = None,
+        tables_to_delete: t.Optional[t.Set[str]] = None,
+    ) -> bool:
         return True
 
     def stop_destroy(self, success: bool = True) -> None:
@@ -1276,16 +1297,40 @@ def stop_cleanup(self, success: bool = False) -> None:
         else:
             self.log_error("Cleanup failed!")
 
-    def start_destroy(self) -> bool:
+    def start_destroy(
+        self,
+        schemas_to_delete: t.Optional[t.Set[str]] = None,
+        views_to_delete: t.Optional[t.Set[str]] = None,
+        tables_to_delete: t.Optional[t.Set[str]] = None,
+    ) -> bool:
         self.log_warning(
-            (
-                "This will permanently delete all engine-managed objects, state tables and SQLMesh cache.\n"
-                "The operation is irreversible and may disrupt any currently running or scheduled plans.\n"
-                "Use this command only when you intend to fully reset the project."
-            )
+            "This will permanently delete all engine-managed objects, state tables and SQLMesh cache.\n"
+            "The operation may disrupt any currently running or scheduled plans.\n"
         )
-        if not self._confirm("Proceed?"):
-            self.log_error("Destroy aborted!")
+
+        if schemas_to_delete or views_to_delete or tables_to_delete:
+            if schemas_to_delete:
+                self.log_error("Schemas to be deleted:")
+                for schema in sorted(schemas_to_delete):
+                    self.log_error(f"  • {schema}")
+
+            if views_to_delete:
+                self.log_error("\nEnvironment views to be deleted:")
+                for view in sorted(views_to_delete):
+                    self.log_error(f"  • {view}")
+
+            if tables_to_delete:
+                self.log_error("\nSnapshot tables to be deleted:")
+                for table in sorted(tables_to_delete):
+                    self.log_error(f"  • {table}")
+
+            self.log_error(
+                "\nThis action will DELETE ALL the above resources managed by SQLMesh AND\n"
+                "potentially external resources created by other tools in these schemas.\n"
+            )
+
+        if not self._confirm("Are you ABSOLUTELY SURE you want to proceed with deletion?"):
+            self.log_error("Destroy operation cancelled.")
             return False
         return True
 
@@ -2711,8 +2756,16 @@ def _cells_match(x: t.Any, y: t.Any) -> bool:
     def _normalize(val: t.Any) -> t.Any:
         # Convert Pandas null to Python null for the purposes of comparison to prevent errors like the following on boolean fields:
         # - TypeError: boolean value of NA is ambiguous
-        if pd.isnull(val):
+        # note pd.isnull() returns either a bool or a ndarray[bool] depending on if the input
+        # is scalar or an array
+        isnull = pd.isnull(val)
+
+        if isinstance(isnull, bool):  # scalar
+            if isnull:
+                val = None
+        elif all(isnull):  # array
             val = None
+
         return list(val) if isinstance(val, (pd.Series, np.ndarray)) else val
 
     return _normalize(x) == _normalize(y)
@@ -3082,15 +3135,23 @@ def consume_captured_errors(self) -> str:
         finally:
             self._errors = []
 
-    def log_warning(self, short_message: str, long_message: t.Optional[str] = None) -> None:
+    def log_warning(
+        self,
+        short_message: str,
+        long_message: t.Optional[str] = None,
+        *args: t.Any,
+        **kwargs: t.Any,
+    ) -> None:
         if short_message not in self._warnings:
             self._warnings.append(short_message)
-        super().log_warning(short_message, long_message)
+        if kwargs.pop("print", True):
+            super().log_warning(short_message, long_message)
 
-    def log_error(self, message: str) -> None:
+    def log_error(self, message: str, *args: t.Any, **kwargs: t.Any) -> None:
         if message not in self._errors:
             self._errors.append(message)
-        super().log_error(message)
+        if kwargs.pop("print", True):
+            super().log_error(message)
 
     def log_skipped_models(self, snapshot_names: t.Set[str]) -> None:
         if snapshot_names:
@@ -3127,6 +3188,11 @@ def __init__(self, **kwargs: t.Any) -> None:
             kwargs.pop("alert_block_collapsible_threshold", 200)
         )
 
+        # capture_only = True: capture but dont print to console
+        # capture_only = False: capture and also print to console
+        self.warning_capture_only = kwargs.pop("warning_capture_only", False)
+        self.error_capture_only = kwargs.pop("error_capture_only", False)
+
         super().__init__(
             **{**kwargs, "console": RichConsole(no_color=True, width=kwargs.pop("width", None))}
         )
@@ -3401,6 +3467,12 @@ def stop_promotion_progress(self, success: bool = True) -> None:
         super().stop_promotion_progress(success)
         self._print("\n")
 
+    def log_warning(self, short_message: str, long_message: t.Optional[str] = None) -> None:
+        super().log_warning(short_message, long_message, print=not self.warning_capture_only)
+
+    def log_error(self, message: str) -> None:
+        super().log_error(message, print=not self.error_capture_only)
+
     def log_success(self, message: str) -> None:
         self._print(message)
 
@@ -3427,19 +3499,24 @@ def log_test_results(self, result: ModelTextTestResult, target_dialect: str) ->
 
     def log_skipped_models(self, snapshot_names: t.Set[str]) -> None:
         if snapshot_names:
-            msg = "  " + "\n  ".join(snapshot_names)
-            self._print(f"**Skipped models**\n\n{msg}")
+            self._print(f"**Skipped models**")
+            for snapshot_name in snapshot_names:
+                self._print(f"* `{snapshot_name}`")
+            self._print("")
 
     def log_failed_models(self, errors: t.List[NodeExecutionFailedError]) -> None:
         if errors:
-            self._print("\n```\nFailed models\n")
+            self._print("**Failed models**")
 
             error_messages = _format_node_errors(errors)
 
             for node_name, msg in error_messages.items():
-                self._print(f"  **{node_name}**\n\n{msg}")
+                self._print(f"* `{node_name}`\n")
+                self._print("  ```")
+                self._print(msg)
+                self._print("  ```")
 
-            self._print("```\n")
+            self._print("")
 
     def show_linter_violations(
         self, violations: t.List[RuleViolation], model: Model, is_error: bool = False