Merge branch 'main' into fix/get-data-objects

Author: themisvaltinos

pyproject.toml

@@ -24,7 +24,7 @@ dependencies = [
     "requests",
     "rich[jupyter]",
     "ruamel.yaml",
-    "sqlglot[rs]~=27.24.2",
+    "sqlglot[rs]~=27.27.0",
     "tenacity",
     "time-machine",
     "json-stream"

sqlmesh/__init__.py

@@ -188,15 +188,23 @@ def configure_logging(
     write_to_file: bool = True,
     log_file_dir: t.Optional[t.Union[str, Path]] = None,
     ignore_warnings: bool = False,
+    log_level: t.Optional[t.Union[str, int]] = None,
 ) -> None:
     # Remove noisy grpc logs that are not useful for users
     os.environ["GRPC_VERBOSITY"] = os.environ.get("GRPC_VERBOSITY", "NONE")
 
     logger = logging.getLogger()
     debug = force_debug or debug_mode_enabled()
 
-    # base logger needs to be the lowest level that we plan to log
-    level = logging.DEBUG if debug else logging.INFO
+    if log_level is not None:
+        if isinstance(log_level, str):
+            level = logging._nameToLevel.get(log_level.upper()) or logging.INFO
+        else:
+            level = log_level
+    else:
+        # base logger needs to be the lowest level that we plan to log
+        level = logging.DEBUG if debug else logging.INFO
+
     logger.setLevel(level)
 
     if debug:

sqlmesh/core/config/connection.py

@@ -58,6 +58,7 @@
     "clickhouse",
 }
 MOTHERDUCK_TOKEN_REGEX = re.compile(r"(\?|\&)(motherduck_token=)(\S*)")
+PASSWORD_REGEX = re.compile(r"(password=)(\S+)")
 
 
 def _get_engine_import_validator(
@@ -479,13 +480,13 @@ def create_engine_adapter(
             adapter = BaseDuckDBConnectionConfig._data_file_to_adapter.get(key)
             if adapter is not None:
                 logger.info(
-                    f"Using existing DuckDB adapter due to overlapping data file: {self._mask_motherduck_token(key)}"
+                    f"Using existing DuckDB adapter due to overlapping data file: {self._mask_sensitive_data(key)}"
                 )
                 return adapter
 
         if data_files:
             masked_files = {
-                self._mask_motherduck_token(file if isinstance(file, str) else file.path)
+                self._mask_sensitive_data(file if isinstance(file, str) else file.path)
                 for file in data_files
             }
             logger.info(f"Creating new DuckDB adapter for data files: {masked_files}")
@@ -507,10 +508,14 @@ def get_catalog(self) -> t.Optional[str]:
             return list(self.catalogs)[0]
         return None
 
-    def _mask_motherduck_token(self, string: str) -> str:
-        return MOTHERDUCK_TOKEN_REGEX.sub(
-            lambda m: f"{m.group(1)}{m.group(2)}{'*' * len(m.group(3))}", string
+    def _mask_sensitive_data(self, string: str) -> str:
+        # Mask MotherDuck tokens with fixed number of asterisks
+        result = MOTHERDUCK_TOKEN_REGEX.sub(
+            lambda m: f"{m.group(1)}{m.group(2)}{'*' * 8 if m.group(3) else ''}", string
         )
+        # Mask PostgreSQL/MySQL passwords with fixed number of asterisks
+        result = PASSWORD_REGEX.sub(lambda m: f"{m.group(1)}{'*' * 8}", result)
+        return result
 
 
 class MotherDuckConnectionConfig(BaseDuckDBConnectionConfig):

sqlmesh/core/config/loader.py

@@ -178,6 +178,7 @@ def load_config_from_paths(
 
             dbt_python_config = sqlmesh_config(
                 project_root=dbt_project_file.parent,
+                profiles_dir=kwargs.pop("profiles_dir", None),
                 dbt_profile_name=kwargs.pop("profile", None),
                 dbt_target_name=kwargs.pop("target", None),
                 variables=variables,

sqlmesh/core/context.py

@@ -139,6 +139,7 @@
 )
 from sqlmesh.utils.config import print_config
 from sqlmesh.utils.jinja import JinjaMacroRegistry
+from sqlmesh.utils.windows import IS_WINDOWS, fix_windows_path
 
 if t.TYPE_CHECKING:
     import pandas as pd
@@ -2590,12 +2591,15 @@ def table_name(
         )
 
     def clear_caches(self) -> None:
-        for path in self.configs:
-            cache_path = path / c.CACHE
-            if cache_path.exists():
-                rmtree(cache_path)
-        if self.cache_dir.exists():
-            rmtree(self.cache_dir)
+        paths_to_remove = [path / c.CACHE for path in self.configs]
+        paths_to_remove.append(self.cache_dir)
+
+        if IS_WINDOWS:
+            paths_to_remove = [fix_windows_path(path) for path in paths_to_remove]
+
+        for path in paths_to_remove:
+            if path.exists():
+                rmtree(path)
 
         if isinstance(self._state_sync, CachingStateSync):
             self._state_sync.clear_cache()

sqlmesh/core/engine_adapter/base.py

@@ -18,7 +18,7 @@
 
 from sqlglot import Dialect, exp
 from sqlglot.errors import ErrorLevel
-from sqlglot.helper import ensure_list
+from sqlglot.helper import ensure_list, seq_get
 from sqlglot.optimizer.qualify_columns import quote_identifiers
 
 from sqlmesh.core.dialect import (
@@ -551,11 +551,13 @@ def replace_query(
                     target_table,
                     source_queries,
                     target_columns_to_types,
+                    **kwargs,
                 )
         return self._insert_overwrite_by_condition(
             target_table,
             source_queries,
             target_columns_to_types,
+            **kwargs,
         )
 
     def create_index(
@@ -1614,7 +1616,7 @@ def _insert_overwrite_by_time_partition(
         **kwargs: t.Any,
     ) -> None:
         return self._insert_overwrite_by_condition(
-            table_name, source_queries, target_columns_to_types, where
+            table_name, source_queries, target_columns_to_types, where, **kwargs
         )
 
     def _values_to_sql(
@@ -1772,7 +1774,7 @@ def scd_type_2_by_column(
         valid_from_col: exp.Column,
         valid_to_col: exp.Column,
         execution_time: t.Union[TimeLike, exp.Column],
-        check_columns: t.Union[exp.Star, t.Sequence[exp.Column]],
+        check_columns: t.Union[exp.Star, t.Sequence[exp.Expression]],
         invalidate_hard_deletes: bool = True,
         execution_time_as_valid_from: bool = False,
         target_columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None,
@@ -1810,7 +1812,7 @@ def _scd_type_2(
         execution_time: t.Union[TimeLike, exp.Column],
         invalidate_hard_deletes: bool = True,
         updated_at_col: t.Optional[exp.Column] = None,
-        check_columns: t.Optional[t.Union[exp.Star, t.Sequence[exp.Column]]] = None,
+        check_columns: t.Optional[t.Union[exp.Star, t.Sequence[exp.Expression]]] = None,
         updated_at_as_valid_from: bool = False,
         execution_time_as_valid_from: bool = False,
         target_columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None,
@@ -1885,8 +1887,10 @@ def remove_managed_columns(
         # they are equal or not, the extra check is not a problem and we gain simplified logic here.
         # If we want to change this, then we just need to check the expressions in unique_key and pull out the
         # column names and then remove them from the unmanaged_columns
-        if check_columns and check_columns == exp.Star():
-            check_columns = [exp.column(col) for col in unmanaged_columns_to_types]
+        if check_columns:
+            # Handle both Star directly and [Star()] (which can happen during serialization/deserialization)
+            if isinstance(seq_get(ensure_list(check_columns), 0), exp.Star):
+                check_columns = [exp.column(col) for col in unmanaged_columns_to_types]
         execution_ts = (
             exp.cast(execution_time, time_data_type, dialect=self.dialect)
             if isinstance(execution_time, exp.Column)
@@ -1923,7 +1927,8 @@ def remove_managed_columns(
                 col_qualified.set("table", exp.to_identifier("joined"))
 
                 t_col = col_qualified.copy()
-                t_col.this.set("this", f"t_{col.name}")
+                for column in t_col.find_all(exp.Column):
+                    column.this.set("this", f"t_{column.name}")
 
                 row_check_conditions.extend(
                     [

sqlmesh/core/engine_adapter/mssql.py

@@ -56,6 +56,7 @@ class MSSQLEngineAdapter(
     COMMENT_CREATION_TABLE = CommentCreationTable.UNSUPPORTED
     COMMENT_CREATION_VIEW = CommentCreationView.UNSUPPORTED
     SUPPORTS_REPLACE_TABLE = False
+    MAX_IDENTIFIER_LENGTH = 128
     SUPPORTS_QUERY_EXECUTION_TRACKING = True
     SCHEMA_DIFFER_KWARGS = {
         "parameterized_type_defaults": {
@@ -422,7 +423,9 @@ def _insert_overwrite_by_condition(
         insert_overwrite_strategy_override: t.Optional[InsertOverwriteStrategy] = None,
         **kwargs: t.Any,
     ) -> None:
-        if not where or where == exp.true():
+        # note that this is passed as table_properties here rather than physical_properties
+        use_merge_strategy = kwargs.get("table_properties", {}).get("mssql_merge_exists")
+        if (not where or where == exp.true()) and not use_merge_strategy:
             # this is a full table replacement, call the base strategy to do DELETE+INSERT
             # which will result in TRUNCATE+INSERT due to how we have overridden self.delete_from()
             return EngineAdapter._insert_overwrite_by_condition(
@@ -435,7 +438,7 @@ def _insert_overwrite_by_condition(
                 **kwargs,
             )
 
-        # For actual conditional overwrites, use MERGE from InsertOverwriteWithMergeMixin
+        # For conditional overwrites or when mssql_merge_exists is set use MERGE
         return super()._insert_overwrite_by_condition(
             table_name=table_name,
             source_queries=source_queries,

sqlmesh/core/engine_adapter/trino.py

@@ -302,7 +302,7 @@ def _scd_type_2(
         execution_time: t.Union[TimeLike, exp.Column],
         invalidate_hard_deletes: bool = True,
         updated_at_col: t.Optional[exp.Column] = None,
-        check_columns: t.Optional[t.Union[exp.Star, t.Sequence[exp.Column]]] = None,
+        check_columns: t.Optional[t.Union[exp.Star, t.Sequence[exp.Expression]]] = None,
         updated_at_as_valid_from: bool = False,
         execution_time_as_valid_from: bool = False,
         target_columns_to_types: t.Optional[t.Dict[str, exp.DataType]] = None,

sqlmesh/core/model/kind.py

@@ -23,7 +23,7 @@
     PydanticModel,
     SQLGlotBool,
     SQLGlotColumn,
-    SQLGlotListOfColumnsOrStar,
+    SQLGlotListOfFieldsOrStar,
     SQLGlotListOfFields,
     SQLGlotPositiveInt,
     SQLGlotString,
@@ -852,7 +852,7 @@ def to_expression(
 
 class SCDType2ByColumnKind(_SCDType2Kind):
     name: t.Literal[ModelKindName.SCD_TYPE_2_BY_COLUMN] = ModelKindName.SCD_TYPE_2_BY_COLUMN
-    columns: SQLGlotListOfColumnsOrStar
+    columns: SQLGlotListOfFieldsOrStar
     execution_time_as_valid_from: SQLGlotBool = False
     updated_at_name: t.Optional[SQLGlotColumn] = None
 

sqlmesh/core/snapshot/evaluator.py

@@ -2188,7 +2188,13 @@ def _get_target_and_source_columns(
         if model.on_destructive_change.is_ignore or model.on_additive_change.is_ignore:
             # We need to identify the columns that are only in the source so we create an empty table with
             # the user query to determine that
-            with self.adapter.temp_table(model.ctas_query(**render_kwargs)) as temp_table:
+            temp_table_name = exp.table_(
+                "diff",
+                db=model.physical_schema,
+            )
+            with self.adapter.temp_table(
+                model.ctas_query(**render_kwargs), name=temp_table_name
+            ) as temp_table:
                 source_columns = list(self.adapter.columns(temp_table))
         else:
             source_columns = None