custom_source_macro.sql

{#
This overrides the default source() macro to include the option
for using BigQuery's time travel functionality. This allows for
for full refreshes in dev to query a source table at an earlier
point-in-time, so that subsequent incremental runs can be tested
in a more realistic manner.
#}

{% macro source(source_name, table_name) %}

{# get the relation using the default source() macro #}
{% set rel = builtins.source(source_name, table_name) %}


{# create a new "relation," as a subquery that optionally use's BQ's time travel functionality #}
{% set new_rel %}
(
select * from {{ rel }}
for system_time as of timestamp_sub(current_timestamp(), interval {{ var("time_travel_days", 6) }} day)
)
{% endset %}

{#
This will return the time-travel-enhanced relation when the following conditions are met:
* --full-refresh flag is used
* target is developement target (default is "dev")
* model uses incremental materialization
* use_time_travel variable == true (--var 'use_time_travel: true' is passed at run time)

or, when the following conditions are met:
* target is developement target (default is "dev")
* model uses incremental materialization
* time_travel_days variable > 0 (--var 'time_travel_days: 5' is passed at run time)
#}

{% if ((flags.FULL_REFRESH and var("use_time_travel", false)) or (is_incremental() and var("time_travel_days", 0) > 0))
    and target.name == var("dev_target_name", "dev")
    and config.get("materialized") == "incremental" %}
    {% do return(new_rel)  %}
{% else %}
    {% do return(rel)  %}
{% endif %}
    
{% endmacro %}