manifest_merge_manager.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 <functional>
#include <memory>
#include <string>
#include <unordered_map>
#include <unordered_set>
#include <vector>
 
#include "iceberg/iceberg_export.h"
#include "iceberg/manifest/manifest_list.h"
#include "iceberg/manifest/manifest_writer.h"
#include "iceberg/result.h"
#include "iceberg/type_fwd.h"
 
namespace iceberg {
 
class ICEBERG_EXPORT ManifestMergeManager {
 public:
  using SnapshotIdSupplier = std::function<int64_t()>;
 
  static Result<std::unique_ptr<ManifestMergeManager>> Make(
      ManifestContent content, int64_t target_size_bytes, int32_t min_count_to_merge,
      bool merge_enabled, std::shared_ptr<FileIO> file_io,
      SnapshotIdSupplier snapshot_id_supplier,
      std::function<Status(const std::string&)> delete_file = {});
 
  ManifestMergeManager(const ManifestMergeManager&) = delete;
  ManifestMergeManager& operator=(const ManifestMergeManager&) = delete;
 
  Result<std::vector<ManifestFile>> MergeManifests(
      const std::vector<ManifestFile>& existing_manifests,
      const std::vector<ManifestFile>& new_manifests, const TableMetadata& metadata,
      const ManifestWriterFactory& writer_factory);
 
  int32_t ReplacedManifestsCount() const { return replaced_manifests_count_; }
 
  Status CleanUncommitted(const std::unordered_set<std::string>& committed);
 
 private:
  ManifestMergeManager(ManifestContent content, int64_t target_size_bytes,
                       int32_t min_count_to_merge, bool merge_enabled,
                       std::shared_ptr<FileIO> file_io,
                       SnapshotIdSupplier snapshot_id_supplier,
                       std::function<Status(const std::string&)> delete_file);
 
  struct MergedManifestCache {
    struct Key {
      std::vector<ManifestFile> bin;
      bool operator==(const Key& other) const = default;
    };
 
    struct KeyHash {
      size_t operator()(const Key& key) const;
    };
 
    const ManifestFile* Find(const std::vector<const ManifestFile*>& bin) const;
    void Add(const std::vector<const ManifestFile*>& bin, const ManifestFile& manifest);
    bool empty() const { return entries_.empty(); }
    Result<int32_t> CleanUncommitted(
        const std::unordered_set<std::string>& committed, int64_t snapshot_id,
        const std::function<Status(const std::string&)>& delete_file);
 
   private:
    static Key MakeKey(const std::vector<const ManifestFile*>& bin);
 
    std::unordered_map<Key, ManifestFile, KeyHash> entries_;
  };
 
  Result<std::vector<ManifestFile>> MergeGroup(
      const std::vector<const ManifestFile*>& group, const ManifestFile* first,
      const TableMetadata& metadata, const ManifestWriterFactory& writer_factory);
 
  Result<ManifestFile> FlushBin(const std::vector<const ManifestFile*>& bin,
                                const TableMetadata& metadata,
                                const ManifestWriterFactory& writer_factory);
 
  const ManifestContent manifest_content_;
  const int64_t target_size_bytes_;
  const int32_t min_count_to_merge_;
  const bool merge_enabled_;
  std::shared_ptr<FileIO> file_io_;
  SnapshotIdSupplier snapshot_id_supplier_;
  std::function<Status(const std::string&)> delete_file_;
  int32_t replaced_manifests_count_{0};
  MergedManifestCache merged_manifests_;
};
 
}  // namespace iceberg