Add asset checks, data version to MaterializeResult #16514
Conversation
|
Current dependencies on/for this PR: This comment was auto-generated by Graphite. |
smackesey
force-pushed
the
sean/asset-checks-materialize-result
branch
from
666971f to
b0166c4
Compare
smackesey
changed the title
smackesey
force-pushed
the
sean/asset-checks-materialize-result
branch
from
b0166c4 to
e9829f5
Compare
|
Deploy preview for dagit-core-storybook ready! ✅ Preview Built with commit 1bfe0bf. |
smackesey
force-pushed
the
sean/asset-checks-materialize-result
branch
3 times, most recently
from
90f7b6f to
bf462da
Compare
| @@ -16,6 +19,8 @@ class MaterializeResult( | |||
| [ | |||
| ("asset_key", PublicAttr[Optional[AssetKey]]), | |||
| ("metadata", PublicAttr[Optional[MetadataUserInput]]), | |||
| ("data_version", PublicAttr[Optional[DataVersion]]), | |||
There was a problem hiding this comment.
let's list data_version after check_results. I think we should consider that feature "more important"
| def _zip_and_iterate_op_result( | ||
| result: Any, context: OpExecutionContext, output_defs: Sequence[OutputDefinition] | ||
| ) -> Iterator[Tuple[int, Any, OutputDefinition]]: | ||
| if len(output_defs) > 1: | ||
| result = _validate_multi_return(context, result, output_defs) | ||
| for position, (output_def, element) in enumerate(zip(output_defs, result)): | ||
| expected_return_outputs = _filter_expected_output_defs(result, context, output_defs) | ||
| if len(expected_return_outputs) > 1: | ||
| result = _validate_multi_return(context, result, expected_return_outputs) | ||
| for position, (output_def, element) in enumerate(zip(expected_return_outputs, result)): | ||
| yield position, output_def, element | ||
| else: | ||
| yield 0, output_defs[0], result | ||
|
|
||
|
|
||
| # Filter out output_defs corresponding to asset check results that already exist on a | ||
| # MaterializeResult. | ||
| def _filter_expected_output_defs( | ||
| result: Any, context: OpExecutionContext, output_defs: Sequence[OutputDefinition] | ||
| ) -> Sequence[OutputDefinition]: | ||
| result_tuple = ( | ||
| (result,) if not isinstance(result, tuple) or is_named_tuple_instance(result) else result | ||
| ) | ||
| materialize_results = [x for x in result_tuple if isinstance(x, MaterializeResult)] | ||
| remove_outputs = [ | ||
| r.get_spec_python_identifier(x.asset_key or context.asset_key) | ||
| for x in materialize_results | ||
| for r in x.check_results or [] | ||
| ] | ||
| return [out for out in output_defs if out.name not in remove_outputs] | ||
|
|
||
|
|
There was a problem hiding this comment.
I just thought on alternative approach here. What if we unpacked MaterializeResult right in _zip_and_iterate_op_result and yielded the appropriate AssetCheckResult and Output objects right there.
There was a problem hiding this comment.
Maybe doable but not trivial because the logic that currently converts MaterializeResult to Output in execute_step.py accesses job_def.asset_layer, which is not available for direct invocation, and _zip_and_iterate_op_result is hit in direct invocation.
Recommend exploring this in the proposed followup refactor rather than this PR.
There was a problem hiding this comment.
Ok. Please leave comment in _zip_and_iterate_op_result that contains all the context in your head about this problematic code path (and is appropriately apologetic in tone :-))
This is a fairly risk change and I want any future people who have to deal with this before you fix it to understand all that context.
smackesey
force-pushed
the
sean/asset-checks-materialize-result
branch
from
bf462da to
5556a49
Compare
smackesey
force-pushed
the
sean/asset-checks-materialize-result
branch
from
5556a49 to
1bfe0bf
Compare
|
Deploy preview for dagster-docs ready! Preview available at https://dagster-docs-cso2j6eoq-elementl.vercel.app Direct link to changed pages: |
|
Deploy preview for dagit-storybook ready! ✅ Preview Built with commit 1bfe0bf. |
## Summary & Motivation Add `check_results` and `data_version` to `MaterializeResult`. This supports streaming asset check results back from an ext process added upstack in #16466. ## How I Tested These Changes New unit test.
## Summary & Motivation Add `check_results` and `data_version` to `MaterializeResult`. This supports streaming asset check results back from an ext process added upstack in dagster-io#16466. ## How I Tested These Changes New unit test.
Summary & Motivation
Add
check_resultsanddata_versiontoMaterializeResult. This supports streaming asset check results back from an ext process added upstack in #16466.How I Tested These Changes
New unit test.