Add coroutine reads for SstFileReader (#15189)

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

Add `CoroDB::CoGet` and `CoMultiGet` overloads for standalone `SstFileReader` instances. Reuse a shared sync/coroutine implementation on the private reader `Rep`, configure and validate the read executor at `Open`, and retain synchronous fallback when unavailable.

Reviewed By: xingbowang

Differential Revision: D119431444

fbshipit-source-id: 1ac1926b6a2ffc11cd3e91a438a8c62cff5d57ec

db/coro_db.cc

@@ -19,7 +19,9 @@
 #include "folly/io/async/EventBase.h"
 #include "rocksdb/coro_db.h"
 #include "rocksdb/file_system.h"
+#include "rocksdb/sst_file_reader.h"
 #include "table/multiget_context.h"
+#include "table/sst_file_reader_impl.h"
 #include "util/coro_stats_util.h"
 
 namespace ROCKSDB_NAMESPACE {
@@ -89,6 +91,52 @@ folly::coro::Task<Status> CoroDB::CoGet(DB* db, const ReadOptions& options,
   return CoGet(db, options, db->DefaultColumnFamily(), key, value, timestamp);
 }
 
+folly::coro::Task<Status> CoroDB::CoGet(SstFileReader* reader,
+                                        const ReadOptions& options,
+                                        const Slice& key,
+                                        PinnableSlice* value) {
+  assert(reader != nullptr);
+  SstFileReader::Rep* rep = reader->rep_.get();
+  auto* read_executor = rep->read_executor;
+  auto* read_event_base =
+      read_executor == nullptr ? nullptr : read_executor->getEventBase();
+  CoroutineStatsConfig stats_config = CaptureAndDisableCoroutineStatsConfig();
+
+  Status status;
+  if (read_event_base == nullptr) {
+    CoroutineStatsContextScope stats_scope(std::move(stats_config),
+                                           rep->options.env);
+    status = reader->Get(options, key, value);
+  } else {
+    auto task = [](CoroutineStatsConfig task_stats_config, Env* task_env,
+                   SstFileReader* task_reader, const ReadOptions& task_options,
+                   Slice task_key,
+                   PinnableSlice* task_value) -> folly::coro::Task<Status> {
+      CoroutineStatsContextScope stats_scope(std::move(task_stats_config),
+                                             task_env);
+      co_return co_await folly::coro::co_nothrow(
+          task_reader->rep_->GetCoroutine(task_options, task_key, task_value));
+    }(std::move(stats_config), rep->options.env, reader, options, key, value);
+    status = co_await folly::coro::co_nothrow(folly::coro::co_withExecutor(
+        folly::Executor::getKeepAliveToken(read_event_base), std::move(task)));
+  }
+  co_return status;
+}
+
+folly::coro::Task<Status> CoroDB::CoGet(SstFileReader* reader,
+                                        const ReadOptions& options,
+                                        const Slice& key, std::string* value) {
+  assert(value != nullptr);
+  PinnableSlice pinnable_value(value);
+  assert(!pinnable_value.IsPinned());
+  Status status = co_await folly::coro::co_nothrow(
+      CoGet(reader, options, key, &pinnable_value));
+  if (status.ok() && pinnable_value.IsPinned()) {
+    value->assign(pinnable_value.data(), pinnable_value.size());
+  }
+  co_return status;
+}
+
 folly::coro::Task<Status> CoroDB::CoGetEntity(DB* db,
                                               const ReadOptions& options,
                                               ColumnFamilyHandle* column_family,
@@ -205,6 +253,54 @@ folly::coro::Task<std::vector<Status>> CoroDB::CoMultiGet(
       CoMultiGet(db, options, column_families, keys, values, timestamps));
 }
 
+folly::coro::Task<std::vector<Status>> CoroDB::CoMultiGet(
+    SstFileReader* reader, const ReadOptions& options,
+    const std::vector<Slice>& keys, std::vector<PinnableSlice>* values) {
+  assert(reader != nullptr);
+  SstFileReader::Rep* rep = reader->rep_.get();
+  auto* read_executor = rep->read_executor;
+  auto* read_event_base =
+      read_executor == nullptr ? nullptr : read_executor->getEventBase();
+  CoroutineStatsConfig stats_config = CaptureAndDisableCoroutineStatsConfig();
+
+  std::vector<Status> statuses;
+  if (read_event_base == nullptr) {
+    CoroutineStatsContextScope stats_scope(std::move(stats_config),
+                                           rep->options.env);
+    statuses = reader->MultiGet(options, keys, values);
+  } else {
+    auto task = [](CoroutineStatsConfig task_stats_config, Env* task_env,
+                   SstFileReader* task_reader, const ReadOptions& task_options,
+                   const std::vector<Slice>& task_keys,
+                   std::vector<PinnableSlice>* task_values)
+        -> folly::coro::Task<std::vector<Status>> {
+      CoroutineStatsContextScope stats_scope(std::move(task_stats_config),
+                                             task_env);
+      co_return co_await folly::coro::co_nothrow(
+          task_reader->rep_->MultiGetCoroutine(task_options, task_keys,
+                                               task_values));
+    }(std::move(stats_config), rep->options.env, reader, options, keys, values);
+    statuses = co_await folly::coro::co_nothrow(folly::coro::co_withExecutor(
+        folly::Executor::getKeepAliveToken(read_event_base), std::move(task)));
+  }
+  co_return statuses;
+}
+
+folly::coro::Task<std::vector<Status>> CoroDB::CoMultiGet(
+    SstFileReader* reader, const ReadOptions& options,
+    const std::vector<Slice>& keys, std::vector<std::string>* values) {
+  std::vector<PinnableSlice> pinnable_values;
+  std::vector<Status> statuses = co_await folly::coro::co_nothrow(
+      CoMultiGet(reader, options, keys, &pinnable_values));
+  values->resize(keys.size());
+  for (size_t i = 0; i < keys.size(); ++i) {
+    if (statuses[i].ok()) {
+      (*values)[i].assign(pinnable_values[i].data(), pinnable_values[i].size());
+    }
+  }
+  co_return statuses;
+}
+
 folly::coro::Task<void> CoroDB::CoMultiGet(
     DB* db, const ReadOptions& options, size_t num_keys,
     ColumnFamilyHandle** column_families, const Slice* keys,

include/rocksdb/coro_db.h

@@ -17,6 +17,7 @@ namespace ROCKSDB_NAMESPACE {
 
 template <typename Base>
 class CoroStackableDBBase;
+class SstFileReader;
 
 // EXPERIMENTAL native coroutine read interface.
 //
@@ -85,6 +86,15 @@ class CoroDB {
                                          const Slice& key, std::string* value,
                                          std::string* timestamp);
 
+  static folly::coro::Task<Status> CoGet(SstFileReader* reader,
+                                         const ReadOptions& options,
+                                         const Slice& key,
+                                         PinnableSlice* value);
+
+  static folly::coro::Task<Status> CoGet(SstFileReader* reader,
+                                         const ReadOptions& options,
+                                         const Slice& key, std::string* value);
+
   static folly::coro::Task<std::vector<Status>> CoMultiGet(
       DB* db, const ReadOptions& options,
       const std::vector<ColumnFamilyHandle*>& column_families,
@@ -109,6 +119,14 @@ class CoroDB {
       DB* db, const ReadOptions& options, const std::vector<Slice>& keys,
       std::vector<std::string>* values, std::vector<std::string>* timestamps);
 
+  static folly::coro::Task<std::vector<Status>> CoMultiGet(
+      SstFileReader* reader, const ReadOptions& options,
+      const std::vector<Slice>& keys, std::vector<PinnableSlice>* values);
+
+  static folly::coro::Task<std::vector<Status>> CoMultiGet(
+      SstFileReader* reader, const ReadOptions& options,
+      const std::vector<Slice>& keys, std::vector<std::string>* values);
+
   static folly::coro::Task<void> CoMultiGet(
       DB* db, const ReadOptions& options, size_t num_keys,
       ColumnFamilyHandle** column_families, const Slice* keys,

include/rocksdb/file_system.h

@@ -1138,6 +1138,9 @@ class FSRandomAccessFile {
   // Returns true when the asynchronous path is used and false when the request
   // is completed using the synchronous Read fallback.
   //
+  // All arguments and any memory they reference remain valid until cb is
+  // invoked.
+  //
   // Default implementation is to read the data synchronously and invoke the
   // callback before returning.
   virtual bool SubmitReadAsync(FSReadRequest& req, const IOOptions& opts,

include/rocksdb/options.h

@@ -795,9 +795,10 @@ struct DBOptions {
   // Default: 16
   int max_file_opening_threads = 16;
 
-  // Requested maximum number of threads in the shared read I/O executor. A DB
-  // open can increase the executor to this size but cannot reduce it. Used
-  // exclusively for asynchronous read requests (e.g. GetAsync, MultiGetAsync).
+  // Requested maximum number of threads in the shared read I/O executor.
+  // Opening a DB or SstFileReader can increase the executor to this size but
+  // cannot reduce it. Used exclusively for asynchronous read requests (e.g.
+  // GetAsync, MultiGetAsync).
   int read_io_executor_threads = 1;
 
   // If true, SST files are opened and validated asynchronously in the

include/rocksdb/sst_file_reader.h

@@ -13,6 +13,8 @@
 
 namespace ROCKSDB_NAMESPACE {
 
+class CoroDB;
+
 // SstFileReader is used to read sst files that are generated by DB or
 // SstFileWriter.
 class SstFileReader {
@@ -78,6 +80,8 @@ class SstFileReader {
   Status VerifyNumEntries(const ReadOptions& /*read_options*/);
 
  private:
+  friend class CoroDB;
+
   struct Rep;
   std::unique_ptr<Rep> rep_;
 };

table/sst_file_reader.cc

@@ -5,44 +5,35 @@
 
 #include "rocksdb/sst_file_reader.h"
 
+#include <algorithm>
+
 #include "db/arena_wrapped_db_iter.h"
 #include "db/db_iter.h"
 #include "db/dbformat.h"
 #include "db/lookup_key.h"
 #include "file/random_access_file_reader.h"
-#include "options/cf_options.h"
 #include "rocksdb/env.h"
 #include "rocksdb/file_checksum.h"
 #include "rocksdb/file_system.h"
 #include "rocksdb/utilities/types_util.h"
 #include "table/get_context.h"
+#include "table/sst_file_reader_impl.h"
 #include "table/table_builder.h"
 #include "table/table_iterator.h"
 #include "table/table_reader.h"
 
+// Generate the regular and coroutine versions of the point-read methods.
+// clang-format off
+#define WITHOUT_COROUTINES
+#include "table/sst_file_reader_sync_and_async.h"
+#undef WITHOUT_COROUTINES
+#define WITH_COROUTINES
+#include "table/sst_file_reader_sync_and_async.h"
+#undef WITH_COROUTINES
+// clang-format on
+
 namespace ROCKSDB_NAMESPACE {
 
-struct SstFileReader::Rep {
-  Options options;
-  EnvOptions soptions;
-  ImmutableOptions ioptions;
-  MutableCFOptions moptions;
-  // Keep a member variable for this, since `NewIterator()` uses a const
-  // reference of `ReadOptions`.
-  ReadOptions roptions_for_table_iter;
-
-  std::unique_ptr<TableReader> table_reader;
-
-  Rep(const Options& opts)
-      : options(opts),
-        soptions(options),
-        ioptions(options),
-        moptions(ColumnFamilyOptions(options)) {
-    roptions_for_table_iter =
-        ReadOptions(/*_verify_checksums=*/true, /*_fill_cache=*/false);
-  }
-};
-
 SstFileReader::SstFileReader(const Options& options) : rep_(new Rep(options)) {}
 
 SstFileReader::~SstFileReader() = default;
@@ -56,6 +47,14 @@ Status SstFileReader::Open(const std::string& file_path) {
   FileOptions fopts(r->soptions);
   fopts.file_checksum_func_name = kNoFileChecksumFuncName;
   const auto& fs = r->options.env->GetFileSystem();
+#if USE_COROUTINES
+  if (r->options.read_io_executor_threads <= 0) {
+    return Status::InvalidArgument(
+        "read_io_executor_threads must be greater than zero");
+  }
+  fs->SetReadIOExecutorThreads(r->options.read_io_executor_threads);
+  r->read_executor = fs->GetReadExecutor();
+#endif  // USE_COROUTINES
 
   s = fs->GetFileSize(file_path, fopts.io_options, &file_size, nullptr);
   if (s.ok()) {
@@ -88,102 +87,7 @@ Status SstFileReader::Open(const std::string& file_path) {
 std::vector<Status> SstFileReader::MultiGet(
     const ReadOptions& roptions, const std::vector<Slice>& keys,
     std::vector<PinnableSlice>* values) {
-  const auto num_keys = keys.size();
-  std::vector<Status> statuses(num_keys, Status::OK());
-  values->resize(num_keys);
-  for (size_t i = 0; i < num_keys; ++i) {
-    (*values)[i].Reset();
-  }
-
-  auto r = rep_.get();
-  const Comparator* user_comparator =
-      r->ioptions.internal_comparator.user_comparator();
-  Statistics* statistics = r->ioptions.stats;
-
-  autovector<KeyContext, MultiGetContext::MAX_BATCH_SIZE> key_context;
-  autovector<KeyContext*, MultiGetContext::MAX_BATCH_SIZE> sorted_keys;
-  autovector<GetContext, MultiGetContext::MAX_BATCH_SIZE> get_ctx;
-  autovector<MergeContext, MultiGetContext::MAX_BATCH_SIZE> merge_ctx;
-  // The loop below hands out pointers to these elements, so they have to be
-  // reserved up front: past the inline capacity an autovector spills into a
-  // std::vector that would otherwise reallocate as more keys are appended.
-  key_context.reserve(num_keys);
-  get_ctx.reserve(num_keys);
-  merge_ctx.reserve(num_keys);
-  sorted_keys.resize(num_keys);
-  for (size_t i = 0; i < num_keys; ++i) {
-    PinnableSlice* val = &(*values)[i];
-    merge_ctx.emplace_back();
-    key_context.emplace_back(nullptr, keys[i], val, nullptr,
-                             nullptr /* timestamp */, &statuses[i]);
-    get_ctx.emplace_back(
-        user_comparator, r->ioptions.merge_operator.get(), nullptr /* logger */,
-        statistics, GetContext::kNotFound, *key_context[i].key, val,
-        nullptr /* columns */, nullptr /* timestamp */,
-        nullptr /* value_found */, &merge_ctx[i], true,
-        &key_context[i].max_covering_tombstone_seq, r->ioptions.clock);
-    key_context[i].get_context = &get_ctx[i];
-  }
-  for (size_t i = 0; i < num_keys; ++i) {
-    sorted_keys[i] = &key_context[i];
-  }
-
-  struct CompareKeyContext {
-    explicit CompareKeyContext(const Comparator* comp) : comparator(comp) {}
-    inline bool operator()(const KeyContext* lhs, const KeyContext* rhs) const {
-      return comparator->CompareWithoutTimestamp(*(lhs->key), false,
-                                                 *(rhs->key), false) < 0;
-    }
-    const Comparator* comparator;
-  };
-
-  std::sort(sorted_keys.begin(), sorted_keys.end(),
-            CompareKeyContext(user_comparator));
-  const auto sequence = roptions.snapshot != nullptr
-                            ? roptions.snapshot->GetSequenceNumber()
-                            : kMaxSequenceNumber;
-  // A MultiGetContext holds at most MAX_BATCH_SIZE keys, so look the sorted
-  // keys up in batches of that size, as DBImpl::MultiGetImpl does. Exceeding
-  // it is undefined behavior, not just a missed optimization.
-  size_t keys_left = num_keys;
-  while (keys_left > 0) {
-    const size_t batch_size = (keys_left > MultiGetContext::MAX_BATCH_SIZE)
-                                  ? MultiGetContext::MAX_BATCH_SIZE
-                                  : keys_left;
-    MultiGetContext ctx(&sorted_keys, num_keys - keys_left, batch_size,
-                        sequence, roptions, r->ioptions.fs.get(), nullptr);
-    MultiGetRange range = ctx.GetMultiGetRange();
-    r->table_reader->MultiGet(roptions, &range,
-                              r->moptions.prefix_extractor.get(),
-                              false /* skip filters */);
-    keys_left -= batch_size;
-  }
-
-  for (size_t i = 0; i < num_keys; ++i) {
-    get_ctx[i].ReportCounters();
-
-    if (statuses[i].ok()) {
-      switch (get_ctx[i].State()) {
-        case GetContext::kFound:
-          break;
-        case GetContext::kNotFound:
-        case GetContext::kDeleted:
-          statuses[i] = Status::NotFound();
-          break;
-        case GetContext::kMerge:
-          statuses[i] = Status::MergeInProgress();
-          break;
-        case GetContext::kCorrupt:
-          statuses[i] = Status::Corruption();
-          break;
-        case GetContext::kUnexpectedBlobIndex:
-        case GetContext::kMergeOperatorFailed:
-          statuses[i] = Status::Corruption();
-          break;
-      };
-    }
-  }
-  return statuses;
+  return rep_->MultiGet(roptions, keys, values);
 }
 
 std::vector<Status> SstFileReader::MultiGet(const ReadOptions& roptions,
@@ -202,50 +106,7 @@ std::vector<Status> SstFileReader::MultiGet(const ReadOptions& roptions,
 
 Status SstFileReader::Get(const ReadOptions& roptions, const Slice& key,
                           PinnableSlice* value) {
-  auto r = rep_.get();
-  value->Reset();
-
-  const Comparator* user_comparator =
-      r->ioptions.internal_comparator.user_comparator();
-  Statistics* statistics = r->ioptions.stats;
-
-  Status status;
-  MergeContext merge_context;
-  SequenceNumber max_covering_tombstone_seq = 0;
-  GetContext get_ctx(user_comparator, r->ioptions.merge_operator.get(),
-                     nullptr /* logger */, statistics, GetContext::kNotFound,
-                     key, value, nullptr /* columns */, nullptr /* timestamp */,
-                     nullptr /* value_found */, &merge_context, true,
-                     &max_covering_tombstone_seq, r->ioptions.clock);
-
-  LookupKey lkey(key, kMaxSequenceNumber);
-  status = r->table_reader->Get(roptions, lkey.internal_key(), &get_ctx,
-                                r->moptions.prefix_extractor.get(),
-                                false /* skip_filters */);
-
-  get_ctx.ReportCounters();
-
-  if (status.ok()) {
-    switch (get_ctx.State()) {
-      case GetContext::kFound:
-        break;
-      case GetContext::kNotFound:
-      case GetContext::kDeleted:
-        status = Status::NotFound();
-        break;
-      case GetContext::kMerge:
-        status = Status::MergeInProgress();
-        break;
-      case GetContext::kCorrupt:
-        status = Status::Corruption();
-        break;
-      case GetContext::kUnexpectedBlobIndex:
-      case GetContext::kMergeOperatorFailed:
-        status = Status::Corruption();
-        break;
-    }
-  }
-  return status;
+  return rep_->Get(roptions, key, value);
 }
 
 Status SstFileReader::Get(const ReadOptions& roptions, const Slice& key,

table/sst_file_reader_impl.h

@@ -0,0 +1,49 @@
+// Copyright (c) Meta Platforms, Inc. and affiliates.
+// This source code is licensed under both the GPLv2 (found in the
+// COPYING file in the root directory) and Apache 2.0 License
+// (found in the LICENSE.Apache file in the root directory).
+
+#pragma once
+
+#include "options/cf_options.h"
+#include "rocksdb/sst_file_reader.h"
+#include "table/table_reader.h"
+#include "util/coro_utils.h"
+#if USE_COROUTINES
+#include "folly/executors/IOExecutor.h"
+#endif  // USE_COROUTINES
+
+namespace ROCKSDB_NAMESPACE {
+
+struct SstFileReader::Rep {
+  Options options;
+  EnvOptions soptions;
+  ImmutableOptions ioptions;
+  MutableCFOptions moptions;
+  // Keep a member variable for this, since `NewIterator()` uses a const
+  // reference of `ReadOptions`.
+  ReadOptions roptions_for_table_iter;
+
+  std::unique_ptr<TableReader> table_reader;
+#if USE_COROUTINES
+  folly::IOExecutor* read_executor = nullptr;
+#endif  // USE_COROUTINES
+
+  DECLARE_SYNC_AND_ASYNC(std::vector<Status>, MultiGet,
+                         const ReadOptions& roptions,
+                         const std::vector<Slice>& keys,
+                         std::vector<PinnableSlice>* values);
+  DECLARE_SYNC_AND_ASYNC(Status, Get, const ReadOptions& roptions,
+                         const Slice& key, PinnableSlice* value);
+
+  explicit Rep(const Options& opts)
+      : options(opts),
+        soptions(options),
+        ioptions(options),
+        moptions(ColumnFamilyOptions(options)) {
+    roptions_for_table_iter =
+        ReadOptions(/*_verify_checksums=*/true, /*_fill_cache=*/false);
+  }
+};
+
+}  // namespace ROCKSDB_NAMESPACE

table/sst_file_reader_sync_and_async.h

@@ -0,0 +1,154 @@
+// Copyright (c) Meta Platforms, Inc. and affiliates.
+// This source code is licensed under both the GPLv2 (found in the
+// COPYING file in the root directory) and Apache 2.0 License
+// (found in the LICENSE.Apache file in the root directory).
+
+#include "util/coro_utils.h"
+
+#if defined(WITHOUT_COROUTINES) || \
+    (defined(USE_COROUTINES) && defined(WITH_COROUTINES))
+
+namespace ROCKSDB_NAMESPACE {
+
+DEFINE_SYNC_AND_ASYNC(std::vector<Status>, SstFileReader::Rep::MultiGet)
+(const ReadOptions& roptions, const std::vector<Slice>& keys,
+ std::vector<PinnableSlice>* values) {
+  const size_t num_keys = keys.size();
+  std::vector<Status> statuses(num_keys);
+  values->resize(num_keys);
+  for (size_t i = 0; i < num_keys; ++i) {
+    (*values)[i].Reset();
+  }
+
+  const Comparator* user_comparator =
+      ioptions.internal_comparator.user_comparator();
+  Statistics* statistics = ioptions.stats;
+
+  autovector<KeyContext, MultiGetContext::MAX_BATCH_SIZE> key_context;
+  autovector<KeyContext*, MultiGetContext::MAX_BATCH_SIZE> sorted_keys;
+  autovector<GetContext, MultiGetContext::MAX_BATCH_SIZE> get_ctx;
+  autovector<MergeContext, MultiGetContext::MAX_BATCH_SIZE> merge_ctx;
+  key_context.reserve(num_keys);
+  get_ctx.reserve(num_keys);
+  merge_ctx.reserve(num_keys);
+  sorted_keys.resize(num_keys);
+  for (size_t i = 0; i < num_keys; ++i) {
+    PinnableSlice* val = &(*values)[i];
+    merge_ctx.emplace_back();
+    key_context.emplace_back(nullptr, keys[i], val, nullptr,
+                             nullptr /* timestamp */, &statuses[i]);
+    get_ctx.emplace_back(
+        user_comparator, ioptions.merge_operator.get(), nullptr /* logger */,
+        statistics, GetContext::kNotFound, *key_context[i].key, val,
+        nullptr /* columns */, nullptr /* timestamp */,
+        nullptr /* value_found */, &merge_ctx[i], true,
+        &key_context[i].max_covering_tombstone_seq, ioptions.clock);
+    key_context[i].get_context = &get_ctx[i];
+  }
+  for (size_t i = 0; i < num_keys; ++i) {
+    sorted_keys[i] = &key_context[i];
+  }
+
+  struct CompareKeyContext {
+    explicit CompareKeyContext(const Comparator* comp) : comparator(comp) {}
+    bool operator()(const KeyContext* lhs, const KeyContext* rhs) const {
+      return comparator->CompareWithoutTimestamp(*(lhs->key), false,
+                                                 *(rhs->key), false) < 0;
+    }
+    const Comparator* comparator;
+  };
+
+  std::sort(sorted_keys.begin(), sorted_keys.end(),
+            CompareKeyContext(user_comparator));
+  const SequenceNumber sequence = roptions.snapshot != nullptr
+                                      ? roptions.snapshot->GetSequenceNumber()
+                                      : kMaxSequenceNumber;
+#ifdef WITH_COROUTINES
+  constexpr bool kUseCoroRead = true;
+#else
+  constexpr bool kUseCoroRead = false;
+#endif
+  for (size_t start_key = 0; start_key < num_keys;
+       start_key += MultiGetContext::MAX_BATCH_SIZE) {
+    const size_t batch_size =
+        std::min<size_t>(MultiGetContext::MAX_BATCH_SIZE, num_keys - start_key);
+    MultiGetContext ctx(&sorted_keys, start_key, batch_size, sequence, roptions,
+                        ioptions.fs.get(), nullptr, kUseCoroRead);
+    MultiGetRange range = ctx.GetMultiGetRange();
+    CO_AWAIT(table_reader->MultiGet, roptions, &range,
+             moptions.prefix_extractor.get(), false /* skip filters */);
+  }
+
+  for (size_t i = 0; i < num_keys; ++i) {
+    get_ctx[i].ReportCounters();
+
+    if (statuses[i].ok()) {
+      switch (get_ctx[i].State()) {
+        case GetContext::kFound:
+          break;
+        case GetContext::kNotFound:
+        case GetContext::kDeleted:
+          statuses[i] = Status::NotFound();
+          break;
+        case GetContext::kMerge:
+          statuses[i] = Status::MergeInProgress();
+          break;
+        case GetContext::kCorrupt:
+        case GetContext::kUnexpectedBlobIndex:
+        case GetContext::kMergeOperatorFailed:
+          statuses[i] = Status::Corruption();
+          break;
+      }
+    }
+  }
+  CO_RETURN statuses;
+}
+
+DEFINE_SYNC_AND_ASYNC(Status, SstFileReader::Rep::Get)
+(const ReadOptions& roptions, const Slice& key, PinnableSlice* value) {
+  value->Reset();
+
+  const Comparator* user_comparator =
+      ioptions.internal_comparator.user_comparator();
+  Statistics* statistics = ioptions.stats;
+
+  MergeContext merge_context;
+  SequenceNumber max_covering_tombstone_seq = 0;
+  GetContext get_ctx(user_comparator, ioptions.merge_operator.get(),
+                     nullptr /* logger */, statistics, GetContext::kNotFound,
+                     key, value, nullptr /* columns */, nullptr /* timestamp */,
+                     nullptr /* value_found */, &merge_context, true,
+                     &max_covering_tombstone_seq, ioptions.clock);
+
+  LookupKey lkey(key, kMaxSequenceNumber);
+  Status status =
+      CO_AWAIT(table_reader->Get, roptions, lkey.internal_key(), &get_ctx,
+               moptions.prefix_extractor.get(), false /* skip_filters */);
+
+  get_ctx.ReportCounters();
+
+  if (status.ok()) {
+    switch (get_ctx.State()) {
+      case GetContext::kFound:
+        break;
+      case GetContext::kNotFound:
+      case GetContext::kDeleted:
+        status = Status::NotFound();
+        break;
+      case GetContext::kMerge:
+        status = Status::MergeInProgress();
+        break;
+      case GetContext::kCorrupt:
+      case GetContext::kUnexpectedBlobIndex:
+      case GetContext::kMergeOperatorFailed:
+        status = Status::Corruption();
+        break;
+    }
+  }
+  CO_RETURN status;
+}
+
+}  // namespace ROCKSDB_NAMESPACE
+
+#endif  // defined(WITHOUT_COROUTINES) ||
+        // (defined(USE_COROUTINES) && defined(WITH_COROUTINES))

table/sst_file_reader_test.cc

@@ -37,6 +37,11 @@
 #include "util/defer.h"
 #include "utilities/merge_operators.h"
 
+#if USE_COROUTINES
+#include "folly/coro/BlockingWait.h"
+#include "rocksdb/coro_db.h"
+#endif
+
 namespace ROCKSDB_NAMESPACE {
 
 std::string EncodeAsString(uint64_t v) {
@@ -159,6 +164,149 @@ class SstFileReaderTest : public testing::Test {
   Env* env_;
 };
 
+#if USE_COROUTINES
+TEST_F(SstFileReaderTest, CoroutinePointReads) {
+  const std::vector<std::string> keys = {"a", "b", "c", "d", "e", "f"};
+  CreateFile(sst_name_, keys);
+
+  SstFileReader reader(options_);
+  ASSERT_OK(reader.Open(sst_name_));
+
+  int coroutine_read_count = 0;
+  SyncPoint::GetInstance()->SetCallBack(
+      "RandomAccessFileReader::ReadCoroutine:SubmitReadAsync",
+      [&](void*) { ++coroutine_read_count; });
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  ReadOptions read_options;
+  read_options.fill_cache = false;
+  std::string string_value;
+  ASSERT_OK(folly::coro::blockingWait(
+      CoroDB::CoGet(&reader, read_options, keys[0], &string_value)));
+  ASSERT_EQ(string_value, keys[0]);
+
+  PinnableSlice pinnable_value;
+  ASSERT_OK(folly::coro::blockingWait(
+      CoroDB::CoGet(&reader, read_options, keys[3], &pinnable_value)));
+  ASSERT_EQ(pinnable_value, keys[3]);
+
+  const std::vector<Slice> read_keys = {keys[0], keys[2], "missing"};
+  std::vector<std::string> string_values;
+  std::vector<Status> statuses = folly::coro::blockingWait(
+      CoroDB::CoMultiGet(&reader, read_options, read_keys, &string_values));
+  ASSERT_OK(statuses[0]);
+  ASSERT_EQ(string_values[0], keys[0]);
+  ASSERT_TRUE(statuses[1].IsNotFound());
+  ASSERT_TRUE(statuses[2].IsNotFound());
+
+  std::vector<PinnableSlice> pinnable_values;
+  statuses = folly::coro::blockingWait(
+      CoroDB::CoMultiGet(&reader, read_options, read_keys, &pinnable_values));
+  ASSERT_OK(statuses[0]);
+  ASSERT_EQ(pinnable_values[0], keys[0]);
+  ASSERT_TRUE(statuses[1].IsNotFound());
+  ASSERT_TRUE(statuses[2].IsNotFound());
+
+  SyncPoint::GetInstance()->DisableProcessing();
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+  if (options_.env->GetFileSystem()->GetReadExecutor() != nullptr) {
+    ASSERT_GT(coroutine_read_count, 0);
+  }
+}
+
+TEST_F(SstFileReaderTest, CoroutinePointReadFallback) {
+  class NoReadExecutorFileSystem final : public FileSystemWrapper {
+   public:
+    explicit NoReadExecutorFileSystem(const std::shared_ptr<FileSystem>& target)
+        : FileSystemWrapper(target) {}
+
+    const char* Name() const override { return "NoReadExecutorFileSystem"; }
+    folly::IOExecutor* GetReadExecutor() override { return nullptr; }
+    void SetReadIOExecutorThreads(int /*number*/) override {}
+  };
+
+  const std::vector<std::string> keys = {"a", "b", "c"};
+  CreateFile(sst_name_, keys);
+
+  auto file_system =
+      std::make_shared<NoReadExecutorFileSystem>(env_->GetFileSystem());
+  std::unique_ptr<Env> env = NewCompositeEnv(file_system);
+  Options options = options_;
+  options.env = env.get();
+  SstFileReader reader(options);
+  ASSERT_OK(reader.Open(sst_name_));
+
+  int coroutine_read_count = 0;
+  SyncPoint::GetInstance()->SetCallBack(
+      "RandomAccessFileReader::ReadCoroutine:SubmitReadAsync",
+      [&](void*) { ++coroutine_read_count; });
+  SyncPoint::GetInstance()->EnableProcessing();
+
+  std::string value;
+  ASSERT_OK(folly::coro::blockingWait(
+      CoroDB::CoGet(&reader, ReadOptions(), keys[0], &value)));
+
+  SyncPoint::GetInstance()->DisableProcessing();
+  SyncPoint::GetInstance()->ClearAllCallBacks();
+  ASSERT_EQ(value, keys[0]);
+  ASSERT_EQ(coroutine_read_count, 0);
+}
+#endif  // USE_COROUTINES
+
+TEST_F(SstFileReaderTest, MultiGetBatchBoundaries) {
+  constexpr size_t kNumKeys = MultiGetContext::MAX_BATCH_SIZE * 2 + 1;
+  std::vector<std::string> key_storage;
+  key_storage.reserve(kNumKeys);
+
+  SstFileWriter writer(soptions_, options_);
+  ASSERT_OK(writer.Open(sst_name_));
+  for (size_t i = 0; i < kNumKeys; ++i) {
+    key_storage.emplace_back(EncodeAsString(i));
+    ASSERT_OK(writer.Put(key_storage.back(), key_storage.back()));
+  }
+  ASSERT_OK(writer.Finish());
+
+  SstFileReader reader(options_);
+  ASSERT_OK(reader.Open(sst_name_));
+
+  std::vector<Slice> keys;
+  std::vector<std::string> expected_values;
+  keys.reserve(key_storage.size());
+  expected_values.reserve(key_storage.size());
+  for (size_t i = 0; i < kNumKeys; ++i) {
+    const size_t key_index = (i * 17) % kNumKeys;
+    keys.emplace_back(key_storage[key_index]);
+    expected_values.emplace_back(key_storage[key_index]);
+  }
+
+  std::vector<std::string> values;
+  std::vector<Status> statuses = reader.MultiGet(ReadOptions(), keys, &values);
+  ASSERT_EQ(statuses.size(), kNumKeys);
+  for (size_t i = 0; i < kNumKeys; ++i) {
+    ASSERT_OK(statuses[i]);
+  }
+  ASSERT_EQ(values, expected_values);
+
+  statuses = reader.MultiGet(ReadOptions(), {}, &values);
+  ASSERT_TRUE(statuses.empty());
+  ASSERT_TRUE(values.empty());
+
+#if USE_COROUTINES
+  statuses = folly::coro::blockingWait(
+      CoroDB::CoMultiGet(&reader, ReadOptions(), keys, &values));
+  ASSERT_EQ(statuses.size(), kNumKeys);
+  for (size_t i = 0; i < kNumKeys; ++i) {
+    ASSERT_OK(statuses[i]);
+  }
+  ASSERT_EQ(values, expected_values);
+
+  statuses = folly::coro::blockingWait(
+      CoroDB::CoMultiGet(&reader, ReadOptions(), {}, &values));
+  ASSERT_TRUE(statuses.empty());
+  ASSERT_TRUE(values.empty());
+#endif  // USE_COROUTINES
+}
+
 class FailingAppendWritableFile : public FSWritableFileOwnerWrapper {
  public:
   FailingAppendWritableFile(std::unique_ptr<FSWritableFile>&& target,

unreleased_history/public_api_changes/coro_sst_file_reader.md

@@ -0,0 +1 @@
+Added experimental `CoroDB::CoGet()` and `CoroDB::CoMultiGet()` overloads for point reads from an `SstFileReader`.