1 // Copyright (c) 2011-present, Facebook, Inc. All rights reserved.
2 // This source code is licensed under both the GPLv2 (found in the
3 // COPYING file in the root directory) and Apache 2.0 License
4 // (found in the LICENSE.Apache file in the root directory).
6 // Copyright (c) 2011 The LevelDB Authors. All rights reserved.
7 // Use of this source code is governed by a BSD-style license that can be
8 // found in the LICENSE file. See the AUTHORS file for names of contributors.
22 #include "db/column_family.h"
23 #include "db/compaction/compaction_job.h"
24 #include "db/dbformat.h"
25 #include "db/error_handler.h"
26 #include "db/event_helpers.h"
27 #include "db/external_sst_file_ingestion_job.h"
28 #include "db/flush_job.h"
29 #include "db/flush_scheduler.h"
30 #include "db/import_column_family_job.h"
31 #include "db/internal_stats.h"
32 #include "db/log_writer.h"
33 #include "db/logs_with_prep_tracker.h"
34 #include "db/memtable_list.h"
35 #include "db/pre_release_callback.h"
36 #include "db/range_del_aggregator.h"
37 #include "db/read_callback.h"
38 #include "db/snapshot_checker.h"
39 #include "db/snapshot_impl.h"
40 #include "db/trim_history_scheduler.h"
41 #include "db/version_edit.h"
42 #include "db/wal_manager.h"
43 #include "db/write_controller.h"
44 #include "db/write_thread.h"
45 #include "logging/event_logger.h"
46 #include "monitoring/instrumented_mutex.h"
47 #include "options/db_options.h"
48 #include "port/port.h"
49 #include "rocksdb/db.h"
50 #include "rocksdb/env.h"
51 #include "rocksdb/memtablerep.h"
52 #include "rocksdb/status.h"
53 #include "rocksdb/trace_reader_writer.h"
54 #include "rocksdb/transaction_log.h"
55 #include "rocksdb/write_buffer_manager.h"
56 #include "table/scoped_arena_iterator.h"
57 #include "trace_replay/block_cache_tracer.h"
58 #include "trace_replay/io_tracer.h"
59 #include "trace_replay/trace_replay.h"
60 #include "util/autovector.h"
61 #include "util/hash.h"
62 #include "util/repeatable_thread.h"
63 #include "util/stop_watch.h"
64 #include "util/thread_local.h"
66 namespace ROCKSDB_NAMESPACE
{
69 class ArenaWrappedDBIter
;
70 class InMemoryStatsHistoryIterator
;
72 class PersistentStatsHistoryIterator
;
73 class PeriodicWorkScheduler
;
75 class PeriodicWorkTestScheduler
;
78 class TaskLimiterToken
;
84 struct ExternalSstFileInfo
;
87 // Class to maintain directories for all database paths other than main one.
90 IOStatus
SetDirectories(FileSystem
* fs
, const std::string
& dbname
,
91 const std::string
& wal_dir
,
92 const std::vector
<DbPath
>& data_paths
);
94 FSDirectory
* GetDataDir(size_t path_id
) const {
95 assert(path_id
< data_dirs_
.size());
96 FSDirectory
* ret_dir
= data_dirs_
[path_id
].get();
97 if (ret_dir
== nullptr) {
104 FSDirectory
* GetWalDir() {
106 return wal_dir_
.get();
108 return db_dir_
.get();
111 FSDirectory
* GetDbDir() { return db_dir_
.get(); }
114 std::unique_ptr
<FSDirectory
> db_dir_
;
115 std::vector
<std::unique_ptr
<FSDirectory
>> data_dirs_
;
116 std::unique_ptr
<FSDirectory
> wal_dir_
;
119 // While DB is the public interface of RocksDB, and DBImpl is the actual
120 // class implementing it. It's the entrance of the core RocksdB engine.
121 // All other DB implementations, e.g. TransactionDB, BlobDB, etc, wrap a
122 // DBImpl internally.
123 // Other than functions implementing the DB interface, some public
124 // functions are there for other internal components to call. For
125 // example, TransactionDB directly calls DBImpl::WriteImpl() and
126 // BlobDB directly calls DBImpl::GetImpl(). Some other functions
127 // are for sub-components to call. For example, ColumnFamilyHandleImpl
128 // calls DBImpl::FindObsoleteFiles().
130 // Since it's a very large class, the definition of the functions is
131 // divided in several db_impl_*.cc files, besides db_impl.cc.
132 class DBImpl
: public DB
{
134 DBImpl(const DBOptions
& options
, const std::string
& dbname
,
135 const bool seq_per_batch
= false, const bool batch_per_txn
= true);
136 // No copying allowed
137 DBImpl(const DBImpl
&) = delete;
138 void operator=(const DBImpl
&) = delete;
142 // ---- Implementations of the DB interface ----
145 virtual Status
Resume() override
;
148 virtual Status
Put(const WriteOptions
& options
,
149 ColumnFamilyHandle
* column_family
, const Slice
& key
,
150 const Slice
& value
) override
;
152 virtual Status
Merge(const WriteOptions
& options
,
153 ColumnFamilyHandle
* column_family
, const Slice
& key
,
154 const Slice
& value
) override
;
156 virtual Status
Delete(const WriteOptions
& options
,
157 ColumnFamilyHandle
* column_family
,
158 const Slice
& key
) override
;
159 using DB::SingleDelete
;
160 virtual Status
SingleDelete(const WriteOptions
& options
,
161 ColumnFamilyHandle
* column_family
,
162 const Slice
& key
) override
;
164 virtual Status
Write(const WriteOptions
& options
,
165 WriteBatch
* updates
) override
;
168 virtual Status
Get(const ReadOptions
& options
,
169 ColumnFamilyHandle
* column_family
, const Slice
& key
,
170 PinnableSlice
* value
) override
;
171 virtual Status
Get(const ReadOptions
& options
,
172 ColumnFamilyHandle
* column_family
, const Slice
& key
,
173 PinnableSlice
* value
, std::string
* timestamp
) override
;
175 using DB::GetMergeOperands
;
176 Status
GetMergeOperands(const ReadOptions
& options
,
177 ColumnFamilyHandle
* column_family
, const Slice
& key
,
178 PinnableSlice
* merge_operands
,
179 GetMergeOperandsOptions
* get_merge_operands_options
,
180 int* number_of_operands
) override
{
181 GetImplOptions get_impl_options
;
182 get_impl_options
.column_family
= column_family
;
183 get_impl_options
.merge_operands
= merge_operands
;
184 get_impl_options
.get_merge_operands_options
= get_merge_operands_options
;
185 get_impl_options
.number_of_operands
= number_of_operands
;
186 get_impl_options
.get_value
= false;
187 return GetImpl(options
, key
, get_impl_options
);
191 virtual std::vector
<Status
> MultiGet(
192 const ReadOptions
& options
,
193 const std::vector
<ColumnFamilyHandle
*>& column_family
,
194 const std::vector
<Slice
>& keys
,
195 std::vector
<std::string
>* values
) override
;
196 virtual std::vector
<Status
> MultiGet(
197 const ReadOptions
& options
,
198 const std::vector
<ColumnFamilyHandle
*>& column_family
,
199 const std::vector
<Slice
>& keys
, std::vector
<std::string
>* values
,
200 std::vector
<std::string
>* timestamps
) override
;
202 // This MultiGet is a batched version, which may be faster than calling Get
203 // multiple times, especially if the keys have some spatial locality that
204 // enables them to be queried in the same SST files/set of files. The larger
205 // the batch size, the more scope for batching and performance improvement
206 // The values and statuses parameters are arrays with number of elements
207 // equal to keys.size(). This allows the storage for those to be alloacted
208 // by the caller on the stack for small batches
209 virtual void MultiGet(const ReadOptions
& options
,
210 ColumnFamilyHandle
* column_family
,
211 const size_t num_keys
, const Slice
* keys
,
212 PinnableSlice
* values
, Status
* statuses
,
213 const bool sorted_input
= false) override
;
214 virtual void MultiGet(const ReadOptions
& options
,
215 ColumnFamilyHandle
* column_family
,
216 const size_t num_keys
, const Slice
* keys
,
217 PinnableSlice
* values
, std::string
* timestamps
,
219 const bool sorted_input
= false) override
;
221 virtual void MultiGet(const ReadOptions
& options
, const size_t num_keys
,
222 ColumnFamilyHandle
** column_families
, const Slice
* keys
,
223 PinnableSlice
* values
, Status
* statuses
,
224 const bool sorted_input
= false) override
;
225 virtual void MultiGet(const ReadOptions
& options
, const size_t num_keys
,
226 ColumnFamilyHandle
** column_families
, const Slice
* keys
,
227 PinnableSlice
* values
, std::string
* timestamps
,
229 const bool sorted_input
= false) override
;
231 virtual void MultiGetWithCallback(
232 const ReadOptions
& options
, ColumnFamilyHandle
* column_family
,
233 ReadCallback
* callback
,
234 autovector
<KeyContext
*, MultiGetContext::MAX_BATCH_SIZE
>* sorted_keys
);
236 virtual Status
CreateColumnFamily(const ColumnFamilyOptions
& cf_options
,
237 const std::string
& column_family
,
238 ColumnFamilyHandle
** handle
) override
;
239 virtual Status
CreateColumnFamilies(
240 const ColumnFamilyOptions
& cf_options
,
241 const std::vector
<std::string
>& column_family_names
,
242 std::vector
<ColumnFamilyHandle
*>* handles
) override
;
243 virtual Status
CreateColumnFamilies(
244 const std::vector
<ColumnFamilyDescriptor
>& column_families
,
245 std::vector
<ColumnFamilyHandle
*>* handles
) override
;
246 virtual Status
DropColumnFamily(ColumnFamilyHandle
* column_family
) override
;
247 virtual Status
DropColumnFamilies(
248 const std::vector
<ColumnFamilyHandle
*>& column_families
) override
;
250 // Returns false if key doesn't exist in the database and true if it may.
251 // If value_found is not passed in as null, then return the value if found in
252 // memory. On return, if value was found, then value_found will be set to true
253 // , otherwise false.
254 using DB::KeyMayExist
;
255 virtual bool KeyMayExist(const ReadOptions
& options
,
256 ColumnFamilyHandle
* column_family
, const Slice
& key
,
257 std::string
* value
, std::string
* timestamp
,
258 bool* value_found
= nullptr) override
;
260 using DB::NewIterator
;
261 virtual Iterator
* NewIterator(const ReadOptions
& options
,
262 ColumnFamilyHandle
* column_family
) override
;
263 virtual Status
NewIterators(
264 const ReadOptions
& options
,
265 const std::vector
<ColumnFamilyHandle
*>& column_families
,
266 std::vector
<Iterator
*>* iterators
) override
;
268 virtual const Snapshot
* GetSnapshot() override
;
269 virtual void ReleaseSnapshot(const Snapshot
* snapshot
) override
;
270 using DB::GetProperty
;
271 virtual bool GetProperty(ColumnFamilyHandle
* column_family
,
272 const Slice
& property
, std::string
* value
) override
;
273 using DB::GetMapProperty
;
274 virtual bool GetMapProperty(
275 ColumnFamilyHandle
* column_family
, const Slice
& property
,
276 std::map
<std::string
, std::string
>* value
) override
;
277 using DB::GetIntProperty
;
278 virtual bool GetIntProperty(ColumnFamilyHandle
* column_family
,
279 const Slice
& property
, uint64_t* value
) override
;
280 using DB::GetAggregatedIntProperty
;
281 virtual bool GetAggregatedIntProperty(const Slice
& property
,
282 uint64_t* aggregated_value
) override
;
283 using DB::GetApproximateSizes
;
284 virtual Status
GetApproximateSizes(const SizeApproximationOptions
& options
,
285 ColumnFamilyHandle
* column_family
,
286 const Range
* range
, int n
,
287 uint64_t* sizes
) override
;
288 using DB::GetApproximateMemTableStats
;
289 virtual void GetApproximateMemTableStats(ColumnFamilyHandle
* column_family
,
291 uint64_t* const count
,
292 uint64_t* const size
) override
;
293 using DB::CompactRange
;
294 virtual Status
CompactRange(const CompactRangeOptions
& options
,
295 ColumnFamilyHandle
* column_family
,
296 const Slice
* begin
, const Slice
* end
) override
;
298 using DB::CompactFiles
;
299 virtual Status
CompactFiles(
300 const CompactionOptions
& compact_options
,
301 ColumnFamilyHandle
* column_family
,
302 const std::vector
<std::string
>& input_file_names
, const int output_level
,
303 const int output_path_id
= -1,
304 std::vector
<std::string
>* const output_file_names
= nullptr,
305 CompactionJobInfo
* compaction_job_info
= nullptr) override
;
307 virtual Status
PauseBackgroundWork() override
;
308 virtual Status
ContinueBackgroundWork() override
;
310 virtual Status
EnableAutoCompaction(
311 const std::vector
<ColumnFamilyHandle
*>& column_family_handles
) override
;
313 virtual void EnableManualCompaction() override
;
314 virtual void DisableManualCompaction() override
;
316 using DB::SetOptions
;
318 ColumnFamilyHandle
* column_family
,
319 const std::unordered_map
<std::string
, std::string
>& options_map
) override
;
321 virtual Status
SetDBOptions(
322 const std::unordered_map
<std::string
, std::string
>& options_map
) override
;
324 using DB::NumberLevels
;
325 virtual int NumberLevels(ColumnFamilyHandle
* column_family
) override
;
326 using DB::MaxMemCompactionLevel
;
327 virtual int MaxMemCompactionLevel(ColumnFamilyHandle
* column_family
) override
;
328 using DB::Level0StopWriteTrigger
;
329 virtual int Level0StopWriteTrigger(
330 ColumnFamilyHandle
* column_family
) override
;
331 virtual const std::string
& GetName() const override
;
332 virtual Env
* GetEnv() const override
;
333 virtual FileSystem
* GetFileSystem() const override
;
334 using DB::GetOptions
;
335 virtual Options
GetOptions(ColumnFamilyHandle
* column_family
) const override
;
336 using DB::GetDBOptions
;
337 virtual DBOptions
GetDBOptions() const override
;
339 virtual Status
Flush(const FlushOptions
& options
,
340 ColumnFamilyHandle
* column_family
) override
;
341 virtual Status
Flush(
342 const FlushOptions
& options
,
343 const std::vector
<ColumnFamilyHandle
*>& column_families
) override
;
344 virtual Status
FlushWAL(bool sync
) override
;
345 bool TEST_WALBufferIsEmpty(bool lock
= true);
346 virtual Status
SyncWAL() override
;
347 virtual Status
LockWAL() override
;
348 virtual Status
UnlockWAL() override
;
350 virtual SequenceNumber
GetLatestSequenceNumber() const override
;
352 virtual bool SetPreserveDeletesSequenceNumber(SequenceNumber seqnum
) override
;
354 virtual Status
GetDbIdentity(std::string
& identity
) const override
;
356 virtual Status
GetDbIdentityFromIdentityFile(std::string
* identity
) const;
358 virtual Status
GetDbSessionId(std::string
& session_id
) const override
;
360 ColumnFamilyHandle
* DefaultColumnFamily() const override
;
362 ColumnFamilyHandle
* PersistentStatsColumnFamily() const;
364 virtual Status
Close() override
;
366 virtual Status
DisableFileDeletions() override
;
368 virtual Status
EnableFileDeletions(bool force
) override
;
370 virtual bool IsFileDeletionsEnabled() const;
372 Status
GetStatsHistory(
373 uint64_t start_time
, uint64_t end_time
,
374 std::unique_ptr
<StatsHistoryIterator
>* stats_iterator
) override
;
377 using DB::ResetStats
;
378 virtual Status
ResetStats() override
;
379 // All the returned filenames start with "/"
380 virtual Status
GetLiveFiles(std::vector
<std::string
>&,
381 uint64_t* manifest_file_size
,
382 bool flush_memtable
= true) override
;
383 virtual Status
GetSortedWalFiles(VectorLogPtr
& files
) override
;
384 virtual Status
GetCurrentWalFile(
385 std::unique_ptr
<LogFile
>* current_log_file
) override
;
386 virtual Status
GetCreationTimeOfOldestFile(
387 uint64_t* creation_time
) override
;
389 virtual Status
GetUpdatesSince(
390 SequenceNumber seq_number
, std::unique_ptr
<TransactionLogIterator
>* iter
,
391 const TransactionLogIterator::ReadOptions
& read_options
=
392 TransactionLogIterator::ReadOptions()) override
;
393 virtual Status
DeleteFile(std::string name
) override
;
394 Status
DeleteFilesInRanges(ColumnFamilyHandle
* column_family
,
395 const RangePtr
* ranges
, size_t n
,
396 bool include_end
= true);
398 virtual void GetLiveFilesMetaData(
399 std::vector
<LiveFileMetaData
>* metadata
) override
;
401 virtual Status
GetLiveFilesChecksumInfo(
402 FileChecksumList
* checksum_list
) override
;
404 // Obtains the meta data of the specified column family of the DB.
405 // Status::NotFound() will be returned if the current DB does not have
406 // any column family match the specified name.
407 // TODO(yhchiang): output parameter is placed in the end in this codebase.
408 virtual void GetColumnFamilyMetaData(ColumnFamilyHandle
* column_family
,
409 ColumnFamilyMetaData
* metadata
) override
;
411 Status
SuggestCompactRange(ColumnFamilyHandle
* column_family
,
412 const Slice
* begin
, const Slice
* end
) override
;
414 Status
PromoteL0(ColumnFamilyHandle
* column_family
,
415 int target_level
) override
;
417 using DB::IngestExternalFile
;
418 virtual Status
IngestExternalFile(
419 ColumnFamilyHandle
* column_family
,
420 const std::vector
<std::string
>& external_files
,
421 const IngestExternalFileOptions
& ingestion_options
) override
;
423 using DB::IngestExternalFiles
;
424 virtual Status
IngestExternalFiles(
425 const std::vector
<IngestExternalFileArg
>& args
) override
;
427 using DB::CreateColumnFamilyWithImport
;
428 virtual Status
CreateColumnFamilyWithImport(
429 const ColumnFamilyOptions
& options
, const std::string
& column_family_name
,
430 const ImportColumnFamilyOptions
& import_options
,
431 const ExportImportFilesMetaData
& metadata
,
432 ColumnFamilyHandle
** handle
) override
;
434 using DB::VerifyFileChecksums
;
435 Status
VerifyFileChecksums(const ReadOptions
& read_options
) override
;
437 using DB::VerifyChecksum
;
438 virtual Status
VerifyChecksum(const ReadOptions
& /*read_options*/) override
;
439 // Verify the checksums of files in db. Currently only tables are checked.
441 // read_options: controls file I/O behavior, e.g. read ahead size while
442 // reading all the live table files.
444 // use_file_checksum: if false, verify the block checksums of all live table
445 // in db. Otherwise, obtain the file checksums and compare
446 // with the MANIFEST. Currently, file checksums are
447 // recomputed by reading all table files.
449 // Returns: OK if there is no file whose file or block checksum mismatches.
450 Status
VerifyChecksumInternal(const ReadOptions
& read_options
,
451 bool use_file_checksum
);
453 Status
VerifySstFileChecksum(const FileMetaData
& fmeta
,
454 const std::string
& fpath
,
455 const ReadOptions
& read_options
);
457 using DB::StartTrace
;
458 virtual Status
StartTrace(
459 const TraceOptions
& options
,
460 std::unique_ptr
<TraceWriter
>&& trace_writer
) override
;
463 virtual Status
EndTrace() override
;
465 using DB::StartBlockCacheTrace
;
466 Status
StartBlockCacheTrace(
467 const TraceOptions
& options
,
468 std::unique_ptr
<TraceWriter
>&& trace_writer
) override
;
470 using DB::EndBlockCacheTrace
;
471 Status
EndBlockCacheTrace() override
;
473 using DB::StartIOTrace
;
474 Status
StartIOTrace(Env
* env
, const TraceOptions
& options
,
475 std::unique_ptr
<TraceWriter
>&& trace_writer
) override
;
477 using DB::EndIOTrace
;
478 Status
EndIOTrace() override
;
480 using DB::GetPropertiesOfAllTables
;
481 virtual Status
GetPropertiesOfAllTables(
482 ColumnFamilyHandle
* column_family
,
483 TablePropertiesCollection
* props
) override
;
484 virtual Status
GetPropertiesOfTablesInRange(
485 ColumnFamilyHandle
* column_family
, const Range
* range
, std::size_t n
,
486 TablePropertiesCollection
* props
) override
;
488 #endif // ROCKSDB_LITE
490 // ---- End of implementations of the DB interface ----
492 struct GetImplOptions
{
493 ColumnFamilyHandle
* column_family
= nullptr;
494 PinnableSlice
* value
= nullptr;
495 std::string
* timestamp
= nullptr;
496 bool* value_found
= nullptr;
497 ReadCallback
* callback
= nullptr;
498 bool* is_blob_index
= nullptr;
499 // If true return value associated with key via value pointer else return
500 // all merge operands for key via merge_operands pointer
501 bool get_value
= true;
502 // Pointer to an array of size
503 // get_merge_operands_options.expected_max_number_of_operands allocated by
505 PinnableSlice
* merge_operands
= nullptr;
506 GetMergeOperandsOptions
* get_merge_operands_options
= nullptr;
507 int* number_of_operands
= nullptr;
510 // Function that Get and KeyMayExist call with no_io true or false
511 // Note: 'value_found' from KeyMayExist propagates here
512 // This function is also called by GetMergeOperands
513 // If get_impl_options.get_value = true get value associated with
514 // get_impl_options.key via get_impl_options.value
515 // If get_impl_options.get_value = false get merge operands associated with
516 // get_impl_options.key via get_impl_options.merge_operands
517 Status
GetImpl(const ReadOptions
& options
, const Slice
& key
,
518 GetImplOptions
& get_impl_options
);
520 // If `snapshot` == kMaxSequenceNumber, set a recent one inside the file.
521 ArenaWrappedDBIter
* NewIteratorImpl(const ReadOptions
& options
,
522 ColumnFamilyData
* cfd
,
523 SequenceNumber snapshot
,
524 ReadCallback
* read_callback
,
525 bool allow_blob
= false,
526 bool allow_refresh
= true);
528 virtual SequenceNumber
GetLastPublishedSequence() const {
529 if (last_seq_same_as_publish_seq_
) {
530 return versions_
->LastSequence();
532 return versions_
->LastPublishedSequence();
536 // REQUIRES: joined the main write queue if two_write_queues is disabled, and
537 // the second write queue otherwise.
538 virtual void SetLastPublishedSequence(SequenceNumber seq
);
539 // Returns LastSequence in last_seq_same_as_publish_seq_
540 // mode and LastAllocatedSequence otherwise. This is useful when visiblility
541 // depends also on data written to the WAL but not to the memtable.
542 SequenceNumber
TEST_GetLastVisibleSequence() const;
545 // Similar to Write() but will call the callback once on the single write
546 // thread to determine whether it is safe to perform the write.
547 virtual Status
WriteWithCallback(const WriteOptions
& write_options
,
548 WriteBatch
* my_batch
,
549 WriteCallback
* callback
);
551 // Returns the sequence number that is guaranteed to be smaller than or equal
552 // to the sequence number of any key that could be inserted into the current
553 // memtables. It can then be assumed that any write with a larger(or equal)
554 // sequence number will be present in this memtable or a later memtable.
556 // If the earliest sequence number could not be determined,
557 // kMaxSequenceNumber will be returned.
559 // If include_history=true, will also search Memtables in MemTableList
561 SequenceNumber
GetEarliestMemTableSequenceNumber(SuperVersion
* sv
,
562 bool include_history
);
564 // For a given key, check to see if there are any records for this key
565 // in the memtables, including memtable history. If cache_only is false,
566 // SST files will also be checked.
568 // If a key is found, *found_record_for_key will be set to true and
569 // *seq will be set to the stored sequence number for the latest
570 // operation on this key or kMaxSequenceNumber if unknown.
571 // If no key is found, *found_record_for_key will be set to false.
573 // Note: If cache_only=false, it is possible for *seq to be set to 0 if
574 // the sequence number has been cleared from the record. If the caller is
575 // holding an active db snapshot, we know the missing sequence must be less
576 // than the snapshot's sequence number (sequence numbers are only cleared
577 // when there are no earlier active snapshots).
579 // If NotFound is returned and found_record_for_key is set to false, then no
580 // record for this key was found. If the caller is holding an active db
581 // snapshot, we know that no key could have existing after this snapshot
582 // (since we do not compact keys that have an earlier snapshot).
584 // Only records newer than or at `lower_bound_seq` are guaranteed to be
585 // returned. Memtables and files may not be checked if it only contains data
586 // older than `lower_bound_seq`.
588 // Returns OK or NotFound on success,
589 // other status on unexpected error.
590 // TODO(andrewkr): this API need to be aware of range deletion operations
591 Status
GetLatestSequenceForKey(SuperVersion
* sv
, const Slice
& key
,
593 SequenceNumber lower_bound_seq
,
595 bool* found_record_for_key
,
596 bool* is_blob_index
= nullptr);
598 Status
TraceIteratorSeek(const uint32_t& cf_id
, const Slice
& key
);
599 Status
TraceIteratorSeekForPrev(const uint32_t& cf_id
, const Slice
& key
);
600 #endif // ROCKSDB_LITE
602 // Similar to GetSnapshot(), but also lets the db know that this snapshot
603 // will be used for transaction write-conflict checking. The DB can then
604 // make sure not to compact any keys that would prevent a write-conflict from
606 const Snapshot
* GetSnapshotForWriteConflictBoundary();
608 // checks if all live files exist on file system and that their file sizes
609 // match to our in-memory records
610 virtual Status
CheckConsistency();
612 // max_file_num_to_ignore allows bottom level compaction to filter out newly
613 // compacted SST files. Setting max_file_num_to_ignore to kMaxUint64 will
614 // disable the filtering
615 Status
RunManualCompaction(ColumnFamilyData
* cfd
, int input_level
,
617 const CompactRangeOptions
& compact_range_options
,
618 const Slice
* begin
, const Slice
* end
,
619 bool exclusive
, bool disallow_trivial_move
,
620 uint64_t max_file_num_to_ignore
);
622 // Return an internal iterator over the current state of the database.
623 // The keys of this iterator are internal keys (see format.h).
624 // The returned iterator should be deleted when no longer needed.
625 // If allow_unprepared_value is true, the returned iterator may defer reading
626 // the value and so will require PrepareValue() to be called before value();
627 // allow_unprepared_value = false is convenient when this optimization is not
628 // useful, e.g. when reading the whole column family.
629 // @param read_options Must outlive the returned iterator.
630 InternalIterator
* NewInternalIterator(
631 const ReadOptions
& read_options
, Arena
* arena
,
632 RangeDelAggregator
* range_del_agg
, SequenceNumber sequence
,
633 ColumnFamilyHandle
* column_family
= nullptr,
634 bool allow_unprepared_value
= false);
636 LogsWithPrepTracker
* logs_with_prep_tracker() {
637 return &logs_with_prep_tracker_
;
644 // Returns maximum background flushes and compactions allowed to be scheduled
645 BGJobLimits
GetBGJobLimits() const;
646 // Need a static version that can be called during SanitizeOptions().
647 static BGJobLimits
GetBGJobLimits(int max_background_flushes
,
648 int max_background_compactions
,
649 int max_background_jobs
,
650 bool parallelize_compactions
);
652 // move logs pending closing from job_context to the DB queue and
654 void ScheduleBgLogWriterClose(JobContext
* job_context
);
656 uint64_t MinLogNumberToKeep();
658 // Returns the lower bound file number for SSTs that won't be deleted, even if
659 // they're obsolete. This lower bound is used internally to prevent newly
660 // created flush/compaction output files from being deleted before they're
661 // installed. This technique avoids the need for tracking the exact numbers of
662 // files pending creation, although it prevents more files than necessary from
664 uint64_t MinObsoleteSstNumberToKeep();
666 // Returns the list of live files in 'live' and the list
667 // of all files in the filesystem in 'candidate_files'.
668 // If force == false and the last call was less than
669 // db_options_.delete_obsolete_files_period_micros microseconds ago,
670 // it will not fill up the job_context
671 void FindObsoleteFiles(JobContext
* job_context
, bool force
,
672 bool no_full_scan
= false);
674 // Diffs the files listed in filenames and those that do not
675 // belong to live files are possibly removed. Also, removes all the
676 // files in sst_delete_files and log_delete_files.
677 // It is not necessary to hold the mutex when invoking this method.
678 // If FindObsoleteFiles() was run, we need to also run
679 // PurgeObsoleteFiles(), even if disable_delete_obsolete_files_ is true
680 void PurgeObsoleteFiles(JobContext
& background_contet
,
681 bool schedule_only
= false);
683 // Schedule a background job to actually delete obsolete files.
684 void SchedulePurge();
686 const SnapshotList
& snapshots() const { return snapshots_
; }
688 // load list of snapshots to `snap_vector` that is no newer than `max_seq`
689 // in ascending order.
690 // `oldest_write_conflict_snapshot` is filled with the oldest snapshot
691 // which satisfies SnapshotImpl.is_write_conflict_boundary_ = true.
692 void LoadSnapshots(std::vector
<SequenceNumber
>* snap_vector
,
693 SequenceNumber
* oldest_write_conflict_snapshot
,
694 const SequenceNumber
& max_seq
) const {
695 InstrumentedMutexLock
l(mutex());
696 snapshots().GetAll(snap_vector
, oldest_write_conflict_snapshot
, max_seq
);
699 const ImmutableDBOptions
& immutable_db_options() const {
700 return immutable_db_options_
;
703 // Cancel all background jobs, including flush, compaction, background
704 // purging, stats dumping threads, etc. If `wait` = true, wait for the
705 // running jobs to abort or finish before returning. Otherwise, only
706 // sends the signals.
707 void CancelAllBackgroundWork(bool wait
);
709 // Find Super version and reference it. Based on options, it might return
710 // the thread local cached one.
711 // Call ReturnAndCleanupSuperVersion() when it is no longer needed.
712 SuperVersion
* GetAndRefSuperVersion(ColumnFamilyData
* cfd
);
714 // Similar to the previous function but looks up based on a column family id.
715 // nullptr will be returned if this column family no longer exists.
716 // REQUIRED: this function should only be called on the write thread or if the
718 SuperVersion
* GetAndRefSuperVersion(uint32_t column_family_id
);
720 // Un-reference the super version and clean it up if it is the last reference.
721 void CleanupSuperVersion(SuperVersion
* sv
);
723 // Un-reference the super version and return it to thread local cache if
724 // needed. If it is the last reference of the super version. Clean it up
725 // after un-referencing it.
726 void ReturnAndCleanupSuperVersion(ColumnFamilyData
* cfd
, SuperVersion
* sv
);
728 // Similar to the previous function but looks up based on a column family id.
729 // nullptr will be returned if this column family no longer exists.
730 // REQUIRED: this function should only be called on the write thread.
731 void ReturnAndCleanupSuperVersion(uint32_t colun_family_id
, SuperVersion
* sv
);
733 // REQUIRED: this function should only be called on the write thread or if the
734 // mutex is held. Return value only valid until next call to this function or
735 // mutex is released.
736 ColumnFamilyHandle
* GetColumnFamilyHandle(uint32_t column_family_id
);
738 // Same as above, should called without mutex held and not on write thread.
739 std::unique_ptr
<ColumnFamilyHandle
> GetColumnFamilyHandleUnlocked(
740 uint32_t column_family_id
);
742 // Returns the number of currently running flushes.
743 // REQUIREMENT: mutex_ must be held when calling this function.
744 int num_running_flushes() {
746 return num_running_flushes_
;
749 // Returns the number of currently running compactions.
750 // REQUIREMENT: mutex_ must be held when calling this function.
751 int num_running_compactions() {
753 return num_running_compactions_
;
756 const WriteController
& write_controller() { return write_controller_
; }
758 // @param read_options Must outlive the returned iterator.
759 InternalIterator
* NewInternalIterator(const ReadOptions
& read_options
,
760 ColumnFamilyData
* cfd
,
761 SuperVersion
* super_version
,
763 RangeDelAggregator
* range_del_agg
,
764 SequenceNumber sequence
,
765 bool allow_unprepared_value
);
767 // hollow transactions shell used for recovery.
768 // these will then be passed to TransactionDB so that
769 // locks can be reacquired before writing can resume.
770 struct RecoveredTransaction
{
775 uint64_t log_number_
;
776 // TODO(lth): For unprepared, the memory usage here can be big for
777 // unprepared transactions. This is only useful for rollbacks, and we
778 // can in theory just keep keyset for that.
780 // Number of sub-batches. A new sub-batch is created if txn attempts to
781 // insert a duplicate key,seq to memtable. This is currently used in
782 // WritePreparedTxn/WriteUnpreparedTxn.
786 // This maps the seq of the first key in the batch to BatchInfo, which
787 // contains WriteBatch and other information relevant to the batch.
789 // For WriteUnprepared, batches_ can have size greater than 1, but for
790 // other write policies, it must be of size 1.
791 std::map
<SequenceNumber
, BatchInfo
> batches_
;
793 explicit RecoveredTransaction(const uint64_t log
, const std::string
& name
,
794 WriteBatch
* batch
, SequenceNumber seq
,
795 size_t batch_cnt
, bool unprepared
)
796 : name_(name
), unprepared_(unprepared
) {
797 batches_
[seq
] = {log
, batch
, batch_cnt
};
800 ~RecoveredTransaction() {
801 for (auto& it
: batches_
) {
802 delete it
.second
.batch_
;
806 void AddBatch(SequenceNumber seq
, uint64_t log_number
, WriteBatch
* batch
,
807 size_t batch_cnt
, bool unprepared
) {
808 assert(batches_
.count(seq
) == 0);
809 batches_
[seq
] = {log_number
, batch
, batch_cnt
};
810 // Prior state must be unprepared, since the prepare batch must be the
813 unprepared_
= unprepared
;
817 bool allow_2pc() const { return immutable_db_options_
.allow_2pc
; }
819 std::unordered_map
<std::string
, RecoveredTransaction
*>
820 recovered_transactions() {
821 return recovered_transactions_
;
824 RecoveredTransaction
* GetRecoveredTransaction(const std::string
& name
) {
825 auto it
= recovered_transactions_
.find(name
);
826 if (it
== recovered_transactions_
.end()) {
833 void InsertRecoveredTransaction(const uint64_t log
, const std::string
& name
,
834 WriteBatch
* batch
, SequenceNumber seq
,
835 size_t batch_cnt
, bool unprepared_batch
) {
836 // For WriteUnpreparedTxn, InsertRecoveredTransaction is called multiple
837 // times for every unprepared batch encountered during recovery.
839 // If the transaction is prepared, then the last call to
840 // InsertRecoveredTransaction will have unprepared_batch = false.
841 auto rtxn
= recovered_transactions_
.find(name
);
842 if (rtxn
== recovered_transactions_
.end()) {
843 recovered_transactions_
[name
] = new RecoveredTransaction(
844 log
, name
, batch
, seq
, batch_cnt
, unprepared_batch
);
846 rtxn
->second
->AddBatch(seq
, log
, batch
, batch_cnt
, unprepared_batch
);
848 logs_with_prep_tracker_
.MarkLogAsContainingPrepSection(log
);
851 void DeleteRecoveredTransaction(const std::string
& name
) {
852 auto it
= recovered_transactions_
.find(name
);
853 assert(it
!= recovered_transactions_
.end());
854 auto* trx
= it
->second
;
855 recovered_transactions_
.erase(it
);
856 for (const auto& info
: trx
->batches_
) {
857 logs_with_prep_tracker_
.MarkLogAsHavingPrepSectionFlushed(
858 info
.second
.log_number_
);
863 void DeleteAllRecoveredTransactions() {
864 for (auto it
= recovered_transactions_
.begin();
865 it
!= recovered_transactions_
.end(); ++it
) {
868 recovered_transactions_
.clear();
871 void AddToLogsToFreeQueue(log::Writer
* log_writer
) {
872 logs_to_free_queue_
.push_back(log_writer
);
875 void AddSuperVersionsToFreeQueue(SuperVersion
* sv
) {
876 superversions_to_free_queue_
.push_back(sv
);
879 void SetSnapshotChecker(SnapshotChecker
* snapshot_checker
);
881 // Fill JobContext with snapshot information needed by flush and compaction.
882 void GetSnapshotContext(JobContext
* job_context
,
883 std::vector
<SequenceNumber
>* snapshot_seqs
,
884 SequenceNumber
* earliest_write_conflict_snapshot
,
885 SnapshotChecker
** snapshot_checker
);
888 void SetRecoverableStatePreReleaseCallback(PreReleaseCallback
* callback
);
890 InstrumentedMutex
* mutex() const { return &mutex_
; }
892 // Initialize a brand new DB. The DB directory is expected to be empty before
893 // calling it. Push new manifest file name into `new_filenames`.
894 Status
NewDB(std::vector
<std::string
>* new_filenames
);
896 // This is to be used only by internal rocksdb classes.
897 static Status
Open(const DBOptions
& db_options
, const std::string
& name
,
898 const std::vector
<ColumnFamilyDescriptor
>& column_families
,
899 std::vector
<ColumnFamilyHandle
*>* handles
, DB
** dbptr
,
900 const bool seq_per_batch
, const bool batch_per_txn
);
902 static IOStatus
CreateAndNewDirectory(
903 FileSystem
* fs
, const std::string
& dirname
,
904 std::unique_ptr
<FSDirectory
>* directory
);
906 // find stats map from stats_history_ with smallest timestamp in
907 // the range of [start_time, end_time)
908 bool FindStatsByTime(uint64_t start_time
, uint64_t end_time
,
910 std::map
<std::string
, uint64_t>* stats_map
);
912 // Print information of all tombstones of all iterators to the std::string
913 // This is only used by ldb. The output might be capped. Tombstones
914 // printed out are not guaranteed to be in any order.
915 Status
TablesRangeTombstoneSummary(ColumnFamilyHandle
* column_family
,
916 int max_entries_to_print
,
917 std::string
* out_str
);
920 // Compact any files in the named level that overlap [*begin, *end]
921 Status
TEST_CompactRange(int level
, const Slice
* begin
, const Slice
* end
,
922 ColumnFamilyHandle
* column_family
= nullptr,
923 bool disallow_trivial_move
= false);
925 void TEST_SwitchWAL();
927 bool TEST_UnableToReleaseOldestLog() { return unable_to_release_oldest_log_
; }
929 bool TEST_IsLogGettingFlushed() {
930 return alive_log_files_
.begin()->getting_flushed
;
933 Status
TEST_SwitchMemtable(ColumnFamilyData
* cfd
= nullptr);
935 // Force current memtable contents to be flushed.
936 Status
TEST_FlushMemTable(bool wait
= true, bool allow_write_stall
= false,
937 ColumnFamilyHandle
* cfh
= nullptr);
939 Status
TEST_FlushMemTable(ColumnFamilyData
* cfd
,
940 const FlushOptions
& flush_opts
);
942 // Flush (multiple) ColumnFamilyData without using ColumnFamilyHandle. This
943 // is because in certain cases, we can flush column families, wait for the
944 // flush to complete, but delete the column family handle before the wait
945 // finishes. For example in CompactRange.
946 Status
TEST_AtomicFlushMemTables(const autovector
<ColumnFamilyData
*>& cfds
,
947 const FlushOptions
& flush_opts
);
949 // Wait for memtable compaction
950 Status
TEST_WaitForFlushMemTable(ColumnFamilyHandle
* column_family
= nullptr);
952 // Wait for any compaction
953 // We add a bool parameter to wait for unscheduledCompactions_ == 0, but this
954 // is only for the special test of CancelledCompactions
955 Status
TEST_WaitForCompact(bool waitUnscheduled
= false);
957 // Return the maximum overlapping data (in bytes) at next level for any
958 // file at a level >= 1.
959 int64_t TEST_MaxNextLevelOverlappingBytes(
960 ColumnFamilyHandle
* column_family
= nullptr);
962 // Return the current manifest file no.
963 uint64_t TEST_Current_Manifest_FileNo();
965 // Returns the number that'll be assigned to the next file that's created.
966 uint64_t TEST_Current_Next_FileNo();
968 // get total level0 file size. Only for testing.
969 uint64_t TEST_GetLevel0TotalSize();
971 void TEST_GetFilesMetaData(ColumnFamilyHandle
* column_family
,
972 std::vector
<std::vector
<FileMetaData
>>* metadata
);
974 void TEST_LockMutex();
976 void TEST_UnlockMutex();
978 // REQUIRES: mutex locked
979 void* TEST_BeginWrite();
981 // REQUIRES: mutex locked
982 // pass the pointer that you got from TEST_BeginWrite()
983 void TEST_EndWrite(void* w
);
985 uint64_t TEST_MaxTotalInMemoryState() const {
986 return max_total_in_memory_state_
;
989 size_t TEST_LogsToFreeSize();
991 uint64_t TEST_LogfileNumber();
993 uint64_t TEST_total_log_size() const { return total_log_size_
; }
995 // Returns column family name to ImmutableCFOptions map.
996 Status
TEST_GetAllImmutableCFOptions(
997 std::unordered_map
<std::string
, const ImmutableCFOptions
*>* iopts_map
);
999 // Return the lastest MutableCFOptions of a column family
1000 Status
TEST_GetLatestMutableCFOptions(ColumnFamilyHandle
* column_family
,
1001 MutableCFOptions
* mutable_cf_options
);
1003 Cache
* TEST_table_cache() { return table_cache_
.get(); }
1005 WriteController
& TEST_write_controler() { return write_controller_
; }
1007 uint64_t TEST_FindMinLogContainingOutstandingPrep();
1008 uint64_t TEST_FindMinPrepLogReferencedByMemTable();
1009 size_t TEST_PreparedSectionCompletedSize();
1010 size_t TEST_LogsWithPrepSize();
1012 int TEST_BGCompactionsAllowed() const;
1013 int TEST_BGFlushesAllowed() const;
1014 size_t TEST_GetWalPreallocateBlockSize(uint64_t write_buffer_size
) const;
1015 void TEST_WaitForStatsDumpRun(std::function
<void()> callback
) const;
1016 size_t TEST_EstimateInMemoryStatsHistorySize() const;
1018 VersionSet
* TEST_GetVersionSet() const { return versions_
.get(); }
1020 const std::unordered_set
<uint64_t>& TEST_GetFilesGrabbedForPurge() const {
1021 return files_grabbed_for_purge_
;
1024 #ifndef ROCKSDB_LITE
1025 PeriodicWorkTestScheduler
* TEST_GetPeriodicWorkScheduler() const;
1026 #endif // !ROCKSDB_LITE
1030 // persist stats to column family "_persistent_stats"
1031 void PersistStats();
1033 // dump rocksdb.stats to LOG
1036 // flush LOG out of application buffer
1037 void FlushInfoLog();
1040 const std::string dbname_
;
1042 // db_session_id_ is an identifier that gets reset
1043 // every time the DB is opened
1044 std::string db_session_id_
;
1045 std::unique_ptr
<VersionSet
> versions_
;
1046 // Flag to check whether we allocated and own the info log file
1048 const DBOptions initial_db_options_
;
1050 std::shared_ptr
<IOTracer
> io_tracer_
;
1051 const ImmutableDBOptions immutable_db_options_
;
1053 MutableDBOptions mutable_db_options_
;
1055 std::unordered_map
<std::string
, RecoveredTransaction
*>
1056 recovered_transactions_
;
1057 std::unique_ptr
<Tracer
> tracer_
;
1058 InstrumentedMutex trace_mutex_
;
1059 BlockCacheTracer block_cache_tracer_
;
1061 // State below is protected by mutex_
1062 // With two_write_queues enabled, some of the variables that accessed during
1063 // WriteToWAL need different synchronization: log_empty_, alive_log_files_,
1064 // logs_, logfile_number_. Refer to the definition of each variable below for
1065 // more description.
1066 mutable InstrumentedMutex mutex_
;
1068 ColumnFamilyHandleImpl
* default_cf_handle_
;
1069 InternalStats
* default_cf_internal_stats_
;
1071 // only used for dynamically adjusting max_total_wal_size. it is a sum of
1072 // [write_buffer_size * max_write_buffer_number] over all column families
1073 uint64_t max_total_in_memory_state_
;
1074 // If true, we have only one (default) column family. We use this to optimize
1076 bool single_column_family_mode_
;
1078 // The options to access storage files
1079 const FileOptions file_options_
;
1081 // Additonal options for compaction and flush
1082 FileOptions file_options_for_compaction_
;
1084 std::unique_ptr
<ColumnFamilyMemTablesImpl
> column_family_memtables_
;
1086 // Increase the sequence number after writing each batch, whether memtable is
1087 // disabled for that or not. Otherwise the sequence number is increased after
1088 // writing each key into memtable. This implies that when disable_memtable is
1089 // set, the seq is not increased at all.
1092 const bool seq_per_batch_
;
1093 // This determines during recovery whether we expect one writebatch per
1094 // recovered transaction, or potentially multiple writebatches per
1095 // transaction. For WriteUnprepared, this is set to false, since multiple
1096 // batches can exist per transaction.
1099 const bool batch_per_txn_
;
1101 // Except in DB::Open(), WriteOptionsFile can only be called when:
1102 // Persist options to options file.
1103 // If need_mutex_lock = false, the method will lock DB mutex.
1104 // If need_enter_write_thread = false, the method will enter write thread.
1105 Status
WriteOptionsFile(bool need_mutex_lock
, bool need_enter_write_thread
);
1107 // The following two functions can only be called when:
1108 // 1. WriteThread::Writer::EnterUnbatched() is used.
1109 // 2. db_mutex is NOT held
1110 Status
RenameTempFileToOptionsFile(const std::string
& file_name
);
1111 Status
DeleteObsoleteOptionsFiles();
1113 void NotifyOnFlushBegin(ColumnFamilyData
* cfd
, FileMetaData
* file_meta
,
1114 const MutableCFOptions
& mutable_cf_options
,
1117 void NotifyOnFlushCompleted(
1118 ColumnFamilyData
* cfd
, const MutableCFOptions
& mutable_cf_options
,
1119 std::list
<std::unique_ptr
<FlushJobInfo
>>* flush_jobs_info
);
1121 void NotifyOnCompactionBegin(ColumnFamilyData
* cfd
, Compaction
* c
,
1123 const CompactionJobStats
& job_stats
, int job_id
);
1125 void NotifyOnCompactionCompleted(ColumnFamilyData
* cfd
, Compaction
* c
,
1127 const CompactionJobStats
& job_stats
,
1129 void NotifyOnMemTableSealed(ColumnFamilyData
* cfd
,
1130 const MemTableInfo
& mem_table_info
);
1132 #ifndef ROCKSDB_LITE
1133 void NotifyOnExternalFileIngested(
1134 ColumnFamilyData
* cfd
, const ExternalSstFileIngestionJob
& ingestion_job
);
1135 #endif // !ROCKSDB_LITE
1137 void NewThreadStatusCfInfo(ColumnFamilyData
* cfd
) const;
1139 void EraseThreadStatusCfInfo(ColumnFamilyData
* cfd
) const;
1141 void EraseThreadStatusDbInfo() const;
1143 // If disable_memtable is set the application logic must guarantee that the
1144 // batch will still be skipped from memtable during the recovery. An excption
1145 // to this is seq_per_batch_ mode, in which since each batch already takes one
1146 // seq, it is ok for the batch to write to memtable during recovery as long as
1147 // it only takes one sequence number: i.e., no duplicate keys.
1148 // In WriteCommitted it is guarnateed since disable_memtable is used for
1149 // prepare batch which will be written to memtable later during the commit,
1150 // and in WritePrepared it is guaranteed since it will be used only for WAL
1151 // markers which will never be written to memtable. If the commit marker is
1152 // accompanied with CommitTimeWriteBatch that is not written to memtable as
1153 // long as it has no duplicate keys, it does not violate the one-seq-per-batch
1155 // batch_cnt is expected to be non-zero in seq_per_batch mode and
1156 // indicates the number of sub-patches. A sub-patch is a subset of the write
1157 // batch that does not have duplicate keys.
1158 Status
WriteImpl(const WriteOptions
& options
, WriteBatch
* updates
,
1159 WriteCallback
* callback
= nullptr,
1160 uint64_t* log_used
= nullptr, uint64_t log_ref
= 0,
1161 bool disable_memtable
= false, uint64_t* seq_used
= nullptr,
1162 size_t batch_cnt
= 0,
1163 PreReleaseCallback
* pre_release_callback
= nullptr);
1165 Status
PipelinedWriteImpl(const WriteOptions
& options
, WriteBatch
* updates
,
1166 WriteCallback
* callback
= nullptr,
1167 uint64_t* log_used
= nullptr, uint64_t log_ref
= 0,
1168 bool disable_memtable
= false,
1169 uint64_t* seq_used
= nullptr);
1171 // Write only to memtables without joining any write queue
1172 Status
UnorderedWriteMemtable(const WriteOptions
& write_options
,
1173 WriteBatch
* my_batch
, WriteCallback
* callback
,
1174 uint64_t log_ref
, SequenceNumber seq
,
1175 const size_t sub_batch_cnt
);
1177 // Whether the batch requires to be assigned with an order
1178 enum AssignOrder
: bool { kDontAssignOrder
, kDoAssignOrder
};
1179 // Whether it requires publishing last sequence or not
1180 enum PublishLastSeq
: bool { kDontPublishLastSeq
, kDoPublishLastSeq
};
1182 // Join the write_thread to write the batch only to the WAL. It is the
1183 // responsibility of the caller to also write the write batch to the memtable
1186 // sub_batch_cnt is expected to be non-zero when assign_order = kDoAssignOrder
1187 // indicating the number of sub-batches in my_batch. A sub-patch is a subset
1188 // of the write batch that does not have duplicate keys. When seq_per_batch is
1189 // not set, each key is a separate sub_batch. Otherwise each duplicate key
1190 // marks start of a new sub-batch.
1191 Status
WriteImplWALOnly(
1192 WriteThread
* write_thread
, const WriteOptions
& options
,
1193 WriteBatch
* updates
, WriteCallback
* callback
, uint64_t* log_used
,
1194 const uint64_t log_ref
, uint64_t* seq_used
, const size_t sub_batch_cnt
,
1195 PreReleaseCallback
* pre_release_callback
, const AssignOrder assign_order
,
1196 const PublishLastSeq publish_last_seq
, const bool disable_memtable
);
1198 // write cached_recoverable_state_ to memtable if it is not empty
1199 // The writer must be the leader in write_thread_ and holding mutex_
1200 Status
WriteRecoverableState();
1202 // Actual implementation of Close()
1205 // Recover the descriptor from persistent storage. May do a significant
1206 // amount of work to recover recently logged updates. Any changes to
1207 // be made to the descriptor are added to *edit.
1208 // recovered_seq is set to less than kMaxSequenceNumber if the log's tail is
1210 virtual Status
Recover(
1211 const std::vector
<ColumnFamilyDescriptor
>& column_families
,
1212 bool read_only
= false, bool error_if_wal_file_exists
= false,
1213 bool error_if_data_exists_in_wals
= false,
1214 uint64_t* recovered_seq
= nullptr);
1216 virtual bool OwnTablesAndLogs() const { return true; }
1218 // Set DB identity file, and write DB ID to manifest if necessary.
1221 // REQUIRES: db mutex held when calling this function, but the db mutex can
1222 // be released and re-acquired. Db mutex will be held when the function
1224 // After recovery, there may be SST files in db/cf paths that are
1225 // not referenced in the MANIFEST (e.g.
1226 // 1. It's best effort recovery;
1227 // 2. The VersionEdits referencing the SST files are appended to
1228 // MANIFEST, DB crashes when syncing the MANIFEST, the VersionEdits are
1229 // still not synced to MANIFEST during recovery.)
1230 // We delete these SST files. In the
1231 // meantime, we find out the largest file number present in the paths, and
1232 // bump up the version set's next_file_number_ to be 1 + largest_file_number.
1233 Status
DeleteUnreferencedSstFiles();
1235 // SetDbSessionId() should be called in the constuctor DBImpl()
1236 // to ensure that db_session_id_ gets updated every time the DB is opened
1237 void SetDbSessionId();
1241 friend class ErrorHandler
;
1242 friend class InternalStats
;
1243 friend class PessimisticTransaction
;
1244 friend class TransactionBaseImpl
;
1245 friend class WriteCommittedTxn
;
1246 friend class WritePreparedTxn
;
1247 friend class WritePreparedTxnDB
;
1248 friend class WriteBatchWithIndex
;
1249 friend class WriteUnpreparedTxnDB
;
1250 friend class WriteUnpreparedTxn
;
1252 #ifndef ROCKSDB_LITE
1253 friend class ForwardIterator
;
1255 friend struct SuperVersion
;
1256 friend class CompactedDBImpl
;
1257 friend class DBTest_ConcurrentFlushWAL_Test
;
1258 friend class DBTest_MixedSlowdownOptionsStop_Test
;
1259 friend class DBCompactionTest_CompactBottomLevelFilesWithDeletions_Test
;
1260 friend class DBCompactionTest_CompactionDuringShutdown_Test
;
1261 friend class StatsHistoryTest_PersistentStatsCreateColumnFamilies_Test
;
1263 friend class DBTest2_ReadCallbackTest_Test
;
1264 friend class WriteCallbackPTest_WriteWithCallbackTest_Test
;
1265 friend class XFTransactionWriteHandler
;
1266 friend class DBBlobIndexTest
;
1267 friend class WriteUnpreparedTransactionTest_RecoveryTest_Test
;
1270 struct CompactionState
;
1271 struct PrepickedCompaction
;
1272 struct PurgeFileInfo
;
1274 struct WriteContext
{
1275 SuperVersionContext superversion_context
;
1276 autovector
<MemTable
*> memtables_to_free_
;
1278 explicit WriteContext(bool create_superversion
= false)
1279 : superversion_context(create_superversion
) {}
1282 superversion_context
.Clean();
1283 for (auto& m
: memtables_to_free_
) {
1289 struct LogFileNumberSize
{
1290 explicit LogFileNumberSize(uint64_t _number
) : number(_number
) {}
1291 void AddSize(uint64_t new_size
) { size
+= new_size
; }
1294 bool getting_flushed
= false;
1297 struct LogWriterNumber
{
1298 // pass ownership of _writer
1299 LogWriterNumber(uint64_t _number
, log::Writer
* _writer
)
1300 : number(_number
), writer(_writer
) {}
1302 log::Writer
* ReleaseWriter() {
1307 Status
ClearWriter() {
1308 Status s
= writer
->WriteBuffer();
1315 // Visual Studio doesn't support deque's member to be noncopyable because
1316 // of a std::unique_ptr as a member.
1317 log::Writer
* writer
; // own
1318 // true for some prefix of logs_
1319 bool getting_synced
= false;
1322 // PurgeFileInfo is a structure to hold information of files to be deleted in
1324 struct PurgeFileInfo
{
1326 std::string dir_to_sync
;
1330 PurgeFileInfo(std::string fn
, std::string d
, FileType t
, uint64_t num
,
1332 : fname(fn
), dir_to_sync(d
), type(t
), number(num
), job_id(jid
) {}
1335 // Argument required by background flush thread.
1338 : cfd_(nullptr), max_memtable_id_(0), superversion_context_(nullptr) {}
1339 BGFlushArg(ColumnFamilyData
* cfd
, uint64_t max_memtable_id
,
1340 SuperVersionContext
* superversion_context
)
1342 max_memtable_id_(max_memtable_id
),
1343 superversion_context_(superversion_context
) {}
1345 // Column family to flush.
1346 ColumnFamilyData
* cfd_
;
1347 // Maximum ID of memtable to flush. In this column family, memtables with
1348 // IDs smaller than this value must be flushed before this flush completes.
1349 uint64_t max_memtable_id_
;
1350 // Pointer to a SuperVersionContext object. After flush completes, RocksDB
1351 // installs a new superversion for the column family. This operation
1352 // requires a SuperVersionContext object (currently embedded in JobContext).
1353 SuperVersionContext
* superversion_context_
;
1356 // Argument passed to flush thread.
1357 struct FlushThreadArg
{
1360 Env::Priority thread_pri_
;
1363 // Information for a manual compaction
1364 struct ManualCompactionState
{
1365 ColumnFamilyData
* cfd
;
1368 uint32_t output_path_id
;
1371 bool in_progress
; // compaction request being processed?
1372 bool incomplete
; // only part of requested range compacted
1373 bool exclusive
; // current behavior of only one manual
1374 bool disallow_trivial_move
; // Force actual compaction to run
1375 const InternalKey
* begin
; // nullptr means beginning of key range
1376 const InternalKey
* end
; // nullptr means end of key range
1377 InternalKey
* manual_end
; // how far we are compacting
1378 InternalKey tmp_storage
; // Used to keep track of compaction progress
1379 InternalKey tmp_storage1
; // Used to keep track of compaction progress
1381 struct PrepickedCompaction
{
1382 // background compaction takes ownership of `compaction`.
1383 Compaction
* compaction
;
1384 // caller retains ownership of `manual_compaction_state` as it is reused
1385 // across background compactions.
1386 ManualCompactionState
* manual_compaction_state
; // nullptr if non-manual
1387 // task limiter token is requested during compaction picking.
1388 std::unique_ptr
<TaskLimiterToken
> task_token
;
1391 struct CompactionArg
{
1392 // caller retains ownership of `db`.
1394 // background compaction takes ownership of `prepicked_compaction`.
1395 PrepickedCompaction
* prepicked_compaction
;
1398 // Initialize the built-in column family for persistent stats. Depending on
1399 // whether on-disk persistent stats have been enabled before, it may either
1400 // create a new column family and column family handle or just a column family
1402 // Required: DB mutex held
1403 Status
InitPersistStatsColumnFamily();
1405 // Persistent Stats column family has two format version key which are used
1406 // for compatibility check. Write format version if it's created for the
1407 // first time, read format version and check compatibility if recovering
1408 // from disk. This function requires DB mutex held at entrance but may
1409 // release and re-acquire DB mutex in the process.
1410 // Required: DB mutex held
1411 Status
PersistentStatsProcessFormatVersion();
1413 Status
ResumeImpl(DBRecoverContext context
);
1415 void MaybeIgnoreError(Status
* s
) const;
1417 const Status
CreateArchivalDirectory();
1419 Status
CreateColumnFamilyImpl(const ColumnFamilyOptions
& cf_options
,
1420 const std::string
& cf_name
,
1421 ColumnFamilyHandle
** handle
);
1423 Status
DropColumnFamilyImpl(ColumnFamilyHandle
* column_family
);
1425 // Delete any unneeded files and stale in-memory entries.
1426 void DeleteObsoleteFiles();
1427 // Delete obsolete files and log status and information of file deletion
1428 void DeleteObsoleteFileImpl(int job_id
, const std::string
& fname
,
1429 const std::string
& path_to_sync
, FileType type
,
1432 // Background process needs to call
1433 // auto x = CaptureCurrentFileNumberInPendingOutputs()
1434 // auto file_num = versions_->NewFileNumber();
1436 // ReleaseFileNumberFromPendingOutputs(x)
1437 // This will protect any file with number `file_num` or greater from being
1438 // deleted while <do something> is running.
1440 // This function will capture current file number and append it to
1441 // pending_outputs_. This will prevent any background process to delete any
1442 // file created after this point.
1443 std::list
<uint64_t>::iterator
CaptureCurrentFileNumberInPendingOutputs();
1444 // This function should be called with the result of
1445 // CaptureCurrentFileNumberInPendingOutputs(). It then marks that any file
1446 // created between the calls CaptureCurrentFileNumberInPendingOutputs() and
1447 // ReleaseFileNumberFromPendingOutputs() can now be deleted (if it's not live
1448 // and blocked by any other pending_outputs_ calls)
1449 void ReleaseFileNumberFromPendingOutputs(
1450 std::unique_ptr
<std::list
<uint64_t>::iterator
>& v
);
1452 IOStatus
SyncClosedLogs(JobContext
* job_context
);
1454 // Flush the in-memory write buffer to storage. Switches to a new
1455 // log-file/memtable and writes a new descriptor iff successful. Then
1456 // installs a new super version for the column family.
1457 Status
FlushMemTableToOutputFile(
1458 ColumnFamilyData
* cfd
, const MutableCFOptions
& mutable_cf_options
,
1459 bool* madeProgress
, JobContext
* job_context
,
1460 SuperVersionContext
* superversion_context
,
1461 std::vector
<SequenceNumber
>& snapshot_seqs
,
1462 SequenceNumber earliest_write_conflict_snapshot
,
1463 SnapshotChecker
* snapshot_checker
, LogBuffer
* log_buffer
,
1464 Env::Priority thread_pri
);
1466 // Flush the memtables of (multiple) column families to multiple files on
1467 // persistent storage.
1468 Status
FlushMemTablesToOutputFiles(
1469 const autovector
<BGFlushArg
>& bg_flush_args
, bool* made_progress
,
1470 JobContext
* job_context
, LogBuffer
* log_buffer
, Env::Priority thread_pri
);
1472 Status
AtomicFlushMemTablesToOutputFiles(
1473 const autovector
<BGFlushArg
>& bg_flush_args
, bool* made_progress
,
1474 JobContext
* job_context
, LogBuffer
* log_buffer
, Env::Priority thread_pri
);
1476 // REQUIRES: log_numbers are sorted in ascending order
1477 // corrupted_log_found is set to true if we recover from a corrupted log file.
1478 Status
RecoverLogFiles(const std::vector
<uint64_t>& log_numbers
,
1479 SequenceNumber
* next_sequence
, bool read_only
,
1480 bool* corrupted_log_found
);
1482 // The following two methods are used to flush a memtable to
1483 // storage. The first one is used at database RecoveryTime (when the
1484 // database is opened) and is heavyweight because it holds the mutex
1485 // for the entire period. The second method WriteLevel0Table supports
1486 // concurrent flush memtables to storage.
1487 Status
WriteLevel0TableForRecovery(int job_id
, ColumnFamilyData
* cfd
,
1488 MemTable
* mem
, VersionEdit
* edit
);
1490 // Restore alive_log_files_ and total_log_size_ after recovery.
1491 // It needs to run only when there's no flush during recovery
1492 // (e.g. avoid_flush_during_recovery=true). May also trigger flush
1493 // in case total_log_size > max_total_wal_size.
1494 Status
RestoreAliveLogFiles(const std::vector
<uint64_t>& log_numbers
);
1496 // num_bytes: for slowdown case, delay time is calculated based on
1497 // `num_bytes` going through.
1498 Status
DelayWrite(uint64_t num_bytes
, const WriteOptions
& write_options
);
1500 Status
ThrottleLowPriWritesIfNeeded(const WriteOptions
& write_options
,
1501 WriteBatch
* my_batch
);
1503 // REQUIRES: mutex locked and in write thread.
1504 Status
ScheduleFlushes(WriteContext
* context
);
1506 void MaybeFlushStatsCF(autovector
<ColumnFamilyData
*>* cfds
);
1508 Status
TrimMemtableHistory(WriteContext
* context
);
1510 Status
SwitchMemtable(ColumnFamilyData
* cfd
, WriteContext
* context
);
1512 void SelectColumnFamiliesForAtomicFlush(autovector
<ColumnFamilyData
*>* cfds
);
1514 // Force current memtable contents to be flushed.
1515 Status
FlushMemTable(ColumnFamilyData
* cfd
, const FlushOptions
& options
,
1516 FlushReason flush_reason
, bool writes_stopped
= false);
1518 Status
AtomicFlushMemTables(
1519 const autovector
<ColumnFamilyData
*>& column_family_datas
,
1520 const FlushOptions
& options
, FlushReason flush_reason
,
1521 bool writes_stopped
= false);
1523 // Wait until flushing this column family won't stall writes
1524 Status
WaitUntilFlushWouldNotStallWrites(ColumnFamilyData
* cfd
,
1525 bool* flush_needed
);
1527 // Wait for memtable flushed.
1528 // If flush_memtable_id is non-null, wait until the memtable with the ID
1529 // gets flush. Otherwise, wait until the column family don't have any
1530 // memtable pending flush.
1531 // resuming_from_bg_err indicates whether the caller is attempting to resume
1532 // from background error.
1533 Status
WaitForFlushMemTable(ColumnFamilyData
* cfd
,
1534 const uint64_t* flush_memtable_id
= nullptr,
1535 bool resuming_from_bg_err
= false) {
1536 return WaitForFlushMemTables({cfd
}, {flush_memtable_id
},
1537 resuming_from_bg_err
);
1539 // Wait for memtables to be flushed for multiple column families.
1540 Status
WaitForFlushMemTables(
1541 const autovector
<ColumnFamilyData
*>& cfds
,
1542 const autovector
<const uint64_t*>& flush_memtable_ids
,
1543 bool resuming_from_bg_err
);
1545 inline void WaitForPendingWrites() {
1546 mutex_
.AssertHeld();
1547 TEST_SYNC_POINT("DBImpl::WaitForPendingWrites:BeforeBlock");
1548 // In case of pipelined write is enabled, wait for all pending memtable
1550 if (immutable_db_options_
.enable_pipelined_write
) {
1551 // Memtable writers may call DB::Get in case max_successive_merges > 0,
1552 // which may lock mutex. Unlocking mutex here to avoid deadlock.
1554 write_thread_
.WaitForMemTableWriters();
1558 if (!immutable_db_options_
.unordered_write
) {
1559 // Then the writes are finished before the next write group starts
1563 // Wait for the ones who already wrote to the WAL to finish their
1565 if (pending_memtable_writes_
.load() != 0) {
1566 std::unique_lock
<std::mutex
> guard(switch_mutex_
);
1567 switch_cv_
.wait(guard
,
1568 [&] { return pending_memtable_writes_
.load() == 0; });
1572 // REQUIRES: mutex locked and in write thread.
1573 void AssignAtomicFlushSeq(const autovector
<ColumnFamilyData
*>& cfds
);
1575 // REQUIRES: mutex locked and in write thread.
1576 Status
SwitchWAL(WriteContext
* write_context
);
1578 // REQUIRES: mutex locked and in write thread.
1579 Status
HandleWriteBufferFull(WriteContext
* write_context
);
1581 // REQUIRES: mutex locked
1582 Status
PreprocessWrite(const WriteOptions
& write_options
, bool* need_log_sync
,
1583 WriteContext
* write_context
);
1585 WriteBatch
* MergeBatch(const WriteThread::WriteGroup
& write_group
,
1586 WriteBatch
* tmp_batch
, size_t* write_with_wal
,
1587 WriteBatch
** to_be_cached_state
);
1589 IOStatus
WriteToWAL(const WriteBatch
& merged_batch
, log::Writer
* log_writer
,
1590 uint64_t* log_used
, uint64_t* log_size
);
1592 IOStatus
WriteToWAL(const WriteThread::WriteGroup
& write_group
,
1593 log::Writer
* log_writer
, uint64_t* log_used
,
1594 bool need_log_sync
, bool need_log_dir_sync
,
1595 SequenceNumber sequence
);
1597 IOStatus
ConcurrentWriteToWAL(const WriteThread::WriteGroup
& write_group
,
1599 SequenceNumber
* last_sequence
, size_t seq_inc
);
1601 // Used by WriteImpl to update bg_error_ if paranoid check is enabled.
1602 // Caller must hold mutex_.
1603 void WriteStatusCheckOnLocked(const Status
& status
);
1605 // Used by WriteImpl to update bg_error_ if paranoid check is enabled.
1606 void WriteStatusCheck(const Status
& status
);
1608 // Used by WriteImpl to update bg_error_ when IO error happens, e.g., write
1609 // WAL, sync WAL fails, if paranoid check is enabled.
1610 void IOStatusCheck(const IOStatus
& status
);
1612 // Used by WriteImpl to update bg_error_ in case of memtable insert error.
1613 void MemTableInsertStatusCheck(const Status
& memtable_insert_status
);
1615 #ifndef ROCKSDB_LITE
1617 Status
CompactFilesImpl(const CompactionOptions
& compact_options
,
1618 ColumnFamilyData
* cfd
, Version
* version
,
1619 const std::vector
<std::string
>& input_file_names
,
1620 std::vector
<std::string
>* const output_file_names
,
1621 const int output_level
, int output_path_id
,
1622 JobContext
* job_context
, LogBuffer
* log_buffer
,
1623 CompactionJobInfo
* compaction_job_info
);
1625 // Wait for current IngestExternalFile() calls to finish.
1626 // REQUIRES: mutex_ held
1627 void WaitForIngestFile();
1630 // IngestExternalFile is not supported in ROCKSDB_LITE so this function
1632 void WaitForIngestFile() {}
1633 #endif // ROCKSDB_LITE
1635 ColumnFamilyData
* GetColumnFamilyDataByName(const std::string
& cf_name
);
1637 void MaybeScheduleFlushOrCompaction();
1639 // A flush request specifies the column families to flush as well as the
1640 // largest memtable id to persist for each column family. Once all the
1641 // memtables whose IDs are smaller than or equal to this per-column-family
1642 // specified value, this flush request is considered to have completed its
1643 // work of flushing this column family. After completing the work for all
1644 // column families in this request, this flush is considered complete.
1645 typedef std::vector
<std::pair
<ColumnFamilyData
*, uint64_t>> FlushRequest
;
1647 void GenerateFlushRequest(const autovector
<ColumnFamilyData
*>& cfds
,
1650 void SchedulePendingFlush(const FlushRequest
& req
, FlushReason flush_reason
);
1652 void SchedulePendingCompaction(ColumnFamilyData
* cfd
);
1653 void SchedulePendingPurge(std::string fname
, std::string dir_to_sync
,
1654 FileType type
, uint64_t number
, int job_id
);
1655 static void BGWorkCompaction(void* arg
);
1656 // Runs a pre-chosen universal compaction involving bottom level in a
1657 // separate, bottom-pri thread pool.
1658 static void BGWorkBottomCompaction(void* arg
);
1659 static void BGWorkFlush(void* arg
);
1660 static void BGWorkPurge(void* arg
);
1661 static void UnscheduleCompactionCallback(void* arg
);
1662 static void UnscheduleFlushCallback(void* arg
);
1663 void BackgroundCallCompaction(PrepickedCompaction
* prepicked_compaction
,
1664 Env::Priority thread_pri
);
1665 void BackgroundCallFlush(Env::Priority thread_pri
);
1666 void BackgroundCallPurge();
1667 Status
BackgroundCompaction(bool* madeProgress
, JobContext
* job_context
,
1668 LogBuffer
* log_buffer
,
1669 PrepickedCompaction
* prepicked_compaction
,
1670 Env::Priority thread_pri
);
1671 Status
BackgroundFlush(bool* madeProgress
, JobContext
* job_context
,
1672 LogBuffer
* log_buffer
, FlushReason
* reason
,
1673 Env::Priority thread_pri
);
1675 bool EnoughRoomForCompaction(ColumnFamilyData
* cfd
,
1676 const std::vector
<CompactionInputFiles
>& inputs
,
1677 bool* sfm_bookkeeping
, LogBuffer
* log_buffer
);
1679 // Request compaction tasks token from compaction thread limiter.
1680 // It always succeeds if force = true or limiter is disable.
1681 bool RequestCompactionToken(ColumnFamilyData
* cfd
, bool force
,
1682 std::unique_ptr
<TaskLimiterToken
>* token
,
1683 LogBuffer
* log_buffer
);
1685 // Schedule background tasks
1686 void StartPeriodicWorkScheduler();
1688 void PrintStatistics();
1690 size_t EstimateInMemoryStatsHistorySize() const;
1692 // Return the minimum empty level that could hold the total data in the
1693 // input level. Return the input level, if such level could not be found.
1694 int FindMinimumEmptyLevelFitting(ColumnFamilyData
* cfd
,
1695 const MutableCFOptions
& mutable_cf_options
,
1698 // Move the files in the input level to the target level.
1699 // If target_level < 0, automatically calculate the minimum level that could
1700 // hold the data set.
1701 Status
ReFitLevel(ColumnFamilyData
* cfd
, int level
, int target_level
= -1);
1703 // helper functions for adding and removing from flush & compaction queues
1704 void AddToCompactionQueue(ColumnFamilyData
* cfd
);
1705 ColumnFamilyData
* PopFirstFromCompactionQueue();
1706 FlushRequest
PopFirstFromFlushQueue();
1708 // Pick the first unthrottled compaction with task token from queue.
1709 ColumnFamilyData
* PickCompactionFromQueue(
1710 std::unique_ptr
<TaskLimiterToken
>* token
, LogBuffer
* log_buffer
);
1712 // helper function to call after some of the logs_ were synced
1713 Status
MarkLogsSynced(uint64_t up_to
, bool synced_dir
);
1714 // WALs with log number up to up_to are not synced successfully.
1715 void MarkLogsNotSynced(uint64_t up_to
);
1717 SnapshotImpl
* GetSnapshotImpl(bool is_write_conflict_boundary
,
1720 uint64_t GetMaxTotalWalSize() const;
1722 FSDirectory
* GetDataDir(ColumnFamilyData
* cfd
, size_t path_id
) const;
1724 Status
CloseHelper();
1726 void WaitForBackgroundWork();
1728 // Background threads call this function, which is just a wrapper around
1729 // the InstallSuperVersion() function. Background threads carry
1730 // sv_context which can have new_superversion already
1732 // All ColumnFamily state changes go through this function. Here we analyze
1733 // the new state and we schedule background work if we detect that the new
1734 // state needs flush or compaction.
1735 void InstallSuperVersionAndScheduleWork(
1736 ColumnFamilyData
* cfd
, SuperVersionContext
* sv_context
,
1737 const MutableCFOptions
& mutable_cf_options
);
1739 bool GetIntPropertyInternal(ColumnFamilyData
* cfd
,
1740 const DBPropertyInfo
& property_info
,
1741 bool is_locked
, uint64_t* value
);
1742 bool GetPropertyHandleOptionsStatistics(std::string
* value
);
1744 bool HasPendingManualCompaction();
1745 bool HasExclusiveManualCompaction();
1746 void AddManualCompaction(ManualCompactionState
* m
);
1747 void RemoveManualCompaction(ManualCompactionState
* m
);
1748 bool ShouldntRunManualCompaction(ManualCompactionState
* m
);
1749 bool HaveManualCompaction(ColumnFamilyData
* cfd
);
1750 bool MCOverlap(ManualCompactionState
* m
, ManualCompactionState
* m1
);
1751 #ifndef ROCKSDB_LITE
1752 void BuildCompactionJobInfo(const ColumnFamilyData
* cfd
, Compaction
* c
,
1754 const CompactionJobStats
& compaction_job_stats
,
1755 const int job_id
, const Version
* current
,
1756 CompactionJobInfo
* compaction_job_info
) const;
1757 // Reserve the next 'num' file numbers for to-be-ingested external SST files,
1758 // and return the current file_number in 'next_file_number'.
1759 // Write a version edit to the MANIFEST.
1760 Status
ReserveFileNumbersBeforeIngestion(
1761 ColumnFamilyData
* cfd
, uint64_t num
,
1762 std::unique_ptr
<std::list
<uint64_t>::iterator
>& pending_output_elem
,
1763 uint64_t* next_file_number
);
1764 #endif //! ROCKSDB_LITE
1766 bool ShouldPurge(uint64_t file_number
) const;
1767 void MarkAsGrabbedForPurge(uint64_t file_number
);
1769 size_t GetWalPreallocateBlockSize(uint64_t write_buffer_size
) const;
1770 Env::WriteLifeTimeHint
CalculateWALWriteHint() { return Env::WLTH_SHORT
; }
1772 IOStatus
CreateWAL(uint64_t log_file_num
, uint64_t recycle_log_number
,
1773 size_t preallocate_block_size
, log::Writer
** new_log
);
1775 // Validate self-consistency of DB options
1776 static Status
ValidateOptions(const DBOptions
& db_options
);
1777 // Validate self-consistency of DB options and its consistency with cf options
1778 static Status
ValidateOptions(
1779 const DBOptions
& db_options
,
1780 const std::vector
<ColumnFamilyDescriptor
>& column_families
);
1782 // Utility function to do some debug validation and sort the given vector
1784 void PrepareMultiGetKeys(
1785 const size_t num_keys
, bool sorted
,
1786 autovector
<KeyContext
*, MultiGetContext::MAX_BATCH_SIZE
>* key_ptrs
);
1788 // A structure to hold the information required to process MultiGet of keys
1789 // belonging to one column family. For a multi column family MultiGet, there
1790 // will be a container of these objects.
1791 struct MultiGetColumnFamilyData
{
1792 ColumnFamilyHandle
* cf
;
1793 ColumnFamilyData
* cfd
;
1795 // For the batched MultiGet which relies on sorted keys, start specifies
1796 // the index of first key belonging to this column family in the sorted
1800 // For the batched MultiGet case, num_keys specifies the number of keys
1801 // belonging to this column family in the sorted list
1804 // SuperVersion for the column family obtained in a manner that ensures a
1805 // consistent view across all column families in the DB
1806 SuperVersion
* super_version
;
1807 MultiGetColumnFamilyData(ColumnFamilyHandle
* column_family
,
1809 : cf(column_family
),
1810 cfd(static_cast<ColumnFamilyHandleImpl
*>(cf
)->cfd()),
1813 super_version(sv
) {}
1815 MultiGetColumnFamilyData(ColumnFamilyHandle
* column_family
, size_t first
,
1816 size_t count
, SuperVersion
* sv
)
1817 : cf(column_family
),
1818 cfd(static_cast<ColumnFamilyHandleImpl
*>(cf
)->cfd()),
1821 super_version(sv
) {}
1823 MultiGetColumnFamilyData() = default;
1826 // A common function to obtain a consistent snapshot, which can be implicit
1827 // if the user doesn't specify a snapshot in read_options, across
1828 // multiple column families for MultiGet. It will attempt to get an implicit
1829 // snapshot without acquiring the db_mutes, but will give up after a few
1830 // tries and acquire the mutex if a memtable flush happens. The template
1831 // allows both the batched and non-batched MultiGet to call this with
1832 // either an std::unordered_map or autovector of column families.
1834 // If callback is non-null, the callback is refreshed with the snapshot
1837 // A return value of true indicates that the SuperVersions were obtained
1838 // from the ColumnFamilyData, whereas false indicates they are thread
1841 bool MultiCFSnapshot(
1842 const ReadOptions
& read_options
, ReadCallback
* callback
,
1843 std::function
<MultiGetColumnFamilyData
*(typename
T::iterator
&)>&
1845 T
* cf_list
, SequenceNumber
* snapshot
);
1847 // The actual implementation of the batching MultiGet. The caller is expected
1848 // to have acquired the SuperVersion and pass in a snapshot sequence number
1849 // in order to construct the LookupKeys. The start_key and num_keys specify
1850 // the range of keys in the sorted_keys vector for a single column family.
1851 Status
MultiGetImpl(
1852 const ReadOptions
& read_options
, size_t start_key
, size_t num_keys
,
1853 autovector
<KeyContext
*, MultiGetContext::MAX_BATCH_SIZE
>* sorted_keys
,
1854 SuperVersion
* sv
, SequenceNumber snap_seqnum
, ReadCallback
* callback
,
1855 bool* is_blob_index
);
1857 Status
DisableFileDeletionsWithLock();
1859 // table_cache_ provides its own synchronization
1860 std::shared_ptr
<Cache
> table_cache_
;
1862 // Lock over the persistent DB state. Non-nullptr iff successfully acquired.
1865 // In addition to mutex_, log_write_mutex_ protected writes to stats_history_
1866 InstrumentedMutex stats_history_mutex_
;
1867 // In addition to mutex_, log_write_mutex_ protected writes to logs_ and
1868 // logfile_number_. With two_write_queues it also protects alive_log_files_,
1869 // and log_empty_. Refer to the definition of each variable below for more
1871 // Note: to avoid dealock, if needed to acquire both log_write_mutex_ and
1872 // mutex_, the order should be first mutex_ and then log_write_mutex_.
1873 InstrumentedMutex log_write_mutex_
;
1875 std::atomic
<bool> shutting_down_
;
1877 // If zero, manual compactions are allowed to proceed. If non-zero, manual
1878 // compactions may still be running, but will quickly fail with
1879 // `Status::Incomplete`. The value indicates how many threads have paused
1880 // manual compactions. It is accessed in read mode outside the DB mutex in
1881 // compaction code paths.
1882 std::atomic
<int> manual_compaction_paused_
;
1884 // This condition variable is signaled on these conditions:
1885 // * whenever bg_compaction_scheduled_ goes down to 0
1886 // * if AnyManualCompaction, whenever a compaction finishes, even if it hasn't
1887 // made any progress
1888 // * whenever a compaction made any progress
1889 // * whenever bg_flush_scheduled_ or bg_purge_scheduled_ value decreases
1890 // (i.e. whenever a flush is done, even if it didn't make any progress)
1891 // * whenever there is an error in background purge, flush or compaction
1892 // * whenever num_running_ingest_file_ goes to 0.
1893 // * whenever pending_purge_obsolete_files_ goes to 0.
1894 // * whenever disable_delete_obsolete_files_ goes to 0.
1895 // * whenever SetOptions successfully updates options.
1896 // * whenever a column family is dropped.
1897 InstrumentedCondVar bg_cv_
;
1898 // Writes are protected by locking both mutex_ and log_write_mutex_, and reads
1899 // must be under either mutex_ or log_write_mutex_. Since after ::Open,
1900 // logfile_number_ is currently updated only in write_thread_, it can be read
1901 // from the same write_thread_ without any locks.
1902 uint64_t logfile_number_
;
1903 std::deque
<uint64_t>
1904 log_recycle_files_
; // a list of log files that we can recycle
1905 bool log_dir_synced_
;
1906 // Without two_write_queues, read and writes to log_empty_ are protected by
1907 // mutex_. Since it is currently updated/read only in write_thread_, it can be
1908 // accessed from the same write_thread_ without any locks. With
1909 // two_write_queues writes, where it can be updated in different threads,
1910 // read and writes are protected by log_write_mutex_ instead. This is to avoid
1911 // expesnive mutex_ lock during WAL write, which update log_empty_.
1914 ColumnFamilyHandleImpl
* persist_stats_cf_handle_
;
1916 bool persistent_stats_cfd_exists_
= true;
1918 // Without two_write_queues, read and writes to alive_log_files_ are
1919 // protected by mutex_. However since back() is never popped, and push_back()
1920 // is done only from write_thread_, the same thread can access the item
1921 // reffered by back() without mutex_. With two_write_queues_, writes
1922 // are protected by locking both mutex_ and log_write_mutex_, and reads must
1923 // be under either mutex_ or log_write_mutex_.
1924 std::deque
<LogFileNumberSize
> alive_log_files_
;
1925 // Log files that aren't fully synced, and the current log file.
1927 // - push_back() is done from write_thread_ with locked mutex_ and
1929 // - pop_front() is done from any thread with locked mutex_ and
1931 // - reads are done with either locked mutex_ or log_write_mutex_
1932 // - back() and items with getting_synced=true are not popped,
1933 // - The same thread that sets getting_synced=true will reset it.
1934 // - it follows that the object referred by back() can be safely read from
1935 // the write_thread_ without using mutex
1936 // - it follows that the items with getting_synced=true can be safely read
1937 // from the same thread that has set getting_synced=true
1938 std::deque
<LogWriterNumber
> logs_
;
1939 // Signaled when getting_synced becomes false for some of the logs_.
1940 InstrumentedCondVar log_sync_cv_
;
1941 // This is the app-level state that is written to the WAL but will be used
1942 // only during recovery. Using this feature enables not writing the state to
1943 // memtable on normal writes and hence improving the throughput. Each new
1944 // write of the state will replace the previous state entirely even if the
1945 // keys in the two consecuitive states do not overlap.
1946 // It is protected by log_write_mutex_ when two_write_queues_ is enabled.
1947 // Otherwise only the heaad of write_thread_ can access it.
1948 WriteBatch cached_recoverable_state_
;
1949 std::atomic
<bool> cached_recoverable_state_empty_
= {true};
1950 std::atomic
<uint64_t> total_log_size_
;
1952 // If this is non-empty, we need to delete these log files in background
1953 // threads. Protected by db mutex.
1954 autovector
<log::Writer
*> logs_to_free_
;
1956 bool is_snapshot_supported_
;
1958 std::map
<uint64_t, std::map
<std::string
, uint64_t>> stats_history_
;
1960 std::map
<std::string
, uint64_t> stats_slice_
;
1962 bool stats_slice_initialized_
= false;
1964 Directories directories_
;
1966 WriteBufferManager
* write_buffer_manager_
;
1968 WriteThread write_thread_
;
1969 WriteBatch tmp_batch_
;
1970 // The write thread when the writers have no memtable write. This will be used
1971 // in 2PC to batch the prepares separately from the serial commit.
1972 WriteThread nonmem_write_thread_
;
1974 WriteController write_controller_
;
1976 // Size of the last batch group. In slowdown mode, next write needs to
1977 // sleep if it uses up the quota.
1978 // Note: This is to protect memtable and compaction. If the batch only writes
1979 // to the WAL its size need not to be included in this.
1980 uint64_t last_batch_group_size_
;
1982 FlushScheduler flush_scheduler_
;
1984 TrimHistoryScheduler trim_history_scheduler_
;
1986 SnapshotList snapshots_
;
1988 // For each background job, pending_outputs_ keeps the current file number at
1989 // the time that background job started.
1990 // FindObsoleteFiles()/PurgeObsoleteFiles() never deletes any file that has
1991 // number bigger than any of the file number in pending_outputs_. Since file
1992 // numbers grow monotonically, this also means that pending_outputs_ is always
1993 // sorted. After a background job is done executing, its file number is
1994 // deleted from pending_outputs_, which allows PurgeObsoleteFiles() to clean
1996 // State is protected with db mutex.
1997 std::list
<uint64_t> pending_outputs_
;
1999 // flush_queue_ and compaction_queue_ hold column families that we need to
2000 // flush and compact, respectively.
2001 // A column family is inserted into flush_queue_ when it satisfies condition
2002 // cfd->imm()->IsFlushPending()
2003 // A column family is inserted into compaction_queue_ when it satisfied
2004 // condition cfd->NeedsCompaction()
2005 // Column families in this list are all Ref()-erenced
2006 // TODO(icanadi) Provide some kind of ReferencedColumnFamily class that will
2007 // do RAII on ColumnFamilyData
2008 // Column families are in this queue when they need to be flushed or
2009 // compacted. Consumers of these queues are flush and compaction threads. When
2010 // column family is put on this queue, we increase unscheduled_flushes_ and
2011 // unscheduled_compactions_. When these variables are bigger than zero, that
2012 // means we need to schedule background threads for flush and compaction.
2013 // Once the background threads are scheduled, we decrease unscheduled_flushes_
2014 // and unscheduled_compactions_. That way we keep track of number of
2015 // compaction and flush threads we need to schedule. This scheduling is done
2016 // in MaybeScheduleFlushOrCompaction()
2017 // invariant(column family present in flush_queue_ <==>
2018 // ColumnFamilyData::pending_flush_ == true)
2019 std::deque
<FlushRequest
> flush_queue_
;
2020 // invariant(column family present in compaction_queue_ <==>
2021 // ColumnFamilyData::pending_compaction_ == true)
2022 std::deque
<ColumnFamilyData
*> compaction_queue_
;
2024 // A map to store file numbers and filenames of the files to be purged
2025 std::unordered_map
<uint64_t, PurgeFileInfo
> purge_files_
;
2027 // A vector to store the file numbers that have been assigned to certain
2028 // JobContext. Current implementation tracks table and blob files only.
2029 std::unordered_set
<uint64_t> files_grabbed_for_purge_
;
2031 // A queue to store log writers to close
2032 std::deque
<log::Writer
*> logs_to_free_queue_
;
2033 std::deque
<SuperVersion
*> superversions_to_free_queue_
;
2034 int unscheduled_flushes_
;
2035 int unscheduled_compactions_
;
2037 // count how many background compactions are running or have been scheduled in
2039 int bg_bottom_compaction_scheduled_
;
2041 // count how many background compactions are running or have been scheduled
2042 int bg_compaction_scheduled_
;
2044 // stores the number of compactions are currently running
2045 int num_running_compactions_
;
2047 // number of background memtable flush jobs, submitted to the HIGH pool
2048 int bg_flush_scheduled_
;
2050 // stores the number of flushes are currently running
2051 int num_running_flushes_
;
2053 // number of background obsolete file purge jobs, submitted to the HIGH pool
2054 int bg_purge_scheduled_
;
2056 std::deque
<ManualCompactionState
*> manual_compaction_dequeue_
;
2058 // shall we disable deletion of obsolete files
2059 // if 0 the deletion is enabled.
2060 // if non-zero, files will not be getting deleted
2061 // This enables two different threads to call
2062 // EnableFileDeletions() and DisableFileDeletions()
2063 // without any synchronization
2064 int disable_delete_obsolete_files_
;
2066 // Number of times FindObsoleteFiles has found deletable files and the
2067 // corresponding call to PurgeObsoleteFiles has not yet finished.
2068 int pending_purge_obsolete_files_
;
2070 // last time when DeleteObsoleteFiles with full scan was executed. Originally
2071 // initialized with startup time.
2072 uint64_t delete_obsolete_files_last_run_
;
2074 // last time stats were dumped to LOG
2075 std::atomic
<uint64_t> last_stats_dump_time_microsec_
;
2077 // The thread that wants to switch memtable, can wait on this cv until the
2078 // pending writes to memtable finishes.
2079 std::condition_variable switch_cv_
;
2080 // The mutex used by switch_cv_. mutex_ should be acquired beforehand.
2081 std::mutex switch_mutex_
;
2082 // Number of threads intending to write to memtable
2083 std::atomic
<size_t> pending_memtable_writes_
= {};
2085 // Each flush or compaction gets its own job id. this counter makes sure
2087 std::atomic
<int> next_job_id_
;
2089 // A flag indicating whether the current rocksdb database has any
2090 // data that is not yet persisted into either WAL or SST file.
2091 // Used when disableWAL is true.
2092 std::atomic
<bool> has_unpersisted_data_
;
2094 // if an attempt was made to flush all column families that
2095 // the oldest log depends on but uncommitted data in the oldest
2096 // log prevents the log from being released.
2097 // We must attempt to free the dependent memtables again
2098 // at a later time after the transaction in the oldest
2099 // log is fully commited.
2100 bool unable_to_release_oldest_log_
;
2102 static const int KEEP_LOG_FILE_NUM
= 1000;
2103 // MSVC version 1800 still does not have constexpr for ::max()
2104 static const uint64_t kNoTimeOut
= port::kMaxUint64
;
2106 std::string db_absolute_path_
;
2108 // Number of running IngestExternalFile() or CreateColumnFamilyWithImport()
2110 // REQUIRES: mutex held
2111 int num_running_ingest_file_
;
2113 #ifndef ROCKSDB_LITE
2114 WalManager wal_manager_
;
2115 #endif // ROCKSDB_LITE
2117 // Unified interface for logging events
2118 EventLogger event_logger_
;
2120 // A value of > 0 temporarily disables scheduling of background work
2121 int bg_work_paused_
;
2123 // A value of > 0 temporarily disables scheduling of background compaction
2124 int bg_compaction_paused_
;
2126 // Guard against multiple concurrent refitting
2127 bool refitting_level_
;
2129 // Indicate DB was opened successfully
2130 bool opened_successfully_
;
2132 // The min threshold to triggere bottommost compaction for removing
2133 // garbages, among all column families.
2134 SequenceNumber bottommost_files_mark_threshold_
= kMaxSequenceNumber
;
2136 LogsWithPrepTracker logs_with_prep_tracker_
;
2138 // Callback for compaction to check if a key is visible to a snapshot.
2139 // REQUIRES: mutex held
2140 std::unique_ptr
<SnapshotChecker
> snapshot_checker_
;
2142 // Callback for when the cached_recoverable_state_ is written to memtable
2143 // Only to be set during initialization
2144 std::unique_ptr
<PreReleaseCallback
> recoverable_state_pre_release_callback_
;
2146 #ifndef ROCKSDB_LITE
2147 // Scheduler to run DumpStats(), PersistStats(), and FlushInfoLog().
2148 // Currently, it always use a global instance from
2149 // PeriodicWorkScheduler::Default(). Only in unittest, it can be overrided by
2150 // PeriodicWorkTestScheduler.
2151 PeriodicWorkScheduler
* periodic_work_scheduler_
;
2154 // When set, we use a separate queue for writes that don't write to memtable.
2155 // In 2PC these are the writes at Prepare phase.
2156 const bool two_write_queues_
;
2157 const bool manual_wal_flush_
;
2159 // LastSequence also indicates last published sequence visibile to the
2160 // readers. Otherwise LastPublishedSequence should be used.
2161 const bool last_seq_same_as_publish_seq_
;
2162 // It indicates that a customized gc algorithm must be used for
2163 // flush/compaction and if it is not provided vis SnapshotChecker, we should
2164 // disable gc to be safe.
2165 const bool use_custom_gc_
;
2166 // Flag to indicate that the DB instance shutdown has been initiated. This
2167 // different from shutting_down_ atomic in that it is set at the beginning
2168 // of shutdown sequence, specifically in order to prevent any background
2169 // error recovery from going on in parallel. The latter, shutting_down_,
2170 // is set a little later during the shutdown after scheduling memtable
2172 std::atomic
<bool> shutdown_initiated_
;
2173 // Flag to indicate whether sst_file_manager object was allocated in
2174 // DB::Open() or passed to us
2177 // Clients must periodically call SetPreserveDeletesSequenceNumber()
2178 // to advance this seqnum. Default value is 0 which means ALL deletes are
2179 // preserved. Note that this has no effect if DBOptions.preserve_deletes
2181 std::atomic
<SequenceNumber
> preserve_deletes_seqnum_
;
2182 const bool preserve_deletes_
;
2184 // Flag to check whether Close() has been called on this DB
2187 ErrorHandler error_handler_
;
2189 // Conditional variable to coordinate installation of atomic flush results.
2190 // With atomic flush, each bg thread installs the result of flushing multiple
2191 // column families, and different threads can flush different column
2192 // families. It's difficult to rely on one thread to perform batch
2193 // installation for all threads. This is different from the non-atomic flush
2195 // atomic_flush_install_cv_ makes sure that threads install atomic flush
2196 // results sequentially. Flush results of memtables with lower IDs get
2197 // installed to MANIFEST first.
2198 InstrumentedCondVar atomic_flush_install_cv_
;
2200 bool wal_in_db_path_
;
2203 extern Options
SanitizeOptions(const std::string
& db
, const Options
& src
);
2205 extern DBOptions
SanitizeOptions(const std::string
& db
, const DBOptions
& src
);
2207 extern CompressionType
GetCompressionFlush(
2208 const ImmutableCFOptions
& ioptions
,
2209 const MutableCFOptions
& mutable_cf_options
);
2211 // Return the earliest log file to keep after the memtable flush is
2213 // `cfd_to_flush` is the column family whose memtable (specified in
2214 // `memtables_to_flush`) will be flushed and thus will not depend on any WAL
2216 // The function is only applicable to 2pc mode.
2217 extern uint64_t PrecomputeMinLogNumberToKeep2PC(
2218 VersionSet
* vset
, const ColumnFamilyData
& cfd_to_flush
,
2219 const autovector
<VersionEdit
*>& edit_list
,
2220 const autovector
<MemTable
*>& memtables_to_flush
,
2221 LogsWithPrepTracker
* prep_tracker
);
2223 // In non-2PC mode, WALs with log number < the returned number can be
2224 // deleted after the cfd_to_flush column family is flushed successfully.
2225 extern uint64_t PrecomputeMinLogNumberToKeepNon2PC(
2226 VersionSet
* vset
, const ColumnFamilyData
& cfd_to_flush
,
2227 const autovector
<VersionEdit
*>& edit_list
);
2229 // `cfd_to_flush` is the column family whose memtable will be flushed and thus
2230 // will not depend on any WAL file. nullptr means no memtable is being flushed.
2231 // The function is only applicable to 2pc mode.
2232 extern uint64_t FindMinPrepLogReferencedByMemTable(
2233 VersionSet
* vset
, const ColumnFamilyData
* cfd_to_flush
,
2234 const autovector
<MemTable
*>& memtables_to_flush
);
2236 // Fix user-supplied options to be reasonable
2237 template <class T
, class V
>
2238 static void ClipToRange(T
* ptr
, V minvalue
, V maxvalue
) {
2239 if (static_cast<V
>(*ptr
) > maxvalue
) *ptr
= maxvalue
;
2240 if (static_cast<V
>(*ptr
) < minvalue
) *ptr
= minvalue
;
2243 } // namespace ROCKSDB_NAMESPACE