feat(native-pos): Add new materialized Exchange path in presto-on-spark native (1/n) (#27573)

Summary:
Pull Request resolved: #27573

#### Summary

Current ShuffleWrite path does single threaded row-by-row processing to maintain thread safety for the shuffle writer (cosco internally at Meta)
But the shuffleWriter is actually thread safe if it is accessed partition exclusive manner. i.e. multi threads can write to it but only
as long as those threads are writing data for disjoint set of partitions.

We can exploit this by just creating a per partition buffer before the shuffle write operator and write multi threaded into this buffer with locks per partition. Everytime we write into a partition buffer, if it is full, we also flush it in the same operation. Also, now that we have batched many rows in our buffer, we write the whole batch as 1 row from shuffle writer POV, which reduces the overhead of per row operations inside the shuffle writer (checksumming, memcpy/buffering)

To accomplish this, instead of updating existing operators, we add new operators as below

- `ExchangeWrite+ExchangeOutputBuffer`
- `ExchangeRead` operators — a new shuffle write/read path

these replaces `PartitionAndSerialize + LocalPartition + ShuffleWrite` with a single operator

### Current architecture

{F1988112256}

### New architecture

 {F1988112251}

#### New classes (all in presto_cpp/main/operators):

- `ExchangeOutputBuffer`: shared per-partition queue buffer owning ShuffleWriter,
  ContinueFuture backpressure, error propagation on close/abort
- `ExchangeWriteNode` + `ExchangeWrite`: Buffers all input data into single contiguous buffer then writes out per partition `enqueue` calls into the ExchangeOutputBuffer
- `ExchangeReadNode` + `ExchangeRead`: reads kFormatBatched pages only
- Translators for both operators

### Configs

- `exchange.materialization.enabled=true`
- `exchange.materialization.partitioning-row-batch-buffer-size=16mb`
- `exchange.materialization.per-partition-buffer-size=130kb`

### Planner wiring

`exchange.materialization.enabled=true` flag gates ExchangeWriteNode (write) and
ExchangeReadNode (read) creation from PartitionedOutputNode/RemoteSourceNode.

Uses allPeersFinished for writer close — no setNumDrivers needed.

Differential Revision: D100365767

Author: shrinidhijoshi

presto-native-execution/presto_cpp/main/PrestoServer.cpp

@@ -39,6 +39,8 @@
 #include "presto_cpp/main/http/filters/StatsFilter.h"
 #include "presto_cpp/main/operators/BroadcastExchangeSource.h"
 #include "presto_cpp/main/operators/BroadcastWrite.h"
+#include "presto_cpp/main/operators/ExchangeRead.h"
+#include "presto_cpp/main/operators/ExchangeWrite.h"
 #include "presto_cpp/main/operators/LocalShuffle.h"
 #include "presto_cpp/main/operators/PartitionAndSerialize.h"
 #include "presto_cpp/main/operators/ShuffleExchangeSource.h"
@@ -1475,6 +1477,10 @@ void PrestoServer::registerCustomOperators() {
       std::make_unique<operators::ShuffleWriteTranslator>());
   velox::exec::Operator::registerOperator(
       std::make_unique<operators::ShuffleReadTranslator>());
+  velox::exec::Operator::registerOperator(
+      std::make_unique<operators::ExchangeWriteTranslator>());
+  velox::exec::Operator::registerOperator(
+      std::make_unique<operators::ExchangeReadTranslator>());
 
   // Todo - Split Presto & Presto-on-Spark server into different classes
   // which will allow server specific operator registration.

presto-native-execution/presto_cpp/main/common/Configs.cpp

@@ -233,6 +233,11 @@ SystemConfig::SystemConfig() {
           BOOL_PROP(kEnableVeloxExprSetLogging, false),
           NUM_PROP(kLocalShuffleMaxPartitionBytes, 268435456),
           STR_PROP(kShuffleName, ""),
+          BOOL_PROP(kExchangeMaterializationEnabled, false),
+          NUM_PROP(
+              kExchangeMaterializationPartitioningRowBatchBufferSize,
+              16L << 20),
+          NUM_PROP(kExchangeMaterializationPerPartitionBufferSize, 130L * 1024),
           STR_PROP(kRemoteFunctionServerCatalogName, ""),
           STR_PROP(kRemoteFunctionServerSerde, "presto_page"),
           BOOL_PROP(kHttpEnableAccessLog, false),
@@ -762,6 +767,24 @@ std::string SystemConfig::shuffleName() const {
   return optionalProperty(kShuffleName).value();
 }
 
+bool SystemConfig::exchangeMaterializationEnabled() const {
+  return optionalProperty<bool>(kExchangeMaterializationEnabled)
+      .value_or(false);
+}
+
+int64_t SystemConfig::exchangeMaterializationPartitioningRowBatchBufferSize()
+    const {
+  return optionalProperty<int64_t>(
+             kExchangeMaterializationPartitioningRowBatchBufferSize)
+      .value_or(16L << 20);
+}
+
+int64_t SystemConfig::exchangeMaterializationPerPartitionBufferSize() const {
+  return optionalProperty<int64_t>(
+             kExchangeMaterializationPerPartitionBufferSize)
+      .value_or(130L * 1024);
+}
+
 bool SystemConfig::enableSerializedPageChecksum() const {
   return optionalProperty<bool>(kEnableSerializedPageChecksum).value();
 }

presto-native-execution/presto_cpp/main/common/Configs.h

@@ -662,6 +662,27 @@ class SystemConfig : public ConfigBase {
   static constexpr std::string_view kLocalShuffleMaxPartitionBytes{
       "shuffle.local.max-partition-bytes"};
   static constexpr std::string_view kShuffleName{"shuffle.name"};
+
+  /// Enable materialized exchange I/O (ExchangeWrite/ExchangeRead operators).
+  /// When false, falls back to PnS + LocalPartition + ShuffleWrite.
+  /// Default: false.
+  static constexpr std::string_view kExchangeMaterializationEnabled{
+      "exchange.materialization.enabled"};
+
+  /// ExchangeWrite flat buffer flush threshold in bytes. Controls how much
+  /// serialized CompactRow data accumulates per driver before flushing to
+  /// the ExchangeOutputBuffer. Default: 16MB.
+  static constexpr std::string_view
+      kExchangeMaterializationPartitioningRowBatchBufferSize{
+          "exchange.materialization.partitioning-row-batch-buffer-size"};
+
+  /// ExchangeOutputBuffer per-partition drain threshold in bytes. When a
+  /// partition accumulates this much data, it is drained to the writer.
+  /// Default: 130KB.
+  static constexpr std::string_view
+      kExchangeMaterializationPerPartitionBufferSize{
+          "exchange.materialization.per-partition-buffer-size"};
+
   static constexpr std::string_view kHttpEnableAccessLog{
       "http-server.enable-access-log"};
   static constexpr std::string_view kHttpEnableStatsFilter{
@@ -1115,6 +1136,12 @@ class SystemConfig : public ConfigBase {
 
   std::string shuffleName() const;
 
+  bool exchangeMaterializationEnabled() const;
+
+  int64_t exchangeMaterializationPartitioningRowBatchBufferSize() const;
+
+  int64_t exchangeMaterializationPerPartitionBufferSize() const;
+
   bool enableSerializedPageChecksum() const;
 
   bool enableVeloxTaskLogging() const;

presto-native-execution/presto_cpp/main/operators/CMakeLists.txt

@@ -13,15 +13,18 @@
 # limitations under the License.
 add_library(
   presto_operators
+  BinarySortableSerializer.cpp
+  BroadcastExchangeSource.cpp
+  BroadcastFile.cpp
+  BroadcastWrite.cpp
+  ExchangeOutputBuffer.cpp
+  ExchangeRead.cpp
+  ExchangeWrite.cpp
+  LocalShuffle.cpp
   PartitionAndSerialize.cpp
   ShuffleExchangeSource.cpp
   ShuffleRead.cpp
   ShuffleWrite.cpp
-  LocalShuffle.cpp
-  BroadcastWrite.cpp
-  BroadcastFile.cpp
-  BroadcastExchangeSource.cpp
-  BinarySortableSerializer.cpp
 )
 
 target_link_libraries(