[Feature] [spark] Support failure-safe staged REPLACE TABLE AS SELECT

[Akash3121]

Search before asking

• I searched in the issues and found nothing similar.

Motivation

Spark can route compatible REPLACE TABLE AS SELECT operations through its staged/atomic execution path, but Paimon's current stageReplace implementation replaces the live table before the query output is written.

SparkCatalog#stageReplaceInternal currently calls catalog.replaceTable immediately and returns a RollbackStagedTable with an empty abort action. The underlying replace implementation is documented as non-atomic and can truncate the table before appending the new schema.

For example:

SET spark.sql.ansi.enabled=true;

CREATE TABLE src(s STRING) USING paimon;
INSERT INTO src VALUES ('bad');

CREATE TABLE target(id BIGINT, data STRING) USING paimon;
INSERT INTO target VALUES (1, 'old');

CREATE OR REPLACE TABLE target USING paimon
AS SELECT CAST(s AS INT) AS id FROM src;

The cast fails during the write. However, the target may already have been truncated and changed to the replacement schema.

A failed staged RTAS should leave the original target schema, snapshots, options, and data unchanged. Losing the live table after a failed replacement is especially surprising when Spark selected an execution node named AtomicReplaceTableAsSelectExec.

Solution

Stage replacement output separately and publish it only from commitStagedChanges.

Possible approaches include:

1. Write into an isolated temporary branch, table, or location.
2. Keep the existing target untouched during the query write.
3. Atomically or transactionally publish the staged replacement from commitStagedChanges where the catalog supports it.
4. Delete staged artifacts from abortStagedChanges.
5. If a catalog cannot provide safe staging, do not advertise/use Spark's atomic replacement path and fail clearly instead of exposing a false rollback contract.

Suggested tests should inject an executor-side failure after staging and verify that the original schema, latest snapshot, rows, and options are unchanged. Coverage should include Spark 3 and Spark 4 and the filesystem, Hive, and REST catalog implementations.

Anything else?

Relevant code:

• PaimonReplaceTableAsSelectStrategy selects Spark's atomic RTAS execution path.
• SparkCatalog#stageReplaceInternal eagerly invokes catalog.replaceTable.
• The returned RollbackStagedTable receives an empty abort action.
• AbstractCatalog documents replace as non-atomic on failure.

This is a larger feature than the other candidates and likely needs maintainer agreement on the staging model before implementation.

Are you willing to submit a PR?

• I'm willing to submit a PR!

[Akash3121]

I'm willing to submit a pr for this issue