Use batched local reads for index projector (facebookincubator#845)

Summary:

Use the positioned `ReadFile::preadv(...)` batch overload from `DirectDataInput` so local direct-I/O files can take the native io_uring batch path and other files fall back to the default loop over `pread`. Split `TabletReader` data and metadata/index read handles so direct data reads can use direct I/O and io_uring while footer, metadata, and index reads use a buffered handle. Add benchmark plumbing for `--use_io_uring` and update focused unit coverage.

Reviewed By: tanjialiang

Differential Revision: D107930292

Author: xiaoxmeng

dwio/nimble/tablet/DataInput.cpp

@@ -16,10 +16,10 @@
 #include "dwio/nimble/tablet/DataInput.h"
 
 #include <algorithm>
+#include <limits>
 #include <numeric>
 
 #include "dwio/nimble/common/Exceptions.h"
-#include "folly/synchronization/Latch.h"
 #include "velox/common/base/BitUtil.h"
 #include "velox/common/base/CoalesceIo.h"
 #include "velox/common/memory/MemoryAllocator.h"
@@ -30,6 +30,20 @@ namespace facebook::nimble {
 
 // --- DirectDataInput ---
 
+namespace {
+
+velox::ReadFile* checkedFile(velox::ReadFile* file) {
+  NIMBLE_CHECK_NOT_NULL(file);
+  return file;
+}
+
+uint64_t readAlignment(const velox::ReadFile& file) {
+  uint64_t alignment{0};
+  return file.directIo(/*alignment=*/alignment) ? alignment : 1;
+}
+
+} // namespace
+
 /*static*/ std::string_view DirectDataInput::stateName(State state) {
   switch (state) {
     case State::kInit:
@@ -45,22 +59,18 @@ namespace facebook::nimble {
 }
 
 DirectDataInput::DirectDataInput(velox::ReadFile* file, const Options& options)
-    : file_{file},
+    : file_{checkedFile(file)},
       pool_{options.pool},
-      executor_{options.executor},
       ioStats_{options.ioStats},
-      alignment_{options.alignment},
+      alignment_{readAlignment(*file_)},
       allocAlignment_{std::max(
           alignment_,
           static_cast<uint64_t>(
               velox::memory::MemoryAllocator::kMinAlignment))},
       maxCoalesceDistance_{
           std::min(options.maxCoalesceDistance, kMaxCoalesceDistance)},
-      maxCoalesceBytes_{options.maxCoalesceBytes},
-      minIoGroupsPerTask_{std::max(1, options.minIoGroupsPerTask)} {
-  NIMBLE_CHECK_NOT_NULL(file_);
+      maxCoalesceBytes_{options.maxCoalesceBytes} {
   NIMBLE_CHECK_NOT_NULL(pool_);
-  NIMBLE_CHECK_NOT_NULL(executor_);
   NIMBLE_CHECK_NOT_NULL(ioStats_);
   NIMBLE_CHECK_GT(alignment_, 0, "alignment must be positive");
   NIMBLE_CHECK(
@@ -230,54 +240,24 @@ std::pair<char*, DataInput::Handle> DirectDataInput::allocateBuffer(
 
 void DirectDataInput::executeIoGroups(
     std::vector<IoGroup>& ioGroups,
-    char* buffer) {
+    char* buffer,
+    uint64_t bufferSize) {
   NIMBLE_CHECK(!ioGroups.empty());
-  const auto numGroups = ioGroups.size();
-
-  // Batch IO groups into tasks. Each task reads multiple groups
-  // sequentially to reduce executor overhead.
-  const auto batchSize = static_cast<size_t>(minIoGroupsPerTask_);
-  const auto numTasks = velox::bits::divRoundUp(numGroups, batchSize);
-
-  std::exception_ptr readError;
-  std::mutex errorMutex;
-  folly::Latch latch(numTasks);
-
-  for (size_t t = 0; t < numTasks; ++t) {
-    const auto start = t * batchSize;
-    const auto end = std::min(start + batchSize, numGroups);
-    executor_->add([this,
-                    &ioGroups,
-                    buffer,
-                    start,
-                    end,
-                    &readError,
-                    &errorMutex,
-                    &latch]() {
-      try {
-        for (size_t i = start; i < end; ++i) {
-          velox::common::testutil::TestValue::adjust(
-              "facebook::nimble::DirectDataInput::executeIoGroups", this);
-          file_->pread(
-              ioGroups[i].readOffset,
-              ioGroups[i].readSize,
-              buffer + ioGroups[i].bufferOffset);
-        }
-      } catch (...) {
-        std::lock_guard<std::mutex> lock(errorMutex);
-        if (readError == nullptr) {
-          readError = std::current_exception();
-        }
-      }
-      latch.count_down();
-    });
-  }
-
-  latch.wait();
-
-  if (readError != nullptr) {
-    std::rethrow_exception(readError);
+  std::vector<velox::common::Region> readRegions;
+  readRegions.reserve(ioGroups.size());
+  std::vector<folly::Range<char*>> readBuffers;
+  readBuffers.reserve(ioGroups.size());
+  for (const auto& ioGroup : ioGroups) {
+    readRegions.emplace_back(ioGroup.readOffset, ioGroup.readSize);
+    readBuffers.emplace_back(
+        buffer + ioGroup.bufferOffset, static_cast<size_t>(ioGroup.readSize));
   }
+  const auto bytesRead = file_->preadv(
+      folly::Range<const velox::common::Region*>(
+          readRegions.data(), readRegions.size()),
+      folly::Range<const folly::Range<char*>*>(
+          readBuffers.data(), readBuffers.size()));
+  NIMBLE_CHECK_EQ(bytesRead, bufferSize, "preadv returned a short read");
 }
 
 DataInput::Handle DirectDataInput::load() {
@@ -325,7 +305,7 @@ DataInput::Handle DirectDataInput::load() {
   uint64_t ioUs{0};
   {
     velox::MicrosecondTimer ioTimer(&ioUs);
-    executeIoGroups(ioGroups, buffer);
+    executeIoGroups(ioGroups, buffer, readBytes);
   }
 
   for (const auto& group : ioGroups) {

dwio/nimble/tablet/DataInput.h

@@ -27,10 +27,6 @@
 #include "velox/common/io/Options.h"
 #include "velox/common/memory/Memory.h"
 
-namespace folly {
-class Executor;
-} // namespace folly
-
 namespace facebook::nimble {
 
 /// Base class for data I/O with an enqueue + batch load pattern.
@@ -84,12 +80,11 @@ class DataInput {
   virtual void clear() = 0;
 };
 
-/// Direct data loading with IO coalescing, alignment for direct I/O,
-/// and optional parallel reads via executor.
+/// Direct data loading with IO coalescing and alignment for direct I/O.
 ///
 /// After coalescing, each I/O group is backed by a non-contiguous
-/// Allocation (page-aligned runs). Reads are parallelized across
-/// I/O groups via the executor.
+/// Allocation (page-aligned runs). Reads are submitted through the file's
+/// positioned vector read API.
 ///
 /// NOTE: DirectDataInput is not thread-safe. Each thread must use its
 /// own instance.
@@ -100,15 +95,9 @@ class DirectDataInput : public DataInput {
 
   struct Options {
     velox::memory::MemoryPool* pool{nullptr};
-    folly::Executor* executor{nullptr};
     std::shared_ptr<velox::io::IoStatistics> ioStats;
-    uint64_t alignment{1};
     int32_t maxCoalesceDistance{kDefaultCoalesceDistance};
     int64_t maxCoalesceBytes{velox::io::ReaderOptions::kDefaultCoalesceBytes};
-    /// Minimum number of IO groups per executor task. Batching multiple
-    /// pread calls into one task reduces executor overhead (task timing,
-    /// lambda allocation, scheduling) at the cost of less parallelism.
-    int32_t minIoGroupsPerTask{4};
   };
 
   DirectDataInput(velox::ReadFile* file, const Options& options);
@@ -173,9 +162,11 @@ class DirectDataInput : public DataInput {
   // it.
   std::pair<char*, Handle> allocateBuffer(uint64_t bytes);
 
-  // Executes all physical IO groups, batching executor tasks to reduce
-  // scheduling overhead.
-  void executeIoGroups(std::vector<IoGroup>& ioGroups, char* buffer);
+  // Executes all physical IO groups through ReadFile's positioned vector API.
+  void executeIoGroups(
+      std::vector<IoGroup>& ioGroups,
+      char* buffer,
+      uint64_t bufferSize);
 
   uint64_t alignDown(uint64_t value) const {
     return value & ~(alignment_ - 1);
@@ -189,8 +180,6 @@ class DirectDataInput : public DataInput {
   velox::ReadFile* const file_;
   // Memory pool for allocating the read buffer.
   velox::memory::MemoryPool* const pool_;
-  // Executor for parallel I/O across coalesced groups.
-  folly::Executor* const executor_;
   // IO statistics for tracking read bytes, latency, and gaps.
   const std::shared_ptr<velox::io::IoStatistics> ioStats_;
   // I/O alignment for file offsets and read sizes.
@@ -202,8 +191,6 @@ class DirectDataInput : public DataInput {
   const int32_t maxCoalesceDistance_;
   // Max total bytes per coalesced I/O group.
   const int64_t maxCoalesceBytes_;
-  // Min IO groups per executor task.
-  const int32_t minIoGroupsPerTask_;
 
   State state_{State::kInit};
 

dwio/nimble/tablet/TabletReaderCache.h

@@ -97,8 +97,8 @@ class TabletReaderCache {
 
   /// Returns a cached or newly created CachedTabletReader for the given file.
   /// Uses readFile->getName() as the cache key. On cache miss, creates the
-  /// TabletReader and deserializes the schema using a sharded system pool
-  /// (keyed by filename hash) and the provided readFile/options.
+  /// TabletReader and deserializes the schema using a sharded system pool and
+  /// the provided readFile/options.
   CachedTabletReader get(
       const std::shared_ptr<velox::ReadFile>& readFile,
       const TabletReader::Options& tabletOptions);