perf: seqno-aware seek in data block point reads

src/table/data_block/iter.rs

@@ -8,7 +8,7 @@ use crate::{
         block::{Decoder, ParsedItem},
         data_block::DataBlockParsedItem,
     },
-    InternalValue,
+    InternalValue, SeqNo,
 };
 
 /// The data block iterator handles double-ended scans over a data block
@@ -34,6 +34,17 @@ impl<'a> Iter<'a> {
         true
     }
 
+    /// Seeks to the restart interval containing the target (needle, seqno) pair.
+    ///
+    /// Exploits internal key ordering (user_key ASC, seqno DESC) to skip
+    /// restart intervals containing only versions newer than the target seqno.
+    pub fn seek_to_key_seqno(&mut self, needle: &[u8], seqno: SeqNo) -> bool {
+        self.decoder.inner_mut().seek(
+            |head_key, head_seqno| head_key < needle || (head_key == needle && head_seqno >= seqno),
+            false,
+        )
+    }
+
     pub fn seek(&mut self, needle: &[u8]) -> bool {
         // Find the restart interval whose head key is the last one strictly below `needle`.
         // The decoder then performs a linear scan within that interval; we stop as soon as we

src/table/data_block/mod.rs

@@ -407,7 +407,6 @@ impl DataBlock {
             .map(|reader| reader.bucket_count())
     }
 
-    // TODO: handle seqno more nicely (make Key generic, so we can do binary search over (key, seqno))
     #[must_use]
     pub fn point_read(&self, needle: &[u8], seqno: SeqNo) -> Option<InternalValue> {
         let iter = if let Some(hash_index_reader) = self.get_hash_index_reader() {
@@ -416,10 +415,10 @@ impl DataBlock {
                     return None;
                 }
                 MARKER_CONFLICT => {
-                    // NOTE: Fallback to binary search
+                    // NOTE: Fallback to seqno-aware binary search
                     let mut iter = self.iter();
 
-                    if !iter.seek(needle) {
+                    if !iter.seek_to_key_seqno(needle, seqno) {
                         return None;
                     }
 
@@ -437,8 +436,9 @@ impl DataBlock {
         } else {
             let mut iter = self.iter();
 
-            // NOTE: Fallback to binary search
-            if !iter.seek(needle) {
+            // NOTE: Seqno-aware binary search skips restart intervals
+            // containing only versions newer than the target seqno
+            if !iter.seek_to_key_seqno(needle, seqno) {
                 return None;
             }
 
@@ -449,14 +449,14 @@ impl DataBlock {
         for item in iter {
             match item.compare_key(needle, &self.inner.data) {
                 std::cmp::Ordering::Greater => {
-                    // We are before our searched key/seqno
+                    // We are past our searched key
                     return None;
                 }
                 std::cmp::Ordering::Equal => {
                     // If key is same as needle, check sequence number
                 }
                 std::cmp::Ordering::Less => {
-                    // We are past our searched key
+                    // We are before our searched key
                     continue;
                 }
             }
@@ -1233,4 +1233,120 @@ mod tests {
 
         Ok(())
     }
+
+    #[test]
+    fn data_block_point_read_seqno_aware_seek() -> crate::Result<()> {
+        // Key "a" with seqno 5,4,3,2,1 — point_read("a", seqno=3) should return v3
+        let items = [
+            InternalValue::from_components(b"a", b"a5", 5, Value),
+            InternalValue::from_components(b"a", b"a4", 4, Value),
+            InternalValue::from_components(b"a", b"a3", 3, Value),
+            InternalValue::from_components(b"a", b"a2", 2, Value),
+            InternalValue::from_components(b"a", b"a1", 1, Value),
+        ];
+
+        // With restart_interval=1, every item is a restart head,
+        // so seqno-aware binary search can skip directly to the target version
+        for restart_interval in 1..=4 {
+            let bytes = DataBlock::encode_into_vec(&items, restart_interval, 0.0)?;
+
+            let data_block = DataBlock::new(Block {
+                data: bytes.into(),
+                header: Header {
+                    block_type: BlockType::Data,
+                    checksum: Checksum::from_raw(0),
+                    data_length: 0,
+                    uncompressed_length: 0,
+                },
+            });
+
+            // seqno=4 → should see version with seqno=3 (first with seqno < 4)
+            assert_eq!(
+                Some(items[2].clone()),
+                data_block.point_read(b"a", 4),
+                "restart_interval={restart_interval}: seqno=4 should return v3",
+            );
+
+            // seqno=3 → should see version with seqno=2
+            assert_eq!(
+                Some(items[3].clone()),
+                data_block.point_read(b"a", 3),
+                "restart_interval={restart_interval}: seqno=3 should return v2",
+            );
+
+            // seqno=6 → should see latest version (seqno=5)
+            assert_eq!(
+                Some(items[0].clone()),
+                data_block.point_read(b"a", 6),
+                "restart_interval={restart_interval}: seqno=6 should return v5",
+            );
+
+            // seqno=1 → no visible version (all seqno >= 1)
+            assert!(
+                data_block.point_read(b"a", 1).is_none(),
+                "restart_interval={restart_interval}: seqno=1 should return None",
+            );
+
+            // Non-existent key
+            assert!(
+                data_block.point_read(b"b", SeqNo::MAX).is_none(),
+                "restart_interval={restart_interval}: key 'b' should not exist",
+            );
+        }
+
+        Ok(())
+    }
+
+    #[test]
+    fn data_block_point_read_seqno_aware_seek_mixed_keys() -> crate::Result<()> {
+        // Multiple keys with multiple versions
+        let items = [
+            InternalValue::from_components(b"a", b"a3", 3, Value),
+            InternalValue::from_components(b"a", b"a2", 2, Value),
+            InternalValue::from_components(b"a", b"a1", 1, Value),
+            InternalValue::from_components(b"b", b"b5", 5, Value),
+            InternalValue::from_components(b"b", b"b4", 4, Value),
+            InternalValue::from_components(b"b", b"b3", 3, Value),
+            InternalValue::from_components(b"b", b"b2", 2, Value),
+            InternalValue::from_components(b"b", b"b1", 1, Value),
+            InternalValue::from_components(b"c", b"c1", 1, Value),
+        ];
+
+        for restart_interval in 1..=4 {
+            let bytes = DataBlock::encode_into_vec(&items, restart_interval, 0.0)?;
+
+            let data_block = DataBlock::new(Block {
+                data: bytes.into(),
+                header: Header {
+                    block_type: BlockType::Data,
+                    checksum: Checksum::from_raw(0),
+                    data_length: 0,
+                    uncompressed_length: 0,
+                },
+            });
+
+            // Read "b" at seqno=4 → should return version with seqno=3
+            assert_eq!(
+                Some(items[5].clone()),
+                data_block.point_read(b"b", 4),
+                "restart_interval={restart_interval}: b@4 should return b3",
+            );
+
+            // Read "a" at seqno=2 → should return version with seqno=1
+            assert_eq!(
+                Some(items[2].clone()),
+                data_block.point_read(b"a", 2),
+                "restart_interval={restart_interval}: a@2 should return a1",
+            );
+
+            // Read "c" at seqno=2 → should return version with seqno=1
+            assert_eq!(
+                Some(items[8].clone()),
+                data_block.point_read(b"c", 2),
+                "restart_interval={restart_interval}: c@2 should return c1",
+            );
+        }
+
+        Ok(())
+    }
 }