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}