Upgrade to slatedb 0.14 (opendata-oss#500)

Upgrades to slatedb 0.14. LogDb makes use of new subrange argument in `scan_prefix` to specify the sequence range of a query.

Author: hachikuji

Cargo.lock

@@ -49,8 +49,8 @@ serde_json = "1.0"
 serde_with = { version = "3", features = ["base64"] }
 serde_yaml = "0.9"
 foyer = "0.22"
-slatedb = { version = "0.13.0", features = ["compaction_filters"] }
-slatedb-common = "0.13.0"
+slatedb = { version = "0.14.0", features = ["compaction_filters"] }
+slatedb-common = "0.14.0"
 thiserror = "2.0"
 tokio = { version = "1.0", features = ["full", "test-util"] }
 tokio-util = "0.7"

Cargo.toml

@@ -2,8 +2,8 @@ use std::sync::Arc;
 use std::time::{Instant, SystemTime, UNIX_EPOCH};
 
 use bytes::Bytes;
-use slatedb::object_store::ObjectStore;
 use slatedb::object_store::path::Path;
+use slatedb::object_store::{ObjectStore, ObjectStoreExt};
 use tokio_util::sync::CancellationToken;
 
 use crate::config::ConsumerConfig;

buffer/src/consumer.rs

@@ -93,7 +93,7 @@ impl GarbageCollector {
         let mut failed: u64 = 0;
         if !to_delete.is_empty() {
             tracing::debug!(count = to_delete.len(), "GC deleting orphaned batch files");
-            let locations = stream::iter(to_delete.iter().cloned().map(Ok));
+            let locations = stream::iter(to_delete.into_iter().map(Ok));
             let mut results = self.object_store.delete_stream(locations.boxed());
             while let Some(result) = results.next().await {
                 match result {
@@ -169,8 +169,8 @@ mod tests {
     use crate::queue::QueueProducer;
     use bytes::Bytes;
     use common::ObjectStoreConfig;
-    use slatedb::object_store::PutPayload;
     use slatedb::object_store::memory::InMemory;
+    use slatedb::object_store::{ObjectStoreExt, PutPayload};
     use std::time::Duration;
 
     const TEST_MANIFEST_PATH: &str = "test/manifest";

buffer/src/gc.rs

@@ -4,7 +4,7 @@ use std::time::{Duration, Instant, SystemTime};
 use bytes::Bytes;
 use common::clock::Clock;
 use slatedb::object_store::path::Path;
-use slatedb::object_store::{ObjectStore, PutPayload};
+use slatedb::object_store::{ObjectStore, ObjectStoreExt, PutPayload};
 use tokio::sync::mpsc;
 use tokio_util::sync::CancellationToken;
 
@@ -421,8 +421,8 @@ mod tests {
     use bytes::Bytes;
     use common::ObjectStoreConfig;
     use common::clock::{MockClock, SystemClock};
-    use slatedb::object_store::ObjectStore;
     use slatedb::object_store::memory::InMemory;
+    use slatedb::object_store::{ObjectStore, ObjectStoreExt};
     use std::time::UNIX_EPOCH;
 
     async fn read_manifest_entries(store: &Arc<dyn ObjectStore>, path: &str) -> Vec<QueueEntry> {

buffer/src/producer.rs

@@ -4,7 +4,7 @@ use std::sync::atomic::{AtomicU64, Ordering};
 use bytes::{BufMut, Bytes, BytesMut};
 use slatedb::object_store::path::Path;
 use slatedb::object_store::{
-    Error as ObjectStoreError, ObjectStore, PutMode, PutPayload, UpdateVersion,
+    Error as ObjectStoreError, ObjectStore, ObjectStoreExt, PutMode, PutPayload, UpdateVersion,
 };
 
 use crate::error::{Error, Result};

buffer/src/queue.rs

@@ -95,6 +95,43 @@ impl BytesRange {
             end: Unbounded,
         }
     }
+
+    /// Builds the absolute key range for a prefix scan narrowed to `subrange`.
+    ///
+    /// `subrange` bounds are key *suffixes* interpreted relative to `prefix`: a
+    /// bound `s` selects the full key `prefix ++ s`. An unbounded subrange end
+    /// falls back to the prefix's natural upper bound (its lexicographic
+    /// successor), so `from_prefix_and_subrange(p, &BytesRange::unbounded())`
+    /// equals [`Self::prefix`].
+    ///
+    /// This mirrors SlateDB's own `scan_prefix` subrange semantics so the
+    /// in-memory fallback in [`StorageRead::scan_prefix_iter`] agrees with the
+    /// SlateDB path byte-for-byte.
+    ///
+    /// [`StorageRead::scan_prefix_iter`]: crate::storage::StorageRead::scan_prefix_iter
+    pub fn from_prefix_and_subrange(prefix: &[u8], subrange: &BytesRange) -> Self {
+        let concat = |suffix: &Bytes| -> Bytes {
+            let mut key = BytesMut::with_capacity(prefix.len() + suffix.len());
+            key.extend_from_slice(prefix);
+            key.extend_from_slice(suffix);
+            key.freeze()
+        };
+        let start = match &subrange.start {
+            Included(s) => Included(concat(s)),
+            Excluded(s) => Excluded(concat(s)),
+            Unbounded if prefix.is_empty() => Unbounded,
+            Unbounded => Included(Bytes::copy_from_slice(prefix)),
+        };
+        let end = match &subrange.end {
+            Included(e) => Included(concat(e)),
+            Excluded(e) => Excluded(concat(e)),
+            Unbounded => match lex_increment(prefix) {
+                Some(successor) => Excluded(successor),
+                None => Unbounded,
+            },
+        };
+        Self { start, end }
+    }
 }
 
 impl RangeBounds<Bytes> for BytesRange {
@@ -106,6 +143,17 @@ impl RangeBounds<Bytes> for BytesRange {
     }
 }
 
+/// SlateDB 0.14 scans are bounded by its own [`slatedb::ByteRangeBounds`] trait
+/// rather than `std::ops::RangeBounds`, so we map our bounds onto byte slices.
+impl slatedb::ByteRangeBounds for BytesRange {
+    fn start_bound(&self) -> Bound<&[u8]> {
+        self.start.as_ref().map(Bytes::as_ref)
+    }
+    fn end_bound(&self) -> Bound<&[u8]> {
+        self.end.as_ref().map(Bytes::as_ref)
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use proptest::prelude::*;

common/src/bytes.rs

@@ -17,7 +17,7 @@ pub use slatedb::db_cache::foyer_hybrid::FoyerHybridCache;
 pub use slatedb::db_cache::{CachedEntry, CachedKey, SplitCache};
 use slatedb::object_store::{self, ObjectStore};
 pub use slatedb::{CompactorBuilder, DbBuilder};
-use slatedb::{DbReader, FilterPolicy, SstReader};
+use slatedb::{DbReader, FilterPolicy, PrefixExtractor, SstReader};
 use tracing::info;
 use uuid::Uuid;
 
@@ -161,6 +161,9 @@ impl StorageBuilder {
                 if let Some(policies) = self.semantics.filter_policies {
                     db_builder = db_builder.with_filter_policies(policies);
                 }
+                if let Some(extractor) = self.semantics.segment_extractor {
+                    db_builder = db_builder.with_segment_extractor(extractor);
+                }
                 let db = db_builder.build().await.map_err(|e| {
                     StorageError::Storage(format!("Failed to create SlateDB: {}", e))
                 })?;
@@ -250,6 +253,7 @@ impl StorageReaderRuntime {
 pub struct StorageSemantics {
     pub(crate) merge_operator: Option<Arc<dyn MergeOperator>>,
     pub(crate) filter_policies: Option<Vec<Arc<dyn FilterPolicy>>>,
+    pub(crate) segment_extractor: Option<Arc<dyn PrefixExtractor>>,
 }
 
 impl StorageSemantics {
@@ -275,6 +279,17 @@ impl StorageSemantics {
         self.filter_policies = Some(policies);
         self
     }
+
+    /// Sets the segment extractor (SlateDB RFC-0024) for writers and readers.
+    ///
+    /// SlateDB 0.14 persists the extractor's name in the manifest and refuses
+    /// to open a writer or `DbReader` whose configured extractor doesn't match.
+    /// Routing this through semantics keeps the writer, compactor, and
+    /// standalone reader paths aligned on a single extractor.
+    pub fn with_segment_extractor(mut self, extractor: Arc<dyn PrefixExtractor>) -> Self {
+        self.segment_extractor = Some(extractor);
+        self
+    }
 }
 
 pub fn new_slatedb_compactor_builder(
@@ -403,6 +418,9 @@ pub async fn create_storage_read(
             if let Some(policies) = semantics.filter_policies {
                 builder = builder.with_filter_policies(policies);
             }
+            if let Some(extractor) = semantics.segment_extractor {
+                builder = builder.with_segment_extractor(extractor);
+            }
             if let Some(cache) = cache.clone() {
                 builder = builder.with_db_cache(cache);
             }

common/src/storage/factory.rs

@@ -243,7 +243,14 @@ pub trait StorageRead: Any + Send + Sync {
         range: BytesRange,
     ) -> StorageResult<Box<dyn StorageIterator + Send + 'static>>;
 
-    /// Returns an iterator over records whose key starts with `prefix`.
+    /// Returns an iterator over records whose key starts with `prefix`,
+    /// restricted to `subrange`.
+    ///
+    /// `subrange` bounds are key *suffixes* interpreted relative to `prefix`: a
+    /// bound `s` selects the full key `prefix ++ s`. Pass
+    /// [`BytesRange::unbounded`] to scan the prefix's entire keyspace. Callers
+    /// with a known sub-key ordering (e.g. the log's trailing sequence bytes)
+    /// use this to bound the scan natively instead of filtering client-side.
     ///
     /// Backends that support prefix-aware SST filters (e.g. SlateDB with a
     /// configured `PrefixExtractor` or custom `FilterPolicy`) consult those
@@ -258,14 +265,17 @@ pub trait StorageRead: Any + Send + Sync {
     /// [`slatedb::FilterContext`].
     ///
     /// The default implementation delegates to [`Self::scan_iter`] with the
-    /// range derived from the prefix, which preserves correctness for any
-    /// backend; backends that can do better should override this.
+    /// absolute range derived from `prefix` and `subrange`, which preserves
+    /// correctness for any backend; backends that can do better should override
+    /// this.
     async fn scan_prefix_iter(
         &self,
         prefix: Bytes,
+        subrange: BytesRange,
         _filter_context: Option<FilterContext>,
     ) -> StorageResult<Box<dyn StorageIterator + Send + 'static>> {
-        self.scan_iter(BytesRange::prefix(prefix)).await
+        self.scan_iter(BytesRange::from_prefix_and_subrange(&prefix, &subrange))
+            .await
     }
 
     /// Collects all records in the range into a Vec.

common/src/storage/mod.rs

@@ -242,12 +242,13 @@ impl StorageRead for SlateDbStorage {
     async fn scan_prefix_iter(
         &self,
         prefix: Bytes,
+        subrange: BytesRange,
         filter_context: Option<FilterContext>,
     ) -> StorageResult<Box<dyn StorageIterator + Send + 'static>> {
         let options = default_scan_options().with_filter_context(filter_context);
         let iter = self
             .db
-            .scan_prefix_with_options(prefix, &options)
+            .scan_prefix_with_options(prefix, subrange, &options)
             .await
             .map_err(StorageError::from_storage)?;
         Ok(Box::new(SlateDbIterator { iter }))
@@ -326,12 +327,13 @@ impl StorageRead for SlateDbStorageSnapshot {
     async fn scan_prefix_iter(
         &self,
         prefix: Bytes,
+        subrange: BytesRange,
         filter_context: Option<FilterContext>,
     ) -> StorageResult<Box<dyn StorageIterator + Send + 'static>> {
         let options = default_scan_options().with_filter_context(filter_context);
         let iter = self
             .snapshot
-            .scan_prefix_with_options(prefix, &options)
+            .scan_prefix_with_options(prefix, subrange, &options)
             .await
             .map_err(StorageError::from_storage)?;
         Ok(Box::new(SlateDbIterator { iter }))
@@ -549,12 +551,13 @@ impl StorageRead for SlateDbStorageReader {
     async fn scan_prefix_iter(
         &self,
         prefix: Bytes,
+        subrange: BytesRange,
         filter_context: Option<FilterContext>,
     ) -> StorageResult<Box<dyn StorageIterator + Send + 'static>> {
         let options = default_scan_options().with_filter_context(filter_context);
         let iter = self
             .reader
-            .scan_prefix_with_options(prefix, &options)
+            .scan_prefix_with_options(prefix, subrange, &options)
             .await
             .map_err(StorageError::from_storage)?;
         Ok(Box::new(SlateDbIterator { iter }))