<commit_before>use std::{
    borrow::Cow;
    cmp::Ordering::{Greater, Less},
    fmt::{self, Debug},
    sync::{
        atomic::{AtomicUsize, Ordering::SeqCst},
        Arc,
    },
};

use pagecache::PagePtr;

use super::*;

type Path<'g> = Vec<(&'g Frag, TreePtr<'g>)>;

impl<'a> IntoIterator for &'a Tree {
    type Item = Result<(Vec<u8>, PinnedValue), ()>;
    type IntoIter = Iter<'a>;

    fn into_iter(self) -> Iter<'a> {
        self.iter()
    }
}

/// A flash-sympathetic persistent lock-free B+ tree
#[derive(Clone)]
pub struct Tree {
    pub(crate) pages: Arc<
        PageCache<BLinkMaterializer, Frag, Vec<(PageId, PageId)>>,
    >,
    config: Config,
    root: Arc<AtomicUsize>,
}

unsafe impl Send for Tree {}
unsafe impl Sync for Tree {}

#[cfg(feature = "event_log")]
impl Drop for Tree {
    fn drop(&mut self) {
        self.config
            .event_log
            .tree_root_before_restart(self.root.load(SeqCst));
    }
}

impl Tree {
    /// Load existing or create a new `Tree` with a default configuration.
    pub fn start_default<P: AsRef<std::path::Path>>(
        path: P,
    ) -> Result<Tree, ()> {
        let config = ConfigBuilder::new().path(path).build();
        Self::start(config)
    }

    /// Load existing or create a new `Tree`.
    pub fn start(config: Config) -> Result<Tree, ()> {
        let _measure = Measure::new(&M.tree_start);

        #[cfg(any(test, feature = "check_snapshot_integrity"))]
        match config
            .verify_snapshot::<BLinkMaterializer, Frag, Vec<(PageId, PageId)>>(
            ) {
            Ok(_) => {}
            #[cfg(feature = "failpoints")]
            Err(Error::FailPoint) => {}
            other => panic!("failed to verify snapshot: {:?}", other),
        }

        let pages = PageCache::start(config.clone())?;

        let roots_opt = pages.recovered_state().clone().and_then(
            |mut roots: Vec<(PageId, PageId)>| {
                if roots.is_empty() {
                    None
                } else {
                    let mut last = std::usize::MAX;
                    let mut last_idx = std::usize::MAX;
                    while !roots.is_empty() {
                        // find the root that links to the last one
                        for (i, &(root, prev_root)) in
                            roots.iter().enumerate()
                        {
                            if prev_root == last {
                                last = root;
                                last_idx = i;
                                break;
                            }
                            assert_ne!(
                                i + 1,
                                roots.len(),
                                "encountered gap in root chain"
                            );
                        }
                        roots.remove(last_idx);
                    }
                    assert_ne!(last, std::usize::MAX);
                    Some(last)
                }
            },
        );

        let root_id = if let Some(root_id) = roots_opt {
            debug!("recovered root {} while starting tree", root_id);
            root_id
        } else {
            let guard = pin();

            let root_id = pages.allocate(&guard)?;
            assert_eq!(
                root_id,
                0,
                "we expect that this is the first page ever allocated"
            );
            debug!("allocated pid {} for root of new tree", root_id);

            let leaf_id = pages.allocate(&guard)?;
            trace!("allocated pid {} for leaf in new", leaf_id);

            let leaf = Frag::Base(
                Node {
                    id: leaf_id,
                    data: Data::Leaf(vec![]),
                    next: None,
                    lo: Bound::Inclusive(vec![]),
                    hi: Bound::Inf,
                },
                None,
            );

            // vec![0] represents a prefix-encoded empty prefix
            let root_index_vec = vec![(vec![0], leaf_id)];

            let root = Frag::Base(
                Node {
                    id: root_id,
                    data: Data::Index(root_index_vec),
                    next: None,
                    lo: Bound::Inclusive(vec![]),
                    hi: Bound::Inf,
                },
                Some(std::usize::MAX),
            );

            pages
                .replace(root_id, PagePtr::allocated(), root, &guard)
                .map_err(|e| e.danger_cast())?;
            pages
                .replace(leaf_id, PagePtr::allocated(), leaf, &guard)
                .map_err(|e| e.danger_cast())?;

            guard.flush();

            root_id
        };

        #[cfg(feature = "event_log")]
        config.event_log.tree_root_after_restart(root_id);

        Ok(Tree {
            pages: Arc::new(pages),
            config,
            root: Arc::new(AtomicUsize::new(root_id)),
        })
    }

    /// Flushes any pending IO buffers to disk to ensure durability.
    pub fn flush(&self) -> Result<(), ()> {
        self.pages.flush()
    }

    /// Retrieve a value from the `Tree` if it exists.
    pub fn get<K: AsRef<[u8]>>(
        &self,
        key: K,
    ) -> Result<Option<PinnedValue>, ()> {
        let _measure = Measure::new(&M.tree_get);

        let guard = pin();

        // the double guard is a hack that maintains
        // correctness of the ret value
        let double_guard = guard.clone();
        let (_, ret) = self.get_internal(key.as_ref(), &guard)?;

        guard.flush();

        Ok(ret.map(|r| PinnedValue::new(r, double_guard)))
    }

    /// Retrieve a value from the `Tree` for the key before the
    /// given key, if one exists.
    ///
    /// # Examples
    ///
    /// ```
    /// use sled::{ConfigBuilder, Error};
    /// let config = ConfigBuilder::new().temporary(true).build();
    ///
    /// let tree = sled::Tree::start(config).unwrap();
    ///
    /// for i in 0..10 {
    ///     tree.set(vec![i], vec![i]).expect("should write successfully");
    /// }
    ///
    /// assert!(tree.get_lt(vec![]).unwrap().is_none());
    /// assert!(tree.get_lt(vec![0]).unwrap().is_none());
    /// assert!(tree.get_lt(vec![1]).unwrap().unwrap().1 == vec![0]);
    /// assert!(tree.get_lt(vec![9]).unwrap().unwrap().1 == vec![8]);
    /// assert!(tree.get_lt(vec![10]).unwrap().unwrap().1 == vec![9]);
    /// assert!(tree.get_lt(vec![255]).unwrap().unwrap().1 == vec![9]);
    /// ```
    pub fn get_lt<K: AsRef<[u8]>>(
        &self,
        key: K,
    ) -> Result<Option<(Key, PinnedValue)>, ()> {
        let _measure = Measure::new(&M.tree_get);

        // the double guard is a hack that maintains
        // correctness of the ret value
        let guard = pin();
        let double_guard = guard.clone();

        let path = self.path_for_key(key.as_ref(), &guard)?;
        let (last_frag, _tree_ptr) = path
            .last()
            .expect("path should always contain a last element");

        let last_node = last_frag.unwrap_base();
        let data = &last_node.data;
        let items =
            data.leaf_ref().expect("last_node should be a leaf");
        let search = leaf_search(Less, items, |&(ref k, ref _v)| {
            prefix_cmp_encoded(k, key.as_ref(), last_node.lo.inner())
        });

        let ret = if search.is_none() {
            let mut iter = Iter {
                id: last_node.id,
                inner: &self,
                last_key: key.as_ref().to_vec(),
                inclusive: false,
                broken: None,
                done: false,
                guard: guard.clone(),
            };

            match iter.next_back() {
                Some(Err(e)) => return Err(e),
                Some(Ok(pair)) => Some(pair),
                None => None,
            }
        } else {
            let idx = search.unwrap();
            let (encoded_key, v) = &items[idx];
            Some((
                prefix_decode(last_node.lo.inner(), &*encoded_key),
                PinnedValue::new(&*v, double_guard),
            ))
        };

        guard.flush();

        Ok(ret)
    }

    /// Retrieve a value from the `Tree` for the key after the
    /// given key, if one exists.
    ///
    /// # Examples
    ///
    /// ```
    /// use sled::{ConfigBuilder, Error};
    /// let config = ConfigBuilder::new().temporary(true).build();
    ///
    /// let tree = sled::Tree::start(config).unwrap();
    ///
    /// for i in 0..10 {
    ///     tree.set(vec![i], vec![i]).expect("should write successfully");
    /// }
    ///
    /// assert!(tree.get_gt(vec![]).unwrap().unwrap().1 == vec![0]);
    /// assert!(tree.get_gt(vec![0]).unwrap().unwrap().1 == vec![1]);
    /// assert!(tree.get_gt(vec![1]).unwrap().unwrap().1 == vec![2]);
    /// assert!(tree.get_gt(vec![8]).unwrap().unwrap().1 == vec![9]);
    /// assert!(tree.get_gt(vec![9]).unwrap().is_none());
    /// ```
    pub fn get_gt<K: AsRef<[u8]>>(
        &self,
        key: K,
    ) -> Result<Option<(Key, PinnedValue)>, ()> {
        let _measure = Measure::new(&M.tree_get);

        // the double guard is a hack that maintains
        // correctness of the ret value
        let guard = pin();
        let double_guard = guard.clone();

        let path = self.path_for_key(key.as_ref(), &guard)?;
        let (last_frag, _tree_ptr) = path
            .last()
            .expect("path should always contain a last element");

        let last_node = last_frag.unwrap_base();
        let data = &last_node.data;
        let items =
            data.leaf_ref().expect("last_node should be a leaf");
        let search =
            leaf_search(Greater, items, |&(ref k, ref _v)| {
                prefix_cmp_encoded(
                    k,
                    key.as_ref(),
                    last_node.lo.inner(),
                )
            });

        let ret = if search.is_none() {
            let mut iter = Iter {
                id: last_node.id,
                inner: &self,
                last_key: key.as_ref().to_vec(),
                inclusive: false,
                broken: None,
                done: false,
                guard: guard.clone(),
            };

            match iter.next() {
                Some(Err(e)) => return Err(e),
                Some(Ok(pair)) => Some(pair),
                None => None,
            }
        } else {
            let idx = search.unwrap();
            let (encoded_key, v) = &items[idx];
            Some((
                prefix_decode(last_node.lo.inner(), &*encoded_key),
                PinnedValue::new(&*v, double_guard),
            ))
        };

        guard.flush();

        Ok(ret)
    }

    /// Compare and swap. Capable of unique creation, conditional modification,
    /// or deletion. If old is None, this will only set the value if it doesn't
    /// exist yet. If new is None, will delete the value if old is correct.
    /// If both old and new are Some, will modify the value if old is correct.
    /// If Tree is read-only, will do nothing.
    ///
    /// # Examples
    ///
    /// ```
    /// use sled::{ConfigBuilder, Error};
    /// let config = ConfigBuilder::new().temporary(true).build();
    /// let t = sled::Tree::start(config).unwrap();
    ///
    /// // unique creation
    /// assert_eq!(t.cas(&[1], None, Some(vec![1])), Ok(()));
    /// // assert_eq!(t.cas(&[1], None, Some(vec![1])), Err(Error::CasFailed(Some(vec![1]))));
    ///
    /// // conditional modification
    /// assert_eq!(t.cas(&[1], Some(&*vec![1]), Some(vec![2])), Ok(()));
    /// // assert_eq!(t.cas(&[1], Some(vec![1]), Some(vec![2])), Err(Error::CasFailed(Some(vec![2]))));
    ///
    /// // conditional deletion
    /// assert_eq!(t.cas(&[1], Some(&[2]), None), Ok(()));
    /// assert_eq!(t.get(&[1]), Ok(None));
    /// ```
    pub fn cas<K: AsRef<[u8]>>(
        &self,
        key: K,
        old: Option<&[u8]>,
        new: Option<Value>,
    ) -> Result<(), Option<PinnedValue>> {
        let _measure = Measure::new(&M.tree_cas);

        if self.config.read_only {
            return Err(Error::CasFailed(None));
        }

        let guard = pin();

        // we need to retry caps until old != cur, since just because
        // cap fails it doesn't mean our value was changed.
        loop {
            let pin_guard = guard.clone();
            let (mut path, cur) = self
                .get_internal(key.as_ref(), &guard)
                .map_err(|e| e.danger_cast())?;

            if old.as_ref().map(|o| o.as_ref()) != cur.map(|v| &*v) {
                return Err(Error::CasFailed(
                    cur.map(|c| PinnedValue::new(c, pin_guard)),
                ));
            }

            let (leaf_frag, leaf_ptr) = path.pop().expect(
                "get_internal somehow returned a path of length zero",
            );

            let (node_id, encoded_key) = {
                let node: &Node = leaf_frag.unwrap_base();
                (
                    node.id,
                    prefix_encode(node.lo.inner(), key.as_ref()),
                )
            };
            let frag = if let Some(ref n) = new {
                Frag::Set(encoded_key, n.clone())
            } else {
                Frag::Del(encoded_key)
            };
            let link = self.pages.link(
                node_id,
                leaf_ptr,
                frag.clone(),
                &guard,
            );
            match link {
                Ok(_) => {
                    guard.flush();
                    return Ok(());
                }
                Err(Error::CasFailed(_)) => {}
                Err(other) => {
                    guard.flush();
                    return Err(other.danger_cast());
                }
            }
            M.tree_looped();
        }
    }

    /// Set a key to a new value, returning the old value if it
    /// was set.
    pub fn set<K: AsRef<[u8]>>(
        &self,
        key: K,
        value: Value,
    ) -> Result<Option<PinnedValue>, ()> {
        let _measure = Measure::new(&M.tree_set);

        if self.config.read_only {
            return Err(Error::Unsupported(
                "the database is in read-only mode".to_owned(),
            ));
        }

        let guard = pin();

        loop {
            let double_guard = guard.clone();
            let (mut path, existing_key) =
                self.get_internal(key.as_ref(), &guard)?;
            let (leaf_frag, leaf_ptr) = path.pop().expect(
                "path_for_key should always return a path \
                 of length >= 2 (root + leaf)",
            );
            let node: &Node = leaf_frag.unwrap_base();
            let encoded_key =
                prefix_encode(node.lo.inner(), key.as_ref());

            let frag = Frag::Set(encoded_key, value.clone());
            let link = self.pages.link(
                node.id,
                leaf_ptr.clone(),
                frag.clone(),
                &guard,
            );
            match link {
                Ok(new_cas_key) => {
                    // success
                    if node.should_split(
                        self.config.blink_node_split_size,
                    ) {
                        let mut path2 = path
                            .iter()
                            .map(|&(f, ref p)| (Cow::Borrowed(f), p.clone()))
                            .collect::<Vec<(Cow<'_, Frag>, _)>>();
                        let mut node2 = node.clone();
                        node2
                            .apply(&frag, self.config.merge_operator);
                        let frag2 = Cow::Owned(Frag::Base(node2, None));
                        path2.push((frag2, new_cas_key));
                        self.recursive_split(path2, &guard)?;
                    }

                    guard.flush();

                    return Ok(existing_key.map(move |r| {
                        PinnedValue::new(r, double_guard)
                    }));
                }
                Err(Error::CasFailed(_)) => {}
                Err(other) => {
                    guard.flush();

                    return Err(other.danger_cast());
                }
            }
            M.tree_looped();
        }
    }

    /// Merge a new value into the total state for a key.
    ///
    /// # Examples
    ///
    /// ```
    /// fn concatenate_merge(
    ///   _key: &[u8],               // the key being merged
    ///   old_value: Option<&[u8]>,  // the previous value, if one existed
    ///   merged_bytes: &[u8]        // the new bytes being merged in
    /// ) -> Option<Vec<u8>> {       // set the new value, return None to delete
    ///   let mut ret = old_value
    ///     .map(|ov| ov.to_vec())
    ///     .unwrap_or_else(|| vec![]);
    ///
    ///   ret.extend_from_slice(merged_bytes);
    ///
    ///   Some(ret)
    /// }
    ///
    /// let config = sled::ConfigBuilder::new()
    ///   .temporary(true)
    ///   .merge_operator(concatenate_merge)
    ///   .build();
    ///
    /// let tree = sled::Tree::start(config).unwrap();
    ///
    /// let k = b"k1";
    ///
    /// tree.set(k, vec![0]);
    /// tree.merge(k, vec![1]);
    /// tree.merge(k, vec![2]);
    /// // assert_eq!(tree.get(k).unwrap().unwrap(), vec![0, 1, 2]);
    ///
    /// // sets replace previously merged data,
    /// // bypassing the merge function.
    /// tree.set(k, vec![3]);
    /// // assert_eq!(tree.get(k), Ok(Some(vec![3])));
    ///
    /// // merges on non-present values will add them
    /// tree.del(k);
    /// tree.merge(k, vec![4]);
    /// // assert_eq!(tree.get(k).unwrap().unwrap(), vec![4]);
    /// ```
    pub fn merge<K: AsRef<[u8]>>(
        &self,
        key: K,
        value: Value,
    ) -> Result<(), ()> {
        let _measure = Measure::new(&M.tree_merge);

        if self.config.read_only {
            return Err(Error::Unsupported(
                "the database is in read-only mode".to_owned(),
            ));
        }

        let guard = pin();

        loop {
            let mut path = self.path_for_key(key.as_ref(), &guard)?;
            let (leaf_frag, leaf_ptr) = path.pop().expect(
                "path_for_key should always return a path \
                 of length >= 2 (root + leaf)",
            );
            let node: &Node = leaf_frag.unwrap_base();

            let encoded_key =
                prefix_encode(node.lo.inner(), key.as_ref());
            let frag = Frag::Merge(encoded_key, value.clone());

            let link = self.pages.link(
                node.id,
                leaf_ptr.clone(),
                frag.clone(),
                &guard,
            );
            match link {
                Ok(new_cas_key) => {
                    // success
                    if node.should_split(
                        self.config.blink_node_split_size,
                    ) {
                        let mut path2 = path
                            .iter()
                            .map(|&(f, ref p)| (Cow::Borrowed(f), p.clone()))
                            .collect::<Vec<(Cow<'_, Frag>, _)>>();
                        let mut node2 = node.clone();
                        node2
                            .apply(&frag, self.config.merge_operator);
                        let frag2 = Cow::Owned(Frag::Base(node2, None));
                        path2.push((frag2, new_cas_key));
                        self.recursive_split(path2, &guard)?;
                    }
                    guard.flush();
                    return Ok(());
                }
                Err(Error::CasFailed(_)) => {}
                Err(other) => {
                    guard.flush();
                    return Err(other.danger_cast());
                }
            }
            M.tree_looped();
        }
    }
<commit_msg>Fix bad conflict resolution<commit_after>use std::{
    borrow::Cow,
    cmp::Ordering::{Greater, Less},
    fmt::{self, Debug},
    sync::{
        atomic::{AtomicUsize, Ordering::SeqCst},
        Arc,
    },
};

use pagecache::PagePtr;

use super::*;

type Path<'g> = Vec<(&'g Frag, TreePtr<'g>)>;

impl<'a> IntoIterator for &'a Tree {
    type Item = Result<(Vec<u8>, PinnedValue), ()>;
    type IntoIter = Iter<'a>;

    fn into_iter(self) -> Iter<'a> {
        self.iter()
    }
}

/// A flash-sympathetic persistent lock-free B+ tree
#[derive(Clone)]
pub struct Tree {
    pub(crate) pages: Arc<
        PageCache<BLinkMaterializer, Frag, Vec<(PageId, PageId)>>,
    >,
    config: Config,
    root: Arc<AtomicUsize>,
}

unsafe impl Send for Tree {}
unsafe impl Sync for Tree {}

#[cfg(feature = "event_log")]
impl Drop for Tree {
    fn drop(&mut self) {
        self.config
            .event_log
            .tree_root_before_restart(self.root.load(SeqCst));
    }
}

impl Tree {
    /// Load existing or create a new `Tree` with a default configuration.
    pub fn start_default<P: AsRef<std::path::Path>>(
        path: P,
    ) -> Result<Tree, ()> {
        let config = ConfigBuilder::new().path(path).build();
        Self::start(config)
    }

    /// Load existing or create a new `Tree`.
    pub fn start(config: Config) -> Result<Tree, ()> {
        let _measure = Measure::new(&M.tree_start);

        #[cfg(any(test, feature = "check_snapshot_integrity"))]
        match config
            .verify_snapshot::<BLinkMaterializer, Frag, Vec<(PageId, PageId)>>(
            ) {
            Ok(_) => {}
            #[cfg(feature = "failpoints")]
            Err(Error::FailPoint) => {}
            other => panic!("failed to verify snapshot: {:?}", other),
        }

        let pages = PageCache::start(config.clone())?;

        let roots_opt = pages.recovered_state().clone().and_then(
            |mut roots: Vec<(PageId, PageId)>| {
                if roots.is_empty() {
                    None
                } else {
                    let mut last = std::usize::MAX;
                    let mut last_idx = std::usize::MAX;
                    while !roots.is_empty() {
                        // find the root that links to the last one
                        for (i, &(root, prev_root)) in
                            roots.iter().enumerate()
                        {
                            if prev_root == last {
                                last = root;
                                last_idx = i;
                                break;
                            }
                            assert_ne!(
                                i + 1,
                                roots.len(),
                                "encountered gap in root chain"
                            );
                        }
                        roots.remove(last_idx);
                    }
                    assert_ne!(last, std::usize::MAX);
                    Some(last)
                }
            },
        );

        let root_id = if let Some(root_id) = roots_opt {
            debug!("recovered root {} while starting tree", root_id);
            root_id
        } else {
            let guard = pin();

            let root_id = pages.allocate(&guard)?;
            assert_eq!(
                root_id,
                0,
                "we expect that this is the first page ever allocated"
            );
            debug!("allocated pid {} for root of new tree", root_id);

            let leaf_id = pages.allocate(&guard)?;
            trace!("allocated pid {} for leaf in new", leaf_id);

            let leaf = Frag::Base(
                Node {
                    id: leaf_id,
                    data: Data::Leaf(vec![]),
                    next: None,
                    lo: Bound::Inclusive(vec![]),
                    hi: Bound::Inf,
                },
                None,
            );

            // vec![0] represents a prefix-encoded empty prefix
            let root_index_vec = vec![(vec![0], leaf_id)];

            let root = Frag::Base(
                Node {
                    id: root_id,
                    data: Data::Index(root_index_vec),
                    next: None,
                    lo: Bound::Inclusive(vec![]),
                    hi: Bound::Inf,
                },
                Some(std::usize::MAX),
            );

            pages
                .replace(root_id, PagePtr::allocated(), root, &guard)
                .map_err(|e| e.danger_cast())?;
            pages
                .replace(leaf_id, PagePtr::allocated(), leaf, &guard)
                .map_err(|e| e.danger_cast())?;

            guard.flush();

            root_id
        };

        #[cfg(feature = "event_log")]
        config.event_log.tree_root_after_restart(root_id);

        Ok(Tree {
            pages: Arc::new(pages),
            config,
            root: Arc::new(AtomicUsize::new(root_id)),
        })
    }

    /// Flushes any pending IO buffers to disk to ensure durability.
    pub fn flush(&self) -> Result<(), ()> {
        self.pages.flush()
    }

    /// Retrieve a value from the `Tree` if it exists.
    pub fn get<K: AsRef<[u8]>>(
        &self,
        key: K,
    ) -> Result<Option<PinnedValue>, ()> {
        let _measure = Measure::new(&M.tree_get);

        let guard = pin();

        // the double guard is a hack that maintains
        // correctness of the ret value
        let double_guard = guard.clone();
        let (_, ret) = self.get_internal(key.as_ref(), &guard)?;

        guard.flush();

        Ok(ret.map(|r| PinnedValue::new(r, double_guard)))
    }

    /// Retrieve a value from the `Tree` for the key before the
    /// given key, if one exists.
    ///
    /// # Examples
    ///
    /// ```
    /// use sled::{ConfigBuilder, Error};
    /// let config = ConfigBuilder::new().temporary(true).build();
    ///
    /// let tree = sled::Tree::start(config).unwrap();
    ///
    /// for i in 0..10 {
    ///     tree.set(vec![i], vec![i]).expect("should write successfully");
    /// }
    ///
    /// assert!(tree.get_lt(vec![]).unwrap().is_none());
    /// assert!(tree.get_lt(vec![0]).unwrap().is_none());
    /// assert!(tree.get_lt(vec![1]).unwrap().unwrap().1 == vec![0]);
    /// assert!(tree.get_lt(vec![9]).unwrap().unwrap().1 == vec![8]);
    /// assert!(tree.get_lt(vec![10]).unwrap().unwrap().1 == vec![9]);
    /// assert!(tree.get_lt(vec![255]).unwrap().unwrap().1 == vec![9]);
    /// ```
    pub fn get_lt<K: AsRef<[u8]>>(
        &self,
        key: K,
    ) -> Result<Option<(Key, PinnedValue)>, ()> {
        let _measure = Measure::new(&M.tree_get);

        // the double guard is a hack that maintains
        // correctness of the ret value
        let guard = pin();
        let double_guard = guard.clone();

        let path = self.path_for_key(key.as_ref(), &guard)?;
        let (last_frag, _tree_ptr) = path
            .last()
            .expect("path should always contain a last element");

        let last_node = last_frag.unwrap_base();
        let data = &last_node.data;
        let items =
            data.leaf_ref().expect("last_node should be a leaf");
        let search = leaf_search(Less, items, |&(ref k, ref _v)| {
            prefix_cmp_encoded(k, key.as_ref(), last_node.lo.inner())
        });

        let ret = if search.is_none() {
            let mut iter = Iter {
                id: last_node.id,
                inner: &self,
                last_key: key.as_ref().to_vec(),
                inclusive: false,
                broken: None,
                done: false,
                guard: guard.clone(),
            };

            match iter.next_back() {
                Some(Err(e)) => return Err(e),
                Some(Ok(pair)) => Some(pair),
                None => None,
            }
        } else {
            let idx = search.unwrap();
            let (encoded_key, v) = &items[idx];
            Some((
                prefix_decode(last_node.lo.inner(), &*encoded_key),
                PinnedValue::new(&*v, double_guard),
            ))
        };

        guard.flush();

        Ok(ret)
    }

    /// Retrieve a value from the `Tree` for the key after the
    /// given key, if one exists.
    ///
    /// # Examples
    ///
    /// ```
    /// use sled::{ConfigBuilder, Error};
    /// let config = ConfigBuilder::new().temporary(true).build();
    ///
    /// let tree = sled::Tree::start(config).unwrap();
    ///
    /// for i in 0..10 {
    ///     tree.set(vec![i], vec![i]).expect("should write successfully");
    /// }
    ///
    /// assert!(tree.get_gt(vec![]).unwrap().unwrap().1 == vec![0]);
    /// assert!(tree.get_gt(vec![0]).unwrap().unwrap().1 == vec![1]);
    /// assert!(tree.get_gt(vec![1]).unwrap().unwrap().1 == vec![2]);
    /// assert!(tree.get_gt(vec![8]).unwrap().unwrap().1 == vec![9]);
    /// assert!(tree.get_gt(vec![9]).unwrap().is_none());
    /// ```
    pub fn get_gt<K: AsRef<[u8]>>(
        &self,
        key: K,
    ) -> Result<Option<(Key, PinnedValue)>, ()> {
        let _measure = Measure::new(&M.tree_get);

        // the double guard is a hack that maintains
        // correctness of the ret value
        let guard = pin();
        let double_guard = guard.clone();

        let path = self.path_for_key(key.as_ref(), &guard)?;
        let (last_frag, _tree_ptr) = path
            .last()
            .expect("path should always contain a last element");

        let last_node = last_frag.unwrap_base();
        let data = &last_node.data;
        let items =
            data.leaf_ref().expect("last_node should be a leaf");
        let search =
            leaf_search(Greater, items, |&(ref k, ref _v)| {
                prefix_cmp_encoded(
                    k,
                    key.as_ref(),
                    last_node.lo.inner(),
                )
            });

        let ret = if search.is_none() {
            let mut iter = Iter {
                id: last_node.id,
                inner: &self,
                last_key: key.as_ref().to_vec(),
                inclusive: false,
                broken: None,
                done: false,
                guard: guard.clone(),
            };

            match iter.next() {
                Some(Err(e)) => return Err(e),
                Some(Ok(pair)) => Some(pair),
                None => None,
            }
        } else {
            let idx = search.unwrap();
            let (encoded_key, v) = &items[idx];
            Some((
                prefix_decode(last_node.lo.inner(), &*encoded_key),
                PinnedValue::new(&*v, double_guard),
            ))
        };

        guard.flush();

        Ok(ret)
    }

    /// Compare and swap. Capable of unique creation, conditional modification,
    /// or deletion. If old is None, this will only set the value if it doesn't
    /// exist yet. If new is None, will delete the value if old is correct.
    /// If both old and new are Some, will modify the value if old is correct.
    /// If Tree is read-only, will do nothing.
    ///
    /// # Examples
    ///
    /// ```
    /// use sled::{ConfigBuilder, Error};
    /// let config = ConfigBuilder::new().temporary(true).build();
    /// let t = sled::Tree::start(config).unwrap();
    ///
    /// // unique creation
    /// assert_eq!(t.cas(&[1], None, Some(vec![1])), Ok(()));
    /// // assert_eq!(t.cas(&[1], None, Some(vec![1])), Err(Error::CasFailed(Some(vec![1]))));
    ///
    /// // conditional modification
    /// assert_eq!(t.cas(&[1], Some(&*vec![1]), Some(vec![2])), Ok(()));
    /// // assert_eq!(t.cas(&[1], Some(vec![1]), Some(vec![2])), Err(Error::CasFailed(Some(vec![2]))));
    ///
    /// // conditional deletion
    /// assert_eq!(t.cas(&[1], Some(&[2]), None), Ok(()));
    /// assert_eq!(t.get(&[1]), Ok(None));
    /// ```
    pub fn cas<K: AsRef<[u8]>>(
        &self,
        key: K,
        old: Option<&[u8]>,
        new: Option<Value>,
    ) -> Result<(), Option<PinnedValue>> {
        let _measure = Measure::new(&M.tree_cas);

        if self.config.read_only {
            return Err(Error::CasFailed(None));
        }

        let guard = pin();

        // we need to retry caps until old != cur, since just because
        // cap fails it doesn't mean our value was changed.
        loop {
            let pin_guard = guard.clone();
            let (mut path, cur) = self
                .get_internal(key.as_ref(), &guard)
                .map_err(|e| e.danger_cast())?;

            if old.as_ref().map(|o| o.as_ref()) != cur.map(|v| &*v) {
                return Err(Error::CasFailed(
                    cur.map(|c| PinnedValue::new(c, pin_guard)),
                ));
            }

            let (leaf_frag, leaf_ptr) = path.pop().expect(
                "get_internal somehow returned a path of length zero",
            );

            let (node_id, encoded_key) = {
                let node: &Node = leaf_frag.unwrap_base();
                (
                    node.id,
                    prefix_encode(node.lo.inner(), key.as_ref()),
                )
            };
            let frag = if let Some(ref n) = new {
                Frag::Set(encoded_key, n.clone())
            } else {
                Frag::Del(encoded_key)
            };
            let link = self.pages.link(
                node_id,
                leaf_ptr,
                frag.clone(),
                &guard,
            );
            match link {
                Ok(_) => {
                    guard.flush();
                    return Ok(());
                }
                Err(Error::CasFailed(_)) => {}
                Err(other) => {
                    guard.flush();
                    return Err(other.danger_cast());
                }
            }
            M.tree_looped();
        }
    }

    /// Set a key to a new value, returning the old value if it
    /// was set.
    pub fn set<K: AsRef<[u8]>>(
        &self,
        key: K,
        value: Value,
    ) -> Result<Option<PinnedValue>, ()> {
        let _measure = Measure::new(&M.tree_set);

        if self.config.read_only {
            return Err(Error::Unsupported(
                "the database is in read-only mode".to_owned(),
            ));
        }

        let guard = pin();

        loop {
            let double_guard = guard.clone();
            let (mut path, existing_key) =
                self.get_internal(key.as_ref(), &guard)?;
            let (leaf_frag, leaf_ptr) = path.pop().expect(
                "path_for_key should always return a path \
                 of length >= 2 (root + leaf)",
            );
            let node: &Node = leaf_frag.unwrap_base();
            let encoded_key =
                prefix_encode(node.lo.inner(), key.as_ref());

            let frag = Frag::Set(encoded_key, value.clone());
            let link = self.pages.link(
                node.id,
                leaf_ptr.clone(),
                frag.clone(),
                &guard,
            );
            match link {
                Ok(new_cas_key) => {
                    // success
                    if node.should_split(
                        self.config.blink_node_split_size,
                    ) {
                        let mut path2 = path
                            .iter()
                            .map(|&(f, ref p)| {
                                (Cow::Borrowed(f), p.clone())
                            }).collect::<Vec<(Cow<'_, Frag>, _)>>();
                        let mut node2 = node.clone();
                        node2
                            .apply(&frag, self.config.merge_operator);
                        let frag2 =
                            Cow::Owned(Frag::Base(node2, None));
                        path2.push((frag2, new_cas_key));
                        self.recursive_split(path2, &guard)?;
                    }

                    guard.flush();

                    return Ok(existing_key.map(move |r| {
                        PinnedValue::new(r, double_guard)
                    }));
                }
                Err(Error::CasFailed(_)) => {}
                Err(other) => {
                    guard.flush();

                    return Err(other.danger_cast());
                }
            }
            M.tree_looped();
        }
    }

    /// Merge a new value into the total state for a key.
    ///
    /// # Examples
    ///
    /// ```
    /// fn concatenate_merge(
    ///   _key: &[u8],               // the key being merged
    ///   old_value: Option<&[u8]>,  // the previous value, if one existed
    ///   merged_bytes: &[u8]        // the new bytes being merged in
    /// ) -> Option<Vec<u8>> {       // set the new value, return None to delete
    ///   let mut ret = old_value
    ///     .map(|ov| ov.to_vec())
    ///     .unwrap_or_else(|| vec![]);
    ///
    ///   ret.extend_from_slice(merged_bytes);
    ///
    ///   Some(ret)
    /// }
    ///
    /// let config = sled::ConfigBuilder::new()
    ///   .temporary(true)
    ///   .merge_operator(concatenate_merge)
    ///   .build();
    ///
    /// let tree = sled::Tree::start(config).unwrap();
    ///
    /// let k = b"k1";
    ///
    /// tree.set(k, vec![0]);
    /// tree.merge(k, vec![1]);
    /// tree.merge(k, vec![2]);
    /// // assert_eq!(tree.get(k).unwrap().unwrap(), vec![0, 1, 2]);
    ///
    /// // sets replace previously merged data,
    /// // bypassing the merge function.
    /// tree.set(k, vec![3]);
    /// // assert_eq!(tree.get(k), Ok(Some(vec![3])));
    ///
    /// // merges on non-present values will add them
    /// tree.del(k);
    /// tree.merge(k, vec![4]);
    /// // assert_eq!(tree.get(k).unwrap().unwrap(), vec![4]);
    /// ```
    pub fn merge<K: AsRef<[u8]>>(
        &self,
        key: K,
        value: Value,
    ) -> Result<(), ()> {
        let _measure = Measure::new(&M.tree_merge);

        if self.config.read_only {
            return Err(Error::Unsupported(
                "the database is in read-only mode".to_owned(),
            ));
        }

        let guard = pin();

        loop {
            let mut path = self.path_for_key(key.as_ref(), &guard)?;
            let (leaf_frag, leaf_ptr) = path.pop().expect(
                "path_for_key should always return a path \
                 of length >= 2 (root + leaf)",
            );
            let node: &Node = leaf_frag.unwrap_base();

            let encoded_key =
                prefix_encode(node.lo.inner(), key.as_ref());
            let frag = Frag::Merge(encoded_key, value.clone());

            let link = self.pages.link(
                node.id,
                leaf_ptr.clone(),
                frag.clone(),
                &guard,
            );
            match link {
                Ok(new_cas_key) => {
                    // success
                    if node.should_split(
                        self.config.blink_node_split_size,
                    ) {
                        let mut path2 = path
                            .iter()
                            .map(|&(f, ref p)| {
                                (Cow::Borrowed(f), p.clone())
                            }).collect::<Vec<(Cow<'_, Frag>, _)>>();
                        let mut node2 = node.clone();
                        node2
                            .apply(&frag, self.config.merge_operator);
                        let frag2 =
                            Cow::Owned(Frag::Base(node2, None));
                        path2.push((frag2, new_cas_key));
                        self.recursive_split(path2, &guard)?;
                    }
                    guard.flush();
                    return Ok(());
                }
                Err(Error::CasFailed(_)) => {}
                Err(other) => {
                    guard.flush();
                    return Err(other.danger_cast());
                }
            }
            M.tree_looped();
        }
    }

{'fraction_non_alphanumeric': 0.15609951845906903, 'fraction_numerical': 0.0047150882825040125, 'mean_word_length': 1.3165814547989774, 'pattern_counts': {'":': 0, '<': 111, '<?xml version=': 0, '>': 173, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1514844', 'n_tokens_mistral': 11776, 'n_tokens_neox': 11399, 'n_words': 2513}