misc: Add cacheable plumbing to FileIoContext (facebookincubator#16495)

Summary:
Pull Request resolved: facebookincubator#16495

Rename cacheable and plumb the cacheable parameter through the velox layer (FileIoContext, BufferedInput, InputStream, CachedBufferedInput) so it's available to pass to the WS layer in a future diff.

This diff only adds the parameter plumbing without actually passing it to WS pread calls.

Reviewed By: xiaoxmeng

Differential Revision: D93788632

fbshipit-source-id: 740b32e5c9a7ca9e395bcaf16dd98a8c8db23cc9

Author: Ke Wang

velox/common/file/File.h

@@ -77,15 +77,22 @@ struct FileIoContext {
   /// Tracer for IO operations, providing call stack context.
   std::shared_ptr<FileIoTracer> ioTracer;
 
+  /// When false, hints to the storage layer that this read should not be cached
+  /// or should be evicted soon after reading. This is useful for one-time reads
+  /// where caching would waste resources.
+  bool cacheable{true};
+
   FileIoContext() = default;
 
   explicit FileIoContext(
       IoStats* stats,
       folly::F14FastMap<std::string, std::string> fileOpts = {},
-      std::shared_ptr<FileIoTracer> tracer = nullptr)
+      std::shared_ptr<FileIoTracer> tracer = nullptr,
+      bool cacheable = true)
       : ioStats(stats),
         fileOpts(std::move(fileOpts)),
-        ioTracer(std::move(tracer)) {}
+        ioTracer(std::move(tracer)),
+        cacheable(cacheable) {}
 };
 
 // A read-only file.  All methods in this object should be thread safe.

velox/common/io/Options.h

@@ -139,12 +139,12 @@ class ReaderOptions {
     return prefetchRowGroups_;
   }
 
-  bool noCacheRetention() const {
-    return noCacheRetention_;
+  bool cacheable() const {
+    return cacheable_;
   }
 
-  void setNoCacheRetention(bool noCacheRetention) {
-    noCacheRetention_ = noCacheRetention;
+  void setCacheable(bool cacheable) {
+    cacheable_ = cacheable;
   }
 
  protected:
@@ -155,6 +155,6 @@ class ReaderOptions {
   int32_t maxCoalesceDistance_{kDefaultCoalesceDistance};
   int64_t maxCoalesceBytes_{kDefaultCoalesceBytes};
   int32_t prefetchRowGroups_{kDefaultPrefetchRowGroups};
-  bool noCacheRetention_{false};
+  bool cacheable_{true};
 };
 } // namespace facebook::velox::io

velox/connectors/hive/HiveConnectorUtil.cpp

@@ -628,7 +628,7 @@ void configureReaderOptions(
   readerOptions.setFooterEstimatedSize(hiveConfig->footerEstimatedSize());
   readerOptions.setFilePreloadThreshold(hiveConfig->filePreloadThreshold());
   readerOptions.setPrefetchRowGroups(hiveConfig->prefetchRowGroups());
-  readerOptions.setNoCacheRetention(!hiveSplit->cacheable);
+  readerOptions.setCacheable(hiveSplit->cacheable);
   const auto& sessionTzName = connectorQueryCtx->sessionTimezone();
   if (!sessionTzName.empty()) {
     const auto timezone = tz::locateZone(sessionTzName);

velox/connectors/hive/tests/HiveConnectorUtilTest.cpp

@@ -293,15 +293,15 @@ TEST_F(HiveConnectorUtilTest, configureReaderOptions) {
 TEST_F(HiveConnectorUtilTest, cacheRetention) {
   struct {
     bool splitCacheable;
-    bool expectedNoCacheRetention;
+    bool expectedCacheable;
 
     std::string debugString() const {
       return fmt::format(
-          "splitCacheable {}, expectedNoCacheRetention {}",
+          "splitCacheable {}, expectedCacheable {}",
           splitCacheable,
-          expectedNoCacheRetention);
+          expectedCacheable);
     }
-  } testSettings[] = {{false, true}, {true, false}};
+  } testSettings[] = {{false, false}, {true, true}};
 
   for (const auto& testData : testSettings) {
     SCOPED_TRACE(testData.debugString());
@@ -358,8 +358,7 @@ TEST_F(HiveConnectorUtilTest, cacheRetention) {
         hiveSplit,
         readerOptions);
 
-    ASSERT_EQ(
-        readerOptions.noCacheRetention(), testData.expectedNoCacheRetention);
+    ASSERT_EQ(readerOptions.cacheable(), testData.expectedCacheable);
   }
 }
 

velox/dwio/common/BufferedInput.h

@@ -38,14 +38,16 @@ class BufferedInput {
       velox::IoStats* ioStats = nullptr,
       uint64_t maxMergeDistance = kMaxMergeDistance,
       std::optional<bool> wsVRLoad = std::nullopt,
-      folly::F14FastMap<std::string, std::string> fileReadOps = {})
+      folly::F14FastMap<std::string, std::string> fileReadOps = {},
+      bool cacheable = true)
       : BufferedInput(
             std::make_shared<ReadFileInputStream>(
                 std::move(readFile),
                 metricsLog,
                 stats,
                 ioStats,
-                std::move(fileReadOps)),
+                std::move(fileReadOps),
+                cacheable),
             pool,
             maxMergeDistance,
             wsVRLoad) {}

velox/dwio/common/CacheInputStream.cpp

@@ -35,14 +35,14 @@ CacheInputStream::CacheInputStream(
     const Region& region,
     std::shared_ptr<ReadFileInputStream> input,
     uint64_t fileNum,
-    bool noCacheRetention,
+    bool cacheable,
     std::shared_ptr<ScanTracker> tracker,
     TrackingId trackingId,
     uint64_t groupId,
     int32_t loadQuantum)
     : bufferedInput_(bufferedInput),
       cache_(bufferedInput_->cache()),
-      noCacheRetention_(noCacheRetention),
+      cacheable_(cacheable),
       region_(region),
       fileNum_(fileNum),
       tracker_(std::move(tracker)),
@@ -58,7 +58,7 @@ CacheInputStream::~CacheInputStream() {
 }
 
 void CacheInputStream::makeCacheEvictable() {
-  if (!noCacheRetention_) {
+  if (cacheable_) {
     return;
   }
   // Walks through the potential prefetch or access cache space of this cache
@@ -256,15 +256,15 @@ void CacheInputStream::loadSync(const Region& region) {
     ioStats_->queryThreadIoLatencyUs().increment(storageReadUs);
     ioStats_->storageReadLatencyUs().increment(storageReadUs);
     ioStats_->incTotalScanTime(storageReadUs * 1'000);
-    entry->setExclusiveToShared(!noCacheRetention_);
+    entry->setExclusiveToShared(cacheable_);
   } while (pin_.empty());
 }
 
 void CacheInputStream::clearCachePin() {
   if (pin_.empty()) {
     return;
   }
-  if (noCacheRetention_) {
+  if (!cacheable_) {
     pin_.checkedEntry()->makeEvictable();
   }
   pin_.clear();
@@ -345,7 +345,7 @@ void CacheInputStream::loadPosition() {
       {
         MicrosecondTimer timer(&loadUs);
         try {
-          if (!load->loadOrFuture(&waitFuture, !noCacheRetention_)) {
+          if (!load->loadOrFuture(&waitFuture, cacheable_)) {
             waitFuture.wait();
           }
         } catch (const std::exception& e) {

velox/dwio/common/CacheInputStream.h

@@ -35,7 +35,7 @@ class CacheInputStream : public SeekableInputStream {
       const velox::common::Region& region,
       std::shared_ptr<ReadFileInputStream> input,
       uint64_t fileNum,
-      bool noCacheRetention,
+      bool cacheable,
       std::shared_ptr<cache::ScanTracker> tracker,
       cache::TrackingId trackingId,
       uint64_t groupId,
@@ -72,7 +72,7 @@ class CacheInputStream : public SeekableInputStream {
         region_,
         input_,
         fileNum_,
-        noCacheRetention_,
+        cacheable_,
         tracker_,
         trackingId_,
         groupId_,
@@ -98,8 +98,8 @@ class CacheInputStream : public SeekableInputStream {
     prefetchPct_ = pct;
   }
 
-  bool testingNoCacheRetention() const {
-    return noCacheRetention_;
+  bool testingCacheable() const {
+    return cacheable_;
   }
 
  private:
@@ -120,7 +120,7 @@ class CacheInputStream : public SeekableInputStream {
       cache::AsyncDataCacheEntry& entry);
 
   // Invoked to clear the cache pin of the accessed cache entry and mark it as
-  // immediate evictable if 'noCacheRetention_' flag is set.
+  // immediate evictable if 'cacheable_' is false.
   void clearCachePin();
 
   void makeCacheEvictable();
@@ -131,10 +131,10 @@ class CacheInputStream : public SeekableInputStream {
 
   CachedBufferedInput* const bufferedInput_;
   cache::AsyncDataCache* const cache_;
-  // True if a pin should be set to the lowest retention score after
+  // False if a pin should be set to the lowest retention score after
   // unpinning. This applies to sequential reads where second access
   // to the page is not expected.
-  const bool noCacheRetention_;
+  const bool cacheable_;
   // The region of 'input' 'this' ranges over.
   const velox::common::Region region_;
   const uint64_t fileNum_;

velox/dwio/common/CachedBufferedInput.cpp

@@ -59,7 +59,7 @@ std::unique_ptr<SeekableInputStream> CachedBufferedInput::enqueue(
       region,
       input_,
       fileNum_.id(),
-      options_.noCacheRetention(),
+      options_.cacheable(),
       tracker_,
       id,
       groupId_.id(),
@@ -520,7 +520,7 @@ void CachedBufferedInput::readRegions(
       auto& load = coalescedLoads_[i];
       if (load->state() == CoalescedLoad::State::kPlanned) {
         executor_->add(
-            [pendingLoad = load, ssdSavable = !options_.noCacheRetention()]() {
+            [pendingLoad = load, ssdSavable = options_.cacheable()]() {
               process::TraceContext trace("Read Ahead");
               pendingLoad->loadOrFuture(nullptr, ssdSavable);
             });
@@ -579,7 +579,7 @@ std::unique_ptr<SeekableInputStream> CachedBufferedInput::read(
       Region{offset, length},
       input_,
       fileNum_.id(),
-      options_.noCacheRetention(),
+      options_.cacheable(),
       nullptr,
       TrackingId(),
       0,

velox/dwio/common/CachedBufferedInput.h

@@ -72,7 +72,8 @@ class CachedBufferedInput : public BufferedInput {
             ioStats.get(),
             kMaxMergeDistance,
             std::nullopt,
-            std::move(fileReadOps)),
+            std::move(fileReadOps),
+            readerOptions.cacheable()),
         cache_(cache),
         fileNum_(std::move(fileNum)),
         tracker_(std::move(tracker)),

velox/dwio/common/InputStream.cpp

@@ -65,9 +65,10 @@ ReadFileInputStream::ReadFileInputStream(
     const MetricsLogPtr& metricsLog,
     IoStatistics* stats,
     velox::IoStats* ioStats,
-    folly::F14FastMap<std::string, std::string> fileOpts)
+    folly::F14FastMap<std::string, std::string> fileOpts,
+    bool cacheable)
     : InputStream(readFile->getName(), metricsLog, stats, ioStats),
-      fileIoContext_(ioStats, std::move(fileOpts)),
+      fileIoContext_(ioStats, std::move(fileOpts), nullptr, cacheable),
       readFile_(std::move(readFile)) {}
 
 void ReadFileInputStream::read(