Skip to main content

rust_rocksdb/
db_iterator.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
15use crate::{
16    Error, ReadOptions, WriteBatch,
17    db::{DB, DBAccess},
18    ffi,
19};
20use libc::{c_char, c_uchar, size_t};
21use std::mem::ManuallyDrop;
22use std::sync::Arc;
23use std::{marker::PhantomData, ops::ControlFlow, slice};
24
25/// A type alias to keep compatibility. See [`DBRawIteratorWithThreadMode`] for details
26pub type DBRawIterator<'a> = DBRawIteratorWithThreadMode<'a, DB>;
27
28/// Keeps an iterator's [`ReadOptions`] alive for as long as the iterator is.
29///
30/// `rocksdb_create_iterators` builds N iterators from one options object, so
31/// that case needs a shared owner. Every other constructor makes one iterator
32/// from one options object and keeps it outright, which costs nothing.
33pub(crate) enum IterReadOptions {
34    Owned(ReadOptions),
35    Shared(Arc<ReadOptions>),
36}
37
38impl IterReadOptions {
39    #[inline]
40    pub(crate) fn as_ptr(&self) -> *mut ffi::rocksdb_readoptions_t {
41        match self {
42            Self::Owned(readopts) => readopts.inner,
43            Self::Shared(readopts) => readopts.inner,
44        }
45    }
46}
47
48impl From<ReadOptions> for IterReadOptions {
49    fn from(readopts: ReadOptions) -> Self {
50        Self::Owned(readopts)
51    }
52}
53
54impl From<Arc<ReadOptions>> for IterReadOptions {
55    fn from(readopts: Arc<ReadOptions>) -> Self {
56        Self::Shared(readopts)
57    }
58}
59
60/// A low-level iterator over a database or column family, created by [`DB::raw_iterator`]
61/// and other `raw_iterator_*` methods.
62///
63/// This iterator replicates RocksDB's API. It should provide better
64/// performance and more features than [`DBIteratorWithThreadMode`], which is a standard
65/// Rust [`std::iter::Iterator`].
66///
67/// ```
68/// use rust_rocksdb::{DB, Options};
69///
70/// let tempdir = tempfile::Builder::new()
71///     .prefix("_path_for_rocksdb_storage4")
72///     .tempdir()
73///     .expect("Failed to create temporary path for the _path_for_rocksdb_storage4.");
74/// let path = tempdir.path();
75/// {
76///     let db = DB::open_default(path).unwrap();
77///     let mut iter = db.raw_iterator();
78///
79///     // Forwards iteration
80///     iter.seek_to_first();
81///     while iter.valid() {
82///         println!("Saw {:?} {:?}", iter.key(), iter.value());
83///         iter.next();
84///     }
85///
86///     // Reverse iteration
87///     iter.seek_to_last();
88///     while iter.valid() {
89///         println!("Saw {:?} {:?}", iter.key(), iter.value());
90///         iter.prev();
91///     }
92///
93///     // Seeking
94///     iter.seek(b"my key");
95///     while iter.valid() {
96///         println!("Saw {:?} {:?}", iter.key(), iter.value());
97///         iter.next();
98///     }
99///
100///     // Reverse iteration from key
101///     // Note, use seek_for_prev when reversing because if this key doesn't exist,
102///     // this will make the iterator start from the previous key rather than the next.
103///     iter.seek_for_prev(b"my key");
104///     while iter.valid() {
105///         println!("Saw {:?} {:?}", iter.key(), iter.value());
106///         iter.prev();
107///     }
108/// }
109/// let _ = DB::destroy(&Options::default(), path);
110/// ```
111pub struct DBRawIteratorWithThreadMode<'a, D: DBAccess> {
112    inner: std::ptr::NonNull<ffi::rocksdb_iterator_t>,
113
114    /// When iterate_lower_bound or iterate_upper_bound are set, the inner
115    /// C iterator keeps a pointer to the upper bound inside `_readopts`.
116    /// Storing this makes sure the upper bound is always alive when the
117    /// iterator is being used.
118    ///
119    /// And yes, we need to store the entire ReadOptions structure since C++
120    /// ReadOptions keep reference to C rocksdb_readoptions_t wrapper which
121    /// point to vectors we own.  See issue #660.
122    ///
123    /// This is deliberately not an `Option`. An iterator without its options
124    /// is the bug in issue #660, so the type does not let one be built.
125    readopts: IterReadOptions,
126
127    db: PhantomData<&'a D>,
128}
129
130impl<'a, D: DBAccess> DBRawIteratorWithThreadMode<'a, D> {
131    /// Unlike [`new_cf`](Self::new_cf), the `db` borrow is not tied to `'a`.
132    /// That is what lets [`OwnedPrefixProber::with_opts`](crate::OwnedPrefixProber::with_opts)
133    /// build a `'static` iterator, so tying it would break that caller.
134    pub(crate) fn new(db: &D, readopts: ReadOptions) -> Self {
135        let inner = unsafe { db.create_iterator(&readopts) };
136        Self::from_inner(inner, readopts)
137    }
138
139    pub(crate) fn new_cf(
140        db: &'a D,
141        cf_handle: *mut ffi::rocksdb_column_family_handle_t,
142        readopts: ReadOptions,
143    ) -> Self {
144        let inner = unsafe { db.create_iterator_cf(cf_handle, &readopts) };
145        Self::from_inner(inner, readopts)
146    }
147
148    /// Same as [`new_cf`](Self::new_cf), but the returned iterator's lifetime
149    /// parameter is not tied to the `db` borrow.
150    ///
151    /// Keeping `db` alive for at least as long as the iterator becomes the
152    /// caller's job, normally by holding an `Arc<D>` next to it and dropping
153    /// the iterator first. Prefer `new_cf` unless you are building an owning
154    /// wrapper such as [`OwnedPrefixProber`](crate::OwnedPrefixProber).
155    pub(crate) fn new_cf_detached(
156        db: &D,
157        cf_handle: *mut ffi::rocksdb_column_family_handle_t,
158        readopts: ReadOptions,
159    ) -> Self {
160        let inner = unsafe { db.create_iterator_cf(cf_handle, &readopts) };
161        Self::from_inner(inner, readopts)
162    }
163
164    pub(crate) fn from_inner(
165        inner: *mut ffi::rocksdb_iterator_t,
166        readopts: impl Into<IterReadOptions>,
167    ) -> Self {
168        // This unwrap will never fail since rocksdb_create_iterator and
169        // rocksdb_create_iterator_cf functions always return non-null. They
170        // use new and deference the result so any nulls would end up with SIGSEGV
171        // there and we would have a bigger issue.
172        let inner = std::ptr::NonNull::new(inner).unwrap();
173        Self {
174            inner,
175            readopts: readopts.into(),
176            db: PhantomData,
177        }
178    }
179
180    pub(crate) fn into_inner(
181        self,
182    ) -> (std::ptr::NonNull<ffi::rocksdb_iterator_t>, IterReadOptions) {
183        let value = ManuallyDrop::new(self);
184        // SAFETY: value won't be used beyond this point
185        let inner = unsafe { std::ptr::read(&raw const value.inner) };
186        let readopts = unsafe { std::ptr::read(&raw const value.readopts) };
187
188        (inner, readopts)
189    }
190
191    /// The options object this iterator was created from, for tests that need
192    /// to assert which iterators share one owner.
193    #[cfg(test)]
194    pub(crate) fn readopts_ptr(&self) -> *mut ffi::rocksdb_readoptions_t {
195        self.readopts.as_ptr()
196    }
197
198    /// Returns `true` if the iterator is valid. An iterator is invalidated when
199    /// it reaches the end of its defined range, or when it encounters an error.
200    ///
201    /// To check whether the iterator encountered an error after `valid` has
202    /// returned `false`, use the [`status`](DBRawIteratorWithThreadMode::status) method. `status` will never
203    /// return an error when `valid` is `true`.
204    pub fn valid(&self) -> bool {
205        unsafe { ffi::rocksdb_iter_valid(self.inner.as_ptr()) != 0 }
206    }
207
208    /// Returns an error `Result` if the iterator has encountered an error
209    /// during operation. When an error is encountered, the iterator is
210    /// invalidated and [`valid`](DBRawIteratorWithThreadMode::valid) will return `false` when called.
211    ///
212    /// Performing a seek will discard the current status.
213    pub fn status(&self) -> Result<(), Error> {
214        unsafe {
215            ffi_try!(ffi::rocksdb_iter_get_error(self.inner.as_ptr()));
216        }
217        Ok(())
218    }
219
220    /// Refreshes the iterator to represent the latest state of the DB.
221    /// The iterator is invalidated after this call and must be re-sought
222    /// before use.
223    ///
224    /// If the iterator was created with a snapshot, the refreshed iterator
225    /// will no longer use that snapshot and will instead read the latest
226    /// DB state. The snapshot itself is not released; it remains valid and
227    /// will be released when the owning [`crate::SnapshotWithThreadMode`] is dropped.
228    pub fn refresh(&mut self) -> Result<(), Error> {
229        unsafe {
230            ffi_try!(ffi::rocksdb_iter_refresh(self.inner.as_ptr()));
231        }
232        Ok(())
233    }
234
235    /// Seeks to the first key in the database.
236    ///
237    /// # Examples
238    ///
239    /// ```rust
240    /// use rust_rocksdb::{DB, Options};
241    ///
242    /// let tempdir = tempfile::Builder::new()
243    ///     .prefix("_path_for_rocksdb_storage5")
244    ///     .tempdir()
245    ///     .expect("Failed to create temporary path for the _path_for_rocksdb_storage5.");
246    /// let path = tempdir.path();
247    /// {
248    ///     let db = DB::open_default(path).unwrap();
249    ///     let mut iter = db.raw_iterator();
250    ///
251    ///     // Iterate all keys from the start in lexicographic order
252    ///     iter.seek_to_first();
253    ///
254    ///     while iter.valid() {
255    ///         println!("{:?} {:?}", iter.key(), iter.value());
256    ///         iter.next();
257    ///     }
258    ///
259    ///     // Read just the first key
260    ///     iter.seek_to_first();
261    ///
262    ///     if iter.valid() {
263    ///         println!("{:?} {:?}", iter.key(), iter.value());
264    ///     } else {
265    ///         // There are no keys in the database
266    ///     }
267    /// }
268    /// let _ = DB::destroy(&Options::default(), path);
269    /// ```
270    pub fn seek_to_first(&mut self) {
271        unsafe {
272            ffi::rocksdb_iter_seek_to_first(self.inner.as_ptr());
273        }
274    }
275
276    /// Seeks to the last key in the database.
277    ///
278    /// # Examples
279    ///
280    /// ```rust
281    /// use rust_rocksdb::{DB, Options};
282    ///
283    /// let tempdir = tempfile::Builder::new()
284    ///     .prefix("_path_for_rocksdb_storage6")
285    ///     .tempdir()
286    ///     .expect("Failed to create temporary path for the _path_for_rocksdb_storage6.");
287    /// let path = tempdir.path();
288    /// {
289    ///     let db = DB::open_default(path).unwrap();
290    ///     let mut iter = db.raw_iterator();
291    ///
292    ///     // Iterate all keys from the end in reverse lexicographic order
293    ///     iter.seek_to_last();
294    ///
295    ///     while iter.valid() {
296    ///         println!("{:?} {:?}", iter.key(), iter.value());
297    ///         iter.prev();
298    ///     }
299    ///
300    ///     // Read just the last key
301    ///     iter.seek_to_last();
302    ///
303    ///     if iter.valid() {
304    ///         println!("{:?} {:?}", iter.key(), iter.value());
305    ///     } else {
306    ///         // There are no keys in the database
307    ///     }
308    /// }
309    /// let _ = DB::destroy(&Options::default(), path);
310    /// ```
311    pub fn seek_to_last(&mut self) {
312        unsafe {
313            ffi::rocksdb_iter_seek_to_last(self.inner.as_ptr());
314        }
315    }
316
317    /// Seeks to the specified key or the first key that lexicographically follows it.
318    ///
319    /// This method will attempt to seek to the specified key. If that key does not exist, it will
320    /// find and seek to the key that lexicographically follows it instead.
321    ///
322    /// # Examples
323    ///
324    /// ```rust
325    /// use rust_rocksdb::{DB, Options};
326    ///
327    /// let tempdir = tempfile::Builder::new()
328    ///     .prefix("_path_for_rocksdb_storage7")
329    ///     .tempdir()
330    ///     .expect("Failed to create temporary path for the _path_for_rocksdb_storage7.");
331    /// let path = tempdir.path();
332    /// {
333    ///     let db = DB::open_default(path).unwrap();
334    ///     let mut iter = db.raw_iterator();
335    ///
336    ///     // Read the first key that starts with 'a'
337    ///     iter.seek(b"a");
338    ///
339    ///     if iter.valid() {
340    ///         println!("{:?} {:?}", iter.key(), iter.value());
341    ///     } else {
342    ///         // There are no keys in the database
343    ///     }
344    /// }
345    /// let _ = DB::destroy(&Options::default(), path);
346    /// ```
347    pub fn seek<K: AsRef<[u8]>>(&mut self, key: K) {
348        let key = key.as_ref();
349
350        unsafe {
351            ffi::rocksdb_iter_seek(
352                self.inner.as_ptr(),
353                key.as_ptr() as *const c_char,
354                key.len() as size_t,
355            );
356        }
357    }
358
359    /// Seeks to the specified key, or the first key that lexicographically precedes it.
360    ///
361    /// Like ``.seek()`` this method will attempt to seek to the specified key.
362    /// The difference with ``.seek()`` is that if the specified key do not exist, this method will
363    /// seek to key that lexicographically precedes it instead.
364    ///
365    /// # Examples
366    ///
367    /// ```rust
368    /// use rust_rocksdb::{DB, Options};
369    ///
370    /// let tempdir = tempfile::Builder::new()
371    ///     .prefix("_path_for_rocksdb_storage8")
372    ///     .tempdir()
373    ///     .expect("Failed to create temporary path for the _path_for_rocksdb_storage8.");
374    /// let path = tempdir.path();
375    /// {
376    ///     let db = DB::open_default(path).unwrap();
377    ///     let mut iter = db.raw_iterator();
378    ///
379    ///     // Read the last key that starts with 'a'
380    ///     iter.seek_for_prev(b"b");
381    ///
382    ///     if iter.valid() {
383    ///         println!("{:?} {:?}", iter.key(), iter.value());
384    ///     } else {
385    ///         // There are no keys in the database
386    ///     }
387    /// }
388    /// let _ = DB::destroy(&Options::default(), path);
389    /// ```
390    pub fn seek_for_prev<K: AsRef<[u8]>>(&mut self, key: K) {
391        let key = key.as_ref();
392
393        unsafe {
394            ffi::rocksdb_iter_seek_for_prev(
395                self.inner.as_ptr(),
396                key.as_ptr() as *const c_char,
397                key.len() as size_t,
398            );
399        }
400    }
401
402    /// Seeks to the next key.
403    pub fn next(&mut self) {
404        if self.valid() {
405            // SAFETY: The validity check above guarantees that RocksDB permits
406            // advancing the iterator.
407            unsafe { self.next_unchecked() }
408        }
409    }
410
411    /// Advances to the next key without checking iterator validity.
412    ///
413    /// # Safety
414    ///
415    /// The iterator must be valid.
416    #[inline]
417    unsafe fn next_unchecked(&mut self) {
418        unsafe {
419            ffi::rocksdb_iter_next(self.inner.as_ptr());
420        }
421    }
422
423    /// Seeks to the previous key.
424    pub fn prev(&mut self) {
425        if self.valid() {
426            // SAFETY: The validity check above guarantees that RocksDB permits
427            // advancing the iterator.
428            unsafe { self.prev_unchecked() }
429        }
430    }
431
432    /// Advances to the previous key without checking iterator validity.
433    ///
434    /// # Safety
435    ///
436    /// The iterator must be valid.
437    #[inline]
438    unsafe fn prev_unchecked(&mut self) {
439        unsafe {
440            ffi::rocksdb_iter_prev(self.inner.as_ptr());
441        }
442    }
443
444    /// Returns a slice of the current key.
445    pub fn key(&self) -> Option<&[u8]> {
446        if self.valid() {
447            // SAFETY: We just checked that the iterator is valid.
448            Some(unsafe { self.key_impl() })
449        } else {
450            None
451        }
452    }
453
454    /// Returns a slice of the current value.
455    pub fn value(&self) -> Option<&[u8]> {
456        if self.valid() {
457            // SAFETY: We just checked that the iterator is valid.
458            Some(unsafe { self.value_impl() })
459        } else {
460            None
461        }
462    }
463
464    /// Returns pair with slice of the current key and current value.
465    pub fn item(&self) -> Option<(&[u8], &[u8])> {
466        if self.valid() {
467            // SAFETY: We just checked that the iterator is valid.
468            Some(unsafe { (self.key_impl(), self.value_impl()) })
469        } else {
470            None
471        }
472    }
473
474    /// Returns a slice of the current key.
475    ///
476    /// # Safety
477    ///
478    /// The iterator must be valid (i.e., `valid()` returns true). Calling this
479    /// method when the iterator is invalid is undefined behavior, as RocksDB
480    /// may return an invalid pointer.
481    ///
482    /// Uses `rocksdb_iter_key_slice` which returns a `rocksdb_slice_t` by value,
483    /// avoiding the overhead of output parameters compared to `rocksdb_iter_key`.
484    #[inline]
485    unsafe fn key_impl(&self) -> &[u8] {
486        unsafe {
487            let slice = ffi::rocksdb_iter_key_slice(self.inner.as_ptr());
488            if slice.size == 0 {
489                // Empty keys and values are legal in RocksDB, and
490                // `slice::from_raw_parts` requires a dereferenceable pointer
491                // even for a zero length, so do not build a slice from
492                // whatever `data` happens to be.
493                return &[];
494            }
495            slice::from_raw_parts(slice.data as *const c_uchar, slice.size)
496        }
497    }
498
499    /// Returns a slice of the current value.
500    ///
501    /// # Safety
502    ///
503    /// The iterator must be valid (i.e., `valid()` returns true). Calling this
504    /// method when the iterator is invalid is undefined behavior, as RocksDB
505    /// may return an invalid pointer.
506    ///
507    /// Uses `rocksdb_iter_value_slice` which returns a `rocksdb_slice_t` by value,
508    /// avoiding the overhead of output parameters compared to `rocksdb_iter_value`.
509    #[inline]
510    unsafe fn value_impl(&self) -> &[u8] {
511        unsafe {
512            let slice = ffi::rocksdb_iter_value_slice(self.inner.as_ptr());
513            if slice.size == 0 {
514                // Empty keys and values are legal in RocksDB, and
515                // `slice::from_raw_parts` requires a dereferenceable pointer
516                // even for a zero length, so do not build a slice from
517                // whatever `data` happens to be.
518                return &[];
519            }
520            slice::from_raw_parts(slice.data as *const c_uchar, slice.size)
521        }
522    }
523
524    /// Returns a slice of the current entry's timestamp.
525    pub fn timestamp(&self) -> Option<&[u8]> {
526        if self.valid() {
527            // SAFETY: We just checked that the iterator is valid.
528            Some(unsafe { self.timestamp_unchecked() })
529        } else {
530            None
531        }
532    }
533
534    /// Returns the timestamp of the current entry without checking iterator validity.
535    ///
536    /// # Safety
537    ///
538    /// The iterator must be valid (i.e., `valid()` returns true). Calling this
539    /// method when the iterator is invalid is undefined behavior, as RocksDB
540    /// may return an invalid pointer.
541    ///
542    /// Uses `rocksdb_iter_timestamp_slice` which returns a `rocksdb_slice_t` by value,
543    /// avoiding the overhead of output parameters compared to `rocksdb_iter_timestamp`.
544    pub unsafe fn timestamp_unchecked(&self) -> &[u8] {
545        unsafe {
546            let slice = ffi::rocksdb_iter_timestamp_slice(self.inner.as_ptr());
547            if slice.size == 0 {
548                return &[];
549            }
550            slice::from_raw_parts(slice.data as *const c_uchar, slice.size)
551        }
552    }
553}
554
555impl<D: DBAccess> Drop for DBRawIteratorWithThreadMode<'_, D> {
556    fn drop(&mut self) {
557        unsafe {
558            ffi::rocksdb_iter_destroy(self.inner.as_ptr());
559        }
560    }
561}
562
563unsafe impl<D: DBAccess> Send for DBRawIteratorWithThreadMode<'_, D> {}
564unsafe impl<D: DBAccess> Sync for DBRawIteratorWithThreadMode<'_, D> {}
565
566/// A type alias to keep compatibility. See [`DBIteratorWithThreadMode`] for details
567pub type DBIterator<'a> = DBIteratorWithThreadMode<'a, DB>;
568
569/// A standard Rust [`Iterator`] over a database or column family.
570///
571/// As an alternative, [`DBRawIteratorWithThreadMode`] is a low level wrapper around
572/// RocksDB's API, which can provide better performance and more features.
573///
574/// ```
575/// use rust_rocksdb::{DB, Direction, IteratorMode, Options};
576///
577/// let tempdir = tempfile::Builder::new()
578///     .prefix("_path_for_rocksdb_storage2")
579///     .tempdir()
580///     .expect("Failed to create temporary path for the _path_for_rocksdb_storage2.");
581/// let path = tempdir.path();
582/// {
583///     let db = DB::open_default(path).unwrap();
584///     let mut iter = db.iterator(IteratorMode::Start); // Always iterates forward
585///     for item in iter {
586///         let (key, value) = item.unwrap();
587///         println!("Saw {:?} {:?}", key, value);
588///     }
589///     iter = db.iterator(IteratorMode::End);  // Always iterates backward
590///     for item in iter {
591///         let (key, value) = item.unwrap();
592///         println!("Saw {:?} {:?}", key, value);
593///     }
594///     iter = db.iterator(IteratorMode::From(b"my key", Direction::Forward)); // From a key in Direction::{forward,reverse}
595///     for item in iter {
596///         let (key, value) = item.unwrap();
597///         println!("Saw {:?} {:?}", key, value);
598///     }
599///
600///     // You can seek with an existing Iterator instance, too
601///     iter = db.iterator(IteratorMode::Start);
602///     iter.set_mode(IteratorMode::From(b"another key", Direction::Reverse));
603///     for item in iter {
604///         let (key, value) = item.unwrap();
605///         println!("Saw {:?} {:?}", key, value);
606///     }
607/// }
608/// let _ = DB::destroy(&Options::default(), path);
609/// ```
610///
611/// An iterator must not outlive the `DB` it iterates over:
612///
613/// ```compile_fail,E0597
614/// use rust_rocksdb::{IteratorMode, DB};
615///
616/// let _iter = {
617///     let db = DB::open_default("foo").unwrap();
618///     db.iterator(IteratorMode::Start)
619/// };
620/// ```
621pub struct DBIteratorWithThreadMode<'a, D: DBAccess> {
622    raw: DBRawIteratorWithThreadMode<'a, D>,
623    direction: Direction,
624    done: bool,
625}
626
627#[derive(Copy, Clone)]
628pub enum Direction {
629    Forward,
630    Reverse,
631}
632
633pub type KVBytes = (Box<[u8]>, Box<[u8]>);
634
635#[derive(Copy, Clone)]
636pub enum IteratorMode<'a> {
637    Start,
638    End,
639    From(&'a [u8], Direction),
640}
641
642impl<'a, D: DBAccess> DBIteratorWithThreadMode<'a, D> {
643    pub(crate) fn new(db: &D, readopts: ReadOptions, mode: IteratorMode) -> Self {
644        Self::from_raw(DBRawIteratorWithThreadMode::new(db, readopts), mode)
645    }
646
647    pub(crate) fn new_cf(
648        db: &'a D,
649        cf_handle: *mut ffi::rocksdb_column_family_handle_t,
650        readopts: ReadOptions,
651        mode: IteratorMode,
652    ) -> Self {
653        Self::from_raw(
654            DBRawIteratorWithThreadMode::new_cf(db, cf_handle, readopts),
655            mode,
656        )
657    }
658
659    fn from_raw(raw: DBRawIteratorWithThreadMode<'a, D>, mode: IteratorMode) -> Self {
660        let mut rv = DBIteratorWithThreadMode {
661            raw,
662            direction: Direction::Forward, // blown away by set_mode()
663            done: false,
664        };
665        rv.set_mode(mode);
666        rv
667    }
668
669    pub fn set_mode(&mut self, mode: IteratorMode) {
670        self.done = false;
671        self.direction = match mode {
672            IteratorMode::Start => {
673                self.raw.seek_to_first();
674                Direction::Forward
675            }
676            IteratorMode::End => {
677                self.raw.seek_to_last();
678                Direction::Reverse
679            }
680            IteratorMode::From(key, Direction::Forward) => {
681                self.raw.seek(key);
682                Direction::Forward
683            }
684            IteratorMode::From(key, Direction::Reverse) => {
685                self.raw.seek_for_prev(key);
686                Direction::Reverse
687            }
688        };
689    }
690
691    /// Refreshes the iterator, then re-seeks using the given mode.
692    ///
693    /// After a refresh the underlying iterator is invalidated, so a mode
694    /// must be provided to reposition it.
695    pub fn refresh(&mut self, mode: IteratorMode) -> Result<(), Error> {
696        self.raw.refresh()?;
697        self.set_mode(mode);
698        Ok(())
699    }
700
701    /// Visits the remaining entries without allocating key or value buffers.
702    ///
703    /// The key and value slices are valid only for the duration of each callback.
704    /// Returning [`ControlFlow::Break`] stops the scan after consuming the current
705    /// entry. The iterator can then resume at the following entry. Callback errors
706    /// and RocksDB iterator errors are returned through `E`.
707    ///
708    /// # Examples
709    ///
710    /// ```
711    /// use std::ops::ControlFlow;
712    ///
713    /// use rust_rocksdb::{DB, Error, IteratorMode};
714    ///
715    /// # let tempdir = tempfile::tempdir().unwrap();
716    /// let db = DB::open_default(tempdir.path()).unwrap();
717    /// db.put(b"k1", b"value").unwrap();
718    ///
719    /// let mut total_bytes = 0;
720    /// let mut iter = db.iterator(IteratorMode::Start);
721    /// let outcome: Result<ControlFlow<()>, Error> = iter.try_for_each_ref(|key, value| {
722    ///     total_bytes += key.len() + value.len();
723    ///     Ok(ControlFlow::Continue(()))
724    /// });
725    ///
726    /// assert!(matches!(outcome, Ok(ControlFlow::Continue(()))));
727    /// assert_eq!(total_bytes, 7);
728    /// ```
729    ///
730    /// Borrowed entries cannot escape the callback:
731    ///
732    /// ```compile_fail,E0521
733    /// use std::ops::ControlFlow;
734    ///
735    /// use rust_rocksdb::{DB, Error, IteratorMode};
736    ///
737    /// # let tempdir = tempfile::tempdir().unwrap();
738    /// let db = DB::open_default(tempdir.path()).unwrap();
739    /// let mut iter = db.iterator(IteratorMode::Start);
740    /// let mut saved_key = None;
741    ///
742    /// let _: Result<ControlFlow<()>, Error> = iter.try_for_each_ref(|key, _| {
743    ///     saved_key = Some(key);
744    ///     Ok(ControlFlow::Break(()))
745    /// });
746    /// ```
747    #[inline]
748    pub fn try_for_each_ref<B, E, F>(&mut self, mut visit: F) -> Result<ControlFlow<B>, E>
749    where
750        E: From<Error>,
751        F: FnMut(&[u8], &[u8]) -> Result<ControlFlow<B>, E>,
752    {
753        loop {
754            let item = self
755                .next_with(|key, value| visit(key, value))
756                .map_err(E::from)?;
757            match item {
758                Some(Ok(ControlFlow::Continue(()))) => {}
759                Some(Ok(ControlFlow::Break(value))) => return Ok(ControlFlow::Break(value)),
760                Some(Err(error)) => return Err(error),
761                None => return Ok(ControlFlow::Continue(())),
762            }
763        }
764    }
765
766    #[inline]
767    fn next_with<T>(&mut self, visit: impl FnOnce(&[u8], &[u8]) -> T) -> Result<Option<T>, Error> {
768        if self.done {
769            return Ok(None);
770        }
771
772        let Some((key, value)) = self.raw.item() else {
773            self.done = true;
774            self.raw.status()?;
775            return Ok(None);
776        };
777
778        let item = visit(key, value);
779        // SAFETY: `raw.item()` returned an entry, which proves that the
780        // iterator is valid until it is advanced.
781        unsafe { self.advance_unchecked() };
782        Ok(Some(item))
783    }
784
785    /// Advances in the configured direction without checking iterator validity.
786    ///
787    /// # Safety
788    ///
789    /// The raw iterator must be valid.
790    #[inline]
791    unsafe fn advance_unchecked(&mut self) {
792        match self.direction {
793            Direction::Forward => unsafe { self.raw.next_unchecked() },
794            Direction::Reverse => unsafe { self.raw.prev_unchecked() },
795        }
796    }
797}
798
799impl<D: DBAccess> Iterator for DBIteratorWithThreadMode<'_, D> {
800    type Item = Result<KVBytes, Error>;
801
802    fn next(&mut self) -> Option<Result<KVBytes, Error>> {
803        match self.next_with(|key, value| (Box::from(key), Box::from(value))) {
804            Ok(Some(item)) => Some(Ok(item)),
805            Ok(None) => None,
806            Err(error) => Some(Err(error)),
807        }
808    }
809}
810
811impl<D: DBAccess> std::iter::FusedIterator for DBIteratorWithThreadMode<'_, D> {}
812
813impl<'a, D: DBAccess> Into<DBRawIteratorWithThreadMode<'a, D>> for DBIteratorWithThreadMode<'a, D> {
814    fn into(self) -> DBRawIteratorWithThreadMode<'a, D> {
815        self.raw
816    }
817}
818
819/// Iterates the batches of writes since a given sequence number.
820///
821/// `DBWALIterator` is returned by `DB::get_updates_since()` and will return the
822/// batches of write operations that have occurred since a given sequence number
823/// (see `DB::latest_sequence_number()`). This iterator cannot be constructed by
824/// the application.
825///
826/// The iterator item type is a tuple of (`u64`, `WriteBatch`) where the first
827/// value is the sequence number of the associated write batch.
828///
829pub struct DBWALIterator {
830    pub(crate) inner: *mut ffi::rocksdb_wal_iterator_t,
831    pub(crate) start_seq_number: u64,
832}
833
834impl DBWALIterator {
835    /// Returns `true` if the iterator is valid. An iterator is invalidated when
836    /// it reaches the end of its defined range, or when it encounters an error.
837    ///
838    /// To check whether the iterator encountered an error after `valid` has
839    /// returned `false`, use the [`status`](DBWALIterator::status) method.
840    /// `status` will never return an error when `valid` is `true`.
841    pub fn valid(&self) -> bool {
842        unsafe { ffi::rocksdb_wal_iter_valid(self.inner) != 0 }
843    }
844
845    /// Returns an error `Result` if the iterator has encountered an error
846    /// during operation. When an error is encountered, the iterator is
847    /// invalidated and [`valid`](DBWALIterator::valid) will return `false` when
848    /// called.
849    pub fn status(&self) -> Result<(), Error> {
850        unsafe {
851            ffi_try!(ffi::rocksdb_wal_iter_status(self.inner));
852        }
853        Ok(())
854    }
855}
856
857impl Iterator for DBWALIterator {
858    type Item = Result<(u64, WriteBatch), Error>;
859
860    fn next(&mut self) -> Option<Self::Item> {
861        if !self.valid() {
862            return None;
863        }
864
865        let mut seq: u64 = 0;
866        let mut batch = WriteBatch {
867            inner: unsafe { ffi::rocksdb_wal_iter_get_batch(self.inner, &raw mut seq) },
868        };
869
870        // if the initial sequence number is what was requested we skip it to
871        // only provide changes *after* it
872        while seq <= self.start_seq_number {
873            unsafe {
874                ffi::rocksdb_wal_iter_next(self.inner);
875            }
876
877            if !self.valid() {
878                return None;
879            }
880
881            // this drops which in turn frees the skipped batch
882            batch = WriteBatch {
883                inner: unsafe { ffi::rocksdb_wal_iter_get_batch(self.inner, &raw mut seq) },
884            };
885        }
886
887        if !self.valid() {
888            return self.status().err().map(Result::Err);
889        }
890
891        // Seek to the next write batch.
892        // Note that WriteBatches live independently of the WAL iterator so this is safe to do
893        unsafe {
894            ffi::rocksdb_wal_iter_next(self.inner);
895        }
896
897        Some(Ok((seq, batch)))
898    }
899}
900
901impl Drop for DBWALIterator {
902    fn drop(&mut self) {
903        unsafe {
904            ffi::rocksdb_wal_iter_destroy(self.inner);
905        }
906    }
907}