fix(scd2): preserve NULL values in historical rows

Select target values using an explicit row-existence flag carried into the SCD2 join. Choose a flag alias that does not overlap source or prefixed target columns, including case-insensitive names.

Co-authored-by: Nicholas <71755971+ionnich@users.noreply.github.com>
Signed-off-by: Arthur031221 <levi74108520963@gmail.com>

Author: Arthur031221

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, seq_get
+from sqlglot.helper import ensure_list, find_new_name, seq_get
 from sqlglot.optimizer.qualify_columns import quote_identifiers
 
 from sqlmesh.core.dialect import (
@@ -2073,6 +2073,11 @@ def remove_managed_columns(
                 prefixed_col = exp.column(column).copy()
                 prefixed_col.this.set("this", f"t_{prefixed_col.name}")
                 prefixed_unmanaged_columns.append(prefixed_col)
+            target_exists_column = find_new_name(
+                {col.name.lower() for col in prefixed_columns_to_types}
+                | {col.lower() for col in unmanaged_columns_to_types},
+                "t__exists",
+            )
             query = (
                 exp.Select()  # type: ignore
                 .select(*table_columns)
@@ -2141,6 +2146,7 @@ def remove_managed_columns(
                     "joined",
                     exp.select(
                         exp.column("_exists", table="source").as_("_exists"),
+                        exp.column("_exists", table="latest").as_(target_exists_column),
                         *(
                             exp.column(col, table="latest").as_(prefixed_columns_to_types[i].this)
                             for i, col in enumerate(target_columns_to_types)
@@ -2164,6 +2170,7 @@ def remove_managed_columns(
                     .union(
                         exp.select(
                             exp.column("_exists", table="source").as_("_exists"),
+                            exp.column("_exists", table="latest").as_(target_exists_column),
                             *(
                                 exp.column(col, table="latest").as_(
                                     prefixed_columns_to_types[i].this
@@ -2195,11 +2202,15 @@ def remove_managed_columns(
                     "updated_rows",
                     exp.select(
                         *(
-                            exp.func(
-                                "COALESCE",
+                            exp.Case()
+                            .when(
+                                exp.column(target_exists_column, table="joined")
+                                .is_(exp.Null())
+                                .not_(),
                                 exp.column(prefixed_unmanaged_columns[i].this, table="joined"),
-                                exp.column(col, table="joined"),
-                            ).as_(col)
+                            )
+                            .else_(exp.column(col, table="joined"))
+                            .as_(col)
                             for i, col in enumerate(unmanaged_columns_to_types)
                         ),
                         valid_from_case_stmt,