dagster-io / dagster

An orchestration platform for the development, production, and observation of data assets.
https://dagster.io
Apache License 2.0
12k stars 1.5k forks source link

Run configuration for non-partitioned asset becomes partitioned once a partitioned asset is created #19389

Open stevenmurphy12 opened 10 months ago

stevenmurphy12 commented 10 months ago

Dagster version

1.6.1 (also observed in 1.5.13)

What's the issue?

When I create a non-partitioned asset and open the run config it is not partitioned (as expected)

When I subsequently create a partitioned asset, and open the run config for the original non-partitioned asset, the run config is now partitioned.

What did you expect to happen?

I expect that the run configuration for the non-partitioned asset remains non-partitioned. And the run config for the partitioned asset to be partitioned. I don't expect all the assets to be partitioned.

How to reproduce?

Create a definitions.py file with the code below, run dagster dev -f definitions.py:

from dagster import asset, Definitions

@asset
def my_asset():
    pass

defs = Definitions(
    assets=[my_asset]
)

In the UI, open the launchpad and observe the asset is not partitioned.

image

Now, add a partitioned asset:

from dagster import asset, Definitions, DailyPartitionsDefinition

@asset
def my_asset():
    pass

@asset(partitions_def=DailyPartitionsDefinition(start_date="2024-01-17"))
def my_partitioned_asset():
    pass

defs = Definitions(
    assets=[my_asset, my_partitioned_asset]
)

Observe in the UI the my_asset is now partitioned even though it hasn't changed.

image

If I attempt to materialise without specifying a partition, I get a warning

image

Now if I add another partitioned asset (using a different PartitionDefinition), the partition drop-down is correct for the run configuration on both of the partitioned asset, but the non-partitioned asset now uses the new PartitionDefinition

from dagster import asset, Definitions, DailyPartitionsDefinition, StaticPartitionsDefinition

@asset
def my_asset():
    pass

@asset(partitions_def=DailyPartitionsDefinition(start_date="2024-01-17"))
def my_partitioned_asset():
    pass

oceans_partitions_def = StaticPartitionsDefinition(
    ["arctic", "atlantic", "indian", "pacific", "southern"]
)

@asset(partitions_def=oceans_partitions_def)
def my_ocean_asset():
    pass

defs = Definitions(
    assets=[my_asset, my_partitioned_asset, my_ocean_asset]
)

image

Deployment type

Dagster Helm chart

Deployment details

Kubernetes deployment, though this is reproducible locally

Additional information

No response

Message from the maintainers

Impacted by this issue? Give it a 👍! We factor engagement into prioritization.

stevenmurphy12 commented 9 months ago

Same issue exists on Dagster 1.6.6