feat: snapshot

Signed-off-by: Junwang Zhao <[email protected]>

Author: zhjwpku

src/iceberg/CMakeLists.txt

@@ -26,7 +26,8 @@ set(ICEBERG_SOURCES
     partition_field.cc
     partition_spec.cc
     transform.cc
-    type.cc)
+    type.cc
+    snapshot.cc)
 
 set(ICEBERG_STATIC_BUILD_INTERFACE_LIBS)
 set(ICEBERG_SHARED_BUILD_INTERFACE_LIBS)

src/iceberg/snapshot.cc

@@ -0,0 +1,123 @@
+/*
+ * 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 "iceberg/snapshot.h"
+
+#include "iceberg/util/formatter.h"
+
+namespace iceberg {
+
+namespace {
+/// \brief Get the relative Operation name
+constexpr std::string_view ToString(Operation operation) {
+  switch (operation) {
+    case Operation::kAppend:
+      return "append";
+    case Operation::kOverwrite:
+      return "overwrite";
+    case Operation::kReplace:
+      return "replace";
+    case Operation::kDelete:
+      return "delete";
+    default:
+      return "invalid";
+  }
+}
+}  // namespace
+
+Summary::Summary(Operation op, std::unordered_map<std::string, std::string> props)
+    : operation_(op), additional_properties_(std::move(props)) {}
+
+Operation Summary::operation() const { return operation_; }
+
+const std::unordered_map<std::string, std::string>& Summary::properties() const {
+  return additional_properties_;
+}
+
+std::string Summary::ToString() const {
+  std::string repr =
+      "summary: { operation: " + std::string(iceberg::ToString(operation_));
+  for (const auto& [key, value] : additional_properties_) {
+    repr += ", " + key + ": " + value;
+  }
+  repr += "}";
+  return repr;
+}
+
+bool Summary::Equals(const Summary& other) const {
+  return operation_ == other.operation_ &&
+         additional_properties_ == other.additional_properties_;
+}
+
+Snapshot::Snapshot(int64_t snapshot_id, std::optional<int64_t> parent_snapshot_id,
+                   int64_t sequence_number, int64_t timestamp_ms,
+                   std::string manifest_list, std::shared_ptr<Summary> summary,
+                   std::optional<int64_t> schema_id)
+    : snapshot_id_(snapshot_id),
+      parent_snapshot_id_(parent_snapshot_id),
+      sequence_number_(sequence_number),
+      timestamp_ms_(timestamp_ms),
+      manifest_list_(std::move(manifest_list)),
+      summary_(std::move(summary)),
+      schema_id_(schema_id) {}
+
+int64_t Snapshot::snapshot_id() const { return snapshot_id_; }
+
+std::optional<int64_t> Snapshot::parent_snapshot_id() const {
+  return parent_snapshot_id_;
+}
+
+int64_t Snapshot::sequence_number() const { return sequence_number_; }
+
+int64_t Snapshot::timestamp_ms() const { return timestamp_ms_; }
+
+const std::string& Snapshot::manifest_list() const { return manifest_list_; }
+
+const std::shared_ptr<Summary>& Snapshot::summary() const { return summary_; }
+
+std::optional<int32_t> Snapshot::schema_id() const { return schema_id_; }
+
+std::string Snapshot::ToString() const {
+  std::string repr = "snapshot: { id: " + std::to_string(snapshot_id_);
+  if (parent_snapshot_id_.has_value()) {
+    repr += ", parent_id: " + std::to_string(parent_snapshot_id_.value());
+  }
+  repr += ", sequence_number: " + std::to_string(sequence_number_);
+  repr += ", timestamp_ms: " + std::to_string(timestamp_ms_);
+  repr += ", manifest_list: " + manifest_list_;
+  repr += ", summary: " + summary_->ToString();
+
+  if (schema_id_.has_value()) {
+    repr += ", schema_id: " + std::to_string(schema_id_.value());
+  }
+
+  repr += " }";
+
+  return repr;
+}
+
+bool Snapshot::Equals(const Snapshot& other) const {
+  return snapshot_id_ == other.snapshot_id_ &&
+         parent_snapshot_id_ == other.parent_snapshot_id_ &&
+         sequence_number_ == other.sequence_number_ &&
+         timestamp_ms_ == other.timestamp_ms_ && manifest_list_ == other.manifest_list_ &&
+         *summary_ == *other.summary_ && schema_id_ == other.schema_id_;
+}
+
+}  // namespace iceberg

src/iceberg/snapshot.h

@@ -0,0 +1,217 @@
+/*
+ * 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 <optional>
+#include <string>
+#include <unordered_map>
+
+#include "iceberg/iceberg_export.h"
+#include "iceberg/util/formattable.h"
+
+namespace iceberg {
+
+/// Optional Snapshot Summary Fields
+/// Metrics
+/// See https://iceberg.apache.org/spec/#metrics
+
+/// \brief Number of data files added in the snapshot
+constexpr std::string_view kAddedDataFilesKey = "added-data-files";
+/// \brief Number of data files deleted in the snapshot
+constexpr std::string_view kDeletedDataFilesKey = "deleted-data-files";
+/// \brief Total number of live data files in the snapshot
+constexpr std::string_view kTotalDataFilesKey = "total-data-files";
+/// \brief Number of positional/equality delete files and deletion vectors added in the
+/// snapshot
+constexpr std::string_view kAddedDeleteFilesKey = "added-delete-files";
+/// \brief Number of equality delete files added in the snapshot
+constexpr std::string_view kAddedEqDeleteFilesKey = "added-equality-delete-files";
+/// \brief Number of equality delete files removed in the snapshot
+constexpr std::string_view kRemovedEqDeleteFilesKey = "removed-equality-delete-files";
+/// \brief Number of position delete files added in the snapshot
+constexpr std::string_view kAddedPosDeleteFilesKey = "added-position-delete-files";
+/// \brief Number of position delete files removed in the snapshot
+constexpr std::string_view kRemovedPosDeleteFilesKey = "removed-position-delete-files";
+/// \brief Number of deletion vectors added in the snapshot
+constexpr std::string_view kAddedDVSKey = "added-dvs";
+/// \brief Number of deletion vectors removed in the snapshot
+constexpr std::string_view kRemovedDVSKey = "removed-dvs";
+/// \brief Number of positional/equality delete files and deletion vectors removed in the
+/// snapshot
+constexpr std::string_view kRemovedDeleteFilesKey = "removed-delete-files";
+/// \brief Total number of live positional/equality delete files and deletion vectors in
+/// the snapshot
+constexpr std::string_view kTotalDeleteFilesKey = "total-delete-files";
+/// \brief Number of records added in the snapshot
+constexpr std::string_view kAddedRecordsKey = "added-records";
+/// \brief Number of records deleted in the snapshot
+constexpr std::string_view kDeletedRecordsKey = "deleted-records";
+/// \brief Total number of records in the snapshot
+constexpr std::string_view kTotalRecordsKey = "total-records";
+/// \brief The size of files added in the snapshot
+constexpr std::string_view kAddedFileSizeKey = "added-files-size";
+/// \brief The size of files removed in the snapshot
+constexpr std::string_view kRemovedFileSizeKey = "removed-files-size";
+/// \brief Total size of live files in the snapshot
+constexpr std::string_view kTotalFileSizeKey = "total-files-size";
+/// \brief Number of position delete records added in the snapshot
+constexpr std::string_view kAddedPosDeletesKey = "added-position-deletes";
+/// \brief Number of position delete records removed in the snapshot
+constexpr std::string_view kRemovedPosDeletesKey = "removed-position-deletes";
+/// \brief Total number of position delete records in the snapshot
+constexpr std::string_view kTotalPosDeletesKey = "total-position-deletes";
+/// \brief Number of equality delete records added in the snapshot
+constexpr std::string_view kAddedEqDeletesKey = "added-equality-deletes";
+/// \brief Number of equality delete records removed in the snapshot
+constexpr std::string_view kRemovedEqDeletesKey = "removed-equality-deletes";
+/// \brief Total number of equality delete records in the snapshot
+constexpr std::string_view kTotalEqDeletesKey = "total-equality-deletes";
+/// \brief Number of duplicate files deleted (duplicates are files recorded more than once
+/// in the manifest)
+constexpr std::string_view kDeletedDuplicatedFilesKey = "deleted-duplicate-files";
+/// \brief Number of partitions with files added or removed in the snapshot
+constexpr std::string_view kChangedPartitionCountProp = "changed-partition-count";
+
+/// Other Fields
+/// See https://iceberg.apache.org/spec/#other-fields
+
+/// \brief The Write-Audit-Publish id of a staged snapshot
+constexpr std::string_view kWAPIDKey = "wap.id";
+/// \brief The Write-Audit-Publish id of a snapshot already been published
+constexpr std::string_view kPublishedWAPIDKey = "published-wap-id";
+/// \brief The original id of a cherry-picked snapshot
+constexpr std::string_view kSourceSnapshotIDKey = "source-snapshot-id";
+/// \brief Name of the engine that created the snapshot
+constexpr std::string_view kEngineNameKey = "engine-name";
+/// \brief Version of the engine that created the snapshot
+constexpr std::string_view kEngineVersionKey = "engine-version";
+
+/// \brief The operation field is used by some operations, like snapshot expiration, to
+/// skip processing certain snapshots.
+enum class Operation {
+  /// Only data files were added and no files were removed.
+  kAppend,
+  /// Data and delete files were added and removed without changing table data; i.e.
+  /// compaction, change the data file format, or relocating data files.
+  kReplace,
+  /// Data and delete files were added and removed in a logical overwrite operation.
+  kOverwrite,
+  /// Data files were removed and their contents logically deleted and/or delete files
+  /// were added to delete rows.
+  kDelete,
+};
+
+/// \brief Summarises the changes in the snapshot.
+class ICEBERG_EXPORT Summary : public iceberg::util::Formattable {
+ public:
+  Summary() = default;
+  /// \brief Construct a summary with the given operation and properties.
+  Summary(Operation op, std::unordered_map<std::string, std::string> props);
+
+  /// \brief Get the operation type of the snapshot.
+  [[nodiscard]] Operation operation() const;
+
+  /// \brief Get the additional properties of the snapshot.
+  [[nodiscard]] const std::unordered_map<std::string, std::string>& properties() const;
+
+  std::string ToString() const override;
+
+  friend bool operator==(const Summary& lhs, const Summary& rhs) {
+    return lhs.Equals(rhs);
+  }
+
+  friend bool operator!=(const Summary& lhs, const Summary& rhs) { return !(lhs == rhs); }
+
+ private:
+  /// \brief Compare two Summaries for equality.
+  [[nodiscard]] bool Equals(const Summary& other) const;
+
+  /// The type of operation in the snapshot
+  Operation operation_{Operation::kAppend};
+  /// Other summary data.
+  std::unordered_map<std::string, std::string> additional_properties_;
+};
+
+/// \brief A snapshot of the data in a table at a point in time.
+///
+/// A snapshot consist of one or more file manifests, and the complete table contents is
+/// the union of all the data files in those manifests.
+///
+/// Snapshots are created by table operations.
+class ICEBERG_EXPORT Snapshot : public iceberg::util::Formattable {
+ public:
+  Snapshot(int64_t snapshot_id, std::optional<int64_t> parent_snapshot_id,
+           int64_t sequence_number, int64_t timestamp_ms, std::string manifest_list,
+           std::shared_ptr<Summary> summary, std::optional<int64_t> schema_id);
+
+  /// \brief Get the id of the snapshot.
+  [[nodiscard]] int64_t snapshot_id() const;
+
+  /// \brief Get parent snapshot id.
+  [[nodiscard]] std::optional<int64_t> parent_snapshot_id() const;
+
+  /// \brief Get the sequence number of the snapshot.
+  [[nodiscard]] int64_t sequence_number() const;
+
+  /// \brief Get the timestamp of the snapshot.
+  [[nodiscard]] int64_t timestamp_ms() const;
+
+  /// \brief Get the manifest list of the snapshot.
+  [[nodiscard]] const std::string& manifest_list() const;
+
+  /// \brief Get the summary of the snapshot.
+  [[nodiscard]] const std::shared_ptr<Summary>& summary() const;
+
+  /// \brief Get the schema ID of the snapshot.
+  [[nodiscard]] std::optional<int32_t> schema_id() const;
+
+  std::string ToString() const override;
+
+  friend bool operator==(const Snapshot& lhs, const Snapshot& rhs) {
+    return lhs.Equals(rhs);
+  }
+
+  friend bool operator!=(const Snapshot& lhs, const Snapshot& rhs) {
+    return !(lhs == rhs);
+  }
+
+ private:
+  /// \brief Compare two snapshots for equality.
+  [[nodiscard]] bool Equals(const Snapshot& other) const;
+
+  /// A unqiue long ID.
+  int64_t snapshot_id_;
+  /// The snapshot ID of the snapshot's parent. Omitted for any snapshot with no parent.
+  std::optional<int64_t> parent_snapshot_id_;
+  /// A monotonically increasing long that tracks the order of changes to a table.
+  int64_t sequence_number_;
+  /// A timestamp when the snapshot was created, used for garbage collection and table
+  /// inspection.
+  int64_t timestamp_ms_;
+  /// The location of a manifest list for this snapshot that tracks manifest files with
+  /// additional metadata.
+  std::string manifest_list_;
+  /// A string map that summaries the snapshot changes, including operation.
+  std::shared_ptr<Summary> summary_;
+  /// ID of the table's current schema when the snapshot was created.
+  std::optional<int32_t> schema_id_;
+};
+
+}  // namespace iceberg