snapshot.h

/*
 * 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 <memory>
#include <optional>
#include <span>
#include <string>
#include <string_view>
#include <unordered_map>
#include <variant>
 
#include "iceberg/iceberg_export.h"
#include "iceberg/manifest/manifest_list.h"
#include "iceberg/result.h"
#include "iceberg/type_fwd.h"
#include "iceberg/util/lazy.h"
#include "iceberg/util/timepoint.h"
 
namespace iceberg {
 
enum class SnapshotRefType {
  kBranch,
  kTag,
};
 
ICEBERG_EXPORT constexpr std::string_view ToString(SnapshotRefType type) noexcept {
  switch (type) {
    case SnapshotRefType::kBranch:
      return "branch";
    case SnapshotRefType::kTag:
      return "tag";
  }
  std::unreachable();
}
ICEBERG_EXPORT constexpr Result<SnapshotRefType> SnapshotRefTypeFromString(
    std::string_view str) noexcept {
  if (str == "branch") return SnapshotRefType::kBranch;
  if (str == "tag") return SnapshotRefType::kTag;
  return InvalidArgument("Invalid snapshot reference type: {}", str);
}
 
struct ICEBERG_EXPORT SnapshotRef {
  static constexpr std::string_view kMainBranch = "main";
 
  struct ICEBERG_EXPORT Branch {
    std::optional<int32_t> min_snapshots_to_keep;
    std::optional<int64_t> max_snapshot_age_ms;
    std::optional<int64_t> max_ref_age_ms;
 
    friend bool operator==(const Branch& lhs, const Branch& rhs) {
      return lhs.Equals(rhs);
    }
 
   private:
    bool Equals(const Branch& other) const;
  };
 
  struct ICEBERG_EXPORT Tag {
    std::optional<int64_t> max_ref_age_ms;
 
    friend bool operator==(const Tag& lhs, const Tag& rhs) { return lhs.Equals(rhs); }
 
   private:
    bool Equals(const Tag& other) const;
  };
 
  int64_t snapshot_id;
  std::variant<Branch, Tag> retention;
 
  SnapshotRefType type() const noexcept;
 
  std::optional<int64_t> max_ref_age_ms() const noexcept;
 
  static Result<std::unique_ptr<SnapshotRef>> MakeBranch(
      int64_t snapshot_id, std::optional<int32_t> min_snapshots_to_keep = std::nullopt,
      std::optional<int64_t> max_snapshot_age_ms = std::nullopt,
      std::optional<int64_t> max_ref_age_ms = std::nullopt);
 
  static Result<std::unique_ptr<SnapshotRef>> MakeTag(
      int64_t snapshot_id, std::optional<int64_t> max_ref_age_ms = std::nullopt);
 
  std::unique_ptr<SnapshotRef> Clone(
      std::optional<int64_t> new_snapshot_id = std::nullopt) const;
 
  Status Validate() const;
 
  friend bool operator==(const SnapshotRef& lhs, const SnapshotRef& rhs) {
    return lhs.Equals(rhs);
  }
 
 private:
  bool Equals(const SnapshotRef& other) const;
};
 
struct ICEBERG_EXPORT SnapshotSummaryFields {
  inline static const std::string kOperation = "operation";
  inline static const std::string kFirstRowId = "first-row-id";
  inline static const std::string kAddedRows = "added-rows";
 
 
  inline static const std::string kAddedDataFiles = "added-data-files";
  inline static const std::string kDeletedDataFiles = "deleted-data-files";
  inline static const std::string kTotalDataFiles = "total-data-files";
  inline static const std::string kAddedDeleteFiles = "added-delete-files";
  inline static const std::string kAddedEqDeleteFiles = "added-equality-delete-files";
  inline static const std::string kRemovedEqDeleteFiles = "removed-equality-delete-files";
  inline static const std::string kAddedPosDeleteFiles = "added-position-delete-files";
  inline static const std::string kRemovedPosDeleteFiles =
      "removed-position-delete-files";
  inline static const std::string kAddedDVs = "added-dvs";
  inline static const std::string kRemovedDVs = "removed-dvs";
  inline static const std::string kRemovedDeleteFiles = "removed-delete-files";
  inline static const std::string kTotalDeleteFiles = "total-delete-files";
  inline static const std::string kAddedRecords = "added-records";
  inline static const std::string kDeletedRecords = "deleted-records";
  inline static const std::string kTotalRecords = "total-records";
  inline static const std::string kAddedFileSize = "added-files-size";
  inline static const std::string kRemovedFileSize = "removed-files-size";
  inline static const std::string kTotalFileSize = "total-files-size";
  inline static const std::string kAddedPosDeletes = "added-position-deletes";
  inline static const std::string kRemovedPosDeletes = "removed-position-deletes";
  inline static const std::string kTotalPosDeletes = "total-position-deletes";
  inline static const std::string kAddedEqDeletes = "added-equality-deletes";
  inline static const std::string kRemovedEqDeletes = "removed-equality-deletes";
  inline static const std::string kTotalEqDeletes = "total-equality-deletes";
  inline static const std::string kDeletedDuplicatedFiles = "deleted-duplicate-files";
  inline static const std::string kChangedPartitionCountProp = "changed-partition-count";
  inline static const std::string kManifestsCreated = "manifests-created";
  inline static const std::string kManifestsKept = "manifests-kept";
  inline static const std::string kManifestsReplaced = "manifests-replaced";
  inline static const std::string kEntriesProcessed = "entries-processed";
  inline static const std::string kChangedPartitionPrefix = "partitions.";
  inline static const std::string kPartitionSummaryProp = "partition-summaries-included";
 
 
  inline static const std::string kWAPId = "wap.id";
  inline static const std::string kPublishedWAPId = "published-wap-id";
  inline static const std::string kSourceSnapshotId = "source-snapshot-id";
  inline static const std::string kEngineName = "engine-name";
  inline static const std::string kEngineVersion = "engine-version";
};
 
class ICEBERG_EXPORT SnapshotSummaryBuilder {
 private:
  class UpdateMetrics {
   public:
    void Clear();
    void AddTo(std::unordered_map<std::string, std::string>& builder) const;
    void AddedFile(const DataFile& file);
    void RemovedFile(const DataFile& file);
    void AddedManifest(const ManifestFile& manifest);
    void Merge(const UpdateMetrics& other);
 
   private:
    int64_t added_size_{0};
    int64_t removed_size_{0};
    int32_t added_files_{0};
    int32_t removed_files_{0};
    int32_t added_eq_delete_files_{0};
    int32_t removed_eq_delete_files_{0};
    int32_t added_pos_delete_files_{0};
    int32_t removed_pos_delete_files_{0};
    int32_t added_dvs_{0};
    int32_t removed_dvs_{0};
    int32_t added_delete_files_{0};
    int32_t removed_delete_files_{0};
    int64_t added_records_{0};
    int64_t deleted_records_{0};
    int64_t added_pos_deletes_{0};
    int64_t removed_pos_deletes_{0};
    int64_t added_eq_deletes_{0};
    int64_t removed_eq_deletes_{0};
    bool trust_size_and_delete_counts_{true};
  };
 
 public:
  SnapshotSummaryBuilder() = default;
 
  void Clear();
 
  void SetPartitionSummaryLimit(int32_t max);
 
  void IncrementDuplicateDeletes(int32_t increment = 1);
 
  Status AddedFile(const PartitionSpec& spec, const DataFile& file);
 
  Status DeletedFile(const PartitionSpec& spec, const DataFile& file);
 
  void AddedManifest(const ManifestFile& manifest);
 
  void Set(const std::string& property, const std::string& value);
 
  void Merge(const SnapshotSummaryBuilder& other);
 
  std::unordered_map<std::string, std::string> Build() const;
 
 private:
  Status UpdatePartitions(const PartitionSpec& spec, const DataFile& file,
                          bool is_addition);
  std::string PartitionSummary(const UpdateMetrics& metrics) const;
 
  std::unordered_map<std::string, std::string> properties_;
  std::unordered_map<std::string, UpdateMetrics> partition_metrics_;
  UpdateMetrics metrics_;
  int32_t max_changed_partitions_for_summaries_{0};
  int64_t deleted_duplicate_files_{0};
  bool trust_partition_metrics_{true};
};
 
struct ICEBERG_EXPORT DataOperation {
  inline static const std::string kAppend = "append";
  inline static const std::string kReplace = "replace";
  inline static const std::string kOverwrite = "overwrite";
  inline static const std::string kDelete = "delete";
};
 
struct ICEBERG_EXPORT Snapshot {
  int64_t snapshot_id;
  std::optional<int64_t> parent_snapshot_id;
  int64_t sequence_number;
  TimePointMs timestamp_ms;
  std::string manifest_list;
  std::unordered_map<std::string, std::string> summary;
  std::optional<int32_t> schema_id;
 
  static Result<std::unique_ptr<Snapshot>> Make(
      int64_t sequence_number, int64_t snapshot_id,
      std::optional<int64_t> parent_snapshot_id, TimePointMs timestamp_ms,
      std::string operation, std::unordered_map<std::string, std::string> summary,
      std::optional<int32_t> schema_id, std::string manifest_list,
      std::optional<int64_t> first_row_id = std::nullopt,
      std::optional<int64_t> added_rows = std::nullopt);
 
  std::optional<std::string_view> Operation() const;
 
  Result<std::optional<int64_t>> FirstRowId() const;
 
  Result<std::optional<int64_t>> AddedRows() const;
 
  friend bool operator==(const Snapshot& lhs, const Snapshot& rhs) {
    return lhs.Equals(rhs);
  }
 
 private:
  bool Equals(const Snapshot& other) const;
};
 
class ICEBERG_EXPORT SnapshotCache {
 public:
  explicit SnapshotCache(const Snapshot* snapshot) : snapshot_(snapshot) {}
 
  const Snapshot& snapshot() const { return *snapshot_; }
 
  Result<std::span<ManifestFile>> Manifests(std::shared_ptr<FileIO> file_io) const;
 
  Result<std::span<ManifestFile>> DataManifests(std::shared_ptr<FileIO> file_io) const;
 
  Result<std::span<ManifestFile>> DeleteManifests(std::shared_ptr<FileIO> file_io) const;
 
 private:
  using ManifestsCache = std::pair<std::vector<ManifestFile>, size_t>;
 
  static Result<ManifestsCache> InitManifestsCache(const Snapshot* snapshot,
                                                   std::shared_ptr<FileIO> file_io);
 
  const Snapshot* snapshot_;
 
  Lazy<InitManifestsCache> manifests_cache_;
};
 
}  // namespace iceberg