Try remaking Builders for every output (#11164)

Author: lll-phill-lll

ydb/library/yql/dq/runtime/dq_output_consumer.cpp

@@ -346,9 +346,6 @@ class TDqOutputHashPartitionConsumerBlock : public IDqOutputConsumer {
                 blockTypes.emplace_back(blockType->GetItemType());
             }
         }
-        ui64 maxBlockLen = CalcMaxBlockLength(blockTypes.begin(), blockTypes.end(), helper);
-        YQL_ENSURE(maxBlockLen > 0);
-        MakeBuilders(maxBlockLen);
 
         TBlockTypeHelper blockHelper;
         for (auto& column : KeyColumns_) {
@@ -391,22 +388,14 @@ class TDqOutputHashPartitionConsumerBlock : public IDqOutputConsumer {
             outputBlockIndexes[GetHashPartitionIndex(datums.data(), i)].push_back(i);
         }
 
-        ui64 maxLen = 0;
-        for (auto& indexes : outputBlockIndexes) {
-            maxLen = std::max(maxLen, indexes.size());
-        }
-
-        if (maxLen > MaxOutputBlockLen_) {
-            MakeBuilders(maxLen);
-        }
-
         TVector<std::unique_ptr<TArgsDechunker>> outputData;
         for (size_t i = 0; i < Outputs_.size(); ++i) {
             ui64 outputBlockLen = outputBlockIndexes[i].size();
             if (!outputBlockLen) {
                 outputData.emplace_back();
                 continue;
             }
+            MakeBuilders(outputBlockLen);
             const ui64* indexes = outputBlockIndexes[i].data();
 
             std::vector<arrow::Datum> output;
@@ -419,7 +408,7 @@ class TDqOutputHashPartitionConsumerBlock : public IDqOutputConsumer {
                     dataItem.Data = src->array().get();
                     dataItem.StartOffset = 0;
                     Builders_[j]->AddMany(&dataItem, 1, indexes, outputBlockLen);
-                    output.emplace_back(Builders_[j]->Build(false));
+                    output.emplace_back(Builders_[j]->Build(true));
                 }
             }
             outputData.emplace_back(std::make_unique<TArgsDechunker>(std::move(output)));
@@ -528,7 +517,6 @@ class TDqOutputHashPartitionConsumerBlock : public IDqOutputConsumer {
                 Builders_.emplace_back();
             }
         }
-        MaxOutputBlockLen_ = maxBlockLen;
     }
 
 private:
@@ -546,7 +534,6 @@ class TDqOutputHashPartitionConsumerBlock : public IDqOutputConsumer {
     TVector<std::unique_ptr<IBlockReader>> Readers_;
     TVector<std::unique_ptr<IArrayBuilder>> Builders_;
 
-    ui64 MaxOutputBlockLen_ = 0;
     mutable bool IsWaitingFlag_ = false;
 };