Add PlanBuilder::partitionedOutputArbitrary (#7060)

Summary:
Create a PartitionedOutputNode to output the input data in arbitrary buffer.

Pull Request resolved: #7060

Reviewed By: xiaoxmeng

Differential Revision: D50308139

Pulled By: kewang1024

fbshipit-source-id: 8cf267c683f863a1def2ba775ac55eb67ce84cf6

velox/exec/tests/utils/PlanBuilder.cpp

@@ -988,6 +988,16 @@ PlanBuilder& PlanBuilder::partitionedOutputBroadcast(
   return *this;
 }
 
+PlanBuilder& PlanBuilder::partitionedOutputArbitrary(
+    const std::vector<std::string>& outputLayout) {
+  auto outputType = outputLayout.empty()
+      ? planNode_->outputType()
+      : extract(planNode_->outputType(), outputLayout);
+  planNode_ = core::PartitionedOutputNode::arbitrary(
+      nextPlanNodeId(), outputType, planNode_);
+  return *this;
+}
+
 PlanBuilder& PlanBuilder::localPartition(
     const std::vector<std::string>& keys,
     const std::vector<core::PlanNodePtr>& sources) {

velox/exec/tests/utils/PlanBuilder.h

@@ -560,7 +560,7 @@ class PlanBuilder {
       core::PartitionFunctionSpecPtr partitionFunctionSpec,
       const std::vector<std::string>& outputLayout = {});
 
-  /// Add a PartitionedOutputNode to broadcast the input data.
+  /// Adds a PartitionedOutputNode to broadcast the input data.
   ///
   /// @param outputLayout Optional output layout in case it is different then
   /// the input. Output columns may appear in different order from the input,
@@ -569,7 +569,11 @@ class PlanBuilder {
   PlanBuilder& partitionedOutputBroadcast(
       const std::vector<std::string>& outputLayout = {});
 
-  /// Add a LocalPartitionNode to hash-partition the input on the specified
+  /// Adds a PartitionedOutputNode to put data into arbitrary buffer.
+  PlanBuilder& partitionedOutputArbitrary(
+      const std::vector<std::string>& outputLayout = {});
+
+  /// Adds a LocalPartitionNode to hash-partition the input on the specified
   /// keys using exec::HashPartitionFunction. Number of partitions is determined
   /// at runtime based on parallelism of the downstream pipeline.
   ///