@@ -70,10 +70,27 @@ func (e *ForeachExecutor) Execute(ctx context.Context, step *core.Step, execCtx
7070 threads = 1
7171 }
7272
73+ // Check for streaming output configuration via exports
74+ var streamingOutput string
75+ if step .Exports != nil {
76+ if so , ok := step .Exports ["streaming_output" ]; ok {
77+ // Render the template
78+ rendered , err := e .templateEngine .Render (so , execCtx .GetVariables ())
79+ if err == nil {
80+ streamingOutput = rendered
81+ }
82+ }
83+ }
84+
7385 // Execute with streaming worker pool
74- outputs , err := e .executeWithWorkerPool (ctx , step , step .Input , threads , execCtx )
86+ outputs , err := e .executeWithWorkerPoolStreaming (ctx , step , step .Input , threads , execCtx , streamingOutput )
7587
76- result .Output = strings .Join (outputs , "\n " )
88+ if streamingOutput != "" {
89+ // In streaming mode, output is written to file
90+ result .Output = fmt .Sprintf ("Results streamed to: %s" , streamingOutput )
91+ } else {
92+ result .Output = strings .Join (outputs , "\n " )
93+ }
7794 result .EndTime = time .Now ()
7895 result .Duration = result .EndTime .Sub (result .StartTime )
7996
@@ -211,6 +228,34 @@ type workResult struct {
211228 err error
212229}
213230
231+ // streamWriter handles concurrent writes to an output file
232+ type streamWriter struct {
233+ file * os.File
234+ mu sync.Mutex
235+ }
236+
237+ func newStreamWriter (path string ) (* streamWriter , error ) {
238+ f , err := os .Create (path )
239+ if err != nil {
240+ return nil , err
241+ }
242+ return & streamWriter {file : f }, nil
243+ }
244+
245+ func (w * streamWriter ) Write (line string ) error {
246+ w .mu .Lock ()
247+ defer w .mu .Unlock ()
248+ _ , err := w .file .WriteString (line + "\n " )
249+ return err
250+ }
251+
252+ func (w * streamWriter ) Close () error {
253+ if w .file != nil {
254+ return w .file .Close ()
255+ }
256+ return nil
257+ }
258+
214259// executeWithWorkerPool executes the inner step using a streaming worker pool pattern
215260// This is memory-efficient: creates only 'threads' goroutines instead of N goroutines
216261// and streams input lines on-demand instead of loading all into memory
@@ -333,6 +378,134 @@ func (e *ForeachExecutor) executeWithWorkerPool(ctx context.Context, step *core.
333378 return outputs , firstError
334379}
335380
381+ // executeWithWorkerPoolStreaming is like executeWithWorkerPool but supports streaming output to file.
382+ // When streamingOutput is set, results are written directly to the file instead of being collected in memory.
383+ // This enables O(1) memory usage for million-line inputs.
384+ func (e * ForeachExecutor ) executeWithWorkerPoolStreaming (ctx context.Context , step * core.Step , inputPath string , threads int , execCtx * core.ExecutionContext , streamingOutput string ) ([]string , error ) {
385+ // If no streaming, delegate to original implementation
386+ if streamingOutput == "" {
387+ return e .executeWithWorkerPool (ctx , step , inputPath , threads , execCtx )
388+ }
389+
390+ log := logger .Get ()
391+ log .Debug ("Foreach streaming mode enabled" ,
392+ zap .String ("input" , inputPath ),
393+ zap .String ("output" , streamingOutput ),
394+ zap .Int ("threads" , threads ),
395+ )
396+
397+ // Create streaming writer
398+ writer , err := newStreamWriter (streamingOutput )
399+ if err != nil {
400+ return nil , fmt .Errorf ("failed to create streaming output file: %w" , err )
401+ }
402+ defer func () { _ = writer .Close () }()
403+
404+ // Create bounded work queue
405+ workQueue := make (chan workItem , threads * 2 )
406+ done := make (chan struct {})
407+
408+ // Track completion
409+ var workerWg sync.WaitGroup
410+ var producerErr error
411+ var writeErr error
412+ var writeErrMu sync.Mutex
413+
414+ // Start fixed worker pool
415+ for i := 0 ; i < threads ; i ++ {
416+ workerWg .Add (1 )
417+ go func () {
418+ defer workerWg .Done ()
419+ for work := range workQueue {
420+ // Check context cancellation
421+ if ctx .Err () != nil {
422+ continue
423+ }
424+
425+ // Apply variable pre-processing if configured
426+ loopValue := work .value
427+ if step .VariablePreProcess != "" {
428+ processedValue , err := e .preProcessVariable (step .VariablePreProcess , step .Variable , work .value , execCtx )
429+ if err != nil {
430+ logger .Get ().Warn ("variable pre-process failed, using original value" ,
431+ zap .String ("expression" , step .VariablePreProcess ),
432+ zap .String ("original_value" , work .value ),
433+ zap .Error (err ))
434+ } else {
435+ loopValue = processedValue
436+ }
437+ }
438+
439+ // Create optimized child context with loop variables pre-set
440+ childCtx := execCtx .CloneForLoop (step .Variable , loopValue , work .index + 1 )
441+
442+ // Clone inner step and render secondary templates [[ ]]
443+ innerStep := e .renderSecondaryTemplates (step .Step , childCtx )
444+
445+ // Execute inner step
446+ stepResult , _ := e .dispatcher .Dispatch (ctx , innerStep , childCtx )
447+
448+ // Stream output directly to file (no memory collection)
449+ if stepResult != nil && stepResult .Output != "" {
450+ if err := writer .Write (stepResult .Output ); err != nil {
451+ writeErrMu .Lock ()
452+ if writeErr == nil {
453+ writeErr = err
454+ }
455+ writeErrMu .Unlock ()
456+ }
457+ }
458+ }
459+ }()
460+ }
461+
462+ // Producer: stream lines into work queue
463+ go func () {
464+ defer close (workQueue )
465+
466+ iter , err := NewLineIterator (inputPath )
467+ if err != nil {
468+ producerErr = err
469+ return
470+ }
471+ defer func () { _ = iter .Close () }()
472+
473+ idx := 0
474+ for iter .Next () {
475+ select {
476+ case workQueue <- workItem {index : idx , value : iter .Value ()}:
477+ idx ++
478+ case <- ctx .Done ():
479+ producerErr = ctx .Err ()
480+ return
481+ }
482+ }
483+
484+ if iter .Err () != nil {
485+ producerErr = iter .Err ()
486+ }
487+ }()
488+
489+ // Wait for all workers to complete
490+ go func () {
491+ workerWg .Wait ()
492+ close (done )
493+ }()
494+
495+ <- done
496+
497+ // Check for errors
498+ if producerErr != nil {
499+ return nil , producerErr
500+ }
501+ if writeErr != nil {
502+ return nil , fmt .Errorf ("streaming write error: %w" , writeErr )
503+ }
504+
505+ // Return empty slice since results were streamed
506+ return nil , nil
507+ }
508+
336509// CanHandle returns true if this executor can handle the given step type
337510func (e * ForeachExecutor ) CanHandle (stepType core.StepType ) bool {
338511 return stepType == core .StepTypeForeach
0 commit comments