snapshot_update.h

Go to the documentation of this file.

/*
 * 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 <atomic>
#include <functional>
#include <memory>
#include <mutex>
#include <optional>
#include <span>
#include <string>
#include <unordered_map>
#include <unordered_set>
#include <vector>
 
#include "iceberg/compat/compiler.h"
#include "iceberg/iceberg_export.h"
#include "iceberg/result.h"
#include "iceberg/snapshot.h"
#include "iceberg/type_fwd.h"
#include "iceberg/update/pending_update.h"
#include "iceberg/util/executor.h"
 
namespace iceberg {
 
class ICEBERG_EXPORT SnapshotUpdate : public PendingUpdate {
 public:
  struct ApplyResult {
    // Null when applying an unconfigured update to an empty table.
    std::shared_ptr<Snapshot> snapshot;
    std::string target_branch;
    bool stage_only = false;
  };
 
  ~SnapshotUpdate() override;
 
  Kind kind() const override { return Kind::kUpdateSnapshot; }
  bool IsRetryable() const override { return true; }
  Status Commit() override;
 
#if ICEBERG_CXX20_COMPAT
  SnapshotUpdate& ReportWith(std::shared_ptr<MetricsReporter> reporter) {
    SetReporter(std::move(reporter));
    return *this;
  }
#else
  auto& ReportWith(this auto& self, std::shared_ptr<MetricsReporter> reporter) {
    static_cast<SnapshotUpdate&>(self).SetReporter(std::move(reporter));
    return self;
  }
#endif
 
#if ICEBERG_CXX20_COMPAT
  SnapshotUpdate& DeleteWith(std::function<Status(const std::string&)> delete_func) {
    SetDeleteFunction(std::move(delete_func));
    return *this;
  }
#else
  auto& DeleteWith(this auto& self,
                   std::function<Status(const std::string&)> delete_func) {
    static_cast<SnapshotUpdate&>(self).SetDeleteFunction(std::move(delete_func));
    return self;
  }
#endif
 
#if ICEBERG_CXX20_COMPAT
  SnapshotUpdate& StageOnly() {
    SetStageOnly();
    return *this;
  }
#else
  auto& StageOnly(this auto& self) {
    static_cast<SnapshotUpdate&>(self).SetStageOnly();
    return self;
  }
#endif
 
#if ICEBERG_CXX20_COMPAT
  SnapshotUpdate& ScanManifestsWith(Executor& executor) {
    SetPlanExecutor(executor);
    return *this;
  }
#else
  auto& ScanManifestsWith(this auto& self, Executor& executor) {
    static_cast<SnapshotUpdate&>(self).SetPlanExecutor(executor);
    return self;
  }
#endif
 
#if ICEBERG_CXX20_COMPAT
  SnapshotUpdate& ToBranch(const std::string& branch) {
    SetTargetBranch(branch);
    return *this;
  }
#else
  auto& ToBranch(this auto& self, const std::string& branch) {
    static_cast<SnapshotUpdate&>(self).SetTargetBranch(branch);
    return self;
  }
#endif
 
#if ICEBERG_CXX20_COMPAT
  SnapshotUpdate& Set(const std::string& property, const std::string& value) {
    SetSummaryProperty(property, value);
    return *this;
  }
#else
  auto& Set(this auto& self, const std::string& property, const std::string& value) {
    static_cast<SnapshotUpdate&>(self).SetSummaryProperty(property, value);
    return self;
  }
#endif
 
#if ICEBERG_CXX20_COMPAT
  SnapshotUpdate& WriteManifestsWith(Executor& executor, int32_t parallelism) {
    SetManifestWriter(executor, parallelism);
    return *this;
  }
#else
  auto& WriteManifestsWith(this auto& self, Executor& executor, int32_t parallelism) {
    static_cast<SnapshotUpdate&>(self).SetManifestWriter(executor, parallelism);
    return self;
  }
#endif
 
 protected:
  friend class Transaction;
 
  virtual Result<ApplyResult> Apply();
  Status Finalize(const TableMetadata& committed) override;
  Status CleanStaged() override;
 
  struct ContentFileWithSequenceNumber {
    std::shared_ptr<DataFile> file;
    std::optional<int64_t> data_sequence_number;
  };
 
  explicit SnapshotUpdate(std::shared_ptr<TransactionContext> ctx);
 
  Result<std::vector<ManifestFile>> WriteDataManifests(
      std::span<const std::shared_ptr<DataFile>> files,
      const std::shared_ptr<PartitionSpec>& spec,
      std::optional<int64_t> data_sequence_number = std::nullopt);
 
  Result<std::vector<ManifestFile>> WriteDeleteManifests(
      std::span<const ContentFileWithSequenceNumber> files,
      const std::shared_ptr<PartitionSpec>& spec);
 
  const std::string& target_branch() const { return target_branch_; }
  bool can_inherit_snapshot_id() const { return can_inherit_snapshot_id_; }
  const std::string& commit_uuid() const { return commit_uuid_; }
  int32_t manifest_count() const {
    return manifest_count_.load(std::memory_order_relaxed);
  }
  int32_t attempt() const { return attempt_.load(std::memory_order_relaxed); }
  int64_t target_manifest_size_bytes() const { return target_manifest_size_bytes_; }
 
  virtual Status CleanUncommitted(const std::unordered_set<std::string>& committed) = 0;
 
  virtual std::string operation() = 0;
 
  virtual Status Validate(const TableMetadata& current_metadata,
                          const std::shared_ptr<Snapshot>& snapshot) {
    return {};
  };
 
  virtual Result<std::vector<ManifestFile>> Apply(
      const TableMetadata& metadata_to_update,
      const std::shared_ptr<Snapshot>& snapshot) = 0;
 
  virtual std::unordered_map<std::string, std::string> Summary() = 0;
 
  virtual void SetSummaryProperty(const std::string& property, const std::string& value);
 
  int64_t SnapshotId();
 
  Status DeleteFile(const std::string& path) noexcept;
  void RegisterStagedFile(const std::string& path);
  void UnregisterStagedFile(const std::string& path);
 
  std::string ManifestPath();
  std::string ManifestListPath();
  OptionalExecutor plan_executor() const { return plan_executor_; }
  SnapshotSummaryBuilder& summary_builder() { return summary_; }
  SnapshotSummaryBuilder BuildManifestCountSummary(
      std::span<const ManifestFile> manifests, int32_t replaced_manifests_count);
 
 private:
  Result<std::unordered_map<std::string, std::string>> ComputeSummary(
      const TableMetadata& previous);
 
  Status ReportCommit() const;
 
 protected:
  SnapshotSummaryBuilder summary_;
 
 private:
  void SetReporter(std::shared_ptr<MetricsReporter> reporter) {
    reporter_ = std::move(reporter);
  }
 
  void SetDeleteFunction(std::function<Status(const std::string&)> delete_func) {
    if (delete_func_) {
      AddError(ErrorKind::kInvalidArgument, "Cannot set delete callback more than once");
      return;
    }
    delete_func_ = std::move(delete_func);
  }
 
  void SetStageOnly() { stage_only_ = true; }
 
  void SetPlanExecutor(Executor& executor) { plan_executor_ = std::ref(executor); }
 
  void SetTargetBranch(const std::string& branch) {
    if (branch.empty()) [[unlikely]] {
      AddError(ErrorKind::kInvalidArgument, "Branch name cannot be empty");
      return;
    }
 
    if (auto ref_it = base().refs.find(branch); ref_it != base().refs.end()) {
      if (ref_it->second->type() != SnapshotRefType::kBranch) {
        AddError(ErrorKind::kInvalidArgument,
                 "{} is a tag, not a branch. Tags cannot be targets for "
                 "producing snapshots",
                 branch);
        return;
      }
    }
 
    target_branch_ = branch;
  }
 
  void SetManifestWriter(Executor& executor, int32_t parallelism) {
    if (parallelism <= 0) [[unlikely]] {
      AddError(ErrorKind::kInvalidArgument,
               "Manifest write parallelism must be greater than 0, but was: {}",
               parallelism);
      return;
    }
 
    write_manifest_executor_ = std::ref(executor);
    write_manifest_parallelism_ = parallelism;
  }
 
  const bool can_inherit_snapshot_id_{true};
  const std::string commit_uuid_;
  OptionalExecutor write_manifest_executor_;
  int32_t write_manifest_parallelism_{1};
  std::atomic<int32_t> manifest_count_{0};
  std::atomic<int32_t> attempt_{0};
  // All paths are registered before writes, including partial/failed writes.
  std::mutex staging_mutex_;
  std::unordered_set<std::string> staged_files_;
  const int64_t target_manifest_size_bytes_;
  std::optional<int64_t> snapshot_id_;
  OptionalExecutor plan_executor_;
  bool stage_only_{false};
  std::function<Status(const std::string&)> delete_func_;
  std::string target_branch_{SnapshotRef::kMainBranch};
  std::shared_ptr<Snapshot> staged_snapshot_;
  std::unique_ptr<CommitMetrics> commit_metrics_;
  std::shared_ptr<MetricsReporter> reporter_;
};
 
}  // namespace iceberg