[Bench] Reduce input cols number for fetch

To estimate output size of result query to allocate buffer we are
running `count*` before some actual queries. This estimation requires
only query body argumnets without `select` arguments, so this commit
changes input_cols for `count*` query.

Resolves: #574

Signed-off-by: Dmitrii Makarenko <[email protected]>

Author: Devjiu

omniscidb/QueryEngine/CardinalityEstimator.cpp

@@ -18,6 +18,7 @@
 #include "ErrorHandling.h"
 #include "ExpressionRewrite.h"
 #include "RelAlgExecutor.h"
+#include "Visitors/InputColumnsCollector.h"
 
 size_t RelAlgExecutor::getNDVEstimation(const WorkUnit& work_unit,
                                         const int64_t range,
@@ -101,8 +102,53 @@ RelAlgExecutionUnit create_ndv_execution_unit(const RelAlgExecutionUnit& ra_exe_
 RelAlgExecutionUnit create_count_all_execution_unit(
     const RelAlgExecutionUnit& ra_exe_unit,
     hdk::ir::ExprPtr replacement_target) {
+  InputColumnsCollector input_columns_collector;
+  for (const auto& join_condition : ra_exe_unit.join_quals) {
+    for (const auto& q : join_condition.quals) {
+      input_columns_collector.visit(q.get());
+    }
+  }
+
+  for (const auto& q : ra_exe_unit.quals) {
+    input_columns_collector.visit(q.get());
+  }
+
+  for (const auto& q : ra_exe_unit.simple_quals) {
+    input_columns_collector.visit(q.get());
+  }
+
+  for (const auto& q : ra_exe_unit.groupby_exprs) {
+    if (!q)
+      break;
+    input_columns_collector.visit(q.get());
+  }
+
+  std::list<std::shared_ptr<const InputColDescriptor>> join_input_col_descs;
+  auto& input_column_descriptors = input_columns_collector.result();
+  for (auto& col_var : input_column_descriptors) {
+    for (auto& icol : ra_exe_unit.input_col_descs) {
+      if (icol->getColId() == col_var.getColId() &&
+          icol->getTableId() == col_var.getTableId()) {
+        join_input_col_descs.emplace_back(icol);
+      }
+    }
+  }
+
+  if (join_input_col_descs.empty()) {
+    join_input_col_descs.insert(join_input_col_descs.end(),
+                                ra_exe_unit.input_col_descs.begin(),
+                                ra_exe_unit.input_col_descs.end());
+  }
+
+  std::stringstream os{};
+  os << "\n\t[Only Join] Table/Col/Levels: ";
+  for (const auto& input_col_desc : join_input_col_descs) {
+    os << "(" << input_col_desc->getTableId() << ", " << input_col_desc->getColId()
+       << ", " << input_col_desc->getNestLevel() << ") ";
+  }
+  LOG(WARNING) << "Generating count* WU with following Input Columns" << os.str();
   return {ra_exe_unit.input_descs,
-          ra_exe_unit.input_col_descs,
+          join_input_col_descs,
           ra_exe_unit.simple_quals,
           ra_exe_unit.quals,
           ra_exe_unit.join_quals,

omniscidb/QueryEngine/Descriptors/QueryFragmentDescriptor.cpp

@@ -151,7 +151,9 @@ void QueryFragmentDescriptor::buildFragmentPerKernelForTable(
     }
 
     ExecutionKernelDescriptor execution_kernel_desc{
-        device_id, {}, fragment.getNumTuples()};
+        /*device_id*/ device_id,
+        /*fragments*/ {},
+        /*outer_tuple_count*/ fragment.getNumTuples()};
     if (table_desc_offset) {
       const auto frag_ids =
           executor->getTableFragmentIndices(ra_exe_unit,

omniscidb/QueryEngine/JoinFilterPushDown.cpp

@@ -18,6 +18,7 @@
 #include "IR/ExprCollector.h"
 #include "IR/ExprRewriter.h"
 #include "RelAlgExecutor.h"
+#include "Visitors/InputColumnsCollector.h"
 
 namespace {
 
@@ -27,15 +28,6 @@ class BindFilterToOutermostVisitor : public hdk::ir::ExprRewriter {
   }
 };
 
-class InputColumnsCollector
-    : public hdk::ir::ExprCollector<std::unordered_set<InputColDescriptor>,
-                                    InputColumnsCollector> {
- protected:
-  void visitColumnVar(const hdk::ir::ColumnVar* col_var) override {
-    result_.insert(InputColDescriptor(col_var->columnInfo(), 0));
-  }
-};
-
 }  // namespace
 
 /**

omniscidb/QueryEngine/Visitors/InputColumnsCollector.h

@@ -0,0 +1,13 @@
+#pragma once
+
+#include <unordered_set>
+#include "IR/ExprCollector.h"
+
+class InputColumnsCollector
+    : public hdk::ir::ExprCollector<std::unordered_set<InputColDescriptor>,
+                                    InputColumnsCollector> {
+ protected:
+  void visitColumnVar(const hdk::ir::ColumnVar* col_var) override {
+    result_.insert(InputColDescriptor(col_var->columnInfo(), 0));
+  }
+};