Pipes and test scripts

Src/AST/Myc.Ast.Evaluator.pas

@@ -83,6 +83,8 @@ uses
     Myc.Ast,
     Myc.Data.Decimal,
     Myc.Data.Series,
+    Myc.Data.Stream,
+    Myc.Data.Stream.Pipes,
     Myc.Data.Scalar.JSON,
     Myc.Ast.Types;
 
@@ -633,20 +635,6 @@ begin
     Result := TDataValue.Void;
 end;
 
-// --- Pipe Visitors Stubs ---
-
-function TEvaluatorVisitor.VisitPipe(const Node: IPipeNode): TDataValue;
-begin
-    // Placeholder for Phase 4 (Runtime Integration)
-    // Here we will:
-    // 1. Resolve Input Streams from Scope
-    // 2. Build TPipeConfig
-    // 3. Compile Lambda to TDataValue.TFunc
-    // 4. Create Adapter
-    // 5. Create TPipeStream
-    raise EEvaluatorException.Create('Pipe execution is not yet implemented.');
-end;
-
 function TEvaluatorVisitor.VisitPipeInput(const Node: IPipeInputNode): TDataValue;
 begin
     Result := TDataValue.Void;
@@ -662,4 +650,114 @@ begin
     Result := TDataValue.Void;
 end;
 
+// --- Pipe Implementation ---
+
+function TEvaluatorVisitor.VisitPipe(const Node: IPipeNode): TDataValue;
+var
+    i: Integer;
+    inputNode: IPipeInputNode;
+    sources: TArray<IStream>;
+    config: TPipeConfig;
+    inputVal: TDataValue;
+    sourceSeries: IScalarRecordSeries;
+    outputDef: IScalarRecordDefinition;
+    lambdaFunc: TDataValue.TFunc;
+    pipeAdapter: TPipeStream.TPipeLambda;
+begin
+    SetLength(sources, Node.Inputs.Count);
+    SetLength(config, Node.Inputs.Count);
+
+    // 1. Resolve Inputs & Build Config
+    for i := 0 to Node.Inputs.Count - 1 do
+    begin
+        inputNode := Node.Inputs[i];
+        inputVal := FScope[inputNode.StreamSource.Address];
+
+        // STRICT CHECK: Only vkStream is allowed.
+        if inputVal.Kind = vkStream then
+        begin
+            sources[i] := inputVal.AsStream;
+        end
+        else
+        begin
+            raise EEvaluatorException
+                .CreateFmt('Variable "%s" must be a Stream. Found type "%s".', [inputNode.StreamSource.Name, inputVal.Kind.ToString]);
+        end;
+
+        sourceSeries := sources[i].Series;
+        var srcDef := sourceSeries.Def;
+        var selectors := inputNode.Selectors;
+
+        SetLength(config[i], selectors.Count);
+        for var k := 0 to selectors.Count - 1 do
+        begin
+            var key := selectors[k].Value;
+            var idx := srcDef.IndexOf(key);
+            // Runtime safety check (TypeChecker usually handles this)
+            if idx < 0 then
+                raise EEvaluatorException.CreateFmt('Field ":%s" missing in stream "%s".', [key.Name, inputNode.StreamSource.Name]);
+
+            var fieldKind := srcDef.Items[idx].Value;
+            config[i][k] := TScalarRecordField.Create(key, fieldKind);
+        end;
+    end;
+
+    // 2. Get Output Definition
+    if (Node.StaticType = nil) or (Node.StaticType.Kind <> stRecordSeries) then
+        raise EEvaluatorException.Create(
+            'Internal Error: Pipe requires a RecordSeries static type definition. Lambda must return a Record.');
+
+    outputDef := Node.StaticType.Definition;
+
+    // 3. Compile Lambda
+    lambdaFunc := Node.Transformation.Accept(Self).AsMethod();
+
+    // 4. Create Adapter
+    pipeAdapter :=
+        function(const SourceSeries: array of ISeries; out Results: array of TScalar.TValue): Boolean
+        var
+            scriptArgs: TArray<TDataValue>;
+            scriptResult: TDataValue;
+            k: Integer;
+            resRec: IScalarRecord;
+        begin
+            SetLength(scriptArgs, Length(SourceSeries));
+            // Pass Scalars (Current Items)
+            for k := 0 to High(SourceSeries) do
+                scriptArgs[k] := TDataValue(SourceSeries[k].Items[0]);
+
+            scriptResult := lambdaFunc(scriptArgs);
+
+            // Void -> Filtered out (no emit)
+            if scriptResult.IsVoid then
+                exit(False);
+
+            // STRICT CHECK: Only ScalarRecord allowed
+            if scriptResult.Kind = vkScalarRecord then
+            begin
+                resRec := scriptResult.AsScalarRecord;
+
+                // Integrity check: Runtime Record must match Static Definition
+                if resRec.Def.Count <> Length(Results) then
+                    raise EEvaluatorException.Create('Pipe lambda returned record with incorrect field count/layout.');
+
+                for k := 0 to resRec.Def.Count - 1 do
+                    Results[k] := resRec.Items[k].Value.Value;
+            end
+            else
+            begin
+                // Scalars are forbidden
+                raise EEvaluatorException
+                    .CreateFmt('Pipe lambda returned invalid type "%s". Must return a Record or Void.', [scriptResult.Kind.ToString]);
+            end;
+
+            Result := True;
+        end;
+
+    var pipe := TPipeStream.Create(config, outputDef, sources, pipeAdapter);
+
+    // Result is a Stream
+    Result := TDataValue.FromStream(pipe);
+end;
+
 end.