rust_rocksdb/db.rs
1// Copyright 2020 Tyler Neely
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7// http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14//
15
16use std::cell::RefCell;
17use std::collections::{BTreeMap, HashMap};
18use std::ffi::{CStr, CString};
19use std::fmt;
20use std::fs;
21use std::iter;
22use std::path::Path;
23use std::path::PathBuf;
24use std::ptr;
25use std::slice;
26use std::str;
27use std::sync::Arc;
28use std::time::Duration;
29
30use crate::column_family::ColumnFamilyTtl;
31use crate::ffi_util::CSlice;
32use crate::{
33 ColumnFamily, ColumnFamilyDescriptor, CompactOptions, DBIteratorWithThreadMode,
34 DBPinnableBatch, DBPinnableSlice, DBRawIteratorWithThreadMode, DBWALIterator,
35 DEFAULT_COLUMN_FAMILY_NAME, Direction, Error, FlushOptions, IngestExternalFileOptions,
36 IteratorMode, Options, ReadOptions, SnapshotWithThreadMode, WaitForCompactOptions, WriteBatch,
37 WriteBatchWithIndex, WriteOptions,
38 column_family::{AsColumnFamilyRef, BoundColumnFamily, UnboundColumnFamily},
39 compaction::CompactionOptions,
40 db_options::{
41 FlushWalOptions, ImportColumnFamilyOptions, OptionsMustOutliveDB, SizeApproximationFlags,
42 SizeApproximationOptions,
43 },
44 event_listener::OwnedCompactionJobInfo,
45 ffi,
46 ffi_util::{
47 CStrLike, convert_rocksdb_error, from_cstr_and_free, from_cstr_without_free,
48 opt_bytes_to_ptr, raw_data, to_cpath,
49 },
50 metadata::{
51 ColumnFamilyMetaDataOptions, LevelMetaData, LiveFilesStorageInfo,
52 LiveFilesStorageInfoOptions, levels_from_cf_metadata_owned,
53 },
54 trace::{BlockCacheTraceOptions, BlockCacheTraceWriterOptions, Replayer, TraceOptions},
55 wal::{OwnedWalFile, WalFiles},
56};
57use rust_librocksdb_sys::{
58 rocksdb_livefile_destroy, rocksdb_livefile_t, rocksdb_livefiles_destroy, rocksdb_livefiles_t,
59};
60
61use libc::{self, c_char, c_int, c_uchar, c_void, size_t};
62use parking_lot::RwLock;
63
64// Default options are kept per-thread to avoid re-allocating on every call while
65// also preventing cross-thread sharing. Some RocksDB option wrappers hold
66// pointers into internal buffers and are not safe to share across threads.
67// Using thread_local allows cheap reuse in the common "default options" path
68// without synchronization overhead. Callers who need non-defaults must pass
69// explicit options.
70thread_local! { static DEFAULT_READ_OPTS: ReadOptions = ReadOptions::default(); }
71thread_local! { static DEFAULT_WRITE_OPTS: WriteOptions = WriteOptions::default(); }
72thread_local! { static DEFAULT_FLUSH_OPTS: FlushOptions = FlushOptions::default(); }
73// Thread-local ReadOptions for hot prefix probes; preconfigured for prefix scans.
74thread_local! { static PREFIX_READ_OPTS: RefCell<ReadOptions> = RefCell::new({ let mut o = ReadOptions::default(); o.set_prefix_same_as_start(true); o }); }
75
76/// Runs `f` with `ReadOptions` bounded to `prefix` and `prefix_same_as_start`
77/// enabled, reusing a thread-local instance when it is available.
78///
79/// The borrow is held across an FFI call that can synchronously re-enter Rust
80/// through a user-supplied comparator or merge operator. If that callback probes
81/// another prefix on the same thread, a plain `borrow_mut` would panic with
82/// `BorrowMutError` — and because the callback runs inside an `extern "C"` frame
83/// the panic aborts the process. Falling back to fresh options on contention
84/// costs an allocation in that rare re-entrant case and keeps the fast path
85/// allocation-free.
86fn with_prefix_read_opts<R>(prefix: &[u8], f: impl FnOnce(&ReadOptions) -> R) -> R {
87 PREFIX_READ_OPTS.with(|rc| {
88 if let Ok(mut opts) = rc.try_borrow_mut() {
89 opts.set_prefix_range_in_place(prefix);
90 f(&opts)
91 } else {
92 let mut opts = ReadOptions::default();
93 opts.set_prefix_same_as_start(true);
94 opts.set_prefix_range_in_place(prefix);
95 f(&opts)
96 }
97 })
98}
99
100/// A range of keys, `start_key` is included, but not `end_key`.
101///
102/// You should make sure `end_key` is not less than `start_key`.
103pub struct Range<'a> {
104 start_key: &'a [u8],
105 end_key: &'a [u8],
106}
107
108impl<'a> Range<'a> {
109 pub fn new(start_key: &'a [u8], end_key: &'a [u8]) -> Range<'a> {
110 Range { start_key, end_key }
111 }
112}
113
114/// Result of a [`get_into_buffer`](DBCommon::get_into_buffer) operation.
115///
116/// This enum represents the outcome of attempting to read a value directly
117/// into a caller-provided buffer, avoiding memory allocation. This is the most
118/// efficient way to read values when you have a pre-allocated buffer available.
119///
120/// # Performance
121///
122/// Using `get_into_buffer` with a reusable buffer can significantly reduce
123/// allocation overhead in hot paths compared to [`get`](DBCommon::get) or even
124/// [`get_pinned`](DBCommon::get_pinned):
125///
126/// - [`get`](DBCommon::get): Allocates a new `Vec<u8>` for each call
127/// - [`get_pinned`](DBCommon::get_pinned): Pins memory in RocksDB's block cache
128/// - `get_into_buffer`: Zero allocation when buffer is large enough
129///
130/// # Example
131///
132/// ```
133/// use rust_rocksdb::{DB, GetIntoBufferResult};
134///
135/// # let tempdir = tempfile::Builder::new().prefix("ex").tempdir().unwrap();
136/// let db = DB::open_default(tempdir.path()).unwrap();
137/// db.put(b"key", b"value").unwrap();
138///
139/// let mut buffer = [0u8; 1024];
140/// match db.get_into_buffer(b"key", &mut buffer).unwrap() {
141/// GetIntoBufferResult::Found(len) => {
142/// println!("Value: {:?}", &buffer[..len]);
143/// }
144/// GetIntoBufferResult::NotFound => {
145/// println!("Key not found");
146/// }
147/// GetIntoBufferResult::BufferTooSmall(needed) => {
148/// println!("Need a buffer of at least {} bytes", needed);
149/// }
150/// }
151/// ```
152#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
153pub enum GetIntoBufferResult {
154 /// The key was not found in the database.
155 NotFound,
156 /// The value was found and successfully copied into the buffer.
157 /// The `usize` contains the actual size of the value (number of bytes written).
158 Found(usize),
159 /// The value was found but the provided buffer was too small to hold it.
160 /// The `usize` contains the actual size of the value, allowing the caller
161 /// to allocate a larger buffer and retry.
162 ///
163 /// Note: When this variant is returned, no data is written to the buffer.
164 BufferTooSmall(usize),
165}
166
167impl GetIntoBufferResult {
168 /// Returns `true` if the key was found (regardless of buffer size).
169 #[inline]
170 pub fn is_found(&self) -> bool {
171 matches!(self, Self::Found(_) | Self::BufferTooSmall(_))
172 }
173
174 /// Returns `true` if the key was not found.
175 #[inline]
176 pub fn is_not_found(&self) -> bool {
177 matches!(self, Self::NotFound)
178 }
179
180 /// Returns the value size if the key was found, `None` otherwise.
181 #[inline]
182 pub fn value_size(&self) -> Option<usize> {
183 match self {
184 Self::Found(size) | Self::BufferTooSmall(size) => Some(*size),
185 Self::NotFound => None,
186 }
187 }
188}
189
190/// Read options tuned for prefix probes.
191fn prefix_probe_read_opts() -> ReadOptions {
192 let mut opts = ReadOptions::default();
193 opts.set_prefix_same_as_start(true);
194 opts
195}
196
197/// A reusable prefix probe that avoids per-call iterator creation/destruction.
198///
199/// Use this when performing many prefix existence checks in a tight loop.
200///
201/// A prober reads the database as of the sequence number that was current when
202/// it was created, and it pins the memtables and SST files that were current
203/// then. Both are what [`refresh`](PrefixProber::refresh) exists to move
204/// forward. A prober that is only used for a single burst of probes and then
205/// dropped needs neither.
206pub struct PrefixProber<'a, D: DBAccess> {
207 raw: DBRawIteratorWithThreadMode<'a, D>,
208}
209
210impl<D: DBAccess> PrefixProber<'_, D> {
211 /// Returns true if any key exists with the given prefix.
212 /// This performs a seek to the prefix and checks the current key.
213 ///
214 /// # Errors
215 ///
216 /// Returns the RocksDB error if the seek failed. A seek that hit an error
217 /// leaves the iterator invalid, so the key check below cannot observe one.
218 pub fn exists(&mut self, prefix: &[u8]) -> Result<bool, Error> {
219 self.raw.seek(prefix);
220 if let Some(key) = self.raw.key() {
221 return Ok(key.starts_with(prefix));
222 }
223 self.raw.status()?;
224 Ok(false)
225 }
226
227 /// Moves the probe to the latest committed state of the database.
228 ///
229 /// Call this before reusing a prober that has been sitting idle. Writes
230 /// that land after the prober was created or last refreshed are invisible
231 /// to it until then.
232 ///
233 /// This is cheap when RocksDB's superversion has not changed, because the
234 /// existing merge tree over the memtables and SST files is kept and only
235 /// the read sequence moves. A flush or a compaction bumps the superversion
236 /// and forces that tree to be rebuilt, which costs roughly what building a
237 /// new prober costs. Refreshing a prober that has not probed yet is a
238 /// pessimisation, because it builds the tree that the first
239 /// [`exists`](PrefixProber::exists) call would otherwise build lazily.
240 ///
241 /// Refreshing also releases the memtables and SST files the prober was
242 /// pinning, which is what lets a flush or a compaction reclaim them. A
243 /// cached prober that is never refreshed holds them for as long as it
244 /// lives, so pool them on a timer rather than on request arrival.
245 ///
246 /// # Errors
247 ///
248 /// Returns the RocksDB error if the refresh failed.
249 pub fn refresh(&mut self) -> Result<(), Error> {
250 self.raw.refresh()
251 }
252}
253
254/// A [`PrefixProber`] that keeps the database open instead of borrowing it.
255///
256/// Use this to cache a prober beyond the scope that built it, for example one
257/// per worker thread. [`PrefixProber`] borrows the database, so it cannot be
258/// held in a thread local or in shared state.
259///
260/// Everything [`PrefixProber`] documents about staleness and pinning applies
261/// here, and matters more, because the point of an owned prober is to outlive
262/// the request that created it. Refresh or drop cached probers on a timer. An
263/// idle one goes on pinning the memtables and SST files it was built over, and
264/// nothing will reclaim them.
265///
266/// `D` is `'static` because the wrapper owns the database rather than borrowing
267/// it. Every `DBWithThreadMode` satisfies that.
268pub struct OwnedPrefixProber<D: DBAccess + 'static> {
269 // Field order is load bearing. Fields drop in declaration order, so the
270 // iterator is destroyed while `_db` still holds the database open.
271 // Reordering these two is a use after free.
272 prober: PrefixProber<'static, D>,
273 _db: Arc<D>,
274}
275
276// No `Deref`/`DerefMut` to `PrefixProber`. `DerefMut` would let two owned
277// probers swap their inner probers, leaving each holding an iterator into the
278// other's database, and dropping one could then free a database the other still
279// points at.
280impl<D: DBAccess + 'static> OwnedPrefixProber<D> {
281 /// Creates an owned prober over the default column family, using read
282 /// options tuned for prefix probes.
283 pub fn new(db: Arc<D>) -> Self {
284 Self::with_opts(db, prefix_probe_read_opts())
285 }
286
287 /// Creates an owned prober over the default column family with the given
288 /// read options.
289 ///
290 /// The prober owns `readopts` so that any buffers it points at, such as
291 /// iterate bounds, stay alive for as long as the iterator.
292 pub fn with_opts(db: Arc<D>, readopts: ReadOptions) -> Self {
293 // A `'static` iterator stores no dangling reference: the database is
294 // recorded on `DBRawIteratorWithThreadMode` as a `PhantomData` lifetime
295 // and nothing reads through it. `_db` is what actually keeps the
296 // database alive, and the field order on the struct is what guarantees
297 // the iterator is destroyed first.
298 let raw = DBRawIteratorWithThreadMode::new(&*db, readopts);
299 Self {
300 prober: PrefixProber { raw },
301 _db: db,
302 }
303 }
304
305 /// Creates an owned prober over one column family, using read options tuned
306 /// for prefix probes.
307 pub fn new_cf(db: Arc<D>, cf_handle: &impl AsColumnFamilyRef) -> Self {
308 Self::cf_with_opts(db, cf_handle, prefix_probe_read_opts())
309 }
310
311 /// Creates an owned prober over one column family with the given read
312 /// options.
313 ///
314 /// `cf_handle` is only read while the iterator is being created. RocksDB
315 /// takes its own reference to the column family, so the handle itself does
316 /// not have to outlive the prober. Dropping the column family while a
317 /// prober over it is alive is still not supported: the prober keeps reading
318 /// the state it was built over, which is no longer meaningful.
319 pub fn cf_with_opts(
320 db: Arc<D>,
321 cf_handle: &impl AsColumnFamilyRef,
322 readopts: ReadOptions,
323 ) -> Self {
324 // See the safety note in `with_opts`.
325 let raw = DBRawIteratorWithThreadMode::new_cf_detached(&*db, cf_handle.inner(), readopts);
326 Self {
327 prober: PrefixProber { raw },
328 _db: db,
329 }
330 }
331
332 /// Returns true if any key exists with the given prefix.
333 ///
334 /// See [`PrefixProber::exists`].
335 ///
336 /// # Errors
337 ///
338 /// Returns the RocksDB error if the seek failed.
339 pub fn exists(&mut self, prefix: &[u8]) -> Result<bool, Error> {
340 self.prober.exists(prefix)
341 }
342
343 /// Moves the probe to the latest committed state of the database.
344 ///
345 /// See [`PrefixProber::refresh`].
346 ///
347 /// # Errors
348 ///
349 /// Returns the RocksDB error if the refresh failed.
350 pub fn refresh(&mut self) -> Result<(), Error> {
351 self.prober.refresh()
352 }
353}
354
355/// Marker trait to specify single or multi threaded column family alternations for
356/// [`DBWithThreadMode<T>`]
357///
358/// This arrangement makes differences in self mutability and return type in
359/// some of `DBWithThreadMode` methods.
360///
361/// While being a marker trait to be generic over `DBWithThreadMode`, this trait
362/// also has a minimum set of not-encapsulated internal methods between
363/// [`SingleThreaded`] and [`MultiThreaded`]. These methods aren't expected to be
364/// called and defined externally.
365pub trait ThreadMode {
366 /// Internal implementation for storing column family handles
367 fn new_cf_map_internal(
368 cf_map: BTreeMap<String, *mut ffi::rocksdb_column_family_handle_t>,
369 ) -> Self;
370 /// Internal implementation for dropping column family handles
371 fn drop_all_cfs_internal(&mut self);
372}
373
374/// Actual marker type for the marker trait `ThreadMode`, which holds
375/// a collection of column families without synchronization primitive, providing
376/// no overhead for the single-threaded column family alternations. The other
377/// mode is [`MultiThreaded`].
378///
379/// See [`DB`] for more details, including performance implications for each mode
380pub struct SingleThreaded {
381 pub(crate) cfs: HashMap<String, ColumnFamily>,
382}
383
384/// Actual marker type for the marker trait `ThreadMode`, which holds
385/// a collection of column families wrapped in a RwLock to be mutated
386/// concurrently. The other mode is [`SingleThreaded`].
387///
388/// See [`DB`] for more details, including performance implications for each mode
389pub struct MultiThreaded {
390 pub(crate) cfs: RwLock<HashMap<String, Arc<UnboundColumnFamily>>>,
391}
392
393impl ThreadMode for SingleThreaded {
394 fn new_cf_map_internal(
395 cfs: BTreeMap<String, *mut ffi::rocksdb_column_family_handle_t>,
396 ) -> Self {
397 Self {
398 cfs: cfs
399 .into_iter()
400 .map(|(n, c)| (n, ColumnFamily { inner: c }))
401 .collect(),
402 }
403 }
404
405 fn drop_all_cfs_internal(&mut self) {
406 // Cause all ColumnFamily objects to be Drop::drop()-ed.
407 self.cfs.clear();
408 }
409}
410
411impl ThreadMode for MultiThreaded {
412 fn new_cf_map_internal(
413 cfs: BTreeMap<String, *mut ffi::rocksdb_column_family_handle_t>,
414 ) -> Self {
415 Self {
416 cfs: RwLock::new(
417 cfs.into_iter()
418 .map(|(n, c)| (n, Arc::new(UnboundColumnFamily { inner: c })))
419 .collect(),
420 ),
421 }
422 }
423
424 fn drop_all_cfs_internal(&mut self) {
425 // Cause all UnboundColumnFamily objects to be Drop::drop()-ed.
426 self.cfs.write().clear();
427 }
428}
429
430/// Get underlying `rocksdb_t`.
431pub trait DBInner {
432 fn inner(&self) -> *mut ffi::rocksdb_t;
433}
434
435/// A helper type to implement some common methods for [`DBWithThreadMode`]
436/// and [`OptimisticTransactionDB`].
437///
438/// [`OptimisticTransactionDB`]: crate::OptimisticTransactionDB
439///
440/// When using [`SingleThreaded`] mode, `create_cf` requires `&mut self`,
441/// preventing multiple immutable references from calling it concurrently:
442///
443/// ```compile_fail,E0596
444/// use rust_rocksdb::{DBWithThreadMode, Options, SingleThreaded};
445///
446/// let db = DBWithThreadMode::<SingleThreaded>::open_default("/path/to/dummy").unwrap();
447/// let db_ref1 = &db;
448/// let db_ref2 = &db;
449/// let opts = Options::default();
450/// db_ref1.create_cf("cf1", &opts).unwrap();
451/// db_ref2.create_cf("cf2", &opts).unwrap();
452/// ```
453///
454/// [`SingleThreaded`]: crate::SingleThreaded
455pub struct DBCommon<T: ThreadMode, D: DBInner> {
456 pub(crate) inner: D,
457 cfs: T, // Column families are held differently depending on thread mode
458 path: PathBuf,
459 _outlive: Vec<OptionsMustOutliveDB>,
460 /// The TTL this DB was opened with, if it was opened with one.
461 ///
462 /// Two things need it. `create_cf_with_ttl` reaches a C function that casts the
463 /// handle to `DBWithTTL` without checking, so calling it on any other kind of DB
464 /// is undefined behaviour, and nothing in the C API can be asked after the fact.
465 /// It is also the TTL that [`ColumnFamilyTtl::SameAsDb`] refers to.
466 opened_with_ttl: Option<Duration>,
467}
468
469/// Minimal set of DB-related methods, intended to be generic over
470/// `DBWithThreadMode<T>`. Mainly used internally
471pub trait DBAccess {
472 unsafe fn create_snapshot(&self) -> *const ffi::rocksdb_snapshot_t;
473
474 unsafe fn release_snapshot(&self, snapshot: *const ffi::rocksdb_snapshot_t);
475
476 unsafe fn create_iterator(&self, readopts: &ReadOptions) -> *mut ffi::rocksdb_iterator_t;
477
478 unsafe fn create_iterator_cf(
479 &self,
480 cf_handle: *mut ffi::rocksdb_column_family_handle_t,
481 readopts: &ReadOptions,
482 ) -> *mut ffi::rocksdb_iterator_t;
483
484 fn get_opt<K: AsRef<[u8]>>(
485 &self,
486 key: K,
487 readopts: &ReadOptions,
488 ) -> Result<Option<Vec<u8>>, Error>;
489
490 fn get_cf_opt<K: AsRef<[u8]>>(
491 &self,
492 cf: &impl AsColumnFamilyRef,
493 key: K,
494 readopts: &ReadOptions,
495 ) -> Result<Option<Vec<u8>>, Error>;
496
497 fn get_pinned_opt<K: AsRef<[u8]>>(
498 &'_ self,
499 key: K,
500 readopts: &ReadOptions,
501 ) -> Result<Option<DBPinnableSlice<'_>>, Error>;
502
503 fn get_pinned_cf_opt<K: AsRef<[u8]>>(
504 &'_ self,
505 cf: &impl AsColumnFamilyRef,
506 key: K,
507 readopts: &ReadOptions,
508 ) -> Result<Option<DBPinnableSlice<'_>>, Error>;
509
510 fn multi_get_opt<K, I>(
511 &self,
512 keys: I,
513 readopts: &ReadOptions,
514 ) -> Vec<Result<Option<Vec<u8>>, Error>>
515 where
516 K: AsRef<[u8]>,
517 I: IntoIterator<Item = K>;
518
519 fn multi_get_cf_opt<'b, K, I, W>(
520 &self,
521 keys_cf: I,
522 readopts: &ReadOptions,
523 ) -> Vec<Result<Option<Vec<u8>>, Error>>
524 where
525 K: AsRef<[u8]>,
526 I: IntoIterator<Item = (&'b W, K)>,
527 W: AsColumnFamilyRef + 'b;
528}
529
530impl<T: ThreadMode, D: DBInner> DBAccess for DBCommon<T, D> {
531 unsafe fn create_snapshot(&self) -> *const ffi::rocksdb_snapshot_t {
532 unsafe { ffi::rocksdb_create_snapshot(self.inner.inner()) }
533 }
534
535 unsafe fn release_snapshot(&self, snapshot: *const ffi::rocksdb_snapshot_t) {
536 unsafe {
537 ffi::rocksdb_release_snapshot(self.inner.inner(), snapshot);
538 }
539 }
540
541 unsafe fn create_iterator(&self, readopts: &ReadOptions) -> *mut ffi::rocksdb_iterator_t {
542 unsafe { ffi::rocksdb_create_iterator(self.inner.inner(), readopts.inner) }
543 }
544
545 unsafe fn create_iterator_cf(
546 &self,
547 cf_handle: *mut ffi::rocksdb_column_family_handle_t,
548 readopts: &ReadOptions,
549 ) -> *mut ffi::rocksdb_iterator_t {
550 unsafe { ffi::rocksdb_create_iterator_cf(self.inner.inner(), readopts.inner, cf_handle) }
551 }
552
553 fn get_opt<K: AsRef<[u8]>>(
554 &self,
555 key: K,
556 readopts: &ReadOptions,
557 ) -> Result<Option<Vec<u8>>, Error> {
558 self.get_opt(key, readopts)
559 }
560
561 fn get_cf_opt<K: AsRef<[u8]>>(
562 &self,
563 cf: &impl AsColumnFamilyRef,
564 key: K,
565 readopts: &ReadOptions,
566 ) -> Result<Option<Vec<u8>>, Error> {
567 self.get_cf_opt(cf, key, readopts)
568 }
569
570 fn get_pinned_opt<K: AsRef<[u8]>>(
571 &'_ self,
572 key: K,
573 readopts: &ReadOptions,
574 ) -> Result<Option<DBPinnableSlice<'_>>, Error> {
575 self.get_pinned_opt(key, readopts)
576 }
577
578 fn get_pinned_cf_opt<K: AsRef<[u8]>>(
579 &'_ self,
580 cf: &impl AsColumnFamilyRef,
581 key: K,
582 readopts: &ReadOptions,
583 ) -> Result<Option<DBPinnableSlice<'_>>, Error> {
584 self.get_pinned_cf_opt(cf, key, readopts)
585 }
586
587 fn multi_get_opt<K, Iter>(
588 &self,
589 keys: Iter,
590 readopts: &ReadOptions,
591 ) -> Vec<Result<Option<Vec<u8>>, Error>>
592 where
593 K: AsRef<[u8]>,
594 Iter: IntoIterator<Item = K>,
595 {
596 self.multi_get_opt(keys, readopts)
597 }
598
599 fn multi_get_cf_opt<'b, K, Iter, W>(
600 &self,
601 keys_cf: Iter,
602 readopts: &ReadOptions,
603 ) -> Vec<Result<Option<Vec<u8>>, Error>>
604 where
605 K: AsRef<[u8]>,
606 Iter: IntoIterator<Item = (&'b W, K)>,
607 W: AsColumnFamilyRef + 'b,
608 {
609 self.multi_get_cf_opt(keys_cf, readopts)
610 }
611}
612
613pub struct DBWithThreadModeInner {
614 inner: *mut ffi::rocksdb_t,
615}
616
617struct OwnedColumnFamilyHandle {
618 inner: *mut ffi::rocksdb_column_family_handle_t,
619}
620
621struct PinnedMultiGetOutput {
622 values: Vec<*mut ffi::rocksdb_pinnableslice_t>,
623 errors: Vec<*mut c_char>,
624}
625
626/// The result of one `rocksdb_create_iterators` call: the iterator handles
627/// plus the single `ReadOptions` they were all created from, which each
628/// iterator must keep alive.
629struct CreatedIterators {
630 readopts: Arc<ReadOptions>,
631 handles: Vec<*mut ffi::rocksdb_iterator_t>,
632}
633
634impl OwnedColumnFamilyHandle {
635 fn default_for(db: *mut ffi::rocksdb_t) -> Self {
636 Self {
637 inner: unsafe { ffi::rocksdb_get_default_column_family_handle(db) },
638 }
639 }
640}
641
642impl Drop for OwnedColumnFamilyHandle {
643 fn drop(&mut self) {
644 unsafe {
645 ffi::rocksdb_column_family_handle_destroy(self.inner);
646 }
647 }
648}
649
650impl DBInner for DBWithThreadModeInner {
651 #[inline]
652 fn inner(&self) -> *mut ffi::rocksdb_t {
653 self.inner
654 }
655}
656
657impl Drop for DBWithThreadModeInner {
658 fn drop(&mut self) {
659 unsafe {
660 ffi::rocksdb_close(self.inner);
661 }
662 }
663}
664
665/// A type alias to RocksDB database.
666///
667/// See crate level documentation for a simple usage example.
668/// See [`DBCommon`] for full list of methods.
669pub type DBWithThreadMode<T> = DBCommon<T, DBWithThreadModeInner>;
670
671/// A type alias to DB instance type with the single-threaded column family
672/// creations/deletions
673///
674/// # Compatibility and multi-threaded mode
675///
676/// Previously, [`DB`] was defined as a direct `struct`. Now, it's type-aliased for
677/// compatibility. Use `DBCommon<MultiThreaded>` for multi-threaded
678/// column family alternations.
679///
680/// # Limited performance implication for single-threaded mode
681///
682/// Even with [`SingleThreaded`], almost all of RocksDB operations is
683/// multi-threaded unless the underlying RocksDB instance is
684/// specifically configured otherwise. `SingleThreaded` only forces
685/// serialization of column family alternations by requiring `&mut self` of DB
686/// instance due to its wrapper implementation details.
687///
688/// # Multi-threaded mode
689///
690/// [`MultiThreaded`] can be appropriate for the situation of multi-threaded
691/// workload including multi-threaded column family alternations, costing the
692/// RwLock overhead inside `DB`.
693#[cfg(not(feature = "multi-threaded-cf"))]
694pub type DB = DBWithThreadMode<SingleThreaded>;
695
696#[cfg(feature = "multi-threaded-cf")]
697pub type DB = DBWithThreadMode<MultiThreaded>;
698
699// Safety note: auto-implementing Send on most db-related types is prevented by the inner FFI
700// pointer. In most cases, however, this pointer is Send-safe because it is never aliased and
701// rocksdb internally does not rely on thread-local information for its user-exposed types.
702unsafe impl<T: ThreadMode + Send, I: DBInner> Send for DBCommon<T, I> {}
703
704// Sync is similarly safe for many types because they do not expose interior mutability, and their
705// use within the rocksdb library is generally behind a const reference
706unsafe impl<T: ThreadMode, I: DBInner> Sync for DBCommon<T, I> {}
707
708// Specifies whether open DB for read only.
709enum AccessType<'a> {
710 ReadWrite,
711 ReadOnly { error_if_log_file_exist: bool },
712 Secondary { secondary_path: &'a Path },
713 WithTTL { ttl: Duration },
714 TrimHistory { trim_ts: &'a [u8] },
715}
716
717/// Methods of `DBWithThreadMode`.
718impl<T: ThreadMode> DBWithThreadMode<T> {
719 /// Opens a database with default options.
720 pub fn open_default<P: AsRef<Path>>(path: P) -> Result<Self, Error> {
721 let mut opts = Options::default();
722 opts.create_if_missing(true);
723 Self::open(&opts, path)
724 }
725
726 /// Opens the database with the specified options.
727 pub fn open<P: AsRef<Path>>(opts: &Options, path: P) -> Result<Self, Error> {
728 Self::open_cf(opts, path, None::<&str>)
729 }
730
731 /// Opens the database for read only with the specified options.
732 pub fn open_for_read_only<P: AsRef<Path>>(
733 opts: &Options,
734 path: P,
735 error_if_log_file_exist: bool,
736 ) -> Result<Self, Error> {
737 Self::open_cf_for_read_only(opts, path, None::<&str>, error_if_log_file_exist)
738 }
739
740 /// Opens the database as a secondary.
741 pub fn open_as_secondary<P: AsRef<Path>>(
742 opts: &Options,
743 primary_path: P,
744 secondary_path: P,
745 ) -> Result<Self, Error> {
746 Self::open_cf_as_secondary(opts, primary_path, secondary_path, None::<&str>)
747 }
748
749 /// Opens the database with a Time to Live compaction filter.
750 ///
751 /// This applies the given `ttl` to all column families created without an explicit TTL.
752 /// See [`DB::open_cf_descriptors_with_ttl`] for more control over individual column family TTLs.
753 ///
754 /// RocksDB stores the TTL as a 32-bit second count, so a `ttl` longer than
755 /// `i32::MAX` seconds (about 68 years) is clamped to that maximum rather
756 /// than wrapping.
757 pub fn open_with_ttl<P: AsRef<Path>>(
758 opts: &Options,
759 path: P,
760 ttl: Duration,
761 ) -> Result<Self, Error> {
762 Self::open_cf_descriptors_with_ttl(opts, path, std::iter::empty(), ttl)
763 }
764
765 /// Opens the database with a Time to Live compaction filter and column family names.
766 ///
767 /// Column families opened using this function will be created with default `Options`.
768 pub fn open_cf_with_ttl<P, I, N>(
769 opts: &Options,
770 path: P,
771 cfs: I,
772 ttl: Duration,
773 ) -> Result<Self, Error>
774 where
775 P: AsRef<Path>,
776 I: IntoIterator<Item = N>,
777 N: AsRef<str>,
778 {
779 let cfs = cfs
780 .into_iter()
781 .map(|name| ColumnFamilyDescriptor::new(name.as_ref(), Options::default()));
782
783 Self::open_cf_descriptors_with_ttl(opts, path, cfs, ttl)
784 }
785
786 /// Opens a database with the given database with a Time to Live compaction filter and
787 /// column family descriptors.
788 ///
789 /// Applies the provided `ttl` as the default TTL for all column families.
790 /// Column families will inherit this TTL by default, unless their descriptor explicitly
791 /// sets a different TTL using [`ColumnFamilyTtl::Duration`] or opts out using [`ColumnFamilyTtl::Disabled`].
792 ///
793 /// *NOTE*: The `default` column family is opened with `Options::default()` unless
794 /// explicitly configured within the `cfs` iterator.
795 /// To customize the `default` column family's options, include a `ColumnFamilyDescriptor`
796 /// with the name "default" in the `cfs` iterator.
797 ///
798 /// If you want to open `default` cf with different options, set them explicitly in `cfs`.
799 pub fn open_cf_descriptors_with_ttl<P, I>(
800 opts: &Options,
801 path: P,
802 cfs: I,
803 ttl: Duration,
804 ) -> Result<Self, Error>
805 where
806 P: AsRef<Path>,
807 I: IntoIterator<Item = ColumnFamilyDescriptor>,
808 {
809 Self::open_cf_descriptors_internal(opts, path, cfs, &AccessType::WithTTL { ttl })
810 }
811
812 /// Opens the database and drops everything written after `trim_ts`.
813 ///
814 /// Column families opened using this function will be created with default
815 /// `Options`. See [`open_cf_descriptors_and_trim_history`][Self::open_cf_descriptors_and_trim_history].
816 ///
817 /// # Errors
818 ///
819 /// See [`open_cf_descriptors_and_trim_history`][Self::open_cf_descriptors_and_trim_history].
820 pub fn open_cf_and_trim_history<P, I, N>(
821 opts: &Options,
822 path: P,
823 cfs: I,
824 trim_ts: &[u8],
825 ) -> Result<Self, Error>
826 where
827 P: AsRef<Path>,
828 I: IntoIterator<Item = N>,
829 N: AsRef<str>,
830 {
831 let cfs = cfs
832 .into_iter()
833 .map(|name| ColumnFamilyDescriptor::new(name.as_ref(), Options::default()));
834
835 Self::open_cf_descriptors_and_trim_history(opts, path, cfs, trim_ts)
836 }
837
838 /// Opens the database and drops everything written after `trim_ts`, given column
839 /// family descriptors.
840 ///
841 /// This is for recovering a column family that uses user-defined timestamps, where
842 /// writes past a known good point have to be undone before anything reads them.
843 /// Entries with a timestamp greater than `trim_ts` are gone once this returns.
844 /// `trim_ts` is compared with the column family's comparator, so it has to be
845 /// encoded the way that comparator expects. Column families without user-defined
846 /// timestamps are left alone.
847 ///
848 /// The trim is permanent, so open a copy first if the discarded writes still
849 /// matter.
850 ///
851 /// RocksDB marks the underlying API experimental and subject to change.
852 ///
853 /// *NOTE*: The `default` column family is opened with `Options::default()` unless
854 /// explicitly configured within the `cfs` iterator. A column family that uses
855 /// user-defined timestamps carries its comparator in its own options, so it has to
856 /// be named here with those options even when it is the default one. Opening it
857 /// with `Options::default()` instead fails with a comparator mismatch.
858 ///
859 /// # Errors
860 ///
861 /// Returns the RocksDB error if the database cannot be opened, which includes
862 /// leaving out a column family that exists on disk. Also errors if a column family
863 /// name contains an interior NUL byte, or if the directory cannot be created.
864 pub fn open_cf_descriptors_and_trim_history<P, I>(
865 opts: &Options,
866 path: P,
867 cfs: I,
868 trim_ts: &[u8],
869 ) -> Result<Self, Error>
870 where
871 P: AsRef<Path>,
872 I: IntoIterator<Item = ColumnFamilyDescriptor>,
873 {
874 // The C function only takes the column family form, so make sure the open goes
875 // down that path even when the caller named no families.
876 let mut cfs: Vec<ColumnFamilyDescriptor> = cfs.into_iter().collect();
877 if cfs.is_empty() {
878 cfs.push(ColumnFamilyDescriptor::new(
879 DEFAULT_COLUMN_FAMILY_NAME,
880 Options::default(),
881 ));
882 }
883
884 Self::open_cf_descriptors_internal(opts, path, cfs, &AccessType::TrimHistory { trim_ts })
885 }
886
887 /// Opens a database with the given database options and column family names.
888 ///
889 /// Column families opened using this function will be created with default `Options`.
890 pub fn open_cf<P, I, N>(opts: &Options, path: P, cfs: I) -> Result<Self, Error>
891 where
892 P: AsRef<Path>,
893 I: IntoIterator<Item = N>,
894 N: AsRef<str>,
895 {
896 let cfs = cfs
897 .into_iter()
898 .map(|name| ColumnFamilyDescriptor::new(name.as_ref(), Options::default()));
899
900 Self::open_cf_descriptors_internal(opts, path, cfs, &AccessType::ReadWrite)
901 }
902
903 /// Opens a database with the given database options and column family names.
904 ///
905 /// Column families opened using given `Options`.
906 pub fn open_cf_with_opts<P, I, N>(opts: &Options, path: P, cfs: I) -> Result<Self, Error>
907 where
908 P: AsRef<Path>,
909 I: IntoIterator<Item = (N, Options)>,
910 N: AsRef<str>,
911 {
912 let cfs = cfs
913 .into_iter()
914 .map(|(name, opts)| ColumnFamilyDescriptor::new(name.as_ref(), opts));
915
916 Self::open_cf_descriptors(opts, path, cfs)
917 }
918
919 /// Opens a database for read only with the given database options and column family names.
920 /// *NOTE*: `default` column family is opened with `Options::default()`.
921 /// If you want to open `default` cf with different options, set them explicitly in `cfs`.
922 pub fn open_cf_for_read_only<P, I, N>(
923 opts: &Options,
924 path: P,
925 cfs: I,
926 error_if_log_file_exist: bool,
927 ) -> Result<Self, Error>
928 where
929 P: AsRef<Path>,
930 I: IntoIterator<Item = N>,
931 N: AsRef<str>,
932 {
933 let cfs = cfs
934 .into_iter()
935 .map(|name| ColumnFamilyDescriptor::new(name.as_ref(), Options::default()));
936
937 Self::open_cf_descriptors_internal(
938 opts,
939 path,
940 cfs,
941 &AccessType::ReadOnly {
942 error_if_log_file_exist,
943 },
944 )
945 }
946
947 /// Opens a database for read only with the given database options and column family names.
948 /// *NOTE*: `default` column family is opened with `Options::default()`.
949 /// If you want to open `default` cf with different options, set them explicitly in `cfs`.
950 pub fn open_cf_with_opts_for_read_only<P, I, N>(
951 db_opts: &Options,
952 path: P,
953 cfs: I,
954 error_if_log_file_exist: bool,
955 ) -> Result<Self, Error>
956 where
957 P: AsRef<Path>,
958 I: IntoIterator<Item = (N, Options)>,
959 N: AsRef<str>,
960 {
961 let cfs = cfs
962 .into_iter()
963 .map(|(name, cf_opts)| ColumnFamilyDescriptor::new(name.as_ref(), cf_opts));
964
965 Self::open_cf_descriptors_internal(
966 db_opts,
967 path,
968 cfs,
969 &AccessType::ReadOnly {
970 error_if_log_file_exist,
971 },
972 )
973 }
974
975 /// Opens a database for ready only with the given database options and
976 /// column family descriptors.
977 /// *NOTE*: `default` column family is opened with `Options::default()`.
978 /// If you want to open `default` cf with different options, set them explicitly in `cfs`.
979 pub fn open_cf_descriptors_read_only<P, I>(
980 opts: &Options,
981 path: P,
982 cfs: I,
983 error_if_log_file_exist: bool,
984 ) -> Result<Self, Error>
985 where
986 P: AsRef<Path>,
987 I: IntoIterator<Item = ColumnFamilyDescriptor>,
988 {
989 Self::open_cf_descriptors_internal(
990 opts,
991 path,
992 cfs,
993 &AccessType::ReadOnly {
994 error_if_log_file_exist,
995 },
996 )
997 }
998
999 /// Opens the database as a secondary with the given database options and column family names.
1000 /// *NOTE*: `default` column family is opened with `Options::default()`.
1001 /// If you want to open `default` cf with different options, set them explicitly in `cfs`.
1002 pub fn open_cf_as_secondary<P, I, N>(
1003 opts: &Options,
1004 primary_path: P,
1005 secondary_path: P,
1006 cfs: I,
1007 ) -> Result<Self, Error>
1008 where
1009 P: AsRef<Path>,
1010 I: IntoIterator<Item = N>,
1011 N: AsRef<str>,
1012 {
1013 let cfs = cfs
1014 .into_iter()
1015 .map(|name| ColumnFamilyDescriptor::new(name.as_ref(), Options::default()));
1016
1017 Self::open_cf_descriptors_internal(
1018 opts,
1019 primary_path,
1020 cfs,
1021 &AccessType::Secondary {
1022 secondary_path: secondary_path.as_ref(),
1023 },
1024 )
1025 }
1026
1027 /// Opens the database as a secondary with the given database options and
1028 /// column family descriptors.
1029 /// *NOTE*: `default` column family is opened with `Options::default()`.
1030 /// If you want to open `default` cf with different options, set them explicitly in `cfs`.
1031 pub fn open_cf_descriptors_as_secondary<P, I>(
1032 opts: &Options,
1033 path: P,
1034 secondary_path: P,
1035 cfs: I,
1036 ) -> Result<Self, Error>
1037 where
1038 P: AsRef<Path>,
1039 I: IntoIterator<Item = ColumnFamilyDescriptor>,
1040 {
1041 Self::open_cf_descriptors_internal(
1042 opts,
1043 path,
1044 cfs,
1045 &AccessType::Secondary {
1046 secondary_path: secondary_path.as_ref(),
1047 },
1048 )
1049 }
1050
1051 /// Opens a database with the given database options and column family descriptors.
1052 /// *NOTE*: `default` column family is opened with `Options::default()`.
1053 /// If you want to open `default` cf with different options, set them explicitly in `cfs`.
1054 pub fn open_cf_descriptors<P, I>(opts: &Options, path: P, cfs: I) -> Result<Self, Error>
1055 where
1056 P: AsRef<Path>,
1057 I: IntoIterator<Item = ColumnFamilyDescriptor>,
1058 {
1059 Self::open_cf_descriptors_internal(opts, path, cfs, &AccessType::ReadWrite)
1060 }
1061
1062 /// Internal implementation for opening RocksDB.
1063 fn open_cf_descriptors_internal<P, I>(
1064 opts: &Options,
1065 path: P,
1066 cfs: I,
1067 access_type: &AccessType,
1068 ) -> Result<Self, Error>
1069 where
1070 P: AsRef<Path>,
1071 I: IntoIterator<Item = ColumnFamilyDescriptor>,
1072 {
1073 let cfs: Vec<_> = cfs.into_iter().collect();
1074 let outlive = iter::once(opts.outlive.clone())
1075 .chain(cfs.iter().map(|cf| cf.options.outlive.clone()))
1076 .collect();
1077
1078 let cpath = to_cpath(&path)?;
1079
1080 if let Err(e) = fs::create_dir_all(&path) {
1081 return Err(Error::new(format!(
1082 "Failed to create RocksDB directory: `{e:?}`."
1083 )));
1084 }
1085
1086 let db: *mut ffi::rocksdb_t;
1087 let mut cf_map = BTreeMap::new();
1088
1089 if cfs.is_empty() {
1090 db = Self::open_raw(opts, &cpath, access_type)?;
1091 } else {
1092 let mut cfs_v = cfs;
1093 // Always open the default column family.
1094 if !cfs_v.iter().any(|cf| cf.name == DEFAULT_COLUMN_FAMILY_NAME) {
1095 cfs_v.push(ColumnFamilyDescriptor {
1096 name: String::from(DEFAULT_COLUMN_FAMILY_NAME),
1097 options: Options::default(),
1098 ttl: ColumnFamilyTtl::SameAsDb,
1099 });
1100 }
1101 // We need to store our CStrings in an intermediate vector
1102 // so that their pointers remain valid.
1103 let c_cfs: Vec<CString> = cfs_v
1104 .iter()
1105 .map(|cf| CString::new(cf.name.as_bytes()).unwrap())
1106 .collect();
1107
1108 let cfnames: Vec<_> = c_cfs.iter().map(|cf| cf.as_ptr()).collect();
1109
1110 // These handles will be populated by DB.
1111 let mut cfhandles: Vec<_> = cfs_v.iter().map(|_| ptr::null_mut()).collect();
1112
1113 let cfopts: Vec<_> = cfs_v
1114 .iter()
1115 .map(|cf| cf.options.inner.cast_const())
1116 .collect();
1117
1118 db = Self::open_cf_raw(
1119 opts,
1120 &cpath,
1121 &cfs_v,
1122 &cfnames,
1123 &cfopts,
1124 &mut cfhandles,
1125 access_type,
1126 )?;
1127 for handle in &cfhandles {
1128 if handle.is_null() {
1129 return Err(Error::new(
1130 "Received null column family handle from DB.".to_owned(),
1131 ));
1132 }
1133 }
1134
1135 for (cf_desc, inner) in cfs_v.iter().zip(cfhandles) {
1136 cf_map.insert(cf_desc.name.clone(), inner);
1137 }
1138 }
1139
1140 if db.is_null() {
1141 return Err(Error::new("Could not initialize database.".to_owned()));
1142 }
1143
1144 Ok(Self {
1145 inner: DBWithThreadModeInner { inner: db },
1146 path: path.as_ref().to_path_buf(),
1147 cfs: T::new_cf_map_internal(cf_map),
1148 _outlive: outlive,
1149 opened_with_ttl: match access_type {
1150 AccessType::WithTTL { ttl } => Some(*ttl),
1151 _ => None,
1152 },
1153 })
1154 }
1155
1156 fn open_raw(
1157 opts: &Options,
1158 cpath: &CString,
1159 access_type: &AccessType,
1160 ) -> Result<*mut ffi::rocksdb_t, Error> {
1161 let db = unsafe {
1162 match *access_type {
1163 AccessType::ReadOnly {
1164 error_if_log_file_exist,
1165 } => ffi_try!(ffi::rocksdb_open_for_read_only(
1166 opts.inner,
1167 cpath.as_ptr(),
1168 c_uchar::from(error_if_log_file_exist),
1169 )),
1170 AccessType::ReadWrite => {
1171 ffi_try!(ffi::rocksdb_open(opts.inner, cpath.as_ptr()))
1172 }
1173 AccessType::TrimHistory { .. } => {
1174 // open_cf_descriptors_and_trim_history always names at least the
1175 // default column family, so this path is not reached.
1176 return Err(Error::new(
1177 "Trimming history requires opening with column families".to_owned(),
1178 ));
1179 }
1180 AccessType::Secondary { secondary_path } => {
1181 ffi_try!(ffi::rocksdb_open_as_secondary(
1182 opts.inner,
1183 cpath.as_ptr(),
1184 to_cpath(secondary_path)?.as_ptr(),
1185 ))
1186 }
1187 AccessType::WithTTL { ttl } => ffi_try!(ffi::rocksdb_open_with_ttl(
1188 opts.inner,
1189 cpath.as_ptr(),
1190 ttl_to_seconds(ttl),
1191 )),
1192 }
1193 };
1194 Ok(db)
1195 }
1196
1197 #[allow(clippy::pedantic)]
1198 fn open_cf_raw(
1199 opts: &Options,
1200 cpath: &CString,
1201 cfs_v: &[ColumnFamilyDescriptor],
1202 cfnames: &[*const c_char],
1203 cfopts: &[*const ffi::rocksdb_options_t],
1204 cfhandles: &mut [*mut ffi::rocksdb_column_family_handle_t],
1205 access_type: &AccessType,
1206 ) -> Result<*mut ffi::rocksdb_t, Error> {
1207 let db = unsafe {
1208 match *access_type {
1209 AccessType::ReadOnly {
1210 error_if_log_file_exist,
1211 } => ffi_try!(ffi::rocksdb_open_for_read_only_column_families(
1212 opts.inner,
1213 cpath.as_ptr(),
1214 cfs_v.len() as c_int,
1215 cfnames.as_ptr(),
1216 cfopts.as_ptr(),
1217 cfhandles.as_mut_ptr(),
1218 c_uchar::from(error_if_log_file_exist),
1219 )),
1220 AccessType::ReadWrite => ffi_try!(ffi::rocksdb_open_column_families(
1221 opts.inner,
1222 cpath.as_ptr(),
1223 cfs_v.len() as c_int,
1224 cfnames.as_ptr(),
1225 cfopts.as_ptr(),
1226 cfhandles.as_mut_ptr(),
1227 )),
1228 AccessType::Secondary { secondary_path } => {
1229 ffi_try!(ffi::rocksdb_open_as_secondary_column_families(
1230 opts.inner,
1231 cpath.as_ptr(),
1232 to_cpath(secondary_path)?.as_ptr(),
1233 cfs_v.len() as c_int,
1234 cfnames.as_ptr(),
1235 cfopts.as_ptr(),
1236 cfhandles.as_mut_ptr(),
1237 ))
1238 }
1239 AccessType::WithTTL { ttl } => {
1240 let ttls: Vec<_> = cfs_v
1241 .iter()
1242 .map(|cf| cf_ttl_to_seconds(cf.ttl, ttl))
1243 .collect();
1244
1245 ffi_try!(ffi::rocksdb_open_column_families_with_ttl(
1246 opts.inner,
1247 cpath.as_ptr(),
1248 cfs_v.len() as c_int,
1249 cfnames.as_ptr(),
1250 cfopts.as_ptr(),
1251 cfhandles.as_mut_ptr(),
1252 ttls.as_ptr(),
1253 ))
1254 }
1255 AccessType::TrimHistory { trim_ts } => {
1256 // The C function copies the timestamp into a std::string before
1257 // doing anything with it, so this only has to stay alive across
1258 // the call. It takes a non-const pointer without writing through
1259 // it, hence the local copy rather than casting the borrow.
1260 let mut trim_ts = trim_ts.to_vec();
1261 ffi_try!(ffi::rocksdb_open_and_trim_history(
1262 opts.inner,
1263 cpath.as_ptr(),
1264 cfs_v.len() as c_int,
1265 cfnames.as_ptr(),
1266 cfopts.as_ptr(),
1267 cfhandles.as_mut_ptr(),
1268 trim_ts.as_mut_ptr().cast::<c_char>(),
1269 trim_ts.len(),
1270 ))
1271 }
1272 }
1273 };
1274 Ok(db)
1275 }
1276
1277 /// Removes the database entries in the range `["from", "to")` using given write options.
1278 pub fn delete_range_cf_opt<K: AsRef<[u8]>>(
1279 &self,
1280 cf: &impl AsColumnFamilyRef,
1281 from: K,
1282 to: K,
1283 writeopts: &WriteOptions,
1284 ) -> Result<(), Error> {
1285 let from = from.as_ref();
1286 let to = to.as_ref();
1287
1288 unsafe {
1289 ffi_try!(ffi::rocksdb_delete_range_cf(
1290 self.inner.inner(),
1291 writeopts.inner,
1292 cf.inner(),
1293 from.as_ptr() as *const c_char,
1294 from.len() as size_t,
1295 to.as_ptr() as *const c_char,
1296 to.len() as size_t,
1297 ));
1298 Ok(())
1299 }
1300 }
1301
1302 /// Removes the database entries in the range `["from", "to")` using default write options.
1303 pub fn delete_range_cf<K: AsRef<[u8]>>(
1304 &self,
1305 cf: &impl AsColumnFamilyRef,
1306 from: K,
1307 to: K,
1308 ) -> Result<(), Error> {
1309 DEFAULT_WRITE_OPTS.with(|opts| self.delete_range_cf_opt(cf, from, to, opts))
1310 }
1311
1312 pub fn write_opt(&self, batch: &WriteBatch, writeopts: &WriteOptions) -> Result<(), Error> {
1313 unsafe {
1314 ffi_try!(ffi::rocksdb_write(
1315 self.inner.inner(),
1316 writeopts.inner,
1317 batch.inner
1318 ));
1319 }
1320 Ok(())
1321 }
1322
1323 pub fn write(&self, batch: &WriteBatch) -> Result<(), Error> {
1324 DEFAULT_WRITE_OPTS.with(|opts| self.write_opt(batch, opts))
1325 }
1326
1327 pub fn write_without_wal(&self, batch: &WriteBatch) -> Result<(), Error> {
1328 let mut wo = WriteOptions::new();
1329 wo.disable_wal(true);
1330 self.write_opt(batch, &wo)
1331 }
1332
1333 pub fn write_wbwi(&self, wbwi: &WriteBatchWithIndex) -> Result<(), Error> {
1334 DEFAULT_WRITE_OPTS.with(|opts| self.write_wbwi_opt(wbwi, opts))
1335 }
1336
1337 pub fn write_wbwi_opt(
1338 &self,
1339 wbwi: &WriteBatchWithIndex,
1340 writeopts: &WriteOptions,
1341 ) -> Result<(), Error> {
1342 unsafe {
1343 ffi_try!(ffi::rocksdb_write_writebatch_wi(
1344 self.inner.inner(),
1345 writeopts.inner,
1346 wbwi.inner
1347 ));
1348
1349 Ok(())
1350 }
1351 }
1352}
1353
1354/// A value read from a DB with user-defined timestamps, and the timestamp it carries.
1355///
1356/// Both halves are RocksDB allocations this owns, freed on drop. Read them as bytes
1357/// through `AsRef<[u8]>`.
1358///
1359/// The timestamp is the raw bytes RocksDB stored, in whatever width and encoding the
1360/// column family's comparator defines, so it is only meaningful to code that knows
1361/// that comparator. RocksDB's own `comparator_with_u64_ts` uses a little endian
1362/// `u64`.
1363pub struct TimestampedValue {
1364 /// The value stored under the key.
1365 pub value: CSlice,
1366 /// The timestamp the value was written with.
1367 pub timestamp: CSlice,
1368}
1369
1370impl fmt::Debug for TimestampedValue {
1371 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1372 f.debug_struct("TimestampedValue")
1373 .field("value", &self.value.as_ref().len())
1374 .field("timestamp", &self.timestamp.as_ref())
1375 .finish()
1376 }
1377}
1378
1379/// What one `rocksdb_create_column_families` call managed to create.
1380///
1381/// The call is not atomic, so it can commit some families and then fail. Both
1382/// fields can be non-empty at once, and the handles have to be recorded even when
1383/// there is an error.
1384struct CreatedCfHandles {
1385 /// Owned handles for the families that were created, in the order the names
1386 /// were given. Each one needs `rocksdb_column_family_handle_destroy`.
1387 handles: Vec<*mut ffi::rocksdb_column_family_handle_t>,
1388 /// Why the remaining families were not created.
1389 error: Option<Error>,
1390}
1391
1392impl CreatedCfHandles {
1393 /// Nothing was created, because the call never got as far as RocksDB.
1394 fn failed(error: Error) -> Self {
1395 Self {
1396 handles: Vec::new(),
1397 error: Some(error),
1398 }
1399 }
1400}
1401
1402/// What a [`DBCommon::compact_files`] call produced.
1403pub struct CompactFilesResult {
1404 /// The SST files the compaction wrote, as RocksDB names them.
1405 ///
1406 /// Empty when the compaction had nothing to write, for instance when every
1407 /// input key was deleted.
1408 pub output_files: Vec<String>,
1409 /// Statistics and file lists for the compaction that just ran.
1410 ///
1411 /// `None` when [`CompactionOptions::set_allow_trivial_move`] is enabled.
1412 /// RocksDB can then satisfy the request by moving files between levels
1413 /// instead of rewriting them, and that path returns success without
1414 /// reporting anything about the work it did. Nothing distinguishes it from
1415 /// the rewriting path afterwards, so the statistics are not collected at all
1416 /// rather than sometimes being made up.
1417 ///
1418 /// [`CompactionOptions::set_allow_trivial_move`]:
1419 /// crate::compaction::CompactionOptions::set_allow_trivial_move
1420 pub job_info: Option<OwnedCompactionJobInfo>,
1421}
1422
1423impl fmt::Debug for CompactFilesResult {
1424 fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
1425 f.debug_struct("CompactFilesResult")
1426 .field("output_files", &self.output_files)
1427 .field("job_info", &self.job_info)
1428 .finish()
1429 }
1430}
1431
1432/// Splits an optional key into the pointer and length a RocksDB range bound wants.
1433///
1434/// A `None` bound becomes a null pointer, which RocksDB reads as unbounded. This is
1435/// why the bound is `Option<&T>` rather than `&[u8]`: an empty slice has a non-null
1436/// pointer and means the empty key, which is a different request.
1437fn optional_key_parts<T: AsRef<[u8]> + ?Sized>(key: Option<&T>) -> (*const c_char, usize) {
1438 (opt_bytes_to_ptr(key), key.map_or(0, |k| k.as_ref().len()))
1439}
1440
1441/// Reads a `char**` array RocksDB allocated into owned `String`s and frees it.
1442///
1443/// The strings are read without freeing them individually because
1444/// `rocksdb_compact_files_output_file_names_destroy` frees the whole array,
1445/// entries included.
1446///
1447/// # Safety
1448///
1449/// `names` must be null, or an array of `count` NUL-terminated strings allocated
1450/// by `rocksdb_compact_files`, which nothing else will free.
1451unsafe fn collect_and_free_output_names(names: *mut *mut c_char, count: usize) -> Vec<String> {
1452 if names.is_null() || count == 0 {
1453 return Vec::new();
1454 }
1455 let collected = (0..count)
1456 .map(|i| unsafe { from_cstr_without_free(*names.add(i)) })
1457 .collect();
1458 unsafe { ffi::rocksdb_compact_files_output_file_names_destroy(names, count) };
1459 collected
1460}
1461
1462/// The pointer arrays the `rocksdb_approximate_sizes*` family takes, kept in one
1463/// place because all five variants want the same six arguments.
1464///
1465/// The key pointers borrow from the `Range` slice, so this must not outlive it.
1466struct ApproximateSizesArgs {
1467 count: c_int,
1468 start_keys: Vec<*const c_char>,
1469 start_key_lens: Vec<usize>,
1470 end_keys: Vec<*const c_char>,
1471 end_key_lens: Vec<usize>,
1472 sizes: Vec<u64>,
1473}
1474
1475impl ApproximateSizesArgs {
1476 fn new(ranges: &[Range]) -> Self {
1477 Self {
1478 count: c_int::try_from(ranges.len()).unwrap_or(c_int::MAX),
1479 start_keys: ranges
1480 .iter()
1481 .map(|x| x.start_key.as_ptr().cast::<c_char>())
1482 .collect(),
1483 start_key_lens: ranges.iter().map(|x| x.start_key.len()).collect(),
1484 end_keys: ranges
1485 .iter()
1486 .map(|x| x.end_key.as_ptr().cast::<c_char>())
1487 .collect(),
1488 end_key_lens: ranges.iter().map(|x| x.end_key.len()).collect(),
1489 sizes: vec![0; ranges.len()],
1490 }
1491 }
1492
1493 /// Turns the `errptr` RocksDB filled in into a `Result`, yielding the sizes
1494 /// on success.
1495 ///
1496 /// RocksDB reports failures here through `errptr`. Ignoring it both leaked
1497 /// the `strdup`ed message and returned a vector of zeros that the caller
1498 /// could not distinguish from "these ranges are empty".
1499 fn finish(&mut self, err: *mut c_char) -> Result<Vec<u64>, Error> {
1500 if !err.is_null() {
1501 return Err(convert_rocksdb_error(err));
1502 }
1503 Ok(std::mem::take(&mut self.sizes))
1504 }
1505}
1506
1507/// Common methods of `DBWithThreadMode` and `OptimisticTransactionDB`.
1508impl<T: ThreadMode, D: DBInner> DBCommon<T, D> {
1509 pub(crate) fn new(inner: D, cfs: T, path: PathBuf, outlive: Vec<OptionsMustOutliveDB>) -> Self {
1510 Self {
1511 inner,
1512 cfs,
1513 path,
1514 _outlive: outlive,
1515 opened_with_ttl: None,
1516 }
1517 }
1518
1519 pub fn list_cf<P: AsRef<Path>>(opts: &Options, path: P) -> Result<Vec<String>, Error> {
1520 let cpath = to_cpath(path)?;
1521 let mut length = 0;
1522
1523 unsafe {
1524 let ptr = ffi_try!(ffi::rocksdb_list_column_families(
1525 opts.inner,
1526 cpath.as_ptr(),
1527 &raw mut length,
1528 ));
1529
1530 let vec = slice::from_raw_parts(ptr, length)
1531 .iter()
1532 .map(|ptr| from_cstr_without_free(*ptr))
1533 .collect();
1534 ffi::rocksdb_list_column_families_destroy(ptr, length);
1535 Ok(vec)
1536 }
1537 }
1538
1539 pub fn destroy<P: AsRef<Path>>(opts: &Options, path: P) -> Result<(), Error> {
1540 let cpath = to_cpath(path)?;
1541 unsafe {
1542 ffi_try!(ffi::rocksdb_destroy_db(opts.inner, cpath.as_ptr()));
1543 }
1544 Ok(())
1545 }
1546
1547 pub fn repair<P: AsRef<Path>>(opts: &Options, path: P) -> Result<(), Error> {
1548 let cpath = to_cpath(path)?;
1549 unsafe {
1550 ffi_try!(ffi::rocksdb_repair_db(opts.inner, cpath.as_ptr()));
1551 }
1552 Ok(())
1553 }
1554
1555 pub fn path(&self) -> &Path {
1556 self.path.as_path()
1557 }
1558
1559 /// Flushes the WAL buffer. If `sync` is set to `true`, also syncs
1560 /// the data to disk.
1561 pub fn flush_wal(&self, sync: bool) -> Result<(), Error> {
1562 unsafe {
1563 ffi_try!(ffi::rocksdb_flush_wal(
1564 self.inner.inner(),
1565 c_uchar::from(sync)
1566 ));
1567 }
1568 Ok(())
1569 }
1570
1571 /// Flushes the WAL buffer, taking the rate limiter priority as well as `sync`.
1572 ///
1573 /// [`flush_wal`](Self::flush_wal) covers the common case. Reach for this one to
1574 /// give the flush a priority other than the default, which decides how the WAL
1575 /// write is charged against a configured rate limiter.
1576 ///
1577 /// # Errors
1578 ///
1579 /// Returns the RocksDB error if the flush or the sync fails.
1580 pub fn flush_wal_with_options(&self, opts: &FlushWalOptions) -> Result<(), Error> {
1581 unsafe {
1582 ffi_try!(ffi::rocksdb_flush_wal_with_options(
1583 self.inner.inner(),
1584 opts.as_ptr()
1585 ));
1586 }
1587 Ok(())
1588 }
1589
1590 /// Stops background flushes and compactions and waits for the ones already
1591 /// running to finish.
1592 ///
1593 /// Writes are not blocked, so they keep filling memtables that cannot be
1594 /// flushed. Enough of them and the DB stalls or hits the stop trigger, so pair
1595 /// this with [`continue_background_work`](Self::continue_background_work) and
1596 /// keep the gap short.
1597 ///
1598 /// Calls nest. Background work resumes once as many `continue` calls have been
1599 /// made as `pause` calls.
1600 ///
1601 /// # Errors
1602 ///
1603 /// Returns the RocksDB error if the DB is shutting down.
1604 pub fn pause_background_work(&self) -> Result<(), Error> {
1605 unsafe {
1606 ffi_try!(ffi::rocksdb_pause_background_work(self.inner.inner()));
1607 }
1608 Ok(())
1609 }
1610
1611 /// Undoes one [`pause_background_work`](Self::pause_background_work) call.
1612 ///
1613 /// # Errors
1614 ///
1615 /// Returns the RocksDB error if background work was not paused, or if the DB is
1616 /// shutting down.
1617 pub fn continue_background_work(&self) -> Result<(), Error> {
1618 unsafe {
1619 ffi_try!(ffi::rocksdb_continue_background_work(self.inner.inner()));
1620 }
1621 Ok(())
1622 }
1623
1624 /// Undoes one [`disable_manual_compaction`](Self::disable_manual_compaction) call.
1625 ///
1626 /// This is a counter, so it takes as many calls here as there were calls there
1627 /// before manual compactions start running again.
1628 ///
1629 /// Pair the two. RocksDB decrements without a floor, and the assertion that would
1630 /// catch it is compiled out of this build, so calling this more often than
1631 /// `disable_manual_compaction` drives the counter below zero. That reads as
1632 /// not paused, which looks fine, but it means the next
1633 /// `disable_manual_compaction` only brings the counter back to zero and does not
1634 /// actually pause anything.
1635 pub fn enable_manual_compaction(&self) {
1636 unsafe { ffi::rocksdb_enable_manual_compaction(self.inner.inner()) }
1637 }
1638
1639 /// Cancels running manual compactions and makes later ones return immediately.
1640 ///
1641 /// Affects only manual compactions, so [`compact_range`](Self::compact_range),
1642 /// [`compact_files`](Self::compact_files) and the like. Automatic background
1643 /// compaction keeps going. Use it to get out of a long manual compaction during
1644 /// shutdown without waiting for it.
1645 ///
1646 /// This increments a counter, so it needs a matching
1647 /// [`enable_manual_compaction`](Self::enable_manual_compaction) call for each
1648 /// call here.
1649 pub fn disable_manual_compaction(&self) {
1650 unsafe { ffi::rocksdb_disable_manual_compaction(self.inner.inner()) }
1651 }
1652
1653 /// Reads every live SST and blob file and checks its block checksums.
1654 ///
1655 /// Reads the whole DB, so it is as slow as the data is large.
1656 ///
1657 /// # Errors
1658 ///
1659 /// Returns the RocksDB error naming the first corrupt file found.
1660 pub fn verify_checksum(&self) -> Result<(), Error> {
1661 unsafe {
1662 ffi_try!(ffi::rocksdb_verify_checksum(self.inner.inner()));
1663 }
1664 Ok(())
1665 }
1666
1667 /// Like [`verify_checksum`](Self::verify_checksum), reading through `readopts`.
1668 ///
1669 /// Worth setting when the default read path is not what you want for a full
1670 /// scan, for instance to keep the verification from filling the block cache or
1671 /// to give it a rate limiter priority.
1672 ///
1673 /// # Errors
1674 ///
1675 /// See [`verify_checksum`](Self::verify_checksum).
1676 pub fn verify_checksum_opt(&self, readopts: &ReadOptions) -> Result<(), Error> {
1677 unsafe {
1678 ffi_try!(ffi::rocksdb_verify_checksum_with_options(
1679 self.inner.inner(),
1680 readopts.inner
1681 ));
1682 }
1683 Ok(())
1684 }
1685
1686 /// Recomputes each live file's whole file checksum and compares it against the
1687 /// one recorded in the manifest.
1688 ///
1689 /// This is the file level check, as opposed to the per block one
1690 /// [`verify_checksum`](Self::verify_checksum) does. It requires the DB to have
1691 /// been written with a file checksum generator, see
1692 /// [`Options::set_file_checksum_gen_factory`](crate::Options::set_file_checksum_gen_factory).
1693 ///
1694 /// # Errors
1695 ///
1696 /// Returns the RocksDB error naming the first file whose checksum does not match
1697 /// what the manifest recorded. Also errors when no generator is configured, since
1698 /// then there is nothing recorded to compare against, rather than treating that
1699 /// as having nothing to check.
1700 pub fn verify_file_checksums(&self) -> Result<(), Error> {
1701 unsafe {
1702 ffi_try!(ffi::rocksdb_verify_file_checksums(self.inner.inner()));
1703 }
1704 Ok(())
1705 }
1706
1707 /// Like [`verify_file_checksums`](Self::verify_file_checksums), reading through
1708 /// `readopts`.
1709 ///
1710 /// # Errors
1711 ///
1712 /// See [`verify_file_checksums`](Self::verify_file_checksums).
1713 pub fn verify_file_checksums_opt(&self, readopts: &ReadOptions) -> Result<(), Error> {
1714 unsafe {
1715 ffi_try!(ffi::rocksdb_verify_file_checksums_with_options(
1716 self.inner.inner(),
1717 readopts.inner
1718 ));
1719 }
1720 Ok(())
1721 }
1722
1723 /// Marks the files overlapping `[start, end)` for compaction and returns
1724 /// without compacting anything.
1725 ///
1726 /// Background compaction picks the marked files up on its own schedule, so this
1727 /// returns as soon as the marking is done. That makes it the cheap way to hint
1728 /// that a range is worth compacting, as opposed to
1729 /// [`compact_range`](Self::compact_range), which does the work before it
1730 /// returns.
1731 ///
1732 /// `None` for either bound means unbounded in that direction. An empty slice is
1733 /// a real empty key, not the same thing.
1734 ///
1735 /// # Errors
1736 ///
1737 /// Returns the RocksDB error if the range cannot be marked.
1738 pub fn suggest_compact_range<S: AsRef<[u8]>, E: AsRef<[u8]>>(
1739 &self,
1740 start: Option<S>,
1741 end: Option<E>,
1742 ) -> Result<(), Error> {
1743 let (start, start_len) = optional_key_parts(start.as_ref());
1744 let (end, end_len) = optional_key_parts(end.as_ref());
1745 unsafe {
1746 ffi_try!(ffi::rocksdb_suggest_compact_range(
1747 self.inner.inner(),
1748 start,
1749 start_len,
1750 end,
1751 end_len
1752 ));
1753 }
1754 Ok(())
1755 }
1756
1757 /// Like [`suggest_compact_range`](Self::suggest_compact_range), for a single
1758 /// column family.
1759 ///
1760 /// # Errors
1761 ///
1762 /// See [`suggest_compact_range`](Self::suggest_compact_range).
1763 pub fn suggest_compact_range_cf<S: AsRef<[u8]>, E: AsRef<[u8]>>(
1764 &self,
1765 cf: &impl AsColumnFamilyRef,
1766 start: Option<S>,
1767 end: Option<E>,
1768 ) -> Result<(), Error> {
1769 let (start, start_len) = optional_key_parts(start.as_ref());
1770 let (end, end_len) = optional_key_parts(end.as_ref());
1771 unsafe {
1772 ffi_try!(ffi::rocksdb_suggest_compact_range_cf(
1773 self.inner.inner(),
1774 cf.inner(),
1775 start,
1776 start_len,
1777 end,
1778 end_len
1779 ));
1780 }
1781 Ok(())
1782 }
1783
1784 /// Reads a key along with the user-defined timestamp its value was written with.
1785 ///
1786 /// Only meaningful on a column family configured for user-defined timestamps, so
1787 /// one whose comparator carries a timestamp size. The plain
1788 /// [`get`](Self::get) reads the same value but throws the timestamp away, and
1789 /// there is no pinned equivalent of this call in the C API.
1790 ///
1791 /// `readopts` decides which timestamp is read. Without a read timestamp set on
1792 /// it RocksDB rejects the read rather than picking one, see
1793 /// [`ReadOptions::set_timestamp`](crate::ReadOptions::set_timestamp).
1794 ///
1795 /// # Errors
1796 ///
1797 /// Returns the RocksDB error if the read fails. A key that is not present is
1798 /// `Ok(None)`, not an error.
1799 pub fn get_with_ts<K: AsRef<[u8]>>(
1800 &self,
1801 key: K,
1802 readopts: &ReadOptions,
1803 ) -> Result<Option<TimestampedValue>, Error> {
1804 let key = key.as_ref();
1805 self.get_with_ts_impl(readopts, |db, ro, vallen, ts, tslen, err| unsafe {
1806 ffi::rocksdb_get_with_ts(
1807 db,
1808 ro,
1809 key.as_ptr().cast::<c_char>(),
1810 key.len(),
1811 vallen,
1812 ts,
1813 tslen,
1814 err,
1815 )
1816 })
1817 }
1818
1819 /// Like [`get_with_ts`](Self::get_with_ts), for a single column family.
1820 ///
1821 /// # Errors
1822 ///
1823 /// See [`get_with_ts`](Self::get_with_ts).
1824 pub fn get_cf_with_ts<K: AsRef<[u8]>>(
1825 &self,
1826 cf: &impl AsColumnFamilyRef,
1827 key: K,
1828 readopts: &ReadOptions,
1829 ) -> Result<Option<TimestampedValue>, Error> {
1830 let key = key.as_ref();
1831 let cf = cf.inner();
1832 self.get_with_ts_impl(readopts, |db, ro, vallen, ts, tslen, err| unsafe {
1833 ffi::rocksdb_get_cf_with_ts(
1834 db,
1835 ro,
1836 cf,
1837 key.as_ptr().cast::<c_char>(),
1838 key.len(),
1839 vallen,
1840 ts,
1841 tslen,
1842 err,
1843 )
1844 })
1845 }
1846
1847 /// Shared tail of the two timestamped gets.
1848 ///
1849 /// `call` is handed the out-params and returns the value pointer. Note that
1850 /// RocksDB only writes through `ts` when the read succeeds, so `ts` starts as
1851 /// null here and a miss leaves it that way rather than leaving it indeterminate
1852 /// (c.cc:2592-2603).
1853 fn get_with_ts_impl(
1854 &self,
1855 readopts: &ReadOptions,
1856 call: impl FnOnce(
1857 *mut ffi::rocksdb_t,
1858 *const ffi::rocksdb_readoptions_t,
1859 *mut usize,
1860 *mut *mut c_char,
1861 *mut usize,
1862 *mut *mut c_char,
1863 ) -> *mut c_char,
1864 ) -> Result<Option<TimestampedValue>, Error> {
1865 let mut vallen: usize = 0;
1866 let mut ts: *mut c_char = ptr::null_mut();
1867 let mut tslen: usize = 0;
1868 let mut err: *mut c_char = ptr::null_mut();
1869
1870 let value = call(
1871 self.inner.inner(),
1872 readopts.inner,
1873 &raw mut vallen,
1874 &raw mut ts,
1875 &raw mut tslen,
1876 &raw mut err,
1877 );
1878
1879 if !err.is_null() {
1880 // RocksDB reports a failure without allocating either output, but free
1881 // anything it did hand back rather than trusting that on an error path.
1882 unsafe {
1883 if !value.is_null() {
1884 ffi::rocksdb_free(value.cast::<c_void>());
1885 }
1886 if !ts.is_null() {
1887 ffi::rocksdb_free(ts.cast::<c_void>());
1888 }
1889 }
1890 return Err(convert_rocksdb_error(err));
1891 }
1892
1893 if value.is_null() {
1894 return Ok(None);
1895 }
1896
1897 // SAFETY: both pointers came from RocksDB's `CopyString`, so they are
1898 // `malloc`ed buffers of the reported length that nothing else frees, which is
1899 // exactly what `CSlice` takes over.
1900 unsafe {
1901 Ok(Some(TimestampedValue {
1902 value: CSlice::from_raw_parts(value, vallen),
1903 timestamp: CSlice::from_raw_parts(ts, tslen),
1904 }))
1905 }
1906 }
1907
1908 /// Reads many keys along with the timestamps their values were written with.
1909 ///
1910 /// One native batch, results in input order, one `Result` per key so a single
1911 /// bad key does not sink the batch. See [`get_with_ts`](Self::get_with_ts) for
1912 /// what the timestamp means and what `readopts` has to carry.
1913 pub fn multi_get_with_ts<K, I>(
1914 &self,
1915 keys: I,
1916 readopts: &ReadOptions,
1917 ) -> Vec<Result<Option<TimestampedValue>, Error>>
1918 where
1919 K: AsRef<[u8]>,
1920 I: IntoIterator<Item = K>,
1921 {
1922 let owned_keys: Vec<K> = keys.into_iter().collect();
1923 let (ptr_keys, keys_sizes) = key_ptrs_and_sizes(&owned_keys);
1924 let mut out = MultiGetTsOut::with_capacity(ptr_keys.len());
1925
1926 unsafe {
1927 ffi::rocksdb_multi_get_with_ts(
1928 self.inner.inner(),
1929 readopts.inner,
1930 ptr_keys.len(),
1931 ptr_keys.as_ptr(),
1932 keys_sizes.as_ptr(),
1933 out.values.as_mut_ptr(),
1934 out.values_sizes.as_mut_ptr(),
1935 out.timestamps.as_mut_ptr(),
1936 out.timestamps_sizes.as_mut_ptr(),
1937 out.errors.as_mut_ptr(),
1938 );
1939 out.assume_filled(ptr_keys.len());
1940 }
1941
1942 out.into_results()
1943 }
1944
1945 /// Like [`multi_get_with_ts`](Self::multi_get_with_ts), for one column family per
1946 /// key.
1947 pub fn multi_get_cf_with_ts<'c, K, I, W>(
1948 &self,
1949 keys: I,
1950 readopts: &ReadOptions,
1951 ) -> Vec<Result<Option<TimestampedValue>, Error>>
1952 where
1953 K: AsRef<[u8]>,
1954 I: IntoIterator<Item = (&'c W, K)>,
1955 W: AsColumnFamilyRef + 'c,
1956 {
1957 let (cfs, owned_keys): (Vec<_>, Vec<K>) = keys.into_iter().unzip();
1958 let cf_ptrs: Vec<*const ffi::rocksdb_column_family_handle_t> =
1959 cfs.iter().map(|cf| cf.inner().cast_const()).collect();
1960 let (ptr_keys, keys_sizes) = key_ptrs_and_sizes(&owned_keys);
1961 let mut out = MultiGetTsOut::with_capacity(ptr_keys.len());
1962
1963 unsafe {
1964 ffi::rocksdb_multi_get_cf_with_ts(
1965 self.inner.inner(),
1966 readopts.inner,
1967 cf_ptrs.as_ptr(),
1968 ptr_keys.len(),
1969 ptr_keys.as_ptr(),
1970 keys_sizes.as_ptr(),
1971 out.values.as_mut_ptr(),
1972 out.values_sizes.as_mut_ptr(),
1973 out.timestamps.as_mut_ptr(),
1974 out.timestamps_sizes.as_mut_ptr(),
1975 out.errors.as_mut_ptr(),
1976 );
1977 out.assume_filled(ptr_keys.len());
1978 }
1979
1980 out.into_results()
1981 }
1982
1983 /// Changes DB wide mutable options at runtime.
1984 ///
1985 /// Takes the same option names and string values the RocksDB configuration
1986 /// strings use, so `max_background_jobs` or `bytes_per_sync`. Only options
1987 /// RocksDB marks mutable at the DB level can be set, and the whole call is
1988 /// rejected if any name or value is not accepted.
1989 ///
1990 /// [`set_options`](Self::set_options) is the column family equivalent.
1991 ///
1992 /// # Aborts
1993 ///
1994 /// Some unparseable values take the process down instead of returning an error.
1995 /// See [`set_options`](Self::set_options) for the details and why this is not
1996 /// caught.
1997 ///
1998 /// # Errors
1999 ///
2000 /// Returns the RocksDB error if a name is unknown or the option is not
2001 /// changeable at runtime, and for an empty `opts`, which RocksDB rejects as
2002 /// `empty input`. Also errors if any name or value contains an interior NUL
2003 /// byte.
2004 pub fn set_db_options(&self, opts: &[(&str, &str)]) -> Result<(), Error> {
2005 let copts = convert_options(opts)?;
2006 let names: Vec<*const c_char> = copts.iter().map(|(n, _)| n.as_ptr()).collect();
2007 let values: Vec<*const c_char> = copts.iter().map(|(_, v)| v.as_ptr()).collect();
2008
2009 unsafe {
2010 ffi_try!(ffi::rocksdb_set_db_options(
2011 self.inner.inner(),
2012 option_count(&copts)?,
2013 names.as_ptr(),
2014 values.as_ptr(),
2015 ));
2016 }
2017 Ok(())
2018 }
2019
2020 /// Suspend deleting obsolete files. Compactions will continue to occur,
2021 /// but no obsolete files will be deleted. To resume file deletions, each
2022 /// call to disable_file_deletions() must be matched by a subsequent call to
2023 /// enable_file_deletions(). For more details, see enable_file_deletions().
2024 pub fn disable_file_deletions(&self) -> Result<(), Error> {
2025 unsafe {
2026 ffi_try!(ffi::rocksdb_disable_file_deletions(self.inner.inner()));
2027 }
2028 Ok(())
2029 }
2030
2031 /// Resume deleting obsolete files, following up on `disable_file_deletions()`.
2032 ///
2033 /// File deletions disabling and enabling is not controlled by a binary flag,
2034 /// instead it's represented as a counter to allow different callers to
2035 /// independently disable file deletion. Disabling file deletion can be
2036 /// critical for operations like making a backup. So the counter implementation
2037 /// makes the file deletion disabled as long as there is one caller requesting
2038 /// so, and only when every caller agrees to re-enable file deletion, it will
2039 /// be enabled. Two threads can call this method concurrently without
2040 /// synchronization -- i.e., file deletions will be enabled only after both
2041 /// threads call enable_file_deletions()
2042 pub fn enable_file_deletions(&self) -> Result<(), Error> {
2043 unsafe {
2044 ffi_try!(ffi::rocksdb_enable_file_deletions(self.inner.inner()));
2045 }
2046 Ok(())
2047 }
2048
2049 /// Flushes database memtables to SST files on the disk.
2050 pub fn flush_opt(&self, flushopts: &FlushOptions) -> Result<(), Error> {
2051 unsafe {
2052 ffi_try!(ffi::rocksdb_flush(self.inner.inner(), flushopts.inner));
2053 }
2054 Ok(())
2055 }
2056
2057 /// Flushes database memtables to SST files on the disk using default options.
2058 pub fn flush(&self) -> Result<(), Error> {
2059 DEFAULT_FLUSH_OPTS.with(|opts| self.flush_opt(opts))
2060 }
2061
2062 /// Flushes database memtables to SST files on the disk for a given column family.
2063 pub fn flush_cf_opt(
2064 &self,
2065 cf: &impl AsColumnFamilyRef,
2066 flushopts: &FlushOptions,
2067 ) -> Result<(), Error> {
2068 unsafe {
2069 ffi_try!(ffi::rocksdb_flush_cf(
2070 self.inner.inner(),
2071 flushopts.inner,
2072 cf.inner()
2073 ));
2074 }
2075 Ok(())
2076 }
2077
2078 /// Flushes multiple column families.
2079 ///
2080 /// If atomic flush is not enabled, it is equivalent to calling flush_cf multiple times.
2081 /// If atomic flush is enabled, it will flush all column families specified in `cfs` up to the latest sequence
2082 /// number at the time when flush is requested.
2083 pub fn flush_cfs_opt(
2084 &self,
2085 cfs: &[&impl AsColumnFamilyRef],
2086 opts: &FlushOptions,
2087 ) -> Result<(), Error> {
2088 let mut cfs = cfs.iter().map(|cf| cf.inner()).collect::<Vec<_>>();
2089 unsafe {
2090 ffi_try!(ffi::rocksdb_flush_cfs(
2091 self.inner.inner(),
2092 opts.inner,
2093 cfs.as_mut_ptr(),
2094 cfs.len() as libc::c_int,
2095 ));
2096 }
2097 Ok(())
2098 }
2099
2100 /// Flushes database memtables to SST files on the disk for a given column family using default
2101 /// options.
2102 pub fn flush_cf(&self, cf: &impl AsColumnFamilyRef) -> Result<(), Error> {
2103 DEFAULT_FLUSH_OPTS.with(|opts| self.flush_cf_opt(cf, opts))
2104 }
2105
2106 /// Return the bytes associated with a key value with read options. If you only intend to use
2107 /// the vector returned temporarily, consider using [`get_pinned_opt`](#method.get_pinned_opt)
2108 /// to avoid unnecessary memory copy.
2109 pub fn get_opt<K: AsRef<[u8]>>(
2110 &self,
2111 key: K,
2112 readopts: &ReadOptions,
2113 ) -> Result<Option<Vec<u8>>, Error> {
2114 self.get_pinned_opt(key, readopts)
2115 .map(|x| x.map(|v| v.as_ref().to_vec()))
2116 }
2117
2118 /// Return the bytes associated with a key value. If you only intend to use the vector returned
2119 /// temporarily, consider using [`get_pinned`](#method.get_pinned) to avoid unnecessary memory
2120 /// copy.
2121 pub fn get<K: AsRef<[u8]>>(&self, key: K) -> Result<Option<Vec<u8>>, Error> {
2122 DEFAULT_READ_OPTS.with(|opts| self.get_opt(key.as_ref(), opts))
2123 }
2124
2125 /// Return the bytes associated with a key value and the given column family with read options.
2126 /// If you only intend to use the vector returned temporarily, consider using
2127 /// [`get_pinned_cf_opt`](#method.get_pinned_cf_opt) to avoid unnecessary memory.
2128 pub fn get_cf_opt<K: AsRef<[u8]>>(
2129 &self,
2130 cf: &impl AsColumnFamilyRef,
2131 key: K,
2132 readopts: &ReadOptions,
2133 ) -> Result<Option<Vec<u8>>, Error> {
2134 self.get_pinned_cf_opt(cf, key, readopts)
2135 .map(|x| x.map(|v| v.as_ref().to_vec()))
2136 }
2137
2138 /// Return the bytes associated with a key value and the given column family. If you only
2139 /// intend to use the vector returned temporarily, consider using
2140 /// [`get_pinned_cf`](#method.get_pinned_cf) to avoid unnecessary memory.
2141 pub fn get_cf<K: AsRef<[u8]>>(
2142 &self,
2143 cf: &impl AsColumnFamilyRef,
2144 key: K,
2145 ) -> Result<Option<Vec<u8>>, Error> {
2146 DEFAULT_READ_OPTS.with(|opts| self.get_cf_opt(cf, key.as_ref(), opts))
2147 }
2148
2149 /// Return the value associated with a key using RocksDB's PinnableSlice
2150 /// so as to avoid unnecessary memory copy.
2151 pub fn get_pinned_opt<K: AsRef<[u8]>>(
2152 &'_ self,
2153 key: K,
2154 readopts: &ReadOptions,
2155 ) -> Result<Option<DBPinnableSlice<'_>>, Error> {
2156 if readopts.inner.is_null() {
2157 return Err(Error::new(
2158 "Unable to create RocksDB read options. This is a fairly trivial call, and its \
2159 failure may be indicative of a mis-compiled or mis-loaded RocksDB library."
2160 .to_owned(),
2161 ));
2162 }
2163
2164 let key = key.as_ref();
2165 unsafe {
2166 let val = ffi_try!(ffi::rocksdb_get_pinned(
2167 self.inner.inner(),
2168 readopts.inner,
2169 key.as_ptr() as *const c_char,
2170 key.len() as size_t,
2171 ));
2172 if val.is_null() {
2173 Ok(None)
2174 } else {
2175 Ok(Some(DBPinnableSlice::from_c(val)))
2176 }
2177 }
2178 }
2179
2180 /// Return the value associated with a key using RocksDB's PinnableSlice
2181 /// so as to avoid unnecessary memory copy. Similar to get_pinned_opt but
2182 /// leverages default options.
2183 pub fn get_pinned<K: AsRef<[u8]>>(
2184 &'_ self,
2185 key: K,
2186 ) -> Result<Option<DBPinnableSlice<'_>>, Error> {
2187 DEFAULT_READ_OPTS.with(|opts| self.get_pinned_opt(key, opts))
2188 }
2189
2190 /// Return the value associated with a key using RocksDB's PinnableSlice
2191 /// so as to avoid unnecessary memory copy. Similar to get_pinned_opt but
2192 /// allows specifying ColumnFamily
2193 pub fn get_pinned_cf_opt<K: AsRef<[u8]>>(
2194 &'_ self,
2195 cf: &impl AsColumnFamilyRef,
2196 key: K,
2197 readopts: &ReadOptions,
2198 ) -> Result<Option<DBPinnableSlice<'_>>, Error> {
2199 if readopts.inner.is_null() {
2200 return Err(Error::new(
2201 "Unable to create RocksDB read options. This is a fairly trivial call, and its \
2202 failure may be indicative of a mis-compiled or mis-loaded RocksDB library."
2203 .to_owned(),
2204 ));
2205 }
2206
2207 let key = key.as_ref();
2208 unsafe {
2209 let val = ffi_try!(ffi::rocksdb_get_pinned_cf(
2210 self.inner.inner(),
2211 readopts.inner,
2212 cf.inner(),
2213 key.as_ptr() as *const c_char,
2214 key.len() as size_t,
2215 ));
2216 if val.is_null() {
2217 Ok(None)
2218 } else {
2219 Ok(Some(DBPinnableSlice::from_c(val)))
2220 }
2221 }
2222 }
2223
2224 /// Return the value associated with a key using RocksDB's PinnableSlice
2225 /// so as to avoid unnecessary memory copy. Similar to get_pinned_cf_opt but
2226 /// leverages default options.
2227 pub fn get_pinned_cf<K: AsRef<[u8]>>(
2228 &'_ self,
2229 cf: &impl AsColumnFamilyRef,
2230 key: K,
2231 ) -> Result<Option<DBPinnableSlice<'_>>, Error> {
2232 DEFAULT_READ_OPTS.with(|opts| self.get_pinned_cf_opt(cf, key, opts))
2233 }
2234
2235 /// Read a value directly into a caller-provided buffer, avoiding memory allocation.
2236 ///
2237 /// This is the most efficient way to read values when you have a pre-allocated
2238 /// buffer. It completely avoids the allocation overhead of [`get`](#method.get)
2239 /// and even the pinning overhead of [`get_pinned`](#method.get_pinned).
2240 ///
2241 /// # Arguments
2242 ///
2243 /// * `key` - The key to look up
2244 /// * `buffer` - A mutable byte slice to write the value into. Can be empty if you
2245 /// only want to check if a key exists and get its value size.
2246 ///
2247 /// # Returns
2248 ///
2249 /// * `Ok(GetIntoBufferResult::NotFound)` - The key doesn't exist
2250 /// * `Ok(GetIntoBufferResult::Found(size))` - Value was copied into the buffer.
2251 /// `size` is the number of bytes written.
2252 /// * `Ok(GetIntoBufferResult::BufferTooSmall(size))` - The value exists but the buffer
2253 /// is too small. `size` is the actual value size needed. No data is written.
2254 /// * `Err(...)` - Database error occurred
2255 ///
2256 /// # Performance
2257 ///
2258 /// This method is ideal for high-throughput scenarios where you can reuse a buffer:
2259 ///
2260 /// ```ignore
2261 /// use rust_rocksdb::{DB, GetIntoBufferResult};
2262 ///
2263 /// let db: DB = /* open database */;
2264 /// let keys_to_lookup: Vec<&[u8]> = /* keys to look up */;
2265 /// let mut buffer = vec![0u8; 4096]; // Reusable buffer
2266 ///
2267 /// for key in keys_to_lookup {
2268 /// match db.get_into_buffer(key, &mut buffer).unwrap() {
2269 /// GetIntoBufferResult::Found(len) => {
2270 /// process_value(&buffer[..len]);
2271 /// }
2272 /// GetIntoBufferResult::BufferTooSmall(needed) => {
2273 /// buffer.resize(needed, 0);
2274 /// // Retry with larger buffer...
2275 /// }
2276 /// GetIntoBufferResult::NotFound => {}
2277 /// }
2278 /// }
2279 /// ```
2280 ///
2281 /// # Example
2282 ///
2283 /// ```
2284 /// use rust_rocksdb::{DB, GetIntoBufferResult};
2285 ///
2286 /// let tempdir = tempfile::Builder::new()
2287 /// .prefix("rocksdb_get_into_buffer")
2288 /// .tempdir()
2289 /// .unwrap();
2290 /// let db = DB::open_default(tempdir.path()).unwrap();
2291 /// db.put(b"key", b"value").unwrap();
2292 ///
2293 /// let mut buffer = [0u8; 100];
2294 /// match db.get_into_buffer(b"key", &mut buffer).unwrap() {
2295 /// GetIntoBufferResult::Found(size) => {
2296 /// assert_eq!(&buffer[..size], b"value");
2297 /// }
2298 /// GetIntoBufferResult::NotFound => panic!("expected value"),
2299 /// GetIntoBufferResult::BufferTooSmall(needed) => {
2300 /// panic!("buffer too small, need {} bytes", needed)
2301 /// }
2302 /// }
2303 /// ```
2304 pub fn get_into_buffer<K: AsRef<[u8]>>(
2305 &self,
2306 key: K,
2307 buffer: &mut [u8],
2308 ) -> Result<GetIntoBufferResult, Error> {
2309 DEFAULT_READ_OPTS.with(|opts| self.get_into_buffer_opt(key, buffer, opts))
2310 }
2311
2312 /// Read a value directly into a caller-provided buffer with custom read options.
2313 ///
2314 /// This is the same as [`get_into_buffer`](#method.get_into_buffer) but allows
2315 /// specifying custom [`ReadOptions`], such as setting a snapshot or fill cache behavior.
2316 ///
2317 /// See [`get_into_buffer`](#method.get_into_buffer) for full documentation.
2318 pub fn get_into_buffer_opt<K: AsRef<[u8]>>(
2319 &self,
2320 key: K,
2321 buffer: &mut [u8],
2322 readopts: &ReadOptions,
2323 ) -> Result<GetIntoBufferResult, Error> {
2324 if readopts.inner.is_null() {
2325 return Err(Error::new(
2326 "Unable to create RocksDB read options. This is a fairly trivial call, and its \
2327 failure may be indicative of a mis-compiled or mis-loaded RocksDB library."
2328 .to_owned(),
2329 ));
2330 }
2331
2332 let key = key.as_ref();
2333 let mut val_len: size_t = 0;
2334 let mut found: c_uchar = 0;
2335
2336 unsafe {
2337 let success = ffi_try!(ffi::rocksdb_get_into_buffer(
2338 self.inner.inner(),
2339 readopts.inner,
2340 key.as_ptr() as *const c_char,
2341 key.len() as size_t,
2342 buffer.as_mut_ptr() as *mut c_char,
2343 buffer.len() as size_t,
2344 &raw mut val_len,
2345 &raw mut found,
2346 ));
2347
2348 if found == 0 {
2349 Ok(GetIntoBufferResult::NotFound)
2350 } else if success != 0 {
2351 Ok(GetIntoBufferResult::Found(val_len))
2352 } else {
2353 Ok(GetIntoBufferResult::BufferTooSmall(val_len))
2354 }
2355 }
2356 }
2357
2358 /// Read a value from a column family directly into a caller-provided buffer.
2359 ///
2360 /// This is the column family variant of [`get_into_buffer`](#method.get_into_buffer).
2361 /// See that method for full documentation on the zero-allocation buffer API.
2362 ///
2363 /// # Arguments
2364 ///
2365 /// * `cf` - The column family to read from
2366 /// * `key` - The key to look up
2367 /// * `buffer` - A mutable byte slice to write the value into
2368 pub fn get_into_buffer_cf<K: AsRef<[u8]>>(
2369 &self,
2370 cf: &impl AsColumnFamilyRef,
2371 key: K,
2372 buffer: &mut [u8],
2373 ) -> Result<GetIntoBufferResult, Error> {
2374 DEFAULT_READ_OPTS.with(|opts| self.get_into_buffer_cf_opt(cf, key, buffer, opts))
2375 }
2376
2377 /// Read a value from a column family directly into a caller-provided buffer
2378 /// with custom read options.
2379 ///
2380 /// This is the column family variant of [`get_into_buffer_opt`](#method.get_into_buffer_opt).
2381 /// See [`get_into_buffer`](#method.get_into_buffer) for full documentation.
2382 pub fn get_into_buffer_cf_opt<K: AsRef<[u8]>>(
2383 &self,
2384 cf: &impl AsColumnFamilyRef,
2385 key: K,
2386 buffer: &mut [u8],
2387 readopts: &ReadOptions,
2388 ) -> Result<GetIntoBufferResult, Error> {
2389 if readopts.inner.is_null() {
2390 return Err(Error::new(
2391 "Unable to create RocksDB read options. This is a fairly trivial call, and its \
2392 failure may be indicative of a mis-compiled or mis-loaded RocksDB library."
2393 .to_owned(),
2394 ));
2395 }
2396
2397 let key = key.as_ref();
2398 let mut val_len: size_t = 0;
2399 let mut found: c_uchar = 0;
2400
2401 unsafe {
2402 let success = ffi_try!(ffi::rocksdb_get_into_buffer_cf(
2403 self.inner.inner(),
2404 readopts.inner,
2405 cf.inner(),
2406 key.as_ptr() as *const c_char,
2407 key.len() as size_t,
2408 buffer.as_mut_ptr() as *mut c_char,
2409 buffer.len() as size_t,
2410 &raw mut val_len,
2411 &raw mut found,
2412 ));
2413
2414 if found == 0 {
2415 Ok(GetIntoBufferResult::NotFound)
2416 } else if success != 0 {
2417 Ok(GetIntoBufferResult::Found(val_len))
2418 } else {
2419 Ok(GetIntoBufferResult::BufferTooSmall(val_len))
2420 }
2421 }
2422 }
2423
2424 /// Return the values associated with the given keys.
2425 pub fn multi_get<K, I>(&self, keys: I) -> Vec<Result<Option<Vec<u8>>, Error>>
2426 where
2427 K: AsRef<[u8]>,
2428 I: IntoIterator<Item = K>,
2429 {
2430 DEFAULT_READ_OPTS.with(|opts| self.multi_get_opt(keys, opts))
2431 }
2432
2433 /// Return the values associated with the given keys using read options.
2434 pub fn multi_get_opt<K, I>(
2435 &self,
2436 keys: I,
2437 readopts: &ReadOptions,
2438 ) -> Vec<Result<Option<Vec<u8>>, Error>>
2439 where
2440 K: AsRef<[u8]>,
2441 I: IntoIterator<Item = K>,
2442 {
2443 let owned_keys: Vec<K> = keys.into_iter().collect();
2444 let (ptr_keys, keys_sizes): (Vec<*const c_char>, Vec<usize>) = owned_keys
2445 .iter()
2446 .map(|k| {
2447 let key = k.as_ref();
2448 (key.as_ptr() as *const c_char, key.len())
2449 })
2450 .unzip();
2451
2452 let mut values: Vec<*mut c_char> = Vec::with_capacity(ptr_keys.len());
2453 let mut values_sizes: Vec<usize> = Vec::with_capacity(ptr_keys.len());
2454 let mut errors: Vec<*mut c_char> = Vec::with_capacity(ptr_keys.len());
2455 unsafe {
2456 ffi::rocksdb_multi_get(
2457 self.inner.inner(),
2458 readopts.inner,
2459 ptr_keys.len(),
2460 ptr_keys.as_ptr(),
2461 keys_sizes.as_ptr(),
2462 values.as_mut_ptr(),
2463 values_sizes.as_mut_ptr(),
2464 errors.as_mut_ptr(),
2465 );
2466 }
2467
2468 unsafe {
2469 values.set_len(ptr_keys.len());
2470 values_sizes.set_len(ptr_keys.len());
2471 errors.set_len(ptr_keys.len());
2472 }
2473
2474 convert_values(values, values_sizes, errors)
2475 }
2476
2477 /// Returns pinned values associated with the given keys using default read options.
2478 ///
2479 /// RocksDB processes the keys in one native batch. Results stay in input order.
2480 pub fn multi_get_pinned<K, I>(
2481 &'_ self,
2482 keys: I,
2483 ) -> Vec<Result<Option<DBPinnableSlice<'_>>, Error>>
2484 where
2485 K: AsRef<[u8]>,
2486 I: IntoIterator<Item = K>,
2487 {
2488 DEFAULT_READ_OPTS.with(|opts| self.multi_get_pinned_opt(keys, opts))
2489 }
2490
2491 /// Returns pinned values associated with the given keys using the provided read options.
2492 ///
2493 /// RocksDB processes the keys in one native batch. Results stay in input order.
2494 pub fn multi_get_pinned_opt<K, I>(
2495 &'_ self,
2496 keys: I,
2497 readopts: &ReadOptions,
2498 ) -> Vec<Result<Option<DBPinnableSlice<'_>>, Error>>
2499 where
2500 K: AsRef<[u8]>,
2501 I: IntoIterator<Item = K>,
2502 {
2503 let mut keys = keys.into_iter();
2504 let Some(first) = keys.next() else {
2505 return Vec::new();
2506 };
2507 // Decide before collecting. A single key does not benefit from the
2508 // native batch, and buying a key-slice vector, two result vectors and
2509 // a default column family handle to do one point lookup is a loss.
2510 let Some(second) = keys.next() else {
2511 return vec![self.get_pinned_opt(first.as_ref(), readopts)];
2512 };
2513 let mut owned_keys = Vec::with_capacity(2 + keys.size_hint().0);
2514 owned_keys.push(first);
2515 owned_keys.push(second);
2516 owned_keys.extend(keys);
2517 self.batched_multi_get_pinned_owned(&owned_keys, false, readopts)
2518 }
2519
2520 /// Returns pinned values associated with the given keys and column families
2521 /// using default read options.
2522 pub fn multi_get_pinned_cf<'a, 'b: 'a, K, I, W>(
2523 &'a self,
2524 keys: I,
2525 ) -> Vec<Result<Option<DBPinnableSlice<'a>>, Error>>
2526 where
2527 K: AsRef<[u8]>,
2528 I: IntoIterator<Item = (&'b W, K)>,
2529 W: 'b + AsColumnFamilyRef,
2530 {
2531 DEFAULT_READ_OPTS.with(|opts| self.multi_get_pinned_cf_opt(keys, opts))
2532 }
2533
2534 /// Returns pinned values associated with the given keys and column families
2535 /// using the provided read options.
2536 pub fn multi_get_pinned_cf_opt<'a, 'b: 'a, K, I, W>(
2537 &'a self,
2538 keys: I,
2539 readopts: &ReadOptions,
2540 ) -> Vec<Result<Option<DBPinnableSlice<'a>>, Error>>
2541 where
2542 K: AsRef<[u8]>,
2543 I: IntoIterator<Item = (&'b W, K)>,
2544 W: 'b + AsColumnFamilyRef,
2545 {
2546 keys.into_iter()
2547 .map(|(cf, k)| self.get_pinned_cf_opt(cf, k, readopts))
2548 .collect()
2549 }
2550
2551 /// Returns pinned values for default-column-family keys in one native batch.
2552 ///
2553 /// Set `sorted_input` only when keys are sorted according to the column
2554 /// family's comparator. Results stay in input order, including duplicates.
2555 pub fn batched_multi_get_pinned<K, I>(
2556 &'_ self,
2557 keys: I,
2558 sorted_input: bool,
2559 ) -> Vec<Result<Option<DBPinnableSlice<'_>>, Error>>
2560 where
2561 K: AsRef<[u8]>,
2562 I: IntoIterator<Item = K>,
2563 {
2564 DEFAULT_READ_OPTS.with(|opts| self.batched_multi_get_pinned_opt(keys, sorted_input, opts))
2565 }
2566
2567 /// Returns pinned values for default-column-family keys in one native batch
2568 /// using the provided read options.
2569 pub fn batched_multi_get_pinned_opt<K, I>(
2570 &'_ self,
2571 keys: I,
2572 sorted_input: bool,
2573 readopts: &ReadOptions,
2574 ) -> Vec<Result<Option<DBPinnableSlice<'_>>, Error>>
2575 where
2576 K: AsRef<[u8]>,
2577 I: IntoIterator<Item = K>,
2578 {
2579 let owned_keys: Vec<K> = keys.into_iter().collect();
2580 self.batched_multi_get_pinned_owned(&owned_keys, sorted_input, readopts)
2581 }
2582
2583 /// Returns pinned values for keys in one column family using one native batch.
2584 ///
2585 /// Set `sorted_input` only when keys are sorted according to the column
2586 /// family's comparator. Results stay in input order, including duplicates.
2587 pub fn batched_multi_get_pinned_cf<K, I>(
2588 &'_ self,
2589 cf: &impl AsColumnFamilyRef,
2590 keys: I,
2591 sorted_input: bool,
2592 ) -> Vec<Result<Option<DBPinnableSlice<'_>>, Error>>
2593 where
2594 K: AsRef<[u8]>,
2595 I: IntoIterator<Item = K>,
2596 {
2597 DEFAULT_READ_OPTS
2598 .with(|opts| self.batched_multi_get_pinned_cf_opt(cf, keys, sorted_input, opts))
2599 }
2600
2601 /// Returns pinned values for keys in one column family using one native batch
2602 /// and the provided read options.
2603 pub fn batched_multi_get_pinned_cf_opt<K, I>(
2604 &'_ self,
2605 cf: &impl AsColumnFamilyRef,
2606 keys: I,
2607 sorted_input: bool,
2608 readopts: &ReadOptions,
2609 ) -> Vec<Result<Option<DBPinnableSlice<'_>>, Error>>
2610 where
2611 K: AsRef<[u8]>,
2612 I: IntoIterator<Item = K>,
2613 {
2614 let owned_keys: Vec<K> = keys.into_iter().collect();
2615 let key_slices = Self::key_slices(&owned_keys);
2616 self.batched_multi_get_pinned_inner(cf.inner(), &key_slices, sorted_input, readopts)
2617 }
2618
2619 /// Returns one owner for all default-column-family pinned results.
2620 ///
2621 /// Values borrow from the returned batch, avoiding one native wrapper
2622 /// allocation and one destroy call per successful key.
2623 pub fn batched_multi_get_pinned_batch<K, I>(
2624 &'_ self,
2625 keys: I,
2626 sorted_input: bool,
2627 ) -> Result<DBPinnableBatch<'_>, Error>
2628 where
2629 K: AsRef<[u8]>,
2630 I: IntoIterator<Item = K>,
2631 {
2632 DEFAULT_READ_OPTS
2633 .with(|opts| self.batched_multi_get_pinned_batch_opt(keys, sorted_input, opts))
2634 }
2635
2636 /// Returns one owner for all default-column-family pinned results using
2637 /// the provided read options.
2638 pub fn batched_multi_get_pinned_batch_opt<K, I>(
2639 &'_ self,
2640 keys: I,
2641 sorted_input: bool,
2642 readopts: &ReadOptions,
2643 ) -> Result<DBPinnableBatch<'_>, Error>
2644 where
2645 K: AsRef<[u8]>,
2646 I: IntoIterator<Item = K>,
2647 {
2648 let owned_keys: Vec<K> = keys.into_iter().collect();
2649 let key_slices = Self::key_slices(&owned_keys);
2650 self.create_pinnable_batch(ptr::null_mut(), &key_slices, sorted_input, readopts)
2651 }
2652
2653 /// Returns one owner for all pinned results from one column family.
2654 pub fn batched_multi_get_pinned_batch_cf<K, I>(
2655 &'_ self,
2656 cf: &impl AsColumnFamilyRef,
2657 keys: I,
2658 sorted_input: bool,
2659 ) -> Result<DBPinnableBatch<'_>, Error>
2660 where
2661 K: AsRef<[u8]>,
2662 I: IntoIterator<Item = K>,
2663 {
2664 DEFAULT_READ_OPTS
2665 .with(|opts| self.batched_multi_get_pinned_batch_cf_opt(cf, keys, sorted_input, opts))
2666 }
2667
2668 /// Returns one owner for all pinned results from one column family using
2669 /// the provided read options.
2670 pub fn batched_multi_get_pinned_batch_cf_opt<K, I>(
2671 &'_ self,
2672 cf: &impl AsColumnFamilyRef,
2673 keys: I,
2674 sorted_input: bool,
2675 readopts: &ReadOptions,
2676 ) -> Result<DBPinnableBatch<'_>, Error>
2677 where
2678 K: AsRef<[u8]>,
2679 I: IntoIterator<Item = K>,
2680 {
2681 let owned_keys: Vec<K> = keys.into_iter().collect();
2682 let key_slices = Self::key_slices(&owned_keys);
2683 self.create_pinnable_batch(cf.inner(), &key_slices, sorted_input, readopts)
2684 }
2685
2686 /// Return the values associated with the given keys and column families.
2687 pub fn multi_get_cf<'a, 'b: 'a, K, I, W>(
2688 &'a self,
2689 keys: I,
2690 ) -> Vec<Result<Option<Vec<u8>>, Error>>
2691 where
2692 K: AsRef<[u8]>,
2693 I: IntoIterator<Item = (&'b W, K)>,
2694 W: 'b + AsColumnFamilyRef,
2695 {
2696 DEFAULT_READ_OPTS.with(|opts| self.multi_get_cf_opt(keys, opts))
2697 }
2698
2699 /// Return the values associated with the given keys and column families using read options.
2700 pub fn multi_get_cf_opt<'a, 'b: 'a, K, I, W>(
2701 &'a self,
2702 keys: I,
2703 readopts: &ReadOptions,
2704 ) -> Vec<Result<Option<Vec<u8>>, Error>>
2705 where
2706 K: AsRef<[u8]>,
2707 I: IntoIterator<Item = (&'b W, K)>,
2708 W: 'b + AsColumnFamilyRef,
2709 {
2710 let cfs_and_owned_keys: Vec<(&'b W, K)> = keys.into_iter().collect();
2711 let (ptr_keys, keys_sizes): (Vec<*const c_char>, Vec<usize>) = cfs_and_owned_keys
2712 .iter()
2713 .map(|(_, k)| {
2714 let key = k.as_ref();
2715 (key.as_ptr() as *const c_char, key.len())
2716 })
2717 .unzip();
2718 let ptr_cfs: Vec<*const ffi::rocksdb_column_family_handle_t> = cfs_and_owned_keys
2719 .iter()
2720 .map(|(c, _)| c.inner().cast_const())
2721 .collect();
2722 let mut values: Vec<*mut c_char> = Vec::with_capacity(ptr_keys.len());
2723 let mut values_sizes: Vec<usize> = Vec::with_capacity(ptr_keys.len());
2724 let mut errors: Vec<*mut c_char> = Vec::with_capacity(ptr_keys.len());
2725 unsafe {
2726 ffi::rocksdb_multi_get_cf(
2727 self.inner.inner(),
2728 readopts.inner,
2729 ptr_cfs.as_ptr(),
2730 ptr_keys.len(),
2731 ptr_keys.as_ptr(),
2732 keys_sizes.as_ptr(),
2733 values.as_mut_ptr(),
2734 values_sizes.as_mut_ptr(),
2735 errors.as_mut_ptr(),
2736 );
2737 }
2738
2739 unsafe {
2740 values.set_len(ptr_keys.len());
2741 values_sizes.set_len(ptr_keys.len());
2742 errors.set_len(ptr_keys.len());
2743 }
2744
2745 convert_values(values, values_sizes, errors)
2746 }
2747
2748 /// Return the values associated with the given keys and the specified column family
2749 /// where internally the read requests are processed in batch if block-based table
2750 /// SST format is used. It is a more optimized version of multi_get_cf.
2751 pub fn batched_multi_get_cf<'a, K, I>(
2752 &'_ self,
2753 cf: &impl AsColumnFamilyRef,
2754 keys: I,
2755 sorted_input: bool,
2756 ) -> Vec<Result<Option<DBPinnableSlice<'_>>, Error>>
2757 where
2758 K: AsRef<[u8]> + 'a + ?Sized,
2759 I: IntoIterator<Item = &'a K>,
2760 {
2761 DEFAULT_READ_OPTS.with(|opts| self.batched_multi_get_cf_opt(cf, keys, sorted_input, opts))
2762 }
2763
2764 /// Return the values associated with the given keys and the specified column family
2765 /// where internally the read requests are processed in batch if block-based table
2766 /// SST format is used. It is a more optimized version of multi_get_cf_opt.
2767 pub fn batched_multi_get_cf_opt<'a, K, I>(
2768 &'_ self,
2769 cf: &impl AsColumnFamilyRef,
2770 keys: I,
2771 sorted_input: bool,
2772 readopts: &ReadOptions,
2773 ) -> Vec<Result<Option<DBPinnableSlice<'_>>, Error>>
2774 where
2775 K: AsRef<[u8]> + 'a + ?Sized,
2776 I: IntoIterator<Item = &'a K>,
2777 {
2778 let key_slices: Vec<_> = keys
2779 .into_iter()
2780 .map(|k| {
2781 let k = k.as_ref();
2782 ffi::rocksdb_slice_t {
2783 data: k.as_ptr() as *const c_char,
2784 size: k.len(),
2785 }
2786 })
2787 .collect();
2788 self.batched_multi_get_pinned_inner(cf.inner(), &key_slices, sorted_input, readopts)
2789 }
2790
2791 /// Return the values associated with the given keys and the specified column family
2792 /// using an optimized slice-based API.
2793 ///
2794 /// This method uses RocksDB's optimized `rocksdb_batched_multi_get_cf_slice` C API,
2795 /// which takes a `rocksdb_slice_t` array directly. This eliminates the internal
2796 /// overhead of converting keys from separate pointer+size arrays to Slice objects.
2797 ///
2798 /// # Arguments
2799 ///
2800 /// * `cf` - The column family to read from
2801 /// * `keys` - An iterator of keys to look up
2802 /// * `sorted_input` - If `true`, indicates the keys are already sorted in ascending
2803 /// order, which allows RocksDB to skip internal sorting and improve performance.
2804 /// **Important**: If you pass `true` but keys are not sorted, results may be incorrect.
2805 ///
2806 /// # Returns
2807 ///
2808 /// A vector of results in the same order as the input keys. Each element is:
2809 /// - `Ok(Some(DBPinnableSlice))` if the key was found
2810 /// - `Ok(None)` if the key was not found
2811 /// - `Err(...)` if an error occurred for that key
2812 ///
2813 /// # Performance
2814 ///
2815 /// This is the fastest batch lookup method when:
2816 /// - You're looking up many keys (10+) from the same column family
2817 /// - You can pre-sort your keys (set `sorted_input = true`)
2818 /// - Block-based table format is used (default)
2819 ///
2820 /// For small numbers of keys, the overhead of batching may not be worth it.
2821 /// Consider using [`get_pinned_cf`](#method.get_pinned_cf) for single key lookups.
2822 ///
2823 /// # Example
2824 ///
2825 /// ```
2826 /// use rust_rocksdb::{DB, Options, ColumnFamilyDescriptor};
2827 ///
2828 /// let tempdir = tempfile::Builder::new().prefix("batch_slice").tempdir().unwrap();
2829 /// let mut opts = Options::default();
2830 /// opts.create_if_missing(true);
2831 /// opts.create_missing_column_families(true);
2832 /// let db = DB::open_cf_descriptors(&opts, tempdir.path(),
2833 /// vec![ColumnFamilyDescriptor::new("cf", Options::default())]).unwrap();
2834 ///
2835 /// let cf = db.cf_handle("cf").unwrap();
2836 /// db.put_cf(&cf, b"k1", b"v1").unwrap();
2837 /// db.put_cf(&cf, b"k2", b"v2").unwrap();
2838 ///
2839 /// // Keys are sorted, so we can set sorted_input = true
2840 /// let keys: Vec<&[u8]> = vec![b"k1", b"k2", b"k3"];
2841 /// let results = db.batched_multi_get_cf_slice(&cf, keys, true);
2842 ///
2843 /// assert!(results[0].as_ref().unwrap().is_some()); // k1 found
2844 /// assert!(results[1].as_ref().unwrap().is_some()); // k2 found
2845 /// assert!(results[2].as_ref().unwrap().is_none()); // k3 not found
2846 /// ```
2847 pub fn batched_multi_get_cf_slice<'a, K, I>(
2848 &'_ self,
2849 cf: &impl AsColumnFamilyRef,
2850 keys: I,
2851 sorted_input: bool,
2852 ) -> Vec<Result<Option<DBPinnableSlice<'_>>, Error>>
2853 where
2854 K: AsRef<[u8]> + 'a + ?Sized,
2855 I: IntoIterator<Item = &'a K>,
2856 {
2857 DEFAULT_READ_OPTS
2858 .with(|opts| self.batched_multi_get_cf_slice_opt(cf, keys, sorted_input, opts))
2859 }
2860
2861 /// Return the values associated with the given keys and the specified column family
2862 /// using an optimized slice-based API with custom read options.
2863 ///
2864 /// This is the same as [`batched_multi_get_cf_slice`](#method.batched_multi_get_cf_slice)
2865 /// but allows specifying custom [`ReadOptions`].
2866 ///
2867 /// See [`batched_multi_get_cf_slice`](#method.batched_multi_get_cf_slice) for full documentation.
2868 pub fn batched_multi_get_cf_slice_opt<'a, K, I>(
2869 &'_ self,
2870 cf: &impl AsColumnFamilyRef,
2871 keys: I,
2872 sorted_input: bool,
2873 readopts: &ReadOptions,
2874 ) -> Vec<Result<Option<DBPinnableSlice<'_>>, Error>>
2875 where
2876 K: AsRef<[u8]> + 'a + ?Sized,
2877 I: IntoIterator<Item = &'a K>,
2878 {
2879 // Convert keys to rocksdb_slice_t array
2880 let slices: Vec<ffi::rocksdb_slice_t> = keys
2881 .into_iter()
2882 .map(|k| {
2883 let k = k.as_ref();
2884 ffi::rocksdb_slice_t {
2885 data: k.as_ptr() as *const c_char,
2886 size: k.len(),
2887 }
2888 })
2889 .collect();
2890
2891 self.batched_multi_get_pinned_inner(cf.inner(), &slices, sorted_input, readopts)
2892 }
2893
2894 fn key_slices<K: AsRef<[u8]>>(keys: &[K]) -> Vec<ffi::rocksdb_slice_t> {
2895 keys.iter()
2896 .map(|key| {
2897 let key = key.as_ref();
2898 ffi::rocksdb_slice_t {
2899 data: key.as_ptr() as *const c_char,
2900 size: key.len(),
2901 }
2902 })
2903 .collect()
2904 }
2905
2906 fn batched_multi_get_pinned_owned<'a, K: AsRef<[u8]>>(
2907 &'a self,
2908 keys: &[K],
2909 sorted_input: bool,
2910 readopts: &ReadOptions,
2911 ) -> Vec<Result<Option<DBPinnableSlice<'a>>, Error>> {
2912 let key_slices = Self::key_slices(keys);
2913 if key_slices.is_empty() {
2914 return Vec::new();
2915 }
2916 let default_cf = OwnedColumnFamilyHandle::default_for(self.inner.inner());
2917 self.batched_multi_get_pinned_inner(default_cf.inner, &key_slices, sorted_input, readopts)
2918 }
2919
2920 fn create_pinnable_batch<'a>(
2921 &'a self,
2922 cf: *mut ffi::rocksdb_column_family_handle_t,
2923 keys: &[ffi::rocksdb_slice_t],
2924 sorted_input: bool,
2925 readopts: &ReadOptions,
2926 ) -> Result<DBPinnableBatch<'a>, Error> {
2927 let batch = unsafe {
2928 ffi_try!(ffi::rust_rocksdb_batched_multi_get_pinned(
2929 self.inner.inner(),
2930 readopts.inner,
2931 cf,
2932 keys.len(),
2933 keys.as_ptr(),
2934 c_uchar::from(sorted_input),
2935 ))
2936 };
2937 if batch.is_null() {
2938 // `ffi_try!` only returns early when the extension set `errptr`.
2939 // A null batch with no error means the extension could not even
2940 // allocate the message, so report it instead of unwrapping.
2941 return Err(Error::new(
2942 "rust_rocksdb_batched_multi_get_pinned returned no batch".to_owned(),
2943 ));
2944 }
2945 // SAFETY: The extension returns a uniquely owned batch.
2946 Ok(unsafe { DBPinnableBatch::from_c(batch) })
2947 }
2948
2949 fn batched_multi_get_pinned_inner<'a>(
2950 &'a self,
2951 cf: *mut ffi::rocksdb_column_family_handle_t,
2952 keys: &[ffi::rocksdb_slice_t],
2953 sorted_input: bool,
2954 readopts: &ReadOptions,
2955 ) -> Vec<Result<Option<DBPinnableSlice<'a>>, Error>> {
2956 if keys.is_empty() {
2957 return Vec::new();
2958 }
2959 let output = match self.execute_batched_multi_get(cf, keys, sorted_input, readopts) {
2960 Ok(output) => output,
2961 Err(error) => {
2962 let message = error.to_string();
2963 return (0..keys.len())
2964 .map(|_| Err(Error::new(message.clone())))
2965 .collect();
2966 }
2967 };
2968 output
2969 .values
2970 .into_iter()
2971 .zip(output.errors)
2972 .map(|(value, error)| unsafe { Self::convert_pinned_result(value, error) })
2973 .collect()
2974 }
2975
2976 fn execute_batched_multi_get(
2977 &self,
2978 cf: *mut ffi::rocksdb_column_family_handle_t,
2979 keys: &[ffi::rocksdb_slice_t],
2980 sorted_input: bool,
2981 readopts: &ReadOptions,
2982 ) -> Result<PinnedMultiGetOutput, Error> {
2983 let mut pinned_values = vec![ptr::null_mut(); keys.len()];
2984 let mut errors = vec![ptr::null_mut(); keys.len()];
2985 unsafe {
2986 ffi_try!(ffi::rust_rocksdb_batched_multi_get_cf_slice_safe(
2987 self.inner.inner(),
2988 readopts.inner,
2989 cf,
2990 keys.len(),
2991 keys.as_ptr(),
2992 pinned_values.as_mut_ptr(),
2993 errors.as_mut_ptr(),
2994 c_uchar::from(sorted_input),
2995 ));
2996 }
2997 Ok(PinnedMultiGetOutput {
2998 values: pinned_values,
2999 errors,
3000 })
3001 }
3002
3003 /// Converts one result returned by `rocksdb_batched_multi_get_cf_slice`.
3004 ///
3005 /// # Safety
3006 ///
3007 /// `value` must be null or an owned pinnable slice. `error` must be null or
3008 /// an owned RocksDB error string.
3009 unsafe fn convert_pinned_result<'a>(
3010 value: *mut ffi::rocksdb_pinnableslice_t,
3011 error: *mut c_char,
3012 ) -> Result<Option<DBPinnableSlice<'a>>, Error> {
3013 if error.is_null() {
3014 return Ok((!value.is_null()).then(|| unsafe { DBPinnableSlice::from_c(value) }));
3015 }
3016 if !value.is_null() {
3017 unsafe {
3018 ffi::rocksdb_pinnableslice_destroy(value);
3019 }
3020 }
3021 Err(convert_rocksdb_error(error))
3022 }
3023
3024 /// Returns `false` if the given key definitely doesn't exist in the database, otherwise returns
3025 /// `true`. This function uses default `ReadOptions`.
3026 pub fn key_may_exist<K: AsRef<[u8]>>(&self, key: K) -> bool {
3027 DEFAULT_READ_OPTS.with(|opts| self.key_may_exist_opt(key, opts))
3028 }
3029
3030 /// Returns `false` if the given key definitely doesn't exist in the database, otherwise returns
3031 /// `true`.
3032 pub fn key_may_exist_opt<K: AsRef<[u8]>>(&self, key: K, readopts: &ReadOptions) -> bool {
3033 let key = key.as_ref();
3034 unsafe {
3035 0 != ffi::rocksdb_key_may_exist(
3036 self.inner.inner(),
3037 readopts.inner,
3038 key.as_ptr() as *const c_char,
3039 key.len() as size_t,
3040 ptr::null_mut(), /*value*/
3041 ptr::null_mut(), /*val_len*/
3042 ptr::null(), /*timestamp*/
3043 0, /*timestamp_len*/
3044 ptr::null_mut(), /*value_found*/
3045 )
3046 }
3047 }
3048
3049 /// Returns `false` if the given key definitely doesn't exist in the specified column family,
3050 /// otherwise returns `true`. This function uses default `ReadOptions`.
3051 pub fn key_may_exist_cf<K: AsRef<[u8]>>(&self, cf: &impl AsColumnFamilyRef, key: K) -> bool {
3052 DEFAULT_READ_OPTS.with(|opts| self.key_may_exist_cf_opt(cf, key, opts))
3053 }
3054
3055 /// Returns `false` if the given key definitely doesn't exist in the specified column family,
3056 /// otherwise returns `true`.
3057 pub fn key_may_exist_cf_opt<K: AsRef<[u8]>>(
3058 &self,
3059 cf: &impl AsColumnFamilyRef,
3060 key: K,
3061 readopts: &ReadOptions,
3062 ) -> bool {
3063 let key = key.as_ref();
3064 0 != unsafe {
3065 ffi::rocksdb_key_may_exist_cf(
3066 self.inner.inner(),
3067 readopts.inner,
3068 cf.inner(),
3069 key.as_ptr() as *const c_char,
3070 key.len() as size_t,
3071 ptr::null_mut(), /*value*/
3072 ptr::null_mut(), /*val_len*/
3073 ptr::null(), /*timestamp*/
3074 0, /*timestamp_len*/
3075 ptr::null_mut(), /*value_found*/
3076 )
3077 }
3078 }
3079
3080 /// If the key definitely does not exist in the database, then this method
3081 /// returns `(false, None)`, else `(true, None)` if it may.
3082 /// If the key is found in memory, then it returns `(true, Some<CSlice>)`.
3083 ///
3084 /// This check is potentially lighter-weight than calling `get()`. One way
3085 /// to make this lighter weight is to avoid doing any IOs.
3086 pub fn key_may_exist_cf_opt_value<K: AsRef<[u8]>>(
3087 &self,
3088 cf: &impl AsColumnFamilyRef,
3089 key: K,
3090 readopts: &ReadOptions,
3091 ) -> (bool, Option<CSlice>) {
3092 let key = key.as_ref();
3093 let mut val: *mut c_char = ptr::null_mut();
3094 let mut val_len: usize = 0;
3095 let mut value_found: c_uchar = 0;
3096 let may_exists = 0
3097 != unsafe {
3098 ffi::rocksdb_key_may_exist_cf(
3099 self.inner.inner(),
3100 readopts.inner,
3101 cf.inner(),
3102 key.as_ptr() as *const c_char,
3103 key.len() as size_t,
3104 &raw mut val, /*value*/
3105 &raw mut val_len, /*val_len*/
3106 ptr::null(), /*timestamp*/
3107 0, /*timestamp_len*/
3108 &raw mut value_found, /*value_found*/
3109 )
3110 };
3111 // The value is only allocated (using malloc) and returned if it is found and
3112 // value_found isn't NULL. In that case the user is responsible for freeing it.
3113 if may_exists && value_found != 0 {
3114 (
3115 may_exists,
3116 Some(unsafe { CSlice::from_raw_parts(val, val_len) }),
3117 )
3118 } else {
3119 (may_exists, None)
3120 }
3121 }
3122
3123 fn create_inner_cf_handle(
3124 &self,
3125 name: impl CStrLike,
3126 opts: &Options,
3127 ) -> Result<*mut ffi::rocksdb_column_family_handle_t, Error> {
3128 let cf_name = name.bake().map_err(|err| {
3129 Error::new(format!(
3130 "Failed to convert path to CString when creating cf: {err}"
3131 ))
3132 })?;
3133
3134 // Can't use ffi_try: rocksdb_create_column_family has a bug where it allocates a
3135 // result that needs to be freed on error
3136 let mut err: *mut ::libc::c_char = ::std::ptr::null_mut();
3137 let cf_handle = unsafe {
3138 ffi::rocksdb_create_column_family(
3139 self.inner.inner(),
3140 opts.inner,
3141 cf_name.as_ptr(),
3142 &raw mut err,
3143 )
3144 };
3145 if !err.is_null() {
3146 if !cf_handle.is_null() {
3147 unsafe { ffi::rocksdb_column_family_handle_destroy(cf_handle) };
3148 }
3149 return Err(convert_rocksdb_error(err));
3150 }
3151 Ok(cf_handle)
3152 }
3153
3154 /// Creates every named column family in one call.
3155 ///
3156 /// This is not atomic. RocksDB creates the families one at a time and stops at
3157 /// the first failure, so the ones before it are already committed and stay that
3158 /// way. Only the options file write at the end is shared, which is the whole
3159 /// saving over calling [`create_cf`](Self::create_cf) in a loop.
3160 ///
3161 /// Returns the handles that were created, in the order the names were given,
3162 /// alongside the error that stopped the rest. The caller owns those handles and
3163 /// has to record them in its column family map even when there is an error,
3164 /// because the families exist either way.
3165 fn create_inner_cf_handles(
3166 &self,
3167 names: &[(String, CString)],
3168 opts: &Options,
3169 ) -> CreatedCfHandles {
3170 let name_ptrs: Vec<*const c_char> = names.iter().map(|(_, name)| name.as_ptr()).collect();
3171 let Ok(count) = c_int::try_from(names.len()) else {
3172 return CreatedCfHandles::failed(Error::new(format!(
3173 "Too many column families to create at once: {}",
3174 names.len()
3175 )));
3176 };
3177
3178 let mut len: usize = 0;
3179 let mut err: *mut ::libc::c_char = ::std::ptr::null_mut();
3180 // Can't use ffi_try: like rocksdb_create_column_family, this allocates a result
3181 // that needs to be freed on error.
3182 let list = unsafe {
3183 ffi::rocksdb_create_column_families(
3184 self.inner.inner(),
3185 opts.inner,
3186 count,
3187 name_ptrs.as_ptr(),
3188 &raw mut len,
3189 &raw mut err,
3190 )
3191 };
3192
3193 // Two allocations come back: the array, and a handle per family. Freeing the
3194 // array does not touch the handles, so take copies of them and release the
3195 // array on its own.
3196 let handles = if list.is_null() {
3197 Vec::new()
3198 } else {
3199 let handles = unsafe { std::slice::from_raw_parts(list, len) }.to_vec();
3200 unsafe { ffi::rocksdb_create_column_families_destroy(list) };
3201 handles
3202 };
3203
3204 if !err.is_null() {
3205 // Hand the handles back even though this failed. The families they name
3206 // were committed before the failing one and are still there, so the caller
3207 // needs them to be able to use or drop those families.
3208 return CreatedCfHandles {
3209 handles,
3210 error: Some(convert_rocksdb_error(err)),
3211 };
3212 }
3213
3214 let created = handles.len();
3215 let error = (created != names.len()).then(|| {
3216 Error::new(format!(
3217 "Expected {} column family handles, got {created}",
3218 names.len(),
3219 ))
3220 });
3221
3222 CreatedCfHandles { handles, error }
3223 }
3224
3225 /// Creates one column family whose entries expire after `ttl`.
3226 ///
3227 /// Only valid on a DB opened with a TTL. The C function casts the handle to
3228 /// `DBWithTTL` unchecked, so the caller has to have proven that already.
3229 fn create_inner_cf_handle_with_ttl(
3230 &self,
3231 name: impl CStrLike,
3232 opts: &Options,
3233 ttl: ColumnFamilyTtl,
3234 ) -> Result<*mut ffi::rocksdb_column_family_handle_t, Error> {
3235 let Some(db_ttl) = self.opened_with_ttl else {
3236 return Err(Error::new(
3237 "create_cf_with_ttl requires a database opened with DB::open_with_ttl \
3238 or one of the open_cf*_with_ttl functions"
3239 .to_owned(),
3240 ));
3241 };
3242
3243 let cf_name = name.bake().map_err(|err| {
3244 Error::new(format!(
3245 "Failed to convert name to CString when creating cf with ttl: {err}"
3246 ))
3247 })?;
3248 let ttl = cf_ttl_to_seconds(ttl, db_ttl);
3249
3250 // Can't use ffi_try: like rocksdb_create_column_family, this allocates a result
3251 // that needs to be freed on error.
3252 let mut err: *mut ::libc::c_char = ::std::ptr::null_mut();
3253 let cf_handle = unsafe {
3254 ffi::rocksdb_create_column_family_with_ttl(
3255 self.inner.inner(),
3256 opts.inner,
3257 cf_name.as_ptr(),
3258 ttl,
3259 &raw mut err,
3260 )
3261 };
3262 if !err.is_null() {
3263 if !cf_handle.is_null() {
3264 unsafe { ffi::rocksdb_column_family_handle_destroy(cf_handle) };
3265 }
3266 return Err(convert_rocksdb_error(err));
3267 }
3268 Ok(cf_handle)
3269 }
3270
3271 pub fn iterator<'a: 'b, 'b>(
3272 &'a self,
3273 mode: IteratorMode,
3274 ) -> DBIteratorWithThreadMode<'b, Self> {
3275 let readopts = ReadOptions::default();
3276 self.iterator_opt(mode, readopts)
3277 }
3278
3279 pub fn iterator_opt<'a: 'b, 'b>(
3280 &'a self,
3281 mode: IteratorMode,
3282 readopts: ReadOptions,
3283 ) -> DBIteratorWithThreadMode<'b, Self> {
3284 DBIteratorWithThreadMode::new(self, readopts, mode)
3285 }
3286
3287 /// Opens an iterator using the provided ReadOptions.
3288 /// This is used when you want to iterate over a specific ColumnFamily with a modified ReadOptions
3289 pub fn iterator_cf_opt<'a: 'b, 'b>(
3290 &'a self,
3291 cf_handle: &impl AsColumnFamilyRef,
3292 readopts: ReadOptions,
3293 mode: IteratorMode,
3294 ) -> DBIteratorWithThreadMode<'b, Self> {
3295 DBIteratorWithThreadMode::new_cf(self, cf_handle.inner(), readopts, mode)
3296 }
3297
3298 /// Opens an iterator with `set_total_order_seek` enabled.
3299 /// This must be used to iterate across prefixes when `set_memtable_factory` has been called
3300 /// with a Hash-based implementation.
3301 pub fn full_iterator<'a: 'b, 'b>(
3302 &'a self,
3303 mode: IteratorMode,
3304 ) -> DBIteratorWithThreadMode<'b, Self> {
3305 let mut opts = ReadOptions::default();
3306 opts.set_total_order_seek(true);
3307 DBIteratorWithThreadMode::new(self, opts, mode)
3308 }
3309
3310 pub fn prefix_iterator<'a: 'b, 'b, P: AsRef<[u8]>>(
3311 &'a self,
3312 prefix: P,
3313 ) -> DBIteratorWithThreadMode<'b, Self> {
3314 let mut opts = ReadOptions::default();
3315 opts.set_prefix_same_as_start(true);
3316 DBIteratorWithThreadMode::new(
3317 self,
3318 opts,
3319 IteratorMode::From(prefix.as_ref(), Direction::Forward),
3320 )
3321 }
3322
3323 pub fn iterator_cf<'a: 'b, 'b>(
3324 &'a self,
3325 cf_handle: &impl AsColumnFamilyRef,
3326 mode: IteratorMode,
3327 ) -> DBIteratorWithThreadMode<'b, Self> {
3328 let opts = ReadOptions::default();
3329 DBIteratorWithThreadMode::new_cf(self, cf_handle.inner(), opts, mode)
3330 }
3331
3332 pub fn full_iterator_cf<'a: 'b, 'b>(
3333 &'a self,
3334 cf_handle: &impl AsColumnFamilyRef,
3335 mode: IteratorMode,
3336 ) -> DBIteratorWithThreadMode<'b, Self> {
3337 let mut opts = ReadOptions::default();
3338 opts.set_total_order_seek(true);
3339 DBIteratorWithThreadMode::new_cf(self, cf_handle.inner(), opts, mode)
3340 }
3341
3342 pub fn prefix_iterator_cf<'a, P: AsRef<[u8]>>(
3343 &'a self,
3344 cf_handle: &impl AsColumnFamilyRef,
3345 prefix: P,
3346 ) -> DBIteratorWithThreadMode<'a, Self> {
3347 let mut opts = ReadOptions::default();
3348 opts.set_prefix_same_as_start(true);
3349 DBIteratorWithThreadMode::<'a, Self>::new_cf(
3350 self,
3351 cf_handle.inner(),
3352 opts,
3353 IteratorMode::From(prefix.as_ref(), Direction::Forward),
3354 )
3355 }
3356
3357 /// Returns `true` if there exists at least one key with the given prefix
3358 /// in the default column family using default read options.
3359 ///
3360 /// When to use: prefer this for one-shot checks. It enables
3361 /// `prefix_same_as_start(true)` and bounds the iterator to the
3362 /// prefix via `PrefixRange`, minimizing stray IO per call.
3363 pub fn prefix_exists<P: AsRef<[u8]>>(&self, prefix: P) -> Result<bool, Error> {
3364 let p = prefix.as_ref();
3365 with_prefix_read_opts(p, |opts| self.prefix_exists_opt(p, opts))
3366 }
3367
3368 /// Returns `true` if there exists at least one key with the given prefix
3369 /// in the default column family using the provided read options.
3370 pub fn prefix_exists_opt<P: AsRef<[u8]>>(
3371 &self,
3372 prefix: P,
3373 readopts: &ReadOptions,
3374 ) -> Result<bool, Error> {
3375 let prefix = prefix.as_ref();
3376 let iter = unsafe { self.create_iterator(readopts) };
3377 let res = unsafe {
3378 ffi::rocksdb_iter_seek(
3379 iter,
3380 prefix.as_ptr() as *const c_char,
3381 prefix.len() as size_t,
3382 );
3383 if ffi::rocksdb_iter_valid(iter) != 0 {
3384 let mut key_len: size_t = 0;
3385 let key_ptr = ffi::rocksdb_iter_key(iter, &raw mut key_len);
3386 // An empty key is legal, and `from_raw_parts` wants a
3387 // dereferenceable pointer even at length 0.
3388 let key = if key_len == 0 {
3389 &[][..]
3390 } else {
3391 slice::from_raw_parts(key_ptr.cast::<u8>(), key_len as usize)
3392 };
3393 Ok(key.starts_with(prefix))
3394 } else if let Err(e) = (|| {
3395 // Check status to differentiate end-of-range vs error
3396 ffi_try!(ffi::rocksdb_iter_get_error(iter));
3397 Ok::<(), Error>(())
3398 })() {
3399 Err(e)
3400 } else {
3401 Ok(false)
3402 }
3403 };
3404 unsafe { ffi::rocksdb_iter_destroy(iter) };
3405 res
3406 }
3407
3408 /// Creates a reusable prefix prober over the default column family using
3409 /// read options optimized for prefix probes.
3410 ///
3411 /// When to use: prefer this in hot loops with many checks per second. It
3412 /// reuses a raw iterator to avoid per-call allocation/FFI overhead. If you
3413 /// need custom tuning (e.g. async IO, readahead, cache-only), use
3414 /// `prefix_prober_with_opts`.
3415 pub fn prefix_prober(&self) -> PrefixProber<'_, Self> {
3416 PrefixProber {
3417 raw: DBRawIteratorWithThreadMode::new(self, prefix_probe_read_opts()),
3418 }
3419 }
3420
3421 /// Creates a reusable prefix prober over the default column family using
3422 /// the provided read options (owned).
3423 ///
3424 /// When to use: advanced tuning for heavy workloads. Callers can set
3425 /// `set_async_io(true)`, `set_readahead_size`, `set_read_tier`, etc. Note:
3426 /// the prober owns `ReadOptions` to keep internal buffers alive.
3427 pub fn prefix_prober_with_opts(&self, readopts: ReadOptions) -> PrefixProber<'_, Self> {
3428 PrefixProber {
3429 raw: DBRawIteratorWithThreadMode::new(self, readopts),
3430 }
3431 }
3432
3433 /// Creates a reusable prefix prober over the specified column family using
3434 /// read options optimized for prefix probes.
3435 pub fn prefix_prober_cf(&self, cf_handle: &impl AsColumnFamilyRef) -> PrefixProber<'_, Self> {
3436 PrefixProber {
3437 raw: DBRawIteratorWithThreadMode::new_cf(
3438 self,
3439 cf_handle.inner(),
3440 prefix_probe_read_opts(),
3441 ),
3442 }
3443 }
3444
3445 /// Creates a reusable prefix prober over the specified column family using
3446 /// the provided read options (owned).
3447 ///
3448 /// When to use: advanced tuning for heavy workloads on a specific CF.
3449 pub fn prefix_prober_cf_with_opts(
3450 &self,
3451 cf_handle: &impl AsColumnFamilyRef,
3452 readopts: ReadOptions,
3453 ) -> PrefixProber<'_, Self> {
3454 PrefixProber {
3455 raw: DBRawIteratorWithThreadMode::new_cf(self, cf_handle.inner(), readopts),
3456 }
3457 }
3458
3459 /// Returns `true` if there exists at least one key with the given prefix
3460 /// in the specified column family using default read options.
3461 ///
3462 /// When to use: one-shot checks on a CF. Enables
3463 /// `prefix_same_as_start(true)` and bounds the iterator via `PrefixRange`.
3464 pub fn prefix_exists_cf<P: AsRef<[u8]>>(
3465 &self,
3466 cf_handle: &impl AsColumnFamilyRef,
3467 prefix: P,
3468 ) -> Result<bool, Error> {
3469 let p = prefix.as_ref();
3470 with_prefix_read_opts(p, |opts| self.prefix_exists_cf_opt(cf_handle, p, opts))
3471 }
3472
3473 /// Returns `true` if there exists at least one key with the given prefix
3474 /// in the specified column family using the provided read options.
3475 pub fn prefix_exists_cf_opt<P: AsRef<[u8]>>(
3476 &self,
3477 cf_handle: &impl AsColumnFamilyRef,
3478 prefix: P,
3479 readopts: &ReadOptions,
3480 ) -> Result<bool, Error> {
3481 let prefix = prefix.as_ref();
3482 let iter = unsafe { self.create_iterator_cf(cf_handle.inner(), readopts) };
3483 let res = unsafe {
3484 ffi::rocksdb_iter_seek(
3485 iter,
3486 prefix.as_ptr() as *const c_char,
3487 prefix.len() as size_t,
3488 );
3489 if ffi::rocksdb_iter_valid(iter) != 0 {
3490 let mut key_len: size_t = 0;
3491 let key_ptr = ffi::rocksdb_iter_key(iter, &raw mut key_len);
3492 // An empty key is legal, and `from_raw_parts` wants a
3493 // dereferenceable pointer even at length 0.
3494 let key = if key_len == 0 {
3495 &[][..]
3496 } else {
3497 slice::from_raw_parts(key_ptr.cast::<u8>(), key_len as usize)
3498 };
3499 Ok(key.starts_with(prefix))
3500 } else if let Err(e) = (|| {
3501 ffi_try!(ffi::rocksdb_iter_get_error(iter));
3502 Ok::<(), Error>(())
3503 })() {
3504 Err(e)
3505 } else {
3506 Ok(false)
3507 }
3508 };
3509 unsafe { ffi::rocksdb_iter_destroy(iter) };
3510 res
3511 }
3512
3513 /// Opens a raw iterator over the database, using the default read options
3514 pub fn raw_iterator<'a: 'b, 'b>(&'a self) -> DBRawIteratorWithThreadMode<'b, Self> {
3515 let opts = ReadOptions::default();
3516 DBRawIteratorWithThreadMode::new(self, opts)
3517 }
3518
3519 /// Opens a raw iterator over the given column family, using the default read options
3520 pub fn raw_iterator_cf<'a: 'b, 'b>(
3521 &'a self,
3522 cf_handle: &impl AsColumnFamilyRef,
3523 ) -> DBRawIteratorWithThreadMode<'b, Self> {
3524 let opts = ReadOptions::default();
3525 DBRawIteratorWithThreadMode::new_cf(self, cf_handle.inner(), opts)
3526 }
3527
3528 /// Opens raw iterators for multiple column families from one consistent
3529 /// RocksDB state.
3530 ///
3531 /// The returned iterators match the input column family order and own their
3532 /// native handles. They share one `ReadOptions`, because one native
3533 /// `rocksdb_create_iterators` call applies a single options object to every
3534 /// iterator it creates.
3535 pub fn raw_iterators_cf<'a, 'b, W, I>(
3536 &'a self,
3537 column_families: I,
3538 ) -> Result<Vec<DBRawIteratorWithThreadMode<'a, Self>>, Error>
3539 where
3540 W: AsColumnFamilyRef + 'b,
3541 I: IntoIterator<Item = &'b W>,
3542 {
3543 let mut cf_handles: Vec<_> = column_families
3544 .into_iter()
3545 .map(AsColumnFamilyRef::inner)
3546 .collect();
3547 if cf_handles.is_empty() {
3548 return Ok(Vec::new());
3549 }
3550 let created = self.create_iterators_cf(&mut cf_handles)?;
3551 Ok(created
3552 .handles
3553 .into_iter()
3554 .map(|handle| {
3555 DBRawIteratorWithThreadMode::from_inner(handle, Arc::clone(&created.readopts))
3556 })
3557 .collect())
3558 }
3559
3560 fn create_iterators_cf(
3561 &self,
3562 cf_handles: &mut [*mut ffi::rocksdb_column_family_handle_t],
3563 ) -> Result<CreatedIterators, Error> {
3564 let mut iterator_handles = vec![ptr::null_mut(); cf_handles.len()];
3565 // Every iterator gets a handle on this. RocksDB's `DBIter` stores raw
3566 // `Slice*` into the options for iterate_lower_bound, iterate_upper_bound
3567 // and the read timestamps, and `ArenaWrappedDBIter::Refresh` re-reads
3568 // them, so the options have to outlive the last iterator rather than
3569 // this function. See issue #660.
3570 let readopts = Arc::new(ReadOptions::default());
3571 unsafe {
3572 ffi_try!(ffi::rust_rocksdb_create_iterators_safe(
3573 self.inner.inner(),
3574 readopts.inner,
3575 cf_handles.as_mut_ptr(),
3576 iterator_handles.as_mut_ptr(),
3577 iterator_handles.len(),
3578 ));
3579 }
3580 Self::validate_created_iterators(&iterator_handles)?;
3581 Ok(CreatedIterators {
3582 readopts,
3583 handles: iterator_handles,
3584 })
3585 }
3586
3587 fn validate_created_iterators(
3588 iterator_handles: &[*mut ffi::rocksdb_iterator_t],
3589 ) -> Result<(), Error> {
3590 if iterator_handles.iter().any(|iterator| iterator.is_null()) {
3591 unsafe {
3592 Self::destroy_iterators(iterator_handles);
3593 }
3594 return Err(Error::new(
3595 "rocksdb_create_iterators returned a null iterator".to_owned(),
3596 ));
3597 }
3598 Ok(())
3599 }
3600
3601 /// Destroys non-null iterator handles owned by the caller.
3602 ///
3603 /// # Safety
3604 ///
3605 /// Every non-null pointer must identify a live, uniquely owned RocksDB iterator.
3606 unsafe fn destroy_iterators(iterators: &[*mut ffi::rocksdb_iterator_t]) {
3607 for &iterator in iterators {
3608 if !iterator.is_null() {
3609 unsafe {
3610 ffi::rocksdb_iter_destroy(iterator);
3611 }
3612 }
3613 }
3614 }
3615
3616 /// Opens a raw iterator over the database, using the given read options
3617 pub fn raw_iterator_opt<'a: 'b, 'b>(
3618 &'a self,
3619 readopts: ReadOptions,
3620 ) -> DBRawIteratorWithThreadMode<'b, Self> {
3621 DBRawIteratorWithThreadMode::new(self, readopts)
3622 }
3623
3624 /// Opens a raw iterator over the given column family, using the given read options
3625 pub fn raw_iterator_cf_opt<'a: 'b, 'b>(
3626 &'a self,
3627 cf_handle: &impl AsColumnFamilyRef,
3628 readopts: ReadOptions,
3629 ) -> DBRawIteratorWithThreadMode<'b, Self> {
3630 DBRawIteratorWithThreadMode::new_cf(self, cf_handle.inner(), readopts)
3631 }
3632
3633 pub fn snapshot(&'_ self) -> SnapshotWithThreadMode<'_, Self> {
3634 SnapshotWithThreadMode::<Self>::new(self)
3635 }
3636
3637 pub fn put_opt<K, V>(&self, key: K, value: V, writeopts: &WriteOptions) -> Result<(), Error>
3638 where
3639 K: AsRef<[u8]>,
3640 V: AsRef<[u8]>,
3641 {
3642 let key = key.as_ref();
3643 let value = value.as_ref();
3644
3645 unsafe {
3646 ffi_try!(ffi::rocksdb_put(
3647 self.inner.inner(),
3648 writeopts.inner,
3649 key.as_ptr() as *const c_char,
3650 key.len() as size_t,
3651 value.as_ptr() as *const c_char,
3652 value.len() as size_t,
3653 ));
3654 Ok(())
3655 }
3656 }
3657
3658 pub fn put_cf_opt<K, V>(
3659 &self,
3660 cf: &impl AsColumnFamilyRef,
3661 key: K,
3662 value: V,
3663 writeopts: &WriteOptions,
3664 ) -> Result<(), Error>
3665 where
3666 K: AsRef<[u8]>,
3667 V: AsRef<[u8]>,
3668 {
3669 let key = key.as_ref();
3670 let value = value.as_ref();
3671
3672 unsafe {
3673 ffi_try!(ffi::rocksdb_put_cf(
3674 self.inner.inner(),
3675 writeopts.inner,
3676 cf.inner(),
3677 key.as_ptr() as *const c_char,
3678 key.len() as size_t,
3679 value.as_ptr() as *const c_char,
3680 value.len() as size_t,
3681 ));
3682 Ok(())
3683 }
3684 }
3685
3686 /// Set the database entry for "key" to "value" with WriteOptions.
3687 /// If "key" already exists, it will coexist with previous entry.
3688 /// `Get` with a timestamp ts specified in ReadOptions will return
3689 /// the most recent key/value whose timestamp is smaller than or equal to ts.
3690 /// Takes an additional argument `ts` as the timestamp.
3691 /// Note: the DB must be opened with user defined timestamp enabled.
3692 pub fn put_with_ts_opt<K, V, S>(
3693 &self,
3694 key: K,
3695 ts: S,
3696 value: V,
3697 writeopts: &WriteOptions,
3698 ) -> Result<(), Error>
3699 where
3700 K: AsRef<[u8]>,
3701 V: AsRef<[u8]>,
3702 S: AsRef<[u8]>,
3703 {
3704 let key = key.as_ref();
3705 let value = value.as_ref();
3706 let ts = ts.as_ref();
3707 unsafe {
3708 ffi_try!(ffi::rocksdb_put_with_ts(
3709 self.inner.inner(),
3710 writeopts.inner,
3711 key.as_ptr() as *const c_char,
3712 key.len() as size_t,
3713 ts.as_ptr() as *const c_char,
3714 ts.len() as size_t,
3715 value.as_ptr() as *const c_char,
3716 value.len() as size_t,
3717 ));
3718 Ok(())
3719 }
3720 }
3721
3722 /// Put with timestamp in a specific column family with WriteOptions.
3723 /// If "key" already exists, it will coexist with previous entry.
3724 /// `Get` with a timestamp ts specified in ReadOptions will return
3725 /// the most recent key/value whose timestamp is smaller than or equal to ts.
3726 /// Takes an additional argument `ts` as the timestamp.
3727 /// Note: the DB must be opened with user defined timestamp enabled.
3728 pub fn put_cf_with_ts_opt<K, V, S>(
3729 &self,
3730 cf: &impl AsColumnFamilyRef,
3731 key: K,
3732 ts: S,
3733 value: V,
3734 writeopts: &WriteOptions,
3735 ) -> Result<(), Error>
3736 where
3737 K: AsRef<[u8]>,
3738 V: AsRef<[u8]>,
3739 S: AsRef<[u8]>,
3740 {
3741 let key = key.as_ref();
3742 let value = value.as_ref();
3743 let ts = ts.as_ref();
3744 unsafe {
3745 ffi_try!(ffi::rocksdb_put_cf_with_ts(
3746 self.inner.inner(),
3747 writeopts.inner,
3748 cf.inner(),
3749 key.as_ptr() as *const c_char,
3750 key.len() as size_t,
3751 ts.as_ptr() as *const c_char,
3752 ts.len() as size_t,
3753 value.as_ptr() as *const c_char,
3754 value.len() as size_t,
3755 ));
3756 Ok(())
3757 }
3758 }
3759
3760 pub fn merge_opt<K, V>(&self, key: K, value: V, writeopts: &WriteOptions) -> Result<(), Error>
3761 where
3762 K: AsRef<[u8]>,
3763 V: AsRef<[u8]>,
3764 {
3765 let key = key.as_ref();
3766 let value = value.as_ref();
3767
3768 unsafe {
3769 ffi_try!(ffi::rocksdb_merge(
3770 self.inner.inner(),
3771 writeopts.inner,
3772 key.as_ptr() as *const c_char,
3773 key.len() as size_t,
3774 value.as_ptr() as *const c_char,
3775 value.len() as size_t,
3776 ));
3777 Ok(())
3778 }
3779 }
3780
3781 pub fn merge_cf_opt<K, V>(
3782 &self,
3783 cf: &impl AsColumnFamilyRef,
3784 key: K,
3785 value: V,
3786 writeopts: &WriteOptions,
3787 ) -> Result<(), Error>
3788 where
3789 K: AsRef<[u8]>,
3790 V: AsRef<[u8]>,
3791 {
3792 let key = key.as_ref();
3793 let value = value.as_ref();
3794
3795 unsafe {
3796 ffi_try!(ffi::rocksdb_merge_cf(
3797 self.inner.inner(),
3798 writeopts.inner,
3799 cf.inner(),
3800 key.as_ptr() as *const c_char,
3801 key.len() as size_t,
3802 value.as_ptr() as *const c_char,
3803 value.len() as size_t,
3804 ));
3805 Ok(())
3806 }
3807 }
3808
3809 pub fn delete_opt<K: AsRef<[u8]>>(
3810 &self,
3811 key: K,
3812 writeopts: &WriteOptions,
3813 ) -> Result<(), Error> {
3814 let key = key.as_ref();
3815
3816 unsafe {
3817 ffi_try!(ffi::rocksdb_delete(
3818 self.inner.inner(),
3819 writeopts.inner,
3820 key.as_ptr() as *const c_char,
3821 key.len() as size_t,
3822 ));
3823 Ok(())
3824 }
3825 }
3826
3827 pub fn delete_cf_opt<K: AsRef<[u8]>>(
3828 &self,
3829 cf: &impl AsColumnFamilyRef,
3830 key: K,
3831 writeopts: &WriteOptions,
3832 ) -> Result<(), Error> {
3833 let key = key.as_ref();
3834
3835 unsafe {
3836 ffi_try!(ffi::rocksdb_delete_cf(
3837 self.inner.inner(),
3838 writeopts.inner,
3839 cf.inner(),
3840 key.as_ptr() as *const c_char,
3841 key.len() as size_t,
3842 ));
3843 Ok(())
3844 }
3845 }
3846
3847 /// Remove the database entry (if any) for "key" with WriteOptions.
3848 /// Takes an additional argument `ts` as the timestamp.
3849 /// Note: the DB must be opened with user defined timestamp enabled.
3850 pub fn delete_with_ts_opt<K, S>(
3851 &self,
3852 key: K,
3853 ts: S,
3854 writeopts: &WriteOptions,
3855 ) -> Result<(), Error>
3856 where
3857 K: AsRef<[u8]>,
3858 S: AsRef<[u8]>,
3859 {
3860 let key = key.as_ref();
3861 let ts = ts.as_ref();
3862 unsafe {
3863 ffi_try!(ffi::rocksdb_delete_with_ts(
3864 self.inner.inner(),
3865 writeopts.inner,
3866 key.as_ptr() as *const c_char,
3867 key.len() as size_t,
3868 ts.as_ptr() as *const c_char,
3869 ts.len() as size_t,
3870 ));
3871 Ok(())
3872 }
3873 }
3874
3875 /// Delete with timestamp in a specific column family with WriteOptions.
3876 /// Takes an additional argument `ts` as the timestamp.
3877 /// Note: the DB must be opened with user defined timestamp enabled.
3878 pub fn delete_cf_with_ts_opt<K, S>(
3879 &self,
3880 cf: &impl AsColumnFamilyRef,
3881 key: K,
3882 ts: S,
3883 writeopts: &WriteOptions,
3884 ) -> Result<(), Error>
3885 where
3886 K: AsRef<[u8]>,
3887 S: AsRef<[u8]>,
3888 {
3889 let key = key.as_ref();
3890 let ts = ts.as_ref();
3891 unsafe {
3892 ffi_try!(ffi::rocksdb_delete_cf_with_ts(
3893 self.inner.inner(),
3894 writeopts.inner,
3895 cf.inner(),
3896 key.as_ptr() as *const c_char,
3897 key.len() as size_t,
3898 ts.as_ptr() as *const c_char,
3899 ts.len() as size_t,
3900 ));
3901 Ok(())
3902 }
3903 }
3904
3905 pub fn put<K, V>(&self, key: K, value: V) -> Result<(), Error>
3906 where
3907 K: AsRef<[u8]>,
3908 V: AsRef<[u8]>,
3909 {
3910 DEFAULT_WRITE_OPTS.with(|opts| self.put_opt(key, value, opts))
3911 }
3912
3913 pub fn put_cf<K, V>(&self, cf: &impl AsColumnFamilyRef, key: K, value: V) -> Result<(), Error>
3914 where
3915 K: AsRef<[u8]>,
3916 V: AsRef<[u8]>,
3917 {
3918 DEFAULT_WRITE_OPTS.with(|opts| self.put_cf_opt(cf, key, value, opts))
3919 }
3920
3921 /// Set the database entry for "key" to "value".
3922 /// If "key" already exists, it will coexist with previous entry.
3923 /// `Get` with a timestamp ts specified in ReadOptions will return
3924 /// the most recent key/value whose timestamp is smaller than or equal to ts.
3925 /// Takes an additional argument `ts` as the timestamp.
3926 /// Note: the DB must be opened with user defined timestamp enabled.
3927 pub fn put_with_ts<K, V, S>(&self, key: K, ts: S, value: V) -> Result<(), Error>
3928 where
3929 K: AsRef<[u8]>,
3930 V: AsRef<[u8]>,
3931 S: AsRef<[u8]>,
3932 {
3933 DEFAULT_WRITE_OPTS
3934 .with(|opts| self.put_with_ts_opt(key.as_ref(), ts.as_ref(), value.as_ref(), opts))
3935 }
3936
3937 /// Put with timestamp in a specific column family.
3938 /// If "key" already exists, it will coexist with previous entry.
3939 /// `Get` with a timestamp ts specified in ReadOptions will return
3940 /// the most recent key/value whose timestamp is smaller than or equal to ts.
3941 /// Takes an additional argument `ts` as the timestamp.
3942 /// Note: the DB must be opened with user defined timestamp enabled.
3943 pub fn put_cf_with_ts<K, V, S>(
3944 &self,
3945 cf: &impl AsColumnFamilyRef,
3946 key: K,
3947 ts: S,
3948 value: V,
3949 ) -> Result<(), Error>
3950 where
3951 K: AsRef<[u8]>,
3952 V: AsRef<[u8]>,
3953 S: AsRef<[u8]>,
3954 {
3955 DEFAULT_WRITE_OPTS.with(|opts| {
3956 self.put_cf_with_ts_opt(cf, key.as_ref(), ts.as_ref(), value.as_ref(), opts)
3957 })
3958 }
3959
3960 pub fn merge<K, V>(&self, key: K, value: V) -> Result<(), Error>
3961 where
3962 K: AsRef<[u8]>,
3963 V: AsRef<[u8]>,
3964 {
3965 DEFAULT_WRITE_OPTS.with(|opts| self.merge_opt(key, value, opts))
3966 }
3967
3968 pub fn merge_cf<K, V>(&self, cf: &impl AsColumnFamilyRef, key: K, value: V) -> Result<(), Error>
3969 where
3970 K: AsRef<[u8]>,
3971 V: AsRef<[u8]>,
3972 {
3973 DEFAULT_WRITE_OPTS.with(|opts| self.merge_cf_opt(cf, key, value, opts))
3974 }
3975
3976 pub fn delete<K: AsRef<[u8]>>(&self, key: K) -> Result<(), Error> {
3977 DEFAULT_WRITE_OPTS.with(|opts| self.delete_opt(key, opts))
3978 }
3979
3980 pub fn delete_cf<K: AsRef<[u8]>>(
3981 &self,
3982 cf: &impl AsColumnFamilyRef,
3983 key: K,
3984 ) -> Result<(), Error> {
3985 DEFAULT_WRITE_OPTS.with(|opts| self.delete_cf_opt(cf, key, opts))
3986 }
3987
3988 /// Remove the database entry (if any) for "key".
3989 /// Takes an additional argument `ts` as the timestamp.
3990 /// Note: the DB must be opened with user defined timestamp enabled.
3991 pub fn delete_with_ts<K: AsRef<[u8]>, S: AsRef<[u8]>>(
3992 &self,
3993 key: K,
3994 ts: S,
3995 ) -> Result<(), Error> {
3996 DEFAULT_WRITE_OPTS.with(|opts| self.delete_with_ts_opt(key, ts, opts))
3997 }
3998
3999 /// Delete with timestamp in a specific column family.
4000 /// Takes an additional argument `ts` as the timestamp.
4001 /// Note: the DB must be opened with user defined timestamp enabled.
4002 pub fn delete_cf_with_ts<K: AsRef<[u8]>, S: AsRef<[u8]>>(
4003 &self,
4004 cf: &impl AsColumnFamilyRef,
4005 key: K,
4006 ts: S,
4007 ) -> Result<(), Error> {
4008 DEFAULT_WRITE_OPTS.with(|opts| self.delete_cf_with_ts_opt(cf, key, ts, opts))
4009 }
4010
4011 /// Remove the database entry for "key" with WriteOptions.
4012 ///
4013 /// Requires that the key exists and was not overwritten. Returns OK on success,
4014 /// and a non-OK status on error. It is not an error if "key" did not exist in the database.
4015 ///
4016 /// If a key is overwritten (by calling Put() multiple times), then the result
4017 /// of calling SingleDelete() on this key is undefined. SingleDelete() only
4018 /// behaves correctly if there has been only one Put() for this key since the
4019 /// previous call to SingleDelete() for this key.
4020 ///
4021 /// This feature is currently an experimental performance optimization
4022 /// for a very specific workload. It is up to the caller to ensure that
4023 /// SingleDelete is only used for a key that is not deleted using Delete() or
4024 /// written using Merge(). Mixing SingleDelete operations with Deletes and
4025 /// Merges can result in undefined behavior.
4026 ///
4027 /// Note: consider setting options.sync = true.
4028 ///
4029 /// For more information, see <https://github.com/facebook/rocksdb/wiki/Single-Delete>
4030 pub fn single_delete_opt<K: AsRef<[u8]>>(
4031 &self,
4032 key: K,
4033 writeopts: &WriteOptions,
4034 ) -> Result<(), Error> {
4035 let key = key.as_ref();
4036
4037 unsafe {
4038 ffi_try!(ffi::rocksdb_singledelete(
4039 self.inner.inner(),
4040 writeopts.inner,
4041 key.as_ptr() as *const c_char,
4042 key.len() as size_t,
4043 ));
4044 Ok(())
4045 }
4046 }
4047
4048 /// Remove the database entry for "key" from a specific column family with WriteOptions.
4049 ///
4050 /// See single_delete_opt() for detailed behavior and restrictions.
4051 pub fn single_delete_cf_opt<K: AsRef<[u8]>>(
4052 &self,
4053 cf: &impl AsColumnFamilyRef,
4054 key: K,
4055 writeopts: &WriteOptions,
4056 ) -> Result<(), Error> {
4057 let key = key.as_ref();
4058
4059 unsafe {
4060 ffi_try!(ffi::rocksdb_singledelete_cf(
4061 self.inner.inner(),
4062 writeopts.inner,
4063 cf.inner(),
4064 key.as_ptr() as *const c_char,
4065 key.len() as size_t,
4066 ));
4067 Ok(())
4068 }
4069 }
4070
4071 /// Remove the database entry for "key" with WriteOptions.
4072 ///
4073 /// Takes an additional argument `ts` as the timestamp.
4074 /// Note: the DB must be opened with user defined timestamp enabled.
4075 ///
4076 /// See single_delete_opt() for detailed behavior and restrictions.
4077 pub fn single_delete_with_ts_opt<K, S>(
4078 &self,
4079 key: K,
4080 ts: S,
4081 writeopts: &WriteOptions,
4082 ) -> Result<(), Error>
4083 where
4084 K: AsRef<[u8]>,
4085 S: AsRef<[u8]>,
4086 {
4087 let key = key.as_ref();
4088 let ts = ts.as_ref();
4089 unsafe {
4090 ffi_try!(ffi::rocksdb_singledelete_with_ts(
4091 self.inner.inner(),
4092 writeopts.inner,
4093 key.as_ptr() as *const c_char,
4094 key.len() as size_t,
4095 ts.as_ptr() as *const c_char,
4096 ts.len() as size_t,
4097 ));
4098 Ok(())
4099 }
4100 }
4101
4102 /// Remove the database entry for "key" from a specific column family with WriteOptions.
4103 ///
4104 /// Takes an additional argument `ts` as the timestamp.
4105 /// Note: the DB must be opened with user defined timestamp enabled.
4106 ///
4107 /// See single_delete_opt() for detailed behavior and restrictions.
4108 pub fn single_delete_cf_with_ts_opt<K, S>(
4109 &self,
4110 cf: &impl AsColumnFamilyRef,
4111 key: K,
4112 ts: S,
4113 writeopts: &WriteOptions,
4114 ) -> Result<(), Error>
4115 where
4116 K: AsRef<[u8]>,
4117 S: AsRef<[u8]>,
4118 {
4119 let key = key.as_ref();
4120 let ts = ts.as_ref();
4121 unsafe {
4122 ffi_try!(ffi::rocksdb_singledelete_cf_with_ts(
4123 self.inner.inner(),
4124 writeopts.inner,
4125 cf.inner(),
4126 key.as_ptr() as *const c_char,
4127 key.len() as size_t,
4128 ts.as_ptr() as *const c_char,
4129 ts.len() as size_t,
4130 ));
4131 Ok(())
4132 }
4133 }
4134
4135 /// Remove the database entry for "key".
4136 ///
4137 /// See single_delete_opt() for detailed behavior and restrictions.
4138 pub fn single_delete<K: AsRef<[u8]>>(&self, key: K) -> Result<(), Error> {
4139 DEFAULT_WRITE_OPTS.with(|opts| self.single_delete_opt(key, opts))
4140 }
4141
4142 /// Remove the database entry for "key" from a specific column family.
4143 ///
4144 /// See single_delete_opt() for detailed behavior and restrictions.
4145 pub fn single_delete_cf<K: AsRef<[u8]>>(
4146 &self,
4147 cf: &impl AsColumnFamilyRef,
4148 key: K,
4149 ) -> Result<(), Error> {
4150 DEFAULT_WRITE_OPTS.with(|opts| self.single_delete_cf_opt(cf, key, opts))
4151 }
4152
4153 /// Remove the database entry for "key".
4154 ///
4155 /// Takes an additional argument `ts` as the timestamp.
4156 /// Note: the DB must be opened with user defined timestamp enabled.
4157 ///
4158 /// See single_delete_opt() for detailed behavior and restrictions.
4159 pub fn single_delete_with_ts<K: AsRef<[u8]>, S: AsRef<[u8]>>(
4160 &self,
4161 key: K,
4162 ts: S,
4163 ) -> Result<(), Error> {
4164 DEFAULT_WRITE_OPTS.with(|opts| self.single_delete_with_ts_opt(key, ts, opts))
4165 }
4166
4167 /// Remove the database entry for "key" from a specific column family.
4168 ///
4169 /// Takes an additional argument `ts` as the timestamp.
4170 /// Note: the DB must be opened with user defined timestamp enabled.
4171 ///
4172 /// See single_delete_opt() for detailed behavior and restrictions.
4173 pub fn single_delete_cf_with_ts<K: AsRef<[u8]>, S: AsRef<[u8]>>(
4174 &self,
4175 cf: &impl AsColumnFamilyRef,
4176 key: K,
4177 ts: S,
4178 ) -> Result<(), Error> {
4179 DEFAULT_WRITE_OPTS.with(|opts| self.single_delete_cf_with_ts_opt(cf, key, ts, opts))
4180 }
4181
4182 /// Runs a manual compaction on the Range of keys given. This is not likely to be needed for typical usage.
4183 pub fn compact_range<S: AsRef<[u8]>, E: AsRef<[u8]>>(&self, start: Option<S>, end: Option<E>) {
4184 unsafe {
4185 let start = start.as_ref().map(AsRef::as_ref);
4186 let end = end.as_ref().map(AsRef::as_ref);
4187
4188 ffi::rocksdb_compact_range(
4189 self.inner.inner(),
4190 opt_bytes_to_ptr(start),
4191 start.map_or(0, <[u8]>::len) as size_t,
4192 opt_bytes_to_ptr(end),
4193 end.map_or(0, <[u8]>::len) as size_t,
4194 );
4195 }
4196 }
4197
4198 /// Same as `compact_range` but with custom options.
4199 pub fn compact_range_opt<S: AsRef<[u8]>, E: AsRef<[u8]>>(
4200 &self,
4201 start: Option<S>,
4202 end: Option<E>,
4203 opts: &CompactOptions,
4204 ) {
4205 unsafe {
4206 let start = start.as_ref().map(AsRef::as_ref);
4207 let end = end.as_ref().map(AsRef::as_ref);
4208
4209 ffi::rocksdb_compact_range_opt(
4210 self.inner.inner(),
4211 opts.inner,
4212 opt_bytes_to_ptr(start),
4213 start.map_or(0, <[u8]>::len) as size_t,
4214 opt_bytes_to_ptr(end),
4215 end.map_or(0, <[u8]>::len) as size_t,
4216 );
4217 }
4218 }
4219
4220 /// Runs a manual compaction on the Range of keys given on the
4221 /// given column family. This is not likely to be needed for typical usage.
4222 pub fn compact_range_cf<S: AsRef<[u8]>, E: AsRef<[u8]>>(
4223 &self,
4224 cf: &impl AsColumnFamilyRef,
4225 start: Option<S>,
4226 end: Option<E>,
4227 ) {
4228 unsafe {
4229 let start = start.as_ref().map(AsRef::as_ref);
4230 let end = end.as_ref().map(AsRef::as_ref);
4231
4232 ffi::rocksdb_compact_range_cf(
4233 self.inner.inner(),
4234 cf.inner(),
4235 opt_bytes_to_ptr(start),
4236 start.map_or(0, <[u8]>::len) as size_t,
4237 opt_bytes_to_ptr(end),
4238 end.map_or(0, <[u8]>::len) as size_t,
4239 );
4240 }
4241 }
4242
4243 /// Same as `compact_range_cf` but with custom options.
4244 pub fn compact_range_cf_opt<S: AsRef<[u8]>, E: AsRef<[u8]>>(
4245 &self,
4246 cf: &impl AsColumnFamilyRef,
4247 start: Option<S>,
4248 end: Option<E>,
4249 opts: &CompactOptions,
4250 ) {
4251 unsafe {
4252 let start = start.as_ref().map(AsRef::as_ref);
4253 let end = end.as_ref().map(AsRef::as_ref);
4254
4255 ffi::rocksdb_compact_range_cf_opt(
4256 self.inner.inner(),
4257 cf.inner(),
4258 opts.inner,
4259 opt_bytes_to_ptr(start),
4260 start.map_or(0, <[u8]>::len) as size_t,
4261 opt_bytes_to_ptr(end),
4262 end.map_or(0, <[u8]>::len) as size_t,
4263 );
4264 }
4265 }
4266
4267 /// Wait for all flush and compactions jobs to finish. Jobs to wait include the
4268 /// unscheduled (queued, but not scheduled yet).
4269 ///
4270 /// NOTE: This may also never return if there's sufficient ongoing writes that
4271 /// keeps flush and compaction going without stopping. The user would have to
4272 /// cease all the writes to DB to make this eventually return in a stable
4273 /// state. The user may also use timeout option in WaitForCompactOptions to
4274 /// make this stop waiting and return when timeout expires.
4275 pub fn wait_for_compact(&self, opts: &WaitForCompactOptions) -> Result<(), Error> {
4276 unsafe {
4277 ffi_try!(ffi::rocksdb_wait_for_compact(
4278 self.inner.inner(),
4279 opts.inner
4280 ));
4281 }
4282 Ok(())
4283 }
4284
4285 /// Changes mutable column family options on the default column family at
4286 /// runtime.
4287 ///
4288 /// [`set_db_options`](Self::set_db_options) is the DB wide equivalent. Either the
4289 /// whole set is applied or none of it is.
4290 ///
4291 /// # Aborts
4292 ///
4293 /// Some unparseable values take the process down instead of returning an error,
4294 /// so validate values before passing them here.
4295 ///
4296 /// RocksDB parses integers with `std::stoi` and friends
4297 /// (`util/string_util.cc:378`), which throw `std::invalid_argument`. It does try
4298 /// to catch that, but not everywhere, and an integer-valued option such as
4299 /// `write_buffer_size` given a non-numeric value aborts the process with
4300 /// "Rust cannot catch foreign exceptions". A boolean-valued option such as
4301 /// `disable_auto_compactions` returns an error instead. Do not rely on which
4302 /// options fall on which side of that line.
4303 ///
4304 /// Catching it here is not an option. `ColumnFamilyData::SetOptions` runs inside
4305 /// a callback that `VersionSet::LogAndApply` invokes while holding the DB mutex
4306 /// as the exclusive manifest writer (`db/db_impl/db_impl.cc:1655`), so unwinding
4307 /// through it leaves that state behind and the next option change on the DB
4308 /// blocks forever in `InstrumentedCondVar::Wait`. That was reproducible under
4309 /// ASAN and silent otherwise, which is worse than aborting.
4310 ///
4311 /// # Errors
4312 ///
4313 /// Returns the RocksDB error if a name is unknown or the option is not changeable
4314 /// at runtime. Also errors if any name or value contains an interior NUL byte.
4315 pub fn set_options(&self, opts: &[(&str, &str)]) -> Result<(), Error> {
4316 let copts = convert_options(opts)?;
4317 let cnames: Vec<*const c_char> = copts.iter().map(|opt| opt.0.as_ptr()).collect();
4318 let cvalues: Vec<*const c_char> = copts.iter().map(|opt| opt.1.as_ptr()).collect();
4319 unsafe {
4320 ffi_try!(ffi::rocksdb_set_options(
4321 self.inner.inner(),
4322 option_count(&copts)?,
4323 cnames.as_ptr(),
4324 cvalues.as_ptr(),
4325 ));
4326 }
4327 Ok(())
4328 }
4329
4330 /// Like [`set_options`](Self::set_options), for a single column family.
4331 ///
4332 /// # Aborts
4333 ///
4334 /// See [`set_options`](Self::set_options).
4335 ///
4336 /// # Errors
4337 ///
4338 /// See [`set_options`](Self::set_options).
4339 pub fn set_options_cf(
4340 &self,
4341 cf: &impl AsColumnFamilyRef,
4342 opts: &[(&str, &str)],
4343 ) -> Result<(), Error> {
4344 let copts = convert_options(opts)?;
4345 let cnames: Vec<*const c_char> = copts.iter().map(|opt| opt.0.as_ptr()).collect();
4346 let cvalues: Vec<*const c_char> = copts.iter().map(|opt| opt.1.as_ptr()).collect();
4347 unsafe {
4348 ffi_try!(ffi::rocksdb_set_options_cf(
4349 self.inner.inner(),
4350 cf.inner(),
4351 option_count(&copts)?,
4352 cnames.as_ptr(),
4353 cvalues.as_ptr(),
4354 ));
4355 }
4356 Ok(())
4357 }
4358
4359 /// Implementation for property_value et al methods.
4360 ///
4361 /// `name` is the name of the property. It will be converted into a CString
4362 /// and passed to `get_property` as argument. `get_property` reads the
4363 /// specified property and either returns NULL or a pointer to a C allocated
4364 /// string; this method takes ownership of that string and will free it at
4365 /// the end. That string is parsed using `parse` callback which produces
4366 /// the returned result.
4367 fn property_value_impl<R>(
4368 name: impl CStrLike,
4369 get_property: impl FnOnce(*const c_char) -> *mut c_char,
4370 parse: impl FnOnce(&str) -> Result<R, Error>,
4371 ) -> Result<Option<R>, Error> {
4372 let value = match name.bake() {
4373 Ok(prop_name) => get_property(prop_name.as_ptr()),
4374 Err(e) => {
4375 return Err(Error::new(format!(
4376 "Failed to convert property name to CString: {e}"
4377 )));
4378 }
4379 };
4380 if value.is_null() {
4381 return Ok(None);
4382 }
4383 let result = match unsafe { CStr::from_ptr(value) }.to_str() {
4384 Ok(s) => parse(s).map(|value| Some(value)),
4385 Err(e) => Err(Error::new(format!(
4386 "Failed to convert property value to string: {e}"
4387 ))),
4388 };
4389 unsafe {
4390 ffi::rocksdb_free(value as *mut c_void);
4391 }
4392 result
4393 }
4394
4395 /// Retrieves a RocksDB property by name.
4396 ///
4397 /// Full list of properties could be find
4398 /// [here](https://github.com/facebook/rocksdb/blob/08809f5e6cd9cc4bc3958dd4d59457ae78c76660/include/rocksdb/db.h#L428-L634).
4399 pub fn property_value(&self, name: impl CStrLike) -> Result<Option<String>, Error> {
4400 Self::property_value_impl(
4401 name,
4402 |prop_name| unsafe { ffi::rocksdb_property_value(self.inner.inner(), prop_name) },
4403 |str_value| Ok(str_value.to_owned()),
4404 )
4405 }
4406
4407 /// Retrieves a RocksDB property by name, for a specific column family.
4408 ///
4409 /// Full list of properties could be find
4410 /// [here](https://github.com/facebook/rocksdb/blob/08809f5e6cd9cc4bc3958dd4d59457ae78c76660/include/rocksdb/db.h#L428-L634).
4411 pub fn property_value_cf(
4412 &self,
4413 cf: &impl AsColumnFamilyRef,
4414 name: impl CStrLike,
4415 ) -> Result<Option<String>, Error> {
4416 Self::property_value_impl(
4417 name,
4418 |prop_name| unsafe {
4419 ffi::rocksdb_property_value_cf(self.inner.inner(), cf.inner(), prop_name)
4420 },
4421 |str_value| Ok(str_value.to_owned()),
4422 )
4423 }
4424
4425 fn property_int_value_impl(
4426 name: impl CStrLike,
4427 get_property: impl FnOnce(*const c_char, *mut u64) -> c_int,
4428 get_string_property: impl FnOnce(*const c_char) -> *mut c_char,
4429 ) -> Result<Option<u64>, Error> {
4430 let prop_name = name.bake().map_err(|err| {
4431 Error::new(format!("Failed to convert property name to CString: {err}"))
4432 })?;
4433 let mut value = 0;
4434 if get_property(prop_name.as_ptr(), &raw mut value) == 0 {
4435 return Ok(Some(value));
4436 }
4437
4438 Self::property_value_impl(
4439 prop_name.as_ref(),
4440 get_string_property,
4441 Self::parse_property_int_value,
4442 )
4443 }
4444
4445 fn parse_property_int_value(value: &str) -> Result<u64, Error> {
4446 value.parse::<u64>().map_err(|err| {
4447 Error::new(format!(
4448 "Failed to convert property value {value} to int: {err}"
4449 ))
4450 })
4451 }
4452
4453 /// Retrieves a RocksDB property and casts it to an integer.
4454 ///
4455 /// Full list of properties that return int values could be find
4456 /// [here](https://github.com/facebook/rocksdb/blob/08809f5e6cd9cc4bc3958dd4d59457ae78c76660/include/rocksdb/db.h#L654-L689).
4457 pub fn property_int_value(&self, name: impl CStrLike) -> Result<Option<u64>, Error> {
4458 Self::property_int_value_impl(
4459 name,
4460 |prop_name, value| unsafe {
4461 ffi::rocksdb_property_int(self.inner.inner(), prop_name, value)
4462 },
4463 |prop_name| unsafe { ffi::rocksdb_property_value(self.inner.inner(), prop_name) },
4464 )
4465 }
4466
4467 /// Retrieves a RocksDB property for a specific column family and casts it to an integer.
4468 ///
4469 /// Full list of properties that return int values could be find
4470 /// [here](https://github.com/facebook/rocksdb/blob/08809f5e6cd9cc4bc3958dd4d59457ae78c76660/include/rocksdb/db.h#L654-L689).
4471 pub fn property_int_value_cf(
4472 &self,
4473 cf: &impl AsColumnFamilyRef,
4474 name: impl CStrLike,
4475 ) -> Result<Option<u64>, Error> {
4476 Self::property_int_value_impl(
4477 name,
4478 |prop_name, value| unsafe {
4479 ffi::rocksdb_property_int_cf(self.inner.inner(), cf.inner(), prop_name, value)
4480 },
4481 |prop_name| unsafe {
4482 ffi::rocksdb_property_value_cf(self.inner.inner(), cf.inner(), prop_name)
4483 },
4484 )
4485 }
4486
4487 /// The sequence number of the most recent transaction.
4488 pub fn latest_sequence_number(&self) -> u64 {
4489 unsafe { ffi::rocksdb_get_latest_sequence_number(self.inner.inner()) }
4490 }
4491
4492 /// Return the approximate file system space used by keys in each ranges.
4493 ///
4494 /// Note that the returned sizes measure file system space usage, so
4495 /// if the user data compresses by a factor of ten, the returned
4496 /// sizes will be one-tenth the size of the corresponding user data size.
4497 ///
4498 /// Due to lack of abi, only data flushed to disk is taken into account.
4499 /// # Errors
4500 ///
4501 /// Returns the RocksDB error if the size estimate fails, for instance on an
4502 /// I/O error reading the manifest. No partial sizes are reported in that
4503 /// case.
4504 pub fn get_approximate_sizes(&self, ranges: &[Range]) -> Result<Vec<u64>, Error> {
4505 self.get_approximate_sizes_cfopt(None::<&ColumnFamily>, ranges)
4506 }
4507
4508 /// Like [`Self::get_approximate_sizes`], for a single column family.
4509 ///
4510 /// # Errors
4511 ///
4512 /// See [`Self::get_approximate_sizes`].
4513 pub fn get_approximate_sizes_cf(
4514 &self,
4515 cf: &impl AsColumnFamilyRef,
4516 ranges: &[Range],
4517 ) -> Result<Vec<u64>, Error> {
4518 self.get_approximate_sizes_cfopt(Some(cf), ranges)
4519 }
4520
4521 fn get_approximate_sizes_cfopt(
4522 &self,
4523 cf: Option<&impl AsColumnFamilyRef>,
4524 ranges: &[Range],
4525 ) -> Result<Vec<u64>, Error> {
4526 let mut args = ApproximateSizesArgs::new(ranges);
4527 let mut err: *mut c_char = ptr::null_mut();
4528 match cf {
4529 None => unsafe {
4530 ffi::rocksdb_approximate_sizes(
4531 self.inner.inner(),
4532 args.count,
4533 args.start_keys.as_ptr(),
4534 args.start_key_lens.as_ptr(),
4535 args.end_keys.as_ptr(),
4536 args.end_key_lens.as_ptr(),
4537 args.sizes.as_mut_ptr(),
4538 &raw mut err,
4539 );
4540 },
4541 Some(cf) => unsafe {
4542 ffi::rocksdb_approximate_sizes_cf(
4543 self.inner.inner(),
4544 cf.inner(),
4545 args.count,
4546 args.start_keys.as_ptr(),
4547 args.start_key_lens.as_ptr(),
4548 args.end_keys.as_ptr(),
4549 args.end_key_lens.as_ptr(),
4550 args.sizes.as_mut_ptr(),
4551 &raw mut err,
4552 );
4553 },
4554 }
4555 args.finish(err)
4556 }
4557
4558 /// Like [`Self::get_approximate_sizes`], but lets the caller say what counts
4559 /// towards the total and how precise the answer has to be.
4560 ///
4561 /// [`Self::get_approximate_sizes`] counts SST files only. Pass a
4562 /// [`SizeApproximationOptions`] with
4563 /// [`set_include_memtables`](SizeApproximationOptions::set_include_memtables)
4564 /// on to include writes that have not been flushed yet.
4565 ///
4566 /// # Errors
4567 ///
4568 /// See [`Self::get_approximate_sizes`].
4569 pub fn get_approximate_sizes_with_options(
4570 &self,
4571 opts: &SizeApproximationOptions,
4572 ranges: &[Range],
4573 ) -> Result<Vec<u64>, Error> {
4574 let mut args = ApproximateSizesArgs::new(ranges);
4575 let mut err: *mut c_char = ptr::null_mut();
4576 unsafe {
4577 ffi::rocksdb_approximate_sizes_with_options(
4578 self.inner.inner(),
4579 opts.as_ptr(),
4580 args.count,
4581 args.start_keys.as_ptr(),
4582 args.start_key_lens.as_ptr(),
4583 args.end_keys.as_ptr(),
4584 args.end_key_lens.as_ptr(),
4585 args.sizes.as_mut_ptr(),
4586 &raw mut err,
4587 );
4588 }
4589 args.finish(err)
4590 }
4591
4592 /// Like [`Self::get_approximate_sizes_with_options`], for a single column
4593 /// family.
4594 ///
4595 /// # Errors
4596 ///
4597 /// See [`Self::get_approximate_sizes`].
4598 pub fn get_approximate_sizes_cf_with_options(
4599 &self,
4600 cf: &impl AsColumnFamilyRef,
4601 opts: &SizeApproximationOptions,
4602 ranges: &[Range],
4603 ) -> Result<Vec<u64>, Error> {
4604 let mut args = ApproximateSizesArgs::new(ranges);
4605 let mut err: *mut c_char = ptr::null_mut();
4606 unsafe {
4607 ffi::rocksdb_approximate_sizes_cf_with_options(
4608 self.inner.inner(),
4609 cf.inner(),
4610 opts.as_ptr(),
4611 args.count,
4612 args.start_keys.as_ptr(),
4613 args.start_key_lens.as_ptr(),
4614 args.end_keys.as_ptr(),
4615 args.end_key_lens.as_ptr(),
4616 args.sizes.as_mut_ptr(),
4617 &raw mut err,
4618 );
4619 }
4620 args.finish(err)
4621 }
4622
4623 /// Like [`Self::get_approximate_sizes_cf_with_options`], but selects what
4624 /// counts with a flag set instead of an options object.
4625 ///
4626 /// Prefer [`Self::get_approximate_sizes_cf_with_options`], which can also
4627 /// set an error margin. This variant exists because it is the form RocksDB
4628 /// offers without allocating.
4629 ///
4630 /// # Errors
4631 ///
4632 /// See [`Self::get_approximate_sizes`].
4633 pub fn get_approximate_sizes_cf_with_flags(
4634 &self,
4635 cf: &impl AsColumnFamilyRef,
4636 flags: SizeApproximationFlags,
4637 ranges: &[Range],
4638 ) -> Result<Vec<u64>, Error> {
4639 let mut args = ApproximateSizesArgs::new(ranges);
4640 let mut err: *mut c_char = ptr::null_mut();
4641 unsafe {
4642 ffi::rocksdb_approximate_sizes_cf_with_flags(
4643 self.inner.inner(),
4644 cf.inner(),
4645 args.count,
4646 args.start_keys.as_ptr(),
4647 args.start_key_lens.as_ptr(),
4648 args.end_keys.as_ptr(),
4649 args.end_key_lens.as_ptr(),
4650 flags.bits(),
4651 args.sizes.as_mut_ptr(),
4652 &raw mut err,
4653 );
4654 }
4655 args.finish(err)
4656 }
4657
4658 /// Iterate over batches of write operations since a given sequence.
4659 ///
4660 /// Produce an iterator that will provide the batches of write operations
4661 /// that have occurred since the given sequence (see
4662 /// `latest_sequence_number()`). Use the provided iterator to retrieve each
4663 /// (`u64`, `WriteBatch`) tuple, and then gather the individual puts and
4664 /// deletes using the `WriteBatch::iterate()` function.
4665 ///
4666 /// Calling `get_updates_since()` with a sequence number that is out of
4667 /// bounds will return an error.
4668 pub fn get_updates_since(&self, seq_number: u64) -> Result<DBWALIterator, Error> {
4669 unsafe {
4670 // rocksdb_wal_readoptions_t does not appear to have any functions
4671 // for creating and destroying it; fortunately we can pass a nullptr
4672 // here to get the default behavior
4673 let opts: *const ffi::rocksdb_wal_readoptions_t = ptr::null();
4674 let iter = ffi_try!(ffi::rocksdb_get_updates_since(
4675 self.inner.inner(),
4676 seq_number,
4677 opts
4678 ));
4679 Ok(DBWALIterator {
4680 inner: iter,
4681 start_seq_number: seq_number,
4682 })
4683 }
4684 }
4685
4686 /// Tries to catch up with the primary by reading as much as possible from the
4687 /// log files.
4688 pub fn try_catch_up_with_primary(&self) -> Result<(), Error> {
4689 unsafe {
4690 ffi_try!(ffi::rocksdb_try_catch_up_with_primary(self.inner.inner()));
4691 }
4692 Ok(())
4693 }
4694
4695 /// Loads a list of external SST files created with SstFileWriter into the DB with default opts
4696 pub fn ingest_external_file<P: AsRef<Path>>(&self, paths: Vec<P>) -> Result<(), Error> {
4697 let opts = IngestExternalFileOptions::default();
4698 self.ingest_external_file_opts(&opts, paths)
4699 }
4700
4701 /// Loads a list of external SST files created with SstFileWriter into the DB
4702 pub fn ingest_external_file_opts<P: AsRef<Path>>(
4703 &self,
4704 opts: &IngestExternalFileOptions,
4705 paths: Vec<P>,
4706 ) -> Result<(), Error> {
4707 let paths_v: Vec<CString> = paths.iter().map(to_cpath).collect::<Result<Vec<_>, _>>()?;
4708 let cpaths: Vec<_> = paths_v.iter().map(|path| path.as_ptr()).collect();
4709
4710 self.ingest_external_file_raw(opts, &paths_v, &cpaths)
4711 }
4712
4713 /// Loads a list of external SST files created with SstFileWriter into the DB for given Column Family
4714 /// with default opts
4715 pub fn ingest_external_file_cf<P: AsRef<Path>>(
4716 &self,
4717 cf: &impl AsColumnFamilyRef,
4718 paths: Vec<P>,
4719 ) -> Result<(), Error> {
4720 let opts = IngestExternalFileOptions::default();
4721 self.ingest_external_file_cf_opts(cf, &opts, paths)
4722 }
4723
4724 /// Loads a list of external SST files created with SstFileWriter into the DB for given Column Family
4725 pub fn ingest_external_file_cf_opts<P: AsRef<Path>>(
4726 &self,
4727 cf: &impl AsColumnFamilyRef,
4728 opts: &IngestExternalFileOptions,
4729 paths: Vec<P>,
4730 ) -> Result<(), Error> {
4731 let paths_v: Vec<CString> = paths.iter().map(to_cpath).collect::<Result<Vec<_>, _>>()?;
4732 let cpaths: Vec<_> = paths_v.iter().map(|path| path.as_ptr()).collect();
4733
4734 self.ingest_external_file_raw_cf(cf, opts, &paths_v, &cpaths)
4735 }
4736
4737 fn ingest_external_file_raw(
4738 &self,
4739 opts: &IngestExternalFileOptions,
4740 paths_v: &[CString],
4741 cpaths: &[*const c_char],
4742 ) -> Result<(), Error> {
4743 unsafe {
4744 ffi_try!(ffi::rocksdb_ingest_external_file(
4745 self.inner.inner(),
4746 cpaths.as_ptr(),
4747 paths_v.len(),
4748 opts.inner.cast_const()
4749 ));
4750 Ok(())
4751 }
4752 }
4753
4754 fn ingest_external_file_raw_cf(
4755 &self,
4756 cf: &impl AsColumnFamilyRef,
4757 opts: &IngestExternalFileOptions,
4758 paths_v: &[CString],
4759 cpaths: &[*const c_char],
4760 ) -> Result<(), Error> {
4761 unsafe {
4762 ffi_try!(ffi::rocksdb_ingest_external_file_cf(
4763 self.inner.inner(),
4764 cf.inner(),
4765 cpaths.as_ptr(),
4766 paths_v.len(),
4767 opts.inner.cast_const()
4768 ));
4769 Ok(())
4770 }
4771 }
4772
4773 /// Obtains the LSM-tree meta data of the default column family of the DB
4774 pub fn get_column_family_metadata(&self) -> ColumnFamilyMetaData {
4775 unsafe {
4776 let ptr = ffi::rocksdb_get_column_family_metadata(self.inner.inner());
4777
4778 let metadata = ColumnFamilyMetaData {
4779 size: ffi::rocksdb_column_family_metadata_get_size(ptr),
4780 name: from_cstr_and_free(ffi::rocksdb_column_family_metadata_get_name(ptr)),
4781 file_count: ffi::rocksdb_column_family_metadata_get_file_count(ptr),
4782 };
4783
4784 // destroy
4785 ffi::rocksdb_column_family_metadata_destroy(ptr);
4786
4787 // return
4788 metadata
4789 }
4790 }
4791
4792 /// Obtains the LSM-tree meta data of the specified column family of the DB
4793 pub fn get_column_family_metadata_cf(
4794 &self,
4795 cf: &impl AsColumnFamilyRef,
4796 ) -> ColumnFamilyMetaData {
4797 unsafe {
4798 let ptr = ffi::rocksdb_get_column_family_metadata_cf(self.inner.inner(), cf.inner());
4799
4800 let metadata = ColumnFamilyMetaData {
4801 size: ffi::rocksdb_column_family_metadata_get_size(ptr),
4802 name: from_cstr_and_free(ffi::rocksdb_column_family_metadata_get_name(ptr)),
4803 file_count: ffi::rocksdb_column_family_metadata_get_file_count(ptr),
4804 };
4805
4806 // destroy
4807 ffi::rocksdb_column_family_metadata_destroy(ptr);
4808
4809 // return
4810 metadata
4811 }
4812 }
4813
4814 /// Compacts the named SST files into `output_level`, giving the caller
4815 /// control over exactly which files are merged.
4816 ///
4817 /// [`Self::compact_range`] picks the files itself from a key range. This
4818 /// picks them by name, which is what a tool driving compaction from
4819 /// [`Self::live_files`] or [`Self::get_column_family_metadata_with_options`]
4820 /// needs. The file names come from that metadata, not from the filesystem.
4821 ///
4822 /// Runs on the calling thread, so it blocks until the compaction finishes.
4823 ///
4824 /// # Errors
4825 ///
4826 /// Returns the RocksDB error if any input file is unknown, if the files do
4827 /// not form a compactible set, or if the compaction itself fails. Nothing is
4828 /// compacted in that case.
4829 pub fn compact_files<I, N>(
4830 &self,
4831 opts: &CompactionOptions,
4832 input_file_names: I,
4833 output_level: i32,
4834 ) -> Result<CompactFilesResult, Error>
4835 where
4836 I: IntoIterator<Item = N>,
4837 N: CStrLike,
4838 {
4839 self.compact_files_impl(None::<&ColumnFamily>, opts, input_file_names, output_level)
4840 }
4841
4842 /// Like [`Self::compact_files`], for a single column family.
4843 ///
4844 /// # Errors
4845 ///
4846 /// See [`Self::compact_files`].
4847 pub fn compact_files_cf<I, N>(
4848 &self,
4849 cf: &impl AsColumnFamilyRef,
4850 opts: &CompactionOptions,
4851 input_file_names: I,
4852 output_level: i32,
4853 ) -> Result<CompactFilesResult, Error>
4854 where
4855 I: IntoIterator<Item = N>,
4856 N: CStrLike,
4857 {
4858 self.compact_files_impl(Some(cf), opts, input_file_names, output_level)
4859 }
4860
4861 fn compact_files_impl<I, N>(
4862 &self,
4863 cf: Option<&impl AsColumnFamilyRef>,
4864 opts: &CompactionOptions,
4865 input_file_names: I,
4866 output_level: i32,
4867 ) -> Result<CompactFilesResult, Error>
4868 where
4869 I: IntoIterator<Item = N>,
4870 N: CStrLike,
4871 {
4872 let names = input_file_names
4873 .into_iter()
4874 .map(CStrLike::into_c_string)
4875 .collect::<Result<Vec<_>, _>>()
4876 .map_err(|e| Error::new(format!("Invalid input file name: {e}")))?;
4877 let name_ptrs: Vec<*const c_char> = names.iter().map(|n| n.as_ptr()).collect();
4878
4879 let mut output_names: *mut *mut c_char = ptr::null_mut();
4880 let mut output_count: usize = 0;
4881
4882 // A trivial move returns from `CompactFilesImpl` before
4883 // `BuildCompactionJobInfo` runs, so it leaves a caller-allocated job info
4884 // untouched while still reporting success, and nothing afterwards says
4885 // which path ran. Asking for the job info only when that path is off
4886 // keeps every value handed out one RocksDB actually wrote.
4887 let mut job_info = (!opts.get_allow_trivial_move()).then(OwnedCompactionJobInfo::new);
4888 let job_info_ptr = job_info
4889 .as_mut()
4890 .map_or(ptr::null_mut(), OwnedCompactionJobInfo::as_mut_ptr);
4891
4892 // `output_path_id` 0 means the first configured DB path. RocksDB has no
4893 // named constant for it and the crate does not expose `cf_paths`
4894 // selection here, so it is fixed rather than a parameter nobody could
4895 // use meaningfully.
4896 let output_path_id = 0;
4897
4898 unsafe {
4899 match cf {
4900 None => ffi_try!(ffi::rocksdb_compact_files(
4901 self.inner.inner(),
4902 opts.inner.cast_const(),
4903 name_ptrs.as_ptr(),
4904 name_ptrs.len(),
4905 output_level,
4906 output_path_id,
4907 &raw mut output_names,
4908 &raw mut output_count,
4909 job_info_ptr,
4910 )),
4911 Some(cf) => ffi_try!(ffi::rocksdb_compact_files_cf(
4912 self.inner.inner(),
4913 cf.inner(),
4914 opts.inner.cast_const(),
4915 name_ptrs.as_ptr(),
4916 name_ptrs.len(),
4917 output_level,
4918 output_path_id,
4919 &raw mut output_names,
4920 &raw mut output_count,
4921 job_info_ptr,
4922 )),
4923 }
4924 }
4925
4926 // On failure RocksDB returns before touching either out-param, so this
4927 // only runs once the call succeeded and both are known good.
4928 let output_files = unsafe { collect_and_free_output_names(output_names, output_count) };
4929
4930 Ok(CompactFilesResult {
4931 output_files,
4932 job_info,
4933 })
4934 }
4935
4936 /// Obtains the LSM-tree meta data of the default column family, reporting
4937 /// only the levels and files `opts` selects.
4938 ///
4939 /// Unlike [`Self::get_column_family_metadata`], which returns only the
4940 /// totals, this reports every level and every SST file in it.
4941 pub fn get_column_family_metadata_with_options(
4942 &self,
4943 opts: &ColumnFamilyMetaDataOptions,
4944 ) -> Vec<LevelMetaData> {
4945 unsafe {
4946 let ptr = ffi::rocksdb_get_column_family_metadata_with_options(
4947 self.inner.inner(),
4948 opts.inner,
4949 );
4950 // The level and file handles borrow from `ptr`, so the returned
4951 // values own it and destroy it when the last one drops.
4952 levels_from_cf_metadata_owned(ptr)
4953 }
4954 }
4955
4956 /// Like [`Self::get_column_family_metadata_with_options`], for a single
4957 /// column family.
4958 pub fn get_column_family_metadata_cf_with_options(
4959 &self,
4960 cf: &impl AsColumnFamilyRef,
4961 opts: &ColumnFamilyMetaDataOptions,
4962 ) -> Vec<LevelMetaData> {
4963 unsafe {
4964 let ptr = ffi::rocksdb_get_column_family_metadata_cf_with_options(
4965 self.inner.inner(),
4966 cf.inner(),
4967 opts.inner,
4968 );
4969 levels_from_cf_metadata_owned(ptr)
4970 }
4971 }
4972
4973 /// Returns every file RocksDB needs in order to restore this DB, which is
4974 /// what a copy-based backup has to capture.
4975 ///
4976 /// This covers more than [`Self::live_files`] does: alongside the SST files
4977 /// it reports the WAL, manifest, options and `CURRENT` files, each with the
4978 /// size and, when `opts` asks for it, the checksum needed to verify a copy.
4979 ///
4980 /// # Errors
4981 ///
4982 /// Returns the RocksDB error if the file list cannot be gathered, for
4983 /// instance when a flush it needs to run fails.
4984 pub fn get_livefiles_storage_info(
4985 &self,
4986 opts: &LiveFilesStorageInfoOptions,
4987 ) -> Result<LiveFilesStorageInfo, Error> {
4988 unsafe {
4989 let ptr = ffi_try!(ffi::rocksdb_get_livefiles_storage_info(
4990 self.inner.inner(),
4991 opts.inner,
4992 ));
4993 if ptr.is_null() {
4994 return Err(Error::new(
4995 "Could not get live files storage info".to_owned(),
4996 ));
4997 }
4998 Ok(LiveFilesStorageInfo::from_ptr(ptr))
4999 }
5000 }
5001
5002 /// Returns every WAL file the DB currently knows about, oldest first,
5003 /// including the one being written to.
5004 ///
5005 /// # Errors
5006 ///
5007 /// Returns the RocksDB error if the WAL directory cannot be listed.
5008 pub fn get_sorted_wal_files(&self) -> Result<WalFiles, Error> {
5009 unsafe {
5010 let ptr = ffi_try!(ffi::rocksdb_get_sorted_wal_files(self.inner.inner()));
5011 if ptr.is_null() {
5012 return Err(Error::new("Could not get sorted WAL files".to_owned()));
5013 }
5014 Ok(WalFiles::from_ptr(ptr))
5015 }
5016 }
5017
5018 /// Returns the WAL file currently being written to.
5019 ///
5020 /// Reported as alive, with its size read from the filesystem, and with a
5021 /// [`start_sequence`](crate::wal::WalFile::start_sequence) of 0 rather than a
5022 /// real sequence number.
5023 ///
5024 /// # Errors
5025 ///
5026 /// Returns the RocksDB error if the current WAL file cannot be identified.
5027 pub fn get_current_wal_file(&self) -> Result<OwnedWalFile, Error> {
5028 unsafe {
5029 let ptr = ffi_try!(ffi::rocksdb_get_current_wal_file(self.inner.inner()));
5030 if ptr.is_null() {
5031 return Err(Error::new("Could not get current WAL file".to_owned()));
5032 }
5033 Ok(OwnedWalFile::from_ptr(ptr))
5034 }
5035 }
5036
5037 /// Starts recording every read and write to a trace file at `trace_path`,
5038 /// which [`Self::new_default_replayer`] can later replay against a DB.
5039 ///
5040 /// The trace file is written through the DB's own `Env` with default file
5041 /// options, so it is not rate limited.
5042 ///
5043 /// Call [`Self::end_trace`] to stop.
5044 ///
5045 /// Starting a second trace without ending the first does not fail. `StartTrace`
5046 /// installs the new tracer unconditionally, and the old one is dropped without
5047 /// its buffered records being written, so the first trace file is left
5048 /// truncated. Unlike [`Self::start_io_trace`] and
5049 /// [`Self::start_block_cache_trace`], which report `Busy`, this one is the
5050 /// caller's to get right.
5051 ///
5052 /// # Errors
5053 ///
5054 /// Returns the RocksDB error if the trace file cannot be created.
5055 pub fn start_trace<P: AsRef<Path>>(
5056 &self,
5057 opts: &TraceOptions,
5058 trace_path: P,
5059 ) -> Result<(), Error> {
5060 let cpath = to_cpath(trace_path)?;
5061 unsafe {
5062 // Null `env` uses the DB's own `Env`, which the DB already keeps
5063 // alive, and null `env_options` uses RocksDB's defaults. Passing an
5064 // `EnvOptions` here would be unsound: the trace writer keeps its
5065 // `rate_limiter` as a borrowed pointer for the life of the trace,
5066 // but the reference counting it lives behind stays in the caller's
5067 // handle.
5068 ffi_try!(ffi::rocksdb_start_trace(
5069 self.inner.inner(),
5070 ptr::null_mut(),
5071 ptr::null(),
5072 opts.inner,
5073 cpath.as_ptr(),
5074 ));
5075 }
5076 Ok(())
5077 }
5078
5079 /// Stops the trace started by [`Self::start_trace`] and closes the file.
5080 ///
5081 /// # Errors
5082 ///
5083 /// Returns the RocksDB error if no trace is running or the file cannot be
5084 /// closed cleanly.
5085 pub fn end_trace(&self) -> Result<(), Error> {
5086 unsafe {
5087 ffi_try!(ffi::rocksdb_end_trace(self.inner.inner()));
5088 }
5089 Ok(())
5090 }
5091
5092 /// Starts recording file system operations to a trace file at `trace_path`,
5093 /// for diagnosing IO behaviour rather than replaying queries.
5094 ///
5095 /// Written the same way as [`Self::start_trace`]. Call [`Self::end_io_trace`]
5096 /// to stop.
5097 ///
5098 /// # Errors
5099 ///
5100 /// Returns the RocksDB error if the trace file cannot be created, or `Busy` if
5101 /// an IO trace is already running.
5102 pub fn start_io_trace<P: AsRef<Path>>(
5103 &self,
5104 opts: &TraceOptions,
5105 trace_path: P,
5106 ) -> Result<(), Error> {
5107 let cpath = to_cpath(trace_path)?;
5108 unsafe {
5109 ffi_try!(ffi::rocksdb_start_io_trace(
5110 self.inner.inner(),
5111 ptr::null_mut(),
5112 ptr::null(),
5113 opts.inner,
5114 cpath.as_ptr(),
5115 ));
5116 }
5117 Ok(())
5118 }
5119
5120 /// Stops the IO trace started by [`Self::start_io_trace`].
5121 ///
5122 /// Does nothing if no IO trace is running.
5123 ///
5124 /// # Errors
5125 ///
5126 /// `EndIOTrace` always reports success, so this only fails if a future RocksDB
5127 /// gives it something to report. It returns a `Result` to keep that a
5128 /// non-breaking change.
5129 pub fn end_io_trace(&self) -> Result<(), Error> {
5130 unsafe {
5131 ffi_try!(ffi::rocksdb_end_io_trace(self.inner.inner()));
5132 }
5133 Ok(())
5134 }
5135
5136 /// Starts recording block cache accesses to a trace file at `trace_path`,
5137 /// which the `block_cache_trace_analyzer` tool can then simulate cache
5138 /// configurations against.
5139 ///
5140 /// Written the same way as [`Self::start_trace`]. Call
5141 /// [`Self::end_block_cache_trace`] to stop.
5142 ///
5143 /// # Errors
5144 ///
5145 /// Returns the RocksDB error if the trace file cannot be created, or `Busy` if
5146 /// a block cache trace is already running.
5147 pub fn start_block_cache_trace<P: AsRef<Path>>(
5148 &self,
5149 opts: &BlockCacheTraceOptions,
5150 writer_opts: &BlockCacheTraceWriterOptions,
5151 trace_path: P,
5152 ) -> Result<(), Error> {
5153 let cpath = to_cpath(trace_path)?;
5154 unsafe {
5155 ffi_try!(ffi::rocksdb_start_block_cache_trace_with_options(
5156 self.inner.inner(),
5157 ptr::null_mut(),
5158 ptr::null(),
5159 opts.inner,
5160 writer_opts.inner,
5161 cpath.as_ptr(),
5162 ));
5163 }
5164 Ok(())
5165 }
5166
5167 /// Stops the block cache trace started by [`Self::start_block_cache_trace`].
5168 ///
5169 /// Does nothing if no block cache trace is running.
5170 ///
5171 /// # Errors
5172 ///
5173 /// `EndBlockCacheTrace` always reports success, so this only fails if a future
5174 /// RocksDB gives it something to report. It returns a `Result` to keep that a
5175 /// non-breaking change.
5176 pub fn end_block_cache_trace(&self) -> Result<(), Error> {
5177 unsafe {
5178 ffi_try!(ffi::rocksdb_end_block_cache_trace(self.inner.inner()));
5179 }
5180 Ok(())
5181 }
5182
5183 /// Builds a replayer that replays the trace at `trace_path` against this DB.
5184 ///
5185 /// `column_families` must name every column family the trace touched, or
5186 /// replay fails with `Corruption: Invalid Column Family ID.`. An empty list
5187 /// means the default column family only.
5188 ///
5189 /// # Errors
5190 ///
5191 /// Returns the RocksDB error if the trace file cannot be opened or read.
5192 pub fn new_default_replayer<'cf, W, I, P>(
5193 &self,
5194 column_families: I,
5195 trace_path: P,
5196 ) -> Result<Replayer<'_>, Error>
5197 where
5198 W: AsColumnFamilyRef + 'cf,
5199 I: IntoIterator<Item = &'cf W>,
5200 P: AsRef<Path>,
5201 {
5202 let cpath = to_cpath(trace_path)?;
5203 unsafe {
5204 Replayer::create_default(
5205 self.inner.inner(),
5206 column_families,
5207 ptr::null_mut(),
5208 ptr::null(),
5209 &cpath,
5210 )
5211 }
5212 }
5213
5214 /// Returns a list of all table files with their level, start key
5215 /// and end key
5216 pub fn live_files(&self) -> Result<Vec<LiveFile>, Error> {
5217 unsafe {
5218 let livefiles_ptr = ffi::rocksdb_livefiles(self.inner.inner());
5219 if livefiles_ptr.is_null() {
5220 Err(Error::new("Could not get live files".to_owned()))
5221 } else {
5222 let files = LiveFile::from_rocksdb_livefiles_ptr(livefiles_ptr);
5223
5224 // destroy livefiles metadata(s)
5225 ffi::rocksdb_livefiles_destroy(livefiles_ptr);
5226
5227 // return
5228 Ok(files)
5229 }
5230 }
5231 }
5232
5233 /// Delete sst files whose keys are entirely in the given range.
5234 ///
5235 /// Could leave some keys in the range which are in files which are not
5236 /// entirely in the range.
5237 ///
5238 /// Note: L0 files are left regardless of whether they're in the range.
5239 ///
5240 /// SnapshotWithThreadModes before the delete might not see the data in the given range.
5241 pub fn delete_file_in_range<K: AsRef<[u8]>>(&self, from: K, to: K) -> Result<(), Error> {
5242 let from = from.as_ref();
5243 let to = to.as_ref();
5244 unsafe {
5245 ffi_try!(ffi::rocksdb_delete_file_in_range(
5246 self.inner.inner(),
5247 from.as_ptr() as *const c_char,
5248 from.len() as size_t,
5249 to.as_ptr() as *const c_char,
5250 to.len() as size_t,
5251 ));
5252 Ok(())
5253 }
5254 }
5255
5256 /// Same as `delete_file_in_range` but only for specific column family
5257 pub fn delete_file_in_range_cf<K: AsRef<[u8]>>(
5258 &self,
5259 cf: &impl AsColumnFamilyRef,
5260 from: K,
5261 to: K,
5262 ) -> Result<(), Error> {
5263 let from = from.as_ref();
5264 let to = to.as_ref();
5265 unsafe {
5266 ffi_try!(ffi::rocksdb_delete_file_in_range_cf(
5267 self.inner.inner(),
5268 cf.inner(),
5269 from.as_ptr() as *const c_char,
5270 from.len() as size_t,
5271 to.as_ptr() as *const c_char,
5272 to.len() as size_t,
5273 ));
5274 Ok(())
5275 }
5276 }
5277
5278 /// Request stopping background work, if wait is true wait until it's done.
5279 pub fn cancel_all_background_work(&self, wait: bool) {
5280 unsafe {
5281 ffi::rocksdb_cancel_all_background_work(self.inner.inner(), c_uchar::from(wait));
5282 }
5283 }
5284
5285 /// Marks the column family as dropped in RocksDB.
5286 ///
5287 /// Deliberately does not take ownership of the handle. Callers must take
5288 /// the handle out of their map first, so that only one caller can ever
5289 /// *destroy* a given handle, and must put it back if this fails: destroying
5290 /// it on failure would leave the column family still present in the DB with
5291 /// no reachable handle, so the only way to touch it again would be to
5292 /// reopen the database.
5293 ///
5294 /// Taking it out of the map does not make the caller the only *reader*. In
5295 /// `MultiThreaded` mode `cf_handle` clones the same `Arc`, so other threads
5296 /// can still hold a live `BoundColumnFamily` for this handle. That is fine:
5297 /// the refcount keeps the handle alive until the last of them is gone.
5298 fn mark_column_family_dropped(
5299 &self,
5300 cf_inner: *mut ffi::rocksdb_column_family_handle_t,
5301 ) -> Result<(), Error> {
5302 unsafe {
5303 ffi_try!(ffi::rocksdb_drop_column_family(
5304 self.inner.inner(),
5305 cf_inner
5306 ));
5307 }
5308 Ok(())
5309 }
5310
5311 /// Increase the full_history_ts of column family. The new ts_low value should
5312 /// be newer than current full_history_ts value.
5313 /// If another thread updates full_history_ts_low concurrently to a higher
5314 /// timestamp than the requested ts_low, a try again error will be returned.
5315 pub fn increase_full_history_ts_low<S: AsRef<[u8]>>(
5316 &self,
5317 cf: &impl AsColumnFamilyRef,
5318 ts: S,
5319 ) -> Result<(), Error> {
5320 let ts = ts.as_ref();
5321 unsafe {
5322 ffi_try!(ffi::rocksdb_increase_full_history_ts_low(
5323 self.inner.inner(),
5324 cf.inner(),
5325 ts.as_ptr() as *const c_char,
5326 ts.len() as size_t,
5327 ));
5328 Ok(())
5329 }
5330 }
5331
5332 /// Get current full_history_ts value.
5333 pub fn get_full_history_ts_low(&self, cf: &impl AsColumnFamilyRef) -> Result<Vec<u8>, Error> {
5334 unsafe {
5335 let mut ts_lowlen = 0;
5336 let ts = ffi_try!(ffi::rocksdb_get_full_history_ts_low(
5337 self.inner.inner(),
5338 cf.inner(),
5339 &raw mut ts_lowlen,
5340 ));
5341
5342 if ts.is_null() {
5343 Err(Error::new("Could not get full_history_ts_low".to_owned()))
5344 } else {
5345 let mut vec = vec![0; ts_lowlen];
5346 ptr::copy_nonoverlapping(ts.cast::<u8>(), vec.as_mut_ptr(), ts_lowlen);
5347 ffi::rocksdb_free(ts as *mut c_void);
5348 Ok(vec)
5349 }
5350 }
5351 }
5352
5353 /// Returns the DB identity. This is typically ASCII bytes, but that is not guaranteed.
5354 pub fn get_db_identity(&self) -> Result<Vec<u8>, Error> {
5355 unsafe {
5356 let mut length: usize = 0;
5357 let identity_ptr = ffi::rocksdb_get_db_identity(self.inner.inner(), &raw mut length);
5358 let identity_vec = raw_data(identity_ptr, length);
5359 ffi::rocksdb_free(identity_ptr as *mut c_void);
5360 // In RocksDB: get_db_identity copies a std::string so it should not fail, but
5361 // the API allows it to be overridden, so it might
5362 identity_vec.ok_or_else(|| Error::new("get_db_identity returned NULL".to_string()))
5363 }
5364 }
5365}
5366
5367impl<I: DBInner> DBCommon<SingleThreaded, I> {
5368 /// Creates column family with given name and options
5369 pub fn create_cf<N: AsRef<str>>(&mut self, name: N, opts: &Options) -> Result<(), Error> {
5370 let inner = self.create_inner_cf_handle(name.as_ref(), opts)?;
5371 self.cfs
5372 .cfs
5373 .insert(name.as_ref().to_string(), ColumnFamily { inner });
5374 Ok(())
5375 }
5376
5377 /// Creates a column family whose entries expire after `ttl`, on a DB that was
5378 /// opened with a TTL.
5379 ///
5380 /// [`ColumnFamilyDescriptor::new_with_ttl`] sets this at open time. This is the
5381 /// way to add such a family to a DB that is already open.
5382 ///
5383 /// # Errors
5384 ///
5385 /// Errors if the DB was not opened with [`DB::open_with_ttl`] or one of the
5386 /// `open_cf*_with_ttl` functions, since RocksDB would otherwise treat the handle
5387 /// as a type it is not. Also returns the RocksDB error if the family already
5388 /// exists or cannot be created, and errors if the name contains an interior NUL
5389 /// byte.
5390 pub fn create_cf_with_ttl<N: AsRef<str>>(
5391 &mut self,
5392 name: N,
5393 opts: &Options,
5394 ttl: ColumnFamilyTtl,
5395 ) -> Result<(), Error> {
5396 let inner = self.create_inner_cf_handle_with_ttl(name.as_ref(), opts, ttl)?;
5397 self.cfs
5398 .cfs
5399 .insert(name.as_ref().to_string(), ColumnFamily { inner });
5400 Ok(())
5401 }
5402
5403 /// Creates the named column families, all sharing `opts`.
5404 ///
5405 /// This saves one options file write over calling
5406 /// [`create_cf`](Self::create_cf) in a loop. It is not a saving on manifest
5407 /// writes, and it is not atomic.
5408 ///
5409 /// # Errors
5410 ///
5411 /// Returns the RocksDB error if a family already exists or cannot be created,
5412 /// and errors if a name contains an interior NUL byte.
5413 ///
5414 /// RocksDB creates the families in order and stops at the first failure, so on
5415 /// error the families named before the failing one already exist and stay that
5416 /// way. Those are recorded here as usual and can be used or dropped, so retrying
5417 /// with the same list will fail again on the ones that now exist.
5418 pub fn create_cfs<Iter, N>(&mut self, names: Iter, opts: &Options) -> Result<(), Error>
5419 where
5420 Iter: IntoIterator<Item = N>,
5421 N: AsRef<str>,
5422 {
5423 let names = convert_cf_names(names)?;
5424 let created = self.create_inner_cf_handles(&names, opts);
5425 for ((name, _), inner) in names.into_iter().zip(created.handles) {
5426 self.cfs.cfs.insert(name, ColumnFamily { inner });
5427 }
5428 match created.error {
5429 Some(err) => Err(err),
5430 None => Ok(()),
5431 }
5432 }
5433
5434 #[doc = include_str!("db_create_column_family_with_import.md")]
5435 pub fn create_column_family_with_import<N: AsRef<str>>(
5436 &mut self,
5437 options: &Options,
5438 column_family_name: N,
5439 import_options: &ImportColumnFamilyOptions,
5440 metadata: &ExportImportFilesMetaData,
5441 ) -> Result<(), Error> {
5442 let name = column_family_name.as_ref();
5443 let c_name = CString::new(name).map_err(|err| {
5444 Error::new(format!(
5445 "Failed to convert name to CString while importing column family: {err}"
5446 ))
5447 })?;
5448 let inner = unsafe {
5449 ffi_try!(ffi::rocksdb_create_column_family_with_import(
5450 self.inner.inner(),
5451 options.inner,
5452 c_name.as_ptr(),
5453 import_options.inner,
5454 metadata.inner
5455 ))
5456 };
5457 self.cfs
5458 .cfs
5459 .insert(column_family_name.as_ref().into(), ColumnFamily { inner });
5460 Ok(())
5461 }
5462
5463 /// Drops the column family with the given name
5464 pub fn drop_cf(&mut self, name: &str) -> Result<(), Error> {
5465 let Some(cf) = self.cfs.cfs.remove(name) else {
5466 return Err(Error::new(format!("Invalid column family: {name}")));
5467 };
5468 match self.mark_column_family_dropped(cf.inner) {
5469 // `cf` is dropped here. In single-threaded mode that destroys the
5470 // handle; in `MultiThreaded` mode it drops one `Arc` reference and
5471 // the handle is destroyed once the last `BoundColumnFamily` clone
5472 // handed out by `cf_handle` is gone.
5473 Ok(()) => Ok(()),
5474 Err(e) => {
5475 // The column family is still there, so put the handle back
5476 // rather than destroying the only way to reach it.
5477 self.cfs.cfs.insert(name.to_owned(), cf);
5478 Err(e)
5479 }
5480 }
5481 }
5482
5483 /// Returns the underlying column family handle
5484 pub fn cf_handle(&self, name: &str) -> Option<&ColumnFamily> {
5485 self.cfs.cfs.get(name)
5486 }
5487
5488 /// Returns the list of column families currently open.
5489 ///
5490 /// The order of names is unspecified and may vary between calls.
5491 pub fn cf_names(&self) -> Vec<String> {
5492 self.cfs.cfs.keys().cloned().collect()
5493 }
5494}
5495
5496impl<I: DBInner> DBCommon<MultiThreaded, I> {
5497 /// Creates column family with given name and options
5498 pub fn create_cf<N: AsRef<str>>(&self, name: N, opts: &Options) -> Result<(), Error> {
5499 // Note that we acquire the cfs lock before inserting: otherwise we might race
5500 // another caller who observed the handle as missing.
5501 let mut cfs = self.cfs.cfs.write();
5502 let inner = self.create_inner_cf_handle(name.as_ref(), opts)?;
5503 cfs.insert(
5504 name.as_ref().to_string(),
5505 Arc::new(UnboundColumnFamily { inner }),
5506 );
5507 Ok(())
5508 }
5509
5510 /// Creates a column family whose entries expire after `ttl`, on a DB that was
5511 /// opened with a TTL.
5512 ///
5513 /// [`ColumnFamilyDescriptor::new_with_ttl`] sets this at open time. This is the
5514 /// way to add such a family to a DB that is already open.
5515 ///
5516 /// # Errors
5517 ///
5518 /// Errors if the DB was not opened with [`DB::open_with_ttl`] or one of the
5519 /// `open_cf*_with_ttl` functions, since RocksDB would otherwise treat the handle
5520 /// as a type it is not. Also returns the RocksDB error if the family already
5521 /// exists or cannot be created, and errors if the name contains an interior NUL
5522 /// byte.
5523 pub fn create_cf_with_ttl<N: AsRef<str>>(
5524 &self,
5525 name: N,
5526 opts: &Options,
5527 ttl: ColumnFamilyTtl,
5528 ) -> Result<(), Error> {
5529 // Note that we acquire the cfs lock before inserting: otherwise we might race
5530 // another caller who observed the handle as missing.
5531 let mut cfs = self.cfs.cfs.write();
5532 let inner = self.create_inner_cf_handle_with_ttl(name.as_ref(), opts, ttl)?;
5533 cfs.insert(
5534 name.as_ref().to_string(),
5535 Arc::new(UnboundColumnFamily { inner }),
5536 );
5537 Ok(())
5538 }
5539
5540 /// Creates the named column families, all sharing `opts`.
5541 ///
5542 /// See [`DBCommon::create_cfs`](DBCommon::<SingleThreaded, I>::create_cfs),
5543 /// including the note that this is not atomic.
5544 ///
5545 /// # Errors
5546 ///
5547 /// Returns the RocksDB error if a family already exists or cannot be created,
5548 /// and errors if a name contains an interior NUL byte. On error the families
5549 /// named before the failing one already exist and are recorded here.
5550 pub fn create_cfs<Iter, N>(&self, names: Iter, opts: &Options) -> Result<(), Error>
5551 where
5552 Iter: IntoIterator<Item = N>,
5553 N: AsRef<str>,
5554 {
5555 let names = convert_cf_names(names)?;
5556 // Note that we acquire the cfs lock before creating: otherwise we might race
5557 // another caller who observed the handles as missing.
5558 let mut cfs = self.cfs.cfs.write();
5559 let created = self.create_inner_cf_handles(&names, opts);
5560 for ((name, _), inner) in names.into_iter().zip(created.handles) {
5561 cfs.insert(name, Arc::new(UnboundColumnFamily { inner }));
5562 }
5563 match created.error {
5564 Some(err) => Err(err),
5565 None => Ok(()),
5566 }
5567 }
5568
5569 #[doc = include_str!("db_create_column_family_with_import.md")]
5570 pub fn create_column_family_with_import<N: AsRef<str>>(
5571 &self,
5572 options: &Options,
5573 column_family_name: N,
5574 import_options: &ImportColumnFamilyOptions,
5575 metadata: &ExportImportFilesMetaData,
5576 ) -> Result<(), Error> {
5577 // Acquire CF lock upfront, before creating the CF, to avoid a race with concurrent creators
5578 let mut cfs = self.cfs.cfs.write();
5579 let name = column_family_name.as_ref();
5580 let c_name = CString::new(name).map_err(|err| {
5581 Error::new(format!(
5582 "Failed to convert name to CString while importing column family: {err}"
5583 ))
5584 })?;
5585 let inner = unsafe {
5586 ffi_try!(ffi::rocksdb_create_column_family_with_import(
5587 self.inner.inner(),
5588 options.inner,
5589 c_name.as_ptr(),
5590 import_options.inner,
5591 metadata.inner
5592 ))
5593 };
5594 cfs.insert(
5595 column_family_name.as_ref().to_string(),
5596 Arc::new(UnboundColumnFamily { inner }),
5597 );
5598 Ok(())
5599 }
5600
5601 /// Drops the column family with the given name by internally locking the inner column
5602 /// family map. This avoids needing `&mut self` reference
5603 pub fn drop_cf(&self, name: &str) -> Result<(), Error> {
5604 // Take the handle out under the write lock before touching RocksDB.
5605 // Looking it up under a read lock and removing it afterwards would let
5606 // two concurrent callers observe the same handle: the first would drop
5607 // and destroy it, and the second would then hand a freed pointer to
5608 // `rocksdb_drop_column_family`.
5609 let Some(cf) = self.cfs.cfs.write().remove(name) else {
5610 return Err(Error::new(format!("Invalid column family: {name}")));
5611 };
5612 match self.mark_column_family_dropped(cf.inner) {
5613 // `cf` is dropped here. In single-threaded mode that destroys the
5614 // handle; in `MultiThreaded` mode it drops one `Arc` reference and
5615 // the handle is destroyed once the last `BoundColumnFamily` clone
5616 // handed out by `cf_handle` is gone.
5617 Ok(()) => Ok(()),
5618 Err(e) => {
5619 // The column family is still there, so put the handle back
5620 // rather than destroying the only way to reach it.
5621 self.cfs.cfs.write().insert(name.to_owned(), cf);
5622 Err(e)
5623 }
5624 }
5625 }
5626
5627 /// Returns the underlying column family handle
5628 pub fn cf_handle(&'_ self, name: &str) -> Option<Arc<BoundColumnFamily<'_>>> {
5629 self.cfs
5630 .cfs
5631 .read()
5632 .get(name)
5633 .cloned()
5634 .map(UnboundColumnFamily::bound_column_family)
5635 }
5636
5637 /// Returns the list of column families currently open.
5638 ///
5639 /// The order of names is unspecified and may vary between calls.
5640 pub fn cf_names(&self) -> Vec<String> {
5641 self.cfs.cfs.read().keys().cloned().collect()
5642 }
5643}
5644
5645impl<T: ThreadMode, I: DBInner> Drop for DBCommon<T, I> {
5646 fn drop(&mut self) {
5647 self.cfs.drop_all_cfs_internal();
5648 }
5649}
5650
5651impl<T: ThreadMode, I: DBInner> fmt::Debug for DBCommon<T, I> {
5652 fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
5653 write!(f, "RocksDB {{ path: {} }}", self.path().display())
5654 }
5655}
5656
5657/// The metadata that describes a column family.
5658#[derive(Debug, Clone)]
5659pub struct ColumnFamilyMetaData {
5660 // The size of this column family in bytes, which is equal to the sum of
5661 // the file size of its "levels".
5662 pub size: u64,
5663 // The name of the column family.
5664 pub name: String,
5665 // The number of files in this column family.
5666 pub file_count: usize,
5667}
5668
5669/// The metadata that describes a SST file
5670#[derive(Debug, Clone)]
5671pub struct LiveFile {
5672 /// Name of the column family the file belongs to
5673 pub column_family_name: String,
5674 /// Name of the file
5675 pub name: String,
5676 /// The directory containing the file, without a trailing '/'. This could be
5677 /// a DB path, wal_dir, etc.
5678 pub directory: String,
5679 /// Size of the file
5680 pub size: usize,
5681 /// Level at which this file resides
5682 pub level: i32,
5683 /// Smallest user defined key in the file
5684 pub start_key: Option<Vec<u8>>,
5685 /// Largest user defined key in the file
5686 pub end_key: Option<Vec<u8>>,
5687 pub smallest_seqno: u64,
5688 pub largest_seqno: u64,
5689 /// Number of entries/alive keys in the file
5690 pub num_entries: u64,
5691 /// Number of deletions/tomb key(s) in the file
5692 pub num_deletions: u64,
5693}
5694
5695impl LiveFile {
5696 /// Create a `Vec<LiveFile>` from a `rocksdb_livefiles_t` pointer
5697 pub(crate) fn from_rocksdb_livefiles_ptr(
5698 files: *const ffi::rocksdb_livefiles_t,
5699 ) -> Vec<LiveFile> {
5700 unsafe {
5701 let n = ffi::rocksdb_livefiles_count(files);
5702
5703 let mut livefiles = Vec::with_capacity(n as usize);
5704 let mut key_size: usize = 0;
5705
5706 for i in 0..n {
5707 // rocksdb_livefiles_* returns pointers to strings, not copies
5708 let column_family_name =
5709 from_cstr_without_free(ffi::rocksdb_livefiles_column_family_name(files, i));
5710 let name = from_cstr_without_free(ffi::rocksdb_livefiles_name(files, i));
5711 let directory = from_cstr_without_free(ffi::rocksdb_livefiles_directory(files, i));
5712 let size = ffi::rocksdb_livefiles_size(files, i);
5713 let level = ffi::rocksdb_livefiles_level(files, i);
5714
5715 // get smallest key inside file
5716 let smallest_key = ffi::rocksdb_livefiles_smallestkey(files, i, &raw mut key_size);
5717 let smallest_key = raw_data(smallest_key, key_size);
5718
5719 // get largest key inside file
5720 let largest_key = ffi::rocksdb_livefiles_largestkey(files, i, &raw mut key_size);
5721 let largest_key = raw_data(largest_key, key_size);
5722
5723 livefiles.push(LiveFile {
5724 column_family_name,
5725 name,
5726 directory,
5727 size,
5728 level,
5729 start_key: smallest_key,
5730 end_key: largest_key,
5731 largest_seqno: ffi::rocksdb_livefiles_largest_seqno(files, i),
5732 smallest_seqno: ffi::rocksdb_livefiles_smallest_seqno(files, i),
5733 num_entries: ffi::rocksdb_livefiles_entries(files, i),
5734 num_deletions: ffi::rocksdb_livefiles_deletions(files, i),
5735 });
5736 }
5737
5738 livefiles
5739 }
5740 }
5741}
5742
5743struct LiveFileGuard(*mut rocksdb_livefile_t);
5744
5745impl LiveFileGuard {
5746 fn into_raw(mut self) -> *mut rocksdb_livefile_t {
5747 let ptr = self.0;
5748 self.0 = ptr::null_mut();
5749 ptr
5750 }
5751}
5752
5753impl Drop for LiveFileGuard {
5754 fn drop(&mut self) {
5755 if !self.0.is_null() {
5756 unsafe {
5757 rocksdb_livefile_destroy(self.0);
5758 }
5759 }
5760 }
5761}
5762
5763struct LiveFilesGuard(*mut rocksdb_livefiles_t);
5764
5765impl LiveFilesGuard {
5766 fn into_raw(mut self) -> *mut rocksdb_livefiles_t {
5767 let ptr = self.0;
5768 self.0 = ptr::null_mut();
5769 ptr
5770 }
5771}
5772
5773impl Drop for LiveFilesGuard {
5774 fn drop(&mut self) {
5775 if !self.0.is_null() {
5776 unsafe {
5777 rocksdb_livefiles_destroy(self.0);
5778 }
5779 }
5780 }
5781}
5782
5783/// Metadata returned as output from [`Checkpoint::export_column_family`][export_column_family] and
5784/// used as input to [`DB::create_column_family_with_import`].
5785///
5786/// [export_column_family]: crate::checkpoint::Checkpoint::export_column_family
5787#[derive(Debug)]
5788pub struct ExportImportFilesMetaData {
5789 pub(crate) inner: *mut ffi::rocksdb_export_import_files_metadata_t,
5790}
5791
5792impl ExportImportFilesMetaData {
5793 pub fn get_db_comparator_name(&self) -> String {
5794 unsafe {
5795 let c_name =
5796 ffi::rocksdb_export_import_files_metadata_get_db_comparator_name(self.inner);
5797 from_cstr_and_free(c_name)
5798 }
5799 }
5800
5801 pub fn set_db_comparator_name(&mut self, name: &str) {
5802 let c_name = CString::new(name.as_bytes()).unwrap();
5803 unsafe {
5804 ffi::rocksdb_export_import_files_metadata_set_db_comparator_name(
5805 self.inner,
5806 c_name.as_ptr(),
5807 );
5808 };
5809 }
5810
5811 pub fn get_files(&self) -> Vec<LiveFile> {
5812 unsafe {
5813 let livefiles_ptr = ffi::rocksdb_export_import_files_metadata_get_files(self.inner);
5814 let files = LiveFile::from_rocksdb_livefiles_ptr(livefiles_ptr);
5815 ffi::rocksdb_livefiles_destroy(livefiles_ptr);
5816 files
5817 }
5818 }
5819
5820 pub fn set_files(&mut self, files: &[LiveFile]) -> Result<(), Error> {
5821 // Use a non-null empty pointer for zero-length keys
5822 static EMPTY: [u8; 0] = [];
5823 let empty_ptr = EMPTY.as_ptr() as *const libc::c_char;
5824
5825 unsafe {
5826 let live_files = LiveFilesGuard(ffi::rocksdb_livefiles_create());
5827
5828 for file in files {
5829 let live_file = LiveFileGuard(ffi::rocksdb_livefile_create());
5830 ffi::rocksdb_livefile_set_level(live_file.0, file.level);
5831
5832 // SAFETY: C strings are copied inside the FFI layer so do not need to be kept alive
5833 let c_cf_name = CString::new(file.column_family_name.as_str()).map_err(|err| {
5834 Error::new(format!("Unable to convert column family to CString: {err}"))
5835 })?;
5836 ffi::rocksdb_livefile_set_column_family_name(live_file.0, c_cf_name.as_ptr());
5837
5838 let c_name = CString::new(file.name.as_str()).map_err(|err| {
5839 Error::new(format!("Unable to convert file name to CString: {err}"))
5840 })?;
5841 ffi::rocksdb_livefile_set_name(live_file.0, c_name.as_ptr());
5842
5843 let c_directory = CString::new(file.directory.as_str()).map_err(|err| {
5844 Error::new(format!("Unable to convert directory to CString: {err}"))
5845 })?;
5846 ffi::rocksdb_livefile_set_directory(live_file.0, c_directory.as_ptr());
5847
5848 ffi::rocksdb_livefile_set_size(live_file.0, file.size);
5849
5850 let (start_key_ptr, start_key_len) = match &file.start_key {
5851 None => (empty_ptr, 0),
5852 Some(key) => (key.as_ptr() as *const libc::c_char, key.len()),
5853 };
5854 ffi::rocksdb_livefile_set_smallest_key(live_file.0, start_key_ptr, start_key_len);
5855
5856 let (largest_key_ptr, largest_key_len) = match &file.end_key {
5857 None => (empty_ptr, 0),
5858 Some(key) => (key.as_ptr() as *const libc::c_char, key.len()),
5859 };
5860 ffi::rocksdb_livefile_set_largest_key(
5861 live_file.0,
5862 largest_key_ptr,
5863 largest_key_len,
5864 );
5865 ffi::rocksdb_livefile_set_smallest_seqno(live_file.0, file.smallest_seqno);
5866 ffi::rocksdb_livefile_set_largest_seqno(live_file.0, file.largest_seqno);
5867 ffi::rocksdb_livefile_set_num_entries(live_file.0, file.num_entries);
5868 ffi::rocksdb_livefile_set_num_deletions(live_file.0, file.num_deletions);
5869
5870 // moves ownership of live_files into live_file
5871 ffi::rocksdb_livefiles_add(live_files.0, live_file.into_raw());
5872 }
5873
5874 // moves ownership of live_files into inner
5875 ffi::rocksdb_export_import_files_metadata_set_files(self.inner, live_files.into_raw());
5876 Ok(())
5877 }
5878 }
5879}
5880
5881impl Default for ExportImportFilesMetaData {
5882 fn default() -> Self {
5883 let inner = unsafe { ffi::rocksdb_export_import_files_metadata_create() };
5884 assert!(
5885 !inner.is_null(),
5886 "Could not create rocksdb_export_import_files_metadata_t"
5887 );
5888
5889 Self { inner }
5890 }
5891}
5892
5893impl Drop for ExportImportFilesMetaData {
5894 fn drop(&mut self) {
5895 unsafe {
5896 ffi::rocksdb_export_import_files_metadata_destroy(self.inner);
5897 }
5898 }
5899}
5900
5901unsafe impl Send for ExportImportFilesMetaData {}
5902unsafe impl Sync for ExportImportFilesMetaData {}
5903
5904/// Converts a TTL to the `int` seconds count RocksDB's TTL API takes,
5905/// saturating instead of wrapping.
5906///
5907/// `Duration::as_secs` is a `u64`, so a plain `as i32` cast wraps: a TTL of
5908/// `Duration::from_secs(4_294_967_301)` (~136 years, i.e. "effectively never")
5909/// became `5`, and RocksDB then compaction-deleted the whole column family a few
5910/// seconds after the data was written.
5911///
5912/// Clamping to `i32::MAX` (~68 years) rather than mapping an over-large TTL to
5913/// RocksDB's never-expire sentinel (`ttl <= 0`, see `DBWithTTLImpl::IsStale`) is
5914/// deliberate: silently turning a finite TTL the caller asked for into "keep
5915/// forever" is a worse surprise than expiring it 68 years out, and `i32::MAX` is
5916/// the longest TTL the C API can express anyway.
5917fn ttl_to_seconds(ttl: Duration) -> c_int {
5918 c_int::try_from(ttl.as_secs()).unwrap_or(c_int::MAX)
5919}
5920
5921/// Resolves a column family's TTL against the TTL the DB was opened with.
5922///
5923/// [`ColumnFamilyTtl::Disabled`] maps to the longest TTL the C API can express rather
5924/// than RocksDB's never-expire sentinel, for the reason on [`ttl_to_seconds`].
5925fn cf_ttl_to_seconds(ttl: ColumnFamilyTtl, db_ttl: Duration) -> c_int {
5926 match ttl {
5927 ColumnFamilyTtl::Disabled => c_int::MAX,
5928 ColumnFamilyTtl::Duration(duration) => ttl_to_seconds(duration),
5929 ColumnFamilyTtl::SameAsDb => ttl_to_seconds(db_ttl),
5930 }
5931}
5932
5933/// Converts column family names to C strings, keeping the given order.
5934///
5935/// # Errors
5936///
5937/// Errors if a name contains an interior NUL byte.
5938fn convert_cf_names<I, N>(names: I) -> Result<Vec<(String, CString)>, Error>
5939where
5940 I: IntoIterator<Item = N>,
5941 N: AsRef<str>,
5942{
5943 names
5944 .into_iter()
5945 .map(|name| {
5946 let name = name.as_ref();
5947 let cname = CString::new(name).map_err(|err| {
5948 Error::new(format!(
5949 "Failed to convert column family name to CString: {err}"
5950 ))
5951 })?;
5952 Ok((name.to_owned(), cname))
5953 })
5954 .collect()
5955}
5956
5957/// The option count as the `int` the C API takes.
5958///
5959/// # Errors
5960///
5961/// Errors rather than truncating if there are more options than an `int` can count.
5962fn option_count(opts: &[(CString, CString)]) -> Result<c_int, Error> {
5963 c_int::try_from(opts.len())
5964 .map_err(|_| Error::new(format!("Too many options to set at once: {}", opts.len())))
5965}
5966
5967fn convert_options(opts: &[(&str, &str)]) -> Result<Vec<(CString, CString)>, Error> {
5968 opts.iter()
5969 .map(|(name, value)| {
5970 let cname = match CString::new(name.as_bytes()) {
5971 Ok(cname) => cname,
5972 Err(e) => return Err(Error::new(format!("Invalid option name `{e}`"))),
5973 };
5974 let cvalue = match CString::new(value.as_bytes()) {
5975 Ok(cvalue) => cvalue,
5976 Err(e) => return Err(Error::new(format!("Invalid option value: `{e}`"))),
5977 };
5978 Ok((cname, cvalue))
5979 })
5980 .collect()
5981}
5982
5983/// Borrows each key as the pointer and length pair the multi-get calls want.
5984///
5985/// The returned pointers borrow `keys`, so `keys` has to outlive them.
5986fn key_ptrs_and_sizes<K: AsRef<[u8]>>(keys: &[K]) -> (Vec<*const c_char>, Vec<usize>) {
5987 keys.iter()
5988 .map(|k| {
5989 let key = k.as_ref();
5990 (key.as_ptr().cast::<c_char>(), key.len())
5991 })
5992 .unzip()
5993}
5994
5995/// The five output arrays `rocksdb_multi_get_*_with_ts` fills in.
5996///
5997/// Kept together because they are only ever allocated, passed and consumed as a set,
5998/// and because the unsafe `set_len` after the call has to cover all five or none.
5999struct MultiGetTsOut {
6000 values: Vec<*mut c_char>,
6001 values_sizes: Vec<usize>,
6002 timestamps: Vec<*mut c_char>,
6003 timestamps_sizes: Vec<usize>,
6004 errors: Vec<*mut c_char>,
6005}
6006
6007impl MultiGetTsOut {
6008 fn with_capacity(n: usize) -> Self {
6009 Self {
6010 values: Vec::with_capacity(n),
6011 values_sizes: Vec::with_capacity(n),
6012 timestamps: Vec::with_capacity(n),
6013 timestamps_sizes: Vec::with_capacity(n),
6014 errors: Vec::with_capacity(n),
6015 }
6016 }
6017
6018 /// Publishes the `n` entries RocksDB wrote into the spare capacity.
6019 ///
6020 /// # Safety
6021 ///
6022 /// `n` must be the key count the call was given, and the call must have returned.
6023 /// RocksDB writes every one of the five arrays at every index, either a pointer it
6024 /// allocated or null (c.cc:2694-2711), so no element is left uninitialised.
6025 unsafe fn assume_filled(&mut self, n: usize) {
6026 unsafe {
6027 self.values.set_len(n);
6028 self.values_sizes.set_len(n);
6029 self.timestamps.set_len(n);
6030 self.timestamps_sizes.set_len(n);
6031 self.errors.set_len(n);
6032 }
6033 }
6034
6035 /// Takes ownership of every buffer RocksDB allocated and pairs it with its key.
6036 ///
6037 /// A null value with no error is a key that was not found. An error is reported
6038 /// with both buffers already null, so there is nothing to free on that path.
6039 fn into_results(self) -> Vec<Result<Option<TimestampedValue>, Error>> {
6040 self.values
6041 .into_iter()
6042 .zip(self.values_sizes)
6043 .zip(self.timestamps.into_iter().zip(self.timestamps_sizes))
6044 .zip(self.errors)
6045 .map(|(((value, vallen), (ts, tslen)), err)| {
6046 if !err.is_null() {
6047 return Err(convert_rocksdb_error(err));
6048 }
6049 if value.is_null() {
6050 return Ok(None);
6051 }
6052 // SAFETY: RocksDB allocated both with `CopyString` at the reported
6053 // lengths and nothing else frees them.
6054 unsafe {
6055 Ok(Some(TimestampedValue {
6056 value: CSlice::from_raw_parts(value, vallen),
6057 timestamp: CSlice::from_raw_parts(ts, tslen),
6058 }))
6059 }
6060 })
6061 .collect()
6062 }
6063}
6064
6065pub(crate) fn convert_values(
6066 values: Vec<*mut c_char>,
6067 values_sizes: Vec<usize>,
6068 errors: Vec<*mut c_char>,
6069) -> Vec<Result<Option<Vec<u8>>, Error>> {
6070 values
6071 .into_iter()
6072 .zip(values_sizes)
6073 .zip(errors)
6074 .map(|((v, s), e)| {
6075 if e.is_null() {
6076 let value = unsafe { crate::ffi_util::raw_data(v, s) };
6077 unsafe {
6078 ffi::rocksdb_free(v as *mut c_void);
6079 }
6080 Ok(value)
6081 } else {
6082 Err(convert_rocksdb_error(e))
6083 }
6084 })
6085 .collect()
6086}
6087
6088#[cfg(test)]
6089mod tests {
6090 use crate::{ColumnFamilyDescriptor, DB, Options};
6091
6092 /// One `rocksdb_create_iterators` call applies a single `ReadOptions` to
6093 /// every iterator it builds, and RocksDB's `DBIter` keeps raw `Slice*`
6094 /// into that object for the iterate bounds and read timestamps. So all the
6095 /// returned iterators have to keep the *same* options object alive, not a
6096 /// copy each and not none at all. Dropping it at the end of
6097 /// `create_iterators_cf` left those pointers dangling. See issue #660.
6098 #[test]
6099 fn raw_iterators_cf_share_one_live_readopts() {
6100 let dir = tempfile::Builder::new()
6101 .prefix("rocksdb-raw-iterators-cf-readopts")
6102 .tempdir()
6103 .unwrap();
6104
6105 let mut opts = Options::default();
6106 opts.create_if_missing(true);
6107 opts.create_missing_column_families(true);
6108 let db = DB::open_cf_descriptors(
6109 &opts,
6110 dir.path(),
6111 [
6112 ColumnFamilyDescriptor::new("first", Options::default()),
6113 ColumnFamilyDescriptor::new("second", Options::default()),
6114 ],
6115 )
6116 .unwrap();
6117
6118 let first = db.cf_handle("first").unwrap();
6119 let second = db.cf_handle("second").unwrap();
6120 let iterators = db.raw_iterators_cf([&first, &second]).unwrap();
6121 assert_eq!(iterators.len(), 2);
6122
6123 let shared = iterators[0].readopts_ptr();
6124 assert!(!shared.is_null());
6125 for iterator in &iterators {
6126 assert_eq!(
6127 iterator.readopts_ptr(),
6128 shared,
6129 "each iterator must hold the options object it was created from"
6130 );
6131 }
6132 }
6133}