Merge 2798fe7 into 21bdc0e

Author: lll-phill-lll

ydb/library/yql/minikql/comp_nodes/mkql_wide_combine.cpp

@@ -132,6 +132,12 @@ struct TCombinerNodes {
         }
     }
 
+    void ExtractValues(TComputationContext& ctx, NUdf::TUnboxedValue** values, NUdf::TUnboxedValue* keys) const {
+        for (size_t i = 0U, j = 0U; i < ItemNodes.size(); ++i) {
+            keys[i] = *(values[j++]);
+        }
+    }
+
     void ProcessItem(TComputationContext& ctx, NUdf::TUnboxedValue* keys, NUdf::TUnboxedValue* state) const {
         if (keys) {
             std::fill_n(keys, KeyResultNodes.size(), NUdf::TUnboxedValuePod());
@@ -346,6 +352,7 @@ class TSpillingSupportState : public TComputationValue<TSpillingSupportState> {
     enum class ETasteResult: i8 {
         Init = -1,
         Update,
+        DataRequired,
         Skip
     };
     TSpillingSupportState(
@@ -445,9 +452,11 @@ class TSpillingSupportState : public TComputationValue<TSpillingSupportState> {
 
         // Corresponding bucket is spilled, we don't need a key anymore, full input will be spilled
         BufferForKeyAndState.resize(0);
-        TryToSpillRawData(bucket, bucketId);
+        // Prepare space for raw data 
+        BufferForUsedInputItems.resize(ItemNodesSize);
+        Throat = BufferForUsedInputItems.data();
 
-        return ETasteResult::Skip;
+        return ETasteResult::DataRequired;
     }
 
     NUdf::TUnboxedValuePod* Extract() {
@@ -1237,6 +1246,7 @@ using TBaseComputation = TStatefulWideFlowCodegeneratorNode<TWideLastCombinerWra
         , AllowSpilling(allowSpilling)
     {}
 
+    // MARK: DoCAlculate
     EFetchResult DoCalculate(NUdf::TUnboxedValue& state, TComputationContext& ctx, NUdf::TUnboxedValue*const* output) const {
         if (!state.HasValue()) {
             MakeState(ctx, state);
@@ -1274,6 +1284,9 @@ using TBaseComputation = TStatefulWideFlowCodegeneratorNode<TWideLastCombinerWra
                         case TSpillingSupportState::ETasteResult::Update:
                             Nodes.ProcessItem(ctx, static_cast<NUdf::TUnboxedValue*>(ptr->Tongue), static_cast<NUdf::TUnboxedValue*>(ptr->Throat));
                             break;
+                        case TSpillingSupportState::ETasteResult::DataRequired:
+                            Nodes.ExtractValues(ctx, fields, static_cast<NUdf::TUnboxedValue*>(ptr->Throat));
+                            break;
                         case TSpillingSupportState::ETasteResult::Skip:
                             break;
                     }