[SPARK-55855][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.

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.

[dongjoon-hyun]

Hi, @andreaschat-db . FYI, GitHub provides Draft feature to help your case. To prevent merging or reviewing systematically, the Apache Spark community recommend to use the standard Draft PR instead of relying on a plain text [WIP] in the PR title.

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

In addition, please get a proper JIRA ID and use it in the PR title before converting it back to a normal PR.

[aokolnychyi]

This seems like a fragile place. Am I right it is a hard requirement to delegate to the original analyzer here?

@andreaschat-db @juliuszsompolski, thoughts?

[aokolnychyi]

How can we ensure once we add more state to the analyzer it is used correctly here?

[andreaschat-db]

Yes it is a hard requirement because otherwise we will silently drop all registered extensions. In general, anyone who adds a new extension needs to also amend this method. The method is defined in the Analyzer itself so it should not be that easy to miss.

In any case, just to be sure, I added a new suite/test, i.e. AnalyzerExtensionPropagationSuite, where I use reflection to verify all known extensions. If anyone adds a new extension the test will fail and point to withCatalogManager.

[aokolnychyi]

I'd tighten the test a bit more, left some comments there.

[aokolnychyi]

I am not sure I understand our plan here. Is this method supposed to resolve only references in views or all of them? Based on usage, it looks like we only resolve references in views. If so, can we rename the method to reflect that and do a check on the context in V2TableReference?

[aokolnychyi]

The comment about the transaction reference is misleading here.

[andreaschat-db]

The comment was wrong here. The V2TableReference resolution for transactions happens in line 1024. Renamed the function and fixed comment.

[aokolnychyi]

Can we re-assess whether this is needed? We have to skip data cache for sure, I am not fully convinced here.

@andreaschat-db, I want you to challenge this. Why isn't it safe to reuse the relation cache in the Analyzer given that we start the transaction before resolution and we unresolve previously resolved relations? Or are we doing this to be safe and we don't think it is going to cause a regression?

The key difference is having one table and multiple scans on it vs two tables with one scan in each.

[aokolnychyi]

To be clear: Justify, not necessarily remove this code.

[andreaschat-db]

That was only meant as a temporary limitation. It was a conservative approach until the interaction of the cache with the transaction is understood. I looked at it. My understanding is the following:

The AnalysisContext.get.relationCache is strictly scoped to a single Analyzer.execute call. So there is no interference outside the transaction and vice-versa. All tables participating in the transaction are resolved within the transaction. Therefore, all cache entries should hold transactional tables. Removed the guard.

[aokolnychyi]

I would probably move this comment as a method comment and add more context.

[andreaschat-db]

Done.

[aokolnychyi]

Can we document either in the method or class doc why we use allowInvokingTransformsInAnalyzer? It is my understanding that we have to use it because resolveOperators would not allow us to iterate over already resolved plans while it is exactly what we need here?

[andreaschat-db]

Yes resolveOperators skips subtrees that have already being marked as analyzed. Furthermore, allowInvokingTransformsInAnalyzer allows to suppress the assertNotAnalysisRule safety check, which forbids calling transform directly inside the analyzer when not within a resolveOperators call.

Added comment.

[aokolnychyi]

Minor: I am not sure this is the best name, it is not clear what transaction relation is. Is it more about unresolving relations in transactions? So more like UnresolveRelationsInTransaction or similar?

[andreaschat-db]

Yes better. Done.

[aokolnychyi]

Shall we add a similar comment for TemporaryViewContext for consistency?

[aokolnychyi]

What about validateTableIdentity? This case is different from temp views. Think about all extra checks that apply here. We also need to cover this with tests.

[andreaschat-db]

Added validation for tables ID and relevant test. There are already test covering validateCapturedColumns. We are currently missing tests that verify metadataColumnsChangedAfterAnalysis. We need to expand testing infra to support that.

[aokolnychyi]

Is this needed?

[andreaschat-db]

This was added by the schema evolution work. SupportsSchemaEvolution was later renamed to WriteWithSchemaEvolution. Corrected the stale comment. Target gives a handle to target to WriteWithSchemaEvolution so it can perform the evolution magic.

[aokolnychyi]

It is a bit unfortunate to do this. I wonder whether we can avoid this.

[aokolnychyi]

Am I right that we sometimes create AppendData directly and some times go via statement?

[andreaschat-db]

Unfortunate indeed. I do not like at all that we are mixing v1 with v2 here but I thought refactoring this is out of the context of this PR.

Am I right that we sometimes create AppendData directly and some times go via statement?

Yes. Dataframe API creates directly AppendData.

[aokolnychyi]

Let's tackle this separately.

[aokolnychyi]

I wonder whether we should use appropriate Spark exceptions here.

[andreaschat-db]

Replaced it with SparkException.internalError.

[aokolnychyi]

Should we block calling this on TransactionAwareCatalogManager?

[andreaschat-db]

Yes great idea. Done.

[aokolnychyi]

How hard is it?

[andreaschat-db]

It is not that bad but I left it out because it would pollute the PR. It can be done in a small follow up PR.

[aokolnychyi]

Agreed.

[aokolnychyi]

This is also fragile so it would be great for extra eyes on it. cc @juliuszsompolski

[aokolnychyi]

I worry about other places that may access this directly? Can we prevent it somehow? Any thoughts?

[andreaschat-db]

External callers cannot access it because it is private. The only issue I see is transactional writes that create internal QE not propagating the transactional analyzer instance to the new QE instance. I added a new comment above analyzerOpt to stress this.

[aokolnychyi]

Will this work with scripting? Streaming? Can we test?

[andreaschat-db]

Expanded functionality to cover streaming. Added new suites for SQLScripting and Streaming.

[aokolnychyi]

This statement can be misleading. It could be a transactional commit from the connector point view. I think a better message would be to highlight that if called within Transaction, this should stage changes but not propagate them.

[andreaschat-db]

Updated both this and the one at BatchWrite.

[aokolnychyi]

Not sure I agree with this, it feels risky to enforce this. What about something like below?

private def resolveTableReferencesInTempView(plan: LogicalPlan): LogicalPlan = {
  plan.resolveOperatorsUp {
    case r: V2TableReference if r.context.isInstanceOf[TemporaryViewContext] =>
      relationResolution.resolveReference(r)
  }
}

[aokolnychyi]

I don't think this is robust enough. Can we instead count all public methods or maybe private fields and force whoever modifies Analyzer to also update the count here? Say today we have 10 fields. Create an analyzer and check we still have 10 fields. Then proceed with cloning and checking the clone like you have today.

[andreaschat-db]

Yes missed that. Fixed.

[aokolnychyi]

I am not sure I understand the purpose of this method.
Won't this always be a no-op for StreamingV2WriteCommand? Why add then?

[andreaschat-db]

The UnresolvedRelation relation we create in MicroBatchExecution has streaming = false. This then matched in resolveWriteTarget !u.isStreaming and gets resolved to a DataSourceV2Relation.

The alternative here would be to set to streaming in UnresolvedRelation to True. This then get resolved as a StreamingRelationV2 instead which we would need to add a case and finally resolve to DataSourceV2Relation (we would also need to drop the !u.isStreaming guard.

IIUC, streaming is a read side concept. For the write perspective, we write batches which are not streams any more. This is I think the reason we originally create a DataSourceV2Relation in MicroBatchExecution instead of a StreamingDataSourceV2Relation. So the path UnresolvedRelation with streaming=true -> StreamingRelationV2 -> StreamingDataSourceV2Relation is not applicable here. I added a comment to explain this.

Thoughts?

[aokolnychyi]

Minor: Drop () for simple getter for Option(relation.table.id)?

[aokolnychyi]

Will V2StreamingWriteCommand be more in line with other classes?

[andreaschat-db]

Is the question about naming? I renamed it to V2StreamingWriteCommand to be consistent with other commands. I also moved right after V2WriteCommand. Is this what you had in mind?

[aokolnychyi]

We have been reviewing this piece back and forth for a while. I don't have blocking comments, let's get this in and have a follow up for the remaining items.

[viirya]

Thanks for the patch. I have a design question about the transparency of transactional semantics.

With interfaces like SupportsDelete or SupportsRowLevelOperations, the capability is tied to a specific SQL command — if the connector doesn't support it, Spark rejects the command. The user gets an explicit signal.

TransactionalCatalogPlugin works differently. The same INSERT INTO (or MERGE, UPDATE, etc.) silently runs with or without transaction protection depending on whether the underlying catalog implements the interface. There is no user-visible difference — no error, no warning, no query plan annotation. A user who writes a pipeline assuming transactional semantics (e.g., skipping idempotency logic) would not know if they switch to a catalog that doesn't implement TransactionalCatalogPlugin.

This is somewhat analogous to StagingTableCatalog, which is also optional — but that one is scoped to CTAS/RTAS (commands with naturally "all-or-nothing" expectations), and its Javadoc explicitly documents the non-atomic fallback path. Here, the affected commands are general-purpose write operations where users may not think to question whether atomicity is present.

Should there be at least a mechanism for users to assert their expectation — e.g., a session config like spark.sql.requireTransactionalWrites that fails the query if the catalog doesn't support TransactionalCatalogPlugin? Or a note in EXPLAIN output indicating whether a transaction will be opened? Even just a WARN-level log at query start would give users a way to detect the difference.