Fix IOStats for Nimble (facebookincubator#10216)

Summary:
Pull Request resolved: facebookincubator#10216

IOStats are being calculated in different layers of the IO stacks.
Since Nimble and DWRF don't share parts of the stack, some IOStats calculation were not affecting Nimble.

Probably the right thing to do is to move all IOStats calculations to the bottom layers (WSFile, cache and SSD reads), where IO is actually performed (and these layers are shared beteen Nimble nad DWRF).
But it seems like that for this change, we need a design, clarifying what we actually want to track and how to track it.

Since we don't have the cycles to create this design right now, I opted for a simple solution, where I create a simple layer on the Nimble side, which will calculate these stats.

Reviewed By: Yuhta

Differential Revision: D58559606

velox/common/file/File.cpp

@@ -61,17 +61,21 @@ uint64_t ReadFile::preadv(
   return numRead;
 }
 
-void ReadFile::preadv(
+uint64_t ReadFile::preadv(
     folly::Range<const common::Region*> regions,
     folly::Range<folly::IOBuf*> iobufs) const {
   VELOX_CHECK_EQ(regions.size(), iobufs.size());
+  uint64_t length = 0;
   for (size_t i = 0; i < regions.size(); ++i) {
     const auto& region = regions[i];
     auto& output = iobufs[i];
     output = folly::IOBuf(folly::IOBuf::CREATE, region.length);
     pread(region.offset, region.length, output.writableData());
     output.append(region.length);
+    length += region.length;
   }
+
+  return length;
 }
 
 std::string_view

velox/common/file/File.h

@@ -74,9 +74,11 @@ class ReadFile {
   // array must be pre-allocated by the caller, with the same size as `regions`,
   // but don't need to be initialized, since each iobuf will be copy-constructed
   // by the preadv.
+  // Returns the total number of bytes read, which might be different than the
+  // sum of all buffer sizes (for example, if coalescing was used).
   //
   // This method should be thread safe.
-  virtual void preadv(
+  virtual uint64_t preadv(
       folly::Range<const common::Region*> regions,
       folly::Range<folly::IOBuf*> iobufs) const;
 

velox/dwio/common/tests/TestBufferedInput.cpp

@@ -41,7 +41,7 @@ class ReadFileMock : public ::facebook::velox::ReadFile {
   MOCK_METHOD(std::string, getName, (), (const, override));
   MOCK_METHOD(uint64_t, getNaturalReadSize, (), (const, override));
   MOCK_METHOD(
-      void,
+      uint64_t,
       preadv,
       (folly::Range<const Region*> regions, folly::Range<folly::IOBuf*> iobufs),
       (const, override));
@@ -74,26 +74,31 @@ void expectPreadvs(
   EXPECT_CALL(file, size()).WillRepeatedly(Return(content.size()));
   EXPECT_CALL(file, preadv(_, _))
       .Times(1)
-      .WillOnce([content, reads](
-                    folly::Range<const Region*> regions,
-                    folly::Range<folly::IOBuf*> iobufs) {
-        ASSERT_EQ(regions.size(), reads.size());
-        for (size_t i = 0; i < reads.size(); ++i) {
-          const auto& region = regions[i];
-          const auto& read = reads[i];
-          auto& iobuf = iobufs[i];
-          ASSERT_EQ(region.offset, read.offset);
-          ASSERT_EQ(region.length, read.length);
-          if (!read.label.empty()) {
-            EXPECT_EQ(read.label, region.label);
-          }
-          ASSERT_LE(region.offset + region.length, content.size());
-          iobuf = folly::IOBuf(
-              folly::IOBuf::COPY_BUFFER,
-              content.data() + region.offset,
-              region.length);
-        }
-      });
+      .WillOnce(
+          [content, reads](
+              folly::Range<const Region*> regions,
+              folly::Range<folly::IOBuf*> iobufs) -> uint64_t {
+            EXPECT_EQ(regions.size(), reads.size());
+            uint64_t length = 0;
+            for (size_t i = 0; i < reads.size(); ++i) {
+              const auto& region = regions[i];
+              const auto& read = reads[i];
+              auto& iobuf = iobufs[i];
+              length += region.length;
+              EXPECT_EQ(region.offset, read.offset);
+              EXPECT_EQ(region.length, read.length);
+              if (!read.label.empty()) {
+                EXPECT_EQ(read.label, region.label);
+              }
+              EXPECT_LE(region.offset + region.length, content.size());
+              iobuf = folly::IOBuf(
+                  folly::IOBuf::COPY_BUFFER,
+                  content.data() + region.offset,
+                  region.length);
+            }
+
+            return length;
+          });
 }
 
 std::optional<std::string> getNext(SeekableInputStream& input) {