Feat: Add Pipe Node

Introduces a new `Pipe` node to the Abstract Syntax Tree (AST).
This node represents a functional composition pattern, allowing
for chaining of operations.

The changes include:
- Defining the `Pipe` variant in `BoundKind` and `UntypedKind`.
- Implementing parsing logic for the `pipe` keyword.
- Adding handling for the `Pipe` node in various compiler passes
  (analyzer, binder, captures, dumper, macros, optimizer, TCO,
  type checker).
- Including a placeholder implementation for `Pipe` execution in the
  VM.
- Updating integration tests to use the new `pipe` syntax.

src/ast/compiler/analyzer.rs

@@ -237,6 +237,20 @@ impl<'a> Analyzer<'a> {
                     Purity::Impure,
                 )
             }
+            BoundKind::Pipe { inputs, lambda } => {
+                let mut analyzed_inputs = Vec::with_capacity(inputs.len());
+                for input in inputs {
+                    analyzed_inputs.push(self.visit(Rc::new(input.clone())));
+                }
+                let a_lambda = Box::new(self.visit(Rc::new((**lambda).clone())));
+                (
+                    BoundKind::Pipe {
+                        inputs: analyzed_inputs,
+                        lambda: a_lambda,
+                    },
+                    Purity::Impure,
+                )
+            }
 
             BoundKind::Block { exprs } => {
                 let mut new_exprs = Vec::with_capacity(exprs.len());

src/ast/compiler/binder.rs

@@ -276,6 +276,21 @@ impl Binder {
                 }
             }
 
+            UntypedKind::Pipe { inputs, lambda } => {
+                let mut bound_inputs = Vec::with_capacity(inputs.len());
+                for input in inputs {
+                    bound_inputs.push(self.bind(&input)?);
+                }
+                let bound_lambda = Box::new(self.bind(lambda.as_ref())?);
+                Ok(self.make_node(
+                    node.identity.clone(),
+                    BoundKind::Pipe {
+                        inputs: bound_inputs,
+                        lambda: bound_lambda,
+                    },
+                ))
+            }
+
             UntypedKind::Lambda { params, body } => {
                 let identity = node.identity.clone();
                 self.functions.push(FunctionCompiler::new(

src/ast/compiler/bound_nodes.rs

@@ -135,7 +135,10 @@ pub enum BoundKind<T = ()> {
     Again {
         args: Box<BoundNode<T>>,
     },
-
+    Pipe {
+        inputs: Vec<BoundNode<T>>,
+        lambda: Box<BoundNode<T>>,
+    },
     Block {
         exprs: Vec<BoundNode<T>>,
     },
@@ -304,6 +307,7 @@ impl<T> BoundKind<T> {
             }
             BoundKind::Call { .. } => "CALL".to_string(),
             BoundKind::Again { .. } => "AGAIN".to_string(),
+            BoundKind::Pipe { .. } => "PIPE".to_string(),
             BoundKind::Block { .. } => "BLOCK".to_string(),
             BoundKind::Tuple { elements } => format!("TUPLE({})", elements.len()),
             BoundKind::Record { values, .. } => format!("RECORD({})", values.len()),

src/ast/compiler/captures.rs

@@ -89,7 +89,16 @@ impl CapturePass {
                     args: Box::new(Self::transform(*args, capture_map)),
                 };
             }
-
+            BoundKind::Pipe { inputs, lambda } => {
+                let mut t_inputs = Vec::with_capacity(inputs.len());
+                for input in inputs {
+                    t_inputs.push(Self::transform(input, capture_map));
+                }
+                node.kind = BoundKind::Pipe {
+                    inputs: t_inputs,
+                    lambda: Box::new(Self::transform(*lambda, capture_map)),
+                };
+            }
             BoundKind::Block { exprs } => {
                 node.kind = BoundKind::Block {
                     exprs: exprs

src/ast/compiler/dumper.rs

@@ -203,6 +203,15 @@ impl Dumper {
                 self.visit(args);
                 self.indent -= 1;
             }
+            BoundKind::Pipe { inputs, lambda } => {
+                self.log("Pipe", node);
+                self.indent += 1;
+                for input in inputs {
+                    self.visit(input);
+                }
+                self.visit(lambda);
+                self.indent -= 1;
+            }
 
             BoundKind::Block { exprs } => {
                 self.log("Block", node);

src/ast/compiler/macros.rs

@@ -152,6 +152,22 @@ impl<E: MacroEvaluator> MacroExpander<E> {
                 })
             }
 
+            UntypedKind::Pipe { inputs, lambda } => {
+                let mut expanded_inputs = Vec::with_capacity(inputs.len());
+                for input in inputs {
+                    expanded_inputs.push(self.expand_recursive(input)?);
+                }
+                let expanded_lambda = Box::new(self.expand_recursive(*lambda)?);
+                Ok(Node {
+                    identity: node.identity,
+                    kind: UntypedKind::Pipe {
+                        inputs: expanded_inputs,
+                        lambda: expanded_lambda,
+                    },
+                    ty: (),
+                })
+            }
+
             UntypedKind::Lambda { params, body } => {
                 self.registry.push();
                 let expanded_params = self.expand_recursive(*params)?;
@@ -498,6 +514,22 @@ impl<E: MacroEvaluator> MacroExpander<E> {
                 })
             }
 
+            UntypedKind::Pipe { inputs, lambda } => {
+                let mut expanded_inputs = Vec::with_capacity(inputs.len());
+                for input in inputs {
+                    expanded_inputs.push(self.expand_template(input, state)?);
+                }
+                let expanded_lambda = Box::new(self.expand_template(*lambda, state)?);
+                Ok(Node {
+                    identity: node.identity,
+                    kind: UntypedKind::Pipe {
+                        inputs: expanded_inputs,
+                        lambda: expanded_lambda,
+                    },
+                    ty: (),
+                })
+            }
+
             UntypedKind::Lambda { params, body } => {
                 let expanded_params = self.expand_template(*params, state)?;
                 let body = self.expand_template(Node::clone(&body), state)?;

src/ast/compiler/optimizer/engine.rs

@@ -386,6 +386,21 @@ impl Optimizer {
                 )
             }
 
+            BoundKind::Pipe { inputs, lambda } => {
+                let mut o_inputs = Vec::with_capacity(inputs.len());
+                for input in inputs {
+                    o_inputs.push(self.visit_node(input, sub, path));
+                }
+                let o_lambda = Box::new(self.visit_node(*lambda, sub, path));
+                (
+                    BoundKind::Pipe {
+                        inputs: o_inputs,
+                        lambda: o_lambda,
+                    },
+                    node.ty.clone(),
+                )
+            }
+
             BoundKind::Block { ref exprs } => {
                 let mut info = UsageInfo::default();
                 if !exprs.is_empty() {

src/ast/compiler/optimizer/utils.rs

@@ -152,6 +152,12 @@ impl UsageInfo {
             BoundKind::Again { args } => {
                 self.collect(args);
             }
+            BoundKind::Pipe { inputs, lambda } => {
+                for input in inputs {
+                    self.collect(input);
+                }
+                self.collect(lambda);
+            }
             BoundKind::Nop | BoundKind::FieldAccessor(_) | BoundKind::Extension(_) => {}
         }
     }

src/ast/compiler/tco.rs

@@ -48,6 +48,16 @@ impl TCO {
                     args: Box::new(Self::transform(Rc::new((**args).clone()), false)),
                 }
             }
+            BoundKind::Pipe { inputs, lambda } => {
+                let mut t_inputs = Vec::with_capacity(inputs.len());
+                for input in inputs {
+                    t_inputs.push(Self::transform(Rc::new(input.clone()), false));
+                }
+                BoundKind::Pipe {
+                    inputs: t_inputs,
+                    lambda: Box::new(Self::transform(Rc::new((**lambda).clone()), false)),
+                }
+            }
 
             BoundKind::If {
                 cond,

src/ast/compiler/type_checker.rs

@@ -387,6 +387,26 @@ impl TypeChecker {
                 )
             }
 
+            BoundKind::Pipe { inputs, lambda } => {
+                let mut typed_inputs = Vec::with_capacity(inputs.len());
+                for input in inputs {
+                    typed_inputs.push(self.check_node(input, ctx)?);
+                }
+                let typed_lambda = self.check_node(*lambda, ctx)?;
+                let ret_ty = if let StaticType::Function(sig) = &typed_lambda.ty {
+                    sig.ret.clone()
+                } else {
+                    StaticType::Any
+                };
+                (
+                    BoundKind::Pipe {
+                        inputs: typed_inputs,
+                        lambda: Box::new(typed_lambda),
+                    },
+                    StaticType::Series(Box::new(ret_ty)),
+                )
+            }
+
             BoundKind::Block { exprs } => {
                 let mut typed_exprs = Vec::new();
                 let mut last_ty = StaticType::Void;

src/ast/nodes.rs

@@ -91,6 +91,10 @@ pub enum UntypedKind {
     Again {
         args: Box<Node<UntypedKind>>,
     },
+    Pipe {
+        inputs: Vec<Node<UntypedKind>>,
+        lambda: Box<Node<UntypedKind>>,
+    },
     Block {
         exprs: Vec<Node<UntypedKind>>,
     },

src/ast/parser.rs

@@ -137,6 +137,7 @@ impl<'a> Parser<'a> {
             match sym.name.as_ref() {
                 "if" => self.parse_if(identity),
                 "fn" => self.parse_fn(identity),
+                "pipe" => self.parse_pipe(identity),
                 "again" => self.parse_again(identity),
                 "def" => self.parse_def(identity),
                 "assign" => self.parse_assign(identity),
@@ -229,6 +230,21 @@ impl<'a> Parser<'a> {
         })
     }
 
+    fn parse_pipe(&mut self, identity: Identity) -> Result<Node<UntypedKind>, String> {
+        let inputs_node = self.parse_expression()?;
+        let inputs = match inputs_node.kind {
+            UntypedKind::Tuple { elements } => elements,
+            _ => vec![inputs_node],
+        };
+        let lambda = Box::new(self.parse_expression()?);
+
+        Ok(Node {
+            identity,
+            kind: UntypedKind::Pipe { inputs, lambda },
+            ty: (),
+        })
+    }
+
     fn parse_fn(&mut self, identity: Identity) -> Result<Node<UntypedKind>, String> {
         let params = Box::new(self.parse_param_vector()?);
         let body = self.parse_expression()?;

src/ast/vm.rs

@@ -368,6 +368,17 @@ impl VM {
                     Ok(Value::Void)
                 }
             }
+            BoundKind::Pipe { inputs, lambda } => {
+                // To be implemented in next step:
+                // 1. Create PipeStream
+                // 2. Link with RootStream and Inputs
+                // 3. Create and return SharedSeries
+                for input in inputs {
+                    self.eval_internal(obs, input)?;
+                }
+                self.eval_internal(obs, lambda)?;
+                Ok(Value::Void)
+            }
             BoundKind::Block { exprs } => {
                 let mut last = Value::Void;
                 for e in exprs {

src/integration_test.rs

@@ -364,7 +364,7 @@ mod tests {
     fn test_multi_level_destructuring() {
         let env = Environment::new();
         let source = "(do
-          (def pipe (fn [conf]
+          (def process_data (fn [conf]
             (do
                (def [str s] conf)
                (def [f ss] s)
@@ -372,7 +372,7 @@ mod tests {
             )
            )
           )
-          (pipe [\"btc\" [:close \"cls\"]]))";
+          (process_data [\"btc\" [:close \"cls\"]]))";
 
         let res = env.run_script(source).unwrap();
         assert_eq!(