[spark] Record the write operation type in snapshot properties

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.

docs/docs/pypaimon/system-tables.md

@@ -91,6 +91,7 @@ One row per persisted snapshot.
 | `changelog_record_count`  | BIGINT          |                                |
 | `watermark`               | BIGINT          |                                |
 | `next_row_id`             | BIGINT          |                                |
+| `operation`               | STRING          | `WRITE`, `DELETE`, `MERGE`, ...|
 
 ### `$schemas`
 

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,
@@ -201,7 +207,8 @@ public Snapshot(
             @Nullable Long watermark,
             @Nullable String statistics,
             @Nullable Map<String, String> properties,
-            @Nullable Long nextRowId) {
+            @Nullable Long nextRowId,
+            @Nullable Operation operation) {
         this(
                 CURRENT_VERSION,
                 id,
@@ -223,7 +230,8 @@ public Snapshot(
                 watermark,
                 statistics,
                 properties,
-                nextRowId);
+                nextRowId,
+                operation);
     }
 
     @JsonCreator
@@ -249,7 +257,8 @@ public Snapshot(
             @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 +280,7 @@ public Snapshot(
         this.statistics = statistics;
         this.properties = properties;
         this.nextRowId = nextRowId;
+        this.operation = operation;
     }
 
     @JsonGetter(FIELD_VERSION)
@@ -388,6 +398,12 @@ public Long nextRowId() {
         return nextRowId;
     }
 
+    @JsonGetter(FIELD_OPERATION)
+    @Nullable
+    public Operation operation() {
+        return operation;
+    }
+
     public String toJson() {
         return JsonSerdeUtil.toJson(this);
     }
@@ -415,7 +431,8 @@ public int hashCode() {
                 watermark,
                 statistics,
                 properties,
-                nextRowId);
+                nextRowId,
+                operation);
     }
 
     @Override
@@ -447,7 +464,8 @@ public boolean equals(Object o) {
                 && 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 +487,19 @@ public enum CommitKind {
         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) {

paimon-core/src/main/java/org/apache/paimon/Changelog.java

@@ -63,7 +63,8 @@ public Changelog(Snapshot snapshot) {
                 snapshot.watermark(),
                 snapshot.statistics(),
                 snapshot.properties,
-                snapshot.nextRowId);
+                snapshot.nextRowId,
+                snapshot.operation);
     }
 
     @JsonCreator
@@ -89,7 +90,8 @@ public Changelog(
             @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 Changelog(
                 watermark,
                 statistics,
                 properties,
-                nextRowId);
+                nextRowId,
+                operation);
     }
 
     public static Changelog fromJson(String json) {

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);
 

paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java

@@ -161,6 +161,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,
@@ -249,6 +250,12 @@ public FileStoreCommit rowIdCheckConflict(@Nullable Long rowIdCheckFromSnapshot)
         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
@@ -1085,7 +1092,8 @@ CommitResult tryCommitOnce(
                             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(
@@ -1190,7 +1198,8 @@ public boolean replaceManifestList(
                         latest.statistics(),
                         // if empty properties, just set to null
                         latest.properties(),
-                        nextRowId);
+                        nextRowId,
+                        latest.operation());
 
         return commitSnapshotImpl(newSnapshot, emptyList());
     }
@@ -1269,7 +1278,8 @@ private boolean compactManifestOnce() {
                         latestSnapshot.watermark(),
                         latestSnapshot.statistics(),
                         latestSnapshot.properties(),
-                        latestSnapshot.nextRowId());
+                        latestSnapshot.nextRowId(),
+                        latestSnapshot.operation());
 
         return commitSnapshotImpl(newSnapshot, emptyList());
     }

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;
+    }
 }

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 TableCommitImpl rowIdCheckConflict(@Nullable Long rowIdCheckFromSnapshot)
         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 void commit(List<CommitMessage> commitMessages) {
     @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);
     }
 

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 @@ private InternalRow toRow(Snapshot snapshot) {
                     snapshot.deltaRecordCount(),
                     snapshot.changelogRecordCount(),
                     snapshot.watermark(),
-                    snapshot.nextRowId());
+                    snapshot.nextRowId(),
+                    snapshot.operation() == null
+                            ? null
+                            : BinaryString.fromString(snapshot.operation().toString()));
         }
     }
 }

paimon-core/src/main/java/org/apache/paimon/tag/Tag.java

@@ -80,6 +80,7 @@ public Tag(
             @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 Tag(
                 watermark,
                 statistics,
                 properties,
-                nextRowId);
+                nextRowId,
+                operation);
         this.tagCreateTime = tagCreateTime;
         this.tagTimeRetained = tagTimeRetained;
     }
@@ -142,6 +144,7 @@ public static Tag fromSnapshotAndTagTtl(
                 snapshot.statistics(),
                 snapshot.properties(),
                 snapshot.nextRowId(),
+                snapshot.operation(),
                 tagCreateTime,
                 tagTimeRetained);
     }
@@ -168,7 +171,8 @@ public Snapshot trimToSnapshot() {
                 watermark,
                 statistics,
                 properties,
-                nextRowId);
+                nextRowId,
+                operation);
     }
 
     @Override

paimon-core/src/test/java/org/apache/paimon/SnapshotTest.java

@@ -78,6 +78,52 @@ public void testSnapshotWithSizes() {
         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);
     }

paimon-core/src/test/java/org/apache/paimon/append/dataevolution/DataEvolutionRowIdReassignerTest.java

@@ -993,7 +993,8 @@ private void replaceGlobalIndexRangesWithPartitionSpanningRanges(FileStoreTable
                         latest.watermark(),
                         latest.statistics(),
                         latest.properties(),
-                        latest.nextRowId());
+                        latest.nextRowId(),
+                        latest.operation());
         SnapshotManager snapshotManager = table.snapshotManager();
         snapshotManager
                 .fileIO()

paimon-core/src/test/java/org/apache/paimon/catalog/RenamingSnapshotCommitTest.java

@@ -137,6 +137,7 @@ private static Snapshot createSnapshot(long id) throws IOException {
                 null,
                 null,
                 null,
+                null,
                 null);
     }