[core] Introduce properties field in Snapshot (apache#5703)

Author: luoyuxia

paimon-api/src/main/java/org/apache/paimon/Snapshot.java

@@ -87,6 +87,7 @@ public class Snapshot implements Serializable {
     protected static final String FIELD_CHANGELOG_RECORD_COUNT = "changelogRecordCount";
     protected static final String FIELD_WATERMARK = "watermark";
     protected static final String FIELD_STATISTICS = "statistics";
+    protected static final String FIELD_PROPERTIES = "properties";
 
     // version of snapshot
     // null for paimon <= 0.2
@@ -195,6 +196,13 @@ public class Snapshot implements Serializable {
     @Nullable
     protected final String statistics;
 
+    // properties
+    // null for paimon <= 1.1 or empty properties
+    @JsonInclude(JsonInclude.Include.NON_NULL)
+    @JsonProperty(FIELD_PROPERTIES)
+    @Nullable
+    protected final Map<String, String> properties;
+
     public Snapshot(
             long id,
             long schemaId,
@@ -214,7 +222,8 @@ public Snapshot(
             @Nullable Long deltaRecordCount,
             @Nullable Long changelogRecordCount,
             @Nullable Long watermark,
-            @Nullable String statistics) {
+            @Nullable String statistics,
+            @Nullable Map<String, String> properties) {
         this(
                 CURRENT_VERSION,
                 id,
@@ -235,7 +244,8 @@ public Snapshot(
                 deltaRecordCount,
                 changelogRecordCount,
                 watermark,
-                statistics);
+                statistics,
+                properties);
     }
 
     @JsonCreator
@@ -260,7 +270,8 @@ public Snapshot(
             @JsonProperty(FIELD_DELTA_RECORD_COUNT) @Nullable Long deltaRecordCount,
             @JsonProperty(FIELD_CHANGELOG_RECORD_COUNT) @Nullable Long changelogRecordCount,
             @JsonProperty(FIELD_WATERMARK) @Nullable Long watermark,
-            @JsonProperty(FIELD_STATISTICS) @Nullable String statistics) {
+            @JsonProperty(FIELD_STATISTICS) @Nullable String statistics,
+            @JsonProperty(FIELD_PROPERTIES) @Nullable Map<String, String> properties) {
         this.version = version;
         this.id = id;
         this.schemaId = schemaId;
@@ -281,6 +292,7 @@ public Snapshot(
         this.changelogRecordCount = changelogRecordCount;
         this.watermark = watermark;
         this.statistics = statistics;
+        this.properties = properties;
     }
 
     @JsonGetter(FIELD_VERSION)
@@ -395,6 +407,12 @@ public String statistics() {
         return statistics;
     }
 
+    @JsonGetter(FIELD_PROPERTIES)
+    @Nullable
+    public Map<String, String> properties() {
+        return properties;
+    }
+
     public String toJson() {
         return JsonSerdeUtil.toJson(this);
     }
@@ -421,7 +439,8 @@ public int hashCode() {
                 deltaRecordCount,
                 changelogRecordCount,
                 watermark,
-                statistics);
+                statistics,
+                properties);
     }
 
     @Override
@@ -452,7 +471,8 @@ public boolean equals(Object o) {
                 && Objects.equals(deltaRecordCount, that.deltaRecordCount)
                 && Objects.equals(changelogRecordCount, that.changelogRecordCount)
                 && Objects.equals(watermark, that.watermark)
-                && Objects.equals(statistics, that.statistics);
+                && Objects.equals(statistics, that.statistics)
+                && Objects.equals(properties, that.properties);
     }
 
     /** Type of changes in this snapshot. */

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

@@ -62,7 +62,8 @@ public Changelog(Snapshot snapshot) {
                 snapshot.deltaRecordCount(),
                 snapshot.changelogRecordCount(),
                 snapshot.watermark(),
-                snapshot.statistics());
+                snapshot.statistics(),
+                snapshot.properties);
     }
 
     @JsonCreator
@@ -87,7 +88,8 @@ public Changelog(
             @JsonProperty(FIELD_DELTA_RECORD_COUNT) @Nullable Long deltaRecordCount,
             @JsonProperty(FIELD_CHANGELOG_RECORD_COUNT) @Nullable Long changelogRecordCount,
             @JsonProperty(FIELD_WATERMARK) @Nullable Long watermark,
-            @JsonProperty(FIELD_STATISTICS) @Nullable String statistics) {
+            @JsonProperty(FIELD_STATISTICS) @Nullable String statistics,
+            @JsonProperty(FIELD_PROPERTIES) Map<String, String> properties) {
         super(
                 version,
                 id,
@@ -108,7 +110,8 @@ public Changelog(
                 deltaRecordCount,
                 changelogRecordCount,
                 watermark,
-                statistics);
+                statistics,
+                properties);
     }
 
     public static Changelog fromJson(String json) {

paimon-core/src/main/java/org/apache/paimon/manifest/ManifestCommittable.java

@@ -34,6 +34,7 @@ public class ManifestCommittable {
     private final long identifier;
     @Nullable private final Long watermark;
     private final Map<Integer, Long> logOffsets;
+    private final Map<String, String> properties;
     private final List<CommitMessage> commitMessages;
 
     public ManifestCommittable(long identifier) {
@@ -45,17 +46,28 @@ public ManifestCommittable(long identifier, @Nullable Long watermark) {
         this.watermark = watermark;
         this.logOffsets = new HashMap<>();
         this.commitMessages = new ArrayList<>();
+        this.properties = new HashMap<>();
     }
 
     public ManifestCommittable(
             long identifier,
             @Nullable Long watermark,
             Map<Integer, Long> logOffsets,
             List<CommitMessage> commitMessages) {
+        this(identifier, watermark, logOffsets, commitMessages, new HashMap<>());
+    }
+
+    public ManifestCommittable(
+            long identifier,
+            @Nullable Long watermark,
+            Map<Integer, Long> logOffsets,
+            List<CommitMessage> commitMessages,
+            Map<String, String> properties) {
         this.identifier = identifier;
         this.watermark = watermark;
         this.logOffsets = logOffsets;
         this.commitMessages = commitMessages;
+        this.properties = properties;
     }
 
     public void addFileCommittable(CommitMessage commitMessage) {
@@ -72,6 +84,10 @@ public void addLogOffset(int bucket, long offset, boolean allowDuplicate) {
         logOffsets.put(bucket, newOffset);
     }
 
+    public void addProperty(String key, String value) {
+        properties.put(key, value);
+    }
+
     public long identifier() {
         return identifier;
     }
@@ -89,6 +105,10 @@ public List<CommitMessage> fileCommittables() {
         return commitMessages;
     }
 
+    public Map<String, String> properties() {
+        return properties;
+    }
+
     @Override
     public boolean equals(Object o) {
         if (this == o) {
@@ -101,12 +121,13 @@ public boolean equals(Object o) {
         return Objects.equals(identifier, that.identifier)
                 && Objects.equals(watermark, that.watermark)
                 && Objects.equals(logOffsets, that.logOffsets)
-                && Objects.equals(commitMessages, that.commitMessages);
+                && Objects.equals(commitMessages, that.commitMessages)
+                && Objects.equals(properties, that.properties);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(identifier, watermark, logOffsets, commitMessages);
+        return Objects.hash(identifier, watermark, logOffsets, commitMessages, properties);
     }
 
     @Override
@@ -116,7 +137,8 @@ public String toString() {
                         + "identifier = %s, "
                         + "watermark = %s, "
                         + "logOffsets = %s, "
-                        + "commitMessages = %s",
-                identifier, watermark, logOffsets, commitMessages);
+                        + "commitMessages = %s, "
+                        + "properties = %s}",
+                identifier, watermark, logOffsets, commitMessages, properties);
     }
 }

paimon-core/src/main/java/org/apache/paimon/manifest/ManifestCommittableSerializer.java

@@ -34,7 +34,7 @@
 /** {@link VersionedSerializer} for {@link ManifestCommittable}. */
 public class ManifestCommittableSerializer implements VersionedSerializer<ManifestCommittable> {
 
-    private static final int CURRENT_VERSION = 3;
+    private static final int CURRENT_VERSION = 4;
 
     private final CommitMessageSerializer commitMessageSerializer;
 
@@ -62,6 +62,7 @@ public byte[] serialize(ManifestCommittable obj) throws IOException {
             view.writeLong(watermark);
         }
         serializeOffsets(view, obj.logOffsets());
+        serializeProperties(view, obj.properties());
         view.writeInt(commitMessageSerializer.getVersion());
         commitMessageSerializer.serializeList(obj.fileCommittables(), view);
         return out.toByteArray();
@@ -76,6 +77,15 @@ private void serializeOffsets(DataOutputViewStreamWrapper view, Map<Integer, Lon
         }
     }
 
+    private void serializeProperties(
+            DataOutputViewStreamWrapper view, Map<String, String> properties) throws IOException {
+        view.writeInt(properties.size());
+        for (Map.Entry<String, String> entry : properties.entrySet()) {
+            view.writeUTF(entry.getKey());
+            view.writeUTF(entry.getValue());
+        }
+    }
+
     @Override
     public ManifestCommittable deserialize(int version, byte[] serialized) throws IOException {
         if (version > CURRENT_VERSION) {
@@ -91,6 +101,8 @@ public ManifestCommittable deserialize(int version, byte[] serialized) throws IO
         long identifier = view.readLong();
         Long watermark = view.readBoolean() ? null : view.readLong();
         Map<Integer, Long> offsets = deserializeOffsets(view);
+        Map<String, String> properties =
+                version == CURRENT_VERSION ? deserializeProperties(view) : new HashMap<>();
         int fileCommittableSerializerVersion = view.readInt();
         List<CommitMessage> fileCommittables;
         try {
@@ -116,7 +128,8 @@ public ManifestCommittable deserialize(int version, byte[] serialized) throws IO
             fileCommittables = legacyV2CommitMessageSerializer.deserializeList(view);
         }
 
-        return new ManifestCommittable(identifier, watermark, offsets, fileCommittables);
+        return new ManifestCommittable(
+                identifier, watermark, offsets, fileCommittables, properties);
     }
 
     private Map<Integer, Long> deserializeOffsets(DataInputDeserializer view) throws IOException {
@@ -127,4 +140,14 @@ private Map<Integer, Long> deserializeOffsets(DataInputDeserializer view) throws
         }
         return offsets;
     }
+
+    private Map<String, String> deserializeProperties(DataInputDeserializer view)
+            throws IOException {
+        int size = view.readInt();
+        Map<String, String> properties = new HashMap<>(size);
+        for (int i = 0; i < size; i++) {
+            properties.put(view.readUTF(), view.readUTF());
+        }
+        return properties;
+    }
 }