Move reduction code from ResultSetStorage.

Signed-off-by: ienkovich <[email protected]>

Author: ienkovich

omniscidb/QueryEngine/Execute.cpp

@@ -59,6 +59,7 @@
 #include "QueryEngine/OutputBufferInitialization.h"
 #include "QueryEngine/QueryRewrite.h"
 #include "QueryEngine/QueryTemplateGenerator.h"
+#include "QueryEngine/ResultSetReduction.h"
 #include "QueryEngine/ResultSetReductionJIT.h"
 #include "QueryEngine/RuntimeFunctions.h"
 #include "QueryEngine/SpeculativeTopN.h"
@@ -1250,12 +1251,18 @@ ResultSetPtr Executor::reduceMultiDeviceResultSets(
     reduced_results->initializeStorage();
     switch (query_mem_desc.getEffectiveKeyWidth()) {
       case 4:
-        first->getStorage()->moveEntriesToBuffer<int32_t>(
-            result_storage->getUnderlyingBuffer(), query_mem_desc.getEntryCount());
+        ResultSetReduction::moveEntriesToBuffer<int32_t>(
+            first->getStorage()->getQueryMemDesc(),
+            first->getStorage()->getUnderlyingBuffer(),
+            result_storage->getUnderlyingBuffer(),
+            query_mem_desc.getEntryCount());
         break;
       case 8:
-        first->getStorage()->moveEntriesToBuffer<int64_t>(
-            result_storage->getUnderlyingBuffer(), query_mem_desc.getEntryCount());
+        ResultSetReduction::moveEntriesToBuffer<int64_t>(
+            first->getStorage()->getQueryMemDesc(),
+            first->getStorage()->getUnderlyingBuffer(),
+            result_storage->getUnderlyingBuffer(),
+            query_mem_desc.getEntryCount());
         break;
       default:
         CHECK(false);
@@ -1278,10 +1285,12 @@ ResultSetPtr Executor::reduceMultiDeviceResultSets(
         (ResultSetStorage*)nullptr,
         [&](auto r, ResultSetStorage* res) {
           for (auto i = r.begin() + 1; i != r.end(); ++i) {
-            (*r.begin())->reduce(**i, {}, reduction_code, getConfig(), this);
+            ResultSetReduction::reduce(
+                **r.begin(), **i, {}, reduction_code, getConfig(), this);
           }
           if (res) {
-            res->reduce(*(*r.begin()), {}, reduction_code, getConfig(), this);
+            ResultSetReduction::reduce(
+                *res, *(*r.begin()), {}, reduction_code, getConfig(), this);
             return res;
           }
           return *r.begin();
@@ -1293,16 +1302,17 @@ ResultSetPtr Executor::reduceMultiDeviceResultSets(
           if (!rhs) {
             return lhs;
           }
-          lhs->reduce(*rhs, {}, reduction_code, getConfig(), this);
+          ResultSetReduction::reduce(*lhs, *rhs, {}, reduction_code, getConfig(), this);
           return lhs;
         });
   } else {
     for (size_t i = 1; i < results_per_device.size(); ++i) {
-      reduced_results->getStorage()->reduce(*(results_per_device[i].first->getStorage()),
-                                            {},
-                                            reduction_code,
-                                            getConfig(),
-                                            this);
+      ResultSetReduction::reduce(*reduced_results->getStorage(),
+                                 *(results_per_device[i].first->getStorage()),
+                                 {},
+                                 reduction_code,
+                                 getConfig(),
+                                 this);
     }
   }
   reduced_results->addCompilationQueueTime(compilation_queue_time);

omniscidb/QueryEngine/QueryExecutionContext.cpp

@@ -25,6 +25,7 @@
 #include "QueryMemoryInitializer.h"
 #include "RelAlgExecutionUnit.h"
 #include "ResultSet.h"
+#include "ResultSetReduction.h"
 #include "Shared/likely.h"
 #include "SpeculativeTopN.h"
 #include "StreamingTopN.h"
@@ -143,14 +144,14 @@ ResultSetPtr QueryExecutionContext::groupBufferToDeinterleavedResults(
     memcpy(&agg_vals[0],
            &executor_->plan_state_->init_agg_vals_[0],
            agg_col_count * sizeof(agg_vals[0]));
-    ResultSetStorage::reduceSingleRow(rows_ptr + bin_base_off,
-                                      executor_->warpSize(),
-                                      false,
-                                      true,
-                                      agg_vals,
-                                      query_mem_desc_,
-                                      result_set->getTargetInfos(),
-                                      executor_->plan_state_->init_agg_vals_);
+    ResultSetReduction::reduceSingleRow(rows_ptr + bin_base_off,
+                                        executor_->warpSize(),
+                                        false,
+                                        true,
+                                        agg_vals,
+                                        query_mem_desc_,
+                                        result_set->getTargetInfos(),
+                                        executor_->plan_state_->init_agg_vals_);
     for (size_t agg_idx = 0; agg_idx < agg_col_count;
          ++agg_idx, ++deinterleaved_buffer_idx) {
       deinterleaved_buffer[deinterleaved_buffer_idx] = agg_vals[agg_idx];