Change Details

This diff enables passing data object by reference between solids (use `Address` created in D4579 as the internal representation) The info we need to pass data (set and get intermediates) 1) a write/read method (this diff) 2) a path where the object will be loaded from or materialized to (D4579) 3) the object **1) write/read method** write - `object_store.set_object(path, obj, serialization_strategy)` - `dagster_type.materializer.materialize_runtime_values(context, config_value, obj)` read - `object_store.get_object(path, serialization_strategy)` - `dagster_type.loader.construct_from_config_value(context, config_value)` **so the processing priority could be** 1. use `dagster_type.materializer/loader` to handle set/get intermediates if `config_value` is either configured in solid_config (see `_create_output_materializations`) or provided through `Output(value, address=Address(config_value))` (see `_set_intermediates`) - limitation 1: materializer and loader have to be paired -- i think pairing materializer and loader makes sense as they are essentially serialization and deserialization in `serialization_strategy`, which also means there's an opportunity to merge the `serialization_strategy` and materializer/loader. - limitation 2: materializer/loader is tied to dagster type. if a user wants to utilize materializer/loader, they would need to write their own custom dagster type, which would introduce boilerplate and doesn't seem ideal. * *will address this later in the stack to allow separating dagster type and materializer/loader. My rough thought is to put custom write/read method on Output or OutputDefinition [TODO 1] 2. else, use `object_store` and `Address(path)` to handle intermediates, which will use `serialization_strategy` or default pickling as its write/read method. -- this also goes back to the previous point of potential refactor. but it doesn't seem critical so i'd like to punt to next iteration. **set/get intermediate logic in intermediate_storage would be** - `write(address, obj) -> set_object event` either * `object_store.set_object(path, obj, serialization_strategy)` [TODO 2: user can specify which object_store to use per output/input] * or `dagster_type.materializer.materialize_runtime_values(context, config_value, obj)` (get `config_value` via `solid_config.outputs` or `Output` of the current step) - `read(address) -> obj, get_object event` either * `object_store.get_object(path, serialization_strategy)` [TODO 2: user can specify which object_store to use per output/input] * or `dagster_type.loader.construct_from_config_value(context, config_value)` (get `config_value` via `solid_config.outputs` or `Output` of the source step) **2) cross-run intermediates based on events** here we created new `ObjectStoreOperation` types: * `ObjectStoreOperationType.SET_EXTERNAL_OBJECT` when an object is set externally (outside the intermediate dir) * `ObjectStoreOperationType.GET_EXTERNAL_OBJECT` when an object is got externally (outside the intermediate dir) we enabled cross-run intermediate storage using event logs (see `instance.get_external_intermediates`) to map step_output_handle->address and therefore enable pass by reference between runs (e.g. re-execution) **TODO** 1. Decouple dagster type and materializer/loader. My rough thought is to put custom write/read method on Output or OutputDefinition. 2. Enable users to configure storage per output via `object_store` - Currently users can configure a system-level or pipeline-level storage, but they are not able to get out-of-box support if they want to store intermediates to different storage in the same pipeline run. - For example: a user wants to persist the output of solid_1 to local filesystem and upload the output of solid_2 to a dev s3 bucket.

see detailed explanation on D4597This diff enables passing data object by reference between solids (use `Address` created in D4579 as the internal representation) The info we need to pass data (set and get intermediates) 1) a write/read method (this diff) 2) a path where the object will be loaded from or materialized to (D4579) 3) the object **1) write/read method** write - `object_store.set_object(path, obj, serialization_strategy)` - `dagster_type.materializer.materialize_runtime_values(context, config_value, obj)` read - `object_store.get_object(path, serialization_strategy)` - `dagster_type.loader.construct_from_config_value(context, config_value)` **so the processing priority could be** 1. use `dagster_type.materializer/loader` to handle set/get intermediates if `config_value` is either configured in solid_config (see `_create_output_materializations`) or provided through `Output(value, address=Address(config_value))` (see `_set_intermediates`) - limitation 1: materializer and loader have to be paired -- i think pairing materializer and loader makes sense as they are essentially serialization and deserialization in `serialization_strategy`, which also means there's an opportunity to merge the `serialization_strategy` and materializer/loader. - limitation 2: materializer/loader is tied to dagster type. if a user wants to utilize materializer/loader, they would need to write their own custom dagster type, which would introduce boilerplate and doesn't seem ideal. * *will address this later in the stack to allow separating dagster type and materializer/loader. My rough thought is to put custom write/read method on Output or OutputDefinition [TODO 1] 2. else, use `object_store` and `Address(path)` to handle intermediates, which will use `serialization_strategy` or default pickling as its write/read method. -- this also goes back to the previous point of potential refactor. but it doesn't seem critical so i'd like to punt to next iteration. **set/get intermediate logic in intermediate_storage would be** - `write(address, obj) -> set_object event` either * `object_store.set_object(path, obj, serialization_strategy)` [TODO 2: user can specify which object_store to use per output/input] * or `dagster_type.materializer.materialize_runtime_values(context, config_value, obj)` (get `config_value` via `solid_config.outputs` or `Output` of the current step) - `read(address) -> obj, get_object event` either * `object_store.get_object(path, serialization_strategy)` [TODO 2: user can specify which object_store to use per output/input] * or `dagster_type.loader.construct_from_config_value(context, config_value)` (get `config_value` via `solid_config.outputs` or `Output` of the source step) **2) cross-run intermediates based on events** here we created new `ObjectStoreOperation` types: * `ObjectStoreOperationType.SET_EXTERNAL_OBJECT` when an object is set externally (outside the intermediate dir) * `ObjectStoreOperationType.GET_EXTERNAL_OBJECT` when an object is got externally (outside the intermediate dir) we enabled cross-run intermediate storage using event logs (see `instance.get_external_intermediates`) to map step_output_handle->address and therefore enable pass by reference between runs (e.g. re-execution) **TODO** 1. Decouple dagster type and materializer/loader. My rough thought is to put custom write/read method on Output or OutputDefinition. 2. Enable users to configure storage per output via `object_store` - Currently users can configure a system-level or pipeline-level storage, but they are not able to get out-of-box support if they want to store intermediates to different storage in the same pipeline run. - For example: a user wants to persist the output of solid_1 to local filesystem and upload the output of solid_2 to a dev s3 bucket.