Correctly record the rawBytesRead_ metric in TableScan (#8147)

Summary:
Based on oerling design:
_rawBytesRead is the quantized size of data read by a query.  Read is the volume copied from storage. The second is 0 if data comes from caches. With DirectBufferedInput  these are close but then read has unused coalesced columns counted where raw read does not. A column can be coalesced into IO but never hit by the query due to filtering._

Currently, `DirectBufferInput` only tracks `rawBytesRead` in the `loadSync` method. It does not track `rawBytesRead` in the asynchronous `loadData` method. This PR adds the `rawBytesRead` metric to the `loadData` method.

Pull Request resolved: #8147

Reviewed By: Yuhta

Differential Revision: D52616884

Pulled By: xiaoxmeng

fbshipit-source-id: f56f72167ff4cd9ab1c1eb083791d7453de55e98

velox/dwio/common/CacheInputStream.cpp

@@ -314,6 +314,9 @@ void CacheInputStream::loadPosition() {
     // 'region_'
     loadRegion.length = std::min<int32_t>(
         loadQuantum_, region_.length - (loadRegion.offset - region_.offset));
+
+    // There is no need to update the metric in the loadData method because
+    // loadSync is always executed regardless and updates the metric.
     loadSync(loadRegion);
   }
   auto* entry = pin_.checkedEntry();

velox/dwio/common/DirectBufferedInput.cpp

@@ -274,6 +274,7 @@ std::vector<cache::CachePin> DirectCoalescedLoad::loadData(bool isPrefetch) {
   }
   input_->read(buffers, requests_[0].region.offset, LogType::FILE);
   ioStats_->read().increment(size);
+  ioStats_->incRawBytesRead(size - overread);
   ioStats_->incRawOverreadBytes(overread);
   if (isPrefetch) {
     ioStats_->prefetch().increment(size);

velox/dwio/common/DirectInputStream.cpp

@@ -190,6 +190,9 @@ void DirectInputStream::loadPosition() {
     loadedRegion_.length = (offsetInRegion_ + loadQuantum_ <= region_.length)
         ? loadQuantum_
         : (region_.length - offsetInRegion_);
+
+    // Since the loadSync method updates the metric, but it is conditionally
+    // executed, we also need to update the metric in the loadData method.
     loadSync();
   }
 

velox/exec/tests/TableScanTest.cpp

@@ -241,7 +241,8 @@ TEST_F(TableScanTest, allColumns) {
   auto scanNodeId = plan->id();
   auto it = planStats.find(scanNodeId);
   ASSERT_TRUE(it != planStats.end());
-  ASSERT_TRUE(it->second.peakMemoryBytes > 0);
+  ASSERT_GT(it->second.peakMemoryBytes, 0);
+  ASSERT_GT(it->second.rawInputBytes, 0);
   EXPECT_LT(0, exec::TableScan::ioWaitNanos());
 }