feat: add commit message and system table utilities

include/paimon/commit_message.h

@@ -0,0 +1,91 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <cstdint>
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "paimon/result.h"
+#include "paimon/type_fwd.h"
+#include "paimon/visibility.h"
+
+namespace paimon {
+class CommitMessageSerializer;
+class MemoryPool;
+
+/// Commit message for partition and bucket. Support Serialize and Deserialize, compatible with java
+/// version.
+// TODO(yonghao.fyh): to add some statistics of write (e.g., write bytes)
+class PAIMON_EXPORT CommitMessage {
+ public:
+    /// Gets the version with which this serializer serializes.
+    static int32_t CurrentVersion();
+
+    virtual ~CommitMessage();
+
+    /// Serializes a single commit message to a binary string format.
+    /// The serialized format is compatible with the Java version of Paimon.
+    /// @param commit_message The commit message to serialize.
+    /// @param pool Memory pool for memory allocation during serialization.
+    /// @return Result containing the serialized string data, or an error if serialization fails.
+    static Result<std::string> Serialize(const std::shared_ptr<CommitMessage>& commit_message,
+                                         const std::shared_ptr<MemoryPool>& pool);
+
+    /// Serializes a list of commit messages to a binary string format.
+    /// @param commit_messages Vector of commit messages to serialize.
+    /// @param pool Memory pool for memory allocation during serialization.
+    /// @return Result containing the serialized string data, or an error if serialization fails.
+    static Result<std::string> SerializeList(
+        const std::vector<std::shared_ptr<CommitMessage>>& commit_messages,
+        const std::shared_ptr<MemoryPool>& pool);
+
+    /// Deserializes a single commit message from binary data.
+    /// @param version The serialization format version used when the data was serialized.
+    /// @param buffer Pointer to the binary data buffer.
+    /// @param length Length of the binary data in bytes.
+    /// @param pool Memory pool for memory allocation during deserialization.
+    /// @return Result containing the deserialized CommitMessage, or an error if deserialization
+    /// fails.
+    static Result<std::shared_ptr<CommitMessage>> Deserialize(
+        int32_t version, const char* buffer, int32_t length,
+        const std::shared_ptr<MemoryPool>& pool);
+
+    /// Deserializes a list of commit messages from binary data.
+    /// This is the counterpart to SerializeList() for batch processing.
+    /// @param version The serialization format version used when the data was serialized.
+    /// @param buffer Pointer to the binary data buffer.
+    /// @param length Length of the binary data in bytes.
+    /// @param pool Memory pool for memory allocation during deserialization.
+    /// @return Result containing a vector of deserialized CommitMessages, or an error if
+    /// deserialization fails.
+    static Result<std::vector<std::shared_ptr<CommitMessage>>> DeserializeList(
+        int32_t version, const char* buffer, int32_t length,
+        const std::shared_ptr<MemoryPool>& pool);
+
+    /// Converts a commit message to a human-readable debug string.
+    /// This is useful for logging, debugging, and troubleshooting purposes.
+    /// @param commit_message The commit message to convert to string.
+    /// @return Result containing the debug string representation, or an error if conversion fails.
+    static Result<std::string> ToDebugString(const std::shared_ptr<CommitMessage>& commit_message);
+};
+
+}  // namespace paimon

src/paimon/core/table/sink/commit_message.cpp

@@ -0,0 +1,101 @@
+/*
+ * 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.
+ */
+
+#include "paimon/commit_message.h"
+
+#include <utility>
+
+#include "paimon/common/io/memory_segment_output_stream.h"
+#include "paimon/common/memory/memory_segment_utils.h"
+#include "paimon/core/table/sink/commit_message_impl.h"
+#include "paimon/core/table/sink/commit_message_serializer.h"
+#include "paimon/io/byte_array_input_stream.h"
+#include "paimon/io/data_input_stream.h"
+#include "paimon/memory/bytes.h"
+#include "paimon/memory/memory_pool.h"
+#include "paimon/result.h"
+#include "paimon/status.h"
+
+namespace paimon {
+
+int32_t CommitMessage::CurrentVersion() {
+    return CommitMessageSerializer::CURRENT_VERSION;
+}
+
+CommitMessage::~CommitMessage() = default;
+
+Result<std::string> CommitMessage::Serialize(const std::shared_ptr<CommitMessage>& commit_message,
+                                             const std::shared_ptr<MemoryPool>& pool) {
+    CommitMessageSerializer serializer(pool);
+    MemorySegmentOutputStream out(MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, pool);
+    PAIMON_RETURN_NOT_OK(serializer.Serialize(commit_message, &out));
+    PAIMON_UNIQUE_PTR<Bytes> bytes =
+        MemorySegmentUtils::CopyToBytes(out.Segments(), 0, out.CurrentSize(), pool.get());
+    return std::string(bytes->data(), bytes->size());
+}
+
+Result<std::string> CommitMessage::SerializeList(
+    const std::vector<std::shared_ptr<CommitMessage>>& commit_messages,
+    const std::shared_ptr<MemoryPool>& pool) {
+    CommitMessageSerializer serializer(pool);
+    MemorySegmentOutputStream out(MemorySegmentOutputStream::DEFAULT_SEGMENT_SIZE, pool);
+    PAIMON_RETURN_NOT_OK(serializer.SerializeList(commit_messages, &out));
+    PAIMON_UNIQUE_PTR<Bytes> bytes =
+        MemorySegmentUtils::CopyToBytes(out.Segments(), 0, out.CurrentSize(), pool.get());
+    return std::string(bytes->data(), bytes->size());
+}
+
+Result<std::shared_ptr<CommitMessage>> CommitMessage::Deserialize(
+    int32_t version, const char* buffer, int32_t length, const std::shared_ptr<MemoryPool>& pool) {
+    if (buffer == nullptr) {
+        return Status::Invalid("buffer is null pointer");
+    }
+    if (length <= 0) {
+        return Status::Invalid("length is equal or less than zero");
+    }
+    CommitMessageSerializer serializer(pool);
+    auto input_stream = std::make_shared<ByteArrayInputStream>(buffer, length);
+    DataInputStream in(input_stream);
+    return serializer.Deserialize(version, &in);
+}
+
+Result<std::vector<std::shared_ptr<CommitMessage>>> CommitMessage::DeserializeList(
+    int32_t version, const char* buffer, int32_t length, const std::shared_ptr<MemoryPool>& pool) {
+    if (buffer == nullptr) {
+        return Status::Invalid("buffer is null pointer");
+    }
+    if (length <= 0) {
+        return Status::Invalid("length is equal or less than zero");
+    }
+    CommitMessageSerializer serializer(pool);
+    auto input_stream = std::make_shared<ByteArrayInputStream>(buffer, length);
+    DataInputStream in(input_stream);
+    return serializer.DeserializeList(version, &in);
+}
+
+Result<std::string> CommitMessage::ToDebugString(
+    const std::shared_ptr<CommitMessage>& commit_message) {
+    auto message = std::dynamic_pointer_cast<CommitMessageImpl>(commit_message);
+    if (message == nullptr) {
+        return Status::Invalid("failed to cast commit message to commit message impl");
+    }
+    return message->ToString();
+}
+
+}  // namespace paimon

src/paimon/core/table/sink/commit_message_impl.cpp

@@ -0,0 +1,65 @@
+/*
+ * 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.
+ */
+
+#include "paimon/core/table/sink/commit_message_impl.h"
+
+#include "fmt/format.h"
+
+namespace paimon {
+
+CommitMessageImpl::CommitMessageImpl(const BinaryRow& partition, int32_t bucket,
+                                     const std::optional<int32_t>& total_buckets,
+                                     const DataIncrement& data_increment,
+                                     const CompactIncrement& compact_increment)
+    : partition_(partition),
+      bucket_(bucket),
+      total_buckets_(total_buckets),
+      data_increment_(data_increment),
+      compact_increment_(compact_increment) {}
+
+std::string CommitMessageImpl::ToString() const {
+    std::string total_buckets_str =
+        (total_buckets_ == std::nullopt ? "null" : std::to_string(total_buckets_.value()));
+    return fmt::format(
+        "FileCommittable {{partition = {}, bucket = {}, totalBuckets = {}, newFilesIncrement = "
+        "{}, compactIncrement = {}}}",
+        partition_.ToString(), bucket_, total_buckets_str, data_increment_.ToString(),
+        compact_increment_.ToString());
+}
+
+bool CommitMessageImpl::operator==(const CommitMessageImpl& other) const {
+    if (this == &other) {
+        return true;
+    }
+    return partition_ == other.partition_ && bucket_ == other.bucket_ &&
+           total_buckets_ == other.total_buckets_ && data_increment_ == other.data_increment_ &&
+           compact_increment_ == other.compact_increment_;
+}
+
+bool CommitMessageImpl::TEST_Equal(const CommitMessageImpl& other) const {
+    if (this == &other) {
+        return true;
+    }
+    return partition_ == other.partition_ && bucket_ == other.bucket_ &&
+           total_buckets_ == other.total_buckets_ &&
+           data_increment_.TEST_Equal(other.data_increment_) &&
+           compact_increment_.TEST_Equal(other.compact_increment_);
+}
+
+}  // namespace paimon

src/paimon/core/table/sink/commit_message_impl.h

@@ -0,0 +1,81 @@
+/*
+ * 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.
+ */
+
+#pragma once
+
+#include <cstdint>
+#include <optional>
+#include <string>
+
+#include "paimon/commit_message.h"
+#include "paimon/common/data/binary_row.h"
+#include "paimon/core/io/compact_increment.h"
+#include "paimon/core/io/data_increment.h"
+
+namespace paimon {
+class CommitMessageImpl : public CommitMessage {
+ public:
+    CommitMessageImpl(const BinaryRow& partition, int32_t bucket,
+                      const std::optional<int32_t>& total_buckets,
+                      const DataIncrement& data_increment,
+                      const CompactIncrement& compact_increment);
+
+    ~CommitMessageImpl() override = default;
+
+    const BinaryRow& Partition() const {
+        return partition_;
+    }
+
+    /// Bucket of this commit message.
+    int32_t Bucket() const {
+        return bucket_;
+    }
+
+    /// Total number of buckets in this partition.
+    const std::optional<int32_t>& TotalBuckets() const {
+        return total_buckets_;
+    }
+
+    const DataIncrement& GetNewFilesIncrement() const {
+        return data_increment_;
+    }
+
+    const CompactIncrement& GetCompactIncrement() const {
+        return compact_increment_;
+    }
+
+    bool IsEmpty() const {
+        return data_increment_.IsEmpty() && compact_increment_.IsEmpty();
+    }
+
+    std::string ToString() const;
+
+    bool operator==(const CommitMessageImpl& other) const;
+
+    bool TEST_Equal(const CommitMessageImpl& other) const;
+
+ private:
+    BinaryRow partition_;
+    int32_t bucket_;
+    std::optional<int32_t> total_buckets_;
+    DataIncrement data_increment_;
+    CompactIncrement compact_increment_;
+};
+
+}  // namespace paimon