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, WriteBatchWithTransaction,
21    db::{DBAccess, convert_values},
22    ffi,
23};
24use libc::{c_char, c_void, size_t};
25
26// Default options are kept per-thread to avoid re-allocating on every call while
27// also preventing cross-thread sharing. Some RocksDB option wrappers hold
28// pointers into internal buffers and are not safe to share across threads.
29// Using thread_local allows cheap reuse in the common "default options" path
30// without synchronization overhead. Callers who need non-defaults must pass
31// explicit options.
32thread_local! { static DEFAULT_READ_OPTS: ReadOptions = ReadOptions::default(); }
33
34/// RocksDB Transaction.
35///
36/// To use transactions, you must first create a [`TransactionDB`] or [`OptimisticTransactionDB`].
37///
38/// [`TransactionDB`]: crate::TransactionDB
39/// [`OptimisticTransactionDB`]: crate::OptimisticTransactionDB
40///
41/// A `Transaction` must not outlive the `TransactionDB` it was created from:
42///
43/// ```compile_fail,E0597
44/// use rust_rocksdb::{SingleThreaded, TransactionDB};
45///
46/// let _txn = {
47///     let db = TransactionDB::<SingleThreaded>::open_default("foo").unwrap();
48///     db.transaction()
49/// };
50/// ```
51///
52/// A `Snapshot` taken from a `Transaction` must not outlive the `Transaction`:
53///
54/// ```compile_fail,E0597
55/// use rust_rocksdb::{SingleThreaded, TransactionDB};
56///
57/// let db = TransactionDB::<SingleThreaded>::open_default("foo").unwrap();
58/// let _snapshot = {
59///     let txn = db.transaction();
60///     txn.snapshot()
61/// };
62/// ```
63pub struct Transaction<'db, DB> {
64    pub(crate) inner: *mut ffi::rocksdb_transaction_t,
65    pub(crate) _marker: PhantomData<&'db DB>,
66}
67
68unsafe impl<DB> Send for Transaction<'_, DB> {}
69
70impl<DB> DBAccess for Transaction<'_, DB> {
71    unsafe fn create_snapshot(&self) -> *const ffi::rocksdb_snapshot_t {
72        unsafe { ffi::rocksdb_transaction_get_snapshot(self.inner) }
73    }
74
75    unsafe fn release_snapshot(&self, snapshot: *const ffi::rocksdb_snapshot_t) {
76        unsafe {
77            ffi::rocksdb_free(snapshot as *mut c_void);
78        }
79    }
80
81    unsafe fn create_iterator(&self, readopts: &ReadOptions) -> *mut ffi::rocksdb_iterator_t {
82        unsafe { ffi::rocksdb_transaction_create_iterator(self.inner, readopts.inner) }
83    }
84
85    unsafe fn create_iterator_cf(
86        &self,
87        cf_handle: *mut ffi::rocksdb_column_family_handle_t,
88        readopts: &ReadOptions,
89    ) -> *mut ffi::rocksdb_iterator_t {
90        unsafe {
91            ffi::rocksdb_transaction_create_iterator_cf(self.inner, readopts.inner, cf_handle)
92        }
93    }
94
95    fn get_opt<K: AsRef<[u8]>>(
96        &self,
97        key: K,
98        readopts: &ReadOptions,
99    ) -> Result<Option<Vec<u8>>, Error> {
100        self.get_opt(key, readopts)
101    }
102
103    fn get_cf_opt<K: AsRef<[u8]>>(
104        &self,
105        cf: &impl AsColumnFamilyRef,
106        key: K,
107        readopts: &ReadOptions,
108    ) -> Result<Option<Vec<u8>>, Error> {
109        self.get_cf_opt(cf, key, readopts)
110    }
111
112    fn get_pinned_opt<K: AsRef<[u8]>>(
113        &'_ self,
114        key: K,
115        readopts: &ReadOptions,
116    ) -> Result<Option<DBPinnableSlice<'_>>, Error> {
117        self.get_pinned_opt(key, readopts)
118    }
119
120    fn get_pinned_cf_opt<K: AsRef<[u8]>>(
121        &'_ self,
122        cf: &impl AsColumnFamilyRef,
123        key: K,
124        readopts: &ReadOptions,
125    ) -> Result<Option<DBPinnableSlice<'_>>, Error> {
126        self.get_pinned_cf_opt(cf, key, readopts)
127    }
128
129    fn multi_get_opt<K, I>(
130        &self,
131        keys: I,
132        readopts: &ReadOptions,
133    ) -> Vec<Result<Option<Vec<u8>>, Error>>
134    where
135        K: AsRef<[u8]>,
136        I: IntoIterator<Item = K>,
137    {
138        self.multi_get_opt(keys, readopts)
139    }
140
141    fn multi_get_cf_opt<'b, K, I, W>(
142        &self,
143        keys_cf: I,
144        readopts: &ReadOptions,
145    ) -> Vec<Result<Option<Vec<u8>>, Error>>
146    where
147        K: AsRef<[u8]>,
148        I: IntoIterator<Item = (&'b W, K)>,
149        W: AsColumnFamilyRef + 'b,
150    {
151        self.multi_get_cf_opt(keys_cf, readopts)
152    }
153}
154
155impl<DB> Transaction<'_, DB> {
156    /// Write all batched keys to the DB atomically.
157    ///
158    /// May return any error that could be returned by `DB::write`.
159    ///
160    /// If this transaction was created by a [`TransactionDB`], an error of
161    /// the [`Expired`] kind may be returned if this transaction has
162    /// lived longer than expiration time in [`TransactionOptions`].
163    ///
164    /// If this transaction was created by an [`OptimisticTransactionDB`], an error of
165    /// the [`Busy`] kind may be returned if the transaction
166    /// could not guarantee that there are no write conflicts.
167    /// An error of the [`TryAgain`] kind may be returned if the memtable
168    /// history size is not large enough (see [`Options::set_max_write_buffer_size_to_maintain`]).
169    ///
170    /// [`Expired`]: crate::ErrorKind::Expired
171    /// [`TransactionOptions`]: crate::TransactionOptions
172    /// [`TransactionDB`]: crate::TransactionDB
173    /// [`OptimisticTransactionDB`]: crate::OptimisticTransactionDB
174    /// [`Busy`]: crate::ErrorKind::Busy
175    /// [`TryAgain`]: crate::ErrorKind::TryAgain
176    /// [`Options::set_max_write_buffer_size_to_maintain`]: crate::Options::set_max_write_buffer_size_to_maintain
177    pub fn commit(self) -> Result<(), Error> {
178        unsafe {
179            ffi_try!(ffi::rocksdb_transaction_commit(self.inner));
180        }
181        Ok(())
182    }
183
184    pub fn set_name(&self, name: &[u8]) -> Result<(), Error> {
185        let ptr = name.as_ptr();
186        let len = name.len();
187        unsafe {
188            ffi_try!(ffi::rocksdb_transaction_set_name(
189                self.inner, ptr as _, len as _
190            ));
191        }
192
193        Ok(())
194    }
195
196    pub fn get_name(&self) -> Option<Vec<u8>> {
197        unsafe {
198            let mut name_len = 0;
199            let name = ffi::rocksdb_transaction_get_name(self.inner, &raw mut name_len);
200            if name.is_null() {
201                None
202            } else {
203                let mut vec = vec![0; name_len];
204                std::ptr::copy_nonoverlapping(name.cast::<u8>(), vec.as_mut_ptr(), name_len);
205                ffi::rocksdb_free(name as *mut c_void);
206                Some(vec)
207            }
208        }
209    }
210
211    pub fn prepare(&self) -> Result<(), Error> {
212        unsafe {
213            ffi_try!(ffi::rocksdb_transaction_prepare(self.inner));
214        }
215        Ok(())
216    }
217
218    /// Returns snapshot associated with transaction if snapshot was enabled in [`TransactionOptions`].
219    /// Otherwise, returns a snapshot with `nullptr` inside which doesn't affect read operations.
220    ///
221    /// [`TransactionOptions`]: crate::TransactionOptions
222    pub fn snapshot(&'_ self) -> SnapshotWithThreadMode<'_, Self> {
223        SnapshotWithThreadMode::new(self)
224    }
225
226    /// Discard all batched writes in this transaction.
227    pub fn rollback(&self) -> Result<(), Error> {
228        unsafe {
229            ffi_try!(ffi::rocksdb_transaction_rollback(self.inner));
230            Ok(())
231        }
232    }
233
234    /// Record the state of the transaction for future calls to [`rollback_to_savepoint`].
235    /// May be called multiple times to set multiple save points.
236    ///
237    /// [`rollback_to_savepoint`]: Self::rollback_to_savepoint
238    pub fn set_savepoint(&self) {
239        unsafe {
240            ffi::rocksdb_transaction_set_savepoint(self.inner);
241        }
242    }
243
244    /// Undo all operations in this transaction since the most recent call to [`set_savepoint`]
245    /// and removes the most recent [`set_savepoint`].
246    ///
247    /// Returns error if there is no previous call to [`set_savepoint`].
248    ///
249    /// [`set_savepoint`]: Self::set_savepoint
250    pub fn rollback_to_savepoint(&self) -> Result<(), Error> {
251        unsafe {
252            ffi_try!(ffi::rocksdb_transaction_rollback_to_savepoint(self.inner));
253            Ok(())
254        }
255    }
256
257    /// Get the bytes associated with a key value.
258    ///
259    /// See [`get_cf_opt`] for details.
260    ///
261    /// [`get_cf_opt`]: Self::get_cf_opt
262    pub fn get<K: AsRef<[u8]>>(&self, key: K) -> Result<Option<Vec<u8>>, Error> {
263        DEFAULT_READ_OPTS.with(|opts| self.get_opt(key, opts))
264    }
265
266    pub fn get_pinned<K: AsRef<[u8]>>(
267        &'_ self,
268        key: K,
269    ) -> Result<Option<DBPinnableSlice<'_>>, Error> {
270        DEFAULT_READ_OPTS.with(|opts| self.get_pinned_opt(key, opts))
271    }
272
273    /// Get the bytes associated with a key value and the given column family.
274    ///
275    /// See [`get_cf_opt`] for details.
276    ///
277    /// [`get_cf_opt`]: Self::get_cf_opt
278    pub fn get_cf<K: AsRef<[u8]>>(
279        &self,
280        cf: &impl AsColumnFamilyRef,
281        key: K,
282    ) -> Result<Option<Vec<u8>>, Error> {
283        DEFAULT_READ_OPTS.with(|opts| self.get_cf_opt(cf, key, opts))
284    }
285
286    pub fn get_pinned_cf<K: AsRef<[u8]>>(
287        &'_ self,
288        cf: &impl AsColumnFamilyRef,
289        key: K,
290    ) -> Result<Option<DBPinnableSlice<'_>>, Error> {
291        DEFAULT_READ_OPTS.with(|opts| self.get_pinned_cf_opt(cf, key, opts))
292    }
293
294    /// Get the key and ensure that this transaction will only
295    /// be able to be committed if this key is not written outside this
296    /// transaction after it has first been read (or after the snapshot if a
297    /// snapshot is set in this transaction).
298    ///
299    /// See [`get_for_update_cf_opt`] for details.
300    ///
301    /// [`get_for_update_cf_opt`]: Self::get_for_update_cf_opt
302    pub fn get_for_update<K: AsRef<[u8]>>(
303        &self,
304        key: K,
305        exclusive: bool,
306    ) -> Result<Option<Vec<u8>>, Error> {
307        DEFAULT_READ_OPTS.with(|opts| self.get_for_update_opt(key, exclusive, opts))
308    }
309
310    pub fn get_pinned_for_update<K: AsRef<[u8]>>(
311        &'_ self,
312        key: K,
313        exclusive: bool,
314    ) -> Result<Option<DBPinnableSlice<'_>>, Error> {
315        DEFAULT_READ_OPTS.with(|opts| self.get_pinned_for_update_opt(key, exclusive, opts))
316    }
317
318    /// Get the key in the given column family and ensure that this transaction will only
319    /// be able to be committed if this key is not written outside this
320    /// transaction after it has first been read (or after the snapshot if a
321    /// snapshot is set in this transaction).
322    ///
323    /// See [`get_for_update_cf_opt`] for details.
324    ///
325    /// [`get_for_update_cf_opt`]: Self::get_for_update_cf_opt
326    pub fn get_for_update_cf<K: AsRef<[u8]>>(
327        &self,
328        cf: &impl AsColumnFamilyRef,
329        key: K,
330        exclusive: bool,
331    ) -> Result<Option<Vec<u8>>, Error> {
332        DEFAULT_READ_OPTS.with(|opts| self.get_for_update_cf_opt(cf, key, exclusive, opts))
333    }
334
335    pub fn get_pinned_for_update_cf<K: AsRef<[u8]>>(
336        &'_ self,
337        cf: &impl AsColumnFamilyRef,
338        key: K,
339        exclusive: bool,
340    ) -> Result<Option<DBPinnableSlice<'_>>, Error> {
341        DEFAULT_READ_OPTS.with(|opts| self.get_pinned_for_update_cf_opt(cf, key, exclusive, opts))
342    }
343
344    /// Returns the bytes associated with a key value with read options.
345    ///
346    /// See [`get_cf_opt`] for details.
347    ///
348    /// [`get_cf_opt`]: Self::get_cf_opt
349    pub fn get_opt<K: AsRef<[u8]>>(
350        &self,
351        key: K,
352        readopts: &ReadOptions,
353    ) -> Result<Option<Vec<u8>>, Error> {
354        self.get_pinned_opt(key, readopts)
355            .map(|x| x.map(|v| v.as_ref().to_vec()))
356    }
357
358    pub fn get_pinned_opt<K: AsRef<[u8]>>(
359        &'_ self,
360        key: K,
361        readopts: &ReadOptions,
362    ) -> Result<Option<DBPinnableSlice<'_>>, Error> {
363        let key = key.as_ref();
364        unsafe {
365            let val = ffi_try!(ffi::rocksdb_transaction_get_pinned(
366                self.inner,
367                readopts.inner,
368                key.as_ptr() as *const c_char,
369                key.len(),
370            ));
371            if val.is_null() {
372                Ok(None)
373            } else {
374                Ok(Some(DBPinnableSlice::from_c(val)))
375            }
376        }
377    }
378
379    /// Get the bytes associated with a key value and the given column family with read options.
380    ///
381    /// This function will also read pending changes in this transaction.
382    /// Currently, this function will return an error of the [`MergeInProgress`] kind
383    /// if the most recent write to the queried key in this batch is a Merge.
384    ///
385    /// [`MergeInProgress`]: crate::ErrorKind::MergeInProgress
386    pub fn get_cf_opt<K: AsRef<[u8]>>(
387        &self,
388        cf: &impl AsColumnFamilyRef,
389        key: K,
390        readopts: &ReadOptions,
391    ) -> Result<Option<Vec<u8>>, Error> {
392        self.get_pinned_cf_opt(cf, key, readopts)
393            .map(|x| x.map(|v| v.as_ref().to_vec()))
394    }
395
396    pub fn get_pinned_cf_opt<K: AsRef<[u8]>>(
397        &'_ self,
398        cf: &impl AsColumnFamilyRef,
399        key: K,
400        readopts: &ReadOptions,
401    ) -> Result<Option<DBPinnableSlice<'_>>, Error> {
402        let key = key.as_ref();
403        unsafe {
404            let val = ffi_try!(ffi::rocksdb_transaction_get_pinned_cf(
405                self.inner,
406                readopts.inner,
407                cf.inner(),
408                key.as_ptr() as *const c_char,
409                key.len(),
410            ));
411            if val.is_null() {
412                Ok(None)
413            } else {
414                Ok(Some(DBPinnableSlice::from_c(val)))
415            }
416        }
417    }
418
419    /// Get the key with read options and ensure that this transaction will only
420    /// be able to be committed if this key is not written outside this
421    /// transaction after it has first been read (or after the snapshot if a
422    /// snapshot is set in this transaction).
423    ///
424    /// See [`get_for_update_cf_opt`] for details.
425    ///
426    /// [`get_for_update_cf_opt`]: Self::get_for_update_cf_opt
427    pub fn get_for_update_opt<K: AsRef<[u8]>>(
428        &self,
429        key: K,
430        exclusive: bool,
431        opts: &ReadOptions,
432    ) -> Result<Option<Vec<u8>>, Error> {
433        self.get_pinned_for_update_opt(key, exclusive, opts)
434            .map(|x| x.map(|v| v.as_ref().to_vec()))
435    }
436
437    pub fn get_pinned_for_update_opt<K: AsRef<[u8]>>(
438        &'_ self,
439        key: K,
440        exclusive: bool,
441        opts: &ReadOptions,
442    ) -> Result<Option<DBPinnableSlice<'_>>, Error> {
443        let key = key.as_ref();
444        unsafe {
445            let val = ffi_try!(ffi::rocksdb_transaction_get_pinned_for_update(
446                self.inner,
447                opts.inner,
448                key.as_ptr() as *const c_char,
449                key.len() as size_t,
450                u8::from(exclusive),
451            ));
452            if val.is_null() {
453                Ok(None)
454            } else {
455                Ok(Some(DBPinnableSlice::from_c(val)))
456            }
457        }
458    }
459
460    /// Get the key in the given column family with read options
461    /// and ensure that this transaction will only
462    /// be able to be committed if this key is not written outside this
463    /// transaction after it has first been read (or after the snapshot if a
464    /// snapshot is set in this transaction).
465    ///
466    /// Currently, this function will return an error of the [`MergeInProgress`]
467    /// if the most recent write to the queried key in this batch is a Merge.
468    ///
469    /// If this transaction was created by a [`TransactionDB`], it can return error of kind:
470    /// * [`Busy`] if there is a write conflict.
471    /// * [`TimedOut`] if a lock could not be acquired.
472    /// * [`TryAgain`] if the memtable history size is not large enough.
473    /// * [`MergeInProgress`] if merge operations cannot be resolved.
474    /// * or other errors if this key could not be read.
475    ///
476    /// If this transaction was created by an `[OptimisticTransactionDB]`, `get_for_update_opt`
477    /// can cause [`commit`] to fail. Otherwise, it could return any error that could
478    /// be returned by `[DB::get]`.
479    ///
480    /// [`Busy`]: crate::ErrorKind::Busy
481    /// [`TimedOut`]: crate::ErrorKind::TimedOut
482    /// [`TryAgain`]: crate::ErrorKind::TryAgain
483    /// [`MergeInProgress`]: crate::ErrorKind::MergeInProgress
484    /// [`TransactionDB`]: crate::TransactionDB
485    /// [`OptimisticTransactionDB`]: crate::OptimisticTransactionDB
486    /// [`commit`]: Self::commit
487    /// [`DB::get`]: crate::DB::get
488    pub fn get_for_update_cf_opt<K: AsRef<[u8]>>(
489        &self,
490        cf: &impl AsColumnFamilyRef,
491        key: K,
492        exclusive: bool,
493        opts: &ReadOptions,
494    ) -> Result<Option<Vec<u8>>, Error> {
495        self.get_pinned_for_update_cf_opt(cf, key, exclusive, opts)
496            .map(|x| x.map(|v| v.as_ref().to_vec()))
497    }
498
499    pub fn get_pinned_for_update_cf_opt<K: AsRef<[u8]>>(
500        &'_ self,
501        cf: &impl AsColumnFamilyRef,
502        key: K,
503        exclusive: bool,
504        opts: &ReadOptions,
505    ) -> Result<Option<DBPinnableSlice<'_>>, Error> {
506        let key = key.as_ref();
507        unsafe {
508            let val = ffi_try!(ffi::rocksdb_transaction_get_pinned_for_update_cf(
509                self.inner,
510                opts.inner,
511                cf.inner(),
512                key.as_ptr() as *const c_char,
513                key.len() as size_t,
514                u8::from(exclusive),
515            ));
516            if val.is_null() {
517                Ok(None)
518            } else {
519                Ok(Some(DBPinnableSlice::from_c(val)))
520            }
521        }
522    }
523
524    /// Return the values associated with the given keys.
525    pub fn multi_get<K, I>(&self, keys: I) -> Vec<Result<Option<Vec<u8>>, Error>>
526    where
527        K: AsRef<[u8]>,
528        I: IntoIterator<Item = K>,
529    {
530        DEFAULT_READ_OPTS.with(|opts| self.multi_get_opt(keys, opts))
531    }
532
533    /// Return the values associated with the given keys using read options.
534    pub fn multi_get_opt<K, I>(
535        &self,
536        keys: I,
537        readopts: &ReadOptions,
538    ) -> Vec<Result<Option<Vec<u8>>, Error>>
539    where
540        K: AsRef<[u8]>,
541        I: IntoIterator<Item = K>,
542    {
543        let owned_keys: Vec<K> = keys.into_iter().collect();
544        let (ptr_keys, keys_sizes): (Vec<*const c_char>, Vec<usize>) = owned_keys
545            .iter()
546            .map(|k| {
547                let key = k.as_ref();
548                (key.as_ptr() as *const c_char, key.len())
549            })
550            .unzip();
551
552        let mut values: Vec<*mut c_char> = Vec::with_capacity(ptr_keys.len());
553        let mut values_sizes: Vec<usize> = Vec::with_capacity(ptr_keys.len());
554        let mut errors: Vec<*mut c_char> = Vec::with_capacity(ptr_keys.len());
555        unsafe {
556            ffi::rocksdb_transaction_multi_get(
557                self.inner,
558                readopts.inner,
559                ptr_keys.len(),
560                ptr_keys.as_ptr(),
561                keys_sizes.as_ptr(),
562                values.as_mut_ptr(),
563                values_sizes.as_mut_ptr(),
564                errors.as_mut_ptr(),
565            );
566        }
567
568        unsafe {
569            values.set_len(ptr_keys.len());
570            values_sizes.set_len(ptr_keys.len());
571            errors.set_len(ptr_keys.len());
572        }
573
574        convert_values(values, values_sizes, errors)
575    }
576
577    /// Return the values associated with the given keys and column families.
578    pub fn multi_get_cf<'a, 'b: 'a, K, I, W>(
579        &'a self,
580        keys: I,
581    ) -> Vec<Result<Option<Vec<u8>>, Error>>
582    where
583        K: AsRef<[u8]>,
584        I: IntoIterator<Item = (&'b W, K)>,
585        W: 'b + AsColumnFamilyRef,
586    {
587        DEFAULT_READ_OPTS.with(|opts| self.multi_get_cf_opt(keys, opts))
588    }
589
590    /// Return the values associated with the given keys and column families using read options.
591    pub fn multi_get_cf_opt<'a, 'b: 'a, K, I, W>(
592        &'a self,
593        keys: I,
594        readopts: &ReadOptions,
595    ) -> Vec<Result<Option<Vec<u8>>, Error>>
596    where
597        K: AsRef<[u8]>,
598        I: IntoIterator<Item = (&'b W, K)>,
599        W: 'b + AsColumnFamilyRef,
600    {
601        let cfs_and_owned_keys: Vec<(&'b W, K)> = keys.into_iter().collect();
602        let (ptr_keys, keys_sizes): (Vec<*const c_char>, Vec<usize>) = cfs_and_owned_keys
603            .iter()
604            .map(|(_, k)| {
605                let key = k.as_ref();
606                (key.as_ptr() as *const c_char, key.len())
607            })
608            .unzip();
609        let ptr_cfs: Vec<*const ffi::rocksdb_column_family_handle_t> = cfs_and_owned_keys
610            .iter()
611            .map(|(c, _)| c.inner().cast_const())
612            .collect();
613        let mut values: Vec<*mut c_char> = Vec::with_capacity(ptr_keys.len());
614        let mut values_sizes: Vec<usize> = Vec::with_capacity(ptr_keys.len());
615        let mut errors: Vec<*mut c_char> = Vec::with_capacity(ptr_keys.len());
616        unsafe {
617            ffi::rocksdb_transaction_multi_get_cf(
618                self.inner,
619                readopts.inner,
620                ptr_cfs.as_ptr(),
621                ptr_keys.len(),
622                ptr_keys.as_ptr(),
623                keys_sizes.as_ptr(),
624                values.as_mut_ptr(),
625                values_sizes.as_mut_ptr(),
626                errors.as_mut_ptr(),
627            );
628        }
629
630        unsafe {
631            values.set_len(ptr_keys.len());
632            values_sizes.set_len(ptr_keys.len());
633            errors.set_len(ptr_keys.len());
634        }
635
636        convert_values(values, values_sizes, errors)
637    }
638
639    /// Put the key value in default column family and do conflict checking on the key.
640    ///
641    /// See [`put_cf`] for details.
642    ///
643    /// [`put_cf`]: Self::put_cf
644    pub fn put<K: AsRef<[u8]>, V: AsRef<[u8]>>(&self, key: K, value: V) -> Result<(), Error> {
645        let key = key.as_ref();
646        let value = value.as_ref();
647        unsafe {
648            ffi_try!(ffi::rocksdb_transaction_put(
649                self.inner,
650                key.as_ptr() as *const c_char,
651                key.len() as size_t,
652                value.as_ptr() as *const c_char,
653                value.len() as size_t,
654            ));
655            Ok(())
656        }
657    }
658
659    /// Put the key value in the given column family and do conflict checking on the key.
660    ///
661    /// If this transaction was created by a [`TransactionDB`], it can return error of kind:
662    /// * [`Busy`] if there is a write conflict.
663    /// * [`TimedOut`] if a lock could not be acquired.
664    /// * [`TryAgain`] if the memtable history size is not large enough.
665    /// * [`MergeInProgress`] if merge operations cannot be resolved.
666    /// * or other errors on unexpected failures.
667    ///
668    /// [`Busy`]: crate::ErrorKind::Busy
669    /// [`TimedOut`]: crate::ErrorKind::TimedOut
670    /// [`TryAgain`]: crate::ErrorKind::TryAgain
671    /// [`MergeInProgress`]: crate::ErrorKind::MergeInProgress
672    /// [`TransactionDB`]: crate::TransactionDB
673    /// [`OptimisticTransactionDB`]: crate::OptimisticTransactionDB
674    pub fn put_cf<K: AsRef<[u8]>, V: AsRef<[u8]>>(
675        &self,
676        cf: &impl AsColumnFamilyRef,
677        key: K,
678        value: V,
679    ) -> Result<(), Error> {
680        let key = key.as_ref();
681        let value = value.as_ref();
682        unsafe {
683            ffi_try!(ffi::rocksdb_transaction_put_cf(
684                self.inner,
685                cf.inner(),
686                key.as_ptr() as *const c_char,
687                key.len() as size_t,
688                value.as_ptr() as *const c_char,
689                value.len() as size_t,
690            ));
691            Ok(())
692        }
693    }
694
695    /// Merge value with existing value of key, and also do conflict checking on the key.
696    ///
697    /// See [`merge_cf`] for details.
698    ///
699    /// [`merge_cf`]: Self::merge_cf
700    pub fn merge<K: AsRef<[u8]>, V: AsRef<[u8]>>(&self, key: K, value: V) -> Result<(), Error> {
701        let key = key.as_ref();
702        let value = value.as_ref();
703        unsafe {
704            ffi_try!(ffi::rocksdb_transaction_merge(
705                self.inner,
706                key.as_ptr() as *const c_char,
707                key.len() as size_t,
708                value.as_ptr() as *const c_char,
709                value.len() as size_t
710            ));
711            Ok(())
712        }
713    }
714
715    /// Merge `value` with existing value of `key` in the given column family,
716    /// and also do conflict checking on the key.
717    ///
718    /// If this transaction was created by a [`TransactionDB`], it can return error of kind:
719    /// * [`Busy`] if there is a write conflict.
720    /// * [`TimedOut`] if a lock could not be acquired.
721    /// * [`TryAgain`] if the memtable history size is not large enough.
722    /// * [`MergeInProgress`] if merge operations cannot be resolved.
723    /// * or other errors on unexpected failures.
724    ///
725    /// [`Busy`]: crate::ErrorKind::Busy
726    /// [`TimedOut`]: crate::ErrorKind::TimedOut
727    /// [`TryAgain`]: crate::ErrorKind::TryAgain
728    /// [`MergeInProgress`]: crate::ErrorKind::MergeInProgress
729    /// [`TransactionDB`]: crate::TransactionDB
730    pub fn merge_cf<K: AsRef<[u8]>, V: AsRef<[u8]>>(
731        &self,
732        cf: &impl AsColumnFamilyRef,
733        key: K,
734        value: V,
735    ) -> Result<(), Error> {
736        let key = key.as_ref();
737        let value = value.as_ref();
738        unsafe {
739            ffi_try!(ffi::rocksdb_transaction_merge_cf(
740                self.inner,
741                cf.inner(),
742                key.as_ptr() as *const c_char,
743                key.len() as size_t,
744                value.as_ptr() as *const c_char,
745                value.len() as size_t
746            ));
747            Ok(())
748        }
749    }
750
751    /// Delete the key value if it exists and do conflict checking on the key.
752    ///
753    /// See [`delete_cf`] for details.
754    ///
755    /// [`delete_cf`]: Self::delete_cf
756    pub fn delete<K: AsRef<[u8]>>(&self, key: K) -> Result<(), Error> {
757        let key = key.as_ref();
758        unsafe {
759            ffi_try!(ffi::rocksdb_transaction_delete(
760                self.inner,
761                key.as_ptr() as *const c_char,
762                key.len() as size_t
763            ));
764        }
765        Ok(())
766    }
767
768    /// Delete the key value in the given column family and do conflict checking.
769    ///
770    /// If this transaction was created by a [`TransactionDB`], it can return error of kind:
771    /// * [`Busy`] if there is a write conflict.
772    /// * [`TimedOut`] if a lock could not be acquired.
773    /// * [`TryAgain`] if the memtable history size is not large enough.
774    /// * [`MergeInProgress`] if merge operations cannot be resolved.
775    /// * or other errors on unexpected failures.
776    ///
777    /// [`Busy`]: crate::ErrorKind::Busy
778    /// [`TimedOut`]: crate::ErrorKind::TimedOut
779    /// [`TryAgain`]: crate::ErrorKind::TryAgain
780    /// [`MergeInProgress`]: crate::ErrorKind::MergeInProgress
781    /// [`TransactionDB`]: crate::TransactionDB
782    pub fn delete_cf<K: AsRef<[u8]>>(
783        &self,
784        cf: &impl AsColumnFamilyRef,
785        key: K,
786    ) -> Result<(), Error> {
787        let key = key.as_ref();
788        unsafe {
789            ffi_try!(ffi::rocksdb_transaction_delete_cf(
790                self.inner,
791                cf.inner(),
792                key.as_ptr() as *const c_char,
793                key.len() as size_t
794            ));
795        }
796        Ok(())
797    }
798
799    pub fn iterator<'a: 'b, 'b>(
800        &'a self,
801        mode: IteratorMode,
802    ) -> DBIteratorWithThreadMode<'b, Self> {
803        let readopts = ReadOptions::default();
804        self.iterator_opt(mode, readopts)
805    }
806
807    pub fn iterator_opt<'a: 'b, 'b>(
808        &'a self,
809        mode: IteratorMode,
810        readopts: ReadOptions,
811    ) -> DBIteratorWithThreadMode<'b, Self> {
812        DBIteratorWithThreadMode::new(self, readopts, mode)
813    }
814
815    /// Opens an iterator using the provided ReadOptions.
816    /// This is used when you want to iterate over a specific ColumnFamily with a modified ReadOptions.
817    pub fn iterator_cf_opt<'a: 'b, 'b>(
818        &'a self,
819        cf_handle: &impl AsColumnFamilyRef,
820        readopts: ReadOptions,
821        mode: IteratorMode,
822    ) -> DBIteratorWithThreadMode<'b, Self> {
823        DBIteratorWithThreadMode::new_cf(self, cf_handle.inner(), readopts, mode)
824    }
825
826    /// Opens an iterator with `set_total_order_seek` enabled.
827    /// This must be used to iterate across prefixes when `set_memtable_factory` has been called
828    /// with a Hash-based implementation.
829    pub fn full_iterator<'a: 'b, 'b>(
830        &'a self,
831        mode: IteratorMode,
832    ) -> DBIteratorWithThreadMode<'b, Self> {
833        let mut opts = ReadOptions::default();
834        opts.set_total_order_seek(true);
835        DBIteratorWithThreadMode::new(self, opts, mode)
836    }
837
838    pub fn prefix_iterator<'a: 'b, 'b, P: AsRef<[u8]>>(
839        &'a self,
840        prefix: P,
841    ) -> DBIteratorWithThreadMode<'b, Self> {
842        let mut opts = ReadOptions::default();
843        opts.set_prefix_same_as_start(true);
844        DBIteratorWithThreadMode::new(
845            self,
846            opts,
847            IteratorMode::From(prefix.as_ref(), Direction::Forward),
848        )
849    }
850
851    pub fn iterator_cf<'a: 'b, 'b>(
852        &'a self,
853        cf_handle: &impl AsColumnFamilyRef,
854        mode: IteratorMode,
855    ) -> DBIteratorWithThreadMode<'b, Self> {
856        let opts = ReadOptions::default();
857        DBIteratorWithThreadMode::new_cf(self, cf_handle.inner(), opts, mode)
858    }
859
860    pub fn full_iterator_cf<'a: 'b, 'b>(
861        &'a self,
862        cf_handle: &impl AsColumnFamilyRef,
863        mode: IteratorMode,
864    ) -> DBIteratorWithThreadMode<'b, Self> {
865        let mut opts = ReadOptions::default();
866        opts.set_total_order_seek(true);
867        DBIteratorWithThreadMode::new_cf(self, cf_handle.inner(), opts, mode)
868    }
869
870    pub fn prefix_iterator_cf<'a, P: AsRef<[u8]>>(
871        &'a self,
872        cf_handle: &impl AsColumnFamilyRef,
873        prefix: P,
874    ) -> DBIteratorWithThreadMode<'a, Self> {
875        let mut opts = ReadOptions::default();
876        opts.set_prefix_same_as_start(true);
877        DBIteratorWithThreadMode::<'a, Self>::new_cf(
878            self,
879            cf_handle.inner(),
880            opts,
881            IteratorMode::From(prefix.as_ref(), Direction::Forward),
882        )
883    }
884
885    /// Opens a raw iterator over the database, using the default read options
886    pub fn raw_iterator<'a: 'b, 'b>(&'a self) -> DBRawIteratorWithThreadMode<'b, Self> {
887        let opts = ReadOptions::default();
888        DBRawIteratorWithThreadMode::new(self, opts)
889    }
890
891    /// Opens a raw iterator over the given column family, using the default read options
892    pub fn raw_iterator_cf<'a: 'b, 'b>(
893        &'a self,
894        cf_handle: &impl AsColumnFamilyRef,
895    ) -> DBRawIteratorWithThreadMode<'b, Self> {
896        let opts = ReadOptions::default();
897        DBRawIteratorWithThreadMode::new_cf(self, cf_handle.inner(), opts)
898    }
899
900    /// Opens a raw iterator over the database, using the given read options
901    pub fn raw_iterator_opt<'a: 'b, 'b>(
902        &'a self,
903        readopts: ReadOptions,
904    ) -> DBRawIteratorWithThreadMode<'b, Self> {
905        DBRawIteratorWithThreadMode::new(self, readopts)
906    }
907
908    /// Opens a raw iterator over the given column family, using the given read options
909    pub fn raw_iterator_cf_opt<'a: 'b, 'b>(
910        &'a self,
911        cf_handle: &impl AsColumnFamilyRef,
912        readopts: ReadOptions,
913    ) -> DBRawIteratorWithThreadMode<'b, Self> {
914        DBRawIteratorWithThreadMode::new_cf(self, cf_handle.inner(), readopts)
915    }
916
917    pub fn get_writebatch(&self) -> WriteBatchWithTransaction<true> {
918        unsafe {
919            let wi = ffi::rocksdb_transaction_get_writebatch_wi(self.inner);
920            let mut len: usize = 0;
921            let ptr = ffi::rocksdb_writebatch_wi_data(wi, std::ptr::from_mut(&mut len));
922            let writebatch = ffi::rocksdb_writebatch_create_from(ptr, len);
923            ffi::rocksdb_free(wi as *mut c_void);
924            WriteBatchWithTransaction { inner: writebatch }
925        }
926    }
927
928    pub fn rebuild_from_writebatch(
929        &self,
930        writebatch: &WriteBatchWithTransaction<true>,
931    ) -> Result<(), Error> {
932        unsafe {
933            ffi_try!(ffi::rocksdb_transaction_rebuild_from_writebatch(
934                self.inner,
935                writebatch.inner
936            ));
937        }
938        Ok(())
939    }
940}
941
942impl<DB> Drop for Transaction<'_, DB> {
943    fn drop(&mut self) {
944        unsafe {
945            ffi::rocksdb_transaction_destroy(self.inner);
946        }
947    }
948}
rust_rocksdb/transactions/transaction.rs

rust_rocksdb/transactions/
transaction.rs
