UPD: Merged Sqlglot into the branch and addressed feedback

Author: alkaline-0

dlt/destinations/impl/sqlalchemy/merge_job.py

@@ -375,9 +375,12 @@ def gen_scd2_sql(
             format_datetime_literal = (
                 DestinationCapabilitiesContext.generic_capabilities().format_datetime_literal
             )
-        created_at = current_load_package()["state"]["created_at"]
         _boundary_ts = cast(Optional[TAnyDateTime], root_table.get("x-boundary-timestamp"))
-        boundary_ts: TAnyDateTime = _boundary_ts if _boundary_ts is not None else created_at
+        boundary_ts: TAnyDateTime = (
+            _boundary_ts
+            if _boundary_ts is not None
+            else current_load_package()["state"]["created_at"]
+        )
         boundary_ts = ensure_pendulum_datetime_utc(boundary_ts)
 
         boundary_literal = format_datetime_literal(boundary_ts, caps.timestamp_precision)

dlt/extract/hints.py

@@ -823,24 +823,21 @@ def validate_write_disposition_hint(template: TResourceHints) -> None:
 
             if wd.get("strategy") == "scd2":
                 wd = cast(TScd2StrategyDict, wd)
-
-                art = wd.get("active_record_timestamp")
-                if art is not None:
-                    try:
-                        ensure_pendulum_datetime_utc(art)
-                    except (ValueError, TypeError) as exc:
-                        raise ValueError(
-                            f"could not parse `active_record_timestamp` value `{art}`"
-                        ) from exc
-
-                bt = wd.get("boundary_timestamp")
-                if bt is not None:
-                    try:
-                        ensure_pendulum_datetime_utc(bt)
-                    except (ValueError, TypeError) as exc:
-                        raise ValueError(
-                            f"could not parse `boundary_timestamp` value `{bt}`"
-                        ) from exc
+                for ts in ("active_record_timestamp", "boundary_timestamp"):
+                    if (
+                        ts == "active_record_timestamp"
+                        and wd.get("active_record_timestamp") is None
+                    ):
+                        continue  # None is allowed for active_record_timestamp
+                    if ts in wd:
+                        if wd[ts] is None:  # type: ignore[literal-required]
+                            continue
+                        try:
+                            ensure_pendulum_datetime_utc(wd[ts])  # type: ignore[literal-required]
+                        except Exception:
+                            raise ValueError(
+                                f"could not parse `{ts}` value `{wd[ts]}`"  # type: ignore[literal-required]
+                            )
 
     @staticmethod
     def validate_reference_hint(template: TResourceHints) -> None: