QueryEngine/RelAlgDagBuilder.cpp
/**
 * Detect the presence of window function operators nested inside expressions. Separate
 * the window function operator from the expression, computing the expression as a
 * subsequent step and replacing the window function operator with a RexInput. Also move
 * all input nodes to the newly created project node.
 * In pseudocode:
 * for each rex in project list:
 *    detect window function expression
 *    if window function expression:
 *        copy window function expression
 *        replace window function expression in base expression w/ input
 *        add base expression to new project node after the current node
 *        replace base expression in current project node with the window function
 expression copy
 */
Detect the presence of window function operators nested inside expressions. Separate
the window function operator from the expression, computing the expression as a
subsequent step and replacing the window function operator with a RexInput. Also move
all input nodes to the newly created project node.
In pseudocode:
for each rex in project list:
detect window function expression
if window function expression:
copy window function expression
replace window function expression in base expression w/ input
add base expression to new project node after the current node
replace base expression in current project node with the window function
expression copy
Detect the presence of window function operators nested inside expressions . Separate the window function operator from the expression computing the expression as a subsequent step and replacing the window function operator with a RexInput . Also move all input nodes to the newly created project node . In pseudocode : for each rex in project list : detect window function expression if window function expression : copy window function expression replace window function expression in base expression w / input add base expression to new project node after the current node replace base expression in current project node with the window function expression copy
void separate_window_function_expressions(
    std::vector<std::shared_ptr<RelAlgNode>>& nodes) {
  std::list<std::shared_ptr<RelAlgNode>> node_list(nodes.begin(), nodes.end());
  WindowFunctionDetectionVisitor visitor;
  for (auto node_itr = node_list.begin(); node_itr != node_list.end(); ++node_itr) {
    const auto node = *node_itr;
    auto window_func_project_node = std::dynamic_pointer_cast<RelProject>(node);
    if (!window_func_project_node) {
      continue;
    }
    std::unordered_map<size_t, const RexScalar*> embedded_window_function_expressions;
    for (size_t i = 0; i < window_func_project_node->size(); i++) {
      const auto scalar_rex = window_func_project_node->getProjectAt(i);
      if (is_window_function_operator(scalar_rex)) {
        continue;
      }
      if (const auto window_func_rex = visitor.visit(scalar_rex)) {
        const auto ret = embedded_window_function_expressions.insert(
            std::make_pair(i, window_func_rex));
        CHECK(ret.second);
      }
    }
    if (!embedded_window_function_expressions.empty()) {
      std::vector<std::unique_ptr<const RexScalar>> new_scalar_exprs;
      auto window_func_scalar_exprs =
          window_func_project_node->getExpressionsAndRelease();
      for (size_t rex_idx = 0; rex_idx < window_func_scalar_exprs.size(); ++rex_idx) {
        const auto embedded_window_func_expr_pair =
            embedded_window_function_expressions.find(rex_idx);
        if (embedded_window_func_expr_pair ==
            embedded_window_function_expressions.end()) {
          new_scalar_exprs.emplace_back(
              std::make_unique<const RexInput>(window_func_project_node.get(), rex_idx));
        } else {
          const auto window_func_rex_idx = embedded_window_func_expr_pair->first;
          CHECK_LT(window_func_rex_idx, window_func_scalar_exprs.size());
          const auto& window_func_rex = embedded_window_func_expr_pair->second;
          RexDeepCopyVisitor copier;
          auto window_func_rex_copy = copier.visit(window_func_rex);
          auto window_func_parent_expr =
              window_func_scalar_exprs[window_func_rex_idx].get();
          auto window_func_result_input = std::make_unique<const RexInput>(
              window_func_project_node.get(), window_func_rex_idx);
          RexWindowFuncReplacementVisitor replacer(std::move(window_func_result_input));
          auto new_parent_rex = replacer.visit(window_func_parent_expr);
          new_scalar_exprs.emplace_back(std::move(new_parent_rex));
          window_func_scalar_exprs[window_func_rex_idx] = std::move(window_func_rex_copy);
        }
      }
      CHECK_EQ(window_func_scalar_exprs.size(), new_scalar_exprs.size());
      window_func_project_node->setExpressions(window_func_scalar_exprs);
      RexInputBackpropagationVisitor input_visitor(window_func_project_node.get());
      for (size_t i = 0; i < new_scalar_exprs.size(); i++) {
        if (dynamic_cast<const RexInput*>(new_scalar_exprs[i].get())) {
          continue;
        }
        new_scalar_exprs[i] = input_visitor.visit(new_scalar_exprs[i].get());
      }
      auto new_project =
          std::make_shared<RelProject>(new_scalar_exprs,
                                       window_func_project_node->getFields(),
                                       window_func_project_node);
      node_list.insert(std::next(node_itr), new_project);
      for (auto rebind_itr = std::next(node_itr, 2); rebind_itr != node_list.end();
           rebind_itr++) {
        (*rebind_itr)->replaceInput(window_func_project_node, new_project);
      }
    }
  }
  nodes.assign(node_list.begin(), node_list.end());
}
void separate_window_function_expressions ( std :: vector < std :: shared_ptr < RelAlgNode > > & nodes ) { std :: list < std :: shared_ptr < RelAlgNode > > node_list ( nodes . begin ( ) , nodes . end ( ) ) ; WindowFunctionDetectionVisitor visitor ; for ( auto node_itr = node_list . begin ( ) ; node_itr != node_list . end ( ) ; ++ node_itr ) { const auto node = * node_itr ; auto window_func_project_node = std :: dynamic_pointer_cast < RelProject > ( node ) ; if ( ! window_func_project_node ) { continue ; } std :: unordered_map < size_t , const RexScalar * > embedded_window_function_expressions ; for ( size_t i = 0 ; i < window_func_project_node -> size ( ) ; i ++ ) { const auto scalar_rex = window_func_project_node -> getProjectAt ( i ) ; if ( is_window_function_operator ( scalar_rex ) ) { continue ; } if ( const auto window_func_rex = visitor . visit ( scalar_rex ) ) { const auto ret = embedded_window_function_expressions . insert ( std :: make_pair ( i , window_func_rex ) ) ; CHECK ( ret . second ) ; } } if ( ! embedded_window_function_expressions . empty ( ) ) { std :: vector < std :: unique_ptr < const RexScalar > > new_scalar_exprs ; auto window_func_scalar_exprs = window_func_project_node -> getExpressionsAndRelease ( ) ; for ( size_t rex_idx = 0 ; rex_idx < window_func_scalar_exprs . size ( ) ; ++ rex_idx ) { const auto embedded_window_func_expr_pair = embedded_window_function_expressions . find ( rex_idx ) ; if ( embedded_window_func_expr_pair == embedded_window_function_expressions . end ( ) ) { new_scalar_exprs . emplace_back ( std :: make_unique < const RexInput > ( window_func_project_node . get ( ) , rex_idx ) ) ; } else { const auto window_func_rex_idx = embedded_window_func_expr_pair -> first ; CHECK_LT ( window_func_rex_idx , window_func_scalar_exprs . size ( ) ) ; const auto & window_func_rex = embedded_window_func_expr_pair -> second ; RexDeepCopyVisitor copier ; auto window_func_rex_copy = copier . visit ( window_func_rex ) ; auto window_func_parent_expr = window_func_scalar_exprs [ window_func_rex_idx ] . get ( ) ; auto window_func_result_input = std :: make_unique < const RexInput > ( window_func_project_node . get ( ) , window_func_rex_idx ) ; RexWindowFuncReplacementVisitor replacer ( std :: move ( window_func_result_input ) ) ; auto new_parent_rex = replacer . visit ( window_func_parent_expr ) ; new_scalar_exprs . emplace_back ( std :: move ( new_parent_rex ) ) ; window_func_scalar_exprs [ window_func_rex_idx ] = std :: move ( window_func_rex_copy ) ; } } CHECK_EQ ( window_func_scalar_exprs . size ( ) , new_scalar_exprs . size ( ) ) ; window_func_project_node -> setExpressions ( window_func_scalar_exprs ) ; RexInputBackpropagationVisitor input_visitor ( window_func_project_node . get ( ) ) ; for ( size_t i = 0 ; i < new_scalar_exprs . size ( ) ; i ++ ) { if ( dynamic_cast < const RexInput * > ( new_scalar_exprs [ i ] . get ( ) ) ) { continue ; } new_scalar_exprs [ i ] = input_visitor . visit ( new_scalar_exprs [ i ] . get ( ) ) ; } auto new_project = std :: make_shared < RelProject > ( new_scalar_exprs , window_func_project_node -> getFields ( ) , window_func_project_node ) ; node_list . insert ( std :: next ( node_itr ) , new_project ) ; for ( auto rebind_itr = std :: next ( node_itr , 2 ) ; rebind_itr != node_list . end ( ) ; rebind_itr ++ ) { ( * rebind_itr ) -> replaceInput ( window_func_project_node , new_project ) ; } } } nodes . assign ( node_list . begin ( ) , node_list . end ( ) ) ; }
Detect the presence of window function operators nested inside expressions.
Detect the presence of window function operators nested inside expressions .
// map scalar expression index in the project node to wiondow function ptr // Iterate the target exprs of the project node and check for window function // expressions. If an embedded expression exists, save it in the // embedded_window_function_expressions map and split the expression into a window // function expression and a parent expression in a subsequent project node // top level window function exprs are fine // Replace window func rex with an input rex // Put the parent expr in the new scalar exprs // Put the window func expr in cur scalar exprs // Ensure any inputs from the node containing the expression (the "new" node) // exist on the window function project node, e.g. if we had a binary operation // involving an aggregate value or column not included in the top level // projection list. // ignore top level inputs, these were copied directly from the previous // node // Build the new project node and insert it into the list after the project node // containing the window function // Rebind all the following inputs