"that has not yet resolved"? or "that has not yet been resolved"?

do we need this if statement? does pipeline_context.for_step(step) break for dynamic steps?

could be nice to convey that only successfully yielded outputs are included. maybe "self._resolved_dynamic_outputs"? "self._successful_dynamic_outputs"?

enumerate(self._pending_resolve) since self._pending_resolve is already a list?

nice

should we check that exactly one of the step_inputs is type UnresolvedStepInput

should we add unit tests for DynamicStepHandle.to_key, DynamicStepHandle.unresolved_form, UnresolvedStepHandle.to_key, UnresolvedStepHandle.resolve, like test_step_handles for StepHandle.from_key

I might add a line clarifying when the UnresolvedStepHandle resolves into DynamicStepHandle (ie after upstream dynamic step yields all dynamic output and step succeeds, then all downstream unresolved step handles resolve into dynamic step handles)

nit: step_output_handle=, input_def=, config_data=

nit: map_key for consistency

nit: step_output_handle=

this intersection is responsible for respecting the ExecutionPlan's step_handles_to_execute?

add a test for this section?

does _create_step_outputs handle this?

what is the case when only one of output.mapping_key and step.get_mapping_key() is set?

nice

the list is to make a copy since we are mutating it while iterating over it, will leave a comment

^

you can have more than one once in the case that you have a downstream dag from a dynamic output - from the regular outputs of those not-yet-resolved steps

the assertion at [1] I think gets at what you were thinking

this is exercised by the multiprocessing test - in each step execution subprocess, let me know if you had something more specific in mind

[1]

good catch

• output.mapping_key is set: this is a normal execution step with dynamic outputs
• step.get_mapping_key() is set: this is a step downstream of a dynamic output

see [2] in https://dagster.phacility.com/D5612 for the case where they would both be set, which as that diff stands we *allow* but you would not be able to use the python api results object to interact with due to this invariant. I think the fix would be probably be double dict of step resolved mapping key -> output mapping key -> results

after exploring this one tough bit is external places like the context that hold a reference to the ExecutionPlan object, so likely any immutable record stack would have to happen internal to an object that maintains stable reference identity. Think this one might have to move to a follow up

from some small research it looks like python caches them already

copy thing worth a comment

doing the top-level include here seems bad. I would dot into dagster.core.types or whatever it is

would rather see if isinstance(step, AllUnresolved)

followed by an invariant check on ExecutionStep

to make sure we are defensive against adding additional subtypes

similar thing here. helper function that does more checks would be good