Merge nest loop join build vectors

Summary: When build vectors in nested loop join are small, we should merge them to get better performance.  In some extreme case, the performance difference can be more than 100 times.

Differential Revision: D65450017

velox/exec/NestedLoopJoinBuild.cpp

@@ -70,6 +70,34 @@ BlockingReason NestedLoopJoinBuild::isBlocked(ContinueFuture* future) {
   return BlockingReason::kWaitForJoinBuild;
 }
 
+std::vector<RowVectorPtr> NestedLoopJoinBuild::mergeDataVectors() const {
+  const auto maxBatchRows =
+      operatorCtx_->task()->queryCtx()->queryConfig().maxOutputBatchRows();
+  std::vector<RowVectorPtr> merged;
+  for (int i = 0; i < dataVectors_.size();) {
+    auto batchSize = dataVectors_[i]->size();
+    auto j = i + 1;
+    while (j < dataVectors_.size() &&
+           batchSize + dataVectors_[j]->size() <= maxBatchRows) {
+      batchSize += dataVectors_[j++]->size();
+    }
+    if (j == i + 1) {
+      merged.push_back(dataVectors_[i++]);
+    } else {
+      auto batch = BaseVector::create<RowVector>(
+          dataVectors_[i]->type(), batchSize, pool());
+      batchSize = 0;
+      while (i < j) {
+        auto* source = dataVectors_[i++].get();
+        batch->copy(source, batchSize, 0, source->size());
+        batchSize += source->size();
+      }
+      merged.push_back(std::move(batch));
+    }
+  }
+  return merged;
+}
+
 void NestedLoopJoinBuild::noMoreInput() {
   Operator::noMoreInput();
   std::vector<ContinuePromise> promises;
@@ -105,6 +133,7 @@ void NestedLoopJoinBuild::noMoreInput() {
     }
   }
 
+  dataVectors_ = mergeDataVectors();
   operatorCtx_->task()
       ->getNestedLoopJoinBridge(
           operatorCtx_->driverCtx()->splitGroupId, planNodeId())

velox/exec/NestedLoopJoinBuild.h

@@ -59,6 +59,8 @@ class NestedLoopJoinBuild : public Operator {
   }
 
  private:
+  std::vector<RowVectorPtr> mergeDataVectors() const;
+
   std::vector<RowVectorPtr> dataVectors_;
 
   // Future for synchronizing with other Drivers of the same pipeline. All build