Remove const from input nodes.

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

Author: ienkovich

omniscidb/IR/Node.cpp

@@ -100,17 +100,17 @@ bool isRenamedInput(const Node* node, const size_t index, const std::string& new
 
 }  // namespace
 
-thread_local unsigned Node::crt_id_ = FIRST_NODE_ID;
+std::atomic<unsigned> Node::crt_id_ = FIRST_NODE_ID;
 
 Node::Node(NodeInputs inputs)
     : inputs_(std::move(inputs))
-    , id_(crt_id_++)
+    , id_(crt_id_.fetch_add(1))
     , context_data_(nullptr)
     , is_nop_(false) {}
 
 void Node::replaceInput(
-    std::shared_ptr<const Node> old_input,
-    std::shared_ptr<const Node> input,
+    NodePtr old_input,
+    NodePtr input,
     std::optional<std::unordered_map<unsigned, unsigned>> old_to_new_index_map) {
   InputRewriter rewriter;
   if (old_to_new_index_map) {
@@ -121,8 +121,8 @@ void Node::replaceInput(
   replaceInput(old_input, input, rewriter);
 }
 
-void Node::replaceInput(std::shared_ptr<const Node> old_input,
-                        std::shared_ptr<const Node> input,
+void Node::replaceInput(NodePtr old_input,
+                        NodePtr input,
                         hdk::ir::ExprRewriter& input_redirector) {
   bool replaced = false;
   for (auto& input_ptr : inputs_) {
@@ -136,10 +136,6 @@ void Node::replaceInput(std::shared_ptr<const Node> old_input,
   }
 }
 
-void Node::resetRelAlgFirstId() noexcept {
-  crt_id_ = FIRST_NODE_ID;
-}
-
 void Node::print() const {
   std::cout << toString() << std::endl;
 }

omniscidb/IR/Node.h

@@ -65,7 +65,8 @@ class SortField {
   const NullSortedPosition nulls_pos_;
 };
 
-using NodeInputs = std::vector<std::shared_ptr<const Node>>;
+using NodePtr = std::shared_ptr<Node>;
+using NodeInputs = std::vector<NodePtr>;
 
 class Node {
  public:
@@ -116,12 +117,22 @@ class Node {
     return inputs_[idx].get();
   }
 
+  Node* getInput(const size_t idx) {
+    CHECK_LT(idx, inputs_.size());
+    return inputs_[idx].get();
+  }
+
   std::shared_ptr<const Node> getAndOwnInput(const size_t idx) const {
     CHECK_LT(idx, inputs_.size());
     return inputs_[idx];
   }
 
-  void addManagedInput(std::shared_ptr<const Node> input) { inputs_.push_back(input); }
+  NodePtr getAndOwnInput(const size_t idx) {
+    CHECK_LT(idx, inputs_.size());
+    return inputs_[idx];
+  }
+
+  void addManagedInput(NodePtr input) { inputs_.push_back(input); }
 
   bool hasInput(const Node* needle) const {
     for (auto& input_ptr : inputs_) {
@@ -134,12 +145,12 @@ class Node {
 
   virtual void rewriteExprs(hdk::ir::ExprRewriter& rewriter) {}
 
-  void replaceInput(std::shared_ptr<const Node> old_input,
-                    std::shared_ptr<const Node> input,
+  void replaceInput(NodePtr old_input,
+                    NodePtr input,
                     std::optional<std::unordered_map<unsigned, unsigned>>
                         old_to_new_index_map = std::nullopt);
-  void replaceInput(std::shared_ptr<const Node> old_input,
-                    std::shared_ptr<const Node> input,
+  void replaceInput(NodePtr old_input,
+                    NodePtr input,
                     hdk::ir::ExprRewriter& input_redirector);
 
   // to keep an assigned DAG node id for data recycler
@@ -162,8 +173,6 @@ class Node {
 
   virtual std::shared_ptr<Node> deepCopy() const = 0;
 
-  static void resetRelAlgFirstId() noexcept;
-
   /**
    * Clears the ptr to the result for this descriptor. Is only used for overriding step
    * results in distributed mode.
@@ -185,7 +194,7 @@ class Node {
   mutable const RaExecutionDesc* context_data_;
   bool is_nop_;
   mutable std::vector<TargetMetaInfo> targets_metainfo_;
-  static thread_local unsigned crt_id_;
+  static std::atomic<unsigned> crt_id_;
   mutable size_t dag_node_id_;
   mutable std::shared_ptr<const ExecutionResult> result_;
 };
@@ -203,8 +212,6 @@ inline std::string inputsToString(const NodeInputs& inputs) {
   return ss.str();
 }
 
-using NodePtr = std::shared_ptr<Node>;
-
 class Scan : public Node {
  public:
   Scan(TableInfoPtr tinfo, std::vector<ColumnInfoPtr> column_infos)
@@ -280,9 +287,7 @@ class Scan : public Node {
 class Project : public Node {
  public:
   // Takes memory ownership of the expressions.
-  Project(ExprPtrVector exprs,
-          std::vector<std::string> fields,
-          std::shared_ptr<const Node> input)
+  Project(ExprPtrVector exprs, std::vector<std::string> fields, NodePtr input)
       : exprs_(std::move(exprs)), fields_(std::move(fields)) {
     inputs_.push_back(input);
   }
@@ -364,7 +369,7 @@ class Aggregate : public Node {
   Aggregate(const size_t groupby_count,
             ExprPtrVector aggs,
             std::vector<std::string> fields,
-            std::shared_ptr<const Node> input)
+            NodePtr input)
       : groupby_count_(groupby_count)
       , aggs_(std::move(aggs))
       , fields_(std::move(fields)) {
@@ -439,10 +444,7 @@ class Aggregate : public Node {
 
 class Join : public Node {
  public:
-  Join(std::shared_ptr<const Node> lhs,
-       std::shared_ptr<const Node> rhs,
-       ExprPtr condition,
-       const JoinType join_type)
+  Join(NodePtr lhs, NodePtr rhs, ExprPtr condition, const JoinType join_type)
       : condition_(std::move(condition)), join_type_(join_type) {
     inputs_.emplace_back(std::move(lhs));
     inputs_.emplace_back(std::move(rhs));
@@ -604,8 +606,7 @@ class TranslatedJoin : public Node {
 
 class Filter : public Node {
  public:
-  Filter(ExprPtr condition, std::shared_ptr<const Node> input)
-      : condition_(std::move(condition)) {
+  Filter(ExprPtr condition, NodePtr input) : condition_(std::move(condition)) {
     CHECK(condition_);
     inputs_.emplace_back(std::move(input));
   }
@@ -663,7 +664,7 @@ class Sort : public Node {
   Sort(std::vector<SortField> collation,
        const size_t limit,
        const size_t offset,
-       std::shared_ptr<const Node> input)
+       NodePtr input)
       : collation_(std::move(collation))
       , limit_(limit)
       , offset_(offset)

omniscidb/QueryEngine/RelAlgDagBuilder.cpp

@@ -2382,7 +2382,7 @@ class RelAlgDispatcher {
     return {key, val};
   }
 
-  std::shared_ptr<const hdk::ir::Node> prev(const rapidjson::Value& crt_node) {
+  hdk::ir::NodePtr prev(const rapidjson::Value& crt_node) {
     const auto id = node_id(crt_node);
     CHECK(id);
     CHECK_EQ(static_cast<size_t>(id), nodes_.size());
@@ -2432,7 +2432,6 @@ RelAlgDagBuilder::RelAlgDagBuilder(const std::string& query_ra,
         "Failed to parse relational algebra tree. Possible query syntax error.");
   }
   CHECK(query_ast.IsObject());
-  hdk::ir::Node::resetRelAlgFirstId();
   build(query_ast, *this);
 }
 

omniscidb/QueryEngine/RelAlgOptimizer.cpp

@@ -31,12 +31,12 @@ namespace {
 
 class ProjectInputRedirector : public hdk::ir::ExprRewriter {
  public:
-  ProjectInputRedirector(const std::unordered_set<const hdk::ir::Project*>& crt_inputs)
+  ProjectInputRedirector(const std::unordered_set<hdk::ir::Project*>& crt_inputs)
       : crt_projects_(crt_inputs) {}
 
   hdk::ir::ExprPtr visitColumnRef(const hdk::ir::ColumnRef* col_ref) override {
     auto source = dynamic_cast<const hdk::ir::Project*>(col_ref->node());
-    if (source && crt_projects_.count(source)) {
+    if (source && crt_projects_.count(const_cast<hdk::ir::Project*>(source))) {
       auto new_source = source->getInput(0);
       auto new_col_ref = dynamic_cast<const hdk::ir::ColumnRef*>(
           source->getExpr(col_ref->index()).get());
@@ -49,7 +49,7 @@ class ProjectInputRedirector : public hdk::ir::ExprRewriter {
   }
 
  private:
-  const std::unordered_set<const hdk::ir::Project*>& crt_projects_;
+  const std::unordered_set<hdk::ir::Project*>& crt_projects_;
 };
 
 // TODO: use more generic InputRenumberVisitor instead.
@@ -81,13 +81,13 @@ class InputSimpleRenumberVisitor : public hdk::ir::ExprRewriter {
 
 size_t get_actual_source_size(
     const hdk::ir::Project* curr_project,
-    const std::unordered_set<const hdk::ir::Project*>& projects_to_remove) {
+    const std::unordered_set<hdk::ir::Project*>& projects_to_remove) {
   auto source = curr_project->getInput(0);
   while (auto filter = dynamic_cast<const hdk::ir::Filter*>(source)) {
     source = filter->getInput(0);
   }
   if (auto src_project = dynamic_cast<const hdk::ir::Project*>(source)) {
-    if (projects_to_remove.count(src_project)) {
+    if (projects_to_remove.count(const_cast<hdk::ir::Project*>(src_project))) {
       return get_actual_source_size(src_project, projects_to_remove);
     }
   }
@@ -113,11 +113,11 @@ bool safe_to_redirect(
 }
 
 bool is_identical_copy(
-    const hdk::ir::Project* project,
+    hdk::ir::Project* project,
     const std::unordered_map<const hdk::ir::Node*,
                              std::unordered_set<const hdk::ir::Node*>>& du_web,
-    const std::unordered_set<const hdk::ir::Project*>& projects_to_remove,
-    std::unordered_set<const hdk::ir::Project*>& permutating_projects) {
+    const std::unordered_set<hdk::ir::Project*>& projects_to_remove,
+    std::unordered_set<hdk::ir::Project*>& permutating_projects) {
   auto source_size = get_actual_source_size(project, projects_to_remove);
   if (project->size() > source_size) {
     return false;
@@ -221,17 +221,17 @@ void propagate_rex_input_renumber(
 // This function appears to redirect/remove redundant Projection input nodes(?)
 void redirect_inputs_of(
     std::shared_ptr<hdk::ir::Node> node,
-    const std::unordered_set<const hdk::ir::Project*>& projects,
-    const std::unordered_set<const hdk::ir::Project*>& permutating_projects,
+    const std::unordered_set<hdk::ir::Project*>& projects,
+    const std::unordered_set<hdk::ir::Project*>& permutating_projects,
     const std::unordered_map<const hdk::ir::Node*,
                              std::unordered_set<const hdk::ir::Node*>>& du_web) {
   if (dynamic_cast<hdk::ir::LogicalUnion*>(node.get())) {
     return;  // UNION keeps all Projection inputs.
   }
-  std::shared_ptr<const hdk::ir::Project> src_project = nullptr;
+  std::shared_ptr<hdk::ir::Project> src_project = nullptr;
   for (size_t i = 0; i < node->inputCount(); ++i) {
     if (auto project =
-            std::dynamic_pointer_cast<const hdk::ir::Project>(node->getAndOwnInput(i))) {
+            std::dynamic_pointer_cast<hdk::ir::Project>(node->getAndOwnInput(i))) {
       if (projects.count(project.get())) {
         src_project = project;
         break;
@@ -244,8 +244,8 @@ void redirect_inputs_of(
   if (auto join = std::dynamic_pointer_cast<hdk::ir::Join>(node)) {
     auto other_project =
         src_project == node->getAndOwnInput(0)
-            ? std::dynamic_pointer_cast<const hdk::ir::Project>(node->getAndOwnInput(1))
-            : std::dynamic_pointer_cast<const hdk::ir::Project>(node->getAndOwnInput(0));
+            ? std::dynamic_pointer_cast<hdk::ir::Project>(node->getAndOwnInput(1))
+            : std::dynamic_pointer_cast<hdk::ir::Project>(node->getAndOwnInput(0));
     join->replaceInput(src_project, src_project->getAndOwnInput(0));
     // Case when join users have to be adjusted is not expected.
     CHECK(src_project != node->getAndOwnInput(0) ||
@@ -459,16 +459,16 @@ bool project_separates_sort(const hdk::ir::Project* project,
 // TODO(miyu): allow more chance if proved safe
 void eliminate_identical_copy(
     std::vector<std::shared_ptr<hdk::ir::Node>>& nodes) noexcept {
-  std::unordered_set<std::shared_ptr<const hdk::ir::Node>> copies;
+  std::unordered_set<hdk::ir::NodePtr> copies;
   auto sink = nodes.back();
   for (auto node : nodes) {
-    auto aggregate = std::dynamic_pointer_cast<const hdk::ir::Aggregate>(node);
+    auto aggregate = std::dynamic_pointer_cast<hdk::ir::Aggregate>(node);
     if (!aggregate || aggregate == sink ||
         !(aggregate->getGroupByCount() == 1 && aggregate->getAggsCount() == 0)) {
       continue;
     }
     auto project =
-        std::dynamic_pointer_cast<const hdk::ir::Project>(aggregate->getAndOwnInput(0));
+        std::dynamic_pointer_cast<hdk::ir::Project>(aggregate->getAndOwnInput(0));
     if (project && project->size() == aggregate->size() &&
         project->getFields() == aggregate->getFields()) {
       CHECK_EQ(size_t(0), copies.count(aggregate));
@@ -483,30 +483,29 @@ void eliminate_identical_copy(
     if (!copies.count(last_source)) {
       continue;
     }
-    auto aggregate = std::dynamic_pointer_cast<const hdk::ir::Aggregate>(last_source);
+    auto aggregate = std::dynamic_pointer_cast<hdk::ir::Aggregate>(last_source);
     CHECK(aggregate);
-    if (!std::dynamic_pointer_cast<const hdk::ir::Join>(node) || aggregate->size() != 1) {
+    if (!std::dynamic_pointer_cast<hdk::ir::Join>(node) || aggregate->size() != 1) {
       continue;
     }
     auto project =
-        std::dynamic_pointer_cast<const hdk::ir::Project>(aggregate->getAndOwnInput(0));
+        std::dynamic_pointer_cast<hdk::ir::Project>(aggregate->getAndOwnInput(0));
     CHECK(project);
     CHECK_EQ(size_t(1), project->size());
     if (!is_distinct(size_t(0), project.get())) {
       continue;
     }
     auto new_source = project->getAndOwnInput(0);
-    if (std::dynamic_pointer_cast<const hdk::ir::Sort>(new_source) ||
-        std::dynamic_pointer_cast<const hdk::ir::Scan>(new_source)) {
+    if (new_source->is<hdk::ir::Sort>() || new_source->is<hdk::ir::Scan>()) {
       node->replaceInput(last_source, new_source);
     }
   }
   decltype(copies)().swap(copies);
 
   auto web = build_du_web(nodes);
 
-  std::unordered_set<const hdk::ir::Project*> projects;
-  std::unordered_set<const hdk::ir::Project*> permutating_projects;
+  std::unordered_set<hdk::ir::Project*> projects;
+  std::unordered_set<hdk::ir::Project*> permutating_projects;
   auto const visible_projs = get_visible_projects(nodes.back().get());
   for (auto node_it = nodes.begin(); node_it != nodes.end(); node_it++) {
     auto node = *node_it;
@@ -1350,8 +1349,8 @@ class InputRedirector : public hdk::ir::ExprRewriter {
 };
 
 void replace_all_usages(
-    std::shared_ptr<const hdk::ir::Node> old_def_node,
-    std::shared_ptr<const hdk::ir::Node> new_def_node,
+    hdk::ir::NodePtr old_def_node,
+    hdk::ir::NodePtr new_def_node,
     std::unordered_map<const hdk::ir::Node*, std::shared_ptr<hdk::ir::Node>>&
         deconst_mapping,
     std::unordered_map<const hdk::ir::Node*, std::unordered_set<const hdk::ir::Node*>>&