/
GraphTreeHeap
/
museumdb2
Обзор
Документация
Войти
/
GraphTreeHeap
/
museumdb2
Код
Запросы
0
Задачи
Вики
Пакеты
0
Релизы
0
Аналитика
Безопасность
main
db/compaction/compaction_iterator.h
671 строка
26 KB
Peter Dillinger
Fix check-sources.sh non-ASCII check and remove non-ASCII from sources (#14729)
12 май 2026, 03:02
12 май 2026, 03:02
795f3bd
Код
Авторство
О чём код?
// Copyright (c) 2011-present, Facebook, Inc. All rights reserved. // This source code is licensed under both the GPLv2 (found in the // COPYING file in the root directory) and Apache 2.0 License // (found in the LICENSE.Apache file in the root directory). #pragma once #include <algorithm> #include <cinttypes> #include <deque> #include <memory> #include <optional> #include <string> #include <vector> #include "db/blob/blob_index.h" #include "db/compaction/compaction.h" #include "db/compaction/compaction_iteration_stats.h" #include "db/merge_helper.h" #include "db/pinned_iterators_manager.h" #include "db/range_del_aggregator.h" #include "db/snapshot_checker.h" #include "options/cf_options.h" #include "rocksdb/compaction_filter.h" #include "rocksdb/slice.h" namespace ROCKSDB_NAMESPACE { class BlobFileBuilder; class BlobFileCache; class BlobFetcher; class Version; class PrefetchBufferCollection; class FilePrefetchBuffer; class Version; // Internal implementation of WideColumnBlobResolver for compaction. // This class enables lazy loading of blob column values during compaction // filter processing. class CompactionBlobResolver : public WideColumnBlobResolver { public: CompactionBlobResolver() : columns_(nullptr), blob_columns_(nullptr), blob_fetcher_(nullptr), prefetch_buffers_(nullptr), iter_stats_(nullptr) {} // Set the fixed context (blob fetcher, prefetch buffers, stats) once. void Init(BlobFetcher* blob_fetcher, PrefetchBufferCollection* prefetch_buffers, CompactionIterationStats* iter_stats); Status ResolveColumn(size_t column_index, Slice* resolved_value) override; bool IsBlobColumn(size_t column_index) const override; size_t NumColumns() const override; Status resolve_status() const { if (!resolve_error_.has_value()) { return Status::OK(); } return *resolve_error_; } // Reset the resolver for a new entity. Clears resolved cache. void Reset(const Slice& user_key, const std::vector<WideColumn>* columns, const std::vector<std::pair<size_t, BlobIndex>>* blob_columns, bool track_resolve_error = false); private: Slice user_key_; const std::vector<WideColumn>* columns_; const std::vector<std::pair<size_t, BlobIndex>>* blob_columns_; BlobFetcher* blob_fetcher_; PrefetchBufferCollection* prefetch_buffers_; CompactionIterationStats* iter_stats_; bool track_resolve_error_ = false; // Cache for resolved blob values to avoid re-fetching. Uses a vector of // (column_index, PinnableSlice) pairs -- typical entities have few blob // columns (<5), making linear scan cheaper than hash map overhead. std::vector<std::pair<size_t, std::unique_ptr<PinnableSlice>>> resolved_cache_; // Sticky resolver error for the current entity. This is enabled only for the // lazy FilterV4 path so compaction can still fail even if the filter // notices the error and returns kKeep. std::optional<Status> resolve_error_; }; // A wrapper of internal iterator whose purpose is to count how // many entries there are in the iterator. class SequenceIterWrapper : public InternalIterator { public: SequenceIterWrapper(InternalIterator* iter, const Comparator* cmp, bool need_count_entries) : icmp_(cmp), inner_iter_(iter), need_count_entries_(need_count_entries) {} bool Valid() const override { return inner_iter_->Valid(); } Status status() const override { return inner_iter_->status(); } void Next() override { if (!inner_iter_->IsDeleteRangeSentinelKey()) { num_itered_++; } inner_iter_->Next(); } void Seek(const Slice& target) override { if (!need_count_entries_) { has_num_itered_ = false; inner_iter_->Seek(target); } else { // Need to count total number of entries, // so we do Next() rather than Seek(). while (inner_iter_->Valid() && icmp_.Compare(inner_iter_->key(), target) < 0) { Next(); } } } Slice key() const override { return inner_iter_->key(); } Slice value() const override { return inner_iter_->value(); } // Unused InternalIterator methods void SeekToFirst() override { assert(false); } void Prev() override { assert(false); } void SeekForPrev(const Slice& /* target */) override { assert(false); } void SeekToLast() override { assert(false); } uint64_t NumItered() const { return num_itered_; } bool HasNumItered() const { return has_num_itered_; } bool IsDeleteRangeSentinelKey() const override { assert(Valid()); return inner_iter_->IsDeleteRangeSentinelKey(); } private: InternalKeyComparator icmp_; InternalIterator* inner_iter_; // not owned uint64_t num_itered_ = 0; bool need_count_entries_; bool has_num_itered_ = true; }; class CompactionIterator { public: // A wrapper around Compaction. Has a much smaller interface, only what // CompactionIterator uses. Tests can override it. class CompactionProxy { public: virtual ~CompactionProxy() = default; virtual int level() const = 0; virtual bool KeyNotExistsBeyondOutputLevel( const Slice& user_key, std::vector<size_t>* level_ptrs) const = 0; virtual bool bottommost_level() const = 0; virtual int number_levels() const = 0; // Result includes timestamp if user-defined timestamp is enabled. virtual Slice GetLargestUserKey() const = 0; virtual bool allow_ingest_behind() const = 0; virtual bool allow_mmap_reads() const = 0; virtual bool enable_blob_garbage_collection() const = 0; virtual double blob_garbage_collection_age_cutoff() const = 0; virtual uint64_t blob_compaction_readahead_size() const = 0; virtual const Version* input_version() const = 0; virtual bool DoesInputReferenceBlobFiles() const = 0; virtual const Compaction* real_compaction() const = 0; virtual bool SupportsPerKeyPlacement() const = 0; }; class RealCompaction : public CompactionProxy { public: explicit RealCompaction(const Compaction* compaction) : compaction_(compaction) { assert(compaction_); } int level() const override { return compaction_->level(); } bool KeyNotExistsBeyondOutputLevel( const Slice& user_key, std::vector<size_t>* level_ptrs) const override { return compaction_->KeyNotExistsBeyondOutputLevel(user_key, level_ptrs); } bool bottommost_level() const override { return compaction_->bottommost_level(); } int number_levels() const override { return compaction_->number_levels(); } // Result includes timestamp if user-defined timestamp is enabled. Slice GetLargestUserKey() const override { return compaction_->GetLargestUserKey(); } bool allow_ingest_behind() const override { return compaction_->immutable_options().cf_allow_ingest_behind || compaction_->immutable_options().allow_ingest_behind; } bool allow_mmap_reads() const override { return compaction_->immutable_options().allow_mmap_reads; } bool enable_blob_garbage_collection() const override { return compaction_->enable_blob_garbage_collection(); } double blob_garbage_collection_age_cutoff() const override { return compaction_->blob_garbage_collection_age_cutoff(); } uint64_t blob_compaction_readahead_size() const override { return compaction_->mutable_cf_options().blob_compaction_readahead_size; } const Version* input_version() const override { return compaction_->input_version(); } bool DoesInputReferenceBlobFiles() const override { return compaction_->DoesInputReferenceBlobFiles(); } const Compaction* real_compaction() const override { return compaction_; } bool SupportsPerKeyPlacement() const override { return compaction_->SupportsPerKeyPlacement(); } private: const Compaction* compaction_; }; // @param must_count_input_entries Controls input entry counting accuracy vs // performance: // - If true: `NumInputEntryScanned()` always returns the exact count of // input keys // scanned. The iterator will use sequential `Next()` calls instead of // `Seek()` to maintain count accuracy as `Seek()` will not count the // skipped input entries, which is slower but guarantees correctness. // - If false: `NumInputEntryScanned()` returns the count only if no // `Seek()` operations // were performed on the input iterator. When compaction filters request // skipping ranges of keys or other optimizations trigger seek operations, // the count becomes unreliable. Always call `HasNumInputEntryScanned()` // first to verify if the count is accurate before using // `NumInputEntryScanned()`. CompactionIterator( InternalIterator* input, const Comparator* cmp, MergeHelper* merge_helper, SequenceNumber last_sequence, std::vector<SequenceNumber>* snapshots, SequenceNumber earliest_snapshot, SequenceNumber earliest_write_conflict_snapshot, SequenceNumber job_snapshot, const SnapshotChecker* snapshot_checker, Env* env, bool report_detailed_time, CompactionRangeDelAggregator* range_del_agg, BlobFileBuilder* blob_file_builder, bool allow_data_in_errors, bool enforce_single_del_contracts, const std::atomic<bool>& manual_compaction_canceled, bool must_count_input_entries, const Compaction* compaction = nullptr, const CompactionFilter* compaction_filter = nullptr, const std::atomic<bool>* shutting_down = nullptr, const std::shared_ptr<Logger> info_log = nullptr, const std::string* full_history_ts_low = nullptr, std::optional<SequenceNumber> preserve_seqno_min = {}, const Version* input_version = nullptr, Env::IOActivity blob_read_io_activity = Env::IOActivity::kCompaction); // Constructor with custom CompactionProxy, used for tests. CompactionIterator( InternalIterator* input, const Comparator* cmp, MergeHelper* merge_helper, SequenceNumber last_sequence, std::vector<SequenceNumber>* snapshots, SequenceNumber earliest_snapshot, SequenceNumber earliest_write_conflict_snapshot, SequenceNumber job_snapshot, const SnapshotChecker* snapshot_checker, Env* env, bool report_detailed_time, CompactionRangeDelAggregator* range_del_agg, BlobFileBuilder* blob_file_builder, bool allow_data_in_errors, bool enforce_single_del_contracts, const std::atomic<bool>& manual_compaction_canceled, std::unique_ptr<CompactionProxy> compaction, bool must_count_input_entries, const CompactionFilter* compaction_filter = nullptr, const std::atomic<bool>* shutting_down = nullptr, const std::shared_ptr<Logger> info_log = nullptr, const std::string* full_history_ts_low = nullptr, std::optional<SequenceNumber> preserve_seqno_min = {}, const Version* input_version = nullptr, Env::IOActivity blob_read_io_activity = Env::IOActivity::kCompaction); ~CompactionIterator(); void ResetRecordCounts(); // Non-compaction table-file creation (e.g., flush/recovery) can still need // blob resolution when direct-write wide entities already contain blob // references. Plumb a fetcher explicitly in those cases. void SetBlobFetcher(const Version* version, BlobFileCache* blob_file_cache, Env::IOActivity io_activity, bool allow_write_path_fallback); // Seek to the beginning of the compaction iterator output. // // REQUIRED: Call only once. void SeekToFirst(); // Produces the next record in the compaction. // // REQUIRED: SeekToFirst() has been called. void Next(); // Getters const Slice& key() const { return key_; } const Slice& value() const { return value_; } const Status& status() const { return status_; } const ParsedInternalKey& ikey() const { return ikey_; } inline bool Valid() const { return validity_info_.IsValid(); } const Slice& user_key() const { if (UNLIKELY(is_range_del_)) { return ikey_.user_key; } return current_user_key_; } const CompactionIterationStats& iter_stats() const { return iter_stats_; } bool HasNumInputEntryScanned() const { return input_.HasNumItered(); } // This method should only be used when `HasNumInputEntryScanned()` returns // true, unless `must_count_input_entries=true` was specified during iterator // creation (which ensures the count is always accurate). uint64_t NumInputEntryScanned() const { return input_.NumItered(); } // Returns true if the current valid key was already scanned/counted during // a lookahead operation in a previous iteration. // // REQUIRED: Valid() must be true bool IsCurrentKeyAlreadyScanned() const { assert(Valid()); return at_next_ || merge_out_iter_.Valid(); } Status InputStatus() const { return input_.status(); } bool IsDeleteRangeSentinelKey() const { return is_range_del_; } private: // Processes the input stream to find the next output void NextFromInput(); // Do final preparations before presenting the output to the callee. void PrepareOutput(); // Passes the output value to the blob file builder (if any), and replaces it // with the corresponding blob reference if it has been actually written to a // blob file (i.e. if it passed the value size check). Returns true if the // value got extracted to a blob file, false otherwise. bool ExtractLargeValueIfNeededImpl(); // Extracts large values as described above, and updates the internal key's // type to kTypeBlobIndex if the value got extracted. Should only be called // for regular values (kTypeValue). void ExtractLargeValueIfNeeded(); // Extracts large column values from wide column entities to blob files. // For each column whose value size >= min_blob_size, the value is written // to a blob file and replaced with a blob index reference. Should only be // called for wide column entities (kTypeWideColumnEntity). void ExtractLargeColumnValuesIfNeeded(); // Relocates valid blobs residing in the oldest blob files if garbage // collection is enabled. Relocated blobs are written to new blob files or // inlined in the LSM tree depending on the current settings (i.e. // enable_blob_files and min_blob_size). Should only be called for blob // references (kTypeBlobIndex). // // Note: the stacked BlobDB implementation's compaction filter based GC // algorithm is also called from here. void GarbageCollectBlobIfNeeded(); // Garbage collects blob references within wide column entities. For entities // that contain blob column references, this method checks if any of the // referenced blob files need garbage collection. If so, the blob values are // fetched and either relocated to new blob files or inlined based on // current settings. Should only be called for wide column entities // (kTypeWideColumnEntity) that have blob columns. void GarbageCollectEntityBlobsIfNeeded(); // Helper for GarbageCollectEntityBlobsIfNeeded: Fetches blob values that need // garbage collection based on the cutoff file number. // Returns true if any blobs were fetched, false otherwise. // On error, sets status_ and returns false. bool FetchBlobsNeedingGC( const std::vector<WideColumn>& columns, const std::vector<std::pair<size_t, BlobIndex>>& blob_columns, std::vector<std::pair<size_t, PinnableSlice>>* fetched_blob_values); // Helper for GarbageCollectEntityBlobsIfNeeded: Relocates fetched blob // values to new blob files if blob_file_builder_ is available. // Returns the new blob column indices. On error, sets status_ and returns // empty. std::vector<std::pair<size_t, BlobIndex>> RelocateBlobValues( const std::vector<std::pair<size_t, PinnableSlice>>& fetched_blob_values); // Helper for GarbageCollectEntityBlobsIfNeeded: Serializes the entity after // garbage collection, combining original columns with fetched values and // new blob indices. On error, sets status_. void SerializeEntityAfterGC( const std::vector<WideColumn>& columns, const std::vector<std::pair<size_t, BlobIndex>>& original_blob_columns, const std::vector<std::pair<size_t, PinnableSlice>>& fetched_blob_values, const std::vector<std::pair<size_t, BlobIndex>>& new_blob_columns); // Invoke compaction filter if needed. // Return true on success, false on failures (e.g.: kIOError). bool InvokeFilterIfNeeded(bool* need_skip, Slice* skip_until); // Given a sequence number, return the sequence number of the // earliest snapshot that this sequence number is visible in. // The snapshots themselves are arranged in ascending order of // sequence numbers. // Employ a sequential search because the total number of // snapshots are typically small. inline SequenceNumber findEarliestVisibleSnapshot( SequenceNumber in, SequenceNumber* prev_snapshot); inline bool KeyCommitted(SequenceNumber sequence) { return snapshot_checker_ == nullptr || snapshot_checker_->CheckInSnapshot(sequence, job_snapshot_) == SnapshotCheckerResult::kInSnapshot; } bool DefinitelyInSnapshot(SequenceNumber seq, SequenceNumber snapshot); bool DefinitelyNotInSnapshot(SequenceNumber seq, SequenceNumber snapshot); // Extract user-defined timestamp from user key if possible and compare it // with *full_history_ts_low_ if applicable. inline void UpdateTimestampAndCompareWithFullHistoryLow() { if (!timestamp_size_) { return; } Slice ts = ExtractTimestampFromUserKey(ikey_.user_key, timestamp_size_); curr_ts_.assign(ts.data(), ts.size()); if (full_history_ts_low_) { cmp_with_history_ts_low_ = cmp_->CompareTimestamp(ts, *full_history_ts_low_); } } static uint64_t ComputeBlobGarbageCollectionCutoffFileNumber( const CompactionProxy* compaction); static std::unique_ptr<BlobFetcher> CreateBlobFetcherIfNeeded( const CompactionProxy* compaction, const Version* input_version, Env::IOActivity blob_read_io_activity); static std::unique_ptr<PrefetchBufferCollection> CreatePrefetchBufferCollectionIfNeeded(const CompactionProxy* compaction); SequenceIterWrapper input_; const Comparator* cmp_; MergeHelper* merge_helper_; const std::vector<SequenceNumber>* snapshots_; // List of snapshots released during compaction. // findEarliestVisibleSnapshot() find them out from return of // snapshot_checker, and make sure they will not be returned as // earliest visible snapshot of an older value. // See WritePreparedTransactionTest::ReleaseSnapshotDuringCompaction3. std::unordered_set<SequenceNumber> released_snapshots_; const SequenceNumber earliest_write_conflict_snapshot_; const SequenceNumber job_snapshot_; const SnapshotChecker* const snapshot_checker_; Env* env_; SystemClock* clock_; const bool report_detailed_time_; CompactionRangeDelAggregator* range_del_agg_; BlobFileBuilder* blob_file_builder_; std::unique_ptr<CompactionProxy> compaction_; const CompactionFilter* compaction_filter_; const bool filter_supports_v4_; const std::atomic<bool>* shutting_down_; const std::atomic<bool>& manual_compaction_canceled_; const bool bottommost_level_; const bool visible_at_tip_; const SequenceNumber earliest_snapshot_; std::shared_ptr<Logger> info_log_; const bool allow_data_in_errors_; const bool enforce_single_del_contracts_; // Comes from comparator. const size_t timestamp_size_; // Lower bound timestamp to retain full history in terms of user-defined // timestamp. If a key's timestamp is older than full_history_ts_low_, then // the key *may* be eligible for garbage collection (GC). The skipping logic // is in `NextFromInput()` and `PrepareOutput()`. // If nullptr, NO GC will be performed and all history will be preserved. const std::string* const full_history_ts_low_; // State // enum ValidContext : uint8_t { kMerge1 = 0, kMerge2 = 1, kParseKeyError = 2, kCurrentKeyUncommitted = 3, kKeepSDAndClearPut = 4, kKeepTsHistory = 5, kKeepSDForConflictCheck = 6, kKeepSDForSnapshot = 7, kKeepSD = 8, kKeepDel = 9, kNewUserKey = 10, kRangeDeletion = 11, kSwapPreferredSeqno = 12, }; struct ValidityInfo { inline bool IsValid() const { return rep & 1; } ValidContext GetContext() const { return static_cast<ValidContext>(rep >> 1); } inline void SetValid(uint8_t ctx) { rep = (ctx << 1) | 1; } inline void Invalidate() { rep = 0; } uint8_t rep{0}; } validity_info_; // Points to a copy of the current compaction iterator output (current_key_) // if valid. Slice key_; // Points to the value in the underlying iterator that corresponds to the // current output. Slice value_; // The status is OK unless compaction iterator encounters a merge operand // while not having a merge operator defined. Status status_; // Stores the user key, sequence number and type of the current compaction // iterator output (or current key in the underlying iterator during // NextFromInput()). ParsedInternalKey ikey_; // Stores whether current_user_key_ is valid. If so, current_user_key_ // stores the user key of the last key seen by the iterator. // If false, treat the next key to read as a new user key. bool has_current_user_key_ = false; // If false, the iterator holds a copy of the current compaction iterator // output (or current key in the underlying iterator during NextFromInput()). bool at_next_ = false; // A copy of the current internal key. IterKey current_key_; Slice current_user_key_; std::string curr_ts_; SequenceNumber current_user_key_sequence_; SequenceNumber current_user_key_snapshot_; // True if the iterator has already returned a record for the current key. bool has_outputted_key_ = false; // Truncate the value of the next key and output it without applying any // compaction rules. This is an optimization for outputting a put after // a single delete. See more in `NextFromInput()` under Optimization 3. bool clear_and_output_next_key_ = false; MergeOutputIterator merge_out_iter_; Status merge_until_status_; // PinnedIteratorsManager used to pin input_ Iterator blocks while reading // merge operands and then releasing them after consuming them. PinnedIteratorsManager pinned_iters_mgr_; uint64_t blob_garbage_collection_cutoff_file_number_; std::unique_ptr<BlobFetcher> blob_fetcher_; std::unique_ptr<PrefetchBufferCollection> prefetch_buffers_; std::string blob_index_; PinnableSlice blob_value_; std::string compaction_filter_value_; InternalKey compaction_filter_skip_until_; // Buffer for storing rewritten entity with blob references after // extracting large column values to blob files. std::string rewritten_entity_; // Cached deserialized columns for the current wide-column entity, reused // across iterations to avoid per-key heap allocations. This is the full // sorted column set. For blob-backed columns, value() initially contains the // serialized BlobIndex bytes from the entity. std::vector<WideColumn> entity_columns_; // Side list for the blob-backed subset of entity_columns_. Each pair is // (column_index_in_entity_columns_, decoded_blob_index), so callers can tell // which entries in entity_columns_ are blob references without reparsing. std::vector<std::pair<size_t, BlobIndex>> entity_blob_columns_; // Temporary Slice-based view over entity_columns_ used when reserializing an // entity after blob extraction, to avoid copying names/values into strings. WideColumns entity_wide_columns_; // Reusable storage for existing_columns in InvokeFilterIfNeeded, // avoiding per-key heap allocations. WideColumns filter_existing_columns_; // Reusable blob resolver for compaction filter, avoiding per-key heap // allocation. Init() is called once; Reset() is called per entity. CompactionBlobResolver blob_resolver_; // True when entity_columns_/entity_blob_columns_ already describe the // current input record (e.g., because InvokeFilterIfNeeded() deserialized // it) and PrepareOutput() can skip redundant work. Reset for each candidate // input record and whenever the filter rewrites the entity value. bool entity_deserialized_{false}; // "level_ptrs" holds indices that remember which file of an associated // level we were last checking during the last call to compaction-> // KeyNotExistsBeyondOutputLevel(). This allows future calls to the function // to pick off where it left off since each subcompaction's key range is // increasing so a later call to the function must be looking for a key that // is in or beyond the last file checked during the previous call std::vector<size_t> level_ptrs_; CompactionIterationStats iter_stats_; // Used to avoid purging uncommitted values. The application can specify // uncommitted values by providing a SnapshotChecker object. bool current_key_committed_; // Saved result of ucmp->CompareTimestamp(current_ts_, *full_history_ts_low_) int cmp_with_history_ts_low_; const int level_; // True if the previous internal key (same user key)'s sequence number has // just been zeroed out during bottommost compaction. bool last_key_seq_zeroed_{false}; // Max seqno that can be zeroed out at last level (various reasons) const SequenceNumber preserve_seqno_after_ = kMaxSequenceNumber; void AdvanceInputIter() { input_.Next(); } void SkipUntil(const Slice& skip_until) { input_.Seek(skip_until); } bool IsShuttingDown() { // This is a best-effort facility, so memory_order_relaxed is sufficient. return shutting_down_ && shutting_down_->load(std::memory_order_relaxed); } bool IsPausingManualCompaction() { // This is a best-effort facility, so memory_order_relaxed is sufficient. return manual_compaction_canceled_.load(std::memory_order_relaxed); } // Stores whether the current compaction iterator output // is a range tombstone start key. bool is_range_del_{false}; }; inline bool CompactionIterator::DefinitelyInSnapshot(SequenceNumber seq, SequenceNumber snapshot) { return DataIsDefinitelyInSnapshot(seq, snapshot, snapshot_checker_); } inline bool CompactionIterator::DefinitelyNotInSnapshot( SequenceNumber seq, SequenceNumber snapshot) { return DataIsDefinitelyNotInSnapshot(seq, snapshot, snapshot_checker_); } } // namespace ROCKSDB_NAMESPACE