The cloud-native orchestrator for the whole development lifecycle, with integrated lineage and observability.

dagster

Hey!

I have an issue that partition config from dynamic ops doesn’t get passed to another op with built-in snowflake_io_manager and it deletes the whole table on every execution.
What’s the best way to approach this? Example in comment

```@op
def op1():
    list_of_df = [df1,df2,df3]

    for df in list_of_df:
        yield DynamicOutput(
            df, mapping_key=ticker_name
        )

@op
def op2(context, df):
    df = ...  # do some processing
    return df


@op(
    name="test",
    out={
        "test": Out(
            io_manager_key="snowflake_io_manager",
            metadata={"partition_expr": "_updated_at"},
        )
    },
)
def op3(context, df):
    return df


@job(name="dynamic", metadata={"partition_expr": "_updated_at"}, config=weekly_partitioned_config)
def dynamic_graph_job():
    result = op1().map(op2)

    op3(result.collect())  ## ??


@weekly_partitioned_config(start_date=datetime(2022, 8, 21))
def weekly_partitioned_config(start: datetime, _end: datetime):
    return {
        "ops": {
            "op1": {
                "config": {"date": start.strftime("%Y-%m-%d")}
            },
            "op2": {
                "config": {"date": start.strftime("%Y-%m-%d")}
            }
        }
    }```

Hi Levan, `snowflake_io_manager` currently only handles partitions when used with assets

Then, is it possible to pass `result.collect()` to asset?

Also, on this line:

<https://github.com/dagster-io/dagster/blob/e1adadd33cecf01474ec1c232b5b52b68ab1210a/python_modules/libraries/dagster-snowflake/dagster_snowflake/db_io_manager.py#L164|https://github.com/dagster-io/dagster/blob/e1adadd33cecf01474ec1c232b5b52b68ab1210[…]/libraries/dagster-snowflake/dagster_snowflake/db_io_manager.py>

`context` still has `_partition_key` available and If `context.has_asset_key` is `False`,  why not assign `time_window` based on `partition_key` (if provided)? Which could solve this issue

I don't think there is a way to mix dynamic outputs with assets, but I will open an issue with your suggestion for making `snowflake_io_manager` use the partition key for ops

<@U018K0G2Y85> issue Support partitions for ops in `snowflake_io_manager`

Created issue at: <https://github.com/dagster-io/dagster/issues/9715>