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 {