feat(fabric): support staging-optimized replace via DDL transactions

[sdebruyn]

Description

Fabric Warehouse, unlike Synapse dedicated SQL pools, supports DDL transactions and ALTER SCHEMA ... TRANSFER. This PR enables the staging-optimized replace strategy for the fabric destination.

Three parts:

1. Capabilities (dlt/destinations/impl/fabric/factory.py): override the Synapse-inherited capabilities to advertise DDL transactions and the optimized strategy:

   • supports_ddl_transactions = True
   • supported_replace_strategies = ["truncate-and-insert", "insert-from-staging", "staging-optimized"]

2. Replace-job routing (dlt/destinations/impl/fabric/fabric.py): SynapseClient._create_replace_followup_jobs deliberately bypasses MsSqlJobClient's replace logic and always falls back to the generic truncate-and-insert job, because Synapse dedicated pools do not support DDL transactions. FabricClient inherited that bypass, so flipping the capabilities alone would advertise staging-optimized without ever emitting ALTER SCHEMA ... TRANSFER, silently behaving like insert-from-staging. FabricClient now overrides _create_replace_followup_jobs to route the staging-optimized strategy to the same MsSqlStagingReplaceJob (ALTER SCHEMA TRANSFER) the mssql destination uses, while keeping the Synapse-style behavior for truncate-and-insert and insert-from-staging.

3. Concurrent-load safety (dlt/destinations/job_client_impl.py, dlt/destinations/impl/fabric/fabric.py).

   The swap generated by MsSqlStagingReplaceJob (DROP TABLE, ALTER SCHEMA ... TRANSFER, SELECT ... INTO) has to run as one atomic transaction, but it was running fully autocommitted, statement by statement. With several table chains replaced concurrently this silently dropped rows: a second chain's swap could interleave with a first still mid-swap, and one chain's data was lost.

   The cause was SqlLoadJob._has_out_of_transaction_commands, which forces a job to run outside a transaction. It had two triggers: a blanket UNLOGGED_COMMANDS = ["ALTER SCHEMA"] match, and a check for SynapseClient. Why UNLOGGED_COMMANDS = ["ALTER SCHEMA"] is removed here:

   • it already carried a # TODO: move to respective sql clients note, i.e. it was flagged as a blanket check living in the wrong place;
   • the SynapseClient trigger (kept) already forces every Synapse sql job into autocommit, so the ALTER SCHEMA match was redundant for Synapse, which in any case never emits this swap (its _create_replace_followup_jobs always falls back to the clone/insert job);
   • the only SQL in the whole codebase that contains ALTER SCHEMA is MsSqlStagingReplaceJob.generate_sql (dlt/destinations/impl/mssql/mssql.py), used by mssql and fabric. Both advertise supports_ddl_transactions = True and are not SynapseClient, so this match was the only thing forcing their swap out of a transaction, which is exactly what broke atomicity.

   Removing the constant therefore leaves Synapse unchanged (still autocommit via the SynapseClient trigger, and it never reaches the swap) and restores the intended atomic swap for both fabric and mssql (the mssql case was a latent bug fixed here as well).

   Atomicity alone is not enough on Fabric: it has no sp_getapplock, so two atomic swaps for different chains committed at the same time can still race against the same destination schema. A fabric-specific FabricStagingReplaceLoadJob serializes the swap with an in-process lock keyed by destination dataset, which is sufficient because the loader always runs sql jobs on a thread pool, never across processes. The lock is fabric-specific: mssql relies on the SQL Server engine's own schema-level locking and is left unchanged.

Related Issues

• Closes Fabric Warehouse: support the staging-optimized replace strategy #4144

Additional Context

Tests in tests/load/fabric/test_fabric_table_builder.py:

• test_fabric_capabilities asserts DDL transactions and staging-optimized are advertised;
• test_fabric_staging_optimized_replace_uses_alter_schema_transfer asserts the generated replace-job SQL contains ALTER SCHEMA, TRANSFER and DROP TABLE IF EXISTS;
• test_fabric_insert_from_staging_replace_unaffected guards that the non-optimized strategies stay unchanged.

tests/load/pipeline/test_replace_disposition.py::test_replace_sql_queries now also covers fabric, which reuses mssql's MsSqlStagingReplaceJob.

Validated end to end against a live Fabric Warehouse: the multi-table staging-optimized replace keeps every child-table row across repeated concurrent runs, and the single-chain and non-optimized strategies are unaffected.