Add descending order to scan (#401)

* Add desc shorting to scan

* Add tests to the in-memory and on disk options

* Fix errors after merge

* Add reverse order to TransactionScan

* Address PR comments

* Fix LevelStream reverse order

* Fix fmt

---------

Co-authored-by: Daniel Martinez Maqueda <daniel@super.ai>

Author: belveryin

README.md

@@ -144,6 +144,7 @@ async fn main() {
     - [x] (Optimistic) Transactions.
     - [x] Leveled compaction strategy.
     - [x] Push down filter, limit and projection.
+    - [x] Reverse scan (descending order) for efficient newest-first queries.
 - [x] Runtime schema definition.
 - [ ] SQL (via [Apache DataFusion](https://datafusion.apache.org/)).
 - [ ] Fusion storage across RAM, flash, SSD, and remote Object Storage Service (OSS) for each column-family, balancing performance and cost efficiency per data block:

examples/declare.rs

@@ -93,6 +93,26 @@ async fn main() {
             }
         }
 
+        {
+            let upper = "Blob".into();
+            // reverse scan of user (descending order)
+            let mut reverse_scan = txn
+                .scan((Bound::Included(&name), Bound::Excluded(&upper)))
+                .reverse() // scan in descending order
+                .projection(&["name", "grade"]) // tonbo supports combining reverse with projection
+                .limit(10) // and with limits
+                .take()
+                .await
+                .unwrap();
+
+            println!("Users in reverse order:");
+            while let Some(entry) = reverse_scan.next().await.transpose().unwrap() {
+                if let Some(user) = entry.value() {
+                    println!("- {}: {:?}", user.name, user.grade.unwrap_or(0.0.into()));
+                }
+            }
+        }
+
         // commit transaction
         txn.commit().await.unwrap();
     }

guide/src/lsm_structure.md

@@ -109,7 +109,12 @@ The query process of immutable memtable is different from memtable:
 
 ```rust
 let range = self.index.range::<TsRef<<<A::Record as Record>::Schema as Schema>::Key>, _>((lower, upper));
-ImmutableScan::<A::Record>::new(range, self.data.as_record_batch(), projection_mask)
+let boxed_range = if order == Some(Order::Desc) {
+    Box::new(range.rev())
+} else {
+    Box::new(range)
+};
+ImmutableScan::<A::Record>::new(boxed_range, self.data.as_record_batch(), projection_mask)
 ```
 
 For `get(key)`, it can be transformed into `scan(key, key)`. The final result returned by `scan` is then converted into `RecordBatchEntry` using Arrow's `.as_record_batch()` method.

guide/src/start.md

@@ -172,6 +172,20 @@ while let Some(entry) = scan.next().await.transpose().unwrap() {
     let data = entry.value(); // type of UserRef
     // ......
 }
+
+// reverse scan of user (descending order) 
+let mut reverse_scan = txn
+    .scan((Bound::Included(&name), Bound::Excluded(&upper)))
+    .reverse() // scan in descending order
+    .limit(10) // optionally limit results
+    .take()
+    .await
+    .unwrap();
+
+while let Some(entry) = reverse_scan.next().await.transpose().unwrap() {
+    let data = entry.value(); // records in reverse order
+    // ......
+}
 ```
 
 ### Persistence

guide/src/usage/tonbo.md

@@ -77,6 +77,18 @@ while let Some(entry) = scan.next().await.transpose().unwrap() {
     let data = entry.value(); // type of UserRef
     // ......
 }
+
+// Reverse scan (descending order)
+let mut reverse_scan = db
+    .scan((Bound::Included(&name), Bound::Excluded(&upper)))
+    .reverse() // scan in descending order
+    .limit(100) // limit results
+    .await
+    .unwrap();
+while let Some(entry) = reverse_scan.next().await.transpose().unwrap() {
+    let data = entry.value(); // newest records first
+    // ......
+}
 ```
 ### Insert/Remove
 

src/compaction/leveled.rs

@@ -285,6 +285,7 @@ where
                                     u32::MAX.into(),
                                     None,
                                     ProjectionMask::all(),
+                                    None,
                                 )
                                 .await?,
                         });
@@ -342,6 +343,7 @@ where
                                 u32::MAX.into(),
                                 None,
                                 ProjectionMask::all(),
+                                None, // Default order for compaction
                             )
                             .await?,
                     });
@@ -359,6 +361,7 @@ where
                     ProjectionMask::all(),
                     level_fs.clone(),
                     ctx.parquet_lru.clone(),
+                    None, // Default order for compaction
                 )
                 .ok_or(CompactionError::EmptyLevel)?;
 
@@ -384,6 +387,7 @@ where
                     ProjectionMask::all(),
                     level_l_fs.clone(),
                     ctx.parquet_lru.clone(),
+                    None, // Default order for compaction
                 )
                 .ok_or(CompactionError::EmptyLevel)?;
 

src/compaction/mod.rs

@@ -53,7 +53,7 @@ where
         schema: &R::Schema,
         fs: &Arc<dyn DynFs>,
     ) -> Result<(), CompactionError<R>> {
-        let mut stream = MergeStream::<R>::from_vec(streams, u32::MAX.into()).await?;
+        let mut stream = MergeStream::<R>::from_vec(streams, u32::MAX.into(), None).await?;
 
         // Kould: is the capacity parameter necessary?
         let mut builder =