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 /// Tries to clear a background error and resume writes.
1556 ///
1557 /// Returns `Ok` right away if no background error is stopping writes or
1558 /// background work. Recovering waits for running flushes and compactions to
1559 /// finish and then flushes every column family, so this can block for a
1560 /// while.
1561 ///
1562 /// Out of space is the only kind of I/O error this can recover from on the
1563 /// built-in filesystems: a full disk, or going over
1564 /// [`SstFileManager::set_max_allowed_space_usage`]. Other I/O errors are
1565 /// fatal with the default `paranoid_checks` and corruption is unrecoverable,
1566 /// so for those this returns the error and the DB has to be reopened.
1567 ///
1568 /// RocksDB normally recovers from out-of-space errors on its own. Every DB
1569 /// gets an [`SstFileManager`] even if you don't set one, and it retries
1570 /// every few seconds until enough space is free. While that is running this
1571 /// returns [`Busy`]. To recover manually instead, set the `&mut bool` passed
1572 /// to [`EventListener::on_error_recovery_begin`] to `false`, and only call
1573 /// this once space has been freed. A recovery flush that fails because the
1574 /// disk is still full can make the error fatal, and then only reopening
1575 /// helps.
1576 ///
1577 /// [`Options::set_max_bgerror_resume_count`] only applies to errors a
1578 /// filesystem marks as retryable, and the built-in ones never produce any
1579 /// that take that path.
1580 ///
1581 /// See also <https://github.com/facebook/rocksdb/wiki/Background-Error-Handling>.
1582 ///
1583 /// # Errors
1584 ///
1585 /// - [`Busy`] if automatic recovery is running. This doesn't wait for it.
1586 /// - The stored background error if it is fatal or unrecoverable.
1587 /// - The new error if the recovery flush fails, e.g. the disk is still full.
1588 ///
1589 /// [`Busy`]: crate::ErrorKind::Busy
1590 /// [`SstFileManager`]: crate::SstFileManager
1591 /// [`SstFileManager::set_max_allowed_space_usage`]: crate::SstFileManager::set_max_allowed_space_usage
1592 /// [`EventListener::on_error_recovery_begin`]: crate::event_listener::EventListener::on_error_recovery_begin
1593 pub fn resume(&self) -> Result<(), Error> {
1594 unsafe {
1595 ffi_try!(ffi::rust_rocksdb_resume(self.inner.inner()));
1596 }
1597
1598 Ok(())
1599 }
1600
1601 pub fn path(&self) -> &Path {
1602 self.path.as_path()
1603 }
1604
1605 /// Flushes the WAL buffer. If `sync` is set to `true`, also syncs
1606 /// the data to disk.
1607 pub fn flush_wal(&self, sync: bool) -> Result<(), Error> {
1608 unsafe {
1609 ffi_try!(ffi::rocksdb_flush_wal(
1610 self.inner.inner(),
1611 c_uchar::from(sync)
1612 ));
1613 }
1614 Ok(())
1615 }
1616
1617 /// Flushes the WAL buffer, taking the rate limiter priority as well as `sync`.
1618 ///
1619 /// [`flush_wal`](Self::flush_wal) covers the common case. Reach for this one to
1620 /// give the flush a priority other than the default, which decides how the WAL
1621 /// write is charged against a configured rate limiter.
1622 ///
1623 /// # Errors
1624 ///
1625 /// Returns the RocksDB error if the flush or the sync fails.
1626 pub fn flush_wal_with_options(&self, opts: &FlushWalOptions) -> Result<(), Error> {
1627 unsafe {
1628 ffi_try!(ffi::rocksdb_flush_wal_with_options(
1629 self.inner.inner(),
1630 opts.as_ptr()
1631 ));
1632 }
1633 Ok(())
1634 }
1635
1636 /// Stops background flushes and compactions and waits for the ones already
1637 /// running to finish.
1638 ///
1639 /// Writes are not blocked, so they keep filling memtables that cannot be
1640 /// flushed. Enough of them and the DB stalls or hits the stop trigger, so pair
1641 /// this with [`continue_background_work`](Self::continue_background_work) and
1642 /// keep the gap short.
1643 ///
1644 /// Calls nest. Background work resumes once as many `continue` calls have been
1645 /// made as `pause` calls.
1646 ///
1647 /// # Errors
1648 ///
1649 /// Returns the RocksDB error if the DB is shutting down.
1650 pub fn pause_background_work(&self) -> Result<(), Error> {
1651 unsafe {
1652 ffi_try!(ffi::rocksdb_pause_background_work(self.inner.inner()));
1653 }
1654 Ok(())
1655 }
1656
1657 /// Undoes one [`pause_background_work`](Self::pause_background_work) call.
1658 ///
1659 /// # Errors
1660 ///
1661 /// Returns the RocksDB error if background work was not paused, or if the DB is
1662 /// shutting down.
1663 pub fn continue_background_work(&self) -> Result<(), Error> {
1664 unsafe {
1665 ffi_try!(ffi::rocksdb_continue_background_work(self.inner.inner()));
1666 }
1667 Ok(())
1668 }
1669
1670 /// Undoes one [`disable_manual_compaction`](Self::disable_manual_compaction) call.
1671 ///
1672 /// This is a counter, so it takes as many calls here as there were calls there
1673 /// before manual compactions start running again.
1674 ///
1675 /// Pair the two. RocksDB decrements without a floor, and the assertion that would
1676 /// catch it is compiled out of this build, so calling this more often than
1677 /// `disable_manual_compaction` drives the counter below zero. That reads as
1678 /// not paused, which looks fine, but it means the next
1679 /// `disable_manual_compaction` only brings the counter back to zero and does not
1680 /// actually pause anything.
1681 pub fn enable_manual_compaction(&self) {
1682 unsafe { ffi::rocksdb_enable_manual_compaction(self.inner.inner()) }
1683 }
1684
1685 /// Cancels running manual compactions and makes later ones return immediately.
1686 ///
1687 /// Affects only manual compactions, so [`compact_range`](Self::compact_range),
1688 /// [`compact_files`](Self::compact_files) and the like. Automatic background
1689 /// compaction keeps going. Use it to get out of a long manual compaction during
1690 /// shutdown without waiting for it.
1691 ///
1692 /// This increments a counter, so it needs a matching
1693 /// [`enable_manual_compaction`](Self::enable_manual_compaction) call for each
1694 /// call here.
1695 pub fn disable_manual_compaction(&self) {
1696 unsafe { ffi::rocksdb_disable_manual_compaction(self.inner.inner()) }
1697 }
1698
1699 /// Reads every live SST and blob file and checks its block checksums.
1700 ///
1701 /// Reads the whole DB, so it is as slow as the data is large.
1702 ///
1703 /// # Errors
1704 ///
1705 /// Returns the RocksDB error naming the first corrupt file found.
1706 pub fn verify_checksum(&self) -> Result<(), Error> {
1707 unsafe {
1708 ffi_try!(ffi::rocksdb_verify_checksum(self.inner.inner()));
1709 }
1710 Ok(())
1711 }
1712
1713 /// Like [`verify_checksum`](Self::verify_checksum), reading through `readopts`.
1714 ///
1715 /// Worth setting when the default read path is not what you want for a full
1716 /// scan, for instance to keep the verification from filling the block cache or
1717 /// to give it a rate limiter priority.
1718 ///
1719 /// # Errors
1720 ///
1721 /// See [`verify_checksum`](Self::verify_checksum).
1722 pub fn verify_checksum_opt(&self, readopts: &ReadOptions) -> Result<(), Error> {
1723 unsafe {
1724 ffi_try!(ffi::rocksdb_verify_checksum_with_options(
1725 self.inner.inner(),
1726 readopts.inner
1727 ));
1728 }
1729 Ok(())
1730 }
1731
1732 /// Recomputes each live file's whole file checksum and compares it against the
1733 /// one recorded in the manifest.
1734 ///
1735 /// This is the file level check, as opposed to the per block one
1736 /// [`verify_checksum`](Self::verify_checksum) does. It requires the DB to have
1737 /// been written with a file checksum generator, see
1738 /// [`Options::set_file_checksum_gen_factory`](crate::Options::set_file_checksum_gen_factory).
1739 ///
1740 /// # Errors
1741 ///
1742 /// Returns the RocksDB error naming the first file whose checksum does not match
1743 /// what the manifest recorded. Also errors when no generator is configured, since
1744 /// then there is nothing recorded to compare against, rather than treating that
1745 /// as having nothing to check.
1746 pub fn verify_file_checksums(&self) -> Result<(), Error> {
1747 unsafe {
1748 ffi_try!(ffi::rocksdb_verify_file_checksums(self.inner.inner()));
1749 }
1750 Ok(())
1751 }
1752
1753 /// Like [`verify_file_checksums`](Self::verify_file_checksums), reading through
1754 /// `readopts`.
1755 ///
1756 /// # Errors
1757 ///
1758 /// See [`verify_file_checksums`](Self::verify_file_checksums).
1759 pub fn verify_file_checksums_opt(&self, readopts: &ReadOptions) -> Result<(), Error> {
1760 unsafe {
1761 ffi_try!(ffi::rocksdb_verify_file_checksums_with_options(
1762 self.inner.inner(),
1763 readopts.inner
1764 ));
1765 }
1766 Ok(())
1767 }
1768
1769 /// Marks the files overlapping `[start, end)` for compaction and returns
1770 /// without compacting anything.
1771 ///
1772 /// Background compaction picks the marked files up on its own schedule, so this
1773 /// returns as soon as the marking is done. That makes it the cheap way to hint
1774 /// that a range is worth compacting, as opposed to
1775 /// [`compact_range`](Self::compact_range), which does the work before it
1776 /// returns.
1777 ///
1778 /// `None` for either bound means unbounded in that direction. An empty slice is
1779 /// a real empty key, not the same thing.
1780 ///
1781 /// # Errors
1782 ///
1783 /// Returns the RocksDB error if the range cannot be marked.
1784 pub fn suggest_compact_range<S: AsRef<[u8]>, E: AsRef<[u8]>>(
1785 &self,
1786 start: Option<S>,
1787 end: Option<E>,
1788 ) -> Result<(), Error> {
1789 let (start, start_len) = optional_key_parts(start.as_ref());
1790 let (end, end_len) = optional_key_parts(end.as_ref());
1791 unsafe {
1792 ffi_try!(ffi::rocksdb_suggest_compact_range(
1793 self.inner.inner(),
1794 start,
1795 start_len,
1796 end,
1797 end_len
1798 ));
1799 }
1800 Ok(())
1801 }
1802
1803 /// Like [`suggest_compact_range`](Self::suggest_compact_range), for a single
1804 /// column family.
1805 ///
1806 /// # Errors
1807 ///
1808 /// See [`suggest_compact_range`](Self::suggest_compact_range).
1809 pub fn suggest_compact_range_cf<S: AsRef<[u8]>, E: AsRef<[u8]>>(
1810 &self,
1811 cf: &impl AsColumnFamilyRef,
1812 start: Option<S>,
1813 end: Option<E>,
1814 ) -> Result<(), Error> {
1815 let (start, start_len) = optional_key_parts(start.as_ref());
1816 let (end, end_len) = optional_key_parts(end.as_ref());
1817 unsafe {
1818 ffi_try!(ffi::rocksdb_suggest_compact_range_cf(
1819 self.inner.inner(),
1820 cf.inner(),
1821 start,
1822 start_len,
1823 end,
1824 end_len
1825 ));
1826 }
1827 Ok(())
1828 }
1829
1830 /// Reads a key along with the user-defined timestamp its value was written with.
1831 ///
1832 /// Only meaningful on a column family configured for user-defined timestamps, so
1833 /// one whose comparator carries a timestamp size. The plain
1834 /// [`get`](Self::get) reads the same value but throws the timestamp away, and
1835 /// there is no pinned equivalent of this call in the C API.
1836 ///
1837 /// `readopts` decides which timestamp is read. Without a read timestamp set on
1838 /// it RocksDB rejects the read rather than picking one, see
1839 /// [`ReadOptions::set_timestamp`](crate::ReadOptions::set_timestamp).
1840 ///
1841 /// # Errors
1842 ///
1843 /// Returns the RocksDB error if the read fails. A key that is not present is
1844 /// `Ok(None)`, not an error.
1845 pub fn get_with_ts<K: AsRef<[u8]>>(
1846 &self,
1847 key: K,
1848 readopts: &ReadOptions,
1849 ) -> Result<Option<TimestampedValue>, Error> {
1850 let key = key.as_ref();
1851 self.get_with_ts_impl(readopts, |db, ro, vallen, ts, tslen, err| unsafe {
1852 ffi::rocksdb_get_with_ts(
1853 db,
1854 ro,
1855 key.as_ptr().cast::<c_char>(),
1856 key.len(),
1857 vallen,
1858 ts,
1859 tslen,
1860 err,
1861 )
1862 })
1863 }
1864
1865 /// Like [`get_with_ts`](Self::get_with_ts), for a single column family.
1866 ///
1867 /// # Errors
1868 ///
1869 /// See [`get_with_ts`](Self::get_with_ts).
1870 pub fn get_cf_with_ts<K: AsRef<[u8]>>(
1871 &self,
1872 cf: &impl AsColumnFamilyRef,
1873 key: K,
1874 readopts: &ReadOptions,
1875 ) -> Result<Option<TimestampedValue>, Error> {
1876 let key = key.as_ref();
1877 let cf = cf.inner();
1878 self.get_with_ts_impl(readopts, |db, ro, vallen, ts, tslen, err| unsafe {
1879 ffi::rocksdb_get_cf_with_ts(
1880 db,
1881 ro,
1882 cf,
1883 key.as_ptr().cast::<c_char>(),
1884 key.len(),
1885 vallen,
1886 ts,
1887 tslen,
1888 err,
1889 )
1890 })
1891 }
1892
1893 /// Shared tail of the two timestamped gets.
1894 ///
1895 /// `call` is handed the out-params and returns the value pointer. Note that
1896 /// RocksDB only writes through `ts` when the read succeeds, so `ts` starts as
1897 /// null here and a miss leaves it that way rather than leaving it indeterminate
1898 /// (c.cc:2592-2603).
1899 fn get_with_ts_impl(
1900 &self,
1901 readopts: &ReadOptions,
1902 call: impl FnOnce(
1903 *mut ffi::rocksdb_t,
1904 *const ffi::rocksdb_readoptions_t,
1905 *mut usize,
1906 *mut *mut c_char,
1907 *mut usize,
1908 *mut *mut c_char,
1909 ) -> *mut c_char,
1910 ) -> Result<Option<TimestampedValue>, Error> {
1911 let mut vallen: usize = 0;
1912 let mut ts: *mut c_char = ptr::null_mut();
1913 let mut tslen: usize = 0;
1914 let mut err: *mut c_char = ptr::null_mut();
1915
1916 let value = call(
1917 self.inner.inner(),
1918 readopts.inner,
1919 &raw mut vallen,
1920 &raw mut ts,
1921 &raw mut tslen,
1922 &raw mut err,
1923 );
1924
1925 if !err.is_null() {
1926 // RocksDB reports a failure without allocating either output, but free
1927 // anything it did hand back rather than trusting that on an error path.
1928 unsafe {
1929 if !value.is_null() {
1930 ffi::rocksdb_free(value.cast::<c_void>());
1931 }
1932 if !ts.is_null() {
1933 ffi::rocksdb_free(ts.cast::<c_void>());
1934 }
1935 }
1936 return Err(convert_rocksdb_error(err));
1937 }
1938
1939 if value.is_null() {
1940 return Ok(None);
1941 }
1942
1943 // SAFETY: both pointers came from RocksDB's `CopyString`, so they are
1944 // `malloc`ed buffers of the reported length that nothing else frees, which is
1945 // exactly what `CSlice` takes over.
1946 unsafe {
1947 Ok(Some(TimestampedValue {
1948 value: CSlice::from_raw_parts(value, vallen),
1949 timestamp: CSlice::from_raw_parts(ts, tslen),
1950 }))
1951 }
1952 }
1953
1954 /// Reads many keys along with the timestamps their values were written with.
1955 ///
1956 /// One native batch, results in input order, one `Result` per key so a single
1957 /// bad key does not sink the batch. See [`get_with_ts`](Self::get_with_ts) for
1958 /// what the timestamp means and what `readopts` has to carry.
1959 pub fn multi_get_with_ts<K, I>(
1960 &self,
1961 keys: I,
1962 readopts: &ReadOptions,
1963 ) -> Vec<Result<Option<TimestampedValue>, Error>>
1964 where
1965 K: AsRef<[u8]>,
1966 I: IntoIterator<Item = K>,
1967 {
1968 let owned_keys: Vec<K> = keys.into_iter().collect();
1969 let (ptr_keys, keys_sizes) = key_ptrs_and_sizes(&owned_keys);
1970 let mut out = MultiGetTsOut::with_capacity(ptr_keys.len());
1971
1972 unsafe {
1973 ffi::rocksdb_multi_get_with_ts(
1974 self.inner.inner(),
1975 readopts.inner,
1976 ptr_keys.len(),
1977 ptr_keys.as_ptr(),
1978 keys_sizes.as_ptr(),
1979 out.values.as_mut_ptr(),
1980 out.values_sizes.as_mut_ptr(),
1981 out.timestamps.as_mut_ptr(),
1982 out.timestamps_sizes.as_mut_ptr(),
1983 out.errors.as_mut_ptr(),
1984 );
1985 out.assume_filled(ptr_keys.len());
1986 }
1987
1988 out.into_results()
1989 }
1990
1991 /// Like [`multi_get_with_ts`](Self::multi_get_with_ts), for one column family per
1992 /// key.
1993 pub fn multi_get_cf_with_ts<'c, K, I, W>(
1994 &self,
1995 keys: I,
1996 readopts: &ReadOptions,
1997 ) -> Vec<Result<Option<TimestampedValue>, Error>>
1998 where
1999 K: AsRef<[u8]>,
2000 I: IntoIterator<Item = (&'c W, K)>,
2001 W: AsColumnFamilyRef + 'c,
2002 {
2003 let (cfs, owned_keys): (Vec<_>, Vec<K>) = keys.into_iter().unzip();
2004 let cf_ptrs: Vec<*const ffi::rocksdb_column_family_handle_t> =
2005 cfs.iter().map(|cf| cf.inner().cast_const()).collect();
2006 let (ptr_keys, keys_sizes) = key_ptrs_and_sizes(&owned_keys);
2007 let mut out = MultiGetTsOut::with_capacity(ptr_keys.len());
2008
2009 unsafe {
2010 ffi::rocksdb_multi_get_cf_with_ts(
2011 self.inner.inner(),
2012 readopts.inner,
2013 cf_ptrs.as_ptr(),
2014 ptr_keys.len(),
2015 ptr_keys.as_ptr(),
2016 keys_sizes.as_ptr(),
2017 out.values.as_mut_ptr(),
2018 out.values_sizes.as_mut_ptr(),
2019 out.timestamps.as_mut_ptr(),
2020 out.timestamps_sizes.as_mut_ptr(),
2021 out.errors.as_mut_ptr(),
2022 );
2023 out.assume_filled(ptr_keys.len());
2024 }
2025
2026 out.into_results()
2027 }
2028
2029 /// Changes DB wide mutable options at runtime.
2030 ///
2031 /// Takes the same option names and string values the RocksDB configuration
2032 /// strings use, so `max_background_jobs` or `bytes_per_sync`. Only options
2033 /// RocksDB marks mutable at the DB level can be set, and the whole call is
2034 /// rejected if any name or value is not accepted.
2035 ///
2036 /// [`set_options`](Self::set_options) is the column family equivalent.
2037 ///
2038 /// # Aborts
2039 ///
2040 /// Some unparseable values take the process down instead of returning an error.
2041 /// See [`set_options`](Self::set_options) for the details and why this is not
2042 /// caught.
2043 ///
2044 /// # Errors
2045 ///
2046 /// Returns the RocksDB error if a name is unknown or the option is not
2047 /// changeable at runtime, and for an empty `opts`, which RocksDB rejects as
2048 /// `empty input`. Also errors if any name or value contains an interior NUL
2049 /// byte.
2050 pub fn set_db_options(&self, opts: &[(&str, &str)]) -> Result<(), Error> {
2051 let copts = convert_options(opts)?;
2052 let names: Vec<*const c_char> = copts.iter().map(|(n, _)| n.as_ptr()).collect();
2053 let values: Vec<*const c_char> = copts.iter().map(|(_, v)| v.as_ptr()).collect();
2054
2055 unsafe {
2056 ffi_try!(ffi::rocksdb_set_db_options(
2057 self.inner.inner(),
2058 option_count(&copts)?,
2059 names.as_ptr(),
2060 values.as_ptr(),
2061 ));
2062 }
2063 Ok(())
2064 }
2065
2066 /// Suspend deleting obsolete files. Compactions will continue to occur,
2067 /// but no obsolete files will be deleted. To resume file deletions, each
2068 /// call to disable_file_deletions() must be matched by a subsequent call to
2069 /// enable_file_deletions(). For more details, see enable_file_deletions().
2070 pub fn disable_file_deletions(&self) -> Result<(), Error> {
2071 unsafe {
2072 ffi_try!(ffi::rocksdb_disable_file_deletions(self.inner.inner()));
2073 }
2074 Ok(())
2075 }
2076
2077 /// Resume deleting obsolete files, following up on `disable_file_deletions()`.
2078 ///
2079 /// File deletions disabling and enabling is not controlled by a binary flag,
2080 /// instead it's represented as a counter to allow different callers to
2081 /// independently disable file deletion. Disabling file deletion can be
2082 /// critical for operations like making a backup. So the counter implementation
2083 /// makes the file deletion disabled as long as there is one caller requesting
2084 /// so, and only when every caller agrees to re-enable file deletion, it will
2085 /// be enabled. Two threads can call this method concurrently without
2086 /// synchronization -- i.e., file deletions will be enabled only after both
2087 /// threads call enable_file_deletions()
2088 pub fn enable_file_deletions(&self) -> Result<(), Error> {
2089 unsafe {
2090 ffi_try!(ffi::rocksdb_enable_file_deletions(self.inner.inner()));
2091 }
2092 Ok(())
2093 }
2094
2095 /// Flushes database memtables to SST files on the disk.
2096 pub fn flush_opt(&self, flushopts: &FlushOptions) -> Result<(), Error> {
2097 unsafe {
2098 ffi_try!(ffi::rocksdb_flush(self.inner.inner(), flushopts.inner));
2099 }
2100 Ok(())
2101 }
2102
2103 /// Flushes database memtables to SST files on the disk using default options.
2104 pub fn flush(&self) -> Result<(), Error> {
2105 DEFAULT_FLUSH_OPTS.with(|opts| self.flush_opt(opts))
2106 }
2107
2108 /// Flushes database memtables to SST files on the disk for a given column family.
2109 pub fn flush_cf_opt(
2110 &self,
2111 cf: &impl AsColumnFamilyRef,
2112 flushopts: &FlushOptions,
2113 ) -> Result<(), Error> {
2114 unsafe {
2115 ffi_try!(ffi::rocksdb_flush_cf(
2116 self.inner.inner(),
2117 flushopts.inner,
2118 cf.inner()
2119 ));
2120 }
2121 Ok(())
2122 }
2123
2124 /// Flushes multiple column families.
2125 ///
2126 /// If atomic flush is not enabled, it is equivalent to calling flush_cf multiple times.
2127 /// If atomic flush is enabled, it will flush all column families specified in `cfs` up to the latest sequence
2128 /// number at the time when flush is requested.
2129 pub fn flush_cfs_opt(
2130 &self,
2131 cfs: &[&impl AsColumnFamilyRef],
2132 opts: &FlushOptions,
2133 ) -> Result<(), Error> {
2134 let mut cfs = cfs.iter().map(|cf| cf.inner()).collect::<Vec<_>>();
2135 unsafe {
2136 ffi_try!(ffi::rocksdb_flush_cfs(
2137 self.inner.inner(),
2138 opts.inner,
2139 cfs.as_mut_ptr(),
2140 cfs.len() as libc::c_int,
2141 ));
2142 }
2143 Ok(())
2144 }
2145
2146 /// Flushes database memtables to SST files on the disk for a given column family using default
2147 /// options.
2148 pub fn flush_cf(&self, cf: &impl AsColumnFamilyRef) -> Result<(), Error> {
2149 DEFAULT_FLUSH_OPTS.with(|opts| self.flush_cf_opt(cf, opts))
2150 }
2151
2152 /// Return the bytes associated with a key value with read options. If you only intend to use
2153 /// the vector returned temporarily, consider using [`get_pinned_opt`](#method.get_pinned_opt)
2154 /// to avoid unnecessary memory copy.
2155 pub fn get_opt<K: AsRef<[u8]>>(
2156 &self,
2157 key: K,
2158 readopts: &ReadOptions,
2159 ) -> Result<Option<Vec<u8>>, Error> {
2160 self.get_pinned_opt(key, readopts)
2161 .map(|x| x.map(|v| v.as_ref().to_vec()))
2162 }
2163
2164 /// Return the bytes associated with a key value. If you only intend to use the vector returned
2165 /// temporarily, consider using [`get_pinned`](#method.get_pinned) to avoid unnecessary memory
2166 /// copy.
2167 pub fn get<K: AsRef<[u8]>>(&self, key: K) -> Result<Option<Vec<u8>>, Error> {
2168 DEFAULT_READ_OPTS.with(|opts| self.get_opt(key.as_ref(), opts))
2169 }
2170
2171 /// Return the bytes associated with a key value and the given column family with read options.
2172 /// If you only intend to use the vector returned temporarily, consider using
2173 /// [`get_pinned_cf_opt`](#method.get_pinned_cf_opt) to avoid unnecessary memory.
2174 pub fn get_cf_opt<K: AsRef<[u8]>>(
2175 &self,
2176 cf: &impl AsColumnFamilyRef,
2177 key: K,
2178 readopts: &ReadOptions,
2179 ) -> Result<Option<Vec<u8>>, Error> {
2180 self.get_pinned_cf_opt(cf, key, readopts)
2181 .map(|x| x.map(|v| v.as_ref().to_vec()))
2182 }
2183
2184 /// Return the bytes associated with a key value and the given column family. If you only
2185 /// intend to use the vector returned temporarily, consider using
2186 /// [`get_pinned_cf`](#method.get_pinned_cf) to avoid unnecessary memory.
2187 pub fn get_cf<K: AsRef<[u8]>>(
2188 &self,
2189 cf: &impl AsColumnFamilyRef,
2190 key: K,
2191 ) -> Result<Option<Vec<u8>>, Error> {
2192 DEFAULT_READ_OPTS.with(|opts| self.get_cf_opt(cf, key.as_ref(), opts))
2193 }
2194
2195 /// Return the value associated with a key using RocksDB's PinnableSlice
2196 /// so as to avoid unnecessary memory copy.
2197 pub fn get_pinned_opt<K: AsRef<[u8]>>(
2198 &'_ self,
2199 key: K,
2200 readopts: &ReadOptions,
2201 ) -> Result<Option<DBPinnableSlice<'_>>, Error> {
2202 if readopts.inner.is_null() {
2203 return Err(Error::new(
2204 "Unable to create RocksDB read options. This is a fairly trivial call, and its \
2205 failure may be indicative of a mis-compiled or mis-loaded RocksDB library."
2206 .to_owned(),
2207 ));
2208 }
2209
2210 let key = key.as_ref();
2211 unsafe {
2212 let val = ffi_try!(ffi::rocksdb_get_pinned(
2213 self.inner.inner(),
2214 readopts.inner,
2215 key.as_ptr() as *const c_char,
2216 key.len() as size_t,
2217 ));
2218 if val.is_null() {
2219 Ok(None)
2220 } else {
2221 Ok(Some(DBPinnableSlice::from_c(val)))
2222 }
2223 }
2224 }
2225
2226 /// Return the value associated with a key using RocksDB's PinnableSlice
2227 /// so as to avoid unnecessary memory copy. Similar to get_pinned_opt but
2228 /// leverages default options.
2229 pub fn get_pinned<K: AsRef<[u8]>>(
2230 &'_ self,
2231 key: K,
2232 ) -> Result<Option<DBPinnableSlice<'_>>, Error> {
2233 DEFAULT_READ_OPTS.with(|opts| self.get_pinned_opt(key, opts))
2234 }
2235
2236 /// Return the value associated with a key using RocksDB's PinnableSlice
2237 /// so as to avoid unnecessary memory copy. Similar to get_pinned_opt but
2238 /// allows specifying ColumnFamily
2239 pub fn get_pinned_cf_opt<K: AsRef<[u8]>>(
2240 &'_ self,
2241 cf: &impl AsColumnFamilyRef,
2242 key: K,
2243 readopts: &ReadOptions,
2244 ) -> Result<Option<DBPinnableSlice<'_>>, Error> {
2245 if readopts.inner.is_null() {
2246 return Err(Error::new(
2247 "Unable to create RocksDB read options. This is a fairly trivial call, and its \
2248 failure may be indicative of a mis-compiled or mis-loaded RocksDB library."
2249 .to_owned(),
2250 ));
2251 }
2252
2253 let key = key.as_ref();
2254 unsafe {
2255 let val = ffi_try!(ffi::rocksdb_get_pinned_cf(
2256 self.inner.inner(),
2257 readopts.inner,
2258 cf.inner(),
2259 key.as_ptr() as *const c_char,
2260 key.len() as size_t,
2261 ));
2262 if val.is_null() {
2263 Ok(None)
2264 } else {
2265 Ok(Some(DBPinnableSlice::from_c(val)))
2266 }
2267 }
2268 }
2269
2270 /// Return the value associated with a key using RocksDB's PinnableSlice
2271 /// so as to avoid unnecessary memory copy. Similar to get_pinned_cf_opt but
2272 /// leverages default options.
2273 pub fn get_pinned_cf<K: AsRef<[u8]>>(
2274 &'_ self,
2275 cf: &impl AsColumnFamilyRef,
2276 key: K,
2277 ) -> Result<Option<DBPinnableSlice<'_>>, Error> {
2278 DEFAULT_READ_OPTS.with(|opts| self.get_pinned_cf_opt(cf, key, opts))
2279 }
2280
2281 /// Read a value directly into a caller-provided buffer, avoiding memory allocation.
2282 ///
2283 /// This is the most efficient way to read values when you have a pre-allocated
2284 /// buffer. It completely avoids the allocation overhead of [`get`](#method.get)
2285 /// and even the pinning overhead of [`get_pinned`](#method.get_pinned).
2286 ///
2287 /// # Arguments
2288 ///
2289 /// * `key` - The key to look up
2290 /// * `buffer` - A mutable byte slice to write the value into. Can be empty if you
2291 /// only want to check if a key exists and get its value size.
2292 ///
2293 /// # Returns
2294 ///
2295 /// * `Ok(GetIntoBufferResult::NotFound)` - The key doesn't exist
2296 /// * `Ok(GetIntoBufferResult::Found(size))` - Value was copied into the buffer.
2297 /// `size` is the number of bytes written.
2298 /// * `Ok(GetIntoBufferResult::BufferTooSmall(size))` - The value exists but the buffer
2299 /// is too small. `size` is the actual value size needed. No data is written.
2300 /// * `Err(...)` - Database error occurred
2301 ///
2302 /// # Performance
2303 ///
2304 /// This method is ideal for high-throughput scenarios where you can reuse a buffer:
2305 ///
2306 /// ```ignore
2307 /// use rust_rocksdb::{DB, GetIntoBufferResult};
2308 ///
2309 /// let db: DB = /* open database */;
2310 /// let keys_to_lookup: Vec<&[u8]> = /* keys to look up */;
2311 /// let mut buffer = vec![0u8; 4096]; // Reusable buffer
2312 ///
2313 /// for key in keys_to_lookup {
2314 /// match db.get_into_buffer(key, &mut buffer).unwrap() {
2315 /// GetIntoBufferResult::Found(len) => {
2316 /// process_value(&buffer[..len]);
2317 /// }
2318 /// GetIntoBufferResult::BufferTooSmall(needed) => {
2319 /// buffer.resize(needed, 0);
2320 /// // Retry with larger buffer...
2321 /// }
2322 /// GetIntoBufferResult::NotFound => {}
2323 /// }
2324 /// }
2325 /// ```
2326 ///
2327 /// # Example
2328 ///
2329 /// ```
2330 /// use rust_rocksdb::{DB, GetIntoBufferResult};
2331 ///
2332 /// let tempdir = tempfile::Builder::new()
2333 /// .prefix("rocksdb_get_into_buffer")
2334 /// .tempdir()
2335 /// .unwrap();
2336 /// let db = DB::open_default(tempdir.path()).unwrap();
2337 /// db.put(b"key", b"value").unwrap();
2338 ///
2339 /// let mut buffer = [0u8; 100];
2340 /// match db.get_into_buffer(b"key", &mut buffer).unwrap() {
2341 /// GetIntoBufferResult::Found(size) => {
2342 /// assert_eq!(&buffer[..size], b"value");
2343 /// }
2344 /// GetIntoBufferResult::NotFound => panic!("expected value"),
2345 /// GetIntoBufferResult::BufferTooSmall(needed) => {
2346 /// panic!("buffer too small, need {} bytes", needed)
2347 /// }
2348 /// }
2349 /// ```
2350 pub fn get_into_buffer<K: AsRef<[u8]>>(
2351 &self,
2352 key: K,
2353 buffer: &mut [u8],
2354 ) -> Result<GetIntoBufferResult, Error> {
2355 DEFAULT_READ_OPTS.with(|opts| self.get_into_buffer_opt(key, buffer, opts))
2356 }
2357
2358 /// Read a value directly into a caller-provided buffer with custom read options.
2359 ///
2360 /// This is the same as [`get_into_buffer`](#method.get_into_buffer) but allows
2361 /// specifying custom [`ReadOptions`], such as setting a snapshot or fill cache behavior.
2362 ///
2363 /// See [`get_into_buffer`](#method.get_into_buffer) for full documentation.
2364 pub fn get_into_buffer_opt<K: AsRef<[u8]>>(
2365 &self,
2366 key: K,
2367 buffer: &mut [u8],
2368 readopts: &ReadOptions,
2369 ) -> Result<GetIntoBufferResult, Error> {
2370 if readopts.inner.is_null() {
2371 return Err(Error::new(
2372 "Unable to create RocksDB read options. This is a fairly trivial call, and its \
2373 failure may be indicative of a mis-compiled or mis-loaded RocksDB library."
2374 .to_owned(),
2375 ));
2376 }
2377
2378 let key = key.as_ref();
2379 let mut val_len: size_t = 0;
2380 let mut found: c_uchar = 0;
2381
2382 unsafe {
2383 let success = ffi_try!(ffi::rocksdb_get_into_buffer(
2384 self.inner.inner(),
2385 readopts.inner,
2386 key.as_ptr() as *const c_char,
2387 key.len() as size_t,
2388 buffer.as_mut_ptr() as *mut c_char,
2389 buffer.len() as size_t,
2390 &raw mut val_len,
2391 &raw mut found,
2392 ));
2393
2394 if found == 0 {
2395 Ok(GetIntoBufferResult::NotFound)
2396 } else if success != 0 {
2397 Ok(GetIntoBufferResult::Found(val_len))
2398 } else {
2399 Ok(GetIntoBufferResult::BufferTooSmall(val_len))
2400 }
2401 }
2402 }
2403
2404 /// Read a value from a column family directly into a caller-provided buffer.
2405 ///
2406 /// This is the column family variant of [`get_into_buffer`](#method.get_into_buffer).
2407 /// See that method for full documentation on the zero-allocation buffer API.
2408 ///
2409 /// # Arguments
2410 ///
2411 /// * `cf` - The column family to read from
2412 /// * `key` - The key to look up
2413 /// * `buffer` - A mutable byte slice to write the value into
2414 pub fn get_into_buffer_cf<K: AsRef<[u8]>>(
2415 &self,
2416 cf: &impl AsColumnFamilyRef,
2417 key: K,
2418 buffer: &mut [u8],
2419 ) -> Result<GetIntoBufferResult, Error> {
2420 DEFAULT_READ_OPTS.with(|opts| self.get_into_buffer_cf_opt(cf, key, buffer, opts))
2421 }
2422
2423 /// Read a value from a column family directly into a caller-provided buffer
2424 /// with custom read options.
2425 ///
2426 /// This is the column family variant of [`get_into_buffer_opt`](#method.get_into_buffer_opt).
2427 /// See [`get_into_buffer`](#method.get_into_buffer) for full documentation.
2428 pub fn get_into_buffer_cf_opt<K: AsRef<[u8]>>(
2429 &self,
2430 cf: &impl AsColumnFamilyRef,
2431 key: K,
2432 buffer: &mut [u8],
2433 readopts: &ReadOptions,
2434 ) -> Result<GetIntoBufferResult, Error> {
2435 if readopts.inner.is_null() {
2436 return Err(Error::new(
2437 "Unable to create RocksDB read options. This is a fairly trivial call, and its \
2438 failure may be indicative of a mis-compiled or mis-loaded RocksDB library."
2439 .to_owned(),
2440 ));
2441 }
2442
2443 let key = key.as_ref();
2444 let mut val_len: size_t = 0;
2445 let mut found: c_uchar = 0;
2446
2447 unsafe {
2448 let success = ffi_try!(ffi::rocksdb_get_into_buffer_cf(
2449 self.inner.inner(),
2450 readopts.inner,
2451 cf.inner(),
2452 key.as_ptr() as *const c_char,
2453 key.len() as size_t,
2454 buffer.as_mut_ptr() as *mut c_char,
2455 buffer.len() as size_t,
2456 &raw mut val_len,
2457 &raw mut found,
2458 ));
2459
2460 if found == 0 {
2461 Ok(GetIntoBufferResult::NotFound)
2462 } else if success != 0 {
2463 Ok(GetIntoBufferResult::Found(val_len))
2464 } else {
2465 Ok(GetIntoBufferResult::BufferTooSmall(val_len))
2466 }
2467 }
2468 }
2469
2470 /// Return the values associated with the given keys.
2471 pub fn multi_get<K, I>(&self, keys: I) -> Vec<Result<Option<Vec<u8>>, Error>>
2472 where
2473 K: AsRef<[u8]>,
2474 I: IntoIterator<Item = K>,
2475 {
2476 DEFAULT_READ_OPTS.with(|opts| self.multi_get_opt(keys, opts))
2477 }
2478
2479 /// Return the values associated with the given keys using read options.
2480 pub fn multi_get_opt<K, I>(
2481 &self,
2482 keys: I,
2483 readopts: &ReadOptions,
2484 ) -> Vec<Result<Option<Vec<u8>>, Error>>
2485 where
2486 K: AsRef<[u8]>,
2487 I: IntoIterator<Item = K>,
2488 {
2489 let owned_keys: Vec<K> = keys.into_iter().collect();
2490 let (ptr_keys, keys_sizes): (Vec<*const c_char>, Vec<usize>) = owned_keys
2491 .iter()
2492 .map(|k| {
2493 let key = k.as_ref();
2494 (key.as_ptr() as *const c_char, key.len())
2495 })
2496 .unzip();
2497
2498 let mut values: Vec<*mut c_char> = Vec::with_capacity(ptr_keys.len());
2499 let mut values_sizes: Vec<usize> = Vec::with_capacity(ptr_keys.len());
2500 let mut errors: Vec<*mut c_char> = Vec::with_capacity(ptr_keys.len());
2501 unsafe {
2502 ffi::rocksdb_multi_get(
2503 self.inner.inner(),
2504 readopts.inner,
2505 ptr_keys.len(),
2506 ptr_keys.as_ptr(),
2507 keys_sizes.as_ptr(),
2508 values.as_mut_ptr(),
2509 values_sizes.as_mut_ptr(),
2510 errors.as_mut_ptr(),
2511 );
2512 }
2513
2514 unsafe {
2515 values.set_len(ptr_keys.len());
2516 values_sizes.set_len(ptr_keys.len());
2517 errors.set_len(ptr_keys.len());
2518 }
2519
2520 convert_values(values, values_sizes, errors)
2521 }
2522
2523 /// Returns pinned values associated with the given keys using default read options.
2524 ///
2525 /// RocksDB processes the keys in one native batch. Results stay in input order.
2526 pub fn multi_get_pinned<K, I>(
2527 &'_ self,
2528 keys: I,
2529 ) -> Vec<Result<Option<DBPinnableSlice<'_>>, Error>>
2530 where
2531 K: AsRef<[u8]>,
2532 I: IntoIterator<Item = K>,
2533 {
2534 DEFAULT_READ_OPTS.with(|opts| self.multi_get_pinned_opt(keys, opts))
2535 }
2536
2537 /// Returns pinned values associated with the given keys using the provided read options.
2538 ///
2539 /// RocksDB processes the keys in one native batch. Results stay in input order.
2540 pub fn multi_get_pinned_opt<K, I>(
2541 &'_ self,
2542 keys: I,
2543 readopts: &ReadOptions,
2544 ) -> Vec<Result<Option<DBPinnableSlice<'_>>, Error>>
2545 where
2546 K: AsRef<[u8]>,
2547 I: IntoIterator<Item = K>,
2548 {
2549 let mut keys = keys.into_iter();
2550 let Some(first) = keys.next() else {
2551 return Vec::new();
2552 };
2553 // Decide before collecting. A single key does not benefit from the
2554 // native batch, and buying a key-slice vector, two result vectors and
2555 // a default column family handle to do one point lookup is a loss.
2556 let Some(second) = keys.next() else {
2557 return vec![self.get_pinned_opt(first.as_ref(), readopts)];
2558 };
2559 let mut owned_keys = Vec::with_capacity(2 + keys.size_hint().0);
2560 owned_keys.push(first);
2561 owned_keys.push(second);
2562 owned_keys.extend(keys);
2563 self.batched_multi_get_pinned_owned(&owned_keys, false, readopts)
2564 }
2565
2566 /// Returns pinned values associated with the given keys and column families
2567 /// using default read options.
2568 pub fn multi_get_pinned_cf<'a, 'b: 'a, K, I, W>(
2569 &'a self,
2570 keys: I,
2571 ) -> Vec<Result<Option<DBPinnableSlice<'a>>, Error>>
2572 where
2573 K: AsRef<[u8]>,
2574 I: IntoIterator<Item = (&'b W, K)>,
2575 W: 'b + AsColumnFamilyRef,
2576 {
2577 DEFAULT_READ_OPTS.with(|opts| self.multi_get_pinned_cf_opt(keys, opts))
2578 }
2579
2580 /// Returns pinned values associated with the given keys and column families
2581 /// using the provided read options.
2582 pub fn multi_get_pinned_cf_opt<'a, 'b: 'a, K, I, W>(
2583 &'a self,
2584 keys: I,
2585 readopts: &ReadOptions,
2586 ) -> Vec<Result<Option<DBPinnableSlice<'a>>, Error>>
2587 where
2588 K: AsRef<[u8]>,
2589 I: IntoIterator<Item = (&'b W, K)>,
2590 W: 'b + AsColumnFamilyRef,
2591 {
2592 keys.into_iter()
2593 .map(|(cf, k)| self.get_pinned_cf_opt(cf, k, readopts))
2594 .collect()
2595 }
2596
2597 /// Returns pinned values for default-column-family keys in one native batch.
2598 ///
2599 /// Set `sorted_input` only when keys are sorted according to the column
2600 /// family's comparator. Results stay in input order, including duplicates.
2601 pub fn batched_multi_get_pinned<K, I>(
2602 &'_ self,
2603 keys: I,
2604 sorted_input: bool,
2605 ) -> Vec<Result<Option<DBPinnableSlice<'_>>, Error>>
2606 where
2607 K: AsRef<[u8]>,
2608 I: IntoIterator<Item = K>,
2609 {
2610 DEFAULT_READ_OPTS.with(|opts| self.batched_multi_get_pinned_opt(keys, sorted_input, opts))
2611 }
2612
2613 /// Returns pinned values for default-column-family keys in one native batch
2614 /// using the provided read options.
2615 pub fn batched_multi_get_pinned_opt<K, I>(
2616 &'_ self,
2617 keys: I,
2618 sorted_input: bool,
2619 readopts: &ReadOptions,
2620 ) -> Vec<Result<Option<DBPinnableSlice<'_>>, Error>>
2621 where
2622 K: AsRef<[u8]>,
2623 I: IntoIterator<Item = K>,
2624 {
2625 let owned_keys: Vec<K> = keys.into_iter().collect();
2626 self.batched_multi_get_pinned_owned(&owned_keys, sorted_input, readopts)
2627 }
2628
2629 /// Returns pinned values for keys in one column family using one native batch.
2630 ///
2631 /// Set `sorted_input` only when keys are sorted according to the column
2632 /// family's comparator. Results stay in input order, including duplicates.
2633 pub fn batched_multi_get_pinned_cf<K, I>(
2634 &'_ self,
2635 cf: &impl AsColumnFamilyRef,
2636 keys: I,
2637 sorted_input: bool,
2638 ) -> Vec<Result<Option<DBPinnableSlice<'_>>, Error>>
2639 where
2640 K: AsRef<[u8]>,
2641 I: IntoIterator<Item = K>,
2642 {
2643 DEFAULT_READ_OPTS
2644 .with(|opts| self.batched_multi_get_pinned_cf_opt(cf, keys, sorted_input, opts))
2645 }
2646
2647 /// Returns pinned values for keys in one column family using one native batch
2648 /// and the provided read options.
2649 pub fn batched_multi_get_pinned_cf_opt<K, I>(
2650 &'_ self,
2651 cf: &impl AsColumnFamilyRef,
2652 keys: I,
2653 sorted_input: bool,
2654 readopts: &ReadOptions,
2655 ) -> Vec<Result<Option<DBPinnableSlice<'_>>, Error>>
2656 where
2657 K: AsRef<[u8]>,
2658 I: IntoIterator<Item = K>,
2659 {
2660 let owned_keys: Vec<K> = keys.into_iter().collect();
2661 let key_slices = Self::key_slices(&owned_keys);
2662 self.batched_multi_get_pinned_inner(cf.inner(), &key_slices, sorted_input, readopts)
2663 }
2664
2665 /// Returns one owner for all default-column-family pinned results.
2666 ///
2667 /// Values borrow from the returned batch, avoiding one native wrapper
2668 /// allocation and one destroy call per successful key.
2669 pub fn batched_multi_get_pinned_batch<K, I>(
2670 &'_ self,
2671 keys: I,
2672 sorted_input: bool,
2673 ) -> Result<DBPinnableBatch<'_>, Error>
2674 where
2675 K: AsRef<[u8]>,
2676 I: IntoIterator<Item = K>,
2677 {
2678 DEFAULT_READ_OPTS
2679 .with(|opts| self.batched_multi_get_pinned_batch_opt(keys, sorted_input, opts))
2680 }
2681
2682 /// Returns one owner for all default-column-family pinned results using
2683 /// the provided read options.
2684 pub fn batched_multi_get_pinned_batch_opt<K, I>(
2685 &'_ self,
2686 keys: I,
2687 sorted_input: bool,
2688 readopts: &ReadOptions,
2689 ) -> Result<DBPinnableBatch<'_>, Error>
2690 where
2691 K: AsRef<[u8]>,
2692 I: IntoIterator<Item = K>,
2693 {
2694 let owned_keys: Vec<K> = keys.into_iter().collect();
2695 let key_slices = Self::key_slices(&owned_keys);
2696 self.create_pinnable_batch(ptr::null_mut(), &key_slices, sorted_input, readopts)
2697 }
2698
2699 /// Returns one owner for all pinned results from one column family.
2700 pub fn batched_multi_get_pinned_batch_cf<K, I>(
2701 &'_ self,
2702 cf: &impl AsColumnFamilyRef,
2703 keys: I,
2704 sorted_input: bool,
2705 ) -> Result<DBPinnableBatch<'_>, Error>
2706 where
2707 K: AsRef<[u8]>,
2708 I: IntoIterator<Item = K>,
2709 {
2710 DEFAULT_READ_OPTS
2711 .with(|opts| self.batched_multi_get_pinned_batch_cf_opt(cf, keys, sorted_input, opts))
2712 }
2713
2714 /// Returns one owner for all pinned results from one column family using
2715 /// the provided read options.
2716 pub fn batched_multi_get_pinned_batch_cf_opt<K, I>(
2717 &'_ self,
2718 cf: &impl AsColumnFamilyRef,
2719 keys: I,
2720 sorted_input: bool,
2721 readopts: &ReadOptions,
2722 ) -> Result<DBPinnableBatch<'_>, Error>
2723 where
2724 K: AsRef<[u8]>,
2725 I: IntoIterator<Item = K>,
2726 {
2727 let owned_keys: Vec<K> = keys.into_iter().collect();
2728 let key_slices = Self::key_slices(&owned_keys);
2729 self.create_pinnable_batch(cf.inner(), &key_slices, sorted_input, readopts)
2730 }
2731
2732 /// Return the values associated with the given keys and column families.
2733 pub fn multi_get_cf<'a, 'b: 'a, K, I, W>(
2734 &'a self,
2735 keys: I,
2736 ) -> Vec<Result<Option<Vec<u8>>, Error>>
2737 where
2738 K: AsRef<[u8]>,
2739 I: IntoIterator<Item = (&'b W, K)>,
2740 W: 'b + AsColumnFamilyRef,
2741 {
2742 DEFAULT_READ_OPTS.with(|opts| self.multi_get_cf_opt(keys, opts))
2743 }
2744
2745 /// Return the values associated with the given keys and column families using read options.
2746 pub fn multi_get_cf_opt<'a, 'b: 'a, K, I, W>(
2747 &'a self,
2748 keys: I,
2749 readopts: &ReadOptions,
2750 ) -> Vec<Result<Option<Vec<u8>>, Error>>
2751 where
2752 K: AsRef<[u8]>,
2753 I: IntoIterator<Item = (&'b W, K)>,
2754 W: 'b + AsColumnFamilyRef,
2755 {
2756 let cfs_and_owned_keys: Vec<(&'b W, K)> = keys.into_iter().collect();
2757 let (ptr_keys, keys_sizes): (Vec<*const c_char>, Vec<usize>) = cfs_and_owned_keys
2758 .iter()
2759 .map(|(_, k)| {
2760 let key = k.as_ref();
2761 (key.as_ptr() as *const c_char, key.len())
2762 })
2763 .unzip();
2764 let ptr_cfs: Vec<*const ffi::rocksdb_column_family_handle_t> = cfs_and_owned_keys
2765 .iter()
2766 .map(|(c, _)| c.inner().cast_const())
2767 .collect();
2768 let mut values: Vec<*mut c_char> = Vec::with_capacity(ptr_keys.len());
2769 let mut values_sizes: Vec<usize> = Vec::with_capacity(ptr_keys.len());
2770 let mut errors: Vec<*mut c_char> = Vec::with_capacity(ptr_keys.len());
2771 unsafe {
2772 ffi::rocksdb_multi_get_cf(
2773 self.inner.inner(),
2774 readopts.inner,
2775 ptr_cfs.as_ptr(),
2776 ptr_keys.len(),
2777 ptr_keys.as_ptr(),
2778 keys_sizes.as_ptr(),
2779 values.as_mut_ptr(),
2780 values_sizes.as_mut_ptr(),
2781 errors.as_mut_ptr(),
2782 );
2783 }
2784
2785 unsafe {
2786 values.set_len(ptr_keys.len());
2787 values_sizes.set_len(ptr_keys.len());
2788 errors.set_len(ptr_keys.len());
2789 }
2790
2791 convert_values(values, values_sizes, errors)
2792 }
2793
2794 /// Return the values associated with the given keys and the specified column family
2795 /// where internally the read requests are processed in batch if block-based table
2796 /// SST format is used. It is a more optimized version of multi_get_cf.
2797 pub fn batched_multi_get_cf<'a, K, I>(
2798 &'_ self,
2799 cf: &impl AsColumnFamilyRef,
2800 keys: I,
2801 sorted_input: bool,
2802 ) -> Vec<Result<Option<DBPinnableSlice<'_>>, Error>>
2803 where
2804 K: AsRef<[u8]> + 'a + ?Sized,
2805 I: IntoIterator<Item = &'a K>,
2806 {
2807 DEFAULT_READ_OPTS.with(|opts| self.batched_multi_get_cf_opt(cf, keys, sorted_input, opts))
2808 }
2809
2810 /// Return the values associated with the given keys and the specified column family
2811 /// where internally the read requests are processed in batch if block-based table
2812 /// SST format is used. It is a more optimized version of multi_get_cf_opt.
2813 pub fn batched_multi_get_cf_opt<'a, K, I>(
2814 &'_ self,
2815 cf: &impl AsColumnFamilyRef,
2816 keys: I,
2817 sorted_input: bool,
2818 readopts: &ReadOptions,
2819 ) -> Vec<Result<Option<DBPinnableSlice<'_>>, Error>>
2820 where
2821 K: AsRef<[u8]> + 'a + ?Sized,
2822 I: IntoIterator<Item = &'a K>,
2823 {
2824 let key_slices: Vec<_> = keys
2825 .into_iter()
2826 .map(|k| {
2827 let k = k.as_ref();
2828 ffi::rocksdb_slice_t {
2829 data: k.as_ptr() as *const c_char,
2830 size: k.len(),
2831 }
2832 })
2833 .collect();
2834 self.batched_multi_get_pinned_inner(cf.inner(), &key_slices, sorted_input, readopts)
2835 }
2836
2837 /// Return the values associated with the given keys and the specified column family
2838 /// using an optimized slice-based API.
2839 ///
2840 /// This method uses RocksDB's optimized `rocksdb_batched_multi_get_cf_slice` C API,
2841 /// which takes a `rocksdb_slice_t` array directly. This eliminates the internal
2842 /// overhead of converting keys from separate pointer+size arrays to Slice objects.
2843 ///
2844 /// # Arguments
2845 ///
2846 /// * `cf` - The column family to read from
2847 /// * `keys` - An iterator of keys to look up
2848 /// * `sorted_input` - If `true`, indicates the keys are already sorted in ascending
2849 /// order, which allows RocksDB to skip internal sorting and improve performance.
2850 /// **Important**: If you pass `true` but keys are not sorted, results may be incorrect.
2851 ///
2852 /// # Returns
2853 ///
2854 /// A vector of results in the same order as the input keys. Each element is:
2855 /// - `Ok(Some(DBPinnableSlice))` if the key was found
2856 /// - `Ok(None)` if the key was not found
2857 /// - `Err(...)` if an error occurred for that key
2858 ///
2859 /// # Performance
2860 ///
2861 /// This is the fastest batch lookup method when:
2862 /// - You're looking up many keys (10+) from the same column family
2863 /// - You can pre-sort your keys (set `sorted_input = true`)
2864 /// - Block-based table format is used (default)
2865 ///
2866 /// For small numbers of keys, the overhead of batching may not be worth it.
2867 /// Consider using [`get_pinned_cf`](#method.get_pinned_cf) for single key lookups.
2868 ///
2869 /// # Example
2870 ///
2871 /// ```
2872 /// use rust_rocksdb::{DB, Options, ColumnFamilyDescriptor};
2873 ///
2874 /// let tempdir = tempfile::Builder::new().prefix("batch_slice").tempdir().unwrap();
2875 /// let mut opts = Options::default();
2876 /// opts.create_if_missing(true);
2877 /// opts.create_missing_column_families(true);
2878 /// let db = DB::open_cf_descriptors(&opts, tempdir.path(),
2879 /// vec![ColumnFamilyDescriptor::new("cf", Options::default())]).unwrap();
2880 ///
2881 /// let cf = db.cf_handle("cf").unwrap();
2882 /// db.put_cf(&cf, b"k1", b"v1").unwrap();
2883 /// db.put_cf(&cf, b"k2", b"v2").unwrap();
2884 ///
2885 /// // Keys are sorted, so we can set sorted_input = true
2886 /// let keys: Vec<&[u8]> = vec![b"k1", b"k2", b"k3"];
2887 /// let results = db.batched_multi_get_cf_slice(&cf, keys, true);
2888 ///
2889 /// assert!(results[0].as_ref().unwrap().is_some()); // k1 found
2890 /// assert!(results[1].as_ref().unwrap().is_some()); // k2 found
2891 /// assert!(results[2].as_ref().unwrap().is_none()); // k3 not found
2892 /// ```
2893 pub fn batched_multi_get_cf_slice<'a, K, I>(
2894 &'_ self,
2895 cf: &impl AsColumnFamilyRef,
2896 keys: I,
2897 sorted_input: bool,
2898 ) -> Vec<Result<Option<DBPinnableSlice<'_>>, Error>>
2899 where
2900 K: AsRef<[u8]> + 'a + ?Sized,
2901 I: IntoIterator<Item = &'a K>,
2902 {
2903 DEFAULT_READ_OPTS
2904 .with(|opts| self.batched_multi_get_cf_slice_opt(cf, keys, sorted_input, opts))
2905 }
2906
2907 /// Return the values associated with the given keys and the specified column family
2908 /// using an optimized slice-based API with custom read options.
2909 ///
2910 /// This is the same as [`batched_multi_get_cf_slice`](#method.batched_multi_get_cf_slice)
2911 /// but allows specifying custom [`ReadOptions`].
2912 ///
2913 /// See [`batched_multi_get_cf_slice`](#method.batched_multi_get_cf_slice) for full documentation.
2914 pub fn batched_multi_get_cf_slice_opt<'a, K, I>(
2915 &'_ self,
2916 cf: &impl AsColumnFamilyRef,
2917 keys: I,
2918 sorted_input: bool,
2919 readopts: &ReadOptions,
2920 ) -> Vec<Result<Option<DBPinnableSlice<'_>>, Error>>
2921 where
2922 K: AsRef<[u8]> + 'a + ?Sized,
2923 I: IntoIterator<Item = &'a K>,
2924 {
2925 // Convert keys to rocksdb_slice_t array
2926 let slices: Vec<ffi::rocksdb_slice_t> = keys
2927 .into_iter()
2928 .map(|k| {
2929 let k = k.as_ref();
2930 ffi::rocksdb_slice_t {
2931 data: k.as_ptr() as *const c_char,
2932 size: k.len(),
2933 }
2934 })
2935 .collect();
2936
2937 self.batched_multi_get_pinned_inner(cf.inner(), &slices, sorted_input, readopts)
2938 }
2939
2940 fn key_slices<K: AsRef<[u8]>>(keys: &[K]) -> Vec<ffi::rocksdb_slice_t> {
2941 keys.iter()
2942 .map(|key| {
2943 let key = key.as_ref();
2944 ffi::rocksdb_slice_t {
2945 data: key.as_ptr() as *const c_char,
2946 size: key.len(),
2947 }
2948 })
2949 .collect()
2950 }
2951
2952 fn batched_multi_get_pinned_owned<'a, K: AsRef<[u8]>>(
2953 &'a self,
2954 keys: &[K],
2955 sorted_input: bool,
2956 readopts: &ReadOptions,
2957 ) -> Vec<Result<Option<DBPinnableSlice<'a>>, Error>> {
2958 let key_slices = Self::key_slices(keys);
2959 if key_slices.is_empty() {
2960 return Vec::new();
2961 }
2962 let default_cf = OwnedColumnFamilyHandle::default_for(self.inner.inner());
2963 self.batched_multi_get_pinned_inner(default_cf.inner, &key_slices, sorted_input, readopts)
2964 }
2965
2966 fn create_pinnable_batch<'a>(
2967 &'a self,
2968 cf: *mut ffi::rocksdb_column_family_handle_t,
2969 keys: &[ffi::rocksdb_slice_t],
2970 sorted_input: bool,
2971 readopts: &ReadOptions,
2972 ) -> Result<DBPinnableBatch<'a>, Error> {
2973 let batch = unsafe {
2974 ffi_try!(ffi::rust_rocksdb_batched_multi_get_pinned(
2975 self.inner.inner(),
2976 readopts.inner,
2977 cf,
2978 keys.len(),
2979 keys.as_ptr(),
2980 c_uchar::from(sorted_input),
2981 ))
2982 };
2983 if batch.is_null() {
2984 // `ffi_try!` only returns early when the extension set `errptr`.
2985 // A null batch with no error means the extension could not even
2986 // allocate the message, so report it instead of unwrapping.
2987 return Err(Error::new(
2988 "rust_rocksdb_batched_multi_get_pinned returned no batch".to_owned(),
2989 ));
2990 }
2991 // SAFETY: The extension returns a uniquely owned batch.
2992 Ok(unsafe { DBPinnableBatch::from_c(batch) })
2993 }
2994
2995 fn batched_multi_get_pinned_inner<'a>(
2996 &'a self,
2997 cf: *mut ffi::rocksdb_column_family_handle_t,
2998 keys: &[ffi::rocksdb_slice_t],
2999 sorted_input: bool,
3000 readopts: &ReadOptions,
3001 ) -> Vec<Result<Option<DBPinnableSlice<'a>>, Error>> {
3002 if keys.is_empty() {
3003 return Vec::new();
3004 }
3005 let output = match self.execute_batched_multi_get(cf, keys, sorted_input, readopts) {
3006 Ok(output) => output,
3007 Err(error) => {
3008 let message = error.to_string();
3009 return (0..keys.len())
3010 .map(|_| Err(Error::new(message.clone())))
3011 .collect();
3012 }
3013 };
3014 output
3015 .values
3016 .into_iter()
3017 .zip(output.errors)
3018 .map(|(value, error)| unsafe { Self::convert_pinned_result(value, error) })
3019 .collect()
3020 }
3021
3022 fn execute_batched_multi_get(
3023 &self,
3024 cf: *mut ffi::rocksdb_column_family_handle_t,
3025 keys: &[ffi::rocksdb_slice_t],
3026 sorted_input: bool,
3027 readopts: &ReadOptions,
3028 ) -> Result<PinnedMultiGetOutput, Error> {
3029 let mut pinned_values = vec![ptr::null_mut(); keys.len()];
3030 let mut errors = vec![ptr::null_mut(); keys.len()];
3031 unsafe {
3032 ffi_try!(ffi::rust_rocksdb_batched_multi_get_cf_slice_safe(
3033 self.inner.inner(),
3034 readopts.inner,
3035 cf,
3036 keys.len(),
3037 keys.as_ptr(),
3038 pinned_values.as_mut_ptr(),
3039 errors.as_mut_ptr(),
3040 c_uchar::from(sorted_input),
3041 ));
3042 }
3043 Ok(PinnedMultiGetOutput {
3044 values: pinned_values,
3045 errors,
3046 })
3047 }
3048
3049 /// Converts one result returned by `rocksdb_batched_multi_get_cf_slice`.
3050 ///
3051 /// # Safety
3052 ///
3053 /// `value` must be null or an owned pinnable slice. `error` must be null or
3054 /// an owned RocksDB error string.
3055 unsafe fn convert_pinned_result<'a>(
3056 value: *mut ffi::rocksdb_pinnableslice_t,
3057 error: *mut c_char,
3058 ) -> Result<Option<DBPinnableSlice<'a>>, Error> {
3059 if error.is_null() {
3060 return Ok((!value.is_null()).then(|| unsafe { DBPinnableSlice::from_c(value) }));
3061 }
3062 if !value.is_null() {
3063 unsafe {
3064 ffi::rocksdb_pinnableslice_destroy(value);
3065 }
3066 }
3067 Err(convert_rocksdb_error(error))
3068 }
3069
3070 /// Returns `false` if the given key definitely doesn't exist in the database, otherwise returns
3071 /// `true`. This function uses default `ReadOptions`.
3072 pub fn key_may_exist<K: AsRef<[u8]>>(&self, key: K) -> bool {
3073 DEFAULT_READ_OPTS.with(|opts| self.key_may_exist_opt(key, opts))
3074 }
3075
3076 /// Returns `false` if the given key definitely doesn't exist in the database, otherwise returns
3077 /// `true`.
3078 pub fn key_may_exist_opt<K: AsRef<[u8]>>(&self, key: K, readopts: &ReadOptions) -> bool {
3079 let key = key.as_ref();
3080 unsafe {
3081 0 != ffi::rocksdb_key_may_exist(
3082 self.inner.inner(),
3083 readopts.inner,
3084 key.as_ptr() as *const c_char,
3085 key.len() as size_t,
3086 ptr::null_mut(), /*value*/
3087 ptr::null_mut(), /*val_len*/
3088 ptr::null(), /*timestamp*/
3089 0, /*timestamp_len*/
3090 ptr::null_mut(), /*value_found*/
3091 )
3092 }
3093 }
3094
3095 /// Returns `false` if the given key definitely doesn't exist in the specified column family,
3096 /// otherwise returns `true`. This function uses default `ReadOptions`.
3097 pub fn key_may_exist_cf<K: AsRef<[u8]>>(&self, cf: &impl AsColumnFamilyRef, key: K) -> bool {
3098 DEFAULT_READ_OPTS.with(|opts| self.key_may_exist_cf_opt(cf, key, opts))
3099 }
3100
3101 /// Returns `false` if the given key definitely doesn't exist in the specified column family,
3102 /// otherwise returns `true`.
3103 pub fn key_may_exist_cf_opt<K: AsRef<[u8]>>(
3104 &self,
3105 cf: &impl AsColumnFamilyRef,
3106 key: K,
3107 readopts: &ReadOptions,
3108 ) -> bool {
3109 let key = key.as_ref();
3110 0 != unsafe {
3111 ffi::rocksdb_key_may_exist_cf(
3112 self.inner.inner(),
3113 readopts.inner,
3114 cf.inner(),
3115 key.as_ptr() as *const c_char,
3116 key.len() as size_t,
3117 ptr::null_mut(), /*value*/
3118 ptr::null_mut(), /*val_len*/
3119 ptr::null(), /*timestamp*/
3120 0, /*timestamp_len*/
3121 ptr::null_mut(), /*value_found*/
3122 )
3123 }
3124 }
3125
3126 /// If the key definitely does not exist in the database, then this method
3127 /// returns `(false, None)`, else `(true, None)` if it may.
3128 /// If the key is found in memory, then it returns `(true, Some<CSlice>)`.
3129 ///
3130 /// This check is potentially lighter-weight than calling `get()`. One way
3131 /// to make this lighter weight is to avoid doing any IOs.
3132 pub fn key_may_exist_cf_opt_value<K: AsRef<[u8]>>(
3133 &self,
3134 cf: &impl AsColumnFamilyRef,
3135 key: K,
3136 readopts: &ReadOptions,
3137 ) -> (bool, Option<CSlice>) {
3138 let key = key.as_ref();
3139 let mut val: *mut c_char = ptr::null_mut();
3140 let mut val_len: usize = 0;
3141 let mut value_found: c_uchar = 0;
3142 let may_exists = 0
3143 != unsafe {
3144 ffi::rocksdb_key_may_exist_cf(
3145 self.inner.inner(),
3146 readopts.inner,
3147 cf.inner(),
3148 key.as_ptr() as *const c_char,
3149 key.len() as size_t,
3150 &raw mut val, /*value*/
3151 &raw mut val_len, /*val_len*/
3152 ptr::null(), /*timestamp*/
3153 0, /*timestamp_len*/
3154 &raw mut value_found, /*value_found*/
3155 )
3156 };
3157 // The value is only allocated (using malloc) and returned if it is found and
3158 // value_found isn't NULL. In that case the user is responsible for freeing it.
3159 if may_exists && value_found != 0 {
3160 (
3161 may_exists,
3162 Some(unsafe { CSlice::from_raw_parts(val, val_len) }),
3163 )
3164 } else {
3165 (may_exists, None)
3166 }
3167 }
3168
3169 fn create_inner_cf_handle(
3170 &self,
3171 name: impl CStrLike,
3172 opts: &Options,
3173 ) -> Result<*mut ffi::rocksdb_column_family_handle_t, Error> {
3174 let cf_name = name.bake().map_err(|err| {
3175 Error::new(format!(
3176 "Failed to convert path to CString when creating cf: {err}"
3177 ))
3178 })?;
3179
3180 // Can't use ffi_try: rocksdb_create_column_family has a bug where it allocates a
3181 // result that needs to be freed on error
3182 let mut err: *mut ::libc::c_char = ::std::ptr::null_mut();
3183 let cf_handle = unsafe {
3184 ffi::rocksdb_create_column_family(
3185 self.inner.inner(),
3186 opts.inner,
3187 cf_name.as_ptr(),
3188 &raw mut err,
3189 )
3190 };
3191 if !err.is_null() {
3192 if !cf_handle.is_null() {
3193 unsafe { ffi::rocksdb_column_family_handle_destroy(cf_handle) };
3194 }
3195 return Err(convert_rocksdb_error(err));
3196 }
3197 Ok(cf_handle)
3198 }
3199
3200 /// Creates every named column family in one call.
3201 ///
3202 /// This is not atomic. RocksDB creates the families one at a time and stops at
3203 /// the first failure, so the ones before it are already committed and stay that
3204 /// way. Only the options file write at the end is shared, which is the whole
3205 /// saving over calling [`create_cf`](Self::create_cf) in a loop.
3206 ///
3207 /// Returns the handles that were created, in the order the names were given,
3208 /// alongside the error that stopped the rest. The caller owns those handles and
3209 /// has to record them in its column family map even when there is an error,
3210 /// because the families exist either way.
3211 fn create_inner_cf_handles(
3212 &self,
3213 names: &[(String, CString)],
3214 opts: &Options,
3215 ) -> CreatedCfHandles {
3216 let name_ptrs: Vec<*const c_char> = names.iter().map(|(_, name)| name.as_ptr()).collect();
3217 let Ok(count) = c_int::try_from(names.len()) else {
3218 return CreatedCfHandles::failed(Error::new(format!(
3219 "Too many column families to create at once: {}",
3220 names.len()
3221 )));
3222 };
3223
3224 let mut len: usize = 0;
3225 let mut err: *mut ::libc::c_char = ::std::ptr::null_mut();
3226 // Can't use ffi_try: like rocksdb_create_column_family, this allocates a result
3227 // that needs to be freed on error.
3228 let list = unsafe {
3229 ffi::rocksdb_create_column_families(
3230 self.inner.inner(),
3231 opts.inner,
3232 count,
3233 name_ptrs.as_ptr(),
3234 &raw mut len,
3235 &raw mut err,
3236 )
3237 };
3238
3239 // Two allocations come back: the array, and a handle per family. Freeing the
3240 // array does not touch the handles, so take copies of them and release the
3241 // array on its own.
3242 let handles = if list.is_null() {
3243 Vec::new()
3244 } else {
3245 let handles = unsafe { std::slice::from_raw_parts(list, len) }.to_vec();
3246 unsafe { ffi::rocksdb_create_column_families_destroy(list) };
3247 handles
3248 };
3249
3250 if !err.is_null() {
3251 // Hand the handles back even though this failed. The families they name
3252 // were committed before the failing one and are still there, so the caller
3253 // needs them to be able to use or drop those families.
3254 return CreatedCfHandles {
3255 handles,
3256 error: Some(convert_rocksdb_error(err)),
3257 };
3258 }
3259
3260 let created = handles.len();
3261 let error = (created != names.len()).then(|| {
3262 Error::new(format!(
3263 "Expected {} column family handles, got {created}",
3264 names.len(),
3265 ))
3266 });
3267
3268 CreatedCfHandles { handles, error }
3269 }
3270
3271 /// Creates one column family whose entries expire after `ttl`.
3272 ///
3273 /// Only valid on a DB opened with a TTL. The C function casts the handle to
3274 /// `DBWithTTL` unchecked, so the caller has to have proven that already.
3275 fn create_inner_cf_handle_with_ttl(
3276 &self,
3277 name: impl CStrLike,
3278 opts: &Options,
3279 ttl: ColumnFamilyTtl,
3280 ) -> Result<*mut ffi::rocksdb_column_family_handle_t, Error> {
3281 let Some(db_ttl) = self.opened_with_ttl else {
3282 return Err(Error::new(
3283 "create_cf_with_ttl requires a database opened with DB::open_with_ttl \
3284 or one of the open_cf*_with_ttl functions"
3285 .to_owned(),
3286 ));
3287 };
3288
3289 let cf_name = name.bake().map_err(|err| {
3290 Error::new(format!(
3291 "Failed to convert name to CString when creating cf with ttl: {err}"
3292 ))
3293 })?;
3294 let ttl = cf_ttl_to_seconds(ttl, db_ttl);
3295
3296 // Can't use ffi_try: like rocksdb_create_column_family, this allocates a result
3297 // that needs to be freed on error.
3298 let mut err: *mut ::libc::c_char = ::std::ptr::null_mut();
3299 let cf_handle = unsafe {
3300 ffi::rocksdb_create_column_family_with_ttl(
3301 self.inner.inner(),
3302 opts.inner,
3303 cf_name.as_ptr(),
3304 ttl,
3305 &raw mut err,
3306 )
3307 };
3308 if !err.is_null() {
3309 if !cf_handle.is_null() {
3310 unsafe { ffi::rocksdb_column_family_handle_destroy(cf_handle) };
3311 }
3312 return Err(convert_rocksdb_error(err));
3313 }
3314 Ok(cf_handle)
3315 }
3316
3317 pub fn iterator<'a: 'b, 'b>(
3318 &'a self,
3319 mode: IteratorMode,
3320 ) -> DBIteratorWithThreadMode<'b, Self> {
3321 let readopts = ReadOptions::default();
3322 self.iterator_opt(mode, readopts)
3323 }
3324
3325 pub fn iterator_opt<'a: 'b, 'b>(
3326 &'a self,
3327 mode: IteratorMode,
3328 readopts: ReadOptions,
3329 ) -> DBIteratorWithThreadMode<'b, Self> {
3330 DBIteratorWithThreadMode::new(self, readopts, mode)
3331 }
3332
3333 /// Opens an iterator using the provided ReadOptions.
3334 /// This is used when you want to iterate over a specific ColumnFamily with a modified ReadOptions
3335 pub fn iterator_cf_opt<'a: 'b, 'b>(
3336 &'a self,
3337 cf_handle: &impl AsColumnFamilyRef,
3338 readopts: ReadOptions,
3339 mode: IteratorMode,
3340 ) -> DBIteratorWithThreadMode<'b, Self> {
3341 DBIteratorWithThreadMode::new_cf(self, cf_handle.inner(), readopts, mode)
3342 }
3343
3344 /// Opens an iterator with `set_total_order_seek` enabled.
3345 /// This must be used to iterate across prefixes when `set_memtable_factory` has been called
3346 /// with a Hash-based implementation.
3347 pub fn full_iterator<'a: 'b, 'b>(
3348 &'a self,
3349 mode: IteratorMode,
3350 ) -> DBIteratorWithThreadMode<'b, Self> {
3351 let mut opts = ReadOptions::default();
3352 opts.set_total_order_seek(true);
3353 DBIteratorWithThreadMode::new(self, opts, mode)
3354 }
3355
3356 pub fn prefix_iterator<'a: 'b, 'b, P: AsRef<[u8]>>(
3357 &'a self,
3358 prefix: P,
3359 ) -> DBIteratorWithThreadMode<'b, Self> {
3360 let mut opts = ReadOptions::default();
3361 opts.set_prefix_same_as_start(true);
3362 DBIteratorWithThreadMode::new(
3363 self,
3364 opts,
3365 IteratorMode::From(prefix.as_ref(), Direction::Forward),
3366 )
3367 }
3368
3369 pub fn iterator_cf<'a: 'b, 'b>(
3370 &'a self,
3371 cf_handle: &impl AsColumnFamilyRef,
3372 mode: IteratorMode,
3373 ) -> DBIteratorWithThreadMode<'b, Self> {
3374 let opts = ReadOptions::default();
3375 DBIteratorWithThreadMode::new_cf(self, cf_handle.inner(), opts, mode)
3376 }
3377
3378 pub fn full_iterator_cf<'a: 'b, 'b>(
3379 &'a self,
3380 cf_handle: &impl AsColumnFamilyRef,
3381 mode: IteratorMode,
3382 ) -> DBIteratorWithThreadMode<'b, Self> {
3383 let mut opts = ReadOptions::default();
3384 opts.set_total_order_seek(true);
3385 DBIteratorWithThreadMode::new_cf(self, cf_handle.inner(), opts, mode)
3386 }
3387
3388 pub fn prefix_iterator_cf<'a, P: AsRef<[u8]>>(
3389 &'a self,
3390 cf_handle: &impl AsColumnFamilyRef,
3391 prefix: P,
3392 ) -> DBIteratorWithThreadMode<'a, Self> {
3393 let mut opts = ReadOptions::default();
3394 opts.set_prefix_same_as_start(true);
3395 DBIteratorWithThreadMode::<'a, Self>::new_cf(
3396 self,
3397 cf_handle.inner(),
3398 opts,
3399 IteratorMode::From(prefix.as_ref(), Direction::Forward),
3400 )
3401 }
3402
3403 /// Returns `true` if there exists at least one key with the given prefix
3404 /// in the default column family using default read options.
3405 ///
3406 /// When to use: prefer this for one-shot checks. It enables
3407 /// `prefix_same_as_start(true)` and bounds the iterator to the
3408 /// prefix via `PrefixRange`, minimizing stray IO per call.
3409 pub fn prefix_exists<P: AsRef<[u8]>>(&self, prefix: P) -> Result<bool, Error> {
3410 let p = prefix.as_ref();
3411 with_prefix_read_opts(p, |opts| self.prefix_exists_opt(p, opts))
3412 }
3413
3414 /// Returns `true` if there exists at least one key with the given prefix
3415 /// in the default column family using the provided read options.
3416 pub fn prefix_exists_opt<P: AsRef<[u8]>>(
3417 &self,
3418 prefix: P,
3419 readopts: &ReadOptions,
3420 ) -> Result<bool, Error> {
3421 let prefix = prefix.as_ref();
3422 let iter = unsafe { self.create_iterator(readopts) };
3423 let res = unsafe {
3424 ffi::rocksdb_iter_seek(
3425 iter,
3426 prefix.as_ptr() as *const c_char,
3427 prefix.len() as size_t,
3428 );
3429 if ffi::rocksdb_iter_valid(iter) != 0 {
3430 let mut key_len: size_t = 0;
3431 let key_ptr = ffi::rocksdb_iter_key(iter, &raw mut key_len);
3432 // An empty key is legal, and `from_raw_parts` wants a
3433 // dereferenceable pointer even at length 0.
3434 let key = if key_len == 0 {
3435 &[][..]
3436 } else {
3437 slice::from_raw_parts(key_ptr.cast::<u8>(), key_len as usize)
3438 };
3439 Ok(key.starts_with(prefix))
3440 } else if let Err(e) = (|| {
3441 // Check status to differentiate end-of-range vs error
3442 ffi_try!(ffi::rocksdb_iter_get_error(iter));
3443 Ok::<(), Error>(())
3444 })() {
3445 Err(e)
3446 } else {
3447 Ok(false)
3448 }
3449 };
3450 unsafe { ffi::rocksdb_iter_destroy(iter) };
3451 res
3452 }
3453
3454 /// Creates a reusable prefix prober over the default column family using
3455 /// read options optimized for prefix probes.
3456 ///
3457 /// When to use: prefer this in hot loops with many checks per second. It
3458 /// reuses a raw iterator to avoid per-call allocation/FFI overhead. If you
3459 /// need custom tuning (e.g. async IO, readahead, cache-only), use
3460 /// `prefix_prober_with_opts`.
3461 pub fn prefix_prober(&self) -> PrefixProber<'_, Self> {
3462 PrefixProber {
3463 raw: DBRawIteratorWithThreadMode::new(self, prefix_probe_read_opts()),
3464 }
3465 }
3466
3467 /// Creates a reusable prefix prober over the default column family using
3468 /// the provided read options (owned).
3469 ///
3470 /// When to use: advanced tuning for heavy workloads. Callers can set
3471 /// `set_async_io(true)`, `set_readahead_size`, `set_read_tier`, etc. Note:
3472 /// the prober owns `ReadOptions` to keep internal buffers alive.
3473 pub fn prefix_prober_with_opts(&self, readopts: ReadOptions) -> PrefixProber<'_, Self> {
3474 PrefixProber {
3475 raw: DBRawIteratorWithThreadMode::new(self, readopts),
3476 }
3477 }
3478
3479 /// Creates a reusable prefix prober over the specified column family using
3480 /// read options optimized for prefix probes.
3481 pub fn prefix_prober_cf(&self, cf_handle: &impl AsColumnFamilyRef) -> PrefixProber<'_, Self> {
3482 PrefixProber {
3483 raw: DBRawIteratorWithThreadMode::new_cf(
3484 self,
3485 cf_handle.inner(),
3486 prefix_probe_read_opts(),
3487 ),
3488 }
3489 }
3490
3491 /// Creates a reusable prefix prober over the specified column family using
3492 /// the provided read options (owned).
3493 ///
3494 /// When to use: advanced tuning for heavy workloads on a specific CF.
3495 pub fn prefix_prober_cf_with_opts(
3496 &self,
3497 cf_handle: &impl AsColumnFamilyRef,
3498 readopts: ReadOptions,
3499 ) -> PrefixProber<'_, Self> {
3500 PrefixProber {
3501 raw: DBRawIteratorWithThreadMode::new_cf(self, cf_handle.inner(), readopts),
3502 }
3503 }
3504
3505 /// Returns `true` if there exists at least one key with the given prefix
3506 /// in the specified column family using default read options.
3507 ///
3508 /// When to use: one-shot checks on a CF. Enables
3509 /// `prefix_same_as_start(true)` and bounds the iterator via `PrefixRange`.
3510 pub fn prefix_exists_cf<P: AsRef<[u8]>>(
3511 &self,
3512 cf_handle: &impl AsColumnFamilyRef,
3513 prefix: P,
3514 ) -> Result<bool, Error> {
3515 let p = prefix.as_ref();
3516 with_prefix_read_opts(p, |opts| self.prefix_exists_cf_opt(cf_handle, p, opts))
3517 }
3518
3519 /// Returns `true` if there exists at least one key with the given prefix
3520 /// in the specified column family using the provided read options.
3521 pub fn prefix_exists_cf_opt<P: AsRef<[u8]>>(
3522 &self,
3523 cf_handle: &impl AsColumnFamilyRef,
3524 prefix: P,
3525 readopts: &ReadOptions,
3526 ) -> Result<bool, Error> {
3527 let prefix = prefix.as_ref();
3528 let iter = unsafe { self.create_iterator_cf(cf_handle.inner(), readopts) };
3529 let res = unsafe {
3530 ffi::rocksdb_iter_seek(
3531 iter,
3532 prefix.as_ptr() as *const c_char,
3533 prefix.len() as size_t,
3534 );
3535 if ffi::rocksdb_iter_valid(iter) != 0 {
3536 let mut key_len: size_t = 0;
3537 let key_ptr = ffi::rocksdb_iter_key(iter, &raw mut key_len);
3538 // An empty key is legal, and `from_raw_parts` wants a
3539 // dereferenceable pointer even at length 0.
3540 let key = if key_len == 0 {
3541 &[][..]
3542 } else {
3543 slice::from_raw_parts(key_ptr.cast::<u8>(), key_len as usize)
3544 };
3545 Ok(key.starts_with(prefix))
3546 } else if let Err(e) = (|| {
3547 ffi_try!(ffi::rocksdb_iter_get_error(iter));
3548 Ok::<(), Error>(())
3549 })() {
3550 Err(e)
3551 } else {
3552 Ok(false)
3553 }
3554 };
3555 unsafe { ffi::rocksdb_iter_destroy(iter) };
3556 res
3557 }
3558
3559 /// Opens a raw iterator over the database, using the default read options
3560 pub fn raw_iterator<'a: 'b, 'b>(&'a self) -> DBRawIteratorWithThreadMode<'b, Self> {
3561 let opts = ReadOptions::default();
3562 DBRawIteratorWithThreadMode::new(self, opts)
3563 }
3564
3565 /// Opens a raw iterator over the given column family, using the default read options
3566 pub fn raw_iterator_cf<'a: 'b, 'b>(
3567 &'a self,
3568 cf_handle: &impl AsColumnFamilyRef,
3569 ) -> DBRawIteratorWithThreadMode<'b, Self> {
3570 let opts = ReadOptions::default();
3571 DBRawIteratorWithThreadMode::new_cf(self, cf_handle.inner(), opts)
3572 }
3573
3574 /// Opens raw iterators for multiple column families from one consistent
3575 /// RocksDB state.
3576 ///
3577 /// The returned iterators match the input column family order and own their
3578 /// native handles. They share one `ReadOptions`, because one native
3579 /// `rocksdb_create_iterators` call applies a single options object to every
3580 /// iterator it creates.
3581 pub fn raw_iterators_cf<'a, 'b, W, I>(
3582 &'a self,
3583 column_families: I,
3584 ) -> Result<Vec<DBRawIteratorWithThreadMode<'a, Self>>, Error>
3585 where
3586 W: AsColumnFamilyRef + 'b,
3587 I: IntoIterator<Item = &'b W>,
3588 {
3589 let mut cf_handles: Vec<_> = column_families
3590 .into_iter()
3591 .map(AsColumnFamilyRef::inner)
3592 .collect();
3593 if cf_handles.is_empty() {
3594 return Ok(Vec::new());
3595 }
3596 let created = self.create_iterators_cf(&mut cf_handles)?;
3597 Ok(created
3598 .handles
3599 .into_iter()
3600 .map(|handle| {
3601 DBRawIteratorWithThreadMode::from_inner(handle, Arc::clone(&created.readopts))
3602 })
3603 .collect())
3604 }
3605
3606 fn create_iterators_cf(
3607 &self,
3608 cf_handles: &mut [*mut ffi::rocksdb_column_family_handle_t],
3609 ) -> Result<CreatedIterators, Error> {
3610 let mut iterator_handles = vec![ptr::null_mut(); cf_handles.len()];
3611 // Every iterator gets a handle on this. RocksDB's `DBIter` stores raw
3612 // `Slice*` into the options for iterate_lower_bound, iterate_upper_bound
3613 // and the read timestamps, and `ArenaWrappedDBIter::Refresh` re-reads
3614 // them, so the options have to outlive the last iterator rather than
3615 // this function. See issue #660.
3616 let readopts = Arc::new(ReadOptions::default());
3617 unsafe {
3618 ffi_try!(ffi::rust_rocksdb_create_iterators_safe(
3619 self.inner.inner(),
3620 readopts.inner,
3621 cf_handles.as_mut_ptr(),
3622 iterator_handles.as_mut_ptr(),
3623 iterator_handles.len(),
3624 ));
3625 }
3626 Self::validate_created_iterators(&iterator_handles)?;
3627 Ok(CreatedIterators {
3628 readopts,
3629 handles: iterator_handles,
3630 })
3631 }
3632
3633 fn validate_created_iterators(
3634 iterator_handles: &[*mut ffi::rocksdb_iterator_t],
3635 ) -> Result<(), Error> {
3636 if iterator_handles.iter().any(|iterator| iterator.is_null()) {
3637 unsafe {
3638 Self::destroy_iterators(iterator_handles);
3639 }
3640 return Err(Error::new(
3641 "rocksdb_create_iterators returned a null iterator".to_owned(),
3642 ));
3643 }
3644 Ok(())
3645 }
3646
3647 /// Destroys non-null iterator handles owned by the caller.
3648 ///
3649 /// # Safety
3650 ///
3651 /// Every non-null pointer must identify a live, uniquely owned RocksDB iterator.
3652 unsafe fn destroy_iterators(iterators: &[*mut ffi::rocksdb_iterator_t]) {
3653 for &iterator in iterators {
3654 if !iterator.is_null() {
3655 unsafe {
3656 ffi::rocksdb_iter_destroy(iterator);
3657 }
3658 }
3659 }
3660 }
3661
3662 /// Opens a raw iterator over the database, using the given read options
3663 pub fn raw_iterator_opt<'a: 'b, 'b>(
3664 &'a self,
3665 readopts: ReadOptions,
3666 ) -> DBRawIteratorWithThreadMode<'b, Self> {
3667 DBRawIteratorWithThreadMode::new(self, readopts)
3668 }
3669
3670 /// Opens a raw iterator over the given column family, using the given read options
3671 pub fn raw_iterator_cf_opt<'a: 'b, 'b>(
3672 &'a self,
3673 cf_handle: &impl AsColumnFamilyRef,
3674 readopts: ReadOptions,
3675 ) -> DBRawIteratorWithThreadMode<'b, Self> {
3676 DBRawIteratorWithThreadMode::new_cf(self, cf_handle.inner(), readopts)
3677 }
3678
3679 pub fn snapshot(&'_ self) -> SnapshotWithThreadMode<'_, Self> {
3680 SnapshotWithThreadMode::<Self>::new(self)
3681 }
3682
3683 pub fn put_opt<K, V>(&self, key: K, value: V, writeopts: &WriteOptions) -> Result<(), Error>
3684 where
3685 K: AsRef<[u8]>,
3686 V: AsRef<[u8]>,
3687 {
3688 let key = key.as_ref();
3689 let value = value.as_ref();
3690
3691 unsafe {
3692 ffi_try!(ffi::rocksdb_put(
3693 self.inner.inner(),
3694 writeopts.inner,
3695 key.as_ptr() as *const c_char,
3696 key.len() as size_t,
3697 value.as_ptr() as *const c_char,
3698 value.len() as size_t,
3699 ));
3700 Ok(())
3701 }
3702 }
3703
3704 pub fn put_cf_opt<K, V>(
3705 &self,
3706 cf: &impl AsColumnFamilyRef,
3707 key: K,
3708 value: V,
3709 writeopts: &WriteOptions,
3710 ) -> Result<(), Error>
3711 where
3712 K: AsRef<[u8]>,
3713 V: AsRef<[u8]>,
3714 {
3715 let key = key.as_ref();
3716 let value = value.as_ref();
3717
3718 unsafe {
3719 ffi_try!(ffi::rocksdb_put_cf(
3720 self.inner.inner(),
3721 writeopts.inner,
3722 cf.inner(),
3723 key.as_ptr() as *const c_char,
3724 key.len() as size_t,
3725 value.as_ptr() as *const c_char,
3726 value.len() as size_t,
3727 ));
3728 Ok(())
3729 }
3730 }
3731
3732 /// Set the database entry for "key" to "value" with WriteOptions.
3733 /// If "key" already exists, it will coexist with previous entry.
3734 /// `Get` with a timestamp ts specified in ReadOptions will return
3735 /// the most recent key/value whose timestamp is smaller than or equal to ts.
3736 /// Takes an additional argument `ts` as the timestamp.
3737 /// Note: the DB must be opened with user defined timestamp enabled.
3738 pub fn put_with_ts_opt<K, V, S>(
3739 &self,
3740 key: K,
3741 ts: S,
3742 value: V,
3743 writeopts: &WriteOptions,
3744 ) -> Result<(), Error>
3745 where
3746 K: AsRef<[u8]>,
3747 V: AsRef<[u8]>,
3748 S: AsRef<[u8]>,
3749 {
3750 let key = key.as_ref();
3751 let value = value.as_ref();
3752 let ts = ts.as_ref();
3753 unsafe {
3754 ffi_try!(ffi::rocksdb_put_with_ts(
3755 self.inner.inner(),
3756 writeopts.inner,
3757 key.as_ptr() as *const c_char,
3758 key.len() as size_t,
3759 ts.as_ptr() as *const c_char,
3760 ts.len() as size_t,
3761 value.as_ptr() as *const c_char,
3762 value.len() as size_t,
3763 ));
3764 Ok(())
3765 }
3766 }
3767
3768 /// Put with timestamp in a specific column family with WriteOptions.
3769 /// If "key" already exists, it will coexist with previous entry.
3770 /// `Get` with a timestamp ts specified in ReadOptions will return
3771 /// the most recent key/value whose timestamp is smaller than or equal to ts.
3772 /// Takes an additional argument `ts` as the timestamp.
3773 /// Note: the DB must be opened with user defined timestamp enabled.
3774 pub fn put_cf_with_ts_opt<K, V, S>(
3775 &self,
3776 cf: &impl AsColumnFamilyRef,
3777 key: K,
3778 ts: S,
3779 value: V,
3780 writeopts: &WriteOptions,
3781 ) -> Result<(), Error>
3782 where
3783 K: AsRef<[u8]>,
3784 V: AsRef<[u8]>,
3785 S: AsRef<[u8]>,
3786 {
3787 let key = key.as_ref();
3788 let value = value.as_ref();
3789 let ts = ts.as_ref();
3790 unsafe {
3791 ffi_try!(ffi::rocksdb_put_cf_with_ts(
3792 self.inner.inner(),
3793 writeopts.inner,
3794 cf.inner(),
3795 key.as_ptr() as *const c_char,
3796 key.len() as size_t,
3797 ts.as_ptr() as *const c_char,
3798 ts.len() as size_t,
3799 value.as_ptr() as *const c_char,
3800 value.len() as size_t,
3801 ));
3802 Ok(())
3803 }
3804 }
3805
3806 pub fn merge_opt<K, V>(&self, key: K, value: V, writeopts: &WriteOptions) -> Result<(), Error>
3807 where
3808 K: AsRef<[u8]>,
3809 V: AsRef<[u8]>,
3810 {
3811 let key = key.as_ref();
3812 let value = value.as_ref();
3813
3814 unsafe {
3815 ffi_try!(ffi::rocksdb_merge(
3816 self.inner.inner(),
3817 writeopts.inner,
3818 key.as_ptr() as *const c_char,
3819 key.len() as size_t,
3820 value.as_ptr() as *const c_char,
3821 value.len() as size_t,
3822 ));
3823 Ok(())
3824 }
3825 }
3826
3827 pub fn merge_cf_opt<K, V>(
3828 &self,
3829 cf: &impl AsColumnFamilyRef,
3830 key: K,
3831 value: V,
3832 writeopts: &WriteOptions,
3833 ) -> Result<(), Error>
3834 where
3835 K: AsRef<[u8]>,
3836 V: AsRef<[u8]>,
3837 {
3838 let key = key.as_ref();
3839 let value = value.as_ref();
3840
3841 unsafe {
3842 ffi_try!(ffi::rocksdb_merge_cf(
3843 self.inner.inner(),
3844 writeopts.inner,
3845 cf.inner(),
3846 key.as_ptr() as *const c_char,
3847 key.len() as size_t,
3848 value.as_ptr() as *const c_char,
3849 value.len() as size_t,
3850 ));
3851 Ok(())
3852 }
3853 }
3854
3855 pub fn delete_opt<K: AsRef<[u8]>>(
3856 &self,
3857 key: K,
3858 writeopts: &WriteOptions,
3859 ) -> Result<(), Error> {
3860 let key = key.as_ref();
3861
3862 unsafe {
3863 ffi_try!(ffi::rocksdb_delete(
3864 self.inner.inner(),
3865 writeopts.inner,
3866 key.as_ptr() as *const c_char,
3867 key.len() as size_t,
3868 ));
3869 Ok(())
3870 }
3871 }
3872
3873 pub fn delete_cf_opt<K: AsRef<[u8]>>(
3874 &self,
3875 cf: &impl AsColumnFamilyRef,
3876 key: K,
3877 writeopts: &WriteOptions,
3878 ) -> Result<(), Error> {
3879 let key = key.as_ref();
3880
3881 unsafe {
3882 ffi_try!(ffi::rocksdb_delete_cf(
3883 self.inner.inner(),
3884 writeopts.inner,
3885 cf.inner(),
3886 key.as_ptr() as *const c_char,
3887 key.len() as size_t,
3888 ));
3889 Ok(())
3890 }
3891 }
3892
3893 /// Remove the database entry (if any) for "key" with WriteOptions.
3894 /// Takes an additional argument `ts` as the timestamp.
3895 /// Note: the DB must be opened with user defined timestamp enabled.
3896 pub fn delete_with_ts_opt<K, S>(
3897 &self,
3898 key: K,
3899 ts: S,
3900 writeopts: &WriteOptions,
3901 ) -> Result<(), Error>
3902 where
3903 K: AsRef<[u8]>,
3904 S: AsRef<[u8]>,
3905 {
3906 let key = key.as_ref();
3907 let ts = ts.as_ref();
3908 unsafe {
3909 ffi_try!(ffi::rocksdb_delete_with_ts(
3910 self.inner.inner(),
3911 writeopts.inner,
3912 key.as_ptr() as *const c_char,
3913 key.len() as size_t,
3914 ts.as_ptr() as *const c_char,
3915 ts.len() as size_t,
3916 ));
3917 Ok(())
3918 }
3919 }
3920
3921 /// Delete with timestamp in a specific column family with WriteOptions.
3922 /// Takes an additional argument `ts` as the timestamp.
3923 /// Note: the DB must be opened with user defined timestamp enabled.
3924 pub fn delete_cf_with_ts_opt<K, S>(
3925 &self,
3926 cf: &impl AsColumnFamilyRef,
3927 key: K,
3928 ts: S,
3929 writeopts: &WriteOptions,
3930 ) -> Result<(), Error>
3931 where
3932 K: AsRef<[u8]>,
3933 S: AsRef<[u8]>,
3934 {
3935 let key = key.as_ref();
3936 let ts = ts.as_ref();
3937 unsafe {
3938 ffi_try!(ffi::rocksdb_delete_cf_with_ts(
3939 self.inner.inner(),
3940 writeopts.inner,
3941 cf.inner(),
3942 key.as_ptr() as *const c_char,
3943 key.len() as size_t,
3944 ts.as_ptr() as *const c_char,
3945 ts.len() as size_t,
3946 ));
3947 Ok(())
3948 }
3949 }
3950
3951 pub fn put<K, V>(&self, key: K, value: V) -> Result<(), Error>
3952 where
3953 K: AsRef<[u8]>,
3954 V: AsRef<[u8]>,
3955 {
3956 DEFAULT_WRITE_OPTS.with(|opts| self.put_opt(key, value, opts))
3957 }
3958
3959 pub fn put_cf<K, V>(&self, cf: &impl AsColumnFamilyRef, key: K, value: V) -> Result<(), Error>
3960 where
3961 K: AsRef<[u8]>,
3962 V: AsRef<[u8]>,
3963 {
3964 DEFAULT_WRITE_OPTS.with(|opts| self.put_cf_opt(cf, key, value, opts))
3965 }
3966
3967 /// Set the database entry for "key" to "value".
3968 /// If "key" already exists, it will coexist with previous entry.
3969 /// `Get` with a timestamp ts specified in ReadOptions will return
3970 /// the most recent key/value whose timestamp is smaller than or equal to ts.
3971 /// Takes an additional argument `ts` as the timestamp.
3972 /// Note: the DB must be opened with user defined timestamp enabled.
3973 pub fn put_with_ts<K, V, S>(&self, key: K, ts: S, value: V) -> Result<(), Error>
3974 where
3975 K: AsRef<[u8]>,
3976 V: AsRef<[u8]>,
3977 S: AsRef<[u8]>,
3978 {
3979 DEFAULT_WRITE_OPTS
3980 .with(|opts| self.put_with_ts_opt(key.as_ref(), ts.as_ref(), value.as_ref(), opts))
3981 }
3982
3983 /// Put with timestamp in a specific column family.
3984 /// If "key" already exists, it will coexist with previous entry.
3985 /// `Get` with a timestamp ts specified in ReadOptions will return
3986 /// the most recent key/value whose timestamp is smaller than or equal to ts.
3987 /// Takes an additional argument `ts` as the timestamp.
3988 /// Note: the DB must be opened with user defined timestamp enabled.
3989 pub fn put_cf_with_ts<K, V, S>(
3990 &self,
3991 cf: &impl AsColumnFamilyRef,
3992 key: K,
3993 ts: S,
3994 value: V,
3995 ) -> Result<(), Error>
3996 where
3997 K: AsRef<[u8]>,
3998 V: AsRef<[u8]>,
3999 S: AsRef<[u8]>,
4000 {
4001 DEFAULT_WRITE_OPTS.with(|opts| {
4002 self.put_cf_with_ts_opt(cf, key.as_ref(), ts.as_ref(), value.as_ref(), opts)
4003 })
4004 }
4005
4006 pub fn merge<K, V>(&self, key: K, value: V) -> Result<(), Error>
4007 where
4008 K: AsRef<[u8]>,
4009 V: AsRef<[u8]>,
4010 {
4011 DEFAULT_WRITE_OPTS.with(|opts| self.merge_opt(key, value, opts))
4012 }
4013
4014 pub fn merge_cf<K, V>(&self, cf: &impl AsColumnFamilyRef, key: K, value: V) -> Result<(), Error>
4015 where
4016 K: AsRef<[u8]>,
4017 V: AsRef<[u8]>,
4018 {
4019 DEFAULT_WRITE_OPTS.with(|opts| self.merge_cf_opt(cf, key, value, opts))
4020 }
4021
4022 pub fn delete<K: AsRef<[u8]>>(&self, key: K) -> Result<(), Error> {
4023 DEFAULT_WRITE_OPTS.with(|opts| self.delete_opt(key, opts))
4024 }
4025
4026 pub fn delete_cf<K: AsRef<[u8]>>(
4027 &self,
4028 cf: &impl AsColumnFamilyRef,
4029 key: K,
4030 ) -> Result<(), Error> {
4031 DEFAULT_WRITE_OPTS.with(|opts| self.delete_cf_opt(cf, key, opts))
4032 }
4033
4034 /// Remove the database entry (if any) for "key".
4035 /// Takes an additional argument `ts` as the timestamp.
4036 /// Note: the DB must be opened with user defined timestamp enabled.
4037 pub fn delete_with_ts<K: AsRef<[u8]>, S: AsRef<[u8]>>(
4038 &self,
4039 key: K,
4040 ts: S,
4041 ) -> Result<(), Error> {
4042 DEFAULT_WRITE_OPTS.with(|opts| self.delete_with_ts_opt(key, ts, opts))
4043 }
4044
4045 /// Delete with timestamp in a specific column family.
4046 /// Takes an additional argument `ts` as the timestamp.
4047 /// Note: the DB must be opened with user defined timestamp enabled.
4048 pub fn delete_cf_with_ts<K: AsRef<[u8]>, S: AsRef<[u8]>>(
4049 &self,
4050 cf: &impl AsColumnFamilyRef,
4051 key: K,
4052 ts: S,
4053 ) -> Result<(), Error> {
4054 DEFAULT_WRITE_OPTS.with(|opts| self.delete_cf_with_ts_opt(cf, key, ts, opts))
4055 }
4056
4057 /// Remove the database entry for "key" with WriteOptions.
4058 ///
4059 /// Requires that the key exists and was not overwritten. Returns OK on success,
4060 /// and a non-OK status on error. It is not an error if "key" did not exist in the database.
4061 ///
4062 /// If a key is overwritten (by calling Put() multiple times), then the result
4063 /// of calling SingleDelete() on this key is undefined. SingleDelete() only
4064 /// behaves correctly if there has been only one Put() for this key since the
4065 /// previous call to SingleDelete() for this key.
4066 ///
4067 /// This feature is currently an experimental performance optimization
4068 /// for a very specific workload. It is up to the caller to ensure that
4069 /// SingleDelete is only used for a key that is not deleted using Delete() or
4070 /// written using Merge(). Mixing SingleDelete operations with Deletes and
4071 /// Merges can result in undefined behavior.
4072 ///
4073 /// Note: consider setting options.sync = true.
4074 ///
4075 /// For more information, see <https://github.com/facebook/rocksdb/wiki/Single-Delete>
4076 pub fn single_delete_opt<K: AsRef<[u8]>>(
4077 &self,
4078 key: K,
4079 writeopts: &WriteOptions,
4080 ) -> Result<(), Error> {
4081 let key = key.as_ref();
4082
4083 unsafe {
4084 ffi_try!(ffi::rocksdb_singledelete(
4085 self.inner.inner(),
4086 writeopts.inner,
4087 key.as_ptr() as *const c_char,
4088 key.len() as size_t,
4089 ));
4090 Ok(())
4091 }
4092 }
4093
4094 /// Remove the database entry for "key" from a specific column family with WriteOptions.
4095 ///
4096 /// See single_delete_opt() for detailed behavior and restrictions.
4097 pub fn single_delete_cf_opt<K: AsRef<[u8]>>(
4098 &self,
4099 cf: &impl AsColumnFamilyRef,
4100 key: K,
4101 writeopts: &WriteOptions,
4102 ) -> Result<(), Error> {
4103 let key = key.as_ref();
4104
4105 unsafe {
4106 ffi_try!(ffi::rocksdb_singledelete_cf(
4107 self.inner.inner(),
4108 writeopts.inner,
4109 cf.inner(),
4110 key.as_ptr() as *const c_char,
4111 key.len() as size_t,
4112 ));
4113 Ok(())
4114 }
4115 }
4116
4117 /// Remove the database entry for "key" with WriteOptions.
4118 ///
4119 /// Takes an additional argument `ts` as the timestamp.
4120 /// Note: the DB must be opened with user defined timestamp enabled.
4121 ///
4122 /// See single_delete_opt() for detailed behavior and restrictions.
4123 pub fn single_delete_with_ts_opt<K, S>(
4124 &self,
4125 key: K,
4126 ts: S,
4127 writeopts: &WriteOptions,
4128 ) -> Result<(), Error>
4129 where
4130 K: AsRef<[u8]>,
4131 S: AsRef<[u8]>,
4132 {
4133 let key = key.as_ref();
4134 let ts = ts.as_ref();
4135 unsafe {
4136 ffi_try!(ffi::rocksdb_singledelete_with_ts(
4137 self.inner.inner(),
4138 writeopts.inner,
4139 key.as_ptr() as *const c_char,
4140 key.len() as size_t,
4141 ts.as_ptr() as *const c_char,
4142 ts.len() as size_t,
4143 ));
4144 Ok(())
4145 }
4146 }
4147
4148 /// Remove the database entry for "key" from a specific column family with WriteOptions.
4149 ///
4150 /// Takes an additional argument `ts` as the timestamp.
4151 /// Note: the DB must be opened with user defined timestamp enabled.
4152 ///
4153 /// See single_delete_opt() for detailed behavior and restrictions.
4154 pub fn single_delete_cf_with_ts_opt<K, S>(
4155 &self,
4156 cf: &impl AsColumnFamilyRef,
4157 key: K,
4158 ts: S,
4159 writeopts: &WriteOptions,
4160 ) -> Result<(), Error>
4161 where
4162 K: AsRef<[u8]>,
4163 S: AsRef<[u8]>,
4164 {
4165 let key = key.as_ref();
4166 let ts = ts.as_ref();
4167 unsafe {
4168 ffi_try!(ffi::rocksdb_singledelete_cf_with_ts(
4169 self.inner.inner(),
4170 writeopts.inner,
4171 cf.inner(),
4172 key.as_ptr() as *const c_char,
4173 key.len() as size_t,
4174 ts.as_ptr() as *const c_char,
4175 ts.len() as size_t,
4176 ));
4177 Ok(())
4178 }
4179 }
4180
4181 /// Remove the database entry for "key".
4182 ///
4183 /// See single_delete_opt() for detailed behavior and restrictions.
4184 pub fn single_delete<K: AsRef<[u8]>>(&self, key: K) -> Result<(), Error> {
4185 DEFAULT_WRITE_OPTS.with(|opts| self.single_delete_opt(key, opts))
4186 }
4187
4188 /// Remove the database entry for "key" from a specific column family.
4189 ///
4190 /// See single_delete_opt() for detailed behavior and restrictions.
4191 pub fn single_delete_cf<K: AsRef<[u8]>>(
4192 &self,
4193 cf: &impl AsColumnFamilyRef,
4194 key: K,
4195 ) -> Result<(), Error> {
4196 DEFAULT_WRITE_OPTS.with(|opts| self.single_delete_cf_opt(cf, key, opts))
4197 }
4198
4199 /// Remove the database entry for "key".
4200 ///
4201 /// Takes an additional argument `ts` as the timestamp.
4202 /// Note: the DB must be opened with user defined timestamp enabled.
4203 ///
4204 /// See single_delete_opt() for detailed behavior and restrictions.
4205 pub fn single_delete_with_ts<K: AsRef<[u8]>, S: AsRef<[u8]>>(
4206 &self,
4207 key: K,
4208 ts: S,
4209 ) -> Result<(), Error> {
4210 DEFAULT_WRITE_OPTS.with(|opts| self.single_delete_with_ts_opt(key, ts, opts))
4211 }
4212
4213 /// Remove the database entry for "key" from a specific column family.
4214 ///
4215 /// Takes an additional argument `ts` as the timestamp.
4216 /// Note: the DB must be opened with user defined timestamp enabled.
4217 ///
4218 /// See single_delete_opt() for detailed behavior and restrictions.
4219 pub fn single_delete_cf_with_ts<K: AsRef<[u8]>, S: AsRef<[u8]>>(
4220 &self,
4221 cf: &impl AsColumnFamilyRef,
4222 key: K,
4223 ts: S,
4224 ) -> Result<(), Error> {
4225 DEFAULT_WRITE_OPTS.with(|opts| self.single_delete_cf_with_ts_opt(cf, key, ts, opts))
4226 }
4227
4228 /// Runs a manual compaction on the Range of keys given. This is not likely to be needed for typical usage.
4229 pub fn compact_range<S: AsRef<[u8]>, E: AsRef<[u8]>>(&self, start: Option<S>, end: Option<E>) {
4230 unsafe {
4231 let start = start.as_ref().map(AsRef::as_ref);
4232 let end = end.as_ref().map(AsRef::as_ref);
4233
4234 ffi::rocksdb_compact_range(
4235 self.inner.inner(),
4236 opt_bytes_to_ptr(start),
4237 start.map_or(0, <[u8]>::len) as size_t,
4238 opt_bytes_to_ptr(end),
4239 end.map_or(0, <[u8]>::len) as size_t,
4240 );
4241 }
4242 }
4243
4244 /// Same as `compact_range` but with custom options.
4245 pub fn compact_range_opt<S: AsRef<[u8]>, E: AsRef<[u8]>>(
4246 &self,
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_opt(
4256 self.inner.inner(),
4257 opts.inner,
4258 opt_bytes_to_ptr(start),
4259 start.map_or(0, <[u8]>::len) as size_t,
4260 opt_bytes_to_ptr(end),
4261 end.map_or(0, <[u8]>::len) as size_t,
4262 );
4263 }
4264 }
4265
4266 /// Runs a manual compaction on the Range of keys given on the
4267 /// given column family. This is not likely to be needed for typical usage.
4268 pub fn compact_range_cf<S: AsRef<[u8]>, E: AsRef<[u8]>>(
4269 &self,
4270 cf: &impl AsColumnFamilyRef,
4271 start: Option<S>,
4272 end: Option<E>,
4273 ) {
4274 unsafe {
4275 let start = start.as_ref().map(AsRef::as_ref);
4276 let end = end.as_ref().map(AsRef::as_ref);
4277
4278 ffi::rocksdb_compact_range_cf(
4279 self.inner.inner(),
4280 cf.inner(),
4281 opt_bytes_to_ptr(start),
4282 start.map_or(0, <[u8]>::len) as size_t,
4283 opt_bytes_to_ptr(end),
4284 end.map_or(0, <[u8]>::len) as size_t,
4285 );
4286 }
4287 }
4288
4289 /// Same as `compact_range_cf` but with custom options.
4290 pub fn compact_range_cf_opt<S: AsRef<[u8]>, E: AsRef<[u8]>>(
4291 &self,
4292 cf: &impl AsColumnFamilyRef,
4293 start: Option<S>,
4294 end: Option<E>,
4295 opts: &CompactOptions,
4296 ) {
4297 unsafe {
4298 let start = start.as_ref().map(AsRef::as_ref);
4299 let end = end.as_ref().map(AsRef::as_ref);
4300
4301 ffi::rocksdb_compact_range_cf_opt(
4302 self.inner.inner(),
4303 cf.inner(),
4304 opts.inner,
4305 opt_bytes_to_ptr(start),
4306 start.map_or(0, <[u8]>::len) as size_t,
4307 opt_bytes_to_ptr(end),
4308 end.map_or(0, <[u8]>::len) as size_t,
4309 );
4310 }
4311 }
4312
4313 /// Wait for all flush and compactions jobs to finish. Jobs to wait include the
4314 /// unscheduled (queued, but not scheduled yet).
4315 ///
4316 /// NOTE: This may also never return if there's sufficient ongoing writes that
4317 /// keeps flush and compaction going without stopping. The user would have to
4318 /// cease all the writes to DB to make this eventually return in a stable
4319 /// state. The user may also use timeout option in WaitForCompactOptions to
4320 /// make this stop waiting and return when timeout expires.
4321 pub fn wait_for_compact(&self, opts: &WaitForCompactOptions) -> Result<(), Error> {
4322 unsafe {
4323 ffi_try!(ffi::rocksdb_wait_for_compact(
4324 self.inner.inner(),
4325 opts.inner
4326 ));
4327 }
4328 Ok(())
4329 }
4330
4331 /// Changes mutable column family options on the default column family at
4332 /// runtime.
4333 ///
4334 /// [`set_db_options`](Self::set_db_options) is the DB wide equivalent. Either the
4335 /// whole set is applied or none of it is.
4336 ///
4337 /// # Aborts
4338 ///
4339 /// Some unparseable values take the process down instead of returning an error,
4340 /// so validate values before passing them here.
4341 ///
4342 /// RocksDB parses integers with `std::stoi` and friends
4343 /// (`util/string_util.cc:378`), which throw `std::invalid_argument`. It does try
4344 /// to catch that, but not everywhere, and an integer-valued option such as
4345 /// `write_buffer_size` given a non-numeric value aborts the process with
4346 /// "Rust cannot catch foreign exceptions". A boolean-valued option such as
4347 /// `disable_auto_compactions` returns an error instead. Do not rely on which
4348 /// options fall on which side of that line.
4349 ///
4350 /// Catching it here is not an option. `ColumnFamilyData::SetOptions` runs inside
4351 /// a callback that `VersionSet::LogAndApply` invokes while holding the DB mutex
4352 /// as the exclusive manifest writer (`db/db_impl/db_impl.cc:1655`), so unwinding
4353 /// through it leaves that state behind and the next option change on the DB
4354 /// blocks forever in `InstrumentedCondVar::Wait`. That was reproducible under
4355 /// ASAN and silent otherwise, which is worse than aborting.
4356 ///
4357 /// # Errors
4358 ///
4359 /// Returns the RocksDB error if a name is unknown or the option is not changeable
4360 /// at runtime. Also errors if any name or value contains an interior NUL byte.
4361 pub fn set_options(&self, opts: &[(&str, &str)]) -> Result<(), Error> {
4362 let copts = convert_options(opts)?;
4363 let cnames: Vec<*const c_char> = copts.iter().map(|opt| opt.0.as_ptr()).collect();
4364 let cvalues: Vec<*const c_char> = copts.iter().map(|opt| opt.1.as_ptr()).collect();
4365 unsafe {
4366 ffi_try!(ffi::rocksdb_set_options(
4367 self.inner.inner(),
4368 option_count(&copts)?,
4369 cnames.as_ptr(),
4370 cvalues.as_ptr(),
4371 ));
4372 }
4373 Ok(())
4374 }
4375
4376 /// Like [`set_options`](Self::set_options), for a single column family.
4377 ///
4378 /// # Aborts
4379 ///
4380 /// See [`set_options`](Self::set_options).
4381 ///
4382 /// # Errors
4383 ///
4384 /// See [`set_options`](Self::set_options).
4385 pub fn set_options_cf(
4386 &self,
4387 cf: &impl AsColumnFamilyRef,
4388 opts: &[(&str, &str)],
4389 ) -> Result<(), Error> {
4390 let copts = convert_options(opts)?;
4391 let cnames: Vec<*const c_char> = copts.iter().map(|opt| opt.0.as_ptr()).collect();
4392 let cvalues: Vec<*const c_char> = copts.iter().map(|opt| opt.1.as_ptr()).collect();
4393 unsafe {
4394 ffi_try!(ffi::rocksdb_set_options_cf(
4395 self.inner.inner(),
4396 cf.inner(),
4397 option_count(&copts)?,
4398 cnames.as_ptr(),
4399 cvalues.as_ptr(),
4400 ));
4401 }
4402 Ok(())
4403 }
4404
4405 /// Implementation for property_value et al methods.
4406 ///
4407 /// `name` is the name of the property. It will be converted into a CString
4408 /// and passed to `get_property` as argument. `get_property` reads the
4409 /// specified property and either returns NULL or a pointer to a C allocated
4410 /// string; this method takes ownership of that string and will free it at
4411 /// the end. That string is parsed using `parse` callback which produces
4412 /// the returned result.
4413 fn property_value_impl<R>(
4414 name: impl CStrLike,
4415 get_property: impl FnOnce(*const c_char) -> *mut c_char,
4416 parse: impl FnOnce(&str) -> Result<R, Error>,
4417 ) -> Result<Option<R>, Error> {
4418 let value = match name.bake() {
4419 Ok(prop_name) => get_property(prop_name.as_ptr()),
4420 Err(e) => {
4421 return Err(Error::new(format!(
4422 "Failed to convert property name to CString: {e}"
4423 )));
4424 }
4425 };
4426 if value.is_null() {
4427 return Ok(None);
4428 }
4429 let result = match unsafe { CStr::from_ptr(value) }.to_str() {
4430 Ok(s) => parse(s).map(|value| Some(value)),
4431 Err(e) => Err(Error::new(format!(
4432 "Failed to convert property value to string: {e}"
4433 ))),
4434 };
4435 unsafe {
4436 ffi::rocksdb_free(value as *mut c_void);
4437 }
4438 result
4439 }
4440
4441 /// Retrieves a RocksDB property by name.
4442 ///
4443 /// Full list of properties could be find
4444 /// [here](https://github.com/facebook/rocksdb/blob/08809f5e6cd9cc4bc3958dd4d59457ae78c76660/include/rocksdb/db.h#L428-L634).
4445 pub fn property_value(&self, name: impl CStrLike) -> Result<Option<String>, Error> {
4446 Self::property_value_impl(
4447 name,
4448 |prop_name| unsafe { ffi::rocksdb_property_value(self.inner.inner(), prop_name) },
4449 |str_value| Ok(str_value.to_owned()),
4450 )
4451 }
4452
4453 /// Retrieves a RocksDB property by name, for a specific column family.
4454 ///
4455 /// Full list of properties could be find
4456 /// [here](https://github.com/facebook/rocksdb/blob/08809f5e6cd9cc4bc3958dd4d59457ae78c76660/include/rocksdb/db.h#L428-L634).
4457 pub fn property_value_cf(
4458 &self,
4459 cf: &impl AsColumnFamilyRef,
4460 name: impl CStrLike,
4461 ) -> Result<Option<String>, Error> {
4462 Self::property_value_impl(
4463 name,
4464 |prop_name| unsafe {
4465 ffi::rocksdb_property_value_cf(self.inner.inner(), cf.inner(), prop_name)
4466 },
4467 |str_value| Ok(str_value.to_owned()),
4468 )
4469 }
4470
4471 fn property_int_value_impl(
4472 name: impl CStrLike,
4473 get_property: impl FnOnce(*const c_char, *mut u64) -> c_int,
4474 get_string_property: impl FnOnce(*const c_char) -> *mut c_char,
4475 ) -> Result<Option<u64>, Error> {
4476 let prop_name = name.bake().map_err(|err| {
4477 Error::new(format!("Failed to convert property name to CString: {err}"))
4478 })?;
4479 let mut value = 0;
4480 if get_property(prop_name.as_ptr(), &raw mut value) == 0 {
4481 return Ok(Some(value));
4482 }
4483
4484 Self::property_value_impl(
4485 prop_name.as_ref(),
4486 get_string_property,
4487 Self::parse_property_int_value,
4488 )
4489 }
4490
4491 fn parse_property_int_value(value: &str) -> Result<u64, Error> {
4492 value.parse::<u64>().map_err(|err| {
4493 Error::new(format!(
4494 "Failed to convert property value {value} to int: {err}"
4495 ))
4496 })
4497 }
4498
4499 /// Retrieves a RocksDB property and casts it to an integer.
4500 ///
4501 /// Full list of properties that return int values could be find
4502 /// [here](https://github.com/facebook/rocksdb/blob/08809f5e6cd9cc4bc3958dd4d59457ae78c76660/include/rocksdb/db.h#L654-L689).
4503 pub fn property_int_value(&self, name: impl CStrLike) -> Result<Option<u64>, Error> {
4504 Self::property_int_value_impl(
4505 name,
4506 |prop_name, value| unsafe {
4507 ffi::rocksdb_property_int(self.inner.inner(), prop_name, value)
4508 },
4509 |prop_name| unsafe { ffi::rocksdb_property_value(self.inner.inner(), prop_name) },
4510 )
4511 }
4512
4513 /// Retrieves a RocksDB property for a specific column family and casts it to an integer.
4514 ///
4515 /// Full list of properties that return int values could be find
4516 /// [here](https://github.com/facebook/rocksdb/blob/08809f5e6cd9cc4bc3958dd4d59457ae78c76660/include/rocksdb/db.h#L654-L689).
4517 pub fn property_int_value_cf(
4518 &self,
4519 cf: &impl AsColumnFamilyRef,
4520 name: impl CStrLike,
4521 ) -> Result<Option<u64>, Error> {
4522 Self::property_int_value_impl(
4523 name,
4524 |prop_name, value| unsafe {
4525 ffi::rocksdb_property_int_cf(self.inner.inner(), cf.inner(), prop_name, value)
4526 },
4527 |prop_name| unsafe {
4528 ffi::rocksdb_property_value_cf(self.inner.inner(), cf.inner(), prop_name)
4529 },
4530 )
4531 }
4532
4533 /// The sequence number of the most recent transaction.
4534 pub fn latest_sequence_number(&self) -> u64 {
4535 unsafe { ffi::rocksdb_get_latest_sequence_number(self.inner.inner()) }
4536 }
4537
4538 /// Return the approximate file system space used by keys in each ranges.
4539 ///
4540 /// Note that the returned sizes measure file system space usage, so
4541 /// if the user data compresses by a factor of ten, the returned
4542 /// sizes will be one-tenth the size of the corresponding user data size.
4543 ///
4544 /// Due to lack of abi, only data flushed to disk is taken into account.
4545 /// # Errors
4546 ///
4547 /// Returns the RocksDB error if the size estimate fails, for instance on an
4548 /// I/O error reading the manifest. No partial sizes are reported in that
4549 /// case.
4550 pub fn get_approximate_sizes(&self, ranges: &[Range]) -> Result<Vec<u64>, Error> {
4551 self.get_approximate_sizes_cfopt(None::<&ColumnFamily>, ranges)
4552 }
4553
4554 /// Like [`Self::get_approximate_sizes`], for a single column family.
4555 ///
4556 /// # Errors
4557 ///
4558 /// See [`Self::get_approximate_sizes`].
4559 pub fn get_approximate_sizes_cf(
4560 &self,
4561 cf: &impl AsColumnFamilyRef,
4562 ranges: &[Range],
4563 ) -> Result<Vec<u64>, Error> {
4564 self.get_approximate_sizes_cfopt(Some(cf), ranges)
4565 }
4566
4567 fn get_approximate_sizes_cfopt(
4568 &self,
4569 cf: Option<&impl AsColumnFamilyRef>,
4570 ranges: &[Range],
4571 ) -> Result<Vec<u64>, Error> {
4572 let mut args = ApproximateSizesArgs::new(ranges);
4573 let mut err: *mut c_char = ptr::null_mut();
4574 match cf {
4575 None => unsafe {
4576 ffi::rocksdb_approximate_sizes(
4577 self.inner.inner(),
4578 args.count,
4579 args.start_keys.as_ptr(),
4580 args.start_key_lens.as_ptr(),
4581 args.end_keys.as_ptr(),
4582 args.end_key_lens.as_ptr(),
4583 args.sizes.as_mut_ptr(),
4584 &raw mut err,
4585 );
4586 },
4587 Some(cf) => unsafe {
4588 ffi::rocksdb_approximate_sizes_cf(
4589 self.inner.inner(),
4590 cf.inner(),
4591 args.count,
4592 args.start_keys.as_ptr(),
4593 args.start_key_lens.as_ptr(),
4594 args.end_keys.as_ptr(),
4595 args.end_key_lens.as_ptr(),
4596 args.sizes.as_mut_ptr(),
4597 &raw mut err,
4598 );
4599 },
4600 }
4601 args.finish(err)
4602 }
4603
4604 /// Like [`Self::get_approximate_sizes`], but lets the caller say what counts
4605 /// towards the total and how precise the answer has to be.
4606 ///
4607 /// [`Self::get_approximate_sizes`] counts SST files only. Pass a
4608 /// [`SizeApproximationOptions`] with
4609 /// [`set_include_memtables`](SizeApproximationOptions::set_include_memtables)
4610 /// on to include writes that have not been flushed yet.
4611 ///
4612 /// # Errors
4613 ///
4614 /// See [`Self::get_approximate_sizes`].
4615 pub fn get_approximate_sizes_with_options(
4616 &self,
4617 opts: &SizeApproximationOptions,
4618 ranges: &[Range],
4619 ) -> Result<Vec<u64>, Error> {
4620 let mut args = ApproximateSizesArgs::new(ranges);
4621 let mut err: *mut c_char = ptr::null_mut();
4622 unsafe {
4623 ffi::rocksdb_approximate_sizes_with_options(
4624 self.inner.inner(),
4625 opts.as_ptr(),
4626 args.count,
4627 args.start_keys.as_ptr(),
4628 args.start_key_lens.as_ptr(),
4629 args.end_keys.as_ptr(),
4630 args.end_key_lens.as_ptr(),
4631 args.sizes.as_mut_ptr(),
4632 &raw mut err,
4633 );
4634 }
4635 args.finish(err)
4636 }
4637
4638 /// Like [`Self::get_approximate_sizes_with_options`], for a single column
4639 /// family.
4640 ///
4641 /// # Errors
4642 ///
4643 /// See [`Self::get_approximate_sizes`].
4644 pub fn get_approximate_sizes_cf_with_options(
4645 &self,
4646 cf: &impl AsColumnFamilyRef,
4647 opts: &SizeApproximationOptions,
4648 ranges: &[Range],
4649 ) -> Result<Vec<u64>, Error> {
4650 let mut args = ApproximateSizesArgs::new(ranges);
4651 let mut err: *mut c_char = ptr::null_mut();
4652 unsafe {
4653 ffi::rocksdb_approximate_sizes_cf_with_options(
4654 self.inner.inner(),
4655 cf.inner(),
4656 opts.as_ptr(),
4657 args.count,
4658 args.start_keys.as_ptr(),
4659 args.start_key_lens.as_ptr(),
4660 args.end_keys.as_ptr(),
4661 args.end_key_lens.as_ptr(),
4662 args.sizes.as_mut_ptr(),
4663 &raw mut err,
4664 );
4665 }
4666 args.finish(err)
4667 }
4668
4669 /// Like [`Self::get_approximate_sizes_cf_with_options`], but selects what
4670 /// counts with a flag set instead of an options object.
4671 ///
4672 /// Prefer [`Self::get_approximate_sizes_cf_with_options`], which can also
4673 /// set an error margin. This variant exists because it is the form RocksDB
4674 /// offers without allocating.
4675 ///
4676 /// # Errors
4677 ///
4678 /// See [`Self::get_approximate_sizes`].
4679 pub fn get_approximate_sizes_cf_with_flags(
4680 &self,
4681 cf: &impl AsColumnFamilyRef,
4682 flags: SizeApproximationFlags,
4683 ranges: &[Range],
4684 ) -> Result<Vec<u64>, Error> {
4685 let mut args = ApproximateSizesArgs::new(ranges);
4686 let mut err: *mut c_char = ptr::null_mut();
4687 unsafe {
4688 ffi::rocksdb_approximate_sizes_cf_with_flags(
4689 self.inner.inner(),
4690 cf.inner(),
4691 args.count,
4692 args.start_keys.as_ptr(),
4693 args.start_key_lens.as_ptr(),
4694 args.end_keys.as_ptr(),
4695 args.end_key_lens.as_ptr(),
4696 flags.bits(),
4697 args.sizes.as_mut_ptr(),
4698 &raw mut err,
4699 );
4700 }
4701 args.finish(err)
4702 }
4703
4704 /// Iterate over batches of write operations since a given sequence.
4705 ///
4706 /// Produce an iterator that will provide the batches of write operations
4707 /// that have occurred since the given sequence (see
4708 /// `latest_sequence_number()`). Use the provided iterator to retrieve each
4709 /// (`u64`, `WriteBatch`) tuple, and then gather the individual puts and
4710 /// deletes using the `WriteBatch::iterate()` function.
4711 ///
4712 /// Calling `get_updates_since()` with a sequence number that is out of
4713 /// bounds will return an error.
4714 pub fn get_updates_since(&self, seq_number: u64) -> Result<DBWALIterator, Error> {
4715 unsafe {
4716 // rocksdb_wal_readoptions_t does not appear to have any functions
4717 // for creating and destroying it; fortunately we can pass a nullptr
4718 // here to get the default behavior
4719 let opts: *const ffi::rocksdb_wal_readoptions_t = ptr::null();
4720 let iter = ffi_try!(ffi::rocksdb_get_updates_since(
4721 self.inner.inner(),
4722 seq_number,
4723 opts
4724 ));
4725 Ok(DBWALIterator {
4726 inner: iter,
4727 start_seq_number: seq_number,
4728 })
4729 }
4730 }
4731
4732 /// Tries to catch up with the primary by reading as much as possible from the
4733 /// log files.
4734 pub fn try_catch_up_with_primary(&self) -> Result<(), Error> {
4735 unsafe {
4736 ffi_try!(ffi::rocksdb_try_catch_up_with_primary(self.inner.inner()));
4737 }
4738 Ok(())
4739 }
4740
4741 /// Loads a list of external SST files created with SstFileWriter into the DB with default opts
4742 pub fn ingest_external_file<P: AsRef<Path>>(&self, paths: Vec<P>) -> Result<(), Error> {
4743 let opts = IngestExternalFileOptions::default();
4744 self.ingest_external_file_opts(&opts, paths)
4745 }
4746
4747 /// Loads a list of external SST files created with SstFileWriter into the DB
4748 pub fn ingest_external_file_opts<P: AsRef<Path>>(
4749 &self,
4750 opts: &IngestExternalFileOptions,
4751 paths: Vec<P>,
4752 ) -> Result<(), Error> {
4753 let paths_v: Vec<CString> = paths.iter().map(to_cpath).collect::<Result<Vec<_>, _>>()?;
4754 let cpaths: Vec<_> = paths_v.iter().map(|path| path.as_ptr()).collect();
4755
4756 self.ingest_external_file_raw(opts, &paths_v, &cpaths)
4757 }
4758
4759 /// Loads a list of external SST files created with SstFileWriter into the DB for given Column Family
4760 /// with default opts
4761 pub fn ingest_external_file_cf<P: AsRef<Path>>(
4762 &self,
4763 cf: &impl AsColumnFamilyRef,
4764 paths: Vec<P>,
4765 ) -> Result<(), Error> {
4766 let opts = IngestExternalFileOptions::default();
4767 self.ingest_external_file_cf_opts(cf, &opts, paths)
4768 }
4769
4770 /// Loads a list of external SST files created with SstFileWriter into the DB for given Column Family
4771 pub fn ingest_external_file_cf_opts<P: AsRef<Path>>(
4772 &self,
4773 cf: &impl AsColumnFamilyRef,
4774 opts: &IngestExternalFileOptions,
4775 paths: Vec<P>,
4776 ) -> Result<(), Error> {
4777 let paths_v: Vec<CString> = paths.iter().map(to_cpath).collect::<Result<Vec<_>, _>>()?;
4778 let cpaths: Vec<_> = paths_v.iter().map(|path| path.as_ptr()).collect();
4779
4780 self.ingest_external_file_raw_cf(cf, opts, &paths_v, &cpaths)
4781 }
4782
4783 fn ingest_external_file_raw(
4784 &self,
4785 opts: &IngestExternalFileOptions,
4786 paths_v: &[CString],
4787 cpaths: &[*const c_char],
4788 ) -> Result<(), Error> {
4789 unsafe {
4790 ffi_try!(ffi::rocksdb_ingest_external_file(
4791 self.inner.inner(),
4792 cpaths.as_ptr(),
4793 paths_v.len(),
4794 opts.inner.cast_const()
4795 ));
4796 Ok(())
4797 }
4798 }
4799
4800 fn ingest_external_file_raw_cf(
4801 &self,
4802 cf: &impl AsColumnFamilyRef,
4803 opts: &IngestExternalFileOptions,
4804 paths_v: &[CString],
4805 cpaths: &[*const c_char],
4806 ) -> Result<(), Error> {
4807 unsafe {
4808 ffi_try!(ffi::rocksdb_ingest_external_file_cf(
4809 self.inner.inner(),
4810 cf.inner(),
4811 cpaths.as_ptr(),
4812 paths_v.len(),
4813 opts.inner.cast_const()
4814 ));
4815 Ok(())
4816 }
4817 }
4818
4819 /// Obtains the LSM-tree meta data of the default column family of the DB
4820 pub fn get_column_family_metadata(&self) -> ColumnFamilyMetaData {
4821 unsafe {
4822 let ptr = ffi::rocksdb_get_column_family_metadata(self.inner.inner());
4823
4824 let metadata = ColumnFamilyMetaData {
4825 size: ffi::rocksdb_column_family_metadata_get_size(ptr),
4826 name: from_cstr_and_free(ffi::rocksdb_column_family_metadata_get_name(ptr)),
4827 file_count: ffi::rocksdb_column_family_metadata_get_file_count(ptr),
4828 };
4829
4830 // destroy
4831 ffi::rocksdb_column_family_metadata_destroy(ptr);
4832
4833 // return
4834 metadata
4835 }
4836 }
4837
4838 /// Obtains the LSM-tree meta data of the specified column family of the DB
4839 pub fn get_column_family_metadata_cf(
4840 &self,
4841 cf: &impl AsColumnFamilyRef,
4842 ) -> ColumnFamilyMetaData {
4843 unsafe {
4844 let ptr = ffi::rocksdb_get_column_family_metadata_cf(self.inner.inner(), cf.inner());
4845
4846 let metadata = ColumnFamilyMetaData {
4847 size: ffi::rocksdb_column_family_metadata_get_size(ptr),
4848 name: from_cstr_and_free(ffi::rocksdb_column_family_metadata_get_name(ptr)),
4849 file_count: ffi::rocksdb_column_family_metadata_get_file_count(ptr),
4850 };
4851
4852 // destroy
4853 ffi::rocksdb_column_family_metadata_destroy(ptr);
4854
4855 // return
4856 metadata
4857 }
4858 }
4859
4860 /// Compacts the named SST files into `output_level`, giving the caller
4861 /// control over exactly which files are merged.
4862 ///
4863 /// [`Self::compact_range`] picks the files itself from a key range. This
4864 /// picks them by name, which is what a tool driving compaction from
4865 /// [`Self::live_files`] or [`Self::get_column_family_metadata_with_options`]
4866 /// needs. The file names come from that metadata, not from the filesystem.
4867 ///
4868 /// Runs on the calling thread, so it blocks until the compaction finishes.
4869 ///
4870 /// # Errors
4871 ///
4872 /// Returns the RocksDB error if any input file is unknown, if the files do
4873 /// not form a compactible set, or if the compaction itself fails. Nothing is
4874 /// compacted in that case.
4875 pub fn compact_files<I, N>(
4876 &self,
4877 opts: &CompactionOptions,
4878 input_file_names: I,
4879 output_level: i32,
4880 ) -> Result<CompactFilesResult, Error>
4881 where
4882 I: IntoIterator<Item = N>,
4883 N: CStrLike,
4884 {
4885 self.compact_files_impl(None::<&ColumnFamily>, opts, input_file_names, output_level)
4886 }
4887
4888 /// Like [`Self::compact_files`], for a single column family.
4889 ///
4890 /// # Errors
4891 ///
4892 /// See [`Self::compact_files`].
4893 pub fn compact_files_cf<I, N>(
4894 &self,
4895 cf: &impl AsColumnFamilyRef,
4896 opts: &CompactionOptions,
4897 input_file_names: I,
4898 output_level: i32,
4899 ) -> Result<CompactFilesResult, Error>
4900 where
4901 I: IntoIterator<Item = N>,
4902 N: CStrLike,
4903 {
4904 self.compact_files_impl(Some(cf), opts, input_file_names, output_level)
4905 }
4906
4907 fn compact_files_impl<I, N>(
4908 &self,
4909 cf: Option<&impl AsColumnFamilyRef>,
4910 opts: &CompactionOptions,
4911 input_file_names: I,
4912 output_level: i32,
4913 ) -> Result<CompactFilesResult, Error>
4914 where
4915 I: IntoIterator<Item = N>,
4916 N: CStrLike,
4917 {
4918 let names = input_file_names
4919 .into_iter()
4920 .map(CStrLike::into_c_string)
4921 .collect::<Result<Vec<_>, _>>()
4922 .map_err(|e| Error::new(format!("Invalid input file name: {e}")))?;
4923 let name_ptrs: Vec<*const c_char> = names.iter().map(|n| n.as_ptr()).collect();
4924
4925 let mut output_names: *mut *mut c_char = ptr::null_mut();
4926 let mut output_count: usize = 0;
4927
4928 // A trivial move returns from `CompactFilesImpl` before
4929 // `BuildCompactionJobInfo` runs, so it leaves a caller-allocated job info
4930 // untouched while still reporting success, and nothing afterwards says
4931 // which path ran. Asking for the job info only when that path is off
4932 // keeps every value handed out one RocksDB actually wrote.
4933 let mut job_info = (!opts.get_allow_trivial_move()).then(OwnedCompactionJobInfo::new);
4934 let job_info_ptr = job_info
4935 .as_mut()
4936 .map_or(ptr::null_mut(), OwnedCompactionJobInfo::as_mut_ptr);
4937
4938 // `output_path_id` 0 means the first configured DB path. RocksDB has no
4939 // named constant for it and the crate does not expose `cf_paths`
4940 // selection here, so it is fixed rather than a parameter nobody could
4941 // use meaningfully.
4942 let output_path_id = 0;
4943
4944 unsafe {
4945 match cf {
4946 None => ffi_try!(ffi::rocksdb_compact_files(
4947 self.inner.inner(),
4948 opts.inner.cast_const(),
4949 name_ptrs.as_ptr(),
4950 name_ptrs.len(),
4951 output_level,
4952 output_path_id,
4953 &raw mut output_names,
4954 &raw mut output_count,
4955 job_info_ptr,
4956 )),
4957 Some(cf) => ffi_try!(ffi::rocksdb_compact_files_cf(
4958 self.inner.inner(),
4959 cf.inner(),
4960 opts.inner.cast_const(),
4961 name_ptrs.as_ptr(),
4962 name_ptrs.len(),
4963 output_level,
4964 output_path_id,
4965 &raw mut output_names,
4966 &raw mut output_count,
4967 job_info_ptr,
4968 )),
4969 }
4970 }
4971
4972 // On failure RocksDB returns before touching either out-param, so this
4973 // only runs once the call succeeded and both are known good.
4974 let output_files = unsafe { collect_and_free_output_names(output_names, output_count) };
4975
4976 Ok(CompactFilesResult {
4977 output_files,
4978 job_info,
4979 })
4980 }
4981
4982 /// Obtains the LSM-tree meta data of the default column family, reporting
4983 /// only the levels and files `opts` selects.
4984 ///
4985 /// Unlike [`Self::get_column_family_metadata`], which returns only the
4986 /// totals, this reports every level and every SST file in it.
4987 pub fn get_column_family_metadata_with_options(
4988 &self,
4989 opts: &ColumnFamilyMetaDataOptions,
4990 ) -> Vec<LevelMetaData> {
4991 unsafe {
4992 let ptr = ffi::rocksdb_get_column_family_metadata_with_options(
4993 self.inner.inner(),
4994 opts.inner,
4995 );
4996 // The level and file handles borrow from `ptr`, so the returned
4997 // values own it and destroy it when the last one drops.
4998 levels_from_cf_metadata_owned(ptr)
4999 }
5000 }
5001
5002 /// Like [`Self::get_column_family_metadata_with_options`], for a single
5003 /// column family.
5004 pub fn get_column_family_metadata_cf_with_options(
5005 &self,
5006 cf: &impl AsColumnFamilyRef,
5007 opts: &ColumnFamilyMetaDataOptions,
5008 ) -> Vec<LevelMetaData> {
5009 unsafe {
5010 let ptr = ffi::rocksdb_get_column_family_metadata_cf_with_options(
5011 self.inner.inner(),
5012 cf.inner(),
5013 opts.inner,
5014 );
5015 levels_from_cf_metadata_owned(ptr)
5016 }
5017 }
5018
5019 /// Returns every file RocksDB needs in order to restore this DB, which is
5020 /// what a copy-based backup has to capture.
5021 ///
5022 /// This covers more than [`Self::live_files`] does: alongside the SST files
5023 /// it reports the WAL, manifest, options and `CURRENT` files, each with the
5024 /// size and, when `opts` asks for it, the checksum needed to verify a copy.
5025 ///
5026 /// # Errors
5027 ///
5028 /// Returns the RocksDB error if the file list cannot be gathered, for
5029 /// instance when a flush it needs to run fails.
5030 pub fn get_livefiles_storage_info(
5031 &self,
5032 opts: &LiveFilesStorageInfoOptions,
5033 ) -> Result<LiveFilesStorageInfo, Error> {
5034 unsafe {
5035 let ptr = ffi_try!(ffi::rocksdb_get_livefiles_storage_info(
5036 self.inner.inner(),
5037 opts.inner,
5038 ));
5039 if ptr.is_null() {
5040 return Err(Error::new(
5041 "Could not get live files storage info".to_owned(),
5042 ));
5043 }
5044 Ok(LiveFilesStorageInfo::from_ptr(ptr))
5045 }
5046 }
5047
5048 /// Returns every WAL file the DB currently knows about, oldest first,
5049 /// including the one being written to.
5050 ///
5051 /// # Errors
5052 ///
5053 /// Returns the RocksDB error if the WAL directory cannot be listed.
5054 pub fn get_sorted_wal_files(&self) -> Result<WalFiles, Error> {
5055 unsafe {
5056 let ptr = ffi_try!(ffi::rocksdb_get_sorted_wal_files(self.inner.inner()));
5057 if ptr.is_null() {
5058 return Err(Error::new("Could not get sorted WAL files".to_owned()));
5059 }
5060 Ok(WalFiles::from_ptr(ptr))
5061 }
5062 }
5063
5064 /// Returns the WAL file currently being written to.
5065 ///
5066 /// Reported as alive, with its size read from the filesystem, and with a
5067 /// [`start_sequence`](crate::wal::WalFile::start_sequence) of 0 rather than a
5068 /// real sequence number.
5069 ///
5070 /// # Errors
5071 ///
5072 /// Returns the RocksDB error if the current WAL file cannot be identified.
5073 pub fn get_current_wal_file(&self) -> Result<OwnedWalFile, Error> {
5074 unsafe {
5075 let ptr = ffi_try!(ffi::rocksdb_get_current_wal_file(self.inner.inner()));
5076 if ptr.is_null() {
5077 return Err(Error::new("Could not get current WAL file".to_owned()));
5078 }
5079 Ok(OwnedWalFile::from_ptr(ptr))
5080 }
5081 }
5082
5083 /// Starts recording every read and write to a trace file at `trace_path`,
5084 /// which [`Self::new_default_replayer`] can later replay against a DB.
5085 ///
5086 /// The trace file is written through the DB's own `Env` with default file
5087 /// options, so it is not rate limited.
5088 ///
5089 /// Call [`Self::end_trace`] to stop.
5090 ///
5091 /// Starting a second trace without ending the first does not fail. `StartTrace`
5092 /// installs the new tracer unconditionally, and the old one is dropped without
5093 /// its buffered records being written, so the first trace file is left
5094 /// truncated. Unlike [`Self::start_io_trace`] and
5095 /// [`Self::start_block_cache_trace`], which report `Busy`, this one is the
5096 /// caller's to get right.
5097 ///
5098 /// # Errors
5099 ///
5100 /// Returns the RocksDB error if the trace file cannot be created.
5101 pub fn start_trace<P: AsRef<Path>>(
5102 &self,
5103 opts: &TraceOptions,
5104 trace_path: P,
5105 ) -> Result<(), Error> {
5106 let cpath = to_cpath(trace_path)?;
5107 unsafe {
5108 // Null `env` uses the DB's own `Env`, which the DB already keeps
5109 // alive, and null `env_options` uses RocksDB's defaults. Passing an
5110 // `EnvOptions` here would be unsound: the trace writer keeps its
5111 // `rate_limiter` as a borrowed pointer for the life of the trace,
5112 // but the reference counting it lives behind stays in the caller's
5113 // handle.
5114 ffi_try!(ffi::rocksdb_start_trace(
5115 self.inner.inner(),
5116 ptr::null_mut(),
5117 ptr::null(),
5118 opts.inner,
5119 cpath.as_ptr(),
5120 ));
5121 }
5122 Ok(())
5123 }
5124
5125 /// Stops the trace started by [`Self::start_trace`] and closes the file.
5126 ///
5127 /// # Errors
5128 ///
5129 /// Returns the RocksDB error if no trace is running or the file cannot be
5130 /// closed cleanly.
5131 pub fn end_trace(&self) -> Result<(), Error> {
5132 unsafe {
5133 ffi_try!(ffi::rocksdb_end_trace(self.inner.inner()));
5134 }
5135 Ok(())
5136 }
5137
5138 /// Starts recording file system operations to a trace file at `trace_path`,
5139 /// for diagnosing IO behaviour rather than replaying queries.
5140 ///
5141 /// Written the same way as [`Self::start_trace`]. Call [`Self::end_io_trace`]
5142 /// to stop.
5143 ///
5144 /// # Errors
5145 ///
5146 /// Returns the RocksDB error if the trace file cannot be created, or `Busy` if
5147 /// an IO trace is already running.
5148 pub fn start_io_trace<P: AsRef<Path>>(
5149 &self,
5150 opts: &TraceOptions,
5151 trace_path: P,
5152 ) -> Result<(), Error> {
5153 let cpath = to_cpath(trace_path)?;
5154 unsafe {
5155 ffi_try!(ffi::rocksdb_start_io_trace(
5156 self.inner.inner(),
5157 ptr::null_mut(),
5158 ptr::null(),
5159 opts.inner,
5160 cpath.as_ptr(),
5161 ));
5162 }
5163 Ok(())
5164 }
5165
5166 /// Stops the IO trace started by [`Self::start_io_trace`].
5167 ///
5168 /// Does nothing if no IO trace is running.
5169 ///
5170 /// # Errors
5171 ///
5172 /// `EndIOTrace` always reports success, so this only fails if a future RocksDB
5173 /// gives it something to report. It returns a `Result` to keep that a
5174 /// non-breaking change.
5175 pub fn end_io_trace(&self) -> Result<(), Error> {
5176 unsafe {
5177 ffi_try!(ffi::rocksdb_end_io_trace(self.inner.inner()));
5178 }
5179 Ok(())
5180 }
5181
5182 /// Starts recording block cache accesses to a trace file at `trace_path`,
5183 /// which the `block_cache_trace_analyzer` tool can then simulate cache
5184 /// configurations against.
5185 ///
5186 /// Written the same way as [`Self::start_trace`]. Call
5187 /// [`Self::end_block_cache_trace`] to stop.
5188 ///
5189 /// # Errors
5190 ///
5191 /// Returns the RocksDB error if the trace file cannot be created, or `Busy` if
5192 /// a block cache trace is already running.
5193 pub fn start_block_cache_trace<P: AsRef<Path>>(
5194 &self,
5195 opts: &BlockCacheTraceOptions,
5196 writer_opts: &BlockCacheTraceWriterOptions,
5197 trace_path: P,
5198 ) -> Result<(), Error> {
5199 let cpath = to_cpath(trace_path)?;
5200 unsafe {
5201 ffi_try!(ffi::rocksdb_start_block_cache_trace_with_options(
5202 self.inner.inner(),
5203 ptr::null_mut(),
5204 ptr::null(),
5205 opts.inner,
5206 writer_opts.inner,
5207 cpath.as_ptr(),
5208 ));
5209 }
5210 Ok(())
5211 }
5212
5213 /// Stops the block cache trace started by [`Self::start_block_cache_trace`].
5214 ///
5215 /// Does nothing if no block cache trace is running.
5216 ///
5217 /// # Errors
5218 ///
5219 /// `EndBlockCacheTrace` always reports success, so this only fails if a future
5220 /// RocksDB gives it something to report. It returns a `Result` to keep that a
5221 /// non-breaking change.
5222 pub fn end_block_cache_trace(&self) -> Result<(), Error> {
5223 unsafe {
5224 ffi_try!(ffi::rocksdb_end_block_cache_trace(self.inner.inner()));
5225 }
5226 Ok(())
5227 }
5228
5229 /// Builds a replayer that replays the trace at `trace_path` against this DB.
5230 ///
5231 /// `column_families` must name every column family other than the default
5232 /// that the trace touched, or replay fails with
5233 /// `Corruption: Invalid Column Family ID.`. The default column family is
5234 /// always included, so an empty list replays it alone.
5235 ///
5236 /// # Errors
5237 ///
5238 /// Returns the RocksDB error if the trace file cannot be opened or read.
5239 pub fn new_default_replayer<'cf, W, I, P>(
5240 &self,
5241 column_families: I,
5242 trace_path: P,
5243 ) -> Result<Replayer<'_>, Error>
5244 where
5245 W: AsColumnFamilyRef + 'cf,
5246 I: IntoIterator<Item = &'cf W>,
5247 P: AsRef<Path>,
5248 {
5249 let cpath = to_cpath(trace_path)?;
5250 unsafe {
5251 Replayer::create_default(
5252 self.inner.inner(),
5253 column_families,
5254 ptr::null_mut(),
5255 ptr::null(),
5256 &cpath,
5257 )
5258 }
5259 }
5260
5261 /// Returns a list of all table files with their level, start key
5262 /// and end key
5263 pub fn live_files(&self) -> Result<Vec<LiveFile>, Error> {
5264 unsafe {
5265 let livefiles_ptr = ffi::rocksdb_livefiles(self.inner.inner());
5266 if livefiles_ptr.is_null() {
5267 Err(Error::new("Could not get live files".to_owned()))
5268 } else {
5269 let files = LiveFile::from_rocksdb_livefiles_ptr(livefiles_ptr);
5270
5271 // destroy livefiles metadata(s)
5272 ffi::rocksdb_livefiles_destroy(livefiles_ptr);
5273
5274 // return
5275 Ok(files)
5276 }
5277 }
5278 }
5279
5280 /// Delete sst files whose keys are entirely in the given range.
5281 ///
5282 /// Could leave some keys in the range which are in files which are not
5283 /// entirely in the range.
5284 ///
5285 /// Note: L0 files are left regardless of whether they're in the range.
5286 ///
5287 /// SnapshotWithThreadModes before the delete might not see the data in the given range.
5288 pub fn delete_file_in_range<K: AsRef<[u8]>>(&self, from: K, to: K) -> Result<(), Error> {
5289 let from = from.as_ref();
5290 let to = to.as_ref();
5291 unsafe {
5292 ffi_try!(ffi::rocksdb_delete_file_in_range(
5293 self.inner.inner(),
5294 from.as_ptr() as *const c_char,
5295 from.len() as size_t,
5296 to.as_ptr() as *const c_char,
5297 to.len() as size_t,
5298 ));
5299 Ok(())
5300 }
5301 }
5302
5303 /// Same as `delete_file_in_range` but only for specific column family
5304 pub fn delete_file_in_range_cf<K: AsRef<[u8]>>(
5305 &self,
5306 cf: &impl AsColumnFamilyRef,
5307 from: K,
5308 to: K,
5309 ) -> Result<(), Error> {
5310 let from = from.as_ref();
5311 let to = to.as_ref();
5312 unsafe {
5313 ffi_try!(ffi::rocksdb_delete_file_in_range_cf(
5314 self.inner.inner(),
5315 cf.inner(),
5316 from.as_ptr() as *const c_char,
5317 from.len() as size_t,
5318 to.as_ptr() as *const c_char,
5319 to.len() as size_t,
5320 ));
5321 Ok(())
5322 }
5323 }
5324
5325 /// Request stopping background work, if wait is true wait until it's done.
5326 pub fn cancel_all_background_work(&self, wait: bool) {
5327 unsafe {
5328 ffi::rocksdb_cancel_all_background_work(self.inner.inner(), c_uchar::from(wait));
5329 }
5330 }
5331
5332 /// Marks the column family as dropped in RocksDB.
5333 ///
5334 /// Deliberately does not take ownership of the handle. Callers must take
5335 /// the handle out of their map first, so that only one caller can ever
5336 /// *destroy* a given handle, and must put it back if this fails: destroying
5337 /// it on failure would leave the column family still present in the DB with
5338 /// no reachable handle, so the only way to touch it again would be to
5339 /// reopen the database.
5340 ///
5341 /// Taking it out of the map does not make the caller the only *reader*. In
5342 /// `MultiThreaded` mode `cf_handle` clones the same `Arc`, so other threads
5343 /// can still hold a live `BoundColumnFamily` for this handle. That is fine:
5344 /// the refcount keeps the handle alive until the last of them is gone.
5345 fn mark_column_family_dropped(
5346 &self,
5347 cf_inner: *mut ffi::rocksdb_column_family_handle_t,
5348 ) -> Result<(), Error> {
5349 unsafe {
5350 ffi_try!(ffi::rocksdb_drop_column_family(
5351 self.inner.inner(),
5352 cf_inner
5353 ));
5354 }
5355 Ok(())
5356 }
5357
5358 /// Increase the full_history_ts of column family. The new ts_low value should
5359 /// be newer than current full_history_ts value.
5360 /// If another thread updates full_history_ts_low concurrently to a higher
5361 /// timestamp than the requested ts_low, a try again error will be returned.
5362 pub fn increase_full_history_ts_low<S: AsRef<[u8]>>(
5363 &self,
5364 cf: &impl AsColumnFamilyRef,
5365 ts: S,
5366 ) -> Result<(), Error> {
5367 let ts = ts.as_ref();
5368 unsafe {
5369 ffi_try!(ffi::rocksdb_increase_full_history_ts_low(
5370 self.inner.inner(),
5371 cf.inner(),
5372 ts.as_ptr() as *const c_char,
5373 ts.len() as size_t,
5374 ));
5375 Ok(())
5376 }
5377 }
5378
5379 /// Get current full_history_ts value.
5380 pub fn get_full_history_ts_low(&self, cf: &impl AsColumnFamilyRef) -> Result<Vec<u8>, Error> {
5381 unsafe {
5382 let mut ts_lowlen = 0;
5383 let ts = ffi_try!(ffi::rocksdb_get_full_history_ts_low(
5384 self.inner.inner(),
5385 cf.inner(),
5386 &raw mut ts_lowlen,
5387 ));
5388
5389 if ts.is_null() {
5390 Err(Error::new("Could not get full_history_ts_low".to_owned()))
5391 } else {
5392 let mut vec = vec![0; ts_lowlen];
5393 ptr::copy_nonoverlapping(ts.cast::<u8>(), vec.as_mut_ptr(), ts_lowlen);
5394 ffi::rocksdb_free(ts as *mut c_void);
5395 Ok(vec)
5396 }
5397 }
5398 }
5399
5400 /// Returns the DB identity. This is typically ASCII bytes, but that is not guaranteed.
5401 pub fn get_db_identity(&self) -> Result<Vec<u8>, Error> {
5402 unsafe {
5403 let mut length: usize = 0;
5404 let identity_ptr = ffi::rocksdb_get_db_identity(self.inner.inner(), &raw mut length);
5405 let identity_vec = raw_data(identity_ptr, length);
5406 ffi::rocksdb_free(identity_ptr as *mut c_void);
5407 // In RocksDB: get_db_identity copies a std::string so it should not fail, but
5408 // the API allows it to be overridden, so it might
5409 identity_vec.ok_or_else(|| Error::new("get_db_identity returned NULL".to_string()))
5410 }
5411 }
5412}
5413
5414impl<I: DBInner> DBCommon<SingleThreaded, I> {
5415 /// Creates column family with given name and options
5416 pub fn create_cf<N: AsRef<str>>(&mut self, name: N, opts: &Options) -> Result<(), Error> {
5417 let inner = self.create_inner_cf_handle(name.as_ref(), opts)?;
5418 self.cfs
5419 .cfs
5420 .insert(name.as_ref().to_string(), ColumnFamily { inner });
5421 Ok(())
5422 }
5423
5424 /// Creates a column family whose entries expire after `ttl`, on a DB that was
5425 /// opened with a TTL.
5426 ///
5427 /// [`ColumnFamilyDescriptor::new_with_ttl`] sets this at open time. This is the
5428 /// way to add such a family to a DB that is already open.
5429 ///
5430 /// # Errors
5431 ///
5432 /// Errors if the DB was not opened with [`DB::open_with_ttl`] or one of the
5433 /// `open_cf*_with_ttl` functions, since RocksDB would otherwise treat the handle
5434 /// as a type it is not. Also returns the RocksDB error if the family already
5435 /// exists or cannot be created, and errors if the name contains an interior NUL
5436 /// byte.
5437 pub fn create_cf_with_ttl<N: AsRef<str>>(
5438 &mut self,
5439 name: N,
5440 opts: &Options,
5441 ttl: ColumnFamilyTtl,
5442 ) -> Result<(), Error> {
5443 let inner = self.create_inner_cf_handle_with_ttl(name.as_ref(), opts, ttl)?;
5444 self.cfs
5445 .cfs
5446 .insert(name.as_ref().to_string(), ColumnFamily { inner });
5447 Ok(())
5448 }
5449
5450 /// Creates the named column families, all sharing `opts`.
5451 ///
5452 /// This saves one options file write over calling
5453 /// [`create_cf`](Self::create_cf) in a loop. It is not a saving on manifest
5454 /// writes, and it is not atomic.
5455 ///
5456 /// # Errors
5457 ///
5458 /// Returns the RocksDB error if a family already exists or cannot be created,
5459 /// and errors if a name contains an interior NUL byte.
5460 ///
5461 /// RocksDB creates the families in order and stops at the first failure, so on
5462 /// error the families named before the failing one already exist and stay that
5463 /// way. Those are recorded here as usual and can be used or dropped, so retrying
5464 /// with the same list will fail again on the ones that now exist.
5465 pub fn create_cfs<Iter, N>(&mut self, names: Iter, opts: &Options) -> Result<(), Error>
5466 where
5467 Iter: IntoIterator<Item = N>,
5468 N: AsRef<str>,
5469 {
5470 let names = convert_cf_names(names)?;
5471 let created = self.create_inner_cf_handles(&names, opts);
5472 for ((name, _), inner) in names.into_iter().zip(created.handles) {
5473 self.cfs.cfs.insert(name, ColumnFamily { inner });
5474 }
5475 match created.error {
5476 Some(err) => Err(err),
5477 None => Ok(()),
5478 }
5479 }
5480
5481 #[doc = include_str!("db_create_column_family_with_import.md")]
5482 pub fn create_column_family_with_import<N: AsRef<str>>(
5483 &mut self,
5484 options: &Options,
5485 column_family_name: N,
5486 import_options: &ImportColumnFamilyOptions,
5487 metadata: &ExportImportFilesMetaData,
5488 ) -> Result<(), Error> {
5489 let name = column_family_name.as_ref();
5490 let c_name = CString::new(name).map_err(|err| {
5491 Error::new(format!(
5492 "Failed to convert name to CString while importing column family: {err}"
5493 ))
5494 })?;
5495 let inner = unsafe {
5496 ffi_try!(ffi::rocksdb_create_column_family_with_import(
5497 self.inner.inner(),
5498 options.inner,
5499 c_name.as_ptr(),
5500 import_options.inner,
5501 metadata.inner
5502 ))
5503 };
5504 self.cfs
5505 .cfs
5506 .insert(column_family_name.as_ref().into(), ColumnFamily { inner });
5507 Ok(())
5508 }
5509
5510 /// Drops the column family with the given name
5511 pub fn drop_cf(&mut self, name: &str) -> Result<(), Error> {
5512 let Some(cf) = self.cfs.cfs.remove(name) else {
5513 return Err(Error::new(format!("Invalid column family: {name}")));
5514 };
5515 match self.mark_column_family_dropped(cf.inner) {
5516 // `cf` is dropped here. In single-threaded mode that destroys the
5517 // handle; in `MultiThreaded` mode it drops one `Arc` reference and
5518 // the handle is destroyed once the last `BoundColumnFamily` clone
5519 // handed out by `cf_handle` is gone.
5520 Ok(()) => Ok(()),
5521 Err(e) => {
5522 // The column family is still there, so put the handle back
5523 // rather than destroying the only way to reach it.
5524 self.cfs.cfs.insert(name.to_owned(), cf);
5525 Err(e)
5526 }
5527 }
5528 }
5529
5530 /// Returns the underlying column family handle
5531 pub fn cf_handle(&self, name: &str) -> Option<&ColumnFamily> {
5532 self.cfs.cfs.get(name)
5533 }
5534
5535 /// Returns the list of column families currently open.
5536 ///
5537 /// The order of names is unspecified and may vary between calls.
5538 pub fn cf_names(&self) -> Vec<String> {
5539 self.cfs.cfs.keys().cloned().collect()
5540 }
5541}
5542
5543impl<I: DBInner> DBCommon<MultiThreaded, I> {
5544 /// Creates column family with given name and options
5545 pub fn create_cf<N: AsRef<str>>(&self, name: N, opts: &Options) -> Result<(), Error> {
5546 // Note that we acquire the cfs lock before inserting: otherwise we might race
5547 // another caller who observed the handle as missing.
5548 let mut cfs = self.cfs.cfs.write();
5549 let inner = self.create_inner_cf_handle(name.as_ref(), opts)?;
5550 cfs.insert(
5551 name.as_ref().to_string(),
5552 Arc::new(UnboundColumnFamily { inner }),
5553 );
5554 Ok(())
5555 }
5556
5557 /// Creates a column family whose entries expire after `ttl`, on a DB that was
5558 /// opened with a TTL.
5559 ///
5560 /// [`ColumnFamilyDescriptor::new_with_ttl`] sets this at open time. This is the
5561 /// way to add such a family to a DB that is already open.
5562 ///
5563 /// # Errors
5564 ///
5565 /// Errors if the DB was not opened with [`DB::open_with_ttl`] or one of the
5566 /// `open_cf*_with_ttl` functions, since RocksDB would otherwise treat the handle
5567 /// as a type it is not. Also returns the RocksDB error if the family already
5568 /// exists or cannot be created, and errors if the name contains an interior NUL
5569 /// byte.
5570 pub fn create_cf_with_ttl<N: AsRef<str>>(
5571 &self,
5572 name: N,
5573 opts: &Options,
5574 ttl: ColumnFamilyTtl,
5575 ) -> Result<(), Error> {
5576 // Note that we acquire the cfs lock before inserting: otherwise we might race
5577 // another caller who observed the handle as missing.
5578 let mut cfs = self.cfs.cfs.write();
5579 let inner = self.create_inner_cf_handle_with_ttl(name.as_ref(), opts, ttl)?;
5580 cfs.insert(
5581 name.as_ref().to_string(),
5582 Arc::new(UnboundColumnFamily { inner }),
5583 );
5584 Ok(())
5585 }
5586
5587 /// Creates the named column families, all sharing `opts`.
5588 ///
5589 /// See [`DBCommon::create_cfs`](DBCommon::<SingleThreaded, I>::create_cfs),
5590 /// including the note that this is not atomic.
5591 ///
5592 /// # Errors
5593 ///
5594 /// Returns the RocksDB error if a family already exists or cannot be created,
5595 /// and errors if a name contains an interior NUL byte. On error the families
5596 /// named before the failing one already exist and are recorded here.
5597 pub fn create_cfs<Iter, N>(&self, names: Iter, opts: &Options) -> Result<(), Error>
5598 where
5599 Iter: IntoIterator<Item = N>,
5600 N: AsRef<str>,
5601 {
5602 let names = convert_cf_names(names)?;
5603 // Note that we acquire the cfs lock before creating: otherwise we might race
5604 // another caller who observed the handles as missing.
5605 let mut cfs = self.cfs.cfs.write();
5606 let created = self.create_inner_cf_handles(&names, opts);
5607 for ((name, _), inner) in names.into_iter().zip(created.handles) {
5608 cfs.insert(name, Arc::new(UnboundColumnFamily { inner }));
5609 }
5610 match created.error {
5611 Some(err) => Err(err),
5612 None => Ok(()),
5613 }
5614 }
5615
5616 #[doc = include_str!("db_create_column_family_with_import.md")]
5617 pub fn create_column_family_with_import<N: AsRef<str>>(
5618 &self,
5619 options: &Options,
5620 column_family_name: N,
5621 import_options: &ImportColumnFamilyOptions,
5622 metadata: &ExportImportFilesMetaData,
5623 ) -> Result<(), Error> {
5624 // Acquire CF lock upfront, before creating the CF, to avoid a race with concurrent creators
5625 let mut cfs = self.cfs.cfs.write();
5626 let name = column_family_name.as_ref();
5627 let c_name = CString::new(name).map_err(|err| {
5628 Error::new(format!(
5629 "Failed to convert name to CString while importing column family: {err}"
5630 ))
5631 })?;
5632 let inner = unsafe {
5633 ffi_try!(ffi::rocksdb_create_column_family_with_import(
5634 self.inner.inner(),
5635 options.inner,
5636 c_name.as_ptr(),
5637 import_options.inner,
5638 metadata.inner
5639 ))
5640 };
5641 cfs.insert(
5642 column_family_name.as_ref().to_string(),
5643 Arc::new(UnboundColumnFamily { inner }),
5644 );
5645 Ok(())
5646 }
5647
5648 /// Drops the column family with the given name by internally locking the inner column
5649 /// family map. This avoids needing `&mut self` reference
5650 pub fn drop_cf(&self, name: &str) -> Result<(), Error> {
5651 // Take the handle out under the write lock before touching RocksDB.
5652 // Looking it up under a read lock and removing it afterwards would let
5653 // two concurrent callers observe the same handle: the first would drop
5654 // and destroy it, and the second would then hand a freed pointer to
5655 // `rocksdb_drop_column_family`.
5656 let Some(cf) = self.cfs.cfs.write().remove(name) else {
5657 return Err(Error::new(format!("Invalid column family: {name}")));
5658 };
5659 match self.mark_column_family_dropped(cf.inner) {
5660 // `cf` is dropped here. In single-threaded mode that destroys the
5661 // handle; in `MultiThreaded` mode it drops one `Arc` reference and
5662 // the handle is destroyed once the last `BoundColumnFamily` clone
5663 // handed out by `cf_handle` is gone.
5664 Ok(()) => Ok(()),
5665 Err(e) => {
5666 // The column family is still there, so put the handle back
5667 // rather than destroying the only way to reach it.
5668 self.cfs.cfs.write().insert(name.to_owned(), cf);
5669 Err(e)
5670 }
5671 }
5672 }
5673
5674 /// Returns the underlying column family handle
5675 pub fn cf_handle(&'_ self, name: &str) -> Option<Arc<BoundColumnFamily<'_>>> {
5676 self.cfs
5677 .cfs
5678 .read()
5679 .get(name)
5680 .cloned()
5681 .map(UnboundColumnFamily::bound_column_family)
5682 }
5683
5684 /// Returns the list of column families currently open.
5685 ///
5686 /// The order of names is unspecified and may vary between calls.
5687 pub fn cf_names(&self) -> Vec<String> {
5688 self.cfs.cfs.read().keys().cloned().collect()
5689 }
5690}
5691
5692impl<T: ThreadMode, I: DBInner> Drop for DBCommon<T, I> {
5693 fn drop(&mut self) {
5694 self.cfs.drop_all_cfs_internal();
5695 }
5696}
5697
5698impl<T: ThreadMode, I: DBInner> fmt::Debug for DBCommon<T, I> {
5699 fn fmt(&self, f: &mut fmt::Formatter) -> fmt::Result {
5700 write!(f, "RocksDB {{ path: {} }}", self.path().display())
5701 }
5702}
5703
5704/// The metadata that describes a column family.
5705#[derive(Debug, Clone)]
5706pub struct ColumnFamilyMetaData {
5707 // The size of this column family in bytes, which is equal to the sum of
5708 // the file size of its "levels".
5709 pub size: u64,
5710 // The name of the column family.
5711 pub name: String,
5712 // The number of files in this column family.
5713 pub file_count: usize,
5714}
5715
5716/// The metadata that describes a SST file
5717#[derive(Debug, Clone)]
5718pub struct LiveFile {
5719 /// Name of the column family the file belongs to
5720 pub column_family_name: String,
5721 /// Name of the file
5722 pub name: String,
5723 /// The directory containing the file, without a trailing '/'. This could be
5724 /// a DB path, wal_dir, etc.
5725 pub directory: String,
5726 /// Size of the file
5727 pub size: usize,
5728 /// Level at which this file resides
5729 pub level: i32,
5730 /// Smallest user defined key in the file
5731 pub start_key: Option<Vec<u8>>,
5732 /// Largest user defined key in the file
5733 pub end_key: Option<Vec<u8>>,
5734 pub smallest_seqno: u64,
5735 pub largest_seqno: u64,
5736 /// Number of entries/alive keys in the file
5737 pub num_entries: u64,
5738 /// Number of deletions/tomb key(s) in the file
5739 pub num_deletions: u64,
5740}
5741
5742impl LiveFile {
5743 /// Create a `Vec<LiveFile>` from a `rocksdb_livefiles_t` pointer
5744 pub(crate) fn from_rocksdb_livefiles_ptr(
5745 files: *const ffi::rocksdb_livefiles_t,
5746 ) -> Vec<LiveFile> {
5747 unsafe {
5748 let n = ffi::rocksdb_livefiles_count(files);
5749
5750 let mut livefiles = Vec::with_capacity(n as usize);
5751 let mut key_size: usize = 0;
5752
5753 for i in 0..n {
5754 // rocksdb_livefiles_* returns pointers to strings, not copies
5755 let column_family_name =
5756 from_cstr_without_free(ffi::rocksdb_livefiles_column_family_name(files, i));
5757 let name = from_cstr_without_free(ffi::rocksdb_livefiles_name(files, i));
5758 let directory = from_cstr_without_free(ffi::rocksdb_livefiles_directory(files, i));
5759 let size = ffi::rocksdb_livefiles_size(files, i);
5760 let level = ffi::rocksdb_livefiles_level(files, i);
5761
5762 // get smallest key inside file
5763 let smallest_key = ffi::rocksdb_livefiles_smallestkey(files, i, &raw mut key_size);
5764 let smallest_key = raw_data(smallest_key, key_size);
5765
5766 // get largest key inside file
5767 let largest_key = ffi::rocksdb_livefiles_largestkey(files, i, &raw mut key_size);
5768 let largest_key = raw_data(largest_key, key_size);
5769
5770 livefiles.push(LiveFile {
5771 column_family_name,
5772 name,
5773 directory,
5774 size,
5775 level,
5776 start_key: smallest_key,
5777 end_key: largest_key,
5778 largest_seqno: ffi::rocksdb_livefiles_largest_seqno(files, i),
5779 smallest_seqno: ffi::rocksdb_livefiles_smallest_seqno(files, i),
5780 num_entries: ffi::rocksdb_livefiles_entries(files, i),
5781 num_deletions: ffi::rocksdb_livefiles_deletions(files, i),
5782 });
5783 }
5784
5785 livefiles
5786 }
5787 }
5788}
5789
5790struct LiveFileGuard(*mut rocksdb_livefile_t);
5791
5792impl LiveFileGuard {
5793 fn into_raw(mut self) -> *mut rocksdb_livefile_t {
5794 let ptr = self.0;
5795 self.0 = ptr::null_mut();
5796 ptr
5797 }
5798}
5799
5800impl Drop for LiveFileGuard {
5801 fn drop(&mut self) {
5802 if !self.0.is_null() {
5803 unsafe {
5804 rocksdb_livefile_destroy(self.0);
5805 }
5806 }
5807 }
5808}
5809
5810struct LiveFilesGuard(*mut rocksdb_livefiles_t);
5811
5812impl LiveFilesGuard {
5813 fn into_raw(mut self) -> *mut rocksdb_livefiles_t {
5814 let ptr = self.0;
5815 self.0 = ptr::null_mut();
5816 ptr
5817 }
5818}
5819
5820impl Drop for LiveFilesGuard {
5821 fn drop(&mut self) {
5822 if !self.0.is_null() {
5823 unsafe {
5824 rocksdb_livefiles_destroy(self.0);
5825 }
5826 }
5827 }
5828}
5829
5830/// Metadata returned as output from [`Checkpoint::export_column_family`][export_column_family] and
5831/// used as input to [`DB::create_column_family_with_import`].
5832///
5833/// [export_column_family]: crate::checkpoint::Checkpoint::export_column_family
5834#[derive(Debug)]
5835pub struct ExportImportFilesMetaData {
5836 pub(crate) inner: *mut ffi::rocksdb_export_import_files_metadata_t,
5837}
5838
5839impl ExportImportFilesMetaData {
5840 pub fn get_db_comparator_name(&self) -> String {
5841 unsafe {
5842 let c_name =
5843 ffi::rocksdb_export_import_files_metadata_get_db_comparator_name(self.inner);
5844 from_cstr_and_free(c_name)
5845 }
5846 }
5847
5848 pub fn set_db_comparator_name(&mut self, name: &str) {
5849 let c_name = CString::new(name.as_bytes()).unwrap();
5850 unsafe {
5851 ffi::rocksdb_export_import_files_metadata_set_db_comparator_name(
5852 self.inner,
5853 c_name.as_ptr(),
5854 );
5855 };
5856 }
5857
5858 pub fn get_files(&self) -> Vec<LiveFile> {
5859 unsafe {
5860 let livefiles_ptr = ffi::rocksdb_export_import_files_metadata_get_files(self.inner);
5861 let files = LiveFile::from_rocksdb_livefiles_ptr(livefiles_ptr);
5862 ffi::rocksdb_livefiles_destroy(livefiles_ptr);
5863 files
5864 }
5865 }
5866
5867 pub fn set_files(&mut self, files: &[LiveFile]) -> Result<(), Error> {
5868 // Use a non-null empty pointer for zero-length keys
5869 static EMPTY: [u8; 0] = [];
5870 let empty_ptr = EMPTY.as_ptr() as *const libc::c_char;
5871
5872 unsafe {
5873 let live_files = LiveFilesGuard(ffi::rocksdb_livefiles_create());
5874
5875 for file in files {
5876 let live_file = LiveFileGuard(ffi::rocksdb_livefile_create());
5877 ffi::rocksdb_livefile_set_level(live_file.0, file.level);
5878
5879 // SAFETY: C strings are copied inside the FFI layer so do not need to be kept alive
5880 let c_cf_name = CString::new(file.column_family_name.as_str()).map_err(|err| {
5881 Error::new(format!("Unable to convert column family to CString: {err}"))
5882 })?;
5883 ffi::rocksdb_livefile_set_column_family_name(live_file.0, c_cf_name.as_ptr());
5884
5885 let c_name = CString::new(file.name.as_str()).map_err(|err| {
5886 Error::new(format!("Unable to convert file name to CString: {err}"))
5887 })?;
5888 ffi::rocksdb_livefile_set_name(live_file.0, c_name.as_ptr());
5889
5890 let c_directory = CString::new(file.directory.as_str()).map_err(|err| {
5891 Error::new(format!("Unable to convert directory to CString: {err}"))
5892 })?;
5893 ffi::rocksdb_livefile_set_directory(live_file.0, c_directory.as_ptr());
5894
5895 ffi::rocksdb_livefile_set_size(live_file.0, file.size);
5896
5897 let (start_key_ptr, start_key_len) = match &file.start_key {
5898 None => (empty_ptr, 0),
5899 Some(key) => (key.as_ptr() as *const libc::c_char, key.len()),
5900 };
5901 ffi::rocksdb_livefile_set_smallest_key(live_file.0, start_key_ptr, start_key_len);
5902
5903 let (largest_key_ptr, largest_key_len) = match &file.end_key {
5904 None => (empty_ptr, 0),
5905 Some(key) => (key.as_ptr() as *const libc::c_char, key.len()),
5906 };
5907 ffi::rocksdb_livefile_set_largest_key(
5908 live_file.0,
5909 largest_key_ptr,
5910 largest_key_len,
5911 );
5912 ffi::rocksdb_livefile_set_smallest_seqno(live_file.0, file.smallest_seqno);
5913 ffi::rocksdb_livefile_set_largest_seqno(live_file.0, file.largest_seqno);
5914 ffi::rocksdb_livefile_set_num_entries(live_file.0, file.num_entries);
5915 ffi::rocksdb_livefile_set_num_deletions(live_file.0, file.num_deletions);
5916
5917 // moves ownership of live_files into live_file
5918 ffi::rocksdb_livefiles_add(live_files.0, live_file.into_raw());
5919 }
5920
5921 // moves ownership of live_files into inner
5922 ffi::rocksdb_export_import_files_metadata_set_files(self.inner, live_files.into_raw());
5923 Ok(())
5924 }
5925 }
5926}
5927
5928impl Default for ExportImportFilesMetaData {
5929 fn default() -> Self {
5930 let inner = unsafe { ffi::rocksdb_export_import_files_metadata_create() };
5931 assert!(
5932 !inner.is_null(),
5933 "Could not create rocksdb_export_import_files_metadata_t"
5934 );
5935
5936 Self { inner }
5937 }
5938}
5939
5940impl Drop for ExportImportFilesMetaData {
5941 fn drop(&mut self) {
5942 unsafe {
5943 ffi::rocksdb_export_import_files_metadata_destroy(self.inner);
5944 }
5945 }
5946}
5947
5948unsafe impl Send for ExportImportFilesMetaData {}
5949unsafe impl Sync for ExportImportFilesMetaData {}
5950
5951/// Converts a TTL to the `int` seconds count RocksDB's TTL API takes,
5952/// saturating instead of wrapping.
5953///
5954/// `Duration::as_secs` is a `u64`, so a plain `as i32` cast wraps: a TTL of
5955/// `Duration::from_secs(4_294_967_301)` (~136 years, i.e. "effectively never")
5956/// became `5`, and RocksDB then compaction-deleted the whole column family a few
5957/// seconds after the data was written.
5958///
5959/// Clamping to `i32::MAX` (~68 years) rather than mapping an over-large TTL to
5960/// RocksDB's never-expire sentinel (`ttl <= 0`, see `DBWithTTLImpl::IsStale`) is
5961/// deliberate: silently turning a finite TTL the caller asked for into "keep
5962/// forever" is a worse surprise than expiring it 68 years out, and `i32::MAX` is
5963/// the longest TTL the C API can express anyway.
5964fn ttl_to_seconds(ttl: Duration) -> c_int {
5965 c_int::try_from(ttl.as_secs()).unwrap_or(c_int::MAX)
5966}
5967
5968/// Resolves a column family's TTL against the TTL the DB was opened with.
5969///
5970/// [`ColumnFamilyTtl::Disabled`] maps to the longest TTL the C API can express rather
5971/// than RocksDB's never-expire sentinel, for the reason on [`ttl_to_seconds`].
5972fn cf_ttl_to_seconds(ttl: ColumnFamilyTtl, db_ttl: Duration) -> c_int {
5973 match ttl {
5974 ColumnFamilyTtl::Disabled => c_int::MAX,
5975 ColumnFamilyTtl::Duration(duration) => ttl_to_seconds(duration),
5976 ColumnFamilyTtl::SameAsDb => ttl_to_seconds(db_ttl),
5977 }
5978}
5979
5980/// Converts column family names to C strings, keeping the given order.
5981///
5982/// # Errors
5983///
5984/// Errors if a name contains an interior NUL byte.
5985fn convert_cf_names<I, N>(names: I) -> Result<Vec<(String, CString)>, Error>
5986where
5987 I: IntoIterator<Item = N>,
5988 N: AsRef<str>,
5989{
5990 names
5991 .into_iter()
5992 .map(|name| {
5993 let name = name.as_ref();
5994 let cname = CString::new(name).map_err(|err| {
5995 Error::new(format!(
5996 "Failed to convert column family name to CString: {err}"
5997 ))
5998 })?;
5999 Ok((name.to_owned(), cname))
6000 })
6001 .collect()
6002}
6003
6004/// The option count as the `int` the C API takes.
6005///
6006/// # Errors
6007///
6008/// Errors rather than truncating if there are more options than an `int` can count.
6009fn option_count(opts: &[(CString, CString)]) -> Result<c_int, Error> {
6010 c_int::try_from(opts.len())
6011 .map_err(|_| Error::new(format!("Too many options to set at once: {}", opts.len())))
6012}
6013
6014fn convert_options(opts: &[(&str, &str)]) -> Result<Vec<(CString, CString)>, Error> {
6015 opts.iter()
6016 .map(|(name, value)| {
6017 let cname = match CString::new(name.as_bytes()) {
6018 Ok(cname) => cname,
6019 Err(e) => return Err(Error::new(format!("Invalid option name `{e}`"))),
6020 };
6021 let cvalue = match CString::new(value.as_bytes()) {
6022 Ok(cvalue) => cvalue,
6023 Err(e) => return Err(Error::new(format!("Invalid option value: `{e}`"))),
6024 };
6025 Ok((cname, cvalue))
6026 })
6027 .collect()
6028}
6029
6030/// Borrows each key as the pointer and length pair the multi-get calls want.
6031///
6032/// The returned pointers borrow `keys`, so `keys` has to outlive them.
6033fn key_ptrs_and_sizes<K: AsRef<[u8]>>(keys: &[K]) -> (Vec<*const c_char>, Vec<usize>) {
6034 keys.iter()
6035 .map(|k| {
6036 let key = k.as_ref();
6037 (key.as_ptr().cast::<c_char>(), key.len())
6038 })
6039 .unzip()
6040}
6041
6042/// The five output arrays `rocksdb_multi_get_*_with_ts` fills in.
6043///
6044/// Kept together because they are only ever allocated, passed and consumed as a set,
6045/// and because the unsafe `set_len` after the call has to cover all five or none.
6046struct MultiGetTsOut {
6047 values: Vec<*mut c_char>,
6048 values_sizes: Vec<usize>,
6049 timestamps: Vec<*mut c_char>,
6050 timestamps_sizes: Vec<usize>,
6051 errors: Vec<*mut c_char>,
6052}
6053
6054impl MultiGetTsOut {
6055 fn with_capacity(n: usize) -> Self {
6056 Self {
6057 values: Vec::with_capacity(n),
6058 values_sizes: Vec::with_capacity(n),
6059 timestamps: Vec::with_capacity(n),
6060 timestamps_sizes: Vec::with_capacity(n),
6061 errors: Vec::with_capacity(n),
6062 }
6063 }
6064
6065 /// Publishes the `n` entries RocksDB wrote into the spare capacity.
6066 ///
6067 /// # Safety
6068 ///
6069 /// `n` must be the key count the call was given, and the call must have returned.
6070 /// RocksDB writes every one of the five arrays at every index, either a pointer it
6071 /// allocated or null (c.cc:2694-2711), so no element is left uninitialised.
6072 unsafe fn assume_filled(&mut self, n: usize) {
6073 unsafe {
6074 self.values.set_len(n);
6075 self.values_sizes.set_len(n);
6076 self.timestamps.set_len(n);
6077 self.timestamps_sizes.set_len(n);
6078 self.errors.set_len(n);
6079 }
6080 }
6081
6082 /// Takes ownership of every buffer RocksDB allocated and pairs it with its key.
6083 ///
6084 /// A null value with no error is a key that was not found. An error is reported
6085 /// with both buffers already null, so there is nothing to free on that path.
6086 fn into_results(self) -> Vec<Result<Option<TimestampedValue>, Error>> {
6087 self.values
6088 .into_iter()
6089 .zip(self.values_sizes)
6090 .zip(self.timestamps.into_iter().zip(self.timestamps_sizes))
6091 .zip(self.errors)
6092 .map(|(((value, vallen), (ts, tslen)), err)| {
6093 if !err.is_null() {
6094 return Err(convert_rocksdb_error(err));
6095 }
6096 if value.is_null() {
6097 return Ok(None);
6098 }
6099 // SAFETY: RocksDB allocated both with `CopyString` at the reported
6100 // lengths and nothing else frees them.
6101 unsafe {
6102 Ok(Some(TimestampedValue {
6103 value: CSlice::from_raw_parts(value, vallen),
6104 timestamp: CSlice::from_raw_parts(ts, tslen),
6105 }))
6106 }
6107 })
6108 .collect()
6109 }
6110}
6111
6112pub(crate) fn convert_values(
6113 values: Vec<*mut c_char>,
6114 values_sizes: Vec<usize>,
6115 errors: Vec<*mut c_char>,
6116) -> Vec<Result<Option<Vec<u8>>, Error>> {
6117 values
6118 .into_iter()
6119 .zip(values_sizes)
6120 .zip(errors)
6121 .map(|((v, s), e)| {
6122 if e.is_null() {
6123 let value = unsafe { crate::ffi_util::raw_data(v, s) };
6124 unsafe {
6125 ffi::rocksdb_free(v as *mut c_void);
6126 }
6127 Ok(value)
6128 } else {
6129 Err(convert_rocksdb_error(e))
6130 }
6131 })
6132 .collect()
6133}
6134
6135#[cfg(test)]
6136mod tests {
6137 use crate::{ColumnFamilyDescriptor, DB, Options};
6138
6139 /// One `rocksdb_create_iterators` call applies a single `ReadOptions` to
6140 /// every iterator it builds, and RocksDB's `DBIter` keeps raw `Slice*`
6141 /// into that object for the iterate bounds and read timestamps. So all the
6142 /// returned iterators have to keep the *same* options object alive, not a
6143 /// copy each and not none at all. Dropping it at the end of
6144 /// `create_iterators_cf` left those pointers dangling. See issue #660.
6145 #[test]
6146 fn raw_iterators_cf_share_one_live_readopts() {
6147 let dir = tempfile::Builder::new()
6148 .prefix("rocksdb-raw-iterators-cf-readopts")
6149 .tempdir()
6150 .unwrap();
6151
6152 let mut opts = Options::default();
6153 opts.create_if_missing(true);
6154 opts.create_missing_column_families(true);
6155 let db = DB::open_cf_descriptors(
6156 &opts,
6157 dir.path(),
6158 [
6159 ColumnFamilyDescriptor::new("first", Options::default()),
6160 ColumnFamilyDescriptor::new("second", Options::default()),
6161 ],
6162 )
6163 .unwrap();
6164
6165 let first = db.cf_handle("first").unwrap();
6166 let second = db.cf_handle("second").unwrap();
6167 let iterators = db.raw_iterators_cf([&first, &second]).unwrap();
6168 assert_eq!(iterators.len(), 2);
6169
6170 let shared = iterators[0].readopts_ptr();
6171 assert!(!shared.is_null());
6172 for iterator in &iterators {
6173 assert_eq!(
6174 iterator.readopts_ptr(),
6175 shared,
6176 "each iterator must hold the options object it was created from"
6177 );
6178 }
6179 }
6180}