Clean Up Nimble Flush Policy Code (facebookincubator#235)

Summary: Pull Request resolved: facebookincubator#235

Differential Revision: D81514657

Author: macvincent

dwio/nimble/velox/FlushPolicy.cpp

@@ -19,12 +19,8 @@ namespace facebook::nimble {
 
 FlushDecision RawStripeSizeFlushPolicy::shouldFlush(
     const StripeProgress& stripeProgress) {
-  return stripeProgress.rawStripeSize >= rawStripeSize_ ? FlushDecision::Stripe
+  return stripeProgress.stripeRawSize >= rawStripeSize_ ? FlushDecision::Stripe
                                                         : FlushDecision::None;
 }
 
-void RawStripeSizeFlushPolicy::onClose() {
-  // No-op
-}
-
 } // namespace facebook::nimble

dwio/nimble/velox/FlushPolicy.h

@@ -22,11 +22,9 @@ namespace facebook::nimble {
 
 struct StripeProgress {
   // Size of the stripe data when it's fully decompressed and decoded
-  const uint64_t rawStripeSize;
+  const uint64_t stripeRawSize;
   // Size of the stripe after buffered data is encoded and optionally compressed
-  const uint64_t stripeSize;
-  // Size of the allocated buffer in the writer
-  const uint64_t bufferSize;
+  const uint64_t stripeEncodedSize;
 };
 
 enum class FlushDecision : uint8_t {
@@ -39,9 +37,6 @@ class FlushPolicy {
  public:
   virtual ~FlushPolicy() = default;
   virtual FlushDecision shouldFlush(const StripeProgress& stripeProgress) = 0;
-  // Required for memory pressure coordination for now. Will remove in the
-  // future.
-  virtual void onClose() = 0;
 };
 
 class RawStripeSizeFlushPolicy final : public FlushPolicy {
@@ -51,8 +46,6 @@ class RawStripeSizeFlushPolicy final : public FlushPolicy {
 
   FlushDecision shouldFlush(const StripeProgress& stripeProgress) override;
 
-  void onClose() override;
-
  private:
   const uint64_t rawStripeSize_;
 };
@@ -61,16 +54,12 @@ class LambdaFlushPolicy : public FlushPolicy {
  public:
   explicit LambdaFlushPolicy(
       std::function<FlushDecision(const StripeProgress&)> lambda)
-      : lambda_{lambda} {}
+      : lambda_{std::move(lambda)} {}
 
   FlushDecision shouldFlush(const StripeProgress& stripeProgress) override {
     return lambda_(stripeProgress);
   }
 
-  void onClose() override {
-    // No-op
-  }
-
  private:
   std::function<FlushDecision(const StripeProgress&)> lambda_;
 };

dwio/nimble/velox/VeloxWriter.cpp

@@ -33,7 +33,6 @@
 #include "dwio/nimble/velox/SchemaSerialization.h"
 #include "dwio/nimble/velox/SchemaTypes.h"
 #include "dwio/nimble/velox/StatsGenerated.h"
-#include "folly/ScopeGuard.h"
 #include "velox/common/time/CpuWallTimer.h"
 #include "velox/dwio/common/ExecutorBarrier.h"
 #include "velox/type/Type.h"
@@ -551,8 +550,6 @@ void VeloxWriter::close() {
 
   if (file_) {
     try {
-      auto exitGuard =
-          folly::makeGuard([this]() { context_->flushPolicy->onClose(); });
       flush();
       root_->close();
 
@@ -845,11 +842,8 @@ bool VeloxWriter::tryWriteStripe(bool force) {
 
   auto shouldFlush = [&]() {
     return context_->flushPolicy->shouldFlush(StripeProgress{
-        .rawStripeSize = context_->memoryUsed,
-        .stripeSize = context_->stripeSize,
-        .bufferSize =
-            static_cast<uint64_t>(context_->bufferMemoryPool->usedBytes()),
-    });
+        .stripeRawSize = context_->memoryUsed,
+        .stripeEncodedSize = context_->stripeSize});
   };
 
   auto decision = force ? FlushDecision::Stripe : shouldFlush();