feat: add snapshot, snapshot_info, and table core utilities with tests (#118)

Author: lucasfang

include/paimon/snapshot/snapshot_info.h

@@ -0,0 +1,53 @@
+/*
+ * 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/visibility.h"
+
+namespace paimon {
+
+/// Plain snapshot metadata returned by Catalog::ListSnapshots().
+struct PAIMON_EXPORT SnapshotInfo {
+    /// Commit kind exposed through the public Catalog API.
+    enum class PAIMON_EXPORT CommitKind : int8_t {
+        APPEND,
+        COMPACT,
+        OVERWRITE,
+        ANALYZE,
+        UNKNOWN,
+    };
+
+    /// Convert a CommitKind to its canonical string representation.
+    static std::string CommitKindToString(CommitKind kind);
+
+    int64_t snapshot_id;
+    int64_t schema_id;
+    std::string commit_user;
+    CommitKind commit_kind;
+    int64_t time_millis;
+    std::optional<int64_t> total_record_count;
+    std::optional<int64_t> delta_record_count;
+    std::optional<int64_t> watermark;
+};
+
+}  // namespace paimon

src/paimon/core/snapshot.cpp

@@ -0,0 +1,329 @@
+/*
+ * 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/snapshot.h"
+
+#include <cassert>
+#include <stdexcept>
+#include <utility>
+
+#include "paimon/common/utils/rapidjson_util.h"
+#include "paimon/fs/file_system.h"
+#include "paimon/result.h"
+#include "paimon/status.h"
+#include "rapidjson/allocators.h"
+#include "rapidjson/document.h"
+#include "rapidjson/rapidjson.h"
+
+namespace paimon {
+
+const Snapshot::CommitKind Snapshot::CommitKind::Append() {
+    static const Snapshot::CommitKind commit_kind = Snapshot::CommitKind(0);
+    return commit_kind;
+}
+
+const Snapshot::CommitKind Snapshot::CommitKind::Compact() {
+    static const Snapshot::CommitKind commit_kind = Snapshot::CommitKind(1);
+    return commit_kind;
+}
+
+const Snapshot::CommitKind Snapshot::CommitKind::Overwrite() {
+    static const Snapshot::CommitKind commit_kind = Snapshot::CommitKind(2);
+    return commit_kind;
+}
+
+const Snapshot::CommitKind Snapshot::CommitKind::Analyze() {
+    static const Snapshot::CommitKind commit_kind = Snapshot::CommitKind(3);
+    return commit_kind;
+}
+
+const Snapshot::CommitKind Snapshot::CommitKind::Unknown() {
+    static const Snapshot::CommitKind commit_kind = Snapshot::CommitKind(-1);
+    return commit_kind;
+}
+bool Snapshot::TEST_Equal(const Snapshot& other) const {
+    if (this == &other) {
+        return true;
+    }
+
+    if ((base_manifest_list_size_ && !other.base_manifest_list_size_) ||
+        (!base_manifest_list_size_ && other.base_manifest_list_size_)) {
+        return false;
+    }
+    if ((delta_manifest_list_size_ && !other.delta_manifest_list_size_) ||
+        (!delta_manifest_list_size_ && other.delta_manifest_list_size_)) {
+        return false;
+    }
+    if ((changelog_manifest_list_ && !other.changelog_manifest_list_) ||
+        (!changelog_manifest_list_ && other.changelog_manifest_list_)) {
+        return false;
+    }
+    if ((changelog_manifest_list_size_ && !other.changelog_manifest_list_size_) ||
+        (!changelog_manifest_list_size_ && other.changelog_manifest_list_size_)) {
+        return false;
+    }
+
+    return version_ == other.version_ && id_ == other.id_ && schema_id_ == other.schema_id_ &&
+           index_manifest_ == other.index_manifest_ && commit_user_ == other.commit_user_ &&
+           commit_identifier_ == other.commit_identifier_ && commit_kind_ == other.commit_kind_ &&
+           log_offsets_ == other.log_offsets_ && total_record_count_ == other.total_record_count_ &&
+           delta_record_count_ == other.delta_record_count_ &&
+           changelog_record_count_ == other.changelog_record_count_ &&
+           watermark_ == other.watermark_ && statistics_ == other.statistics_ &&
+           properties_ == other.properties_ && next_row_id_ == other.next_row_id_;
+}
+
+bool Snapshot::operator==(const Snapshot& other) const {
+    if (this == &other) {
+        return true;
+    }
+    return version_ == other.version_ && id_ == other.id_ && schema_id_ == other.schema_id_ &&
+           base_manifest_list_ == other.base_manifest_list_ &&
+           base_manifest_list_size_ == other.base_manifest_list_size_ &&
+           delta_manifest_list_ == other.delta_manifest_list_ &&
+           delta_manifest_list_size_ == other.delta_manifest_list_size_ &&
+           changelog_manifest_list_ == other.changelog_manifest_list_ &&
+           changelog_manifest_list_size_ == other.changelog_manifest_list_size_ &&
+           index_manifest_ == other.index_manifest_ && commit_user_ == other.commit_user_ &&
+           commit_identifier_ == other.commit_identifier_ && commit_kind_ == other.commit_kind_ &&
+           time_millis_ == other.time_millis_ && log_offsets_ == other.log_offsets_ &&
+           total_record_count_ == other.total_record_count_ &&
+           delta_record_count_ == other.delta_record_count_ &&
+           changelog_record_count_ == other.changelog_record_count_ &&
+           watermark_ == other.watermark_ && statistics_ == other.statistics_ &&
+           properties_ == other.properties_ && next_row_id_ == other.next_row_id_;
+}
+
+std::string Snapshot::CommitKind::ToString(const Snapshot::CommitKind& kind) {
+    switch (kind.value_) {
+        case 0:
+            return "APPEND";
+        case 1:
+            return "COMPACT";
+        case 2:
+            return "OVERWRITE";
+        case 3:
+            return "ANALYZE";
+        default:
+            assert(false);
+            return "UNKNOWN";
+    }
+}
+Snapshot::CommitKind Snapshot::CommitKind::FromString(const std::string& kind) {
+    if (kind == "APPEND") {
+        return Append();
+    } else if (kind == "COMPACT") {
+        return Compact();
+    } else if (kind == "OVERWRITE") {
+        return Overwrite();
+    } else if (kind == "ANALYZE") {
+        return Analyze();
+    }
+    assert(false);
+    return Unknown();
+}
+
+Snapshot::Snapshot(const std::optional<int32_t>& version, int64_t id, int64_t schema_id,
+                   const std::string& base_manifest_list,
+                   const std::optional<int64_t>& base_manifest_list_size,
+                   const std::string& delta_manifest_list,
+                   const std::optional<int64_t>& delta_manifest_list_size,
+                   const std::optional<std::string>& changelog_manifest_list,
+                   const std::optional<int64_t>& changelog_manifest_list_size,
+                   const std::optional<std::string>& index_manifest, const std::string& commit_user,
+                   int64_t commit_identifier, CommitKind commit_kind, int64_t time_millis,
+                   const std::optional<std::map<int32_t, int64_t>>& log_offsets,
+                   const std::optional<int64_t>& total_record_count,
+                   const std::optional<int64_t>& delta_record_count,
+                   const std::optional<int64_t>& changelog_record_count,
+                   const std::optional<int64_t>& watermark,
+                   const std::optional<std::string>& statistics,
+                   const std::optional<std::map<std::string, std::string>>& properties,
+                   const std::optional<int64_t>& next_row_id)
+    : version_(version),
+      id_(id),
+      schema_id_(schema_id),
+      base_manifest_list_(base_manifest_list),
+      base_manifest_list_size_(base_manifest_list_size),
+      delta_manifest_list_(delta_manifest_list),
+      delta_manifest_list_size_(delta_manifest_list_size),
+      changelog_manifest_list_(changelog_manifest_list),
+      changelog_manifest_list_size_(changelog_manifest_list_size),
+      index_manifest_(index_manifest),
+      commit_user_(commit_user),
+      commit_identifier_(commit_identifier),
+      commit_kind_(commit_kind),
+      time_millis_(time_millis),
+      log_offsets_(log_offsets),
+      total_record_count_(total_record_count),
+      delta_record_count_(delta_record_count),
+      changelog_record_count_(changelog_record_count),
+      watermark_(watermark),
+      statistics_(statistics),
+      properties_(properties),
+      next_row_id_(next_row_id) {}
+
+rapidjson::Value Snapshot::ToJson(rapidjson::Document::AllocatorType* allocator) const
+    noexcept(false) {
+    rapidjson::Value obj(rapidjson::kObjectType);
+    obj.AddMember(rapidjson::StringRef(FIELD_VERSION),
+                  RapidJsonUtil::SerializeValue(Version(), allocator).Move(), *allocator);
+    obj.AddMember(rapidjson::StringRef(FIELD_ID),
+                  RapidJsonUtil::SerializeValue(id_, allocator).Move(), *allocator);
+    obj.AddMember(rapidjson::StringRef(FIELD_SCHEMA_ID),
+                  RapidJsonUtil::SerializeValue(schema_id_, allocator).Move(), *allocator);
+    obj.AddMember(rapidjson::StringRef(FIELD_BASE_MANIFEST_LIST),
+                  RapidJsonUtil::SerializeValue(base_manifest_list_, allocator).Move(), *allocator);
+    if (base_manifest_list_size_) {
+        obj.AddMember(
+            rapidjson::StringRef(FIELD_BASE_MANIFEST_LIST_SIZE),
+            RapidJsonUtil::SerializeValue(base_manifest_list_size_.value(), allocator).Move(),
+            *allocator);
+    }
+    obj.AddMember(rapidjson::StringRef(FIELD_DELTA_MANIFEST_LIST),
+                  RapidJsonUtil::SerializeValue(delta_manifest_list_, allocator).Move(),
+                  *allocator);
+    if (delta_manifest_list_size_) {
+        obj.AddMember(rapidjson::StringRef(FIELD_DELTA_MANIFEST_LIST_SIZE),
+                      RapidJsonUtil::SerializeValue(delta_manifest_list_size_, allocator).Move(),
+                      *allocator);
+    }
+    obj.AddMember(rapidjson::StringRef(FIELD_CHANGELOG_MANIFEST_LIST),
+                  RapidJsonUtil::SerializeValue(changelog_manifest_list_, allocator).Move(),
+                  *allocator);
+    if (changelog_manifest_list_size_) {
+        obj.AddMember(
+            rapidjson::StringRef(FIELD_CHANGELOG_MANIFEST_LIST_SIZE),
+            RapidJsonUtil::SerializeValue(changelog_manifest_list_size_, allocator).Move(),
+            *allocator);
+    }
+    if (index_manifest_ != std::nullopt) {
+        obj.AddMember(rapidjson::StringRef(FIELD_INDEX_MANIFEST),
+                      RapidJsonUtil::SerializeValue(index_manifest_.value(), allocator).Move(),
+                      *allocator);
+    }
+
+    obj.AddMember(rapidjson::StringRef(FIELD_COMMIT_USER),
+                  RapidJsonUtil::SerializeValue(commit_user_, allocator).Move(), *allocator);
+    obj.AddMember(rapidjson::StringRef(FIELD_COMMIT_IDENTIFIER),
+                  RapidJsonUtil::SerializeValue(commit_identifier_, allocator).Move(), *allocator);
+    obj.AddMember(
+        rapidjson::StringRef(FIELD_COMMIT_KIND),
+        RapidJsonUtil::SerializeValue(Snapshot::CommitKind::ToString(commit_kind_), allocator)
+            .Move(),
+        *allocator);
+
+    obj.AddMember(rapidjson::StringRef(FIELD_TIME_MILLIS),
+                  RapidJsonUtil::SerializeValue(time_millis_, allocator).Move(), *allocator);
+    if (log_offsets_ != std::nullopt) {
+        obj.AddMember(rapidjson::StringRef(FIELD_LOG_OFFSETS),
+                      RapidJsonUtil::SerializeValue(log_offsets_.value(), allocator).Move(),
+                      *allocator);
+    }
+    obj.AddMember(rapidjson::StringRef(FIELD_TOTAL_RECORD_COUNT),
+                  RapidJsonUtil::SerializeValue(total_record_count_.value(), allocator).Move(),
+                  *allocator);
+    obj.AddMember(rapidjson::StringRef(FIELD_DELTA_RECORD_COUNT),
+                  RapidJsonUtil::SerializeValue(delta_record_count_.value(), allocator).Move(),
+                  *allocator);
+
+    if (changelog_record_count_ != std::nullopt) {
+        obj.AddMember(
+            rapidjson::StringRef(FIELD_CHANGELOG_RECORD_COUNT),
+            RapidJsonUtil::SerializeValue(changelog_record_count_.value(), allocator).Move(),
+            *allocator);
+    }
+
+    if (watermark_ != std::nullopt) {
+        obj.AddMember(rapidjson::StringRef(FIELD_WATERMARK),
+                      RapidJsonUtil::SerializeValue(watermark_.value(), allocator).Move(),
+                      *allocator);
+    }
+
+    if (statistics_ != std::nullopt) {
+        obj.AddMember(rapidjson::StringRef(FIELD_STATISTICS),
+                      RapidJsonUtil::SerializeValue(statistics_.value(), allocator).Move(),
+                      *allocator);
+    }
+    if (properties_ != std::nullopt) {
+        obj.AddMember(rapidjson::StringRef(FIELD_PROPERTIES),
+                      RapidJsonUtil::SerializeValue(properties_.value(), allocator).Move(),
+                      *allocator);
+    }
+    if (next_row_id_ != std::nullopt) {
+        obj.AddMember(rapidjson::StringRef(FIELD_NEXT_ROW_ID),
+                      RapidJsonUtil::SerializeValue(next_row_id_.value(), allocator).Move(),
+                      *allocator);
+    }
+
+    return obj;
+}
+
+void Snapshot::FromJson(const rapidjson::Value& obj) noexcept(false) {
+    version_ = RapidJsonUtil::DeserializeKeyValue<int32_t>(obj, FIELD_VERSION, -1);
+    id_ = RapidJsonUtil::DeserializeKeyValue<int64_t>(obj, FIELD_ID);
+    schema_id_ = RapidJsonUtil::DeserializeKeyValue<int64_t>(obj, FIELD_SCHEMA_ID);
+    base_manifest_list_ =
+        RapidJsonUtil::DeserializeKeyValue<std::string>(obj, FIELD_BASE_MANIFEST_LIST);
+    base_manifest_list_size_ = RapidJsonUtil::DeserializeKeyValue<std::optional<int64_t>>(
+        obj, FIELD_BASE_MANIFEST_LIST_SIZE);
+    delta_manifest_list_ =
+        RapidJsonUtil::DeserializeKeyValue<std::string>(obj, FIELD_DELTA_MANIFEST_LIST);
+    delta_manifest_list_size_ = RapidJsonUtil::DeserializeKeyValue<std::optional<int64_t>>(
+        obj, FIELD_DELTA_MANIFEST_LIST_SIZE);
+    changelog_manifest_list_ = RapidJsonUtil::DeserializeKeyValue<std::optional<std::string>>(
+        obj, FIELD_CHANGELOG_MANIFEST_LIST);
+    changelog_manifest_list_size_ = RapidJsonUtil::DeserializeKeyValue<std::optional<int64_t>>(
+        obj, FIELD_CHANGELOG_MANIFEST_LIST_SIZE);
+    index_manifest_ =
+        RapidJsonUtil::DeserializeKeyValue<std::optional<std::string>>(obj, FIELD_INDEX_MANIFEST);
+    commit_user_ = RapidJsonUtil::DeserializeKeyValue<std::string>(obj, FIELD_COMMIT_USER);
+    commit_identifier_ = RapidJsonUtil::DeserializeKeyValue<int64_t>(obj, FIELD_COMMIT_IDENTIFIER);
+    commit_kind_ = Snapshot::CommitKind::FromString(
+        RapidJsonUtil::DeserializeKeyValue<std::string>(obj, FIELD_COMMIT_KIND));
+    if (commit_kind_ == Snapshot::CommitKind::Unknown()) {
+        throw std::invalid_argument("deserialize CommitKind failed");
+    }
+    time_millis_ = RapidJsonUtil::DeserializeKeyValue<int64_t>(obj, FIELD_TIME_MILLIS);
+    log_offsets_ = RapidJsonUtil::DeserializeKeyValue<std::optional<std::map<int32_t, int64_t>>>(
+        obj, FIELD_LOG_OFFSETS);
+    total_record_count_ =
+        RapidJsonUtil::DeserializeKeyValue<std::optional<int64_t>>(obj, FIELD_TOTAL_RECORD_COUNT);
+    delta_record_count_ =
+        RapidJsonUtil::DeserializeKeyValue<std::optional<int64_t>>(obj, FIELD_DELTA_RECORD_COUNT);
+    changelog_record_count_ = RapidJsonUtil::DeserializeKeyValue<std::optional<int64_t>>(
+        obj, FIELD_CHANGELOG_RECORD_COUNT);
+    watermark_ = RapidJsonUtil::DeserializeKeyValue<std::optional<int64_t>>(obj, FIELD_WATERMARK);
+    statistics_ =
+        RapidJsonUtil::DeserializeKeyValue<std::optional<std::string>>(obj, FIELD_STATISTICS);
+    properties_ =
+        RapidJsonUtil::DeserializeKeyValue<std::optional<std::map<std::string, std::string>>>(
+            obj, FIELD_PROPERTIES);
+    next_row_id_ =
+        RapidJsonUtil::DeserializeKeyValue<std::optional<int64_t>>(obj, FIELD_NEXT_ROW_ID);
+}
+
+Result<Snapshot> Snapshot::FromPath(const std::shared_ptr<FileSystem>& fs,
+                                    const std::string& path) {
+    std::string json_str;
+    PAIMON_RETURN_NOT_OK(fs->ReadFile(path, &json_str));
+    PAIMON_ASSIGN_OR_RAISE(Snapshot snapshot, Snapshot::FromJsonString(json_str));
+    return snapshot;
+}
+
+}  // namespace paimon