rust_rocksdb/transactions/
transaction.rs

1// Copyright 2021 Yiyuan Liu
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::marker::PhantomData;
17
18use crate::{
19    AsColumnFamilyRef, DBIteratorWithThreadMode, DBPinnableSlice, DBRawIteratorWithThreadMode,
20    Direction, Error, IteratorMode, ReadOptions, SnapshotWithThreadMode, WriteBatchWithIndex,
21    WriteBatchWithTransaction,
22    db::{DBAccess, convert_values},
23    ffi,
24};
25use libc::{c_char, c_void, size_t};
26
27// Default options are kept per-thread to avoid re-allocating on every call while
28// also preventing cross-thread sharing. Some RocksDB option wrappers hold
29// pointers into internal buffers and are not safe to share across threads.
30// Using thread_local allows cheap reuse in the common "default options" path
31// without synchronization overhead. Callers who need non-defaults must pass
32// explicit options.
33thread_local! { static DEFAULT_READ_OPTS: ReadOptions = ReadOptions::default(); }
34
35/// RocksDB Transaction.
36///
37/// To use transactions, you must first create a [`TransactionDB`] or [`OptimisticTransactionDB`].
38///
39/// [`TransactionDB`]: crate::TransactionDB
40/// [`OptimisticTransactionDB`]: crate::OptimisticTransactionDB
41///
42/// A `Transaction` must not outlive the `TransactionDB` it was created from:
43///
44/// ```compile_fail,E0597
45/// use rust_rocksdb::{SingleThreaded, TransactionDB};
46///
47/// let _txn = {
48///     let db = TransactionDB::<SingleThreaded>::open_default("foo").unwrap();
49///     db.transaction()
50/// };
51/// ```
52///
53/// A `Snapshot` taken from a `Transaction` must not outlive the `Transaction`:
54///
55/// ```compile_fail,E0597
56/// use rust_rocksdb::{SingleThreaded, TransactionDB};
57///
58/// let db = TransactionDB::<SingleThreaded>::open_default("foo").unwrap();
59/// let _snapshot = {
60///     let txn = db.transaction();
61///     txn.snapshot()
62/// };
63/// ```
64pub struct Transaction<'db, DB> {
65    pub(crate) inner: *mut ffi::rocksdb_transaction_t,
66    pub(crate) _marker: PhantomData<&'db DB>,
67}
68
69unsafe impl<DB> Send for Transaction<'_, DB> {}
70
71impl<DB> DBAccess for Transaction<'_, DB> {
72    unsafe fn create_snapshot(&self) -> *const ffi::rocksdb_snapshot_t {
73        unsafe { ffi::rocksdb_transaction_get_snapshot(self.inner) }
74    }
75
76    unsafe fn release_snapshot(&self, snapshot: *const ffi::rocksdb_snapshot_t) {
77        unsafe {
78            ffi::rocksdb_free(snapshot as *mut c_void);
79        }
80    }
81
82    unsafe fn create_iterator(&self, readopts: &ReadOptions) -> *mut ffi::rocksdb_iterator_t {
83        unsafe { ffi::rocksdb_transaction_create_iterator(self.inner, readopts.inner) }
84    }
85
86    unsafe fn create_iterator_cf(
87        &self,
88        cf_handle: *mut ffi::rocksdb_column_family_handle_t,
89        readopts: &ReadOptions,
90    ) -> *mut ffi::rocksdb_iterator_t {
91        unsafe {
92            ffi::rocksdb_transaction_create_iterator_cf(self.inner, readopts.inner, cf_handle)
93        }
94    }
95
96    fn get_opt<K: AsRef<[u8]>>(
97        &self,
98        key: K,
99        readopts: &ReadOptions,
100    ) -> Result<Option<Vec<u8>>, Error> {
101        self.get_opt(key, readopts)
102    }
103
104    fn get_cf_opt<K: AsRef<[u8]>>(
105        &self,
106        cf: &impl AsColumnFamilyRef,
107        key: K,
108        readopts: &ReadOptions,
109    ) -> Result<Option<Vec<u8>>, Error> {
110        self.get_cf_opt(cf, key, readopts)
111    }
112
113    fn get_pinned_opt<K: AsRef<[u8]>>(
114        &'_ self,
115        key: K,
116        readopts: &ReadOptions,
117    ) -> Result<Option<DBPinnableSlice<'_>>, Error> {
118        self.get_pinned_opt(key, readopts)
119    }
120
121    fn get_pinned_cf_opt<K: AsRef<[u8]>>(
122        &'_ self,
123        cf: &impl AsColumnFamilyRef,
124        key: K,
125        readopts: &ReadOptions,
126    ) -> Result<Option<DBPinnableSlice<'_>>, Error> {
127        self.get_pinned_cf_opt(cf, key, readopts)
128    }
129
130    fn multi_get_opt<K, I>(
131        &self,
132        keys: I,
133        readopts: &ReadOptions,
134    ) -> Vec<Result<Option<Vec<u8>>, Error>>
135    where
136        K: AsRef<[u8]>,
137        I: IntoIterator<Item = K>,
138    {
139        self.multi_get_opt(keys, readopts)
140    }
141
142    fn multi_get_cf_opt<'b, K, I, W>(
143        &self,
144        keys_cf: I,
145        readopts: &ReadOptions,
146    ) -> Vec<Result<Option<Vec<u8>>, Error>>
147    where
148        K: AsRef<[u8]>,
149        I: IntoIterator<Item = (&'b W, K)>,
150        W: AsColumnFamilyRef + 'b,
151    {
152        self.multi_get_cf_opt(keys_cf, readopts)
153    }
154}
155
156impl<DB> Transaction<'_, DB> {
157    /// Write all batched keys to the DB atomically.
158    ///
159    /// May return any error that could be returned by `DB::write`.
160    ///
161    /// If this transaction was created by a [`TransactionDB`], an error of
162    /// the [`Expired`] kind may be returned if this transaction has
163    /// lived longer than expiration time in [`TransactionOptions`].
164    ///
165    /// If this transaction was created by an [`OptimisticTransactionDB`], an error of
166    /// the [`Busy`] kind may be returned if the transaction
167    /// could not guarantee that there are no write conflicts.
168    /// An error of the [`TryAgain`] kind may be returned if the memtable
169    /// history size is not large enough (see [`Options::set_max_write_buffer_size_to_maintain`]).
170    ///
171    /// [`Expired`]: crate::ErrorKind::Expired
172    /// [`TransactionOptions`]: crate::TransactionOptions
173    /// [`TransactionDB`]: crate::TransactionDB
174    /// [`OptimisticTransactionDB`]: crate::OptimisticTransactionDB
175    /// [`Busy`]: crate::ErrorKind::Busy
176    /// [`TryAgain`]: crate::ErrorKind::TryAgain
177    /// [`Options::set_max_write_buffer_size_to_maintain`]: crate::Options::set_max_write_buffer_size_to_maintain
178    pub fn commit(self) -> Result<(), Error> {
179        unsafe {
180            ffi_try!(ffi::rocksdb_transaction_commit(self.inner));
181        }
182        Ok(())
183    }
184
185    pub fn set_name(&self, name: &[u8]) -> Result<(), Error> {
186        let ptr = name.as_ptr();
187        let len = name.len();
188        unsafe {
189            ffi_try!(ffi::rocksdb_transaction_set_name(
190                self.inner, ptr as _, len as _
191            ));
192        }
193
194        Ok(())
195    }
196
197    pub fn get_name(&self) -> Option<Vec<u8>> {
198        unsafe {
199            let mut name_len = 0;
200            let name = ffi::rocksdb_transaction_get_name(self.inner, &raw mut name_len);
201            if name.is_null() {
202                None
203            } else {
204                let mut vec = vec![0; name_len];
205                std::ptr::copy_nonoverlapping(name.cast::<u8>(), vec.as_mut_ptr(), name_len);
206                ffi::rocksdb_free(name as *mut c_void);
207                Some(vec)
208            }
209        }
210    }
211
212    pub fn prepare(&self) -> Result<(), Error> {
213        unsafe {
214            ffi_try!(ffi::rocksdb_transaction_prepare(self.inner));
215        }
216        Ok(())
217    }
218
219    /// Returns snapshot associated with transaction if snapshot was enabled in [`TransactionOptions`].
220    /// Otherwise, returns a snapshot with `nullptr` inside which doesn't affect read operations.
221    ///
222    /// [`TransactionOptions`]: crate::TransactionOptions
223    pub fn snapshot(&'_ self) -> SnapshotWithThreadMode<'_, Self> {
224        SnapshotWithThreadMode::new(self)
225    }
226
227    /// Discard all batched writes in this transaction.
228    pub fn rollback(&self) -> Result<(), Error> {
229        unsafe {
230            ffi_try!(ffi::rocksdb_transaction_rollback(self.inner));
231            Ok(())
232        }
233    }
234
235    /// Record the state of the transaction for future calls to [`rollback_to_savepoint`].
236    /// May be called multiple times to set multiple save points.
237    ///
238    /// [`rollback_to_savepoint`]: Self::rollback_to_savepoint
239    pub fn set_savepoint(&self) {
240        unsafe {
241            ffi::rocksdb_transaction_set_savepoint(self.inner);
242        }
243    }
244
245    /// Undo all operations in this transaction since the most recent call to [`set_savepoint`]
246    /// and removes the most recent [`set_savepoint`].
247    ///
248    /// Returns error if there is no previous call to [`set_savepoint`].
249    ///
250    /// [`set_savepoint`]: Self::set_savepoint
251    pub fn rollback_to_savepoint(&self) -> Result<(), Error> {
252        unsafe {
253            ffi_try!(ffi::rocksdb_transaction_rollback_to_savepoint(self.inner));
254            Ok(())
255        }
256    }
257
258    /// Get the bytes associated with a key value.
259    ///
260    /// See [`get_cf_opt`] for details.
261    ///
262    /// [`get_cf_opt`]: Self::get_cf_opt
263    pub fn get<K: AsRef<[u8]>>(&self, key: K) -> Result<Option<Vec<u8>>, Error> {
264        DEFAULT_READ_OPTS.with(|opts| self.get_opt(key, opts))
265    }
266
267    pub fn get_pinned<K: AsRef<[u8]>>(
268        &'_ self,
269        key: K,
270    ) -> Result<Option<DBPinnableSlice<'_>>, Error> {
271        DEFAULT_READ_OPTS.with(|opts| self.get_pinned_opt(key, opts))
272    }
273
274    /// Get the bytes associated with a key value and the given column family.
275    ///
276    /// See [`get_cf_opt`] for details.
277    ///
278    /// [`get_cf_opt`]: Self::get_cf_opt
279    pub fn get_cf<K: AsRef<[u8]>>(
280        &self,
281        cf: &impl AsColumnFamilyRef,
282        key: K,
283    ) -> Result<Option<Vec<u8>>, Error> {
284        DEFAULT_READ_OPTS.with(|opts| self.get_cf_opt(cf, key, opts))
285    }
286
287    pub fn get_pinned_cf<K: AsRef<[u8]>>(
288        &'_ self,
289        cf: &impl AsColumnFamilyRef,
290        key: K,
291    ) -> Result<Option<DBPinnableSlice<'_>>, Error> {
292        DEFAULT_READ_OPTS.with(|opts| self.get_pinned_cf_opt(cf, key, opts))
293    }
294
295    /// Get the key and ensure that this transaction will only
296    /// be able to be committed if this key is not written outside this
297    /// transaction after it has first been read (or after the snapshot if a
298    /// snapshot is set in this transaction).
299    ///
300    /// See [`get_for_update_cf_opt`] for details.
301    ///
302    /// [`get_for_update_cf_opt`]: Self::get_for_update_cf_opt
303    pub fn get_for_update<K: AsRef<[u8]>>(
304        &self,
305        key: K,
306        exclusive: bool,
307    ) -> Result<Option<Vec<u8>>, Error> {
308        DEFAULT_READ_OPTS.with(|opts| self.get_for_update_opt(key, exclusive, opts))
309    }
310
311    pub fn get_pinned_for_update<K: AsRef<[u8]>>(
312        &'_ self,
313        key: K,
314        exclusive: bool,
315    ) -> Result<Option<DBPinnableSlice<'_>>, Error> {
316        DEFAULT_READ_OPTS.with(|opts| self.get_pinned_for_update_opt(key, exclusive, opts))
317    }
318
319    /// Get the key in the given column family and ensure that this transaction will only
320    /// be able to be committed if this key is not written outside this
321    /// transaction after it has first been read (or after the snapshot if a
322    /// snapshot is set in this transaction).
323    ///
324    /// See [`get_for_update_cf_opt`] for details.
325    ///
326    /// [`get_for_update_cf_opt`]: Self::get_for_update_cf_opt
327    pub fn get_for_update_cf<K: AsRef<[u8]>>(
328        &self,
329        cf: &impl AsColumnFamilyRef,
330        key: K,
331        exclusive: bool,
332    ) -> Result<Option<Vec<u8>>, Error> {
333        DEFAULT_READ_OPTS.with(|opts| self.get_for_update_cf_opt(cf, key, exclusive, opts))
334    }
335
336    pub fn get_pinned_for_update_cf<K: AsRef<[u8]>>(
337        &'_ self,
338        cf: &impl AsColumnFamilyRef,
339        key: K,
340        exclusive: bool,
341    ) -> Result<Option<DBPinnableSlice<'_>>, Error> {
342        DEFAULT_READ_OPTS.with(|opts| self.get_pinned_for_update_cf_opt(cf, key, exclusive, opts))
343    }
344
345    /// Returns the bytes associated with a key value with read options.
346    ///
347    /// See [`get_cf_opt`] for details.
348    ///
349    /// [`get_cf_opt`]: Self::get_cf_opt
350    pub fn get_opt<K: AsRef<[u8]>>(
351        &self,
352        key: K,
353        readopts: &ReadOptions,
354    ) -> Result<Option<Vec<u8>>, Error> {
355        self.get_pinned_opt(key, readopts)
356            .map(|x| x.map(|v| v.as_ref().to_vec()))
357    }
358
359    pub fn get_pinned_opt<K: AsRef<[u8]>>(
360        &'_ self,
361        key: K,
362        readopts: &ReadOptions,
363    ) -> Result<Option<DBPinnableSlice<'_>>, Error> {
364        let key = key.as_ref();
365        unsafe {
366            let val = ffi_try!(ffi::rocksdb_transaction_get_pinned(
367                self.inner,
368                readopts.inner,
369                key.as_ptr() as *const c_char,
370                key.len(),
371            ));
372            if val.is_null() {
373                Ok(None)
374            } else {
375                Ok(Some(DBPinnableSlice::from_c(val)))
376            }
377        }
378    }
379
380    /// Get the bytes associated with a key value and the given column family with read options.
381    ///
382    /// This function will also read pending changes in this transaction.
383    /// Currently, this function will return an error of the [`MergeInProgress`] kind
384    /// if the most recent write to the queried key in this batch is a Merge.
385    ///
386    /// [`MergeInProgress`]: crate::ErrorKind::MergeInProgress
387    pub fn get_cf_opt<K: AsRef<[u8]>>(
388        &self,
389        cf: &impl AsColumnFamilyRef,
390        key: K,
391        readopts: &ReadOptions,
392    ) -> Result<Option<Vec<u8>>, Error> {
393        self.get_pinned_cf_opt(cf, key, readopts)
394            .map(|x| x.map(|v| v.as_ref().to_vec()))
395    }
396
397    pub fn get_pinned_cf_opt<K: AsRef<[u8]>>(
398        &'_ self,
399        cf: &impl AsColumnFamilyRef,
400        key: K,
401        readopts: &ReadOptions,
402    ) -> Result<Option<DBPinnableSlice<'_>>, Error> {
403        let key = key.as_ref();
404        unsafe {
405            let val = ffi_try!(ffi::rocksdb_transaction_get_pinned_cf(
406                self.inner,
407                readopts.inner,
408                cf.inner(),
409                key.as_ptr() as *const c_char,
410                key.len(),
411            ));
412            if val.is_null() {
413                Ok(None)
414            } else {
415                Ok(Some(DBPinnableSlice::from_c(val)))
416            }
417        }
418    }
419
420    /// Get the key with read options and ensure that this transaction will only
421    /// be able to be committed if this key is not written outside this
422    /// transaction after it has first been read (or after the snapshot if a
423    /// snapshot is set in this transaction).
424    ///
425    /// See [`get_for_update_cf_opt`] for details.
426    ///
427    /// [`get_for_update_cf_opt`]: Self::get_for_update_cf_opt
428    pub fn get_for_update_opt<K: AsRef<[u8]>>(
429        &self,
430        key: K,
431        exclusive: bool,
432        opts: &ReadOptions,
433    ) -> Result<Option<Vec<u8>>, Error> {
434        self.get_pinned_for_update_opt(key, exclusive, opts)
435            .map(|x| x.map(|v| v.as_ref().to_vec()))
436    }
437
438    pub fn get_pinned_for_update_opt<K: AsRef<[u8]>>(
439        &'_ self,
440        key: K,
441        exclusive: bool,
442        opts: &ReadOptions,
443    ) -> Result<Option<DBPinnableSlice<'_>>, Error> {
444        let key = key.as_ref();
445        unsafe {
446            let val = ffi_try!(ffi::rocksdb_transaction_get_pinned_for_update(
447                self.inner,
448                opts.inner,
449                key.as_ptr() as *const c_char,
450                key.len() as size_t,
451                u8::from(exclusive),
452            ));
453            if val.is_null() {
454                Ok(None)
455            } else {
456                Ok(Some(DBPinnableSlice::from_c(val)))
457            }
458        }
459    }
460
461    /// Get the key in the given column family with read options
462    /// and ensure that this transaction will only
463    /// be able to be committed if this key is not written outside this
464    /// transaction after it has first been read (or after the snapshot if a
465    /// snapshot is set in this transaction).
466    ///
467    /// Currently, this function will return an error of the [`MergeInProgress`]
468    /// if the most recent write to the queried key in this batch is a Merge.
469    ///
470    /// If this transaction was created by a [`TransactionDB`], it can return error of kind:
471    /// * [`Busy`] if there is a write conflict.
472    /// * [`TimedOut`] if a lock could not be acquired.
473    /// * [`TryAgain`] if the memtable history size is not large enough.
474    /// * [`MergeInProgress`] if merge operations cannot be resolved.
475    /// * or other errors if this key could not be read.
476    ///
477    /// If this transaction was created by an `[OptimisticTransactionDB]`, `get_for_update_opt`
478    /// can cause [`commit`] to fail. Otherwise, it could return any error that could
479    /// be returned by `[DB::get]`.
480    ///
481    /// [`Busy`]: crate::ErrorKind::Busy
482    /// [`TimedOut`]: crate::ErrorKind::TimedOut
483    /// [`TryAgain`]: crate::ErrorKind::TryAgain
484    /// [`MergeInProgress`]: crate::ErrorKind::MergeInProgress
485    /// [`TransactionDB`]: crate::TransactionDB
486    /// [`OptimisticTransactionDB`]: crate::OptimisticTransactionDB
487    /// [`commit`]: Self::commit
488    /// [`DB::get`]: crate::DB::get
489    pub fn get_for_update_cf_opt<K: AsRef<[u8]>>(
490        &self,
491        cf: &impl AsColumnFamilyRef,
492        key: K,
493        exclusive: bool,
494        opts: &ReadOptions,
495    ) -> Result<Option<Vec<u8>>, Error> {
496        self.get_pinned_for_update_cf_opt(cf, key, exclusive, opts)
497            .map(|x| x.map(|v| v.as_ref().to_vec()))
498    }
499
500    pub fn get_pinned_for_update_cf_opt<K: AsRef<[u8]>>(
501        &'_ self,
502        cf: &impl AsColumnFamilyRef,
503        key: K,
504        exclusive: bool,
505        opts: &ReadOptions,
506    ) -> Result<Option<DBPinnableSlice<'_>>, Error> {
507        let key = key.as_ref();
508        unsafe {
509            let val = ffi_try!(ffi::rocksdb_transaction_get_pinned_for_update_cf(
510                self.inner,
511                opts.inner,
512                cf.inner(),
513                key.as_ptr() as *const c_char,
514                key.len() as size_t,
515                u8::from(exclusive),
516            ));
517            if val.is_null() {
518                Ok(None)
519            } else {
520                Ok(Some(DBPinnableSlice::from_c(val)))
521            }
522        }
523    }
524
525    /// Return the values associated with the given keys.
526    pub fn multi_get<K, I>(&self, keys: I) -> Vec<Result<Option<Vec<u8>>, Error>>
527    where
528        K: AsRef<[u8]>,
529        I: IntoIterator<Item = K>,
530    {
531        DEFAULT_READ_OPTS.with(|opts| self.multi_get_opt(keys, opts))
532    }
533
534    /// Return the values associated with the given keys using read options.
535    pub fn multi_get_opt<K, I>(
536        &self,
537        keys: I,
538        readopts: &ReadOptions,
539    ) -> Vec<Result<Option<Vec<u8>>, Error>>
540    where
541        K: AsRef<[u8]>,
542        I: IntoIterator<Item = K>,
543    {
544        let owned_keys: Vec<K> = keys.into_iter().collect();
545        let (ptr_keys, keys_sizes): (Vec<*const c_char>, Vec<usize>) = owned_keys
546            .iter()
547            .map(|k| {
548                let key = k.as_ref();
549                (key.as_ptr() as *const c_char, key.len())
550            })
551            .unzip();
552
553        let mut values: Vec<*mut c_char> = Vec::with_capacity(ptr_keys.len());
554        let mut values_sizes: Vec<usize> = Vec::with_capacity(ptr_keys.len());
555        let mut errors: Vec<*mut c_char> = Vec::with_capacity(ptr_keys.len());
556        unsafe {
557            ffi::rocksdb_transaction_multi_get(
558                self.inner,
559                readopts.inner,
560                ptr_keys.len(),
561                ptr_keys.as_ptr(),
562                keys_sizes.as_ptr(),
563                values.as_mut_ptr(),
564                values_sizes.as_mut_ptr(),
565                errors.as_mut_ptr(),
566            );
567        }
568
569        unsafe {
570            values.set_len(ptr_keys.len());
571            values_sizes.set_len(ptr_keys.len());
572            errors.set_len(ptr_keys.len());
573        }
574
575        convert_values(values, values_sizes, errors)
576    }
577
578    /// Return the values associated with the given keys and column families.
579    pub fn multi_get_cf<'a, 'b: 'a, K, I, W>(
580        &'a self,
581        keys: I,
582    ) -> Vec<Result<Option<Vec<u8>>, Error>>
583    where
584        K: AsRef<[u8]>,
585        I: IntoIterator<Item = (&'b W, K)>,
586        W: 'b + AsColumnFamilyRef,
587    {
588        DEFAULT_READ_OPTS.with(|opts| self.multi_get_cf_opt(keys, opts))
589    }
590
591    /// Return the values associated with the given keys and column families using read options.
592    pub fn multi_get_cf_opt<'a, 'b: 'a, K, I, W>(
593        &'a self,
594        keys: I,
595        readopts: &ReadOptions,
596    ) -> Vec<Result<Option<Vec<u8>>, Error>>
597    where
598        K: AsRef<[u8]>,
599        I: IntoIterator<Item = (&'b W, K)>,
600        W: 'b + AsColumnFamilyRef,
601    {
602        let cfs_and_owned_keys: Vec<(&'b W, K)> = keys.into_iter().collect();
603        let (ptr_keys, keys_sizes): (Vec<*const c_char>, Vec<usize>) = cfs_and_owned_keys
604            .iter()
605            .map(|(_, k)| {
606                let key = k.as_ref();
607                (key.as_ptr() as *const c_char, key.len())
608            })
609            .unzip();
610        let ptr_cfs: Vec<*const ffi::rocksdb_column_family_handle_t> = cfs_and_owned_keys
611            .iter()
612            .map(|(c, _)| c.inner().cast_const())
613            .collect();
614        let mut values: Vec<*mut c_char> = Vec::with_capacity(ptr_keys.len());
615        let mut values_sizes: Vec<usize> = Vec::with_capacity(ptr_keys.len());
616        let mut errors: Vec<*mut c_char> = Vec::with_capacity(ptr_keys.len());
617        unsafe {
618            ffi::rocksdb_transaction_multi_get_cf(
619                self.inner,
620                readopts.inner,
621                ptr_cfs.as_ptr(),
622                ptr_keys.len(),
623                ptr_keys.as_ptr(),
624                keys_sizes.as_ptr(),
625                values.as_mut_ptr(),
626                values_sizes.as_mut_ptr(),
627                errors.as_mut_ptr(),
628            );
629        }
630
631        unsafe {
632            values.set_len(ptr_keys.len());
633            values_sizes.set_len(ptr_keys.len());
634            errors.set_len(ptr_keys.len());
635        }
636
637        convert_values(values, values_sizes, errors)
638    }
639
640    /// Return the values associated with the given keys, marking every key read for update.
641    ///
642    /// See [`multi_get_for_update_cf_opt`] for details.
643    ///
644    /// [`multi_get_for_update_cf_opt`]: Self::multi_get_for_update_cf_opt
645    pub fn multi_get_for_update<K, I>(&self, keys: I) -> Vec<Result<Option<Vec<u8>>, Error>>
646    where
647        K: AsRef<[u8]>,
648        I: IntoIterator<Item = K>,
649    {
650        DEFAULT_READ_OPTS.with(|opts| self.multi_get_for_update_opt(keys, opts))
651    }
652
653    /// Return the values associated with the given keys using read options, marking every key
654    /// read for update.
655    ///
656    /// See [`multi_get_for_update_cf_opt`] for details.
657    ///
658    /// [`multi_get_for_update_cf_opt`]: Self::multi_get_for_update_cf_opt
659    pub fn multi_get_for_update_opt<K, I>(
660        &self,
661        keys: I,
662        readopts: &ReadOptions,
663    ) -> Vec<Result<Option<Vec<u8>>, Error>>
664    where
665        K: AsRef<[u8]>,
666        I: IntoIterator<Item = K>,
667    {
668        let owned_keys: Vec<K> = keys.into_iter().collect();
669        let (ptr_keys, keys_sizes): (Vec<*const c_char>, Vec<usize>) = owned_keys
670            .iter()
671            .map(|k| {
672                let key = k.as_ref();
673                (key.as_ptr() as *const c_char, key.len())
674            })
675            .unzip();
676
677        let mut values: Vec<*mut c_char> = Vec::with_capacity(ptr_keys.len());
678        let mut values_sizes: Vec<usize> = Vec::with_capacity(ptr_keys.len());
679        let mut errors: Vec<*mut c_char> = Vec::with_capacity(ptr_keys.len());
680        unsafe {
681            ffi::rocksdb_transaction_multi_get_for_update(
682                self.inner,
683                readopts.inner,
684                ptr_keys.len(),
685                ptr_keys.as_ptr(),
686                keys_sizes.as_ptr(),
687                values.as_mut_ptr(),
688                values_sizes.as_mut_ptr(),
689                errors.as_mut_ptr(),
690            );
691        }
692
693        unsafe {
694            values.set_len(ptr_keys.len());
695            values_sizes.set_len(ptr_keys.len());
696            errors.set_len(ptr_keys.len());
697        }
698
699        convert_values(values, values_sizes, errors)
700    }
701
702    /// Return the values associated with the given keys and column families, marking every key
703    /// read for update.
704    ///
705    /// See [`multi_get_for_update_cf_opt`] for details.
706    ///
707    /// [`multi_get_for_update_cf_opt`]: Self::multi_get_for_update_cf_opt
708    pub fn multi_get_for_update_cf<'a, 'b: 'a, K, I, W>(
709        &'a self,
710        keys: I,
711    ) -> Vec<Result<Option<Vec<u8>>, Error>>
712    where
713        K: AsRef<[u8]>,
714        I: IntoIterator<Item = (&'b W, K)>,
715        W: 'b + AsColumnFamilyRef,
716    {
717        DEFAULT_READ_OPTS.with(|opts| self.multi_get_for_update_cf_opt(keys, opts))
718    }
719
720    /// Return the values associated with the given keys and column families using read options,
721    /// marking every key read for update.
722    ///
723    /// This is [`get_for_update_cf_opt`] over a batch of keys: each key is fetched and
724    /// conflict checked, so the transaction only commits if none of them were written
725    /// outside it after the read (or after the snapshot, if one is set). There is no
726    /// `exclusive` parameter because RocksDB always takes the lock exclusively here.
727    ///
728    /// Locking is all or nothing. RocksDB locks every key before reading any of them,
729    /// and if one lock fails it returns that same error for every entry without
730    /// performing the reads. Once the locks are held, each key is read on its own, so
731    /// after that point the results can differ per key.
732    ///
733    /// [`get_for_update_cf_opt`]: Self::get_for_update_cf_opt
734    pub fn multi_get_for_update_cf_opt<'a, 'b: 'a, K, I, W>(
735        &'a self,
736        keys: I,
737        readopts: &ReadOptions,
738    ) -> Vec<Result<Option<Vec<u8>>, Error>>
739    where
740        K: AsRef<[u8]>,
741        I: IntoIterator<Item = (&'b W, K)>,
742        W: 'b + AsColumnFamilyRef,
743    {
744        let cfs_and_owned_keys: Vec<(&'b W, K)> = keys.into_iter().collect();
745        let (ptr_keys, keys_sizes): (Vec<*const c_char>, Vec<usize>) = cfs_and_owned_keys
746            .iter()
747            .map(|(_, k)| {
748                let key = k.as_ref();
749                (key.as_ptr() as *const c_char, key.len())
750            })
751            .unzip();
752        let ptr_cfs: Vec<*const ffi::rocksdb_column_family_handle_t> = cfs_and_owned_keys
753            .iter()
754            .map(|(c, _)| c.inner().cast_const())
755            .collect();
756        let mut values: Vec<*mut c_char> = Vec::with_capacity(ptr_keys.len());
757        let mut values_sizes: Vec<usize> = Vec::with_capacity(ptr_keys.len());
758        let mut errors: Vec<*mut c_char> = Vec::with_capacity(ptr_keys.len());
759        unsafe {
760            ffi::rocksdb_transaction_multi_get_for_update_cf(
761                self.inner,
762                readopts.inner,
763                ptr_cfs.as_ptr(),
764                ptr_keys.len(),
765                ptr_keys.as_ptr(),
766                keys_sizes.as_ptr(),
767                values.as_mut_ptr(),
768                values_sizes.as_mut_ptr(),
769                errors.as_mut_ptr(),
770            );
771        }
772
773        unsafe {
774            values.set_len(ptr_keys.len());
775            values_sizes.set_len(ptr_keys.len());
776            errors.set_len(ptr_keys.len());
777        }
778
779        convert_values(values, values_sizes, errors)
780    }
781
782    /// Put the key value in default column family and do conflict checking on the key.
783    ///
784    /// See [`put_cf`] for details.
785    ///
786    /// [`put_cf`]: Self::put_cf
787    pub fn put<K: AsRef<[u8]>, V: AsRef<[u8]>>(&self, key: K, value: V) -> Result<(), Error> {
788        let key = key.as_ref();
789        let value = value.as_ref();
790        unsafe {
791            ffi_try!(ffi::rocksdb_transaction_put(
792                self.inner,
793                key.as_ptr() as *const c_char,
794                key.len() as size_t,
795                value.as_ptr() as *const c_char,
796                value.len() as size_t,
797            ));
798            Ok(())
799        }
800    }
801
802    /// Put the key value in the given column family and do conflict checking on the key.
803    ///
804    /// If this transaction was created by a [`TransactionDB`], it can return error of kind:
805    /// * [`Busy`] if there is a write conflict.
806    /// * [`TimedOut`] if a lock could not be acquired.
807    /// * [`TryAgain`] if the memtable history size is not large enough.
808    /// * [`MergeInProgress`] if merge operations cannot be resolved.
809    /// * or other errors on unexpected failures.
810    ///
811    /// [`Busy`]: crate::ErrorKind::Busy
812    /// [`TimedOut`]: crate::ErrorKind::TimedOut
813    /// [`TryAgain`]: crate::ErrorKind::TryAgain
814    /// [`MergeInProgress`]: crate::ErrorKind::MergeInProgress
815    /// [`TransactionDB`]: crate::TransactionDB
816    /// [`OptimisticTransactionDB`]: crate::OptimisticTransactionDB
817    pub fn put_cf<K: AsRef<[u8]>, V: AsRef<[u8]>>(
818        &self,
819        cf: &impl AsColumnFamilyRef,
820        key: K,
821        value: V,
822    ) -> Result<(), Error> {
823        let key = key.as_ref();
824        let value = value.as_ref();
825        unsafe {
826            ffi_try!(ffi::rocksdb_transaction_put_cf(
827                self.inner,
828                cf.inner(),
829                key.as_ptr() as *const c_char,
830                key.len() as size_t,
831                value.as_ptr() as *const c_char,
832                value.len() as size_t,
833            ));
834            Ok(())
835        }
836    }
837
838    /// Merge value with existing value of key, and also do conflict checking on the key.
839    ///
840    /// See [`merge_cf`] for details.
841    ///
842    /// [`merge_cf`]: Self::merge_cf
843    pub fn merge<K: AsRef<[u8]>, V: AsRef<[u8]>>(&self, key: K, value: V) -> Result<(), Error> {
844        let key = key.as_ref();
845        let value = value.as_ref();
846        unsafe {
847            ffi_try!(ffi::rocksdb_transaction_merge(
848                self.inner,
849                key.as_ptr() as *const c_char,
850                key.len() as size_t,
851                value.as_ptr() as *const c_char,
852                value.len() as size_t
853            ));
854            Ok(())
855        }
856    }
857
858    /// Merge `value` with existing value of `key` in the given column family,
859    /// and also do conflict checking on the key.
860    ///
861    /// If this transaction was created by a [`TransactionDB`], it can return error of kind:
862    /// * [`Busy`] if there is a write conflict.
863    /// * [`TimedOut`] if a lock could not be acquired.
864    /// * [`TryAgain`] if the memtable history size is not large enough.
865    /// * [`MergeInProgress`] if merge operations cannot be resolved.
866    /// * or other errors on unexpected failures.
867    ///
868    /// [`Busy`]: crate::ErrorKind::Busy
869    /// [`TimedOut`]: crate::ErrorKind::TimedOut
870    /// [`TryAgain`]: crate::ErrorKind::TryAgain
871    /// [`MergeInProgress`]: crate::ErrorKind::MergeInProgress
872    /// [`TransactionDB`]: crate::TransactionDB
873    pub fn merge_cf<K: AsRef<[u8]>, V: AsRef<[u8]>>(
874        &self,
875        cf: &impl AsColumnFamilyRef,
876        key: K,
877        value: V,
878    ) -> Result<(), Error> {
879        let key = key.as_ref();
880        let value = value.as_ref();
881        unsafe {
882            ffi_try!(ffi::rocksdb_transaction_merge_cf(
883                self.inner,
884                cf.inner(),
885                key.as_ptr() as *const c_char,
886                key.len() as size_t,
887                value.as_ptr() as *const c_char,
888                value.len() as size_t
889            ));
890            Ok(())
891        }
892    }
893
894    /// Delete the key value if it exists and do conflict checking on the key.
895    ///
896    /// See [`delete_cf`] for details.
897    ///
898    /// [`delete_cf`]: Self::delete_cf
899    pub fn delete<K: AsRef<[u8]>>(&self, key: K) -> Result<(), Error> {
900        let key = key.as_ref();
901        unsafe {
902            ffi_try!(ffi::rocksdb_transaction_delete(
903                self.inner,
904                key.as_ptr() as *const c_char,
905                key.len() as size_t
906            ));
907        }
908        Ok(())
909    }
910
911    /// Delete the key value in the given column family and do conflict checking.
912    ///
913    /// If this transaction was created by a [`TransactionDB`], it can return error of kind:
914    /// * [`Busy`] if there is a write conflict.
915    /// * [`TimedOut`] if a lock could not be acquired.
916    /// * [`TryAgain`] if the memtable history size is not large enough.
917    /// * [`MergeInProgress`] if merge operations cannot be resolved.
918    /// * or other errors on unexpected failures.
919    ///
920    /// [`Busy`]: crate::ErrorKind::Busy
921    /// [`TimedOut`]: crate::ErrorKind::TimedOut
922    /// [`TryAgain`]: crate::ErrorKind::TryAgain
923    /// [`MergeInProgress`]: crate::ErrorKind::MergeInProgress
924    /// [`TransactionDB`]: crate::TransactionDB
925    pub fn delete_cf<K: AsRef<[u8]>>(
926        &self,
927        cf: &impl AsColumnFamilyRef,
928        key: K,
929    ) -> Result<(), Error> {
930        let key = key.as_ref();
931        unsafe {
932            ffi_try!(ffi::rocksdb_transaction_delete_cf(
933                self.inner,
934                cf.inner(),
935                key.as_ptr() as *const c_char,
936                key.len() as size_t
937            ));
938        }
939        Ok(())
940    }
941
942    pub fn iterator<'a: 'b, 'b>(
943        &'a self,
944        mode: IteratorMode,
945    ) -> DBIteratorWithThreadMode<'b, Self> {
946        let readopts = ReadOptions::default();
947        self.iterator_opt(mode, readopts)
948    }
949
950    pub fn iterator_opt<'a: 'b, 'b>(
951        &'a self,
952        mode: IteratorMode,
953        readopts: ReadOptions,
954    ) -> DBIteratorWithThreadMode<'b, Self> {
955        DBIteratorWithThreadMode::new(self, readopts, mode)
956    }
957
958    /// Opens an iterator using the provided ReadOptions.
959    /// This is used when you want to iterate over a specific ColumnFamily with a modified ReadOptions.
960    pub fn iterator_cf_opt<'a: 'b, 'b>(
961        &'a self,
962        cf_handle: &impl AsColumnFamilyRef,
963        readopts: ReadOptions,
964        mode: IteratorMode,
965    ) -> DBIteratorWithThreadMode<'b, Self> {
966        DBIteratorWithThreadMode::new_cf(self, cf_handle.inner(), readopts, mode)
967    }
968
969    /// Opens an iterator with `set_total_order_seek` enabled.
970    /// This must be used to iterate across prefixes when `set_memtable_factory` has been called
971    /// with a Hash-based implementation.
972    pub fn full_iterator<'a: 'b, 'b>(
973        &'a self,
974        mode: IteratorMode,
975    ) -> DBIteratorWithThreadMode<'b, Self> {
976        let mut opts = ReadOptions::default();
977        opts.set_total_order_seek(true);
978        DBIteratorWithThreadMode::new(self, opts, mode)
979    }
980
981    pub fn prefix_iterator<'a: 'b, 'b, P: AsRef<[u8]>>(
982        &'a self,
983        prefix: P,
984    ) -> DBIteratorWithThreadMode<'b, Self> {
985        let mut opts = ReadOptions::default();
986        opts.set_prefix_same_as_start(true);
987        DBIteratorWithThreadMode::new(
988            self,
989            opts,
990            IteratorMode::From(prefix.as_ref(), Direction::Forward),
991        )
992    }
993
994    pub fn iterator_cf<'a: 'b, 'b>(
995        &'a self,
996        cf_handle: &impl AsColumnFamilyRef,
997        mode: IteratorMode,
998    ) -> DBIteratorWithThreadMode<'b, Self> {
999        let opts = ReadOptions::default();
1000        DBIteratorWithThreadMode::new_cf(self, cf_handle.inner(), opts, mode)
1001    }
1002
1003    pub fn full_iterator_cf<'a: 'b, 'b>(
1004        &'a self,
1005        cf_handle: &impl AsColumnFamilyRef,
1006        mode: IteratorMode,
1007    ) -> DBIteratorWithThreadMode<'b, Self> {
1008        let mut opts = ReadOptions::default();
1009        opts.set_total_order_seek(true);
1010        DBIteratorWithThreadMode::new_cf(self, cf_handle.inner(), opts, mode)
1011    }
1012
1013    pub fn prefix_iterator_cf<'a, P: AsRef<[u8]>>(
1014        &'a self,
1015        cf_handle: &impl AsColumnFamilyRef,
1016        prefix: P,
1017    ) -> DBIteratorWithThreadMode<'a, Self> {
1018        let mut opts = ReadOptions::default();
1019        opts.set_prefix_same_as_start(true);
1020        DBIteratorWithThreadMode::<'a, Self>::new_cf(
1021            self,
1022            cf_handle.inner(),
1023            opts,
1024            IteratorMode::From(prefix.as_ref(), Direction::Forward),
1025        )
1026    }
1027
1028    /// Opens a raw iterator over the database, using the default read options
1029    pub fn raw_iterator<'a: 'b, 'b>(&'a self) -> DBRawIteratorWithThreadMode<'b, Self> {
1030        let opts = ReadOptions::default();
1031        DBRawIteratorWithThreadMode::new(self, opts)
1032    }
1033
1034    /// Opens a raw iterator over the given column family, using the default read options
1035    pub fn raw_iterator_cf<'a: 'b, 'b>(
1036        &'a self,
1037        cf_handle: &impl AsColumnFamilyRef,
1038    ) -> DBRawIteratorWithThreadMode<'b, Self> {
1039        let opts = ReadOptions::default();
1040        DBRawIteratorWithThreadMode::new_cf(self, cf_handle.inner(), opts)
1041    }
1042
1043    /// Opens a raw iterator over the database, using the given read options
1044    pub fn raw_iterator_opt<'a: 'b, 'b>(
1045        &'a self,
1046        readopts: ReadOptions,
1047    ) -> DBRawIteratorWithThreadMode<'b, Self> {
1048        DBRawIteratorWithThreadMode::new(self, readopts)
1049    }
1050
1051    /// Opens a raw iterator over the given column family, using the given read options
1052    pub fn raw_iterator_cf_opt<'a: 'b, 'b>(
1053        &'a self,
1054        cf_handle: &impl AsColumnFamilyRef,
1055        readopts: ReadOptions,
1056    ) -> DBRawIteratorWithThreadMode<'b, Self> {
1057        DBRawIteratorWithThreadMode::new_cf(self, cf_handle.inner(), readopts)
1058    }
1059
1060    pub fn get_writebatch(&self) -> WriteBatchWithTransaction<true> {
1061        unsafe {
1062            let wi = ffi::rocksdb_transaction_get_writebatch_wi(self.inner);
1063            let mut len: usize = 0;
1064            let ptr = ffi::rocksdb_writebatch_wi_data(wi, std::ptr::from_mut(&mut len));
1065            let writebatch = ffi::rocksdb_writebatch_create_from(ptr, len);
1066            ffi::rocksdb_free(wi as *mut c_void);
1067            WriteBatchWithTransaction { inner: writebatch }
1068        }
1069    }
1070
1071    pub fn rebuild_from_writebatch(
1072        &self,
1073        writebatch: &WriteBatchWithTransaction<true>,
1074    ) -> Result<(), Error> {
1075        unsafe {
1076            ffi_try!(ffi::rocksdb_transaction_rebuild_from_writebatch(
1077                self.inner,
1078                writebatch.inner
1079            ));
1080        }
1081        Ok(())
1082    }
1083
1084    /// Replays the writes indexed by `writebatch` into this transaction.
1085    ///
1086    /// RocksDB walks the batch underlying the index and re-applies each record through
1087    /// this transaction's own put, merge and delete, so the keys end up tracked and
1088    /// conflict checked as if they had been written on the transaction directly. Under
1089    /// [`TxnDBWritePolicy::WriteUnprepared`] the records are already in the DB from WAL
1090    /// replay, so RocksDB only re-tracks the keys instead of rewriting them.
1091    ///
1092    /// Nothing is cleared first, so this layers on top of whatever the transaction
1093    /// already holds. Run it on a fresh transaction that has not been prepared or
1094    /// committed, the way RocksDB uses it when rebuilding prepared transactions during
1095    /// recovery. `writebatch` is only read, never taken over or modified.
1096    ///
1097    /// Returns an error if the batch carries two-phase commit markers, or if a replayed
1098    /// write fails for any of the reasons listed on [`put_cf`].
1099    ///
1100    /// [`TxnDBWritePolicy::WriteUnprepared`]: crate::TxnDBWritePolicy::WriteUnprepared
1101    /// [`put_cf`]: Self::put_cf
1102    pub fn rebuild_from_writebatch_wi(
1103        &self,
1104        writebatch: &WriteBatchWithIndex,
1105    ) -> Result<(), Error> {
1106        unsafe {
1107            ffi_try!(ffi::rocksdb_transaction_rebuild_from_writebatch_wi(
1108                self.inner,
1109                writebatch.inner
1110            ));
1111        }
1112        Ok(())
1113    }
1114
1115    /// Appends a blob of arbitrary size to the records in this transaction.
1116    ///
1117    /// The blob goes into the WAL only. It is never written to an SST file and is never
1118    /// visible to any read, so no `get` or iterator will ever return it. Iterating the
1119    /// transaction's write batch surfaces it through
1120    /// [`WriteBatchIterator::log_data`], interleaved with the puts, deletes and merges
1121    /// in insertion order. It consumes no sequence number and does not change the
1122    /// batch's count.
1123    ///
1124    /// A typical use is stamping the transaction log with data needed for replication.
1125    ///
1126    /// [`WriteBatchIterator::log_data`]: crate::WriteBatchIterator::log_data
1127    pub fn put_log_data<V: AsRef<[u8]>>(&self, log_data: V) {
1128        let log_data = log_data.as_ref();
1129        unsafe {
1130            ffi::rocksdb_transaction_put_log_data(
1131                self.inner,
1132                log_data.as_ptr() as *const c_char,
1133                log_data.len() as size_t,
1134            );
1135        }
1136    }
1137
1138    /// Sets the commit timestamp for this transaction.
1139    ///
1140    /// Only meaningful when the column family uses user-defined timestamps. RocksDB
1141    /// reports an unsupported timestamp through a status that the C API discards, so
1142    /// calling this on a transaction without timestamp support does nothing.
1143    pub fn set_commit_timestamp(&self, val: u64) {
1144        unsafe {
1145            ffi::rocksdb_transaction_set_commit_timestamp(self.inner, val);
1146        }
1147    }
1148
1149    /// Sets the read timestamp used to validate this transaction's reads.
1150    ///
1151    /// Only meaningful when the column family uses user-defined timestamps. RocksDB
1152    /// reports an unsupported timestamp through a status that the C API discards, so
1153    /// calling this on a transaction without timestamp support does nothing.
1154    pub fn set_read_timestamp_for_validation(&self, val: u64) {
1155        unsafe {
1156            ffi::rocksdb_transaction_set_read_timestamp_for_validation(self.inner, val);
1157        }
1158    }
1159}
1160
1161impl<DB> Drop for Transaction<'_, DB> {
1162    fn drop(&mut self) {
1163        unsafe {
1164            ffi::rocksdb_transaction_destroy(self.inner);
1165        }
1166    }
1167}
rust_rocksdb/transactions/transaction.rs

rust_rocksdb/transactions/
transaction.rs
