From 595bcf09e5a57fc0873e808735f3d7cd08cc9f7d Mon Sep 17 00:00:00 2001 From: Michael Schimmel Date: Sat, 7 Mar 2026 23:48:09 +0100 Subject: [PATCH] Update examples to use StreamNode and Series with lookback The `PipelineNode` has been refactored into `StreamNode`. This commit updates the example files to reflect this change and also updates the `series` constructor to accept a lookback parameter. The `PipelineNode` was an artifact of a previous implementation and is no longer needed. The `StreamNode` is the appropriate type for representing the output of a pipe operation. The `series` function now requires a `lookback` argument to specify the maximum number of items to store. This makes the behavior of series more explicit and prevents accidental unbounded memory growth. --- examples/pipeline.myc | 6 +- examples/pipeline_script_ticker.myc | 2 +- examples/record_specialization_test.myc | 8 +- examples/sma.myc | 24 ++-- examples/soa_series.myc | 2 +- rtl/sma.myc | 2 +- src/ast/compiler/bound_nodes.rs | 1 + src/ast/compiler/optimizer/engine.rs | 3 +- src/ast/compiler/type_checker.rs | 2 +- src/ast/rtl/prelude.myc | 22 ++++ src/ast/rtl/series.rs | 91 +++++++-------- src/ast/rtl/streams.rs | 145 +----------------------- src/ast/types.rs | 7 +- src/ast/vm.rs | 13 +-- src/integration_test.rs | 13 +-- 15 files changed, 108 insertions(+), 233 deletions(-) diff --git a/examples/pipeline.myc b/examples/pipeline.myc index 69ca49c..5516b74 100644 --- a/examples/pipeline.myc +++ b/examples/pipeline.myc @@ -1,6 +1,6 @@ -;; Benchmark: 2.1us -;; Benchmark-Repeat: 921 -;; Output: PipelineNode[last: Some(110.45243843391206)] +;; Benchmark: 2.1us +;; Benchmark-Repeat: 921 +;; Output: (do (def src1 (create-random-ohlc 42 3)) diff --git a/examples/pipeline_script_ticker.myc b/examples/pipeline_script_ticker.myc index 1887fa6..10b13b3 100644 --- a/examples/pipeline_script_ticker.myc +++ b/examples/pipeline_script_ticker.myc @@ -1,6 +1,6 @@ ;; Benchmark: 3.2us ;; Benchmark-Repeat: 613 -;; Output: PipelineNode[last: Some(110.45243843391206)] +;; Output: (do ;; Set the random seed to match the existing test output exactly diff --git a/examples/record_specialization_test.myc b/examples/record_specialization_test.myc index 3020c90..3fd677f 100644 --- a/examples/record_specialization_test.myc +++ b/examples/record_specialization_test.myc @@ -1,11 +1,11 @@ -;; Benchmark: 1.5us -;; Benchmark-Repeat: 1348 +;; Benchmark: 1.5us +;; Benchmark-Repeat: 1348 ;; Test Record SoA specialization in Pipelines ;; Dank der neuen Spezialisierung wird hierfür im Hintergrund ;; eine SharedRecordSeries mit SoA-Layout (Float-Puffer für mid und range) erstellt. -;; Output: PipelineNode[last: Some({:mid 101.35623617501585, :range 0.9314056584827455})] - +;; Output: + (do (def ohlc_stream (create-random-ohlc 42 10)) diff --git a/examples/sma.myc b/examples/sma.myc index 7a30853..c5790a1 100644 --- a/examples/sma.myc +++ b/examples/sma.myc @@ -1,12 +1,22 @@ #use rtl (do - (def src (create-random-ohlc 42 200)) + (def src (create-random-ohlc 42 20000)) + + (macro printx [s] `(fn [x] (print ~s x))) + +; (pipe src print) +; (pipe [(pipe [(.close src)] (SMA 20))] (printx "sma20=")) +; (pipe [(pipe [(.close src)] (SMA 5))] (printx "sma5=")) +; (pipe [(pipe [(.close src)] (SMA 100))] (printx "sma100=")) - (macro printx [s] `(fn [x] (print ~s x))) +(macro cache [lookback type src] `(do (def data (series ~lookback ~type)) (pipe [~src] (fn [s] (do (push data s)))) data)) - (pipe src print) - (pipe [(pipe [(.close src)] (SMA 20))] (printx "sma20=")) - (pipe [(pipe [(.close src)] (SMA 5))] (printx "sma5=")) - (pipe [(pipe [(.close src)] (SMA 100))] (printx "sma100=")) -) + + +; (def bars (series 20 :float)) +; (pipe [src] (fn [s] (do (push bars (.close s))))) + +(cache 20 :float src) + +) \ No newline at end of file diff --git a/examples/soa_series.myc b/examples/soa_series.myc index 8f02c15..b6713cc 100644 --- a/examples/soa_series.myc +++ b/examples/soa_series.myc @@ -2,7 +2,7 @@ ;; Benchmark: 1.4us ;; Benchmark-Repeat: 1406 (do - (def my_ticks (series {:price :float :volume :int :msg :text})) + (def my_ticks (series 100 {:price :float :volume :int :msg :text})) (push my_ticks {:price 10.5 :volume 100 :msg "A"}) (push my_ticks {:price 11.2 :volume 200 :msg "B"}) diff --git a/rtl/sma.myc b/rtl/sma.myc index cca8d12..8ee3190 100644 --- a/rtl/sma.myc +++ b/rtl/sma.myc @@ -1,5 +1,5 @@ (do - (def SMA + (def SMA (fn [length] (do (def history (series :float)) diff --git a/src/ast/compiler/bound_nodes.rs b/src/ast/compiler/bound_nodes.rs index fe650f6..acf1bb6 100644 --- a/src/ast/compiler/bound_nodes.rs +++ b/src/ast/compiler/bound_nodes.rs @@ -141,6 +141,7 @@ pub enum BoundKind { Again { args: Rc>, }, + Pipe { inputs: Vec>>, lambda: Rc>, diff --git a/src/ast/compiler/optimizer/engine.rs b/src/ast/compiler/optimizer/engine.rs index e6c8192..d26d513 100644 --- a/src/ast/compiler/optimizer/engine.rs +++ b/src/ast/compiler/optimizer/engine.rs @@ -439,7 +439,7 @@ impl Optimizer { 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; } @@ -453,7 +453,6 @@ impl Optimizer { node.ty.clone(), ) } - BoundKind::Block { exprs } => { let mut info = UsageInfo::default(); if !exprs.is_empty() { diff --git a/src/ast/compiler/type_checker.rs b/src/ast/compiler/type_checker.rs index 686006d..d299de2 100644 --- a/src/ast/compiler/type_checker.rs +++ b/src/ast/compiler/type_checker.rs @@ -453,7 +453,7 @@ impl TypeChecker { lambda: Rc::new(typed_lambda), out_type: ret_ty.clone(), }, - StaticType::Series(Box::new(ret_ty)), + StaticType::Stream(Box::new(ret_ty)), ) } diff --git a/src/ast/rtl/prelude.myc b/src/ast/rtl/prelude.myc index 2854a43..c6402cd 100644 --- a/src/ast/rtl/prelude.myc +++ b/src/ast/rtl/prelude.myc @@ -7,3 +7,25 @@ (do ~body (again)) ))) ) + +;; Creates a stateful cache (Series) from a stateless stream. +;; +;; Pipes are highly optimized, stateless stream-routers that do not allocate memory +;; for historical data. If multiple indicators need to read historical data from +;; a stream, or if you need to access previous items via index (e.g. `(my_data 5)`), +;; you must explicitly cache the stream. +;; +;; Parameters: +;; lookback: The maximum number of items to keep in memory (e.g. 100). +;; type: The data type or schema of the series (e.g. :float or {:price :float}). +;; src: The source stream to observe and cache. +;; +;; Returns: +;; A stateful Series object that automatically updates when the source stream fires. +(macro cache [lookback type src] + `(do + (def data (series ~lookback ~type)) + (pipe [~src] (fn [s] (do (push data s)))) + data + ) +) diff --git a/src/ast/rtl/series.rs b/src/ast/rtl/series.rs index f0f802c..cb2b382 100644 --- a/src/ast/rtl/series.rs +++ b/src/ast/rtl/series.rs @@ -18,32 +18,28 @@ use crate::ast::types::{Purity, Signature}; pub struct RingBuffer { buffer: VecDeque, total_count: u64, -} - -impl Default for RingBuffer { - fn default() -> Self { - Self::new() - } + lookback_limit: usize, } impl RingBuffer { - pub fn new() -> Self { + pub fn new(lookback_limit: usize) -> Self { + // Pre-allocate up to 500 elements to minimize heap fragmentation + let initial_capacity = std::cmp::min(lookback_limit, 500); Self { - buffer: VecDeque::new(), + buffer: VecDeque::with_capacity(initial_capacity), total_count: 0, + lookback_limit, } } /// Adds an element. Removes the oldest if the lookback limit is reached. #[inline] - pub fn push(&mut self, item: T, lookback: Option) { + pub fn push(&mut self, item: T) { self.buffer.push_back(item); self.total_count += 1; - if let Some(limit) = lookback { - while self.buffer.len() > limit { - self.buffer.pop_front(); - } + while self.buffer.len() > self.lookback_limit { + self.buffer.pop_front(); } } @@ -107,9 +103,9 @@ pub struct ScalarSeries { } impl ScalarSeries { - pub fn new(type_name: &'static str) -> Self { + pub fn new(type_name: &'static str, lookback_limit: usize) -> Self { Self { - data: std::cell::RefCell::new(RingBuffer::new()), + data: std::cell::RefCell::new(RingBuffer::new(lookback_limit)), type_name, } } @@ -161,16 +157,10 @@ pub struct ValueSeries { pub data: std::cell::RefCell>, } -impl Default for ValueSeries { - fn default() -> Self { - Self::new() - } -} - impl ValueSeries { - pub fn new() -> Self { + pub fn new(lookback_limit: usize) -> Self { Self { - data: std::cell::RefCell::new(RingBuffer::new()), + data: std::cell::RefCell::new(RingBuffer::new(lookback_limit)), } } } @@ -213,36 +203,36 @@ impl Series for ValueSeries { /// of the exact type (Type Erasure for member arrays). pub trait SeriesMember: Object + Series { /// Pushes a dynamic Value into the series, performing the necessary downcast. - fn push_value(&self, value: Value, limit: Option); + fn push_value(&self, value: Value); } impl SeriesMember for ScalarSeries { - fn push_value(&self, value: Value, limit: Option) { + fn push_value(&self, value: Value) { if let Some(v) = value.as_float() { - self.data.borrow_mut().push(v, limit); + self.data.borrow_mut().push(v); } } } impl SeriesMember for ScalarSeries { - fn push_value(&self, value: Value, limit: Option) { + fn push_value(&self, value: Value) { if let Some(v) = value.as_int() { - self.data.borrow_mut().push(v, limit); + self.data.borrow_mut().push(v); } } } impl SeriesMember for ScalarSeries { - fn push_value(&self, value: Value, limit: Option) { + fn push_value(&self, value: Value) { if let Value::Bool(v) = value { - self.data.borrow_mut().push(v, limit); + self.data.borrow_mut().push(v); } } } impl SeriesMember for ValueSeries { - fn push_value(&self, value: Value, limit: Option) { - self.data.borrow_mut().push(value, limit); + fn push_value(&self, value: Value) { + self.data.borrow_mut().push(value); } } @@ -260,7 +250,7 @@ pub struct RecordSeries { impl RecordSeries { /// Creates a new RecordSeries matching the given layout. - pub fn new(layout: Arc) -> Self { + pub fn new(layout: Arc, lookback_limit: usize) -> Self { let mut fields: Vec>> = Vec::with_capacity(layout.fields.len()); @@ -269,15 +259,17 @@ impl RecordSeries { let member: Rc> = match static_type { StaticType::Float => Rc::new(std::cell::RefCell::new(ScalarSeries::::new( "FloatSeries", + lookback_limit, ))), StaticType::Int | StaticType::DateTime => Rc::new(std::cell::RefCell::new( - ScalarSeries::::new("IntSeries"), + ScalarSeries::::new("IntSeries", lookback_limit), )), StaticType::Bool => Rc::new(std::cell::RefCell::new(ScalarSeries::::new( "BoolSeries", + lookback_limit, ))), // Fallback for everything else (Text, Lists, nested Records without SoA, etc.) - _ => Rc::new(std::cell::RefCell::new(ValueSeries::new())), + _ => Rc::new(std::cell::RefCell::new(ValueSeries::new(lookback_limit))), }; fields.push(member); } @@ -547,23 +539,24 @@ impl Series for SharedRecordSeries { // ============================================================================ pub fn register(env: &Environment) { - // (series template_or_type_keyword) -> Series + // (series lookback_limit template_or_type_keyword) -> Series env.register_native_fn( "series", StaticType::Function(Box::new(Signature { - params: StaticType::Tuple(vec![StaticType::Any]), + params: StaticType::Tuple(vec![StaticType::Int, StaticType::Any]), ret: StaticType::Any, })), Purity::Impure, |args: &[Value]| { - let arg = &args[0]; + let lookback_limit = args[0].as_int().unwrap_or(0) as usize; + let arg = &args[1]; match arg { Value::Keyword(k) => { match k.name().to_lowercase().as_str() { - "float" => Value::Object(Rc::new(ScalarSeries::::new("FloatSeries"))), - "int" | "datetime" => Value::Object(Rc::new(ScalarSeries::::new("IntSeries"))), - "bool" => Value::Object(Rc::new(ScalarSeries::::new("BoolSeries"))), - "text" => Value::Object(Rc::new(ValueSeries::new())), + "float" => Value::Object(Rc::new(ScalarSeries::::new("FloatSeries", lookback_limit))), + "int" | "datetime" => Value::Object(Rc::new(ScalarSeries::::new("IntSeries", lookback_limit))), + "bool" => Value::Object(Rc::new(ScalarSeries::::new("BoolSeries", lookback_limit))), + "text" => Value::Object(Rc::new(ValueSeries::new(lookback_limit))), _ => panic!("Unknown or unsupported series type keyword: :{}", k.name()), } } @@ -586,9 +579,9 @@ pub fn register(env: &Environment) { fields.push((*name, ty)); } let final_layout = RecordLayout::get_or_create(fields); - Value::Object(Rc::new(RecordSeries::new(final_layout))) + Value::Object(Rc::new(RecordSeries::new(final_layout, lookback_limit))) } - _ => panic!("series expects a type keyword (:float, :int...) or a schema record (e.g., {{:price :float}})"), + _ => panic!("series expects a lookback_limit and a type keyword (:float, :int...) or a schema record (e.g., {{:price :float}})"), } }, ); @@ -612,20 +605,20 @@ pub fn register(env: &Environment) { if let Value::Record(_, values) = val { for (i, v) in values.iter().enumerate() { if let Some(f) = rs.fields.get(i) { - f.borrow().push_value(v.clone(), None); + f.borrow().push_value(v.clone()); } } } else { panic!("push to RecordSeries expects a record"); } } else if let Some(s) = any.downcast_ref::>() { - s.push_value(val.clone(), None); + s.push_value(val.clone()); } else if let Some(s) = any.downcast_ref::>() { - s.push_value(val.clone(), None); + s.push_value(val.clone()); } else if let Some(s) = any.downcast_ref::>() { - s.push_value(val.clone(), None); + s.push_value(val.clone()); } else if let Some(s) = any.downcast_ref::() { - s.push_value(val.clone(), None); + s.push_value(val.clone()); } else { panic!("Object is not a mutable series"); } diff --git a/src/ast/rtl/streams.rs b/src/ast/rtl/streams.rs index 7c48c64..9063abd 100644 --- a/src/ast/rtl/streams.rs +++ b/src/ast/rtl/streams.rs @@ -71,32 +71,6 @@ impl crate::ast::types::Object for StreamNode { } } -/// A node that represents the result of a `pipe` statement. -/// It acts as a Stream (to connect downstream pipes) AND as a Series (for script lookbacks). -#[derive(Clone)] -pub struct PipelineNode { - pub stream: Rc, - pub series: Rc, -} - -impl std::fmt::Debug for PipelineNode { - fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - write!(f, "PipelineNode") - } -} - -impl crate::ast::types::Object for PipelineNode { - fn type_name(&self) -> &'static str { - "PipelineNode" - } - fn as_any(&self) -> &dyn std::any::Any { - self - } - fn as_series(&self) -> Option<&dyn crate::ast::types::Series> { - Some(self.series.as_ref()) - } -} - /// The RootStream is the "Clock" and data source of the entire pipeline. /// It generates the monotonic `cycle_id` and triggers the observers. pub struct RootStream { @@ -245,14 +219,13 @@ impl Observer for PipeStream { /// A specialized observer that pushes incoming signals into a SharedSeries buffer. pub struct SeriesPusher { pub buffer: Rc>>, - pub lookback: Option, pub extractor: fn(Value) -> Option, } impl Observer for SeriesPusher { fn notify(&mut self, _source_index: usize, _cycle_id: u64, value: Value) { if let Some(v) = (self.extractor)(value) { - self.buffer.borrow_mut().push(v, self.lookback); + self.buffer.borrow_mut().push(v); } } } @@ -260,19 +233,17 @@ impl Observer for SeriesPusher { /// A specialized observer that pushes incoming signals into a generic SharedValueSeries buffer. pub struct ValuePusher { pub buffer: Rc>>, - pub lookback: Option, } impl Observer for ValuePusher { fn notify(&mut self, _source_index: usize, _cycle_id: u64, value: Value) { - self.buffer.borrow_mut().push(value, self.lookback); + self.buffer.borrow_mut().push(value); } } /// A specialized observer that splits a Record into its fields and pushes them into SoA buffers. pub struct RecordPusher { pub field_buffers: Vec>>, - pub lookback: Option, } impl Observer for RecordPusher { @@ -280,7 +251,7 @@ impl Observer for RecordPusher { if let Value::Record(_, values) = value { for (i, v) in values.iter().enumerate() { if let Some(buf) = self.field_buffers.get(i) { - buf.borrow_mut().push_value(v.clone(), self.lookback); + buf.borrow_mut().push_value(v.clone()); } } } @@ -292,10 +263,8 @@ impl Observer for RecordPusher { pub fn build_pipeline_node( inputs: Vec>, executor: Box, - out_type: &StaticType, -) -> Rc { - use crate::ast::rtl::series::*; - + _out_type: &StaticType, +) -> Rc { let pipe = Rc::new(RefCell::new(PipeStream::new( "pipe".to_string(), inputs.len(), @@ -311,109 +280,7 @@ pub fn build_pipeline_node( input.add_observer(adapter); } - let series: Rc = match out_type { - StaticType::Float => { - let buffer = Rc::new(RefCell::new(RingBuffer::::new())); - let pusher = Rc::new(RefCell::new(SeriesPusher { - buffer: buffer.clone(), - lookback: Some(100), // TODO: Configuration - extractor: |v| { - if let Value::Float(f) = v { - Some(f) - } else { - None - } - }, - })); - pipe.borrow().add_observer(pusher); - Rc::new(SharedSeries { - buffer, - type_name: "SharedFloatSeries", - converter: |v| Value::Float(v), - }) - } - StaticType::Int | StaticType::DateTime => { - let buffer = Rc::new(RefCell::new(RingBuffer::::new())); - let pusher = Rc::new(RefCell::new(SeriesPusher { - buffer: buffer.clone(), - lookback: Some(100), - extractor: |v| match v { - Value::Int(i) => Some(i), - Value::DateTime(i) => Some(i), - _ => None, - }, - })); - pipe.borrow().add_observer(pusher); - Rc::new(SharedSeries { - buffer, - type_name: "SharedIntSeries", - converter: |v| Value::Int(v), - }) - } - StaticType::Bool => { - let buffer = Rc::new(RefCell::new(RingBuffer::::new())); - let pusher = Rc::new(RefCell::new(SeriesPusher { - buffer: buffer.clone(), - lookback: Some(100), - extractor: |v| { - if let Value::Bool(b) = v { - Some(b) - } else { - None - } - }, - })); - pipe.borrow().add_observer(pusher); - Rc::new(SharedSeries { - buffer, - type_name: "SharedBoolSeries", - converter: |v| Value::Bool(v), - }) - } - StaticType::Record(layout) => { - let mut field_buffers = Vec::with_capacity(layout.fields.len()); - for (_, ty) in &layout.fields { - let member: Rc> = match ty { - StaticType::Float => { - Rc::new(RefCell::new(ScalarSeries::::new("FloatSeries"))) - } - StaticType::Int | StaticType::DateTime => { - Rc::new(RefCell::new(ScalarSeries::::new("IntSeries"))) - } - StaticType::Bool => { - Rc::new(RefCell::new(ScalarSeries::::new("BoolSeries"))) - } - _ => Rc::new(RefCell::new(ValueSeries::new())), - }; - field_buffers.push(member); - } - - let pusher = Rc::new(RefCell::new(RecordPusher { - field_buffers: field_buffers.clone(), - lookback: Some(100), - })); - pipe.borrow().add_observer(pusher); - - Rc::new(SharedRecordSeries { - layout: layout.clone(), - field_buffers, - }) - } - _ => { - let buffer = Rc::new(RefCell::new(RingBuffer::::new())); - let pusher = Rc::new(RefCell::new(ValuePusher { - buffer: buffer.clone(), - lookback: Some(100), - })); - pipe.borrow().add_observer(pusher); - Rc::new(SharedValueSeries { buffer }) - } - }; - - Rc::new(PipelineNode { - stream: pipe, - series, - }) + Rc::new(StreamNode { inner: pipe }) } pub fn build_map_stream(input: Rc, field: Keyword) -> StreamNode { diff --git a/src/ast/types.rs b/src/ast/types.rs index 540e741..42f3d19 100644 --- a/src/ast/types.rs +++ b/src/ast/types.rs @@ -656,12 +656,7 @@ impl fmt::Display for Value { Value::FieldAccessor(k) => write!(f, ".{}", k.name()), Value::Function(f_meta) => write!(f, "", f_meta.purity), Value::Object(o) => { - if let Some(pipe) = o - .as_any() - .downcast_ref::() - { - write!(f, "PipelineNode[last: {:?}]", pipe.series.get_item(0)) - } else if let Some(val_series) = + if let Some(val_series) = o.as_any() .downcast_ref::() { diff --git a/src/ast/vm.rs b/src/ast/vm.rs index 236cdba..efdbf21 100644 --- a/src/ast/vm.rs +++ b/src/ast/vm.rs @@ -387,9 +387,6 @@ impl VM { } else if let Some(sn) = obj.as_any().downcast_ref::() { let mapped = crate::ast::rtl::streams::build_map_stream(sn.inner.clone(), *field); return Ok(Value::Object(Rc::new(mapped))); - } else if let Some(pn) = obj.as_any().downcast_ref::() { - let mapped = crate::ast::rtl::streams::build_map_stream(pn.stream.clone(), *field); - return Ok(Value::Object(Rc::new(mapped))); } // Fallback if it's another type of object that is not a RecordSeries. @@ -431,7 +428,7 @@ impl VM { lambda, out_type, } => { - use crate::ast::rtl::streams::{PipelineNode, StreamNode}; + use crate::ast::rtl::streams::StreamNode; let mut obs_streams = Vec::new(); for input in inputs { @@ -439,8 +436,6 @@ impl VM { if let Value::Object(obj) = val { if let Some(s) = obj.as_any().downcast_ref::() { obs_streams.push(s.inner.clone()); - } else if let Some(p) = obj.as_any().downcast_ref::() { - obs_streams.push(p.stream.clone()); } else { return Err(format!( "Pipe input must be a stream, found {}", @@ -565,9 +560,6 @@ impl VM { } else if let Some(sn) = obj.as_any().downcast_ref::() { let mapped = crate::ast::rtl::streams::build_map_stream(sn.inner.clone(), k); return Ok(Value::Object(Rc::new(mapped))); - } else if let Some(pn) = obj.as_any().downcast_ref::() { - let mapped = crate::ast::rtl::streams::build_map_stream(pn.stream.clone(), k); - return Ok(Value::Object(Rc::new(mapped))); } return Err(format!( "Field accessor .{} expects a record, RecordSeries or Stream, got {}", @@ -637,9 +629,6 @@ impl VM { } else if let Some(sn) = obj.as_any().downcast_ref::() { let mapped = crate::ast::rtl::streams::build_map_stream(sn.inner.clone(), *k); Ok(Value::Object(Rc::new(mapped))) - } else if let Some(pn) = obj.as_any().downcast_ref::() { - let mapped = crate::ast::rtl::streams::build_map_stream(pn.stream.clone(), *k); - Ok(Value::Object(Rc::new(mapped))) } else { Err(format!( "Field accessor .{} expects a record, RecordSeries or Stream, got {}", diff --git a/src/integration_test.rs b/src/integration_test.rs index 804b5dd..e11dde9 100644 --- a/src/integration_test.rs +++ b/src/integration_test.rs @@ -391,9 +391,9 @@ mod tests { let val = res.unwrap(); if let crate::ast::types::Value::Object(obj) = val { - assert_eq!(obj.type_name(), "PipelineNode"); + assert_eq!(obj.type_name(), "StreamNode"); } else { - panic!("Expected an Object(PipelineNode)"); + panic!("Expected an Object(StreamNode)"); } } @@ -772,7 +772,7 @@ mod tests { (do (def outer (fn [length] (do - (def history (series :float)) + (def history (series 100 :float)) (fn [val] length) ) )) @@ -794,7 +794,7 @@ mod tests { (def MY_SMA (fn [length] (do - (def history (series :float)) + (def history (series 100 :float)) (fn [val] (do (push history val) @@ -803,9 +803,8 @@ mod tests { (def src (create-random-ohlc 42 100)) (def s (.close src)) - [(pipe [s] (MY_SMA 5))] - [(pipe [s] (SMA 20))] -) + [(pipe [s] (MY_SMA 5))] + [(pipe [s] (SMA 20))]) "#; // Note: Using SMA from RTL and MY_SMA locally to mix it up. // Actually, for full independence, let's use MY_SMA twice.