[WIP][SQL][DML] DSv2 Transaction Management

[andreaschat-db]

What changes were proposed in this pull request?

Currently, DSv2 is lacking the required abstractions and machinery for allowing transactionality in DML operations.

First, this PR introduces the public Java interfaces that connectors need to implement. These are the following:

TransactionInfo — carries the transaction metadata.
Transaction — represents a transaction. Exposes catalog(), commit(), abort(), and close().
TransactionalCatalogPlugin — extends CatalogPlugin with beginTransaction(TransactionInfo) method.

Second, it expands Spark to manage the Transaction lifecycle. This is done as follows:

• Pre-analysis. At pre-analysis, we look at the plan for logical nodes that implement the TransactionalWrite trait. When a plan contains such a node we initiate a transaction. Commands that do not result in execution, e.g. EXPLAIN, we should never initiate a transaction.

• Analysis. We create a copy of the analyzer that contains the TransactionAwareCatalogManager instance. This is a CatalogManager that is aware of the current transaction and intercepts catalog(name) and currentCatalog lookups by returning the transaction catalog instead of the default session catalog. All catalog look ups within the query go through the TransactionAwareCatalogManager.

• Planning. At planning we need to propagate the transaction to the execution nodes. This is necessary so we can commit the transaction before the relation cache is refreshed. This process occurs post planning where the plan is fully formed. We walk the plan and inject the transaction callback to each TransactionalExec node.

• Execution. V2ExistingTableWriteExec nodes invoke the transaction commit right after the write operation and before the relation cache is refreshed.

• Abort/close. Every QE operation, i.e. analyzed, commandExecuted, optimizedPlan, sparkPlan, executedPlan etc., is wrapped with executeWithTransactionContext closure. This ensures that if an issue occurs at any point throughout the QE, the transition is aborted and closed.

Note, since we can have multiple QueryExecution instantiations for the same operation, the abort/close operations need to be idempotent. Due the nested structure of query execution, we might end up triggering nested abort/close calls.

Future work

• Introduce transactionality to Streaming.
• Introduce transactionality to atomic CTAS/RTAS.

This PR is based on @aokolnychyi's prototype.

Why are the changes needed?

We are currently lacking the required abstractions and machinery for DSv2 connectors to implement transactions in write operations.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Introduced new suites and added tests in existing suites.

Was this patch authored or co-authored using generative AI tooling?

Claude Sonnet 4.6.