fix(db2): _truncate_table — IMMEDIATE outside transactions, DELETE inside

Two separate Db2 constraints require this dual approach:

  SQL0104N — TRUNCATE TABLE without IMMEDIATE fails; the keyword is
             mandatory in Db2 syntax and the base class does not add it.

  SQL0428N — TRUNCATE TABLE ... IMMEDIATE commits instantly and must be
             the first statement in a unit of work; it cannot run inside
             an open transaction and cannot be rolled back.

When a transaction is already active, fall back to DELETE which
participates in the transaction normally and can be rolled back.
When no transaction is active, TRUNCATE TABLE ... IMMEDIATE runs as
the first statement in a fresh unit of work and succeeds.

This mirrors the intent of NonTransactionalTruncateMixin (used by
MySQL and Redshift) but that mixin delegates to base._truncate_table()
which omits IMMEDIATE — making it unsuitable for Db2 without an
additional override.

Author: iawanish

sqlmesh/core/engine_adapter/db2.py

@@ -733,11 +733,19 @@ def _create_table_like(
         )
 
     def _truncate_table(self, table_name: TableName) -> None:
-        # Db2 requires the IMMEDIATE keyword after the table name; without it
-        # the statement fails with SQL0104N (unexpected token END-OF-STATEMENT,
-        # expected IMMEDIATE).
-        table = exp.to_table(table_name)
-        self.execute(f"TRUNCATE TABLE {table.sql(dialect=self.dialect, identify=True)} IMMEDIATE")
+        # Db2's TRUNCATE TABLE ... IMMEDIATE commits instantly and cannot be
+        # rolled back (SQL0428N if inside an open transaction). When a
+        # transaction is already active, use DELETE which participates in the
+        # transaction normally and can be rolled back. When no transaction is
+        # active, use TRUNCATE TABLE ... IMMEDIATE — the IMMEDIATE keyword is
+        # mandatory in Db2 syntax (SQL0104N without it).
+        if self._connection_pool.is_transaction_active:
+            self.execute(exp.Delete(this=exp.to_table(table_name)))
+        else:
+            table = exp.to_table(table_name)
+            self.execute(
+                f"TRUNCATE TABLE {table.sql(dialect=self.dialect, identify=True)} IMMEDIATE"
+            )
 
     def _convert_df_datetime(self, df: DF, columns_to_types: t.Dict[str, exp.DataType]) -> None:
         """