Add prefetch support for reverse scan

Author: jaykorean

file/file_prefetch_buffer.cc

@@ -259,11 +259,17 @@ void FilePrefetchBuffer::CopyDataToOverlapBuffer(BufferInfo* src,
 // Clear the buffers if it contains outdated data. Outdated data can be because
 // previous sequential reads were read from the cache instead of these buffer.
 // In that case outdated IOs should be aborted.
-void FilePrefetchBuffer::AbortOutdatedIO(uint64_t offset) {
+void FilePrefetchBuffer::AbortOutdatedIO(uint64_t offset, size_t length) {
   std::vector<void*> handles;
   std::vector<BufferInfo*> tmp_buf;
   for (auto& buf : bufs_) {
-    if (buf->IsBufferOutdatedWithAsyncProgress(offset)) {
+    bool outdated = false;
+    if (direction_ == PrefetchDirection::kForward) {
+      outdated = buf->IsBufferOutdatedWithAsyncProgress(offset);
+    } else {
+      outdated = buf->IsBufferOutdatedBackwardWithAsyncProgress(offset, length);
+    }
+    if (outdated) {
       handles.emplace_back(buf->io_handle_);
       tmp_buf.emplace_back(buf);
     }
@@ -315,8 +321,15 @@ void FilePrefetchBuffer::AbortAllIOs() {
 void FilePrefetchBuffer::ClearOutdatedData(uint64_t offset, size_t length) {
   while (!IsBufferQueueEmpty()) {
     BufferInfo* buf = GetFirstBuffer();
-    // Offset is greater than this buffer's end offset.
-    if (buf->IsBufferOutdated(offset)) {
+    // Offset is greater than this buffer's end offset for forward,
+    // or offset+length <= buf start for backward.
+    bool outdated = false;
+    if (direction_ == PrefetchDirection::kForward) {
+      outdated = buf->IsBufferOutdated(offset);
+    } else {
+      outdated = buf->IsBufferOutdatedBackward(offset, length);
+    }
+    if (outdated) {
       FreeFrontBuffer();
     } else {
       break;
@@ -339,11 +352,22 @@ void FilePrefetchBuffer::ClearOutdatedData(uint64_t offset, size_t length) {
 
   if (buf->DoesBufferContainData() && buf->IsOffsetInBuffer(offset)) {
     BufferInfo* next_buf = bufs_[1];
-    if (/* next buffer doesn't align with first buffer and requested data
-       overlaps with next buffer */
-        ((buf->offset_ + buf->CurrentSize() != next_buf->offset_) &&
-         (offset + length > buf->offset_ + buf->CurrentSize()))) {
-      abort_io = true;
+    if (direction_ == PrefetchDirection::kForward) {
+      if (/* next buffer doesn't align with first buffer and requested data
+         overlaps with next buffer */
+          ((buf->offset_ + buf->CurrentSize() != next_buf->offset_) &&
+           (offset + length > buf->offset_ + buf->CurrentSize()))) {
+        abort_io = true;
+      }
+    } else {
+      // Backward: next buffer should start where current ends in reverse
+      // direction? For backward, buffers are ordered with decreasing offsets in
+      // queue front to back? Simplistic check: if requested range extends
+      // before current buffer start, and next buffer doesn't align.
+      if (((next_buf->offset_ + next_buf->CurrentSize() != buf->offset_) &&
+           (offset < buf->offset_))) {
+        abort_io = true;
+      }
     }
   } else {
     // buffer with offset doesn't contain data or offset doesn't lie in this
@@ -636,7 +660,7 @@ Status FilePrefetchBuffer::PrefetchInternal(const IOOptions& opts,
 
   // Abort outdated IO.
   if (!explicit_prefetch_submitted_) {
-    AbortOutdatedIO(offset);
+    AbortOutdatedIO(offset, length);
     FreeEmptyBuffers();
   }
   ClearOutdatedData(offset, length);

file/file_prefetch_buffer.h

@@ -34,6 +34,18 @@ namespace ROCKSDB_NAMESPACE {
 struct IOOptions;
 class RandomAccessFileReader;
 
+enum class FilePrefetchBufferUsage {
+  kTableOpenPrefetchTail,
+  kUserScanPrefetch,
+  kCompactionPrefetch,
+  kUnknown,
+};
+
+enum class PrefetchDirection {
+  kForward,
+  kBackward,
+};
+
 struct ReadaheadParams {
   ReadaheadParams() {}
 
@@ -58,6 +70,10 @@ struct ReadaheadParams {
   // Number of buffers to maintain that contains prefetched data. If num_buffers
   // > 1 then buffers will be filled asynchronously whenever they get emptied.
   size_t num_buffers = 1;
+
+  // Direction of prefetch for sequential scans. Forward is default for backward
+  // compatibility. Backward is used for reverse iteration.
+  PrefetchDirection direction = PrefetchDirection::kForward;
 };
 
 struct BufferInfo {
@@ -118,11 +134,22 @@ struct BufferInfo {
             offset >= offset_ + buffer_.CurrentSize());
   }
 
+  bool IsBufferOutdatedBackward(uint64_t offset, size_t length) {
+    return (!async_read_in_progress_ && DoesBufferContainData() &&
+            offset + length <= offset_);
+  }
+
   bool IsBufferOutdatedWithAsyncProgress(uint64_t offset) {
     return (async_read_in_progress_ && io_handle_ != nullptr &&
             offset >= offset_ + async_req_len_);
   }
 
+  bool IsBufferOutdatedBackwardWithAsyncProgress(uint64_t offset,
+                                                 size_t length) {
+    return (async_read_in_progress_ && io_handle_ != nullptr &&
+            offset + length <= offset_);
+  }
+
   bool IsOffsetInBufferWithAsyncProgress(uint64_t offset) {
     return (async_read_in_progress_ && offset >= offset_ &&
             offset < offset_ + async_req_len_);
@@ -131,13 +158,6 @@ struct BufferInfo {
   size_t CurrentSize() { return buffer_.CurrentSize(); }
 };
 
-enum class FilePrefetchBufferUsage {
-  kTableOpenPrefetchTail,
-  kUserScanPrefetch,
-  kCompactionPrefetch,
-  kUnknown,
-};
-
 // Implementation:
 // FilePrefetchBuffer maintains a dequeu of free buffers (free_bufs_) with no
 // data and bufs_ which contains the prefetched data. Whenever a buffer is
@@ -208,7 +228,8 @@ class FilePrefetchBuffer {
         stats_(stats),
         usage_(usage),
         readaheadsize_cb_(cb),
-        num_buffers_(readahead_params.num_buffers) {
+        num_buffers_(readahead_params.num_buffers),
+        direction_(readahead_params.direction) {
     assert((num_file_reads_ >= num_file_reads_for_auto_readahead_ + 1) ||
            (num_file_reads_ == 0));
 
@@ -424,7 +445,7 @@ class FilePrefetchBuffer {
                             size_t roundup_len, bool refit_tail,
                             bool use_fs_buffer, uint64_t& aligned_useful_len);
 
-  void AbortOutdatedIO(uint64_t offset);
+  void AbortOutdatedIO(uint64_t offset, size_t length);
 
   void AbortAllIOs();
 
@@ -711,5 +732,8 @@ class FilePrefetchBuffer {
   // num_buffers_ is the number of buffers maintained by FilePrefetchBuffer to
   // prefetch the data at a time.
   size_t num_buffers_;
+
+  // Direction of prefetch for sequential scans.
+  PrefetchDirection direction_;
 };
 }  // namespace ROCKSDB_NAMESPACE

table/block_based/block_based_table_iterator.cc

@@ -269,7 +269,7 @@ void BlockBasedTableIterator::SeekForPrev(const Slice& target) {
     }
   }
 
-  InitDataBlock();
+  InitDataBlock(PrefetchDirection::kBackward);
 
   block_iter_.SeekForPrev(target);
 
@@ -297,7 +297,7 @@ void BlockBasedTableIterator::SeekToLast() {
     return;
   }
 
-  InitDataBlock();
+  InitDataBlock(PrefetchDirection::kBackward);
   block_iter_.SeekToLast();
   FindKeyBackward();
   CheckDataBlockWithinUpperBound();
@@ -362,7 +362,7 @@ void BlockBasedTableIterator::Prev() {
       return;
     }
 
-    InitDataBlock();
+    InitDataBlock(PrefetchDirection::kBackward);
     block_iter_.SeekToLast();
   } else {
     assert(block_iter_points_to_real_block_);
@@ -372,7 +372,7 @@ void BlockBasedTableIterator::Prev() {
   FindKeyBackward();
 }
 
-void BlockBasedTableIterator::InitDataBlock() {
+void BlockBasedTableIterator::InitDataBlock(PrefetchDirection direction) {
   // MultiScan path: load block from ReadSet
   if (multi_scan_read_set_) {
     BlockHandle data_block_handle = index_iter_->value().handle;
@@ -443,7 +443,7 @@ void BlockBasedTableIterator::InitDataBlock() {
 
       std::function<void(bool, uint64_t&, uint64_t&)> readaheadsize_cb =
           nullptr;
-      if (readahead_cache_lookup_) {
+      if (readahead_cache_lookup_ && direction == PrefetchDirection::kForward) {
         readaheadsize_cb = std::bind(
             &BlockBasedTableIterator::BlockCacheLookupForReadAheadSize, this,
             std::placeholders::_1, std::placeholders::_2,
@@ -460,7 +460,7 @@ void BlockBasedTableIterator::InitDataBlock() {
           rep, data_block_handle, read_options_.readahead_size,
           is_for_compaction,
           /*no_sequential_checking=*/false, read_options_, readaheadsize_cb,
-          read_options_.async_io);
+          read_options_.async_io, direction);
 
       Status s;
       table_->NewDataBlockIterator<DataBlockIter>(
@@ -484,7 +484,8 @@ void BlockBasedTableIterator::InitDataBlock() {
   }
 }
 
-void BlockBasedTableIterator::AsyncInitDataBlock(bool is_first_pass) {
+void BlockBasedTableIterator::AsyncInitDataBlock(bool is_first_pass,
+                                                 PrefetchDirection direction) {
   BlockHandle data_block_handle;
   bool is_for_compaction =
       lookup_context_.caller == TableReaderCaller::kCompaction;
@@ -501,7 +502,7 @@ void BlockBasedTableIterator::AsyncInitDataBlock(bool is_first_pass) {
 
       std::function<void(bool, uint64_t&, uint64_t&)> readaheadsize_cb =
           nullptr;
-      if (readahead_cache_lookup_) {
+      if (readahead_cache_lookup_ && direction == PrefetchDirection::kForward) {
         readaheadsize_cb = std::bind(
             &BlockBasedTableIterator::BlockCacheLookupForReadAheadSize, this,
             std::placeholders::_1, std::placeholders::_2,
@@ -520,7 +521,7 @@ void BlockBasedTableIterator::AsyncInitDataBlock(bool is_first_pass) {
       block_prefetcher_.PrefetchIfNeeded(
           rep, data_block_handle, read_options_.readahead_size,
           is_for_compaction, /*no_sequential_checking=*/read_options_.async_io,
-          read_options_, readaheadsize_cb, read_options_.async_io);
+          read_options_, readaheadsize_cb, read_options_.async_io, direction);
 
       Status s;
       table_->NewDataBlockIterator<DataBlockIter>(
@@ -740,7 +741,7 @@ void BlockBasedTableIterator::FindKeyBackward() {
     index_iter_->Prev();
 
     if (index_iter_->Valid()) {
-      InitDataBlock();
+      InitDataBlock(PrefetchDirection::kBackward);
       block_iter_.SeekToLast();
     } else {
       return;

table/block_based/block_based_table_iterator.h

@@ -417,8 +417,9 @@ class BlockBasedTableIterator : public InternalIteratorBase<Slice> {
   // If `target` is null, seek to first.
   void SeekImpl(const Slice* target, bool async_prefetch);
 
-  void InitDataBlock();
-  void AsyncInitDataBlock(bool is_first_pass);
+  void InitDataBlock(PrefetchDirection direction = PrefetchDirection::kForward);
+  void AsyncInitDataBlock(bool is_first_pass, PrefetchDirection direction =
+                                                  PrefetchDirection::kForward);
   bool MaterializeCurrentBlock();
   void FindKeyForward();
   void FindBlockForward();

table/block_based/block_prefetcher.cc

@@ -17,7 +17,7 @@ void BlockPrefetcher::PrefetchIfNeeded(
     const size_t readahead_size, bool is_for_compaction,
     const bool no_sequential_checking, const ReadOptions& read_options,
     const std::function<void(bool, uint64_t&, uint64_t&)>& readaheadsize_cb,
-    bool is_async_io_prefetch) {
+    bool is_async_io_prefetch, PrefetchDirection direction) {
   if (read_options.read_tier == ReadTier::kBlockCacheTier) {
     // Disable prefetching when IO disallowed. (Note that we haven't allocated
     // any buffers yet despite the various tracked settings.)
@@ -28,13 +28,15 @@ void BlockPrefetcher::PrefetchIfNeeded(
   readahead_params.initial_readahead_size = readahead_size;
   readahead_params.max_readahead_size = readahead_size;
   readahead_params.num_buffers = is_async_io_prefetch ? 2 : 1;
+  readahead_params.direction = direction;
 
   const size_t len = BlockBasedTable::BlockSizeWithTrailer(handle);
   const size_t offset = handle.offset();
   if (is_for_compaction) {
     if (!rep->file->use_direct_io() && compaction_readahead_size_ > 0) {
       // If FS supports prefetching (readahead_limit_ will be non zero in that
       // case) and current block exists in prefetch buffer then return.
+      // Compaction is always forward.
       if (offset + len <= readahead_limit_) {
         return;
       }
@@ -108,12 +110,29 @@ void BlockPrefetcher::PrefetchIfNeeded(
 
   // If FS supports prefetching (readahead_limit_ will be non zero in that case)
   // and current block exists in prefetch buffer then return.
-  if (offset + len <= readahead_limit_) {
+  bool in_prefetch_range = false;
+  if (direction == PrefetchDirection::kForward) {
+    in_prefetch_range = (offset + len <= readahead_limit_);
+  } else {
+    // For backward, prefetch_start_ tracks lower bound of prefetched range.
+    // If prefetch_start_ is 0 it means not set yet, treat as miss.
+    in_prefetch_range = (prefetch_start_ != 0 && offset >= prefetch_start_ &&
+                         offset + len <= readahead_limit_);
+    // Alternative simple check for backward: offset >= prefetch_start_
+    // We'll use offset >= prefetch_start_ as indicator current block is within
+    // prefetched window.
+    if (prefetch_start_ != 0) {
+      in_prefetch_range = (offset >= prefetch_start_);
+    } else {
+      in_prefetch_range = false;
+    }
+  }
+  if (in_prefetch_range) {
     UpdateReadPattern(offset, len);
     return;
   }
 
-  if (!IsBlockSequential(offset)) {
+  if (!IsBlockSequential(offset, len, direction)) {
     UpdateReadPattern(offset, len);
     ResetValues(rep->table_options.initial_auto_readahead_size);
     return;
@@ -148,11 +167,31 @@ void BlockPrefetcher::PrefetchIfNeeded(
   }
 
   if (rep->fs_prefetch_support) {
-    s = rep->file->Prefetch(
-        opts, handle.offset(),
-        BlockBasedTable::BlockSizeWithTrailer(handle) + readahead_size_);
+    uint64_t prefetch_offset;
+    size_t prefetch_len;
+    if (direction == PrefetchDirection::kForward) {
+      prefetch_offset = handle.offset();
+      prefetch_len =
+          BlockBasedTable::BlockSizeWithTrailer(handle) + readahead_size_;
+    } else {
+      // Backward: prefetch from offset - readahead_size_ to offset + len
+      uint64_t curr_offset = handle.offset();
+      uint64_t start =
+          curr_offset > readahead_size_ ? curr_offset - readahead_size_ : 0;
+      prefetch_offset = start;
+      prefetch_len =
+          curr_offset - start + BlockBasedTable::BlockSizeWithTrailer(handle);
+    }
+    s = rep->file->Prefetch(opts, prefetch_offset, prefetch_len);
     if (s.ok()) {
-      readahead_limit_ = offset + len + readahead_size_;
+      if (direction == PrefetchDirection::kForward) {
+        readahead_limit_ = offset + len + readahead_size_;
+        prefetch_start_ =
+            offset;  // not really used for forward but for completeness
+      } else {
+        prefetch_start_ = prefetch_offset;
+        readahead_limit_ = offset + len;
+      }
       // Keep exponentially increasing readahead size until
       // max_auto_readahead_size.
       readahead_size_ = std::min(max_auto_readahead_size, readahead_size_ * 2);