Concurrent runs of same job write to wrong target tables despite different parameters

[mperezcofrai]

What's the issue?

We're using Dagster for a specific use case where we have jobs that execute database operations with different parameters. For example, we use these jobs to correct data in specific database tables.

Our Use Case: We need the ability to run the same job concurrently with different parameters. Until recently, this worked as expected.

The Issue: We're experiencing what appears to be a bug with concurrent job executions. When we:

• Start Job1 with param_set_1 (e.g., targeting table_A)
• Simultaneously start the same Job1 with param_set_2 (e.g., targeting table_B)
  The second run (with param_set_2) seems to interfere with the first run. Specifically, Run 2's data is being incorrectly ingested into Run 1's target table.

Has anyone encountered similar issues with concurrent Dagster job executions using different parameters? Are there known configuration requirements for ensuring proper run isolation?

What did you expect to happen?

Expected Behavior: Each concurrent run should maintain isolation and write to its intended target based on its specific parameters.
Actual Behavior: The runs appear to "collapse" or interfere with each other, with Run 2's data ending up in Run 1's destination.

How to reproduce?

• Create a parameterized job that performs database operations where the target table is determined by a configuration parameter (e.g., a job that writes data to a table specified in the run config)
• Launch the first run with parameters targeting one destination (e.g., target_table: "table_A", data: "dataset_1")
• Immediately launch a second concurrent run of the same job with different parameters targeting a different destination (e.g., target_table: "table_B", data: "dataset_2")
• Check the database tables after both runs complete

Dagster version

1.11.16

Deployment type

Dagster Cloud

Deployment details

No response

Additional information

No response

Message from the maintainers

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

[zyd14]

Is the state you're seeing loaded into the wrong table being written manually within your asset, or are you using an IO manager? If you're using an IO manager, which one are you using? Is one run completing before the other? It's a bit unclear how the data written by Run2 could be used in Run1 if Run2 didn't write data before Run1 loaded data.

Could you provide a reproducible example? Right now the description of how the data is being generated, and where each run is loading its input data from is still pretty vague.

[mperezcofrai]

Is the state you're seeing loaded into the wrong table being written manually within your asset, or are you using an IO manager? If you're using an IO manager, which one are you using? Is one run completing before the other? It's a bit unclear how the data written by Run2 could be used in Run1 if Run2 didn't write data before Run1 loaded data.

Could you provide a reproducible example? Right now the description of how the data is being generated, and where each run is loading its input data from is still pretty vague.

---

Thanks for looking into this @zyd14!

To answer your question: I'm using manual writes within assets, not IO managers.

Data Write Pattern

Manual writes within assets (not IO managers):

1. First asset loads config and constructs target table IDs from run parameters (client_key and clients_table)
2. Middle assets process data as pandas DataFrames passed between assets
3. Final asset writes DataFrames to BigQuery using df.to_gbq() with table IDs from step 1

Resource Configuration

Resources are defined as singletons at the Definitions level:

# cofrai_onboarding/__init__.py
defs = Definitions(
    assets=[...],
    jobs=[...],
    resources={
        "bq_resource": CustomBigQueryResource(
            project="onboardings-427410",
        ),
        # ... other resources
    }
)

The CustomBigQueryResource is a simple wrapper:

class CustomBigQueryResource(ConfigurableResource):
    project: str

    def get_bigquery(self):
        return bigquery

    def get_client(self):
        return bigquery.Client(project=self.project)

Simplified Reproduction Pattern

# Asset 1: Load config and build table IDs
@asset
def standardize_clients_config_loaded(
    context,
    config: StandardizeClientsConfig,
    bq_resource: CustomBigQueryResource
) -> Dict:
    client_key = config.client_key
    input_table = f"{client_key}.{config.clients_table}"
    output_table = f"{client_key}.{config.clients_table}_standardized"
    
    # Validation logic...
    
    return {
        "client_key": client_key,
        "input_clients_table": input_table,
        "output_clients_table": output_table,
        # ... more config
    }

# Asset 2-N: Data processing (DataFrames passed through)
@asset
def bq_clients_table_loaded(
    context,
    standardize_clients_config_loaded: dict,
    bq_resource: CustomBigQueryResource,
) -> pd.DataFrame:
    return create_df_from_bq_table(...)

# ... more processing assets ...

# Final Asset: Write to BigQuery
@asset
def standardized_clients_stored_in_bq(
    context,
    standardize_clients_config_loaded: dict,
    bq_resource: CustomBigQueryResource,
    clients_final_processed: pd.DataFrame,
    # ... other dependencies
) -> None:
    table_id = standardize_clients_config_loaded["output_clients_table"]
    
    # Uses df.to_gbq() internally
    create_bq_table_from_df(context, clients_final_processed, table_id, bq_resource)

Where create_bq_table_from_df() does:

def create_bq_table_from_df(context, df, bq_table_id, bq_resource, description=None):
    df.to_gbq(
        destination_table=bq_table_id,  # e.g., "client_a.table1_standardized"
        project_id=bq_resource.get_client().project,
        if_exists="replace",
        table_schema=table_schema,
    )

The Bug Behavior

When launching two concurrent runs:

Run 1 (launched first):

config:
  client_key: "client_a"
  clients_table: "source_table_1"
Expected output: client_a.source_table_1_standardized

Run 2 (launched immediately after):

config:
  client_key: "client_b"
  clients_table: "source_table_2"
Expected output: client_b.source_table_2_standardized

Actual behavior: Run 2's processed data ends up written to client_a.source_table_1_standardized instead of its own target table.

Key Observations

1. Both runs use the same job definition (standardize_clients_job)
2. Both runs use the same singleton bq_resource instance
3. The config dict from the first asset is passed through asset dependencies
4. Each run should have isolated asset outputs, but something is being shared

Questions

1. Are asset outputs (like the config dict) supposed to be isolated per run ID, or could there be state sharing between concurrent runs?
2. Could the singleton bq_resource be caching or sharing state that affects concurrent writes?
3. Is there a recommended pattern for parameterized jobs that write to different BigQuery tables based on run config in a Dagster Cloud Serverless environment?
4. Could this be related to how asset materializations are tracked/cached in concurrent scenarios?

[zyd14]

I think the way Dagster might be involved in this issue is related to how you're passing data between Dagster's assets. An important thing to highlight about assets is that by default assets expect some kind of return value, and that return value will always be passed to an IO manager - if you don't provide an IO manager, the default IO manager is used. To opt out, you need to declare dependencies in the asset definition with deps=[<upstream_asset_keys_or_defs>]:

@asset(
  name="upstream",
  ...
):
...

@asset(
  name="example",
  deps=["upstream"]
):
...

It's also important to understand that unless you're using the in-process executor, every asset runs in its own subprocess, which means any data that is passed "directly" from one asset to another is actually being written to disk and then loaded into memory by the downstream asset. deps= allows you to opt-out of this behavior, but then you're responsible for manually loading any necessary data inputs / dependencies inside the asset definition.

The default IO manager that gets used is the FilesystemIOManager, which pickles the outputs and saves them to disk as a file named using the asset's key. This IO manager does not provide any run-to-run isolation of outputs, it simply overwrites whatever is in the output location when the asset is materialized.

So this relates to your pipeline in that you have a race condition where if two jobs are running concurrently, the second step of the job will receive the output of whichever instance of the first step most recently completed. More concretely if job1step1 starts at around the same time as job2step1, and job2step1 finishes after job1step1 but before job2step2 has started, then both job1step2 and job2step2 will receive the same input.

As to how to work around this you have a number of options:

1. use an IO manager that is capable of append-style operations - you can create a BigqueryIOManager that takes pandas dataframes like this, and it appears to use the underlying bigquery client library's default of append mode.
2. use deps= to declare dependencies, and then manually load the data you want to pass between assets in their definitions
3. use @ops instead of @assets - this would align better with how you're modeling your workflow
4. use an asset factory to parameterize the creation of individual assets for each of the tables you're trying to model. This would give the bonus of having visual UI elements and materialization history
5. partition the assets on client ID.