Use same max-output-buffer-size config for all types of output buffers (

#8426)

Summary:
We need 2 configuration settings to control the behavior of PartitionedOutput
operator.

First, a limit on how many bytes to buffer inside the PartitionedOutput operator
itself. That limit is split N ways (where N is number of partitions) to compute
per-destination packet size. A packet here is a SerializedPage.

Second, a limit on how many bytes to buffer in OutputBuffer across all
destinations (a sum of all ready packets / SerializedPages). This limit needs
to be significantly higher than the first limit.

We have introduced the second limit earlier to use for arbitrary buffer, but it
should be used for all kinds of buffers (partitioned, broadcast and
arbitrary).

This change is to start using this second limit for all kinds of buffers. A
series of follow-up changes needs to rename the configuration property to
remove 'arbitrary' from the name in a backwards compatible way
(max_arbitrary_buffer_size -> max_buffer_size).

Pull Request resolved: #8426

Reviewed By: spershin

Differential Revision: D52857143

Pulled By: mbasmanova

fbshipit-source-id: cf790e48d02652b45f2c46bc683ba0301dba323d

velox/core/QueryConfig.h

@@ -156,6 +156,7 @@ class QueryConfig {
   static constexpr const char* kMaxPartitionedOutputBufferSize =
       "max_page_partitioning_buffer_size";
 
+  /// TODO Rename configuration property to remove 'arbitrary' from the name.
   static constexpr const char* kMaxArbitraryBufferSize =
       "max_arbitrary_buffer_size";
 
@@ -445,9 +446,7 @@ class QueryConfig {
     return get<uint64_t>(kMaxPartitionedOutputBufferSize, kDefault);
   }
 
-  /// Returns the maximum size in bytes for the task's buffered output when
-  /// output is distributed randomly among consumers. See
-  /// PartitionedOutputNode::Kind::kArbitrary.
+  /// Returns the maximum size in bytes for the task's buffered output.
   uint64_t maxArbitraryBufferSize() const {
     static constexpr uint64_t kDefault = 32UL << 20;
     return get<uint64_t>(kMaxArbitraryBufferSize, kDefault);

velox/exec/OutputBuffer.cpp

@@ -261,16 +261,6 @@ void releaseAfterAcknowledge(
   }
 }
 
-uint64_t maxBufferSize(
-    const core::QueryConfig& config,
-    PartitionedOutputNode::Kind bufferKind) {
-  if (bufferKind == PartitionedOutputNode::Kind::kArbitrary) {
-    return config.maxArbitraryBufferSize();
-  }
-
-  return config.maxPartitionedOutputBufferSize();
-}
-
 } // namespace
 
 OutputBuffer::OutputBuffer(
@@ -280,7 +270,7 @@ OutputBuffer::OutputBuffer(
     uint32_t numDrivers)
     : task_(std::move(task)),
       kind_(kind),
-      maxSize_(maxBufferSize(task_->queryCtx()->queryConfig(), kind)),
+      maxSize_(task_->queryCtx()->queryConfig().maxArbitraryBufferSize()),
       continueSize_((maxSize_ * kContinuePct) / 100),
       arbitraryBuffer_(
           isArbitrary() ? std::make_unique<ArbitraryBuffer>() : nullptr),