FK Contraint creation fails if FK relation doesn't exist

[kokorin]

We use DBT Slim CI to test MR changes (with --deffer and --state options).

Recently we started to add FK tests and CI failed:

1. Model Customer has PK test
2. Model Address has PK + FK (to Customer) tests
3. Model Address was changed and executed by Slim CI, FK test was executed (and passed) too
4. dbt_contraints run on-run-end hook and fails because it can't find Customer table

It appeared that dbt_constraints doesn't check if table exists in lookup_table_columns macro.

Details:
create_constraints_by_type creates DBT Relations similar to this:

{%- set table_relation = api.Relation.create(
    database=table_models[0].database,
    schema=table_models[0].schema,
    identifier=table_models[0].alias ) -%}

In deferred mode DB or Schema can be different from model's DB or Schema. It is correct to not create constraints on deferred models, but the problem is that target table may not exist due to Slim CI. But even if target table exists it may be outdated, so constraints may be incorrect. The conclusion is that we can create constraints only if FK model was actually executed.

Our workaround:

{%- macro snowflake__lookup_table_columns(table_relation, lookup_cache) -%}
{#-
    This macro tweaks default dbt_constraints behavior: during Slim CI some model can be deferred
    and actual tables/views may be absent in target DB/schema.
    We make sure that lookup_columns returns columns only for executed models.
-#}

{%- if table_relation not in lookup_cache.table_columns -%}
    {%- set model_executed = results |  map(attribute="node")
                        | selectattr("resource_type", "equalto", "model")
                        | selectattr("database", "equalto", table_relation.database)
                        | selectattr("schema", "equalto", table_relation.schema)
                        | selectattr("name", "equalto", table_relation.identifier)
                        | list | count -%}
    {%- if not model_executed -%}
        {%- do lookup_cache.table_columns.update({ table_relation: [] }) -%}
        {%- do lookup_cache.not_null_col.update({ table_relation: [] }) -%}
    {%- endif -%}
{% endif -%}

{{ return(dbt_constraints.snowflake__lookup_table_columns(table_relation, lookup_cache)) }}

{%- endmacro -%}