[Bug] The key has to be less than 250 characters

[ms32035]

Astronomer Cosmos Version

1.8.1

dbt-core version

1.9.0

Versions of dbt adapters

No response

LoadMode

DBT_LS_MANIFEST

ExecutionMode

KUBERNETES

InvocationMode

None

airflow version

2.10.4

Operating System

Debian GNU/Linux 12 (bookworm)

If a you think it's an UI issue, what browsers are you seeing the problem on?

No response

Deployment

Official Apache Airflow Helm Chart

Deployment details

No response

What happened?

Looks like for #1433 in case of some tests provided by packages like dbt_utils or dbt_expectations the names generated are absurdly long
and result in:

Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/models/baseoperator.py", line 968, in __init__
    validate_key(task_id)
  File "/home/airflow/.local/lib/python3.12/site-packages/airflow/utils/helpers.py", line 55, in validate_key
    raise AirflowException(f"The key has to be less than {max_length} characters")
airflow.exceptions.AirflowException: The key has to be less than 250 characters

Additionally, it seems that tests are generated as their own task regardless of the test_behaviour setting. Specifically, these are generated when the value is AFTER_ALL

Relevant log output

How to reproduce

Create a test using dbt_utils.relationships_where where table and column names are long

Anything else :)?

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Contact Details

No response

[dosubot]

To address the issue of long task names generated by dbt_utils or dbt_expectations tests in Airflow, which exceed the 250 character limit, you can customize the task ID generation logic. Specifically, you can modify the _get_task_id_and_args function to create shorter task IDs. This can be achieved by adjusting the task_display_name construction or implementing a custom normalize_task_id function to generate concise task IDs ^[1].

Regarding the test_behaviour setting, when set to AFTER_ALL, tests are generated as their own task because this configuration ensures that tests run only after all models have been successfully executed. This behavior is intentional to allow tests to execute collectively at the end of the DAG execution, rather than after each model ^[2].

To continue talking to Dosu, mention @dosu.

---

^{Help Dosu learn! Give it feedback: Great Response | Irrelevant Answer | Incorrect Sources | Too Verbose | Hallucination | Other}

[jtrobles-cdd]

After upgrading Cosmos from 1.7.1 to 1.8.1, I also started experiencing problems caused by excessively long task names, though in my case it's another part of Airflow that's complaining about the 250 character limit:

[2025-01-06, 19:49:30 -03] {validators.py:135} ERROR - Invalid stat name: task.mem_usage.cordada.airflow_data_pipelines.dbt_data_pipelines.refresh_******_******_*****.run_dbt.relationships_stg_**_***_******_******_business_ops__**********_*****_****_********_*******_id_*****__id_*****__ref_stg_**_***_******_******_business_ops__**********_*****_index__test.
Traceback (most recent call last):
  File "/usr/local/lib/python3.10/site-packages/airflow/metrics/validators.py", line 132, in wrapper
    stat = handler_stat_name_func(stat)
  File "/usr/local/lib/python3.10/site-packages/airflow/metrics/validators.py", line 216, in stat_name_default_handler
    raise InvalidStatsNameException(
airflow.exceptions.InvalidStatsNameException: The stat_name (task.mem_usage.cordada.airflow_data_pipelines.dbt_data_pipelines.refresh_******_******_*****.run_dbt.relationships_stg_**_***_******_******_business_ops__**********_*****_****_********_*******_id_*****__id_*****__ref_stg_**_***_******_******_business_ops__**********_*****_index__test) has to be less than 250 characters.

[pankajkoti]

Thank you for reporting the issue.

@ms32035 @jtrobles-cdd, I’m curious—did you only upgrade Cosmos, or was there also an upgrade to Airflow?

Additionally, since Airflow limits task keys to fewer than 250 characters, do you have any suggestions on how we could shorten the task keys while still making them identifiable in terms of the node they represent? Any ideas here would help us address the issue in a more user-friendly way.

@ms32035, regarding your comment:

It seems that tests are generated as their own task regardless of the test_behaviour setting, especially when the value is AFTER_ALL.

This is expected behavior, as outlined in the documentation: https://github.com/astronomer/astronomer-cosmos/blob/main/docs/configuration/testing-behavior.rst. You can refer to the section titled Example when changing the behavior to use TestBehavior.AFTER_ALL.

[ms32035]

@pankajkoti it's just Cosmos upgrade

One workaround idea I have, which is not exactly user friendly is to use the hash of the test that dbt generates

On the AFTER_ALL this isn't the expected behaviour as all tests run as a part of a single dbt test operator, and these have their own node and would probably run twice

[jtrobles-cdd]

did you only upgrade Cosmos, or was there also an upgrade to Airflow?

Only Cosmos. I've been using Airflow 2.10.4 (with Astro Runtime 12.6.0) for many weeks without issues.

[internetcoffeephone]

Running into this too since cosmos==1.8.1.

@pankajkoti We could group these tests by joint parents. E.g. for model_a and model_b, we could create a new Airflow task named model_a-model_b.test, which runs all of the relevant tests. Still verbose, but less so than using the entire test configuration as a unique_id / task_id. I'm not sure about the Cosmos internals here re: failures on tests, so we might have to make model_a-model_b an Airflow TaskGroup and put the tests from individual parents inside of that TaskGroup separately.

[tatiana]

@ms32035 @jtrobles-cdd @internetcoffeephone thanks a lot for reporting / discussing this issue. I'm sorry the fix I made to solve another problem had these unintended side effects.

It feels like we have two bugs:

1. Name of the test task, when it is detached, is ginormous and breaks Airflow
2. Detached test tasks even when TestBehavior.AFTER_ALL is selected

I liked the suggestion of @internetcoffeephone for (1) - how do the others feel about it?

Can we log a separate ticket for (2)? @ms32035, could you do this?

Would anyone be interested in working on this this week? If not, the Astronomer team can take over these two issues starting next week.

In the meantime, to avoid these problems, the recommendation is to use 1.8.0 until we've released 1.8.2 with the fixes.

[internetcoffeephone]

@tatiana @pankajkoti I've given it some more thought and I think there are some issues with my previous suggestion.
The main issue is that it breaks the abstraction of having all Airflow tasks relating to a specific model in a specific TaskGroup.

The reason that this abstraction is important is because it makes it easier to reason about the DAG state - if you are going to make some external DAG depend on a specific model, you don't want to hunt down all downstream tests that may or may not be defined in the model YAML. You just want to create a sensor for the model's TaskGroup, which doesn't work if there exist tests for that model outside of its TaskGroup.

It's unclear to me why before #1433 was implemented, custom_test_combined_model would run for both model_a.test and combined_model.test, since the test is defined here as belonging to combined_model, not to model_a.

Since this test is defined under a specific model according to the YAML file, even if it has other dependencies, shouldn't it be included only in that specific model's TaskGroup? Or am I missing an edge case where you would be able to define out-of-order tests? E.g. adding the test in the YAML file under model_a instead, and in the test SQL, swapping model_a for combined_model (I don't know whether this is possible).

[tatiana]

@internetcoffeephone #1433 was implemented because tests that depend on more than one dbt node had chances of failing. This happened because they were executing every time each model was run, regardless of all the dependencies that had been previously executed.

It's unclear to me why before custom_test_combined_model would run for both model_a.test and combined_model.test, since the test is defined here as belonging to combined_model, not to model_a.

Since the test macro itself depends on both models, as shown below, both dbt test --model model_a and dbt test --model combined_model would run the test custom_test_combined_model:

astronomer-cosmos/dev/dags/dbt/multiple_parents_test/macros/custom_test_combined_model.sql

Lines 2 to 7 in 184e45f

	WITH source_data AS (
	SELECT id FROM {{ ref('model_a') }}
	),
	combined_data AS (
	SELECT id FROM {{ model }}
	)

Please feel free to try it yourself. As part of #1433, we created a straightforward dbt project in #1433 and a DAG that uses it - so we could reproduce the original problem and make sure we would not have regressions once it was solved.

While this is just an example, the problem was reported by several companies that use Cosmos, using real-case dbt projects.

Since this test is defined under a specific model according to the YAML file, even if it has other dependencies, shouldn't it be included only in that particular model's TaskGroup? Or am I missing an edge case where you would be able to define out-of-order tests? E.g. adding the test in the YAML file under model_a instead, and in the test SQL, swapping model_a for combined_model (I don't know whether this is possible).

It would be great if this were the case, but that's not what dbt-core currently does.

We could not find a solution for keeping these multiple-parent tests in the same task groups of individual dependencies while guaranteeing all the dependencies had been resolved.

[internetcoffeephone]

@tatiana Thank you for your explanation, I think I understand now. I tried flipping the dependencies in the yaml file and running dbt list --output json.

If we naively add the test to the Airflow test task of the model TaskGroup wherever it is assigned in the YAML file, it may fail. E.g. if you move the test to model_a like so:

- name: model_a
    description: "A simple model with user data"
    tests:
      - unique:
          column_name: id
      - custom_test_combined_model: {}

and change the test to:

{% test custom_test_combined_model(model) %}
WITH source_data AS (
    SELECT id FROM {{ ref('combined_model') }}
),
...

then we get an almost identical output, except for the test name (and a few similar name changes): instead of custom_test_combined_model_combined_model_ we get custom_test_combined_model_model_a_.

So yes, the case where you can define out-of-order tests exists and we cannot blindly trust the YAML structure. In fact, the YAML structure doesn't even explicitly appear in the JSON output. dbt run would not have this problem, as it would only run these tests after all models have ran.

Then the question becomes: what behavior do we want? What we definitely want is that a test only executes once per run, and only after all of its upstream models have ran.

For any given test with severity=error with multiple parents: if the test fails, do we want to fail:

1. All parents' TaskGroups
2. Only the most downstream TaskGroup
3. Only the test itself

For case 2, there is the problem that a "most downstream" TaskGroup may not explicitly exist, e.g. in a setup where A->B, A->C and you want to test B+C. In this case, we could use alphabetical ordering and pick the first or last.

For both case 2 and 3, we would be breaking the abstraction of a TaskGroup for a model failing if any of its tests fail.

Therefore I'm inclined towards 1. The main objection to case 1 would be that we add more complexity to the DAG. However, it is possible, and it preserves the TaskGroup abstraction. E.g. for models A and B: the test should be added either to the most downstream TaskGroup as defined above, or to a new TaskGroup named e.g. A_B which contains a task that runs all tests with A and B as dependencies. These should then loop back to an EmptyOperator named something like dependent_tests situated within any upstream models' TaskGroups.

The above approach fails if dependencies within the Airflow DAG are specified at TaskGroup level - we'd have to make the dependency explicit on either the model.run or model.test Airflow task.

Happy to hear your opinion here, I'm not 100% sure whether this is what we want - as we're veering outside of regular dbt functionality (better functionality I might add), we can't just try to replicate existing dbt behavior.