merging_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 <cstdint>
#include <memory>
#include <optional>
#include <string>
#include <unordered_map>
#include <unordered_set>
#include <vector>
 
#include "iceberg/delete_file_index.h"
#include "iceberg/iceberg_export.h"
#include "iceberg/manifest/manifest_filter_manager.h"
#include "iceberg/manifest/manifest_merge_manager.h"
#include "iceberg/result.h"
#include "iceberg/type_fwd.h"
#include "iceberg/update/snapshot_update.h"
#include "iceberg/util/data_file_set.h"
 
namespace iceberg {
 
class ICEBERG_EXPORT MergingSnapshotUpdate : public SnapshotUpdate {
 public:
  ~MergingSnapshotUpdate() override = default;
 
 protected:
  // SnapshotUpdate overrides
  Result<std::vector<ManifestFile>> Apply(
      const TableMetadata& metadata_to_update,
      const std::shared_ptr<Snapshot>& snapshot) override;
 
  Status CleanUncommitted(const std::unordered_set<std::string>& committed) override;
 
  std::unordered_map<std::string, std::string> Summary() override;
 
  explicit MergingSnapshotUpdate(std::string table_name,
                                 std::shared_ptr<TransactionContext> ctx);
 
  Status AddDataFile(std::shared_ptr<DataFile> file);
 
  Status AddDeleteFile(std::shared_ptr<DataFile> file);
 
  Status ValidateNewDeleteFile(const TableMetadata& metadata, const DataFile& file);
 
  Status AddDeleteFile(std::shared_ptr<DataFile> file, int64_t data_sequence_number);
 
  Status AddManifest(ManifestFile manifest);
 
  Status DeleteDataFile(std::shared_ptr<DataFile> file);
 
  Status DeleteDeleteFile(std::shared_ptr<DataFile> file);
 
  Status DeleteByPath(std::string_view path);
 
  Status DeleteByRowFilter(std::shared_ptr<Expression> expr);
 
  Status DropPartition(int32_t spec_id, PartitionValues partition);
 
  void FailMissingDeletePaths();
 
  void FailAnyDelete();
 
  void SetNewDataFilesDataSequenceNumber(int64_t sequence_number);
 
  bool HasDataSequenceNumber() const {
    return new_data_files_data_seq_number_.has_value();
  }
 
  void CaseSensitive(bool case_sensitive);
 
  bool IsCaseSensitive() const { return case_sensitive_; }
 
  bool AddsDataFiles() const;
 
  bool AddsDeleteFiles() const;
 
  bool DeletesDataFiles() const;
 
  bool DeletesDeleteFiles() const;
 
  const std::shared_ptr<Expression>& RowFilter() const { return delete_expression_; }
 
  Result<std::shared_ptr<PartitionSpec>> DataSpec() const;
 
  std::vector<std::shared_ptr<DataFile>> AddedDataFiles() const;
 
  static Status ValidateAddedDataFiles(const TableMetadata& metadata,
                                       std::optional<int64_t> starting_snapshot_id,
                                       std::shared_ptr<Expression> filter,
                                       const std::shared_ptr<Snapshot>& parent,
                                       std::shared_ptr<FileIO> io,
                                       bool case_sensitive = true);
 
  static Status ValidateAddedDataFiles(const TableMetadata& metadata,
                                       std::optional<int64_t> starting_snapshot_id,
                                       const PartitionSet& partition_set,
                                       const std::shared_ptr<Snapshot>& parent,
                                       std::shared_ptr<FileIO> io);
 
  static Status ValidateDataFilesExist(
      const TableMetadata& metadata, std::optional<int64_t> starting_snapshot_id,
      const std::unordered_set<std::string>& file_paths, bool skip_deletes,
      std::shared_ptr<Expression> filter, const std::shared_ptr<Snapshot>& parent,
      std::shared_ptr<FileIO> io, bool case_sensitive = true);
 
  Status ValidateNoNewDeletesForDataFiles(const TableMetadata& metadata,
                                          std::optional<int64_t> starting_snapshot_id,
                                          const DataFileSet& replaced_files,
                                          const std::shared_ptr<Snapshot>& parent,
                                          std::shared_ptr<FileIO> io) const {
    const bool ignore_equality_deletes = new_data_files_data_seq_number_.has_value();
    return ValidateNoNewDeletesForDataFiles(metadata, starting_snapshot_id,
                                            replaced_files, parent, io,
                                            ignore_equality_deletes);
  }
 
  static Status ValidateNoNewDeletesForDataFiles(
      const TableMetadata& metadata, std::optional<int64_t> starting_snapshot_id,
      const DataFileSet& replaced_files, const std::shared_ptr<Snapshot>& parent,
      std::shared_ptr<FileIO> io, bool ignore_equality_deletes = false);
 
  static Status ValidateNoNewDeletesForDataFiles(
      const TableMetadata& metadata, std::optional<int64_t> starting_snapshot_id,
      std::shared_ptr<Expression> data_filter, const DataFileSet& replaced_files,
      const std::shared_ptr<Snapshot>& parent, std::shared_ptr<FileIO> io,
      bool case_sensitive = true);
 
  static Status ValidateNoNewDeleteFiles(const TableMetadata& metadata,
                                         std::optional<int64_t> starting_snapshot_id,
                                         std::shared_ptr<Expression> data_filter,
                                         const std::shared_ptr<Snapshot>& parent,
                                         std::shared_ptr<FileIO> io,
                                         bool case_sensitive = true);
 
  static Status ValidateNoNewDeleteFiles(const TableMetadata& metadata,
                                         std::optional<int64_t> starting_snapshot_id,
                                         const PartitionSet& partition_set,
                                         const std::shared_ptr<Snapshot>& parent,
                                         std::shared_ptr<FileIO> io);
 
  static Status ValidateDeletedDataFiles(const TableMetadata& metadata,
                                         std::optional<int64_t> starting_snapshot_id,
                                         std::shared_ptr<Expression> data_filter,
                                         const std::shared_ptr<Snapshot>& parent,
                                         std::shared_ptr<FileIO> io,
                                         bool case_sensitive = true);
 
  static Status ValidateDeletedDataFiles(const TableMetadata& metadata,
                                         std::optional<int64_t> starting_snapshot_id,
                                         const PartitionSet& partition_set,
                                         const std::shared_ptr<Snapshot>& parent,
                                         std::shared_ptr<FileIO> io);
 
  static Result<std::unique_ptr<DeleteFileIndex>> AddedDeleteFiles(
      const TableMetadata& metadata, std::optional<int64_t> starting_snapshot_id,
      std::shared_ptr<Expression> data_filter,
      std::shared_ptr<PartitionSet> partition_set,
      const std::shared_ptr<Snapshot>& parent, std::shared_ptr<FileIO> io,
      bool case_sensitive = true);
 
  static Status ValidateAddedDVs(
      const TableMetadata& metadata, std::optional<int64_t> starting_snapshot_id,
      std::shared_ptr<Expression> conflict_filter,
      const std::unordered_set<std::string>& referenced_data_files,
      const std::shared_ptr<Snapshot>& parent, std::shared_ptr<FileIO> io,
      bool case_sensitive = true);
 
  Status ValidateAddedDVs(const TableMetadata& metadata,
                          std::optional<int64_t> starting_snapshot_id,
                          std::shared_ptr<Expression> conflict_filter,
                          const std::shared_ptr<Snapshot>& parent,
                          std::shared_ptr<FileIO> io) const;
 
 private:
  struct PendingDeleteFile {
    std::shared_ptr<DataFile> file;
    std::optional<int64_t> data_sequence_number;
  };
 
  struct PendingDeleteFilesByReferencedFile {
    struct Entry {
      std::string referenced_file;
      std::vector<PendingDeleteFile> files;
    };
 
    void Add(std::string referenced_file, PendingDeleteFile file);
    bool empty() const { return entries_.empty(); }
    size_t size() const { return entries_.size(); }
    const std::vector<Entry>& entries() const { return entries_; }
 
   private:
    std::vector<Entry> entries_;
    std::unordered_map<std::string, size_t> index_by_referenced_file_;
  };
 
  ManifestWriterFactory MakeWriterFactory(const std::shared_ptr<Schema>& schema);
 
  Result<ManifestFile> CopyManifest(const ManifestFile& manifest, bool update_summary);
 
  Status AddDeleteFile(std::shared_ptr<DataFile> file,
                       std::optional<int64_t> data_sequence_number);
 
  Status ManagersReady() const;
 
  void SetSummaryProperty(const std::string& property, const std::string& value) override;
 
  Result<std::vector<PendingDeleteFile>> MergeDVs();
 
  Result<std::vector<ManifestFile>> WriteNewDataManifests();
 
  Result<std::vector<ManifestFile>> WriteNewDeleteManifests();
 
  Status CleanUncommittedAppends(const std::unordered_set<std::string>& committed);
 
  Status DeleteUncommitted(std::vector<ManifestFile>& manifests,
                           const std::unordered_set<std::string>& committed, bool clear);
 
  // Used for commit event notifications and diagnostic log messages.
  std::string table_name_;
  std::shared_ptr<Expression> delete_expression_;
  bool case_sensitive_ = true;
 
  // Stable sub-builders for added files — accumulated across retries and merged
  // into summary_builder_ at the start of each Apply() call.
  SnapshotSummaryBuilder added_data_files_summary_;
  SnapshotSummaryBuilder added_delete_files_summary_;
  SnapshotSummaryBuilder appended_manifests_summary_;
  std::unordered_map<std::string, std::string> custom_summary_properties_;
 
  std::unique_ptr<ManifestFilterManager> data_filter_manager_;
  std::unique_ptr<ManifestFilterManager> delete_filter_manager_;
  std::unique_ptr<ManifestMergeManager> data_merge_manager_;
  std::unique_ptr<ManifestMergeManager> delete_merge_manager_;
 
  std::unordered_map<int32_t, DataFileSet> new_data_files_by_spec_;
  std::vector<PendingDeleteFile> v2_deletes_;
  PendingDeleteFilesByReferencedFile dvs_by_referenced_file_;
  std::optional<int64_t> new_data_files_data_seq_number_;
 
  // Manifests passed via AddManifest(): inherit path (no copy needed) and
  // rewrite path (must be copied with the current snapshot ID).
  std::vector<ManifestFile> append_manifests_;
  // Original manifests kept to recreate copied manifests after retry cleanup.
  std::vector<ManifestFile> append_manifests_to_copy_;
  std::vector<ManifestFile> rewritten_append_manifests_;
 
  // Set to true when new files are staged after the cache was populated, so the
  // cache is invalidated and re-written on the next Apply() call (commit retry).
  bool has_new_data_files_ = false;
  bool has_new_delete_files_ = false;
 
  std::vector<ManifestFile> cached_new_data_manifests_;
  std::vector<ManifestFile> cached_new_delete_manifests_;
  std::vector<PendingDeleteFile> merged_dvs_;
  int32_t dv_merge_attempt_ = 0;
};
 
}  // namespace iceberg