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

@@ -1,6 +1,6 @@
-;; Benchmark: 2.1us
+;; Benchmark: 2.1us
-;; Benchmark-Repeat: 921
+;; Benchmark-Repeat: 921
-;; Output: PipelineNode[last: Some(110.45243843391206)]
+;; Output: <StreamNode>
 
 (do
   (def src1 (create-random-ohlc 42 3))

examples/pipeline_script_ticker.myc

@@ -1,6 +1,6 @@
 ;; Benchmark: 3.2us
 ;; Benchmark-Repeat: 613
-;; Output: PipelineNode[last: Some(110.45243843391206)]
+;; Output: <StreamNode>
 
 (do
   ;; Set the random seed to match the existing test output exactly

examples/record_specialization_test.myc

@@ -1,11 +1,11 @@
-;; Benchmark: 1.5us
+;; Benchmark: 1.5us
-;; Benchmark-Repeat: 1348
+;; 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: <StreamNode>
-
+
 (do
   (def ohlc_stream (create-random-ohlc 42 10))
 

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="))
+;  (def bars (series 20 :float))
-  (pipe [(pipe [(.close src)] (SMA 100))] (printx "sma100="))
+;  (pipe [src] (fn [s] (do (push bars (.close s)))))
-)
+
+(cache 20 :float src)
+  
+)

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"})

rtl/sma.myc

@@ -1,5 +1,5 @@
 (do
-  (def SMA
+ (def SMA
     (fn [length]
       (do
         (def history (series :float))

src/ast/compiler/bound_nodes.rs

@@ -141,6 +141,7 @@ pub enum BoundKind<T = ()> {
     Again {
         args: Rc<BoundNode<T>>,
     },
+
     Pipe {
         inputs: Vec<Rc<BoundNode<T>>>,
         lambda: Rc<BoundNode<T>>,

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() {

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)),
                 )
             }
 

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
+   )
+)

src/ast/rtl/series.rs

@@ -18,32 +18,28 @@ use crate::ast::types::{Purity, Signature};
 pub struct RingBuffer<T> {
     buffer: VecDeque<T>,
     total_count: u64,
-}
+    lookback_limit: usize,
-
-impl<T> Default for RingBuffer<T> {
-    fn default() -> Self {
-        Self::new()
-    }
 }
 
 impl<T> RingBuffer<T> {
-    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<usize>) {
+    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() > self.lookback_limit {
-            while self.buffer.len() > limit {
+            self.buffer.pop_front();
-                self.buffer.pop_front();
-            }
         }
     }
 
@@ -107,9 +103,9 @@ pub struct ScalarSeries<T: ScalarValue> {
 }
 
 impl<T: ScalarValue> ScalarSeries<T> {
-    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<RingBuffer<Value>>,
 }
 
-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<usize>);
+    fn push_value(&self, value: Value);
 }
 
 impl SeriesMember for ScalarSeries<f64> {
-    fn push_value(&self, value: Value, limit: Option<usize>) {
+    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<i64> {
-    fn push_value(&self, value: Value, limit: Option<usize>) {
+    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<bool> {
-    fn push_value(&self, value: Value, limit: Option<usize>) {
+    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<usize>) {
+    fn push_value(&self, value: Value) {
-        self.data.borrow_mut().push(value, limit);
+        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<RecordLayout>) -> Self {
+    pub fn new(layout: Arc<RecordLayout>, lookback_limit: usize) -> Self {
         let mut fields: Vec<Rc<std::cell::RefCell<dyn SeriesMember>>> =
             Vec::with_capacity(layout.fields.len());
 
@@ -269,15 +259,17 @@ impl RecordSeries {
             let member: Rc<std::cell::RefCell<dyn SeriesMember>> = match static_type {
                 StaticType::Float => Rc::new(std::cell::RefCell::new(ScalarSeries::<f64>::new(
                     "FloatSeries",
+                    lookback_limit,
                 ))),
                 StaticType::Int | StaticType::DateTime => Rc::new(std::cell::RefCell::new(
-                    ScalarSeries::<i64>::new("IntSeries"),
+                    ScalarSeries::<i64>::new("IntSeries", lookback_limit),
                 )),
                 StaticType::Bool => Rc::new(std::cell::RefCell::new(ScalarSeries::<bool>::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::<f64>::new("FloatSeries"))),
+                        "float" => Value::Object(Rc::new(ScalarSeries::<f64>::new("FloatSeries", lookback_limit))),
-                        "int" | "datetime" => Value::Object(Rc::new(ScalarSeries::<i64>::new("IntSeries"))),
+                        "int" | "datetime" => Value::Object(Rc::new(ScalarSeries::<i64>::new("IntSeries", lookback_limit))),
-                        "bool" => Value::Object(Rc::new(ScalarSeries::<bool>::new("BoolSeries"))),
+                        "bool" => Value::Object(Rc::new(ScalarSeries::<bool>::new("BoolSeries", lookback_limit))),
-                        "text" => Value::Object(Rc::new(ValueSeries::new())),
+                        "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::<ScalarSeries<f64>>() {
-                    s.push_value(val.clone(), None);
+                    s.push_value(val.clone());
                 } else if let Some(s) = any.downcast_ref::<ScalarSeries<i64>>() {
-                    s.push_value(val.clone(), None);
+                    s.push_value(val.clone());
                 } else if let Some(s) = any.downcast_ref::<ScalarSeries<bool>>() {
-                    s.push_value(val.clone(), None);
+                    s.push_value(val.clone());
                 } else if let Some(s) = any.downcast_ref::<ValueSeries>() {
-                    s.push_value(val.clone(), None);
+                    s.push_value(val.clone());
                 } else {
                     panic!("Object is not a mutable series");
                 }

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<dyn ObservableStream>,
-    pub series: Rc<dyn crate::ast::types::Series>,
-}
-
-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<T: crate::ast::rtl::series::ScalarValue> {
     pub buffer: Rc<RefCell<crate::ast::rtl::series::RingBuffer<T>>>,
-    pub lookback: Option<usize>,
     pub extractor: fn(Value) -> Option<T>,
 }
 
 impl<T: crate::ast::rtl::series::ScalarValue> Observer for SeriesPusher<T> {
     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<T: crate::ast::rtl::series::ScalarValue> Observer for SeriesPusher<T> {
 /// A specialized observer that pushes incoming signals into a generic SharedValueSeries buffer.
 pub struct ValuePusher {
     pub buffer: Rc<RefCell<crate::ast::rtl::series::RingBuffer<Value>>>,
-    pub lookback: Option<usize>,
 }
 
 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<Rc<RefCell<dyn crate::ast::rtl::series::SeriesMember>>>,
-    pub lookback: Option<usize>,
 }
 
 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<Rc<dyn ObservableStream>>,
     executor: Box<PipeFn>,
-    out_type: &StaticType,
+    _out_type: &StaticType,
-) -> Rc<PipelineNode> {
+) -> Rc<StreamNode> {
-    use crate::ast::rtl::series::*;
-
     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<dyn crate::ast::types::Series> = match out_type {
+    Rc::new(StreamNode { inner: pipe })
-        StaticType::Float => {
-            let buffer = Rc::new(RefCell::new(RingBuffer::<f64>::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::<i64>::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::<bool>::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<RefCell<dyn SeriesMember>> = match ty {
-                    StaticType::Float => {
-                        Rc::new(RefCell::new(ScalarSeries::<f64>::new("FloatSeries")))
-                    }
-                    StaticType::Int | StaticType::DateTime => {
-                        Rc::new(RefCell::new(ScalarSeries::<i64>::new("IntSeries")))
-                    }
-                    StaticType::Bool => {
-                        Rc::new(RefCell::new(ScalarSeries::<bool>::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::<Value>::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,
-    })
 }
 
 pub fn build_map_stream(input: Rc<dyn ObservableStream>, field: Keyword) -> StreamNode {

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, "<native fn, {:?}>", f_meta.purity),
             Value::Object(o) => {
-                if let Some(pipe) = o
+                if let Some(val_series) =
-                    .as_any()
-                    .downcast_ref::<crate::ast::rtl::streams::PipelineNode>()
-                {
-                    write!(f, "PipelineNode[last: {:?}]", pipe.series.get_item(0))
-                } else if let Some(val_series) =
                     o.as_any()
                         .downcast_ref::<crate::ast::rtl::series::SharedValueSeries>()
                 {

src/ast/vm.rs

@@ -387,9 +387,6 @@ impl VM {
                         } else if let Some(sn) = obj.as_any().downcast_ref::<crate::ast::rtl::streams::StreamNode>() {
                             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::<crate::ast::rtl::streams::PipelineNode>() {
-                            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::<StreamNode>() {
                             obs_streams.push(s.inner.clone());
-                        } else if let Some(p) = obj.as_any().downcast_ref::<PipelineNode>() {
-                            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::<crate::ast::rtl::streams::StreamNode>() {
                                         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::<crate::ast::rtl::streams::PipelineNode>() {
-                                        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::<crate::ast::rtl::streams::StreamNode>() {
                                         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::<crate::ast::rtl::streams::PipelineNode>() {
-                                        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 {}",

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] (MY_SMA 5))]
-  [(pipe [s] (SMA 20))]
+  [(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.