fix: to_params ignored repeating nodes

- to_params uses subtree iterator to create
parameters for cached query. But our plan
is a DAG, not a tree. We have multiple nodes
referring to the same node, this leads to
situation when we traverse the same subquery
node multiple times. This in term leads to
wrong number of parameters produces by
to_params
- fix this by keeping track of already traversed
parameter nodes in to_params

sbroad-core/src/backend/sql/ir.rs

@@ -8,6 +8,7 @@ use crate::ir::node::{
 use crate::ir::relation::Column;
 use crate::ir::transformation::redistribution::MotionPolicy;
 use crate::ir::tree::traversal::{LevelNode, PostOrderWithFilter};
+use ahash::AHashSet;
 use opentelemetry::Context;
 use serde::{Deserialize, Serialize};
 use smol_str::format_smolstr;
@@ -165,7 +166,15 @@ impl ExecutionPlan {
     pub fn to_params(&self) -> Result<Vec<Value>, SbroadError> {
         let plan = self.get_ir_plan();
         let capacity = plan.nodes.len();
-        let filter = |id: NodeId| -> bool { matches!(plan.get_node(id), Ok(Node::Parameter(_))) };
+        let mut was: AHashSet<NodeId> = AHashSet::with_capacity(plan.constants.len());
+        let filter = |id: NodeId| -> bool {
+            if matches!(plan.get_node(id), Ok(Node::Parameter(_))) && !was.contains(&id) {
+                was.insert(id);
+                true
+            } else {
+                false
+            }
+        };
         let mut tree = PostOrderWithFilter::with_capacity(
             |node| plan.flashback_subtree_iter(node),
             capacity,

sbroad-core/src/ir/parameters.rs

@@ -5,6 +5,7 @@ use std::collections::HashMap;
 
 use crate::ir::node::{Node64, NodeId};
 
+// TODO: why this wrapper is needed? remove it
 #[derive(Clone, Debug, PartialEq, Eq, Deserialize, Serialize)]
 pub struct Parameters(HashMap<NodeId, Node64>);
 
@@ -32,4 +33,12 @@ impl Parameters {
     pub fn drain(&mut self) -> HashMap<NodeId, Node64> {
         std::mem::take(&mut self.0)
     }
+
+    pub fn len(&self) -> usize {
+        self.0.len()
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.0.is_empty()
+    }
 }

sbroad-core/src/ir/tree/traversal.rs

@@ -49,7 +49,7 @@ where
     }
 }
 
-pub type FilterFn<'filter, T> = Box<dyn Fn(T) -> bool + 'filter>;
+pub type FilterFn<'filter, T> = Box<dyn FnMut(T) -> bool + 'filter>;
 
 pub struct PostOrderWithFilter<'filter, F, I, T>
 where