This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new e2b5eb9d8c [spark] Record the write operation type in snapshot
properties (#8236)
e2b5eb9d8c is described below
commit e2b5eb9d8cffcd5e86763fba94c7d53c941fcedf
Author: Zouxxyy <[email protected]>
AuthorDate: Tue Jun 23 14:25:04 2026 +0800
[spark] Record the write operation type in snapshot properties (#8236)
Add a first-class `operation` field (`Snapshot.Operation` enum) to
`Snapshot`, recording the logical operation type that produced it. This
complements the physical `CommitKind` (APPEND/COMPACT/OVERWRITE/ANALYZE)
and lets downstream tooling distinguish, e.g., an APPEND from INSERT vs.
one from MERGE.
**Design:**
- `Snapshot.Operation` enum: `WRITE`, `OVERWRITE`, `DELETE`, `TRUNCATE`,
`UPDATE`, `MERGE`, `CREATE_TABLE_AS_SELECT`, `REPLACE_TABLE_AS_SELECT`,
`CREATE_OR_REPLACE_TABLE_AS_SELECT`
- Nullable field with `@JsonInclude(NON_NULL)` — old snapshots
deserialize as `null`, old readers ignore the unknown field via
`@JsonIgnoreProperties(ignoreUnknown = true)`
- `BatchTableCommit.withOperation(Operation)` — `default` method, no
breaking change for existing implementations
- `FileStoreCommit.withOperation(Operation)` — internal API, propagated
to `Snapshot` construction in `FileStoreCommitImpl`
- `TRUNCATE` is automatically set by
`TableCommitImpl.truncateTable()`/`truncatePartitions()` in core, so
callers don't need to handle it
**Spark coverage (both v1 and v2 write paths):**
| SQL | operation |
| --- | --- |
| INSERT INTO | `WRITE` |
| INSERT OVERWRITE | `OVERWRITE` |
| DELETE (row-level) | `DELETE` |
| DELETE (full-table / partition) | `TRUNCATE` |
| TRUNCATE TABLE | `TRUNCATE` |
| UPDATE | `UPDATE` |
| MERGE INTO | `MERGE` |
| CREATE TABLE AS SELECT | `CREATE_TABLE_AS_SELECT` |
| (CREATE OR) REPLACE TABLE AS SELECT | `REPLACE_TABLE_AS_SELECT` /
`CREATE_OR_REPLACE_TABLE_AS_SELECT` |
---
docs/docs/concepts/spec/snapshot.md | 3 +
.../src/main/java/org/apache/paimon/Snapshot.java | 131 ++++++++++++++++++++-
.../src/main/java/org/apache/paimon/Changelog.java | 9 +-
.../apache/paimon/operation/FileStoreCommit.java | 2 +
.../paimon/operation/FileStoreCommitImpl.java | 16 ++-
.../apache/paimon/table/sink/BatchTableCommit.java | 6 +
.../apache/paimon/table/sink/TableCommitImpl.java | 9 ++
.../apache/paimon/table/system/SnapshotsTable.java | 9 +-
.../src/main/java/org/apache/paimon/tag/Tag.java | 8 +-
.../test/java/org/apache/paimon/SnapshotTest.java | 46 ++++++++
.../DataEvolutionRowIdReassignerTest.java | 3 +-
.../paimon/catalog/RenamingSnapshotCommitTest.java | 1 +
.../paimon/operation/ExpireSnapshotsTest.java | 1 +
.../operation/commit/ConflictDetectionTest.java | 1 +
.../paimon/table/system/SnapshotsTableTest.java | 5 +-
.../org/apache/paimon/tag/TagAutoManagerTest.java | 2 +
.../test/java/org/apache/paimon/tag/TagTest.java | 1 +
.../apache/paimon/utils/SnapshotManagerTest.java | 5 +
.../shim/PaimonCreateTableAsSelectStrategy.scala | 7 +-
.../shim/PaimonCreateTableAsSelectStrategy.scala | 7 +-
.../shim/PaimonCreateTableAsSelectStrategy.scala | 7 +-
.../shim/PaimonReplaceTableAsSelectStrategy.scala | 8 +-
.../MergeIntoPaimonDataEvolutionTable.scala | 3 +-
.../spark/commands/MergeIntoPaimonTable.scala | 3 +-
.../paimon/spark/write/PaimonBatchWrite.scala | 23 +++-
.../apache/spark/sql/paimon/shims/Spark4Shim.scala | 12 +-
.../commands/DeleteFromPaimonTableCommand.scala | 3 +-
.../MergeIntoPaimonDataEvolutionTable.scala | 3 +-
.../spark/commands/MergeIntoPaimonTable.scala | 3 +-
.../paimon/spark/commands/PaimonSparkWriter.scala | 7 ++
.../spark/commands/UpdatePaimonTableCommand.scala | 3 +-
.../spark/commands/WriteIntoPaimonTable.scala | 13 +-
.../rowops/PaimonSparkCopyOnWriteOperation.scala | 2 +
.../paimon/spark/write/PaimonBatchWriteBase.scala | 7 +-
.../apache/paimon/spark/write/PaimonV2Write.scala | 7 +-
.../paimon/spark/write/PaimonV2WriteBuilder.scala | 17 ++-
...WriteBuilder.scala => PaimonWriteOptions.scala} | 25 +---
.../shim/PaimonCreateTableAsSelectStrategy.scala | 8 +-
.../shim/PaimonReplaceTableAsSelectStrategy.scala | 11 +-
.../apache/spark/sql/paimon/shims/SparkShim.scala | 4 +-
.../paimon/spark/sql/SnapshotOperationTest.scala | 110 +++++++++++++++++
.../paimon/spark/write/PaimonBatchWrite.scala | 23 +++-
.../apache/spark/sql/paimon/shims/Spark3Shim.scala | 12 +-
.../paimon/spark/write/PaimonBatchWrite.scala | 23 +++-
.../apache/spark/sql/paimon/shims/Spark4Shim.scala | 12 +-
45 files changed, 546 insertions(+), 75 deletions(-)
diff --git a/docs/docs/concepts/spec/snapshot.md
b/docs/docs/concepts/spec/snapshot.md
index f9762b0599..dd2bf28ed2 100644
--- a/docs/docs/concepts/spec/snapshot.md
+++ b/docs/docs/concepts/spec/snapshot.md
@@ -62,3 +62,6 @@ Snapshot File is JSON, it includes:
15. changelogRecordCount: record count of all changelog produced in this
snapshot.
16. watermark: watermark for input records, from Flink watermark mechanism,
Long.MIN_VALUE if there is no watermark.
17. statistics: stats file name for statistics of this table.
+18. properties: additional key-value properties of this snapshot.
+19. nextRowId: next row id for row tracking.
+20. operation: logical operation type, e.g. WRITE, DELETE, UPDATE, MERGE. Null
if not set.
diff --git a/paimon-api/src/main/java/org/apache/paimon/Snapshot.java
b/paimon-api/src/main/java/org/apache/paimon/Snapshot.java
index 93f525a7e3..4a200dfecc 100644
--- a/paimon-api/src/main/java/org/apache/paimon/Snapshot.java
+++ b/paimon-api/src/main/java/org/apache/paimon/Snapshot.java
@@ -69,6 +69,7 @@ public class Snapshot implements Serializable {
protected static final String FIELD_STATISTICS = "statistics";
protected static final String FIELD_PROPERTIES = "properties";
protected static final String FIELD_NEXT_ROW_ID = "nextRowId";
+ protected static final String FIELD_OPERATION = "operation";
// version of snapshot
@JsonProperty(FIELD_VERSION)
@@ -181,6 +182,11 @@ public class Snapshot implements Serializable {
@Nullable
protected final Long nextRowId;
+ @JsonInclude(JsonInclude.Include.NON_NULL)
+ @JsonProperty(FIELD_OPERATION)
+ @Nullable
+ protected final Operation operation;
+
public Snapshot(
long id,
long schemaId,
@@ -202,6 +208,52 @@ public class Snapshot implements Serializable {
@Nullable String statistics,
@Nullable Map<String, String> properties,
@Nullable Long nextRowId) {
+ this(
+ id,
+ schemaId,
+ baseManifestList,
+ baseManifestListSize,
+ deltaManifestList,
+ deltaManifestListSize,
+ changelogManifestList,
+ changelogManifestListSize,
+ indexManifest,
+ commitUser,
+ commitIdentifier,
+ commitKind,
+ timeMillis,
+ totalRecordCount,
+ deltaRecordCount,
+ changelogRecordCount,
+ watermark,
+ statistics,
+ properties,
+ nextRowId,
+ null);
+ }
+
+ public Snapshot(
+ long id,
+ long schemaId,
+ String baseManifestList,
+ @Nullable Long baseManifestListSize,
+ String deltaManifestList,
+ @Nullable Long deltaManifestListSize,
+ @Nullable String changelogManifestList,
+ @Nullable Long changelogManifestListSize,
+ @Nullable String indexManifest,
+ String commitUser,
+ long commitIdentifier,
+ CommitKind commitKind,
+ long timeMillis,
+ long totalRecordCount,
+ long deltaRecordCount,
+ @Nullable Long changelogRecordCount,
+ @Nullable Long watermark,
+ @Nullable String statistics,
+ @Nullable Map<String, String> properties,
+ @Nullable Long nextRowId,
+ @Nullable Operation operation) {
this(
CURRENT_VERSION,
id,
@@ -223,7 +275,55 @@ public class Snapshot implements Serializable {
watermark,
statistics,
properties,
- nextRowId);
+ nextRowId,
+ operation);
+ }
+
+ public Snapshot(
+ int version,
+ long id,
+ long schemaId,
+ String baseManifestList,
+ @Nullable Long baseManifestListSize,
+ String deltaManifestList,
+ @Nullable Long deltaManifestListSize,
+ @Nullable String changelogManifestList,
+ @Nullable Long changelogManifestListSize,
+ @Nullable String indexManifest,
+ String commitUser,
+ long commitIdentifier,
+ CommitKind commitKind,
+ long timeMillis,
+ long totalRecordCount,
+ long deltaRecordCount,
+ @Nullable Long changelogRecordCount,
+ @Nullable Long watermark,
+ @Nullable String statistics,
+ @Nullable Map<String, String> properties,
+ @Nullable Long nextRowId) {
+ this(
+ version,
+ id,
+ schemaId,
+ baseManifestList,
+ baseManifestListSize,
+ deltaManifestList,
+ deltaManifestListSize,
+ changelogManifestList,
+ changelogManifestListSize,
+ indexManifest,
+ commitUser,
+ commitIdentifier,
+ commitKind,
+ timeMillis,
+ totalRecordCount,
+ deltaRecordCount,
+ changelogRecordCount,
+ watermark,
+ statistics,
+ properties,
+ nextRowId,
+ null);
}
@JsonCreator
@@ -249,7 +349,8 @@ public class Snapshot implements Serializable {
@JsonProperty(FIELD_WATERMARK) @Nullable Long watermark,
@JsonProperty(FIELD_STATISTICS) @Nullable String statistics,
@JsonProperty(FIELD_PROPERTIES) @Nullable Map<String, String>
properties,
- @JsonProperty(FIELD_NEXT_ROW_ID) @Nullable Long nextRowId) {
+ @JsonProperty(FIELD_NEXT_ROW_ID) @Nullable Long nextRowId,
+ @JsonProperty(FIELD_OPERATION) @Nullable Operation operation) {
this.version = version;
this.id = id;
this.schemaId = schemaId;
@@ -271,6 +372,7 @@ public class Snapshot implements Serializable {
this.statistics = statistics;
this.properties = properties;
this.nextRowId = nextRowId;
+ this.operation = operation;
}
@JsonGetter(FIELD_VERSION)
@@ -388,6 +490,12 @@ public class Snapshot implements Serializable {
return nextRowId;
}
+ @JsonGetter(FIELD_OPERATION)
+ @Nullable
+ public Operation operation() {
+ return operation;
+ }
+
public String toJson() {
return JsonSerdeUtil.toJson(this);
}
@@ -415,7 +523,8 @@ public class Snapshot implements Serializable {
watermark,
statistics,
properties,
- nextRowId);
+ nextRowId,
+ operation);
}
@Override
@@ -447,7 +556,8 @@ public class Snapshot implements Serializable {
&& Objects.equals(watermark, that.watermark)
&& Objects.equals(statistics, that.statistics)
&& Objects.equals(properties, that.properties)
- && Objects.equals(nextRowId, that.nextRowId);
+ && Objects.equals(nextRowId, that.nextRowId)
+ && operation == that.operation;
}
/** Type of changes in this snapshot. */
@@ -469,6 +579,19 @@ public class Snapshot implements Serializable {
ANALYZE
}
+ /** Logical operation type that produced this snapshot. */
+ public enum Operation {
+ WRITE,
+ OVERWRITE,
+ DELETE,
+ TRUNCATE,
+ UPDATE,
+ MERGE,
+ CREATE_TABLE_AS_SELECT,
+ REPLACE_TABLE_AS_SELECT,
+ CREATE_OR_REPLACE_TABLE_AS_SELECT
+ }
+
// =================== Utils for reading =========================
public static Snapshot fromJson(String json) {
diff --git a/paimon-core/src/main/java/org/apache/paimon/Changelog.java
b/paimon-core/src/main/java/org/apache/paimon/Changelog.java
index 0ab3429dfc..05b76d7153 100644
--- a/paimon-core/src/main/java/org/apache/paimon/Changelog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/Changelog.java
@@ -63,7 +63,8 @@ public class Changelog extends Snapshot {
snapshot.watermark(),
snapshot.statistics(),
snapshot.properties,
- snapshot.nextRowId);
+ snapshot.nextRowId,
+ snapshot.operation);
}
@JsonCreator
@@ -89,7 +90,8 @@ public class Changelog extends Snapshot {
@JsonProperty(FIELD_WATERMARK) @Nullable Long watermark,
@JsonProperty(FIELD_STATISTICS) @Nullable String statistics,
@JsonProperty(FIELD_PROPERTIES) Map<String, String> properties,
- @JsonProperty(FIELD_NEXT_ROW_ID) @Nullable Long nextRowId) {
+ @JsonProperty(FIELD_NEXT_ROW_ID) @Nullable Long nextRowId,
+ @JsonProperty(FIELD_OPERATION) @Nullable Operation operation) {
super(
version,
id,
@@ -111,7 +113,8 @@ public class Changelog extends Snapshot {
watermark,
statistics,
properties,
- nextRowId);
+ nextRowId,
+ operation);
}
public static Changelog fromJson(String json) {
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java
index 31fb3c52ca..dbd316d8fa 100644
--- a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java
+++ b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommit.java
@@ -42,6 +42,8 @@ public interface FileStoreCommit extends AutoCloseable {
FileStoreCommit rowIdCheckConflict(@Nullable Long rowIdCheckFromSnapshot);
+ FileStoreCommit withOperation(Snapshot.Operation operation);
+
/** Find out which committables need to be retried when recovering from
the failure. */
List<ManifestCommittable> filterCommitted(List<ManifestCommittable>
committables);
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
index 47442f1b78..b97b12f74e 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
@@ -163,6 +163,7 @@ public class FileStoreCommitImpl implements FileStoreCommit
{
private CommitMetrics commitMetrics;
private boolean appendCommitCheckConflict = false;
private long lastCommittedSnapshotId = -1L;
+ @Nullable private Snapshot.Operation operation;
public FileStoreCommitImpl(
SnapshotCommit snapshotCommit,
@@ -251,6 +252,12 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
return this;
}
+ @Override
+ public FileStoreCommit withOperation(Snapshot.Operation operation) {
+ this.operation = operation;
+ return this;
+ }
+
@Override
public List<ManifestCommittable> filterCommitted(List<ManifestCommittable>
committables) {
// nothing to filter, fast exit
@@ -1107,7 +1114,8 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
statsFileName,
// if empty properties, just set to null
properties.isEmpty() ? null : properties,
- nextRowIdStart);
+ nextRowIdStart,
+ operation);
} catch (Throwable e) {
// fails when preparing for commit, we should clean up
commitCleaner.cleanUpReuseTmpManifests(
@@ -1261,7 +1269,8 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
latest.statistics(),
// if empty properties, just set to null
latest.properties(),
- nextRowId);
+ nextRowId,
+ null);
return commitSnapshotImpl(newSnapshot, emptyList());
}
@@ -1340,7 +1349,8 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
latestSnapshot.watermark(),
latestSnapshot.statistics(),
latestSnapshot.properties(),
- latestSnapshot.nextRowId());
+ latestSnapshot.nextRowId(),
+ null);
return commitSnapshotImpl(newSnapshot, emptyList());
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/sink/BatchTableCommit.java
b/paimon-core/src/main/java/org/apache/paimon/table/sink/BatchTableCommit.java
index 84215980d5..2cfc181ead 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/sink/BatchTableCommit.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/sink/BatchTableCommit.java
@@ -18,6 +18,7 @@
package org.apache.paimon.table.sink;
+import org.apache.paimon.Snapshot;
import org.apache.paimon.Snapshot.CommitKind;
import org.apache.paimon.annotation.Public;
import org.apache.paimon.stats.Statistics;
@@ -71,4 +72,9 @@ public interface BatchTableCommit extends TableCommit {
/** Compact the manifest entries. Generates a snapshot with {@link
CommitKind#COMPACT}. */
void compactManifests();
+
+ /** Set the logical operation type (e.g. WRITE, DELETE, MERGE) recorded in
the snapshot. */
+ default BatchTableCommit withOperation(Snapshot.Operation operation) {
+ return this;
+ }
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java
b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java
index 0311c9bbe4..6a9e3046cc 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/sink/TableCommitImpl.java
@@ -18,6 +18,7 @@
package org.apache.paimon.table.sink;
+import org.apache.paimon.Snapshot;
import org.apache.paimon.annotation.VisibleForTesting;
import org.apache.paimon.consumer.ConsumerManager;
import org.apache.paimon.fs.Path;
@@ -170,6 +171,12 @@ public class TableCommitImpl implements InnerTableCommit {
return this;
}
+ @Override
+ public TableCommitImpl withOperation(Snapshot.Operation operation) {
+ commit.withOperation(operation);
+ return this;
+ }
+
@Override
public InnerTableCommit withMetricRegistry(MetricRegistry registry) {
commit.withMetrics(new CommitMetrics(registry, tableName));
@@ -185,11 +192,13 @@ public class TableCommitImpl implements InnerTableCommit {
@Override
public void truncateTable() {
checkCommitted();
+ commit.withOperation(Snapshot.Operation.TRUNCATE);
commit.truncateTable(COMMIT_IDENTIFIER);
}
@Override
public void truncatePartitions(List<Map<String, String>> partitionSpecs) {
+ commit.withOperation(Snapshot.Operation.TRUNCATE);
commit.dropPartitions(partitionSpecs, COMMIT_IDENTIFIER);
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/table/system/SnapshotsTable.java
b/paimon-core/src/main/java/org/apache/paimon/table/system/SnapshotsTable.java
index 394dc4d58a..201192fde6 100644
---
a/paimon-core/src/main/java/org/apache/paimon/table/system/SnapshotsTable.java
+++
b/paimon-core/src/main/java/org/apache/paimon/table/system/SnapshotsTable.java
@@ -108,7 +108,9 @@ public class SnapshotsTable implements ReadonlyTable {
new DataField(10, "delta_record_count", new
BigIntType(true)),
new DataField(11, "changelog_record_count", new
BigIntType(true)),
new DataField(12, "watermark", new
BigIntType(true)),
- new DataField(13, "next_row_id", new
BigIntType(true))));
+ new DataField(13, "next_row_id", new
BigIntType(true)),
+ new DataField(
+ 14, "operation",
SerializationUtils.newStringType(true))));
private final FileIO fileIO;
private final Path location;
@@ -339,7 +341,10 @@ public class SnapshotsTable implements ReadonlyTable {
snapshot.deltaRecordCount(),
snapshot.changelogRecordCount(),
snapshot.watermark(),
- snapshot.nextRowId());
+ snapshot.nextRowId(),
+ snapshot.operation() == null
+ ? null
+ :
BinaryString.fromString(snapshot.operation().toString()));
}
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/tag/Tag.java
b/paimon-core/src/main/java/org/apache/paimon/tag/Tag.java
index 192df2d470..22d86b6fca 100644
--- a/paimon-core/src/main/java/org/apache/paimon/tag/Tag.java
+++ b/paimon-core/src/main/java/org/apache/paimon/tag/Tag.java
@@ -80,6 +80,7 @@ public class Tag extends Snapshot {
@JsonProperty(FIELD_STATISTICS) @Nullable String statistics,
@JsonProperty(FIELD_PROPERTIES) Map<String, String> properties,
@JsonProperty(FIELD_NEXT_ROW_ID) @Nullable Long nextRowId,
+ @JsonProperty(FIELD_OPERATION) @Nullable Operation operation,
@JsonProperty(FIELD_TAG_CREATE_TIME) @Nullable LocalDateTime
tagCreateTime,
@JsonProperty(FIELD_TAG_TIME_RETAINED) @Nullable Duration
tagTimeRetained) {
super(
@@ -103,7 +104,8 @@ public class Tag extends Snapshot {
watermark,
statistics,
properties,
- nextRowId);
+ nextRowId,
+ operation);
this.tagCreateTime = tagCreateTime;
this.tagTimeRetained = tagTimeRetained;
}
@@ -142,6 +144,7 @@ public class Tag extends Snapshot {
snapshot.statistics(),
snapshot.properties(),
snapshot.nextRowId(),
+ snapshot.operation(),
tagCreateTime,
tagTimeRetained);
}
@@ -168,7 +171,8 @@ public class Tag extends Snapshot {
watermark,
statistics,
properties,
- nextRowId);
+ nextRowId,
+ operation);
}
@Override
diff --git a/paimon-core/src/test/java/org/apache/paimon/SnapshotTest.java
b/paimon-core/src/test/java/org/apache/paimon/SnapshotTest.java
index 0cd75f3528..3a035c8a33 100644
--- a/paimon-core/src/test/java/org/apache/paimon/SnapshotTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/SnapshotTest.java
@@ -78,6 +78,52 @@ public class SnapshotTest {
assertThat(Snapshot.fromJson(snapshot.toJson())).isEqualTo(snapshot);
}
+ @Test
+ public void testSnapshotWithOperation() {
+ // Old snapshot without operation field: operation should be null
+ String oldJson =
+ "{\n"
+ + " \"version\" : 3,\n"
+ + " \"id\" : 1,\n"
+ + " \"schemaId\" : 0,\n"
+ + " \"baseManifestList\" : \"m-0\",\n"
+ + " \"deltaManifestList\" : \"m-1\",\n"
+ + " \"commitUser\" : \"user\",\n"
+ + " \"commitIdentifier\" : 0,\n"
+ + " \"commitKind\" : \"APPEND\",\n"
+ + " \"timeMillis\" : 1000,\n"
+ + " \"totalRecordCount\" : 10,\n"
+ + " \"deltaRecordCount\" : 5\n"
+ + "}";
+ Snapshot old = Snapshot.fromJson(oldJson);
+ assertThat(old.operation()).isNull();
+
+ // New snapshot with operation field
+ String newJson =
+ "{\n"
+ + " \"version\" : 3,\n"
+ + " \"id\" : 2,\n"
+ + " \"schemaId\" : 0,\n"
+ + " \"baseManifestList\" : \"m-0\",\n"
+ + " \"deltaManifestList\" : \"m-1\",\n"
+ + " \"commitUser\" : \"user\",\n"
+ + " \"commitIdentifier\" : 1,\n"
+ + " \"commitKind\" : \"APPEND\",\n"
+ + " \"timeMillis\" : 2000,\n"
+ + " \"totalRecordCount\" : 20,\n"
+ + " \"deltaRecordCount\" : 10,\n"
+ + " \"operation\" : \"MERGE\"\n"
+ + "}";
+ Snapshot withOp = Snapshot.fromJson(newJson);
+ assertThat(withOp.operation()).isEqualTo(Snapshot.Operation.MERGE);
+
+ // Round-trip: toJson -> fromJson preserves operation
+ assertThat(Snapshot.fromJson(withOp.toJson())).isEqualTo(withOp);
+
+ // Null operation is omitted in JSON
+ assertThat(old.toJson()).doesNotContain("operation");
+ }
+
public static SnapshotManager newSnapshotManager(FileIO fileIO, Path
tablePath) {
return newSnapshotManager(fileIO, tablePath, DEFAULT_MAIN_BRANCH);
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java
index b17c912aaf..ffdc12539a 100644
---
a/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java
@@ -993,7 +993,8 @@ public class DataEvolutionRowIdReassignerTest extends
TableTestBase {
latest.watermark(),
latest.statistics(),
latest.properties(),
- latest.nextRowId());
+ latest.nextRowId(),
+ latest.operation());
SnapshotManager snapshotManager = table.snapshotManager();
snapshotManager
.fileIO()
diff --git
a/paimon-core/src/test/java/org/apache/paimon/catalog/RenamingSnapshotCommitTest.java
b/paimon-core/src/test/java/org/apache/paimon/catalog/RenamingSnapshotCommitTest.java
index 6e8ae6f36a..0924b94203 100644
---
a/paimon-core/src/test/java/org/apache/paimon/catalog/RenamingSnapshotCommitTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/catalog/RenamingSnapshotCommitTest.java
@@ -137,6 +137,7 @@ public class RenamingSnapshotCommitTest {
null,
null,
null,
+ null,
null);
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/ExpireSnapshotsTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/ExpireSnapshotsTest.java
index 1da28a5bef..c4de0f44e1 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/ExpireSnapshotsTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/ExpireSnapshotsTest.java
@@ -420,6 +420,7 @@ public class ExpireSnapshotsTest {
null,
null,
null,
+ null,
null);
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/commit/ConflictDetectionTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/commit/ConflictDetectionTest.java
index c3371e1c19..0695406bde 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/commit/ConflictDetectionTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/commit/ConflictDetectionTest.java
@@ -740,6 +740,7 @@ class ConflictDetectionTest {
null,
null,
null,
+ null,
null);
}
}
diff --git
a/paimon-core/src/test/java/org/apache/paimon/table/system/SnapshotsTableTest.java
b/paimon-core/src/test/java/org/apache/paimon/table/system/SnapshotsTableTest.java
index 7a3bbf961c..c1a6b6f714 100644
---
a/paimon-core/src/test/java/org/apache/paimon/table/system/SnapshotsTableTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/table/system/SnapshotsTableTest.java
@@ -118,7 +118,10 @@ public class SnapshotsTableTest extends TableTestBase {
snapshot.deltaRecordCount(),
snapshot.changelogRecordCount(),
snapshot.watermark(),
- snapshot.nextRowId()));
+ snapshot.nextRowId(),
+ snapshot.operation() == null
+ ? null
+ :
BinaryString.fromString(snapshot.operation().toString())));
}
return expectedRow;
diff --git
a/paimon-core/src/test/java/org/apache/paimon/tag/TagAutoManagerTest.java
b/paimon-core/src/test/java/org/apache/paimon/tag/TagAutoManagerTest.java
index ef0e1627ee..d262110f03 100644
--- a/paimon-core/src/test/java/org/apache/paimon/tag/TagAutoManagerTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/tag/TagAutoManagerTest.java
@@ -410,6 +410,7 @@ public class TagAutoManagerTest extends
PrimaryKeyTableTestBase {
null,
null,
null,
+ null,
null);
tagManager.createTag(
snapshot1,
@@ -439,6 +440,7 @@ public class TagAutoManagerTest extends
PrimaryKeyTableTestBase {
null,
null,
null,
+ null,
null);
tagManager.createTag(
snapshot2,
diff --git a/paimon-core/src/test/java/org/apache/paimon/tag/TagTest.java
b/paimon-core/src/test/java/org/apache/paimon/tag/TagTest.java
index d7dbeacb7f..436e86c211 100644
--- a/paimon-core/src/test/java/org/apache/paimon/tag/TagTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/tag/TagTest.java
@@ -51,6 +51,7 @@ public class TagTest {
null,
null,
null,
+ null,
null);
@Test
diff --git
a/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java
b/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java
index 8e47b74aa3..02b21ba906 100644
--- a/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java
+++ b/paimon-core/src/test/java/org/apache/paimon/utils/SnapshotManagerTest.java
@@ -293,6 +293,7 @@ public class SnapshotManagerTest {
null,
null,
null,
+ null,
null);
}
@@ -317,6 +318,7 @@ public class SnapshotManagerTest {
watermark,
null,
null,
+ null,
null);
}
@@ -342,6 +344,7 @@ public class SnapshotManagerTest {
null,
null,
null,
+ null,
null));
}
@@ -373,6 +376,7 @@ public class SnapshotManagerTest {
null,
null,
null,
+ null,
null);
localFileIO.tryToWriteAtomic(snapshotManager.snapshotPath(i),
snapshot.toJson());
}
@@ -426,6 +430,7 @@ public class SnapshotManagerTest {
null,
null,
null,
+ null,
null);
localFileIO.tryToWriteAtomic(snapshotManager.snapshotPath(i),
snapshot.toJson());
}
diff --git
a/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/spark/sql/execution/shim/PaimonCreateTableAsSelectStrategy.scala
b/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/spark/sql/execution/shim/PaimonCreateTableAsSelectStrategy.scala
index 44095b7d2e..4aa8fb7840 100644
---
a/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/spark/sql/execution/shim/PaimonCreateTableAsSelectStrategy.scala
+++
b/paimon-spark/paimon-spark-3.2/src/main/scala/org/apache/spark/sql/execution/shim/PaimonCreateTableAsSelectStrategy.scala
@@ -18,8 +18,10 @@
package org.apache.spark.sql.execution.shim
+import org.apache.paimon.Snapshot
import org.apache.paimon.spark.SparkCatalog
import org.apache.paimon.spark.catalog.FormatTableCatalog
+import org.apache.paimon.spark.write.PaimonWriteOptions
import org.apache.spark.sql.{SparkSession, Strategy}
import org.apache.spark.sql.catalyst.plans.logical.{CreateTableAsSelect,
LogicalPlan}
@@ -66,7 +68,10 @@ case class PaimonCreateTableAsSelectStrategy(spark:
SparkSession) extends Strate
query,
planLater(query),
newProps,
- new CaseInsensitiveStringMap(writeOptions.asJava),
+ new CaseInsensitiveStringMap(
+ (writeOptions +
+ (PaimonWriteOptions.OPERATION_OPTION ->
Snapshot.Operation.CREATE_TABLE_AS_SELECT
+ .name())).asJava),
ifNotExists
) :: Nil
case _ => Nil
diff --git
a/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/spark/sql/execution/shim/PaimonCreateTableAsSelectStrategy.scala
b/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/spark/sql/execution/shim/PaimonCreateTableAsSelectStrategy.scala
index ca10cb259f..eb3e044459 100644
---
a/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/spark/sql/execution/shim/PaimonCreateTableAsSelectStrategy.scala
+++
b/paimon-spark/paimon-spark-3.3/src/main/scala/org/apache/spark/sql/execution/shim/PaimonCreateTableAsSelectStrategy.scala
@@ -18,8 +18,10 @@
package org.apache.spark.sql.execution.shim
+import org.apache.paimon.Snapshot
import org.apache.paimon.spark.SparkCatalog
import org.apache.paimon.spark.catalog.FormatTableCatalog
+import org.apache.paimon.spark.write.PaimonWriteOptions
import org.apache.spark.sql.{SparkSession, Strategy}
import org.apache.spark.sql.catalyst.analysis.ResolvedDBObjectName
@@ -69,7 +71,10 @@ case class PaimonCreateTableAsSelectStrategy(spark:
SparkSession)
query,
planLater(query),
qualifiedSpec,
- new CaseInsensitiveStringMap(writeOptions.asJava),
+ new CaseInsensitiveStringMap(
+ (writeOptions +
+ (PaimonWriteOptions.OPERATION_OPTION ->
Snapshot.Operation.CREATE_TABLE_AS_SELECT
+ .name())).asJava),
ifNotExists
) :: Nil
case _ => Nil
diff --git
a/paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/spark/sql/execution/shim/PaimonCreateTableAsSelectStrategy.scala
b/paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/spark/sql/execution/shim/PaimonCreateTableAsSelectStrategy.scala
index be377c17d1..0e0f3037d2 100644
---
a/paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/spark/sql/execution/shim/PaimonCreateTableAsSelectStrategy.scala
+++
b/paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/spark/sql/execution/shim/PaimonCreateTableAsSelectStrategy.scala
@@ -18,8 +18,10 @@
package org.apache.spark.sql.execution.shim
+import org.apache.paimon.Snapshot
import org.apache.paimon.spark.SparkCatalog
import org.apache.paimon.spark.catalog.FormatTableCatalog
+import org.apache.paimon.spark.write.PaimonWriteOptions
import org.apache.spark.sql.{SparkSession, Strategy}
import org.apache.spark.sql.catalyst.analysis.ResolvedIdentifier
@@ -71,7 +73,10 @@ case class PaimonCreateTableAsSelectStrategy(spark:
SparkSession)
analyzedQuery.get,
planLater(query),
qualifiedSpec,
- new CaseInsensitiveStringMap(writeOptions.asJava),
+ new CaseInsensitiveStringMap(
+ (writeOptions +
+ (PaimonWriteOptions.OPERATION_OPTION ->
Snapshot.Operation.CREATE_TABLE_AS_SELECT
+ .name())).asJava),
ifNotExists
) :: Nil
case _ => Nil
diff --git
a/paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/spark/sql/execution/shim/PaimonReplaceTableAsSelectStrategy.scala
b/paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/spark/sql/execution/shim/PaimonReplaceTableAsSelectStrategy.scala
index ec4c0498e1..ee9064dcc3 100644
---
a/paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/spark/sql/execution/shim/PaimonReplaceTableAsSelectStrategy.scala
+++
b/paimon-spark/paimon-spark-3.4/src/main/scala/org/apache/spark/sql/execution/shim/PaimonReplaceTableAsSelectStrategy.scala
@@ -18,7 +18,9 @@
package org.apache.spark.sql.execution.shim
+import org.apache.paimon.Snapshot
import org.apache.paimon.spark.catalog.SparkBaseCatalog
+import org.apache.paimon.spark.write.PaimonWriteOptions
import org.apache.spark.sql.{SparkSession, Strategy}
import org.apache.spark.sql.catalyst.analysis.ResolvedIdentifier
@@ -62,7 +64,11 @@ case class PaimonReplaceTableAsSelectStrategy(spark:
SparkSession)
val (tableOptions, writeOptions) =
splitTableAndWriteOptions(options)
val qualifiedSpec = qualifyTableSpec(tableSpec, tableOptions)
- val writeOpts = new CaseInsensitiveStringMap(writeOptions.asJava)
+ val operation =
+ if (orCreate) Snapshot.Operation.CREATE_OR_REPLACE_TABLE_AS_SELECT
+ else Snapshot.Operation.REPLACE_TABLE_AS_SELECT
+ val writeOpts = new CaseInsensitiveStringMap(
+ (writeOptions + (PaimonWriteOptions.OPERATION_OPTION ->
operation.name())).asJava)
val pinnedQuery =
pinSnapshotInQuery(catalog, ident, analyzedQuery.get)
if (canAtomicReplace(catalog, ident, qualifiedSpec, parts)) {
diff --git
a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
index 4e5080c0d5..c14ee04648 100644
---
a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
+++
b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
@@ -19,6 +19,7 @@
package org.apache.paimon.spark.commands
import org.apache.paimon.CoreOptions.GlobalIndexColumnUpdateAction
+import org.apache.paimon.Snapshot
import org.apache.paimon.data.BinaryRow
import org.apache.paimon.format.blob.BlobFileFormat.isBlobFile
import org.apache.paimon.index.GlobalIndexMeta
@@ -236,7 +237,7 @@ case class MergeIntoPaimonDataEvolutionTable(
if (plan.snapshotId() != null) {
writer.rowIdCheckConflict(plan.snapshotId())
}
- writer.commit(updateCommit ++ insertCommit)
+ writer.commit(updateCommit ++ insertCommit, Snapshot.Operation.MERGE)
} finally {
if (persistSourceDss.isDefined) {
persistSourceDss.get.unpersist(blocking = false)
diff --git
a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonTable.scala
b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonTable.scala
index 89689e108c..c24d4d6675 100644
---
a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonTable.scala
+++
b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonTable.scala
@@ -18,6 +18,7 @@
package org.apache.paimon.spark.commands
+import org.apache.paimon.Snapshot
import org.apache.paimon.spark.SparkTable
import org.apache.paimon.spark.catalyst.analysis.{PaimonRelation,
PaimonUpdateAction}
import org.apache.paimon.spark.schema.{PaimonMetadataColumn,
SparkSystemColumns}
@@ -78,7 +79,7 @@ case class MergeIntoPaimonTable(
} else {
performMergeForNonPkTable(sparkSession)
}
- writer.commit(commitMessages)
+ writer.commit(commitMessages, Snapshot.Operation.MERGE)
Seq.empty[Row]
}
diff --git
a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/write/PaimonBatchWrite.scala
b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/write/PaimonBatchWrite.scala
index c78f783924..6a5a141f08 100644
---
a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/write/PaimonBatchWrite.scala
+++
b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/paimon/spark/write/PaimonBatchWrite.scala
@@ -18,6 +18,7 @@
package org.apache.paimon.spark.write
+import org.apache.paimon.Snapshot
import org.apache.paimon.spark.rowops.PaimonCopyOnWriteScan
import org.apache.paimon.table.FileStoreTable
@@ -36,8 +37,15 @@ class PaimonBatchWrite(
writeSchema: StructType,
dataSchema: StructType,
overwritePartitions: Option[Map[String, String]],
- copyOnWriteScan: Option[PaimonCopyOnWriteScan])
- extends PaimonBatchWriteBase(table, writeSchema, dataSchema,
overwritePartitions, copyOnWriteScan)
+ copyOnWriteScan: Option[PaimonCopyOnWriteScan],
+ operationType: Option[Snapshot.Operation] = None)
+ extends PaimonBatchWriteBase(
+ table,
+ writeSchema,
+ dataSchema,
+ overwritePartitions,
+ copyOnWriteScan,
+ operationType)
with BatchWrite
with Serializable {
@@ -57,6 +65,13 @@ object PaimonBatchWrite {
writeSchema: StructType,
dataSchema: StructType,
overwritePartitions: Option[Map[String, String]],
- copyOnWriteScan: Option[PaimonCopyOnWriteScan]): PaimonBatchWrite =
- new PaimonBatchWrite(table, writeSchema, dataSchema, overwritePartitions,
copyOnWriteScan)
+ copyOnWriteScan: Option[PaimonCopyOnWriteScan],
+ operationType: Option[Snapshot.Operation] = None): PaimonBatchWrite =
+ new PaimonBatchWrite(
+ table,
+ writeSchema,
+ dataSchema,
+ overwritePartitions,
+ copyOnWriteScan,
+ operationType)
}
diff --git
a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
index 9e97adc96e..b92411d8bd 100644
---
a/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
+++
b/paimon-spark/paimon-spark-4.0/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
@@ -18,6 +18,7 @@
package org.apache.spark.sql.paimon.shims
+import org.apache.paimon.Snapshot
import org.apache.paimon.data.variant.{GenericVariant, Variant}
import org.apache.paimon.spark.catalyst.analysis.Spark4ResolutionRules
import
org.apache.paimon.spark.catalyst.parser.extensions.PaimonSpark4SqlExtensionsParser
@@ -209,8 +210,15 @@ class Spark4Shim extends SparkShim {
writeSchema: StructType,
dataSchema: StructType,
overwritePartitions: Option[Map[String, String]],
- copyOnWriteScan: Option[PaimonCopyOnWriteScan]): BatchWrite =
- new PaimonBatchWrite(table, writeSchema, dataSchema, overwritePartitions,
copyOnWriteScan)
+ copyOnWriteScan: Option[PaimonCopyOnWriteScan],
+ operationType: Option[Snapshot.Operation]): BatchWrite =
+ new PaimonBatchWrite(
+ table,
+ writeSchema,
+ dataSchema,
+ overwritePartitions,
+ copyOnWriteScan,
+ operationType)
override def createFormatTableBatchWrite(
table: FormatTable,
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DeleteFromPaimonTableCommand.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DeleteFromPaimonTableCommand.scala
index d052d35471..cd29efa152 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DeleteFromPaimonTableCommand.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/DeleteFromPaimonTableCommand.scala
@@ -18,6 +18,7 @@
package org.apache.paimon.spark.commands
+import org.apache.paimon.Snapshot
import org.apache.paimon.spark.catalyst.analysis.expressions.ExpressionHelper
import org.apache.paimon.spark.schema.SparkSystemColumns.ROW_KIND_COL
import org.apache.paimon.table.FileStoreTable
@@ -46,7 +47,7 @@ case class DeleteFromPaimonTableCommand(
} else {
performNonPrimaryKeyDelete(sparkSession)
}
- writer.commit(commitMessages)
+ writer.commit(commitMessages, Snapshot.Operation.DELETE)
Seq.empty[Row]
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
index 321eb6f147..7464176c87 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonDataEvolutionTable.scala
@@ -19,6 +19,7 @@
package org.apache.paimon.spark.commands
import org.apache.paimon.CoreOptions.GlobalIndexColumnUpdateAction
+import org.apache.paimon.Snapshot
import org.apache.paimon.data.BinaryRow
import org.apache.paimon.format.blob.BlobFileFormat.isBlobFile
import org.apache.paimon.index.GlobalIndexMeta
@@ -235,7 +236,7 @@ case class MergeIntoPaimonDataEvolutionTable(
if (plan.snapshotId() != null) {
writer.rowIdCheckConflict(plan.snapshotId())
}
- writer.commit(updateCommit ++ insertCommit)
+ writer.commit(updateCommit ++ insertCommit, Snapshot.Operation.MERGE)
} finally {
if (persistSourceDss.isDefined) {
persistSourceDss.get.unpersist(blocking = false)
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonTable.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonTable.scala
index 89689e108c..c24d4d6675 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonTable.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/MergeIntoPaimonTable.scala
@@ -18,6 +18,7 @@
package org.apache.paimon.spark.commands
+import org.apache.paimon.Snapshot
import org.apache.paimon.spark.SparkTable
import org.apache.paimon.spark.catalyst.analysis.{PaimonRelation,
PaimonUpdateAction}
import org.apache.paimon.spark.schema.{PaimonMetadataColumn,
SparkSystemColumns}
@@ -78,7 +79,7 @@ case class MergeIntoPaimonTable(
} else {
performMergeForNonPkTable(sparkSession)
}
- writer.commit(commitMessages)
+ writer.commit(commitMessages, Snapshot.Operation.MERGE)
Seq.empty[Row]
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
index 8e310dc9e0..92df5a7f06 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/PaimonSparkWriter.scala
@@ -414,6 +414,10 @@ case class PaimonSparkWriter(
}
def commit(commitMessages: Seq[CommitMessage]): Unit = {
+ commit(commitMessages, null)
+ }
+
+ def commit(commitMessages: Seq[CommitMessage], operation:
Snapshot.Operation): Unit = {
val finalWriteBuilder = if (postponeBatchWriteFixedBucket) {
writeBuilder
.asInstanceOf[BatchWriteBuilderImpl]
@@ -424,6 +428,9 @@ case class PaimonSparkWriter(
writeBuilder
}
val tableCommit = finalWriteBuilder.newCommit()
+ if (operation != null) {
+ tableCommit.withOperation(operation)
+ }
try {
tableCommit.commit(commitMessages.toList.asJava)
} catch {
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/UpdatePaimonTableCommand.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/UpdatePaimonTableCommand.scala
index 1feb0f1f76..1f1622fe2b 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/UpdatePaimonTableCommand.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/UpdatePaimonTableCommand.scala
@@ -18,6 +18,7 @@
package org.apache.paimon.spark.commands
+import org.apache.paimon.Snapshot
import org.apache.paimon.spark.schema.PaimonMetadataColumn.{ROW_ID_COLUMN,
SEQUENCE_NUMBER_COLUMN}
import org.apache.paimon.spark.schema.SparkSystemColumns.ROW_KIND_COL
import org.apache.paimon.table.FileStoreTable
@@ -52,7 +53,7 @@ case class UpdatePaimonTableCommand(
} else {
performUpdateForNonPkTable(sparkSession)
}
- writer.commit(commitMessages)
+ writer.commit(commitMessages, Snapshot.Operation.UPDATE)
Seq.empty[Row]
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/WriteIntoPaimonTable.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/WriteIntoPaimonTable.scala
index fcf061ec67..f400409fda 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/WriteIntoPaimonTable.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/commands/WriteIntoPaimonTable.scala
@@ -19,10 +19,12 @@
package org.apache.paimon.spark.commands
import org.apache.paimon.CoreOptions.DYNAMIC_PARTITION_OVERWRITE
+import org.apache.paimon.Snapshot
import org.apache.paimon.options.Options
import org.apache.paimon.spark._
import org.apache.paimon.spark.catalyst.analysis.ReplacePaimonFunctions
import org.apache.paimon.spark.catalyst.analysis.expressions.ExpressionHelper
+import org.apache.paimon.spark.write.PaimonWriteOptions
import org.apache.paimon.table.FileStoreTable
import org.apache.spark.internal.Logging
@@ -60,8 +62,17 @@ case class WriteIntoPaimonTable(
if (overwritePartition != null) {
writer.writeBuilder.withOverwrite(overwritePartition.asJava)
}
+ val operation = Option(options.get(PaimonWriteOptions.OPERATION_OPTION))
+ .map(Snapshot.Operation.valueOf)
+ .getOrElse {
+ if (overwritePartition != null || dynamicPartitionOverwriteMode) {
+ Snapshot.Operation.OVERWRITE
+ } else {
+ Snapshot.Operation.WRITE
+ }
+ }
val commitMessages = writer.write(replacedData)
- writer.commit(commitMessages)
+ writer.commit(commitMessages, operation)
Seq.empty
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/rowops/PaimonSparkCopyOnWriteOperation.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/rowops/PaimonSparkCopyOnWriteOperation.scala
index 24c3c19761..5db630bec7 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/rowops/PaimonSparkCopyOnWriteOperation.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/rowops/PaimonSparkCopyOnWriteOperation.scala
@@ -18,6 +18,7 @@
package org.apache.paimon.spark.rowops
+import org.apache.paimon.Snapshot
import org.apache.paimon.options.Options
import org.apache.paimon.spark.PaimonBaseScanBuilder
import org.apache.paimon.spark.schema.PaimonMetadataColumn.{FILE_PATH_COLUMN,
ROW_ID_COLUMN, SEQUENCE_NUMBER_COLUMN}
@@ -52,6 +53,7 @@ class PaimonSparkCopyOnWriteOperation(table: FileStoreTable,
info: RowLevelOpera
override def newWriteBuilder(info: LogicalWriteInfo): WriteBuilder = {
val options = Options.fromMap(info.options)
val builder = new PaimonV2WriteBuilder(table, info.schema(), options)
+ builder.withOperationType(Snapshot.Operation.valueOf(command().toString))
assert(copyOnWriteScan.isDefined)
builder.overwriteFiles(copyOnWriteScan.get)
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonBatchWriteBase.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonBatchWriteBase.scala
index 42d2ebcd85..e925cad76b 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonBatchWriteBase.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonBatchWriteBase.scala
@@ -18,6 +18,7 @@
package org.apache.paimon.spark.write
+import org.apache.paimon.Snapshot
import org.apache.paimon.io.{CompactIncrement, DataFileMeta, DataIncrement}
import org.apache.paimon.spark.SparkTypeUtils
import org.apache.paimon.spark.catalyst.Compatibility
@@ -55,7 +56,8 @@ abstract class PaimonBatchWriteBase(
val writeSchema: StructType,
val dataSchema: StructType,
val overwritePartitions: Option[Map[String, String]],
- val copyOnWriteScan: Option[PaimonCopyOnWriteScan])
+ val copyOnWriteScan: Option[PaimonCopyOnWriteScan],
+ operationType: Option[Snapshot.Operation] = None)
extends WriteHelper
with Serializable {
@@ -114,6 +116,9 @@ abstract class PaimonBatchWriteBase(
logInfo(s"Committing to table ${table.name()}")
val batchTableCommit = batchWriteBuilder.newCommit()
batchTableCommit.withMetricRegistry(metricRegistry)
+ val operation = operationType.getOrElse(
+ if (overwritePartitions.isDefined) Snapshot.Operation.OVERWRITE else
Snapshot.Operation.WRITE)
+ batchTableCommit.withOperation(operation)
val addCommitMessage = WriteTaskResult.merge(messages)
val deletedCommitMessage = copyOnWriteScan match {
case Some(scan) => buildDeletedCommitMessage(scan.scannedFiles)
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2Write.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2Write.scala
index 2ae1dd53a3..6dd02172a6 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2Write.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2Write.scala
@@ -19,6 +19,7 @@
package org.apache.paimon.spark.write
import org.apache.paimon.CoreOptions.ChangelogProducer
+import org.apache.paimon.Snapshot
import org.apache.paimon.options.Options
import org.apache.paimon.spark._
import org.apache.paimon.spark.commands.SchemaEvolutionHelper
@@ -41,7 +42,8 @@ class PaimonV2Write(
overwritePartitions: Option[Map[String, String]],
copyOnWriteScan: Option[PaimonCopyOnWriteScan],
dataSchema: StructType,
- options: Options
+ options: Options,
+ operationType: Option[Snapshot.Operation] = None
) extends Write
with RequiresDistributionAndOrdering
with SchemaEvolutionHelper
@@ -69,7 +71,8 @@ class PaimonV2Write(
writeSchema,
dataSchema,
overwritePartitions,
- copyOnWriteScan)
+ copyOnWriteScan,
+ operationType)
}
override def supportedCustomMetrics(): Array[CustomMetric] = {
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2WriteBuilder.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2WriteBuilder.scala
index 91f4f861ce..f57c4e99aa 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2WriteBuilder.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2WriteBuilder.scala
@@ -19,6 +19,7 @@
package org.apache.paimon.spark.write
import org.apache.paimon.CoreOptions
+import org.apache.paimon.Snapshot
import org.apache.paimon.options.Options
import org.apache.paimon.table.FileStoreTable
import org.apache.paimon.types.RowType
@@ -30,13 +31,27 @@ import scala.collection.JavaConverters._
class PaimonV2WriteBuilder(table: FileStoreTable, dataSchema: StructType,
options: Options)
extends BaseV2WriteBuilder(table) {
+ private var _operationType: Option[Snapshot.Operation] =
+
Option(options.get(PaimonWriteOptions.OPERATION_OPTION)).map(Snapshot.Operation.valueOf)
+
+ def withOperationType(operationType: Snapshot.Operation):
PaimonV2WriteBuilder = {
+ _operationType = Option(operationType)
+ this
+ }
+
override def build: PaimonV2Write = {
val finalTable = overwriteDynamic match {
case Some(o) =>
table.copy(Map(CoreOptions.DYNAMIC_PARTITION_OVERWRITE.key ->
o.toString).asJava)
case _ => table
}
- new PaimonV2Write(finalTable, overwritePartitions, copyOnWriteScan,
dataSchema, options)
+ new PaimonV2Write(
+ finalTable,
+ overwritePartitions,
+ copyOnWriteScan,
+ dataSchema,
+ options,
+ _operationType)
}
override def partitionRowType(): RowType =
table.schema().logicalPartitionType()
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2WriteBuilder.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonWriteOptions.scala
similarity index 51%
copy from
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2WriteBuilder.scala
copy to
paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonWriteOptions.scala
index 91f4f861ce..16633cad37 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonV2WriteBuilder.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/paimon/spark/write/PaimonWriteOptions.scala
@@ -18,26 +18,9 @@
package org.apache.paimon.spark.write
-import org.apache.paimon.CoreOptions
-import org.apache.paimon.options.Options
-import org.apache.paimon.table.FileStoreTable
-import org.apache.paimon.types.RowType
+/** Internal Spark write-option keys. */
+object PaimonWriteOptions {
-import org.apache.spark.sql.types.StructType
-
-import scala.collection.JavaConverters._
-
-class PaimonV2WriteBuilder(table: FileStoreTable, dataSchema: StructType,
options: Options)
- extends BaseV2WriteBuilder(table) {
-
- override def build: PaimonV2Write = {
- val finalTable = overwriteDynamic match {
- case Some(o) =>
- table.copy(Map(CoreOptions.DYNAMIC_PARTITION_OVERWRITE.key ->
o.toString).asJava)
- case _ => table
- }
- new PaimonV2Write(finalTable, overwritePartitions, copyOnWriteScan,
dataSchema, options)
- }
-
- override def partitionRowType(): RowType =
table.schema().logicalPartitionType()
+ /** Carries the snapshot operation type through Spark's CTAS/RTAS
write-option map. */
+ val OPERATION_OPTION: String = "__paimon.operation"
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/execution/shim/PaimonCreateTableAsSelectStrategy.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/execution/shim/PaimonCreateTableAsSelectStrategy.scala
index fbfb6e4eca..1a8e3ffe4b 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/execution/shim/PaimonCreateTableAsSelectStrategy.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/execution/shim/PaimonCreateTableAsSelectStrategy.scala
@@ -18,8 +18,10 @@
package org.apache.spark.sql.execution.shim
+import org.apache.paimon.Snapshot
import org.apache.paimon.spark.SparkCatalog
import org.apache.paimon.spark.catalog.FormatTableCatalog
+import org.apache.paimon.spark.write.PaimonWriteOptions
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.analysis.ResolvedIdentifier
@@ -66,8 +68,10 @@ case class PaimonCreateTableAsSelectStrategy(spark:
SparkSession)
parts,
query,
qualifiedSpec,
- writeOptions,
- ifNotExists) :: Nil
+ writeOptions +
+ (PaimonWriteOptions.OPERATION_OPTION ->
Snapshot.Operation.CREATE_TABLE_AS_SELECT.name()),
+ ifNotExists
+ ) :: Nil
case _ => Nil
}
}
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/execution/shim/PaimonReplaceTableAsSelectStrategy.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/execution/shim/PaimonReplaceTableAsSelectStrategy.scala
index 4c21a448e9..4bbcc0af25 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/execution/shim/PaimonReplaceTableAsSelectStrategy.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/execution/shim/PaimonReplaceTableAsSelectStrategy.scala
@@ -18,7 +18,9 @@
package org.apache.spark.sql.execution.shim
+import org.apache.paimon.Snapshot
import org.apache.paimon.spark.catalog.SparkBaseCatalog
+import org.apache.paimon.spark.write.PaimonWriteOptions
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.catalyst.analysis.ResolvedIdentifier
@@ -53,6 +55,11 @@ case class PaimonReplaceTableAsSelectStrategy(spark:
SparkSession)
val (tableOptions, writeOptions) = splitTableAndWriteOptions(options)
val qualifiedSpec = qualifyTableSpec(tableSpec, tableOptions)
+ val operation =
+ if (orCreate) Snapshot.Operation.CREATE_OR_REPLACE_TABLE_AS_SELECT
+ else Snapshot.Operation.REPLACE_TABLE_AS_SELECT
+ val finalWriteOptions =
+ writeOptions + (PaimonWriteOptions.OPERATION_OPTION ->
operation.name())
// Pin snapshot in query to prevent self-referencing RTAS from reading
truncated data
val pinnedQuery = pinSnapshotInQuery(catalog, ident, query)
if (canAtomicReplace(catalog, ident, qualifiedSpec, parts)) {
@@ -62,7 +69,7 @@ case class PaimonReplaceTableAsSelectStrategy(spark:
SparkSession)
parts,
pinnedQuery,
qualifiedSpec,
- writeOptions,
+ finalWriteOptions,
orCreate = orCreate) :: Nil
} else {
SparkShimLoader.shim.createReplaceTableAsSelectExec(
@@ -71,7 +78,7 @@ case class PaimonReplaceTableAsSelectStrategy(spark:
SparkSession)
parts,
pinnedQuery,
qualifiedSpec,
- writeOptions,
+ finalWriteOptions,
orCreate = orCreate) :: Nil
}
case _ => Nil
diff --git
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/paimon/shims/SparkShim.scala
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/paimon/shims/SparkShim.scala
index 5c8094682f..b925036463 100644
---
a/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/paimon/shims/SparkShim.scala
+++
b/paimon-spark/paimon-spark-common/src/main/scala/org/apache/spark/sql/paimon/shims/SparkShim.scala
@@ -18,6 +18,7 @@
package org.apache.spark.sql.paimon.shims
+import org.apache.paimon.Snapshot
import org.apache.paimon.data.variant.Variant
import org.apache.paimon.function.{Function => PaimonFunction}
import org.apache.paimon.spark.data.{SparkArrayData, SparkInternalRow}
@@ -129,7 +130,8 @@ trait SparkShim {
writeSchema: StructType,
dataSchema: StructType,
overwritePartitions: Option[Map[String, String]],
- copyOnWriteScan: Option[PaimonCopyOnWriteScan]): BatchWrite
+ copyOnWriteScan: Option[PaimonCopyOnWriteScan],
+ operationType: Option[Snapshot.Operation]): BatchWrite
/** Same `BatchWrite` mixin problem as [[createPaimonBatchWrite]], but for
`FormatTable` writes. */
def createFormatTableBatchWrite(
diff --git
a/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/SnapshotOperationTest.scala
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/SnapshotOperationTest.scala
new file mode 100644
index 0000000000..fcc4d06650
--- /dev/null
+++
b/paimon-spark/paimon-spark-ut/src/test/scala/org/apache/paimon/spark/sql/SnapshotOperationTest.scala
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.spark.sql
+
+import org.apache.paimon.Snapshot
+import org.apache.paimon.spark.PaimonSparkTestBase
+
+/** Verifies the logical operation type recorded in the committed snapshot. */
+class SnapshotOperationTest extends PaimonSparkTestBase {
+
+ private def latestOperation(tableName: String): Snapshot.Operation = {
+ val snapshot = loadTable(tableName).snapshotManager().latestSnapshot()
+ assert(snapshot != null, s"table $tableName has no snapshot")
+ snapshot.operation()
+ }
+
+ test("Snapshot operation: INSERT / OVERWRITE / UPDATE / DELETE / MERGE") {
+ for (useV2Write <- Seq("true", "false")) {
+ withSparkSQLConf("spark.paimon.write.use-v2-write" -> useV2Write) {
+ withTable("t", "s") {
+ sql("CREATE TABLE t (id INT, name STRING)")
+ sql("CREATE TABLE s (id INT, name STRING)")
+ sql("INSERT INTO s VALUES (1, 'merged'), (9, 'new')")
+ sql("INSERT INTO t VALUES (1, 'a'), (2, 'b'), (3, 'c')")
+ assert(latestOperation("t") == Snapshot.Operation.WRITE)
+
+ sql("INSERT OVERWRITE t VALUES (1, 'a2'), (2, 'b2'), (3, 'c2')")
+ assert(latestOperation("t") == Snapshot.Operation.OVERWRITE)
+
+ sql("UPDATE t SET name = 'updated' WHERE id = 1")
+ assert(latestOperation("t") == Snapshot.Operation.UPDATE)
+
+ sql("DELETE FROM t WHERE id = 2")
+ assert(latestOperation("t") == Snapshot.Operation.DELETE)
+
+ sql("""
+ |MERGE INTO t USING s ON t.id = s.id
+ |WHEN MATCHED THEN UPDATE SET t.name = s.name
+ |WHEN NOT MATCHED THEN INSERT *
+ |""".stripMargin)
+ assert(latestOperation("t") == Snapshot.Operation.MERGE)
+ }
+ }
+ }
+ }
+
+ test("Snapshot operation: CTAS / RTAS") {
+ for (useV2Write <- Seq("true", "false")) {
+ withSparkSQLConf("spark.paimon.write.use-v2-write" -> useV2Write) {
+ withTable("src", "ctas", "rtas") {
+ sql("CREATE TABLE src (id INT, name STRING)")
+ sql("INSERT INTO src VALUES (1, 'a'), (2, 'b')")
+
+ sql("CREATE TABLE ctas AS SELECT * FROM src")
+ assert(latestOperation("ctas") ==
Snapshot.Operation.CREATE_TABLE_AS_SELECT)
+
+ sql("CREATE OR REPLACE TABLE rtas AS SELECT * FROM src")
+ assert(latestOperation("rtas") ==
Snapshot.Operation.CREATE_OR_REPLACE_TABLE_AS_SELECT)
+
+ // rtas twice
+ sql("CREATE OR REPLACE TABLE rtas AS SELECT * FROM src")
+ assert(latestOperation("rtas") ==
Snapshot.Operation.CREATE_OR_REPLACE_TABLE_AS_SELECT)
+
+ // pure REPLACE TABLE (no CREATE)
+ sql("REPLACE TABLE rtas AS SELECT * FROM src")
+ assert(latestOperation("rtas") ==
Snapshot.Operation.REPLACE_TABLE_AS_SELECT)
+ }
+ }
+ }
+ }
+
+ test("Snapshot operation: TRUNCATE") {
+ withTable("t") {
+ sql("CREATE TABLE t (id INT, name STRING, dt STRING) PARTITIONED BY
(dt)")
+ sql("INSERT INTO t VALUES (1, 'a', '2024-01-01'), (2, 'b',
'2024-01-02')")
+
+ // Full-table DELETE (no WHERE) is optimized to truncateTable
+ sql("DELETE FROM t")
+ assert(latestOperation("t") == Snapshot.Operation.TRUNCATE)
+
+ sql("INSERT INTO t VALUES (3, 'c', '2024-01-01'), (4, 'd',
'2024-01-02')")
+
+ // Partition DELETE is optimized to truncatePartitions
+ sql("DELETE FROM t WHERE dt = '2024-01-01'")
+ assert(latestOperation("t") == Snapshot.Operation.TRUNCATE)
+
+ sql("INSERT INTO t VALUES (5, 'e', '2024-01-01')")
+
+ // TRUNCATE TABLE
+ sql("TRUNCATE TABLE t")
+ assert(latestOperation("t") == Snapshot.Operation.TRUNCATE)
+ }
+ }
+}
diff --git
a/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/paimon/spark/write/PaimonBatchWrite.scala
b/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/paimon/spark/write/PaimonBatchWrite.scala
index 89eb15054c..57b89270d8 100644
---
a/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/paimon/spark/write/PaimonBatchWrite.scala
+++
b/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/paimon/spark/write/PaimonBatchWrite.scala
@@ -18,6 +18,7 @@
package org.apache.paimon.spark.write
+import org.apache.paimon.Snapshot
import org.apache.paimon.spark.rowops.PaimonCopyOnWriteScan
import org.apache.paimon.table.FileStoreTable
@@ -33,8 +34,15 @@ class PaimonBatchWrite(
writeSchema: StructType,
dataSchema: StructType,
overwritePartitions: Option[Map[String, String]],
- copyOnWriteScan: Option[PaimonCopyOnWriteScan])
- extends PaimonBatchWriteBase(table, writeSchema, dataSchema,
overwritePartitions, copyOnWriteScan)
+ copyOnWriteScan: Option[PaimonCopyOnWriteScan],
+ operationType: Option[Snapshot.Operation] = None)
+ extends PaimonBatchWriteBase(
+ table,
+ writeSchema,
+ dataSchema,
+ overwritePartitions,
+ copyOnWriteScan,
+ operationType)
with BatchWrite
with Serializable {
@@ -54,6 +62,13 @@ object PaimonBatchWrite {
writeSchema: StructType,
dataSchema: StructType,
overwritePartitions: Option[Map[String, String]],
- copyOnWriteScan: Option[PaimonCopyOnWriteScan]): PaimonBatchWrite =
- new PaimonBatchWrite(table, writeSchema, dataSchema, overwritePartitions,
copyOnWriteScan)
+ copyOnWriteScan: Option[PaimonCopyOnWriteScan],
+ operationType: Option[Snapshot.Operation] = None): PaimonBatchWrite =
+ new PaimonBatchWrite(
+ table,
+ writeSchema,
+ dataSchema,
+ overwritePartitions,
+ copyOnWriteScan,
+ operationType)
}
diff --git
a/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark3Shim.scala
b/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark3Shim.scala
index 362e2bebf0..d4993fdce4 100644
---
a/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark3Shim.scala
+++
b/paimon-spark/paimon-spark3-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark3Shim.scala
@@ -18,6 +18,7 @@
package org.apache.spark.sql.paimon.shims
+import org.apache.paimon.Snapshot
import org.apache.paimon.data.variant.Variant
import org.apache.paimon.spark.catalyst.analysis.Spark3ResolutionRules
import
org.apache.paimon.spark.catalyst.parser.extensions.PaimonSpark3SqlExtensionsParser
@@ -195,8 +196,15 @@ class Spark3Shim extends SparkShim {
writeSchema: StructType,
dataSchema: StructType,
overwritePartitions: Option[Map[String, String]],
- copyOnWriteScan: Option[PaimonCopyOnWriteScan]): BatchWrite =
- new PaimonBatchWrite(table, writeSchema, dataSchema, overwritePartitions,
copyOnWriteScan)
+ copyOnWriteScan: Option[PaimonCopyOnWriteScan],
+ operationType: Option[Snapshot.Operation]): BatchWrite =
+ new PaimonBatchWrite(
+ table,
+ writeSchema,
+ dataSchema,
+ overwritePartitions,
+ copyOnWriteScan,
+ operationType)
override def createFormatTableBatchWrite(
table: FormatTable,
diff --git
a/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/paimon/spark/write/PaimonBatchWrite.scala
b/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/paimon/spark/write/PaimonBatchWrite.scala
index 38683e79c0..374062c6b4 100644
---
a/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/paimon/spark/write/PaimonBatchWrite.scala
+++
b/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/paimon/spark/write/PaimonBatchWrite.scala
@@ -18,6 +18,7 @@
package org.apache.paimon.spark.write
+import org.apache.paimon.Snapshot
import org.apache.paimon.spark.rowops.PaimonCopyOnWriteScan
import org.apache.paimon.table.FileStoreTable
@@ -35,8 +36,15 @@ class PaimonBatchWrite(
writeSchema: StructType,
dataSchema: StructType,
overwritePartitions: Option[Map[String, String]],
- copyOnWriteScan: Option[PaimonCopyOnWriteScan])
- extends PaimonBatchWriteBase(table, writeSchema, dataSchema,
overwritePartitions, copyOnWriteScan)
+ copyOnWriteScan: Option[PaimonCopyOnWriteScan],
+ operationType: Option[Snapshot.Operation] = None)
+ extends PaimonBatchWriteBase(
+ table,
+ writeSchema,
+ dataSchema,
+ overwritePartitions,
+ copyOnWriteScan,
+ operationType)
with BatchWrite
with Serializable {
@@ -56,6 +64,13 @@ object PaimonBatchWrite {
writeSchema: StructType,
dataSchema: StructType,
overwritePartitions: Option[Map[String, String]],
- copyOnWriteScan: Option[PaimonCopyOnWriteScan]): PaimonBatchWrite =
- new PaimonBatchWrite(table, writeSchema, dataSchema, overwritePartitions,
copyOnWriteScan)
+ copyOnWriteScan: Option[PaimonCopyOnWriteScan],
+ operationType: Option[Snapshot.Operation] = None): PaimonBatchWrite =
+ new PaimonBatchWrite(
+ table,
+ writeSchema,
+ dataSchema,
+ overwritePartitions,
+ copyOnWriteScan,
+ operationType)
}
diff --git
a/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
b/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
index 3a49c223b6..00d6024362 100644
---
a/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
+++
b/paimon-spark/paimon-spark4-common/src/main/scala/org/apache/spark/sql/paimon/shims/Spark4Shim.scala
@@ -18,6 +18,7 @@
package org.apache.spark.sql.paimon.shims
+import org.apache.paimon.Snapshot
import org.apache.paimon.data.variant.{GenericVariant, Variant}
import org.apache.paimon.spark.catalyst.analysis.Spark4ResolutionRules
import
org.apache.paimon.spark.catalyst.parser.extensions.PaimonSpark4SqlExtensionsParser
@@ -193,8 +194,15 @@ class Spark4Shim extends SparkShim {
writeSchema: StructType,
dataSchema: StructType,
overwritePartitions: Option[Map[String, String]],
- copyOnWriteScan: Option[PaimonCopyOnWriteScan]): BatchWrite =
- new PaimonBatchWrite(table, writeSchema, dataSchema, overwritePartitions,
copyOnWriteScan)
+ copyOnWriteScan: Option[PaimonCopyOnWriteScan],
+ operationType: Option[Snapshot.Operation]): BatchWrite =
+ new PaimonBatchWrite(
+ table,
+ writeSchema,
+ dataSchema,
+ overwritePartitions,
+ copyOnWriteScan,
+ operationType)
override def createFormatTableBatchWrite(
table: FormatTable,
