Add prefetch feature enum to FSSupportedOps (#13917)

Summary:
**Problem:** RocksDB was making unnecessary prefetch system calls on file systems that don't support prefetch operations, potentially leading to wasted CPU cycles.

**Fix:** Add kFSPrefetch to FSSupportedOps enum to allow file systems to indicate prefetch support capability. File systems can now opt out of prefetch calls by not setting this field.

**Backwards compatibility:** File systems that don't override SupportedOps() continue to receive prefetch calls exactly as before. Only file systems that explicitly opt out by not setting kFSPrefetch will avoid the calls.

Pull Request resolved: https://github.com/facebook/rocksdb/pull/13917

Test Plan:
- Added a new test in block_based_table_reader.
- Run existing tests: ```make prefetch_test && ./prefetch_test```

Reviewed By: anand1976

Differential Revision: D81607145

Pulled By: nmk70

fbshipit-source-id: 3bbefa05919034e8776ea4e4540cdc695cdc6d3f

env/fs_posix.cc

@@ -1272,6 +1272,7 @@ class PosixFileSystem : public FileSystem {
       supported_ops |= (1 << FSSupportedOps::kAsyncIO);
     }
 #endif
+    supported_ops |= (1 << FSSupportedOps::kFSPrefetch);
   }
 
 #if defined(ROCKSDB_IOURING_PRESENT)

include/rocksdb/file_system.h

@@ -90,6 +90,7 @@ enum FSSupportedOps {
   kVerifyAndReconstructRead,  // Supports a higher level of data integrity. See
                               // the verify_and_reconstruct_read flag in
                               // IOOptions.
+  kFSPrefetch,                // Supports prefetch operations
 };
 
 // Per-request options that can be passed down to the FileSystem
@@ -771,12 +772,13 @@ class FileSystem : public Customizable {
   //  If async_io is supported by the underlying FileSystem, then supported_ops
   //  will have corresponding bit (i.e FSSupportedOps::kAsyncIO) set to 1.
   //
-  // By default, async_io operation is set and FS should override this API and
-  // set all the operations they support provided in FSSupportedOps (including
-  // async_io).
+  // By default, async_io and prefetch operation are set and FS should override
+  // this API and set all the operations they support provided in FSSupportedOps
+  // (including async_io and prefetch).
   virtual void SupportedOps(int64_t& supported_ops) {
     supported_ops = 0;
     supported_ops |= (1 << FSSupportedOps::kAsyncIO);
+    supported_ops |= (1 << FSSupportedOps::kFSPrefetch);
   }
 
   // If you're adding methods here, remember to add them to EnvWrapper too.

table/block_based/block_based_table_reader.h

@@ -610,7 +610,9 @@ struct BlockBasedTable::Rep {
         file_size(_file_size),
         level(_level),
         immortal_table(_immortal_table),
-        user_defined_timestamps_persisted(_user_defined_timestamps_persisted) {}
+        user_defined_timestamps_persisted(_user_defined_timestamps_persisted),
+        fs_prefetch_support(CheckFSFeatureSupport(
+            _ioptions.fs.get(), FSSupportedOps::kFSPrefetch)) {}
   ~Rep() { status.PermitUncheckedError(); }
   const ImmutableOptions& ioptions;
   const EnvOptions& env_options;
@@ -699,6 +701,8 @@ struct BlockBasedTable::Rep {
   // `end_key` for range deletion entries.
   const bool user_defined_timestamps_persisted;
 
+  const bool fs_prefetch_support;
+
   // Set to >0 when the file is known to be obsolete and should have its block
   // cache entries evicted on close. NOTE: when the file becomes obsolete,
   // there could be multiple table cache references that all mark this file as

table/block_based/block_based_table_reader_test.cc

@@ -1623,6 +1623,91 @@ TEST_P(BlockBasedTableReaderMultiScanTest, MultiScanUnpinPreviousBlocks) {
   }
 }
 
+// Test that fs_prefetch_support flag is correctly initialized during table
+// construction based on filesystem capabilities
+TEST_P(BlockBasedTableReaderTest, FSPrefetchSupportInitializedCorrectly) {
+  class ConfigurablePrefetchFS : public FileSystemWrapper {
+   public:
+    ConfigurablePrefetchFS(const std::shared_ptr<FileSystem>& target,
+                           bool support_prefetch)
+        : FileSystemWrapper(target), support_prefetch_(support_prefetch) {}
+
+    static const char* kClassName() { return "ConfigurablePrefetchFS"; }
+    const char* Name() const override { return kClassName(); }
+
+    void SupportedOps(int64_t& supported_ops) override {
+      target()->SupportedOps(supported_ops);
+      if (!support_prefetch_) {  // Disable prefetch support if requested
+        supported_ops &= ~(1 << FSSupportedOps::kFSPrefetch);
+      }
+    }
+
+   private:
+    bool support_prefetch_;
+  };
+
+  // Prepare test table
+  Options options;
+  options.persist_user_defined_timestamps = persist_udt_;
+  if (udt_enabled_) {
+    options.comparator = test::BytewiseComparatorWithU64TsWrapper();
+  }
+  size_t ts_sz = options.comparator->timestamp_size();
+  auto kv = BlockBasedTableReaderBaseTest::GenerateKVMap(5, true, ts_sz);
+  std::string table_name = "BlockBasedTableReaderTest_BlockPrefetcherTest" +
+                           CompressionTypeToString(compression_type_);
+  ImmutableOptions ioptions(options);
+  CreateTable(table_name, ioptions, compression_type_, kv,
+              compression_parallel_threads_, compression_dict_bytes_);
+
+  // Test Case 1: Filesystem supports prefetch, fs_prefetch_support should be
+  // true
+  {
+    auto fs_with_prefetch = std::make_shared<ConfigurablePrefetchFS>(
+        env_->GetFileSystem(), true /* support_prefetch */);
+    std::unique_ptr<Env> env_wrapper(
+        new CompositeEnvWrapper(env_, fs_with_prefetch));
+    options.env = env_wrapper.get();
+
+    FileOptions fopts;
+    fopts.use_direct_reads = use_direct_reads_;
+    InternalKeyComparator cmp(options.comparator);
+    ImmutableOptions iopts(options);
+
+    std::unique_ptr<BlockBasedTable> table;
+    NewBlockBasedTableReader(fopts, iopts, cmp, table_name, &table,
+                             false /* prefetch_index_and_filter_in_cache */,
+                             nullptr, persist_udt_);
+
+    ASSERT_TRUE(table->get_rep()->fs_prefetch_support);
+    ASSERT_TRUE(CheckFSFeatureSupport(fs_with_prefetch.get(),
+                                      FSSupportedOps::kFSPrefetch));
+  }
+
+  // Test Case 2: Filesystem doesn't support prefetch, fs_prefetch_support
+  // should be false
+  {
+    auto fs_without_prefetch = std::make_shared<ConfigurablePrefetchFS>(
+        env_->GetFileSystem(), false /* support_prefetch */);
+    std::unique_ptr<Env> env_wrapper(
+        new CompositeEnvWrapper(env_, fs_without_prefetch));
+    options.env = env_wrapper.get();
+
+    FileOptions fopts;
+    fopts.use_direct_reads = use_direct_reads_;
+    InternalKeyComparator cmp(options.comparator);
+    ImmutableOptions iopts(options);
+
+    std::unique_ptr<BlockBasedTable> table;
+    NewBlockBasedTableReader(fopts, iopts, cmp, table_name, &table,
+                             false /* prefetch_index_and_filter_in_cache */,
+                             nullptr, persist_udt_);
+
+    ASSERT_FALSE(table->get_rep()->fs_prefetch_support);
+    ASSERT_FALSE(CheckFSFeatureSupport(fs_without_prefetch.get(),
+                                       FSSupportedOps::kFSPrefetch));
+  }
+}
 std::vector<BlockBasedTableReaderTestParam> GenerateCombinedParameters(
     const std::vector<CompressionType>& compression_types,
     const std::vector<bool>& use_direct_read_flags,

table/block_based/block_prefetcher.cc

@@ -44,12 +44,16 @@ void BlockPrefetcher::PrefetchIfNeeded(
       if (!s.ok()) {
         return;
       }
-      s = rep->file->Prefetch(opts, offset, len + compaction_readahead_size_);
-      if (s.ok()) {
-        readahead_limit_ = offset + len + compaction_readahead_size_;
-        return;
-      } else if (!s.IsNotSupported()) {
-        return;
+      if (rep->fs_prefetch_support) {
+        s = rep->file->Prefetch(opts, offset, len + compaction_readahead_size_);
+        if (s.ok()) {
+          readahead_limit_ = offset + len + compaction_readahead_size_;
+          return;
+        } else if (!s.IsNotSupported()) {
+          return;
+        }
+        // If FS prefetch returned NotSupported despite feature bit being set,
+        // fall through to use internal prefetch buffer.
       }
     }
     // If FS prefetch is not supported, fall back to use internal prefetch
@@ -142,19 +146,23 @@ void BlockPrefetcher::PrefetchIfNeeded(
   if (!s.ok()) {
     return;
   }
-  s = rep->file->Prefetch(
-      opts, handle.offset(),
-      BlockBasedTable::BlockSizeWithTrailer(handle) + readahead_size_);
-  if (s.IsNotSupported()) {
-    rep->CreateFilePrefetchBufferIfNotExists(
-        readahead_params, &prefetch_buffer_, readaheadsize_cb,
-        /*usage=*/FilePrefetchBufferUsage::kUserScanPrefetch);
-    return;
-  }
 
-  readahead_limit_ = offset + len + readahead_size_;
-  // Keep exponentially increasing readahead size until
-  // max_auto_readahead_size.
-  readahead_size_ = std::min(max_auto_readahead_size, readahead_size_ * 2);
+  if (rep->fs_prefetch_support) {
+    s = rep->file->Prefetch(
+        opts, handle.offset(),
+        BlockBasedTable::BlockSizeWithTrailer(handle) + readahead_size_);
+    if (s.ok()) {
+      readahead_limit_ = offset + len + readahead_size_;
+      // Keep exponentially increasing readahead size until
+      // max_auto_readahead_size.
+      readahead_size_ = std::min(max_auto_readahead_size, readahead_size_ * 2);
+      return;
+    }
+  }
+  // If FS prefetch is not supported or returned NotSupported, fall back to use
+  // internal prefetch buffer.
+  rep->CreateFilePrefetchBufferIfNotExists(
+      readahead_params, &prefetch_buffer_, readaheadsize_cb,
+      /*usage=*/FilePrefetchBufferUsage::kUserScanPrefetch);
 }
 }  // namespace ROCKSDB_NAMESPACE

unreleased_history/new_features/fs_prefetch_support.md

@@ -0,0 +1 @@
+Add kFSPrefetch to FSSupportedOps enum to allow file systems to indicate prefetch support capability, avoiding unnecessary prefetch system calls on file systems that don't support them.