[POAE7-2913] Vectorized Filter Support (#414)

* Remove column num function from C2R.

* Add bit operations.

* Support vectorized filter.

* Fix format.

* Fix broken UTs.

Author: bigPYJ1151

cpp/src/cider/exec/nextgen/context/CodegenContext.h

@@ -86,6 +86,12 @@ struct CodegenOptions {
   jitlib::CompilationOptions co = jitlib::CompilationOptions{};
 };
 
+struct FilterDescriptor {
+  bool applied_filter_mask{false};
+  jitlib::JITValuePointer filter_i64_mask{nullptr};
+  jitlib::JITValuePointer start_index{nullptr};
+};
+
 class CodegenContext {
  public:
   CodegenContext() : jit_func_(nullptr) {}
@@ -102,6 +108,33 @@ class CodegenContext {
     jit_module_ = std::move(jit_module);
   }
 
+  void setFilterMask(jitlib::JITValuePointer& mask,
+                     jitlib::JITValuePointer& start_index) {
+    CHECK(mask->getValueTypeTag() == jitlib::JITTypeTag::INT64);
+    CHECK(start_index->getValueTypeTag() == jitlib::JITTypeTag::INT64);
+
+    filter_desc_.applied_filter_mask = true;
+    filter_desc_.filter_i64_mask.replace(mask);
+    filter_desc_.start_index.replace(start_index);
+  }
+
+  bool isAppliedFilterMask() { return filter_desc_.applied_filter_mask; }
+
+  std::pair<jitlib::JITValuePointer&, jitlib::JITValuePointer&> getFilterMask() {
+    return {filter_desc_.filter_i64_mask, filter_desc_.start_index};
+  }
+
+  template <typename FuncT>
+  void appendDeferFunc(FuncT&& func) {
+    defer_func_list_.emplace_back(func);
+  }
+
+  void clearDeferFunc() { defer_func_list_.clear(); }
+
+  const std::vector<std::function<void()>>& getDeferFunc() const {
+    return defer_func_list_;
+  }
+
   void setInputLength(jitlib::JITValuePointer& len) { input_len_.replace(len); }
 
   jitlib::JITValuePointer& getInputLength() { return input_len_; }
@@ -269,7 +302,9 @@ class CodegenContext {
 
   jitlib::JITFunctionPointer jit_func_;
   jitlib::JITModulePointer jit_module_;
-  jitlib::JITValuePointer input_len_;
+  jitlib::JITValuePointer input_len_{nullptr};
+  FilterDescriptor filter_desc_;
+  std::vector<std::function<void()>> defer_func_list_{};
 
   int64_t id_counter_{0};
   CodegenOptions codegen_options_;

cpp/src/cider/exec/nextgen/operators/CMakeLists.txt

@@ -30,6 +30,7 @@ add_opnode(RowToColumnNode)
 add_opnode(HashJoinNode)
 add_opnode(VectorizedProjectNode)
 add_opnode(CrossJoinNode)
+add_opnode(VectorizedFilterNode)
 
 list(APPEND OPERATORS_SOURCE
      ${CMAKE_CURRENT_LIST_DIR}/extractor/AggExtractorBuilder.cpp)

cpp/src/cider/exec/nextgen/operators/ColumnToRowNode.cpp

@@ -229,25 +229,49 @@ void ColumnToRowTranslator::codegenImpl(SuccessorEmitter successor_wrapper,
 
   // for row loop
   auto index = func->createVariable(JITTypeTag::INT64, "index", 0);
-  auto len = context.getInputLength();
-  static_cast<ColumnToRowNode*>(node_.get())->setColumnRowNum(len);
-
   auto builder = func->createLoopBuilder();
-  builder->condition([&index, &len]() { return index < len; })
-      ->loop([&](LoopBuilder*) {
-        for (auto& input : inputs) {
-          ColumnReader(context, input, index).read();
-        }
-        successor_wrapper(successor, context);
-      })
-      ->update([&index]() { index = index + 1l; });
+
+  if (!context.isAppliedFilterMask()) {
+    auto len = context.getInputLength();
+    builder->condition([&index, &len]() { return index < len; })
+        ->loop([&](LoopBuilder*) {
+          for (auto& input : inputs) {
+            ColumnReader(context, input, index).read();
+          }
+          successor_wrapper(successor, context);
+        })
+        ->update([&index]() { index = index + 1l; });
+  } else {
+    auto&& filter_mask = context.getFilterMask();
+    builder->condition([&filter_mask]() { return filter_mask.first != 0; })
+        ->loop([&](LoopBuilder*) {
+          auto offset = func->emitRuntimeFunctionCall(
+              "get_lowest_set_bit",
+              JITFunctionEmitDescriptor{.ret_type = JITTypeTag::INT64,
+                                        .params_vector = {filter_mask.first.get()}});
+          index = *filter_mask.second + offset;
+          for (auto& input : inputs) {
+            ColumnReader(context, input, index).read();
+          }
+          successor_wrapper(successor, context);
+        })
+        ->update([&filter_mask, &func]() {
+          filter_mask.first = func->emitRuntimeFunctionCall(
+              "reset_tail_set_bit",
+              JITFunctionEmitDescriptor{.ret_type = JITTypeTag::INT64,
+                                        .params_vector = {filter_mask.first.get()}});
+        });
+  }
 
   auto c2r_node = static_cast<ColumnToRowNode*>(node_.get());
   builder->setNoAlias(c2r_node->isVectorizable())->build();
 
   // Execute defer build functions.
-  for (auto& defer_func : c2r_node->getDeferFunctions()) {
-    defer_func();
+  if (!context.isAppliedFilterMask()) {
+    for (auto& defer_func : context.getDeferFunc()) {
+      defer_func();
+    }
+    context.clearDeferFunc();
   }
 }
 }  // namespace cider::exec::nextgen::operators

cpp/src/cider/exec/nextgen/operators/ColumnToRowNode.h

@@ -37,22 +37,6 @@ class ColumnToRowNode : public OpNode {
 
   TranslatorPtr toTranslator(const TranslatorPtr& successor = nullptr) override;
 
-  jitlib::JITValuePointer& getColumnRowNum() { return column_row_num_; }
-
-  void setColumnRowNum(jitlib::JITValuePointer& row_num) {
-    CHECK(column_row_num_.get() == nullptr);
-    column_row_num_.replace(row_num);
-  }
-
-  using DeferFunc = void (*)(void*);
-
-  template <typename FuncT>
-  void registerDeferFunc(FuncT&& func) {
-    defer_func_list_.emplace_back(func);
-  }
-
-  std::vector<std::function<void()>>& getDeferFunctions() { return defer_func_list_; }
-
   bool isVectorizable() const { return vectorizable_; }
 
  private:

cpp/src/cider/exec/nextgen/operators/OperatorRuntimeFunctions.h

@@ -19,8 +19,8 @@
  * under the License.
  */
 
-#ifndef NEXTGEN_CIDER_FUNCTION_RUNTIME_FUNCTIONS_H
-#define NEXTGEN_CIDER_FUNCTION_RUNTIME_FUNCTIONS_H
+#ifndef NEXTGEN_OPERATORS_OPERATORRUNTIMEFUNCTIONS_H
+#define NEXTGEN_OPERATORS_OPERATORRUNTIMEFUNCTIONS_H
 
 #include "exec/nextgen/context/RuntimeContext.h"
 #include "type/data/funcannotations.h"
@@ -211,4 +211,20 @@ extern "C" NEVER_INLINE void convert_bool_to_bit(uint8_t* byte,
   CiderBitUtils::byteToBit(byte, bit, len);
 }
 
-#endif  // NEXTGEN_CIDER_FUNCTION_RUNTIME_FUNCTIONS_H
+extern "C" ALWAYS_INLINE size_t get_lowest_set_bit(size_t data) {
+  return CiderBitUtils::countTailZero(data);
+}
+
+extern "C" ALWAYS_INLINE size_t reset_tail_set_bit(size_t data) {
+  return CiderBitUtils::setTailOneToZero(data);
+}
+
+extern "C" ALWAYS_INLINE void reset_tail_bits_64_align(uint8_t* data, size_t len) {
+  uint64_t* data_i64 = (uint64_t*)data;
+  data_i64 += (len >> 6);
+  size_t invalid_bits = 64 - (len & 63);
+  uint64_t mask = ((0xFFFFFFFFFFFFFFFF << invalid_bits) >> invalid_bits);
+  *data_i64 &= mask;
+}
+
+#endif  // NEXTGEN_OPERATORS_OPERATORRUNTIMEFUNCTIONS_H

cpp/src/cider/exec/nextgen/operators/RowToColumnNode.cpp

@@ -458,9 +458,7 @@ void RowToColumnTranslator::codegenImpl(SuccessorEmitter successor_wrapper,
 
   auto output_index = func->createVariable(JITTypeTag::INT64, "output_index", 0);
 
-  // Get input ArrowArray length from previous C2RNode
-  auto prev_c2r_node = static_cast<RowToColumnNode*>(node_.get())->getColumnToRowNode();
-  auto input_array_len = prev_c2r_node->getColumnRowNum();
+  auto input_array_len = context.getInputLength();
   bool bitwise_bool = static_cast<RowToColumnNode*>(node_.get())->writeBitwiseBool();
 
   for (int64_t i = 0; i < exprs.size(); ++i) {
@@ -474,7 +472,7 @@ void RowToColumnTranslator::codegenImpl(SuccessorEmitter successor_wrapper,
   successor_wrapper(successor, context);
 
   // Execute length field updating build function after C2R loop finished.
-  prev_c2r_node->registerDeferFunc([output_index, &output_exprs, &context]() mutable {
+  context.appendDeferFunc([output_index, &output_exprs, &context]() mutable {
     for (auto& expr : output_exprs) {
       size_t local_offset = expr->getLocalIndex();
       CHECK_NE(local_offset, 0);

cpp/src/cider/exec/nextgen/operators/VectorizedFilterNode.cpp

@@ -0,0 +1,143 @@
+/*
+ * Copyright(c) 2022-2023 Intel Corporation.
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+#include "exec/nextgen/operators/VectorizedFilterNode.h"
+
+#include <cstddef>
+#include <memory>
+
+#include "exec/nextgen/context/CodegenContext.h"
+#include "exec/nextgen/jitlib/base/JITControlFlow.h"
+#include "exec/nextgen/jitlib/base/JITFunction.h"
+#include "exec/nextgen/jitlib/base/ValueTypes.h"
+#include "exec/nextgen/operators/OpNode.h"
+#include "exec/nextgen/operators/VectorizedProjectNode.h"
+#include "exec/nextgen/utils/JITExprValue.h"
+#include "type/data/sqltypes.h"
+#include "util/sqldefs.h"
+
+namespace cider::exec::nextgen::operators {
+
+using namespace context;
+using namespace jitlib;
+
+TranslatorPtr VectorizedFilterNode::toTranslator(const TranslatorPtr& successor) {
+  return createOpTranslator<VectorizedFilterTranslator>(shared_from_this(), successor);
+}
+
+jitlib::JITValuePointer VectorizedFilterTranslator::generateFilterCondition(
+    CodegenContext& context) {
+  auto&& [_, exprs] = node_->getOutputExprs();
+  // Combine conditions with AND.
+  ExprPtr conditon = exprs.front();
+  CHECK_EQ(conditon->get_type_info().get_type(), kBOOLEAN);
+  for (size_t i = 1; i < exprs.size(); ++i) {
+    CHECK_EQ(exprs[i]->get_type_info().get_type(), kBOOLEAN);
+    conditon = std::make_shared<Analyzer::BinOper>(
+        SQLTypeInfo(kBOOLEAN, !(conditon->getNullable() || exprs[i]->getNullable())),
+        false,
+        kAND,
+        kONE,
+        conditon,
+        exprs[i]);
+  }
+
+  // Register batch for the condition.
+  auto condition_array = context.registerBatch(conditon->get_type_info(), "condition");
+  conditon->setLocalIndex(context.appendArrowArrayValues(
+      condition_array, utils::JITExprValue(0, JITExprValueType::BATCH)));
+
+  // Generate the condition.
+  ExprPtrVector condition_target{conditon};
+  auto vectorized_proj_node = createOpNode<VectorizedProjectNode>(condition_target);
+  auto vectorized_proj_translator = vectorized_proj_node->toTranslator();
+  vectorized_proj_translator->codegen(context, [](CodegenContext&) {});
+
+  // Remove null data
+  // TODO bigPYJ1151): Support ISNULL NOTNULL vectorization.
+  auto&& buffers = context.getArrowArrayValues(conditon->getLocalIndex()).second;
+  utils::FixSizeJITExprValue bool_data(buffers);
+  CHECK(bool_data.getValue().get());
+  if (conditon->getNullable()) {
+    CHECK(bool_data.getNull().get());
+    context::codegen_utils::bitBufferAnd(bool_data.getValue(),
+                                         bool_data.getValue(),
+                                         bool_data.getNull(),
+                                         context.getInputLength());
+  }
+
+  // Reset tail bits.
+  context.getJITFunction()->emitRuntimeFunctionCall(
+      "reset_tail_bits_64_align",
+      JITFunctionEmitDescriptor{
+          .params_vector = {bool_data.getValue().get(), context.getInputLength().get()}});
+
+  return bool_data.getValue();
+}
+
+void VectorizedFilterTranslator::consume(context::CodegenContext& context) {
+  codegen(context, [this](context::CodegenContext& context) {
+    if (successor_) {
+      successor_->consume(context);
+    }
+  });
+}
+
+void VectorizedFilterTranslator::codegenImpl(SuccessorEmitter successor_wrapper,
+                                             context::CodegenContext& context,
+                                             void* successor) {
+  auto selected_row_mask = generateFilterCondition(context);
+
+  auto&& func = context.getJITFunction();
+  auto loop_builder = func->createLoopBuilder();
+
+  auto mask_index = func->createVariable(JITTypeTag::INT64, "selected_row_mask_index", 0);
+  auto row_index_start =
+      func->createVariable(JITTypeTag::INT64, "selected_row_start_index", 0);
+  auto upper = (context.getInputLength() + 63) / 64;
+
+  loop_builder->condition([&upper, &mask_index]() { return mask_index < upper; })
+      ->loop([&](LoopBuilder* builder) {
+        auto&& selected_row_mask_i64 =
+            selected_row_mask->castPointerSubType(JITTypeTag::INT64);
+        auto current_selected_mask = selected_row_mask_i64[mask_index];
+
+        // Set current i64 selected mask.
+        context.setFilterMask(current_selected_mask, row_index_start);
+
+        // All the rows are filtered.
+        builder->loopContinue(current_selected_mask == 0);
+
+        // TODO (bigPYJ1151): Support full execution when no rows are filtered.
+        // Row-based execution.
+        successor_wrapper(successor, context);
+      })
+      ->update([&mask_index, &row_index_start]() {
+        mask_index = mask_index + 1;
+        row_index_start = row_index_start + 64;
+      })
+      ->build();
+
+  for (auto& defer_func : context.getDeferFunc()) {
+    defer_func();
+  }
+  context.clearDeferFunc();
+}
+}  // namespace cider::exec::nextgen::operators