Call dbt_constraints.create_constraints() via run-operation?

[squashed94]

Our airflow/cosmos implementation does a dbt invocation for each model in the DAG rather than a single invocation for all models. Not ideal but it's currently what we have...

Calling dbt_constraints.create_constraints() via an on-run-hook will cause an error for yet to be processed models that don't exist in the db (snowflake) - which will typically present during the first run of a code deployment.

Can run-operation be used to overcome this?

Thanks

[sfc-gh-dflippo]

I'm surprised that you are receiving an error because the code should be testing for the existence of all the tables. Does Airflow run each model with --select [model] ? That is something that I expected people to be using. Can you give me more details so I can reproduce it with the integration tests? I don't use Airflow/Cosmos, and I've been cynical that it was mostly created to increase the number of jobs executed on Astronomer. :)

[sfc-gh-dflippo]

Is your error occurring during the dbt Constraints process or when the test executes?

This is my test to try to recreate your issue:

• added on-run-start to drop dim_orders if it exists
• Executed dbt build --select fact_order_line

This is the resulting error:

14:27:53    Database Error in test relationships_fact_order_line_l_orderkey__o_orderkey__ref_dim_orders_ (models/schema.yml)
  002003 (42S02): SQL compilation error:
  Object 'DEV.DBT_DEMO.DIM_ORDERS' does not exist or not authorized.
  compiled Code at target/run/dbt_constraints_integration_tests/models/schema.yml/relationships_fact_order_line_ba65f24463b9568f96716793036e580e.sql
14:27:53  
14:27:53    on-run-end failed, error:
 002003 (42S02): SQL compilation error:
Table 'DEV.DBT_DEMO.DIM_ORDERS' does not exist or not authorized.

My error is not occuring because of dbt Constraints however. If you note carefully, the error is occuring in the out of the box "relationships" test that is part of dbt Core. They did not implement any logic to test whether the parent table exists and produce an error. I also did not put code in my tests to conditionally execute them. I do have lots of code to verify that objects exist when the on-run-end macro executes.

[sfc-gh-dflippo]

I figured it out. You are running dbt run, not dbt build and it is still trying to create the primary key on the parent table when it doesn't exist. I've added logic to test that every table is checked if it exists and is a table, not a view. I'll be deploying the code shortly after I test it on more than just Snowflake.

[squashed94]

Hi Dan...thanks for the quick turnaround. It hasn't quite solved the issue for me.

If tests are detected in the config then a dbt test command is sent, but you're correct, up until now (the new config being added) we do a dbt run --select

fact_leave_test - is the new table added in the config that did not exist in the db.

[2024-09-30T21:15:23.208+0000] {subprocess.py:73} INFO - Running command: ['/usr/local/airflow/dbt_venv/bin/dbt', 'test', '--vars', 'raw_database_name: DB_BDH_RAW_SYST02\n', '--models', 'dim_time', '--project-dir', '/tmp/tmp92wsgc70', '--profiles-dir', '/tmp/cosmos/profile/df425265a6ba65f881f517de1216dee898687418e3f36788053e9b26d74f9c18', '--profile', 'default', '--target', 'dev']
[2024-09-30T21:15:23.224+0000] {subprocess.py:84} INFO - Command output:
[2024-09-30T21:15:25.173+0000] {subprocess.py:94} INFO - �[0m21:15:25 Running with dbt=1.7.17
[2024-09-30T21:15:25.518+0000] {subprocess.py:94} INFO - �[0m21:15:25 Registered adapter: snowflake=1.7.1
[2024-09-30T21:15:25.814+0000] {subprocess.py:94} INFO - �[0m21:15:25 Found 133 models, 8 seeds, 2 operations, 245 tests, 4 sources, 0 exposures, 0 metrics, 669 macros, 0 groups, 0 semantic models
[2024-09-30T21:15:25.822+0000] {subprocess.py:94} INFO - �[0m21:15:25
[2024-09-30T21:15:26.516+0000] {subprocess.py:94} INFO - �[0m21:15:26 Concurrency: 32 threads (target='dev')
[2024-09-30T21:15:26.517+0000] {subprocess.py:94} INFO - �[0m21:15:26
[2024-09-30T21:15:26.538+0000] {subprocess.py:94} INFO - �[0m21:15:26 1 of 16 START test dbt_constraints_foreign_key_fact_colleague_confidential_snapshot_day_time_key__time_key__ref_dim_time_ [RUN]
[2024-09-30T21:15:26.541+0000] {subprocess.py:94} INFO - �[0m21:15:26 2 of 16 START test dbt_constraints_foreign_key_fact_colleague_confidential_time_key__time_key__ref_dim_time_ [RUN]
[2024-09-30T21:15:26.544+0000] {subprocess.py:94} INFO - �[0m21:15:26 3 of 16 START test dbt_constraints_foreign_key_fact_colleague_snapshot_day_time_key__time_key__ref_dim_time_ [RUN]
[2024-09-30T21:15:26.547+0000] {subprocess.py:94} INFO - �[0m21:15:26 4 of 16 START test dbt_constraints_foreign_key_fact_colleague_time_key__time_key__ref_dim_time_ [RUN]
[2024-09-30T21:15:26.554+0000] {subprocess.py:94} INFO - �[0m21:15:26 5 of 16 START test dbt_constraints_foreign_key_fact_leave_liability_snapshot_day_time_key__time_key__ref_dim_time_ [RUN]
[2024-09-30T21:15:26.556+0000] {subprocess.py:94} INFO - �[0m21:15:26 6 of 16 START test dbt_constraints_foreign_key_fact_leave_liability_time_key__time_key__ref_dim_time_ [RUN]
[2024-09-30T21:15:26.557+0000] {subprocess.py:94} INFO - �[0m21:15:26 7 of 16 START test dbt_constraints_foreign_key_fact_leave_snapshot_day_time_key__time_key__ref_dim_time_ [RUN]
[2024-09-30T21:15:26.559+0000] {subprocess.py:94} INFO - �[0m21:15:26 8 of 16 START test dbt_constraints_foreign_key_fact_leave_test_time_key__time_key__ref_dim_time_ [RUN]
[2024-09-30T21:15:26.560+0000] {subprocess.py:94} INFO - �[0m21:15:26 9 of 16 START test dbt_constraints_foreign_key_fact_leave_time_key__time_key__ref_dim_time_ [RUN]
[2024-09-30T21:15:26.561+0000] {subprocess.py:94} INFO - �[0m21:15:26 10 of 16 START test dbt_constraints_foreign_key_fact_organisation_unit_snapshot_day_time_key__time_key__ref_dim_time_ [RUN]
[2024-09-30T21:15:26.561+0000] {subprocess.py:94} INFO - �[0m21:15:26 11 of 16 START test dbt_constraints_foreign_key_fact_organisation_unit_time_key__time_key__ref_dim_time_ [RUN]
[2024-09-30T21:15:26.564+0000] {subprocess.py:94} INFO - �[0m21:15:26 12 of 16 START test dbt_constraints_foreign_key_fact_pay_variance_time_key__time_key__ref_dim_time_ [RUN]
[2024-09-30T21:15:26.564+0000] {subprocess.py:94} INFO - �[0m21:15:26 13 of 16 START test dbt_constraints_foreign_key_fact_personnel_expense_time_key__time_key__ref_dim_time_ [RUN]
[2024-09-30T21:15:26.566+0000] {subprocess.py:94} INFO - �[0m21:15:26 14 of 16 START test dbt_constraints_foreign_key_fact_position_snapshot_day_time_key__time_key__ref_dim_time_ [RUN]
[2024-09-30T21:15:26.567+0000] {subprocess.py:94} INFO - �[0m21:15:26 15 of 16 START test dbt_constraints_foreign_key_fact_position_time_key__time_key__ref_dim_time_ [RUN]
[2024-09-30T21:15:26.568+0000] {subprocess.py:94} INFO - �[0m21:15:26 16 of 16 START test dbt_constraints_primary_key_dim_time_time_key .............. [RUN]
[2024-09-30T21:15:27.598+0000] {subprocess.py:94} INFO - �[0m21:15:27 2 of 16 PASS dbt_constraints_foreign_key_fact_colleague_confidential_time_key__time_key__ref_dim_time_ [�[32mPASS�[0m in 1.04s]
[2024-09-30T21:15:27.613+0000] {subprocess.py:94} INFO - �[0m21:15:27 8 of 16 ERROR dbt_constraints_foreign_key_fact_leave_test_time_key__time_key__ref_dim_time_ [�[31mERROR�[0m in 1.04s]
[2024-09-30T21:15:27.805+0000] {subprocess.py:94} INFO - �[0m21:15:27 6 of 16 PASS dbt_constraints_foreign_key_fact_leave_liability_time_key__time_key__ref_dim_time_ [�[32mPASS�[0m in 1.23s]
[2024-09-30T21:15:27.825+0000] {subprocess.py:94} INFO - �[0m21:15:27 15 of 16 PASS dbt_constraints_foreign_key_fact_position_time_key__time_key__ref_dim_time_ [�[32mPASS�[0m in 1.21s]
[2024-09-30T21:15:27.849+0000] {subprocess.py:94} INFO - �[0m21:15:27 12 of 16 PASS dbt_constraints_foreign_key_fact_pay_variance_time_key__time_key__ref_dim_time_ [�[32mPASS�[0m in 1.27s]
...

[2024-09-30T21:15:28.171+0000] {subprocess.py:94} INFO - �[0m21:15:28 Running 2 on-run-end hooks
[2024-09-30T21:15:28.892+0000] {subprocess.py:94} INFO - �[0m21:15:28 1 of 2 START hook: dbt_main.on-run-end.0 ....................................... [RUN]
[2024-09-30T21:15:28.892+0000] {subprocess.py:94} INFO - �[0m21:15:28 1 of 2 OK hook: dbt_main.on-run-end.0 .......................................... [�[32mOK�[0m in 0.00s]
[2024-09-30T21:15:28.903+0000] {subprocess.py:94} INFO - �[0m21:15:28 Running dbt Constraints
[2024-09-30T21:15:29.563+0000] {subprocess.py:94} INFO - �[0m21:15:29 Creating foreign key: FACT_COLLEAGUE_CONFIDENTIAL_TIME_KEY_FK referencing DIM_TIME ['time_key'] RELY
[2024-09-30T21:15:29.751+0000] {subprocess.py:94} INFO - �[0m21:15:29 Database error while running on-run-end
[2024-09-30T21:15:29.844+0000] {subprocess.py:94} INFO - �[0m21:15:29
[2024-09-30T21:15:29.845+0000] {subprocess.py:94} INFO - �[0m21:15:29 Finished running 16 tests, 1 hook in 0 hours 0 minutes and 4.02 seconds (4.02s).
[2024-09-30T21:15:29.905+0000] {subprocess.py:94} INFO - �[0m21:15:29
[2024-09-30T21:15:29.906+0000] {subprocess.py:94} INFO - �[0m21:15:29 �[31mCompleted with 2 errors and 0 warnings:�[0m
[2024-09-30T21:15:29.906+0000] {subprocess.py:94} INFO - �[0m21:15:29
[2024-09-30T21:15:29.907+0000] {subprocess.py:94} INFO - �[0m21:15:29 Database Error in test dbt_constraints_foreign_key_fact_leave_test_time_key__time_key__ref_dim_time_ (models/people/people_properties.yml)
[2024-09-30T21:15:29.908+0000] {subprocess.py:94} INFO - 002003 (42S02): SQL compilation error:
[2024-09-30T21:15:29.908+0000] {subprocess.py:94} INFO - Object 'DB_PEOPLE_SYST02.DP_PEOPLE.FACT_LEAVE_TEST' does not exist or not authorized.
[2024-09-30T21:15:29.908+0000] {subprocess.py:94} INFO - compiled Code at target/run/dbt_main/models/people/people_properties.yml/dbt_constraints_foreign_key_fa_f63a7f8b38211da8ab21a08c84384c40.sql
[2024-09-30T21:15:29.908+0000] {subprocess.py:94} INFO - �[0m21:15:29
[2024-09-30T21:15:29.909+0000] {subprocess.py:94} INFO - �[0m21:15:29 on-run-end failed, error:

...

These error seem to be prefixed and originating from dbt_constraints. Is that what you were expecting?

Thanks,

[squashed94]

I suppose the other issue with the on-end hook acting as the trigger in this Airflow setup is that dbt_constraints will be trying to apply constraints (for the same tables) concurrently with as many tasks that are executing in parallel.

A single pass at the end of the dag's processing would be preferable and I suppose I can manufacture that by toggling the config (dbt_constraints_enabled: true) when running for a single model.