Always copy the wide consumed data by DqOutput (#12373)

Author: abyss7

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

@@ -368,7 +368,7 @@ class TDqOutputHashPartitionConsumerBlock : public IDqOutputConsumer {
         YQL_ENSURE(false, "Consume() called on wide block stream");
     }
 
-    void WideConsume(TUnboxedValue* values, ui32 count) final {
+    void WideConsume(TUnboxedValue values[], ui32 count) final {
         YQL_ENSURE(!IsWaitingFlag_);
         YQL_ENSURE(count == OutputWidth_);
 
@@ -379,7 +379,7 @@ class TDqOutputHashPartitionConsumerBlock : public IDqOutputConsumer {
 
         TVector<const arrow::Datum*> datums;
         datums.reserve(count - 1);
-        for (ui32 i = 0; i + 1 < count; ++i) {
+        for (ui32 i = 0; i < count - 1; ++i) {
             datums.push_back(&TArrowBlock::From(values[i]).GetDatum());
         }
 
@@ -404,9 +404,10 @@ class TDqOutputHashPartitionConsumerBlock : public IDqOutputConsumer {
                 if (src->is_scalar()) {
                     output.emplace_back(*src);
                 } else {
-                    IArrayBuilder::TArrayDataItem dataItem;
-                    dataItem.Data = src->array().get();
-                    dataItem.StartOffset = 0;
+                    IArrayBuilder::TArrayDataItem dataItem {
+                        .Data = src->array().get(),
+                        .StartOffset = 0,
+                    };
                     Builders_[j]->AddMany(&dataItem, 1, indexes, outputBlockLen);
                     output.emplace_back(Builders_[j]->Build(true));
                 }
@@ -418,11 +419,14 @@ class TDqOutputHashPartitionConsumerBlock : public IDqOutputConsumer {
     }
 
     void DoConsume(TVector<std::unique_ptr<TArgsDechunker>>&& outputData) const {
+        Y_ENSURE(outputData.size() == Outputs_.size());
+
         while (!outputData.empty()) {
             bool hasData = false;
             for (size_t i = 0; i < Outputs_.size(); ++i) {
                 if (Outputs_[i]->IsFull()) {
                     IsWaitingFlag_ = true;
+                    Y_ENSURE(OutputData_.empty());
                     OutputData_ = std::move(outputData);
                     return;
                 }
@@ -474,7 +478,7 @@ class TDqOutputHashPartitionConsumerBlock : public IDqOutputConsumer {
         return !IsWaitingFlag_;
     }
 
-    size_t GetHashPartitionIndex(const arrow::Datum** values, ui64 blockIndex) {
+    size_t GetHashPartitionIndex(const arrow::Datum* values[], ui64 blockIndex) {
         ui64 hash = 0;
         for (size_t keyId = 0; keyId < KeyColumns_.size(); keyId++) {
             const ui32 columnIndex = KeyColumns_[keyId].Index;
@@ -512,7 +516,7 @@ class TDqOutputHashPartitionConsumerBlock : public IDqOutputConsumer {
             if (blockType->GetShape() == NMiniKQL::TBlockType::EShape::Many) {
                 auto itemType = blockType->GetItemType();
                 YQL_ENSURE(!itemType->IsPg(), "pg types are not supported yet");
-                Builders_.emplace_back(MakeArrayBuilder(helper, itemType, *NYql::NUdf::GetYqlMemoryPool(), maxBlockLen, nullptr));
+                Builders_.emplace_back(MakeArrayBuilder(helper, itemType, *NYql::NUdf::GetYqlMemoryPool(), maxBlockLen, nullptr, {.MinFillPercentage=100}));
             } else {
                 Builders_.emplace_back();
             }

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

@@ -28,7 +28,7 @@ class IDqOutputConsumer : public TSimpleRefCount<IDqOutputConsumer>,
     }
     virtual bool IsFull() const = 0;
     virtual void Consume(NKikimr::NUdf::TUnboxedValue&& value) = 0;
-    virtual void WideConsume(NKikimr::NUdf::TUnboxedValue* values, ui32 count) = 0;
+    virtual void WideConsume(NKikimr::NUdf::TUnboxedValue values[], ui32 count) = 0;
     virtual void Consume(NDqProto::TCheckpoint&& checkpoint) = 0;
     virtual void Finish() = 0;
     bool IsFinishing() const {