From f7cb6655af4b09eb025183263d841d739a45dc27 Mon Sep 17 00:00:00 2001 From: Michael Schimmel Date: Mon, 2 Mar 2026 22:03:24 +0100 Subject: [PATCH] feat: Specialize pipeline types for performance Introduces type specialization for reactive pipeline nodes and their data buffers. This eliminates the overhead of generic `Value` enums and `SharedValueSeries` when dealing with known scalar or record types. The architecture shifts buffer instantiation from the VM to the runtime (RTL), leveraging type information from the AST. A new `out_type` field is added to `BoundKind::Pipe` to store the static type of the pipeline's output, determined by the type checker. This enables the creation of specialized `RingBuffer` and `SharedRecordSeries` (for Struct-of-Arrays layout) when the output type is known, significantly improving memory usage and processing speed for time-series data, especially in financial analysis. --- docs/Reactive_Pipeline_Type_Specialization.md | 73 +++++++++ examples/object_records.myc | 4 +- examples/record_specialization_test.myc | 23 +++ src/ast/compiler/analyzer.rs | 3 +- src/ast/compiler/binder.rs | 1 + src/ast/compiler/bound_nodes.rs | 1 + src/ast/compiler/captures.rs | 3 +- src/ast/compiler/dumper.rs | 2 +- src/ast/compiler/optimizer/engine.rs | 3 +- src/ast/compiler/optimizer/utils.rs | 2 +- src/ast/compiler/tco.rs | 3 +- src/ast/compiler/type_checker.rs | 19 ++- src/ast/rtl/streams.rs | 148 +++++++++++++++++- src/ast/vm.rs | 68 +++----- 14 files changed, 295 insertions(+), 58 deletions(-) create mode 100644 docs/Reactive_Pipeline_Type_Specialization.md create mode 100644 examples/record_specialization_test.myc diff --git a/docs/Reactive_Pipeline_Type_Specialization.md b/docs/Reactive_Pipeline_Type_Specialization.md new file mode 100644 index 0000000..14567db --- /dev/null +++ b/docs/Reactive_Pipeline_Type_Specialization.md @@ -0,0 +1,73 @@ +# Typ-Spezialisierung der Reaktiven Pipeline + +Dieses Dokument beschreibt die Architektur und Motivation hinter der typspezifischen Initialisierung von Pipeline-Knoten (`PipeStream`) und deren Datenpuffern (`SharedSeries`). + +## 1. Motivation + +In der Myc Script Engine ist die reaktive Pipeline (Pipes und Streams) das zentrale Konstrukt für die Verarbeitung von Zeitreihendaten (z.B. Finanzdaten, Indikatoren). Eine Pipe konsumiert Daten aus einem oder mehreren Streams, führt ein Lambda aus und speichert das Ergebnis in einem Ringpuffer, der wiederum als Stream für nachfolgende Pipes oder als Serie für Lookbacks (`series[0]`) dient. + +### Das Problem der generischen Puffer +Ohne Typinformationen muss die VM zur Laufzeit einen generischen Puffer (`RingBuffer`) und eine `SharedValueSeries` instanziieren. Dies hat mehrere Nachteile: +* **Speicher-Overhead:** Jedes Element ist ein `Value`-Enum, das mehr Platz verbraucht als ein nativer Typ wie `f64`. +* **Performance-Overhead:** Bei jedem Zugriff (Lookback) oder Push muss das `Value`-Enum verarbeitet werden. +* **Kein Struct-of-Arrays (SoA):** Der wichtigste Anwendungsfall in der Finanzanalyse sind Indikatoren, die mehrere Werte produzieren (z.B. MACD mit `macd`, `signal` und `hist`). In einer generischen Architektur entstünde hier ein Puffer aus Record-Objekten (Array of Structs). Für die Performance und Vektorisierung ist jedoch ein "Struct of Arrays" (SoA) zwingend erforderlich, bei dem jedes Record-Feld seinen eigenen, flachen `RingBuffer` erhält. + +### Die Rolle des TypeCheckers und der VM +Der `TypeChecker` analysiert bereits erfolgreich den Rückgabetyp der Lambdas innerhalb einer Pipe und kennt somit den exakten Typ der resultierenden Serie (z.B. `StaticType::Float` oder `StaticType::Record`). +Die Virtual Machine (VM) hingegen sollte so "dumm", universell und schnell wie möglich bleiben. Sie sollte keine komplexen Typbäume (`StaticType`) auswerten oder Puffer-Strategien verwalten müssen. + +## 2. Design-Ziele + +1. **Dumb VM:** Die VM soll keine Logik zur Auswahl oder Instanziierung von typisierten Ringpuffern enthalten. +2. **First-Class Record Support:** Wenn eine Pipe Records produziert (wie beim MACD), muss das System automatisch eine `SharedRecordSeries` aufbauen, die intern für jedes Record-Feld einen eigenen, typisierten `RingBuffer` (SoA) verwendet. +3. **Zero-Cost Abstractions für Skalare:** Skalare Werte (`f64`, `i64`, `bool`) sollen in flachen, nativen Arrays (`RingBuffer`) landen. +4. **Klare Trennung von Zuständigkeiten:** Das RTL-Modul für Streams (`src/ast/rtl/streams.rs`) besitzt das Domänenwissen über Puffer und Observer, nicht der Compiler und nicht die VM. + +## 3. Architektur und Umsetzung + +Um diese Ziele zu erreichen, wird die Puffer-Initialisierung vollständig aus der VM in die Laufzeitumgebung (RTL) verschoben. + +### Schritt 1: Transport der Typinformation im AST +Der AST-Knoten für Pipes (`BoundKind::Pipe`) wird um ein Feld für den resultierenden Typ erweitert. +```rust +BoundKind::Pipe { + inputs: Vec>, + lambda: Box>, + out_type: StaticType, // Wird vom TypeChecker injiziert +} +``` +Damit steht die Typinformation direkt an der Stelle zur Verfügung, wo der Knoten evaluiert wird, ohne dass die VM Typ-Metadaten parsen muss. + +### Schritt 2: Factory in der RTL (`streams.rs`) +Die VM sammelt zur Laufzeit nur noch die evaluierten Input-Streams und erzeugt die ausführbare VM-Closure für das Lambda. Anschließend delegiert sie die Konstruktion der Pipeline an eine zentrale Factory-Funktion in `streams.rs`: + +```rust +pub fn build_pipeline_node( + inputs: Vec>, + executor: Box) -> Value>, + out_type: &StaticType +) -> Result, String> +``` + +### Schritt 3: Typ-Spezialisierung und SoA-Aufbau +Innerhalb von `build_pipeline_node` wird per Pattern-Matching auf den `out_type` die optimale Puffer-Architektur aufgebaut: + +1. **`StaticType::Record(layout)`:** + * Die Funktion durchläuft das Layout des Records. + * Für jedes Feld wird ein passender, typisierter Ringpuffer (z.B. `RingBuffer`) und ein entsprechender `SeriesPusher` erzeugt. + * Alle Puffer werden in einer `SharedRecordSeries` gebündelt. Diese verhält sich nach außen wie eine Liste von Records, speichert die Daten intern aber extrem effizient als Spalten (SoA). +2. **`StaticType::Float` / `Int` / `Bool`:** + * Es wird direkt ein `RingBuffer` mit einer `SharedSeries` und einem `SeriesPusher` instanziiert. +3. **Fallback (`StaticType::Any`):** + * Nur wenn der Typ zur Compile-Zeit nicht ermittelbar ist oder es sich um unstrukturierte Daten handelt, fällt das System auf den generischen `RingBuffer` und die `SharedValueSeries` zurück. + +### Schritt 4: Der Typsichere `SeriesPusher` +Da die Signale aus dem `PipeStream` generische `Value`-Objekte sind (da die VM-Closure `Value` zurückgibt), muss der Wert vor dem Einfügen in den flachen Puffer konvertiert (downcasted) werden. +Dies geschieht durch eine funktionale Injektion im `SeriesPusher`: +```rust +pub struct SeriesPusher { + pub buffer: Rc>>, + pub extractor: fn(Value) -> Option, +} +``` +Der Extractor (z.B. `|v| if let Value::Float(f) = v { Some(f) } else { None }`) wird bei der Instanziierung in der Factory mitgegeben. Dadurch bleibt der Pusher völlig generisch, weiß aber exakt, wie er "seinen" nativen Typ aus dem VM-Rückgabewert extrahieren muss. diff --git a/examples/object_records.myc b/examples/object_records.myc index 4c925b9..32dddf8 100644 --- a/examples/object_records.myc +++ b/examples/object_records.myc @@ -1,5 +1,5 @@ -;; Benchmark: 1.9us -;; Benchmark-Repeat: 1072 +;; Benchmark: 1.6us +;; Benchmark-Repeat: 1233 ;; Output: [150 130 "Insufficient funds" 130] ; --------------------------------------------------------- diff --git a/examples/record_specialization_test.myc b/examples/record_specialization_test.myc new file mode 100644 index 0000000..c67dc09 --- /dev/null +++ b/examples/record_specialization_test.myc @@ -0,0 +1,23 @@ +;; Benchmark: 1.5us +;; Benchmark-Repeat: 1367 +;; 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. + +(do + (def ohlc_stream (create-random-ohlc 42 10)) + + ;; Pipe die einen Record erzeugt + (def indicator + (pipe [ohlc_stream] + (fn [ohlc] + { + :mid (/ (+ (.high ohlc) (.low ohlc)) 2.0) + :range (- (.high ohlc) (.low ohlc)) + } + ) + ) + ) + + indicator +) diff --git a/src/ast/compiler/analyzer.rs b/src/ast/compiler/analyzer.rs index c91c61e..7f2e3ba 100644 --- a/src/ast/compiler/analyzer.rs +++ b/src/ast/compiler/analyzer.rs @@ -237,7 +237,7 @@ impl<'a> Analyzer<'a> { Purity::Impure, ) } - BoundKind::Pipe { inputs, lambda } => { + BoundKind::Pipe { inputs, lambda, out_type } => { let mut analyzed_inputs = Vec::with_capacity(inputs.len()); for input in inputs { analyzed_inputs.push(self.visit(Rc::new(input.clone()))); @@ -247,6 +247,7 @@ impl<'a> Analyzer<'a> { BoundKind::Pipe { inputs: analyzed_inputs, lambda: a_lambda, + out_type: out_type.clone(), }, Purity::Impure, ) diff --git a/src/ast/compiler/binder.rs b/src/ast/compiler/binder.rs index da82f4e..be0527e 100644 --- a/src/ast/compiler/binder.rs +++ b/src/ast/compiler/binder.rs @@ -287,6 +287,7 @@ impl Binder { BoundKind::Pipe { inputs: bound_inputs, lambda: bound_lambda, + out_type: crate::ast::types::StaticType::Any, }, )) } diff --git a/src/ast/compiler/bound_nodes.rs b/src/ast/compiler/bound_nodes.rs index 00070b1..73eab09 100644 --- a/src/ast/compiler/bound_nodes.rs +++ b/src/ast/compiler/bound_nodes.rs @@ -138,6 +138,7 @@ pub enum BoundKind { Pipe { inputs: Vec>, lambda: Box>, + out_type: crate::ast::types::StaticType, }, Block { exprs: Vec>, diff --git a/src/ast/compiler/captures.rs b/src/ast/compiler/captures.rs index 5ec2268..9dfd0a3 100644 --- a/src/ast/compiler/captures.rs +++ b/src/ast/compiler/captures.rs @@ -89,7 +89,7 @@ impl CapturePass { args: Box::new(Self::transform(*args, capture_map)), }; } - BoundKind::Pipe { inputs, lambda } => { + BoundKind::Pipe { inputs, lambda, out_type } => { let mut t_inputs = Vec::with_capacity(inputs.len()); for input in inputs { t_inputs.push(Self::transform(input, capture_map)); @@ -97,6 +97,7 @@ impl CapturePass { node.kind = BoundKind::Pipe { inputs: t_inputs, lambda: Box::new(Self::transform(*lambda, capture_map)), + out_type: out_type.clone(), }; } BoundKind::Block { exprs } => { diff --git a/src/ast/compiler/dumper.rs b/src/ast/compiler/dumper.rs index 1ee9d9f..196e4ce 100644 --- a/src/ast/compiler/dumper.rs +++ b/src/ast/compiler/dumper.rs @@ -203,7 +203,7 @@ impl Dumper { self.visit(args); self.indent -= 1; } - BoundKind::Pipe { inputs, lambda } => { + BoundKind::Pipe { inputs, lambda, .. } => { self.log("Pipe", node); self.indent += 1; for input in inputs { diff --git a/src/ast/compiler/optimizer/engine.rs b/src/ast/compiler/optimizer/engine.rs index 91e6be0..dd20c11 100644 --- a/src/ast/compiler/optimizer/engine.rs +++ b/src/ast/compiler/optimizer/engine.rs @@ -386,7 +386,7 @@ impl Optimizer { ) } - BoundKind::Pipe { inputs, lambda } => { + BoundKind::Pipe { inputs, lambda, out_type } => { let mut o_inputs = Vec::with_capacity(inputs.len()); for input in inputs { o_inputs.push(self.visit_node(input, sub, path)); @@ -396,6 +396,7 @@ impl Optimizer { BoundKind::Pipe { inputs: o_inputs, lambda: o_lambda, + out_type: out_type.clone(), }, node.ty.clone(), ) diff --git a/src/ast/compiler/optimizer/utils.rs b/src/ast/compiler/optimizer/utils.rs index d9b5f56..b8e2157 100644 --- a/src/ast/compiler/optimizer/utils.rs +++ b/src/ast/compiler/optimizer/utils.rs @@ -152,7 +152,7 @@ impl UsageInfo { BoundKind::Again { args } => { self.collect(args); } - BoundKind::Pipe { inputs, lambda } => { + BoundKind::Pipe { inputs, lambda, .. } => { for input in inputs { self.collect(input); } diff --git a/src/ast/compiler/tco.rs b/src/ast/compiler/tco.rs index 3ad2ca6..775654e 100644 --- a/src/ast/compiler/tco.rs +++ b/src/ast/compiler/tco.rs @@ -48,7 +48,7 @@ impl TCO { args: Box::new(Self::transform(Rc::new((**args).clone()), false)), } } - BoundKind::Pipe { inputs, lambda } => { + BoundKind::Pipe { inputs, lambda, out_type } => { let mut t_inputs = Vec::with_capacity(inputs.len()); for input in inputs { t_inputs.push(Self::transform(Rc::new(input.clone()), false)); @@ -56,6 +56,7 @@ impl TCO { BoundKind::Pipe { inputs: t_inputs, lambda: Box::new(Self::transform(Rc::new((**lambda).clone()), false)), + out_type: out_type.clone(), } } diff --git a/src/ast/compiler/type_checker.rs b/src/ast/compiler/type_checker.rs index a70ae38..ff1687b 100644 --- a/src/ast/compiler/type_checker.rs +++ b/src/ast/compiler/type_checker.rs @@ -387,12 +387,24 @@ impl TypeChecker { ) } - BoundKind::Pipe { inputs, lambda } => { + BoundKind::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 { - typed_inputs.push(self.check_node(input, ctx)?); + let typed_input = self.check_node(input, ctx)?; + // Unwrap Series(T) to T for the lambda argument + let arg_ty = if let StaticType::Series(inner) = &typed_input.ty { + *inner.clone() + } else { + StaticType::Any + }; + arg_types.push(arg_ty); + typed_inputs.push(typed_input); } - let typed_lambda = self.check_node(*lambda, ctx)?; + + // Specialized check for the lambda using the input types! + let typed_lambda = self.check(*lambda, &arg_types)?; + let ret_ty = if let StaticType::Function(sig) = &typed_lambda.ty { // If the lambda returns an Optional(T), the pipeline filters Void and stores T! if let StaticType::Optional(inner) = &sig.ret { @@ -407,6 +419,7 @@ impl TypeChecker { BoundKind::Pipe { inputs: typed_inputs, lambda: Box::new(typed_lambda), + out_type: ret_ty.clone(), }, StaticType::Series(Box::new(ret_ty)), ) diff --git a/src/ast/rtl/streams.rs b/src/ast/rtl/streams.rs index 50ba5e8..095732a 100644 --- a/src/ast/rtl/streams.rs +++ b/src/ast/rtl/streams.rs @@ -240,11 +240,14 @@ impl Observer for PipeStream { 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) { - // ... (Downcast and push logic) + 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); + } } } @@ -260,6 +263,137 @@ impl Observer for ValuePusher { } } +/// 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 { + fn notify(&mut self, _source_index: usize, _cycle_id: u64, value: Value) { + 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); + } + } + } + } +} + +/// Factory function to build a specialized pipeline node based on the output type. +/// This keeps the VM "dumb" and moves the buffer selection logic to the RTL. +pub fn build_pipeline_node( + inputs: Vec>, + executor: Box) -> Value>, + out_type: &StaticType, +) -> Rc { + use crate::ast::rtl::series::*; + + let pipe = Rc::new(RefCell::new(PipeStream::new( + "pipe".to_string(), + inputs.len(), + Some(executor), + ))); + + // Connect inputs to the pipe + for (i, input) in inputs.into_iter().enumerate() { + let adapter = Rc::new(RefCell::new(SourceAdapter { + target: pipe.clone(), + target_index: i, + })); + 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, + }) +} + // ============================================================================ // Script Integration (RTL Registration) // ============================================================================ @@ -271,11 +405,19 @@ pub fn register(env: &Environment) { // (create-random-ohlc seed limit) -> StreamNode let generators = env.pipeline_generators.clone(); + // Define the OHLC layout for typing + let ohlc_layout = RecordLayout::get_or_create(vec![ + (Keyword::intern("open"), StaticType::Float), + (Keyword::intern("high"), StaticType::Float), + (Keyword::intern("low"), StaticType::Float), + (Keyword::intern("close"), StaticType::Float), + ]); + env.register_native_fn( "create-random-ohlc", StaticType::Function(Box::new(Signature { params: StaticType::Tuple(vec![StaticType::Int, StaticType::Int]), - ret: StaticType::Any, // Returns a StreamNode + ret: StaticType::Series(Box::new(StaticType::Record(ohlc_layout.clone()))), })), Purity::Impure, // Modifies global generator registry move |args: std::vec::Vec| { diff --git a/src/ast/vm.rs b/src/ast/vm.rs index 4e366a3..a6ac6d7 100644 --- a/src/ast/vm.rs +++ b/src/ast/vm.rs @@ -368,10 +368,13 @@ impl VM { Ok(Value::Void) } } - BoundKind::Pipe { inputs, lambda } => { - use crate::ast::rtl::streams::{PipelineNode, PipeStream, StreamNode, ValuePusher, ObservableStream}; - use crate::ast::rtl::series::{RingBuffer, SharedValueSeries}; - + BoundKind::Pipe { + inputs, + lambda, + out_type, + } => { + use crate::ast::rtl::streams::{PipelineNode, StreamNode}; + let mut obs_streams = Vec::new(); for input in inputs { let val = self.eval_internal(obs, input)?; @@ -381,7 +384,10 @@ impl VM { } 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 {}", obj.type_name())); + 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()); @@ -398,46 +404,20 @@ impl VM { // Create the persistent execution closure for the PipeStream let mut pipe_vm = crate::ast::vm::VM::new(self.globals.clone()); let my_closure = lambda_obj.clone(); - let executor: Box) -> Value> = Box::new(move |args: Vec| -> Value { - match pipe_vm.run_with_args(my_closure.clone(), args) { - Ok(res) => res, - Err(e) => panic!("Pipeline lambda execution failed: {}", e), - } - }); + let executor: Box) -> Value> = + Box::new(move |args: Vec| -> Value { + match pipe_vm.run_with_args(my_closure.clone(), args) { + Ok(res) => res, + Err(e) => panic!("Pipeline lambda execution failed: {}", e), + } + }); - let pipe = Rc::new(RefCell::new(PipeStream::new( - "pipe".to_string(), - inputs.len(), - Some(executor) - ))); - - for (i, stream) in obs_streams.into_iter().enumerate() { - let adapter = Rc::new(RefCell::new(crate::ast::rtl::streams::SourceAdapter { - target: pipe.clone() as Rc>, - target_index: i, - })); - stream.add_observer(adapter); - } - - // Create the data buffer and pusher for the pipe's output - let buffer = Rc::new(RefCell::new(RingBuffer::::new())); - let pusher = Rc::new(RefCell::new(ValuePusher { - buffer: buffer.clone(), - lookback: Some(100), // Default lookback for now - })); - - // The pipe pushes to the buffer - pipe.borrow().add_observer(pusher); - - // Create the Series view for script access - let series = Rc::new(SharedValueSeries { - buffer, - }); - - let node = Rc::new(PipelineNode { - stream: pipe, - series, - }); + // Delegate to the RTL Factory for specialized buffer instantiation + let node = crate::ast::rtl::streams::build_pipeline_node( + obs_streams, + executor, + out_type, + ); Ok(Value::Object(node)) }