From 1875cfdc9b8910c782db43ea5bba42fbcb35fcc1 Mon Sep 17 00:00:00 2001 From: Brummel Date: Mon, 23 Mar 2026 10:22:12 +0100 Subject: [PATCH] Remove unused pipe node definitions The `Pipe` node and its associated logic have been removed from the AST. This commit cleans up the code by removing all references to this defunct node in the analyzer, binder, captures, dumper, lowering, macros, optimizer, and type checker. The parser no longer recognizes the `pipe` keyword. --- src/ast/compiler/analyzer.rs | 18 ------- src/ast/compiler/binder.rs | 16 ------ src/ast/compiler/captures.rs | 14 ----- src/ast/compiler/dumper.rs | 10 ---- src/ast/compiler/lowering.rs | 14 ----- src/ast/compiler/macros.rs | 32 ------------ src/ast/compiler/optimizer/engine.rs | 27 ---------- src/ast/compiler/optimizer/utils.rs | 6 --- src/ast/compiler/type_checker.rs | 36 ------------- src/ast/nodes.rs | 14 ----- src/ast/parser.rs | 16 ------ src/ast/rtl/streams.rs | 77 ++++++++++++++++++++++++++++ src/ast/types.rs | 6 +++ src/ast/vm.rs | 55 +------------------- 14 files changed, 85 insertions(+), 256 deletions(-) diff --git a/src/ast/compiler/analyzer.rs b/src/ast/compiler/analyzer.rs index 89a5ad0..fdc0a1a 100644 --- a/src/ast/compiler/analyzer.rs +++ b/src/ast/compiler/analyzer.rs @@ -226,24 +226,6 @@ impl<'a> Analyzer<'a> { Purity::Impure, ) } - NodeKind::Pipe { - inputs, - lambda, - } => { - let mut analyzed_inputs = Vec::with_capacity(inputs.len()); - for input in inputs { - analyzed_inputs.push(Rc::new(self.visit(input.clone()))); - } - let a_lambda = Rc::new(self.visit(lambda.clone())); - ( - NodeKind::Pipe { - inputs: analyzed_inputs, - lambda: a_lambda, - }, - Purity::Impure, - ) - } - NodeKind::Block { exprs } => { let mut new_exprs = Vec::with_capacity(exprs.len()); let mut p = Purity::Pure; diff --git a/src/ast/compiler/binder.rs b/src/ast/compiler/binder.rs index f6e7b0b..e09d7a9 100644 --- a/src/ast/compiler/binder.rs +++ b/src/ast/compiler/binder.rs @@ -358,22 +358,6 @@ impl Binder { } } - SyntaxKind::Pipe { inputs, lambda } => { - let mut bound_inputs = Vec::with_capacity(inputs.len()); - for input in inputs { - bound_inputs.push(Rc::new(self.bind(input, ExprContext::Expression, diag))); - } - let bound_lambda = - Rc::new(self.bind(lambda.as_ref(), ExprContext::Expression, diag)); - self.make_node( - node.identity.clone(), - NodeKind::Pipe { - inputs: bound_inputs, - lambda: bound_lambda, - }, - ) - } - SyntaxKind::Lambda { params, body, .. } => { let identity = node.identity.clone(); self.functions diff --git a/src/ast/compiler/captures.rs b/src/ast/compiler/captures.rs index 72d0c69..a815762 100644 --- a/src/ast/compiler/captures.rs +++ b/src/ast/compiler/captures.rs @@ -63,20 +63,6 @@ impl CapturePass { info, }, - NodeKind::Pipe { - inputs, - lambda, - } => { - let mut t_inputs = Vec::with_capacity(inputs.len()); - for input in inputs { - t_inputs.push(Rc::new(Self::transform(input.as_ref().clone(), capture_map))); - } - NodeKind::Pipe { - inputs: t_inputs, - lambda: Rc::new(Self::transform(lambda.as_ref().clone(), capture_map)), - } - } - NodeKind::Block { exprs } => NodeKind::Block { exprs: exprs .into_iter() diff --git a/src/ast/compiler/dumper.rs b/src/ast/compiler/dumper.rs index 0c4735d..dbb66f8 100644 --- a/src/ast/compiler/dumper.rs +++ b/src/ast/compiler/dumper.rs @@ -138,16 +138,6 @@ impl Dumper { self.visit(args); self.indent -= 1; } - NodeKind::Pipe { inputs, lambda } => { - self.log("Pipe", node); - self.indent += 1; - for input in inputs { - self.visit(input); - } - self.visit(lambda); - self.indent -= 1; - } - NodeKind::Block { exprs } => { self.log("Block", node); self.indent += 1; diff --git a/src/ast/compiler/lowering.rs b/src/ast/compiler/lowering.rs index 0006ef1..13b6e58 100644 --- a/src/ast/compiler/lowering.rs +++ b/src/ast/compiler/lowering.rs @@ -99,20 +99,6 @@ impl Lowering { args: Rc::new(Self::transform(args.clone(), false, allocator)), } } - NodeKind::Pipe { - inputs, - lambda, - } => { - let mut t_inputs = Vec::with_capacity(inputs.len()); - for input in inputs { - t_inputs.push(Rc::new(Self::transform(input.clone(), false, allocator))); - } - NodeKind::Pipe { - inputs: t_inputs, - lambda: Rc::new(Self::transform(lambda.clone(), false, allocator)), - } - } - NodeKind::If { cond, then_br, diff --git a/src/ast/compiler/macros.rs b/src/ast/compiler/macros.rs index 1d347d5..3138f29 100644 --- a/src/ast/compiler/macros.rs +++ b/src/ast/compiler/macros.rs @@ -154,22 +154,6 @@ impl MacroExpander { }) } - SyntaxKind::Pipe { inputs, lambda } => { - let mut expanded_inputs = Vec::with_capacity(inputs.len()); - for input in inputs { - expanded_inputs.push(Rc::new(self.expand_recursive((*input).clone())?)); - } - let expanded_lambda = Rc::new(self.expand_recursive((*lambda).clone())?); - Ok(SyntaxNode { - identity: node.identity, - kind: SyntaxKind::Pipe { - inputs: expanded_inputs, - lambda: expanded_lambda, - }, - ty: (), - }) - } - SyntaxKind::Lambda { params, body, .. } => { self.registry.push(); let expanded_params = self.expand_recursive((*params).clone())?; @@ -536,22 +520,6 @@ impl MacroExpander { }) } - SyntaxKind::Pipe { inputs, lambda } => { - let mut expanded_inputs = Vec::with_capacity(inputs.len()); - for input in inputs { - expanded_inputs.push(Rc::new(self.expand_template((*input).clone(), state)?)); - } - let expanded_lambda = Rc::new(self.expand_template((*lambda).clone(), state)?); - Ok(SyntaxNode { - identity: node.identity, - kind: SyntaxKind::Pipe { - inputs: expanded_inputs, - lambda: expanded_lambda, - }, - ty: (), - }) - } - SyntaxKind::Lambda { params, body, .. } => { let expanded_params = self.expand_template((*params).clone(), state)?; let body = self.expand_template((*body).clone(), state)?; diff --git a/src/ast/compiler/optimizer/engine.rs b/src/ast/compiler/optimizer/engine.rs index 7c55641..53fda84 100644 --- a/src/ast/compiler/optimizer/engine.rs +++ b/src/ast/compiler/optimizer/engine.rs @@ -455,33 +455,6 @@ impl Optimizer { ) } - NodeKind::Pipe { - inputs, - lambda, - } => { - let mut o_inputs = Vec::with_capacity(inputs.len()); - let mut inputs_changed = false; - for input in inputs { - let opt = self.visit_node(input.clone(), sub, path); - if !Rc::ptr_eq(&opt, input) { - inputs_changed = true; - } - o_inputs.push(opt); - } - let o_lambda = self.visit_node(lambda.clone(), sub, path); - - if !inputs_changed && Rc::ptr_eq(&o_lambda, lambda) { - return node_rc; - } - - ( - NodeKind::Pipe { - inputs: o_inputs, - lambda: o_lambda, - }, - node.ty.clone(), - ) - } NodeKind::Block { exprs } => { let mut info = UsageInfo::default(); if !exprs.is_empty() { diff --git a/src/ast/compiler/optimizer/utils.rs b/src/ast/compiler/optimizer/utils.rs index 8399512..e26454e 100644 --- a/src/ast/compiler/optimizer/utils.rs +++ b/src/ast/compiler/optimizer/utils.rs @@ -158,12 +158,6 @@ impl UsageInfo { NodeKind::Again { args } => { self.collect(args); } - NodeKind::Pipe { inputs, lambda, .. } => { - for input in inputs { - self.collect(input); - } - self.collect(lambda); - } NodeKind::Nop | NodeKind::FieldAccessor(_) | NodeKind::Extension(_) diff --git a/src/ast/compiler/type_checker.rs b/src/ast/compiler/type_checker.rs index 1aa6361..b4c9434 100644 --- a/src/ast/compiler/type_checker.rs +++ b/src/ast/compiler/type_checker.rs @@ -550,42 +550,6 @@ impl TypeChecker { ) } - NodeKind::Pipe { inputs, lambda } => { - let mut typed_inputs = Vec::with_capacity(inputs.len()); - let mut arg_types = Vec::with_capacity(inputs.len()); - for input in inputs { - let typed_input = self.check_node(input, ctx, diag); - let arg_ty = if let StaticType::Series(inner) = &typed_input.ty { - *inner.clone() - } else if let StaticType::Stream(inner) = &typed_input.ty { - *inner.clone() - } else { - StaticType::Any - }; - arg_types.push(arg_ty); - typed_inputs.push(Rc::new(typed_input)); - } - - let typed_lambda = self.check_node_as_bound(lambda.as_ref(), &arg_types, diag); - - let ret_ty = if let StaticType::Function(sig) = &typed_lambda.ty { - if let StaticType::Optional(inner) = &sig.ret { - *inner.clone() - } else { - sig.ret.clone() - } - } else { - StaticType::Any - }; - ( - NodeKind::Pipe { - inputs: typed_inputs, - lambda: Rc::new(typed_lambda), - }, - StaticType::Stream(Box::new(ret_ty)), - ) - } - NodeKind::Block { exprs } => { let mut typed_exprs = Vec::new(); let mut last_ty = StaticType::Void; diff --git a/src/ast/nodes.rs b/src/ast/nodes.rs index 356ac8b..b00b8cd 100644 --- a/src/ast/nodes.rs +++ b/src/ast/nodes.rs @@ -389,10 +389,6 @@ pub enum NodeKind { Again { args: Rc>, }, - Pipe { - inputs: Vec>>, - lambda: Rc>, - }, Block { exprs: Vec>>, }, @@ -464,10 +460,6 @@ impl Clone for NodeKind

{ args: args.clone(), }, NodeKind::Again { args } => NodeKind::Again { args: args.clone() }, - NodeKind::Pipe { inputs, lambda } => NodeKind::Pipe { - inputs: inputs.clone(), - lambda: lambda.clone(), - }, NodeKind::Block { exprs } => NodeKind::Block { exprs: exprs.clone() }, NodeKind::Tuple { elements } => NodeKind::Tuple { elements: elements.clone() }, NodeKind::Record { fields, layout } => NodeKind::Record { @@ -525,11 +517,6 @@ impl PartialEq for NodeKind

{ Rc::ptr_eq(ca, cb) && Rc::ptr_eq(aa, ab) } (NodeKind::Again { args: aa }, NodeKind::Again { args: ab }) => Rc::ptr_eq(aa, ab), - (NodeKind::Pipe { inputs: ia, lambda: la }, NodeKind::Pipe { inputs: ib, lambda: lb }) => { - ia.len() == ib.len() - && ia.iter().zip(ib.iter()).all(|(a, b)| Rc::ptr_eq(a, b)) - && Rc::ptr_eq(la, lb) - } (NodeKind::Block { exprs: ea }, NodeKind::Block { exprs: eb }) => { ea.len() == eb.len() && ea.iter().zip(eb.iter()).all(|(a, b)| Rc::ptr_eq(a, b)) } @@ -572,7 +559,6 @@ impl NodeKind

{ NodeKind::Lambda { .. } => "LAMBDA".to_string(), NodeKind::Call { .. } => "CALL".to_string(), NodeKind::Again { .. } => "AGAIN".to_string(), - NodeKind::Pipe { .. } => "PIPE".to_string(), NodeKind::Block { .. } => "BLOCK".to_string(), NodeKind::Tuple { elements } => format!("TUPLE({})", elements.len()), NodeKind::Record { fields, .. } => format!("RECORD({})", fields.len()), diff --git a/src/ast/parser.rs b/src/ast/parser.rs index 3adedfe..f99373b 100644 --- a/src/ast/parser.rs +++ b/src/ast/parser.rs @@ -185,7 +185,6 @@ impl<'a> Parser<'a> { match symbol.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), @@ -286,21 +285,6 @@ impl<'a> Parser<'a> { } } - fn parse_pipe(&mut self, identity: Identity) -> SyntaxNode { - let inputs_node = self.parse_expression(); - let inputs = match inputs_node.kind { - SyntaxKind::Tuple { elements } => elements, - _ => vec![Rc::new(inputs_node)], - }; - let lambda = Rc::new(self.parse_expression()); - - SyntaxNode { - identity, - kind: SyntaxKind::Pipe { inputs, lambda }, - ty: (), - } - } - fn parse_fn(&mut self, identity: Identity) -> SyntaxNode { let params = Rc::new(self.parse_param_vector()); let body = self.parse_expression(); diff --git a/src/ast/rtl/streams.rs b/src/ast/rtl/streams.rs index 45d27eb..9d3c774 100644 --- a/src/ast/rtl/streams.rs +++ b/src/ast/rtl/streams.rs @@ -323,6 +323,25 @@ pub fn build_map_stream(input: Rc, field: Keyword) -> Stre use crate::ast::environment::Environment; use crate::ast::types::{Keyword, Purity, RecordLayout, Signature, StaticType}; +/// Resolves the return type of a `pipe` call from its argument types. +/// Argument layout: `Tuple([inputs, lambda])` where inputs is a Tuple of streams +/// or a single StreamNode, and lambda is a `Function U>`. +/// Returns `Stream`, unwrapping `Optional` to `U` (filter pattern). +fn pipe_type_resolver(args_ty: &StaticType) -> Option { + if let StaticType::Tuple(elements) = args_ty + && elements.len() == 2 + && let StaticType::Function(sig) = &elements[1] + { + let inner = if let StaticType::Optional(inner) = &sig.ret { + *inner.clone() + } else { + sig.ret.clone() + }; + return Some(StaticType::Stream(Box::new(inner))); + } + None +} + pub fn register(env: &Environment) { // (create-random-ohlc seed limit) -> StreamNode let generators = env.pipeline_generators.clone(); @@ -473,6 +492,64 @@ pub fn register(env: &Environment) { Value::Object(Rc::new(stream_node)) }, ); + + // (pipe inputs lambda) -> StreamNode + // inputs: a Tuple of StreamNodes or a single StreamNode + // lambda: a Closure or Function + let globals = env.root_values.clone(); + env.register_native_fn( + "pipe", + StaticType::PolymorphicFn(pipe_type_resolver), + Purity::Impure, + move |args: &[Value]| { + assert!(args.len() == 2, "pipe expects exactly 2 arguments (inputs, lambda)"); + + // Extract observable streams from args[0] + let obs_streams: Vec> = match &args[0] { + Value::Tuple(elements) => elements + .iter() + .map(|v| { + if let Value::Object(obj) = v + && let Some(sn) = obj.as_any().downcast_ref::() + { + sn.inner.clone() + } else { + panic!("pipe: each input must be a StreamNode"); + } + }) + .collect(), + Value::Object(obj) => { + if let Some(sn) = obj.as_any().downcast_ref::() { + vec![sn.inner.clone()] + } else { + panic!("pipe: input must be a StreamNode"); + } + } + _ => panic!("pipe: first argument must be a stream or tuple of streams"), + }; + + // Build the executor closure from args[1] + let mut pipe_vm = VM::new(globals.clone()); + let executor: Box = match &args[1] { + Value::Closure(rc) => { + let my_closure = rc.clone(); + Box::new(move |call_args: &[Value]| -> Value { + pipe_vm + .run_with_args(my_closure.clone(), call_args) + .unwrap_or_else(|e| panic!("Pipeline lambda execution failed: {}", e)) + }) + } + Value::Function(f) => { + let my_func = f.clone(); + Box::new(move |call_args: &[Value]| -> Value { (my_func.func)(call_args) }) + } + _ => panic!("pipe: second argument must be a function or closure"), + }; + + let node = build_pipeline_node(obs_streams, executor, &StaticType::Any); + Value::Object(node) + }, + ); } #[cfg(test)] diff --git a/src/ast/types.rs b/src/ast/types.rs index 80c607e..ae77722 100644 --- a/src/ast/types.rs +++ b/src/ast/types.rs @@ -315,6 +315,7 @@ pub struct Signature { } #[derive(Debug, Clone, PartialEq, Eq, Hash)] +#[allow(unpredictable_function_pointer_comparisons)] pub enum StaticType { Any, Void, @@ -336,6 +337,9 @@ pub enum StaticType { Function(Box), FunctionOverloads(Vec), Object(&'static str), + /// A polymorphic native function whose return type is computed from its argument types. + /// The function pointer receives the full argument type and returns the resolved return type. + PolymorphicFn(fn(&StaticType) -> Option), /// A diagnostic poison type, allowing type-checking to continue after an error. Error, } @@ -394,6 +398,7 @@ impl fmt::Display for StaticType { write!(f, "overloads({} variants)", sigs.len()) } StaticType::Object(name) => write!(f, "{}", name), + StaticType::PolymorphicFn(_) => write!(f, ""), StaticType::Error => write!(f, ""), } } @@ -516,6 +521,7 @@ impl StaticType { .find(|sig| sig.params == *args_ty) // 1. Try exact match first .or_else(|| sigs.iter().find(|sig| sig.params.is_assignable_from(args_ty))) // 2. Fallback to implicit coercion .map(|sig| sig.ret.clone()), + StaticType::PolymorphicFn(resolver) => resolver(args_ty), _ => None, } } diff --git a/src/ast/vm.rs b/src/ast/vm.rs index 62b7893..9432413 100644 --- a/src/ast/vm.rs +++ b/src/ast/vm.rs @@ -1,8 +1,8 @@ use crate::ast::closure::Closure; use crate::ast::nodes::{Address, ExecNode, IdentifierBinding, NodeKind, StackOffset}; use crate::ast::rtl::series::{RecordSeries, SeriesView}; -use crate::ast::rtl::streams::{build_map_stream, build_pipeline_node, StreamNode}; -use crate::ast::types::{PipeFn, Value}; +use crate::ast::rtl::streams::{build_map_stream, StreamNode}; +use crate::ast::types::Value; use std::cell::RefCell; use std::rc::Rc; @@ -392,57 +392,6 @@ impl VM { Ok(Value::Void) } } - NodeKind::Pipe { - inputs, - lambda, - } => { - let mut obs_streams = Vec::new(); - for input in inputs { - let val = self.eval_internal(obs, input)?; - if let Value::Object(obj) = val { - if let Some(s) = obj.as_any().downcast_ref::() { - obs_streams.push(s.inner.clone()); - } else { - return Err(format!( - "Pipe input must be a stream, found {}", - obj.type_name() - )); - } - } else { - return Err("Pipe input must be an object (stream)".to_string()); - } - } - - let lambda_val = self.eval_internal(obs, lambda)?; - - // Create the persistent execution closure for the PipeStream - let mut pipe_vm = VM::new(self.globals.clone()); - - let executor: Box = match lambda_val { - Value::Closure(rc) => { - let my_closure = rc.clone(); - Box::new(move |args: &[Value]| -> Value { - match pipe_vm.run_with_args(my_closure.clone(), args) { - Ok(res) => res, - Err(e) => panic!("Pipeline lambda execution failed: {}", e), - } - }) - } - Value::Function(func) => { - let my_func = func.clone(); - Box::new(move |args: &[Value]| -> Value { - (my_func.func)(args) - }) - } - _ => return Err("Pipe lambda must be a function/closure".to_string()), - }; - - // Delegate to the RTL Factory for specialized buffer instantiation - let node = - build_pipeline_node(obs_streams, executor, &node.ty.ty); - - Ok(Value::Object(node)) - } NodeKind::Block { exprs } => { let mut last = Value::Void; for e in exprs {