Repository navigation
Feat/foreach step - #6473
Open
nayan3107 wants to merge 5 commits into
Open
Feat/foreach step#6473nayan3107 wants to merge 5 commits into
nayan3107 wants to merge 5 commits into
Conversation
Adds a new ForEachStep to sagemaker.mlops.workflow that iterates a body of steps over a collection at pipeline execution time. Supports CurrentItem / ChildOutput references, a Selector for picking values from prior child outputs, and a configurable max_concurrency in the range [1, 50]. - New for_each_step.py exposing ForEachStep, ForEachSelector, ForEachSelectorPickType, CurrentItem, and ChildOutput. - Add FOR_EACH to StepTypeEnum. - Extend StepsCompiler to compile ForEachStep bodies and share duplicate-name detection with ConditionStep by generalizing _flatten_condition_step into _flatten_compound_step. - Track for_each_body dependencies in PipelineGraph and enforce unique step names across ForEachStep bodies. - Re-export the new symbols from sagemaker.mlops.workflow. - Unit tests covering the new step type, compiler paths, and pipeline wire format.
Remove Status, InputItemCount, TotalStepCount, SucceededStepCount, FailedStepCount, and StoppedStepCount from ForEachStep.properties. Only ChildOutputs[i] and Selected remain as downstream-referenceable properties. This aligns with the backend policy change: any child failure now causes the entire ForEach step to fail (no partial success). The PartiallySucceeded status is removed, making aggregate counters no longer meaningful to downstream steps.
Remove _TerminalPropertyReference class and CurrentItem overrides.
CurrentItem now inherits full drilling from _ForEachPropertyReference:
- Deep dot paths: CurrentItem().a.b.c
- Bracket indexing: CurrentItem()['score'], CurrentItem()[0]
All forms produce valid {"Get": "ForEach.CurrentItem..."} expressions.
The service resolves these at expansion time for each child lane.
Remove the "exactly one body step" restriction so ForEach bodies can contain any number of steps. Each iteration spawns one child instance of every body step. Rename ChildOutput / ChildOutputs to IterationOutput / IterationOutputs across the public API. Property reference changes: - properties.IterationOutputs[i].<bodyStepName>.<member> now requires a body step name qualifier, since an iteration may span multiple body steps. - properties.Selected.<member> resolves directly to the selector- picked iteration's output — no body step name qualifier, because the selector's field already identifies the target body step.
The service still resolves the ForEach aggregate members (Status, InputItemCount and the step counts) as downstream references, and accepts a literal IterableItems list of up to 100 scalars or property references. Expose both and add ForEach to the step-type guard test. --- X-AI-Prompt: Fix the ForEachStep gaps against the Tioga backend X-AI-Tool: kiro-cli
nayan3107
requested a deployment
to
manual-approval
October 14, 2026 17:56 — with 
GitHub Actions
Waiting
GitHub Actions
Waiting
nayan3107
requested a deployment
to
manual-approval
October 14, 2026 17:56 — with
GitHub Actions
Waiting
GitHub Actions
Waiting
nayan3107
requested a deployment
to
manual-approval
October 14, 2026 17:56 — with
GitHub Actions
Waiting
GitHub Actions
Waiting
This branch is waiting to be deployed
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What: Add ForEachStep to sagemaker.mlops.workflow. A ForEach step runs its body step(s) once per item of a list resolved at execution time, with a concurrency cap of 1–50.
iterable_items: a property reference that resolves to a JSON list, or a literal list of up to 100 scalars/property references.
for_each_body: one or more steps (Condition, Fail and nested ForEach are not allowed). Body steps read the current item via CurrentItem(), including deep paths and bracket indexing (CurrentItem().a.b, CurrentItem()["score"]).
ForEachSelector(pick="MaxBy"|"MinBy", field=IterationOutput(body_step).) picks one iteration.
Downstream references: properties.IterationOutputs[i].., properties.Selected., and the aggregates Status, InputItemCount, TotalStepCount, SucceededStepCount, FailedStepCount, StoppedStepCount.
StepsCompiler and PipelineGraph compile ForEach bodies and enforce unique step names across bodies.
Why: SageMaker Pipelines adds a ForEach step type; this exposes it in the PySDK.
Test: Unit tests for the new step, compiler paths and pipeline wire format, including full Pipeline.definition() checks; all 413 tests in sagemaker-mlops/tests/unit/workflow pass.
Original implementation by Daniel Castro; rebased onto master and aligned with the current service contract.