diff --git a/velox/dwio/common/DirectBufferedInput.cpp b/velox/dwio/common/DirectBufferedInput.cpp
index 3accdaf075a8..caf667858080 100644
--- a/velox/dwio/common/DirectBufferedInput.cpp
+++ b/velox/dwio/common/DirectBufferedInput.cpp
@@ -221,6 +221,23 @@ std::shared_ptr<DirectCoalescedLoad> DirectBufferedInput::coalescedLoad(
       });
 }
 
+std::unique_ptr<SeekableInputStream> DirectBufferedInput::read(
+    uint64_t offset,
+    uint64_t length,
+    LogType /*logType*/) const {
+  VELOX_CHECK_LE(offset + length, fileSize_);
+  return std::make_unique<DirectInputStream>(
+      const_cast<DirectBufferedInput*>(this),
+      ioStats_.get(),
+      Region{offset, length},
+      input_,
+      fileNum_,
+      nullptr,
+      TrackingId(),
+      0,
+      options_.loadQuantum());
+}
+
 namespace {
 void appendRanges(
     memory::Allocation& allocation,
diff --git a/velox/dwio/common/DirectBufferedInput.h b/velox/dwio/common/DirectBufferedInput.h
index 51a8bf65ff8d..cf81d9744d47 100644
--- a/velox/dwio/common/DirectBufferedInput.h
+++ b/velox/dwio/common/DirectBufferedInput.h
@@ -172,6 +172,9 @@ class DirectBufferedInput : public BufferedInput {
   std::shared_ptr<DirectCoalescedLoad> coalescedLoad(
       const SeekableInputStream* stream);
 
+  std::unique_ptr<SeekableInputStream>
+  read(uint64_t offset, uint64_t length, LogType logType) const override;
+
   folly::Executor* executor() const override {
     return executor_;
   }
diff --git a/velox/exec/tests/TableScanTest.cpp b/velox/exec/tests/TableScanTest.cpp
index 2370db795a01..6ecfa3d71789 100644
--- a/velox/exec/tests/TableScanTest.cpp
+++ b/velox/exec/tests/TableScanTest.cpp
@@ -314,7 +314,7 @@ TEST_F(TableScanTest, directBufferInputRawInputBytes) {
   ASSERT_TRUE(it != planStats.end());
   auto rawInputBytes = it->second.rawInputBytes;
   auto overreadBytes = getTableScanRuntimeStats(task).at("overreadBytes").sum;
-  ASSERT_EQ(rawInputBytes, 26);
+  ASSERT_GE(rawInputBytes, 500);
   ASSERT_EQ(overreadBytes, 13);
   ASSERT_EQ(
       getTableScanRuntimeStats(task).at("storageReadBytes").sum,