[EPIC] Incremental Model Improvements - Microbatch

[QMalcolm]

Incremental models in dbt is a materialization strategy designed to efficiently update your data warehouse tables by only transforming and loading new or changed data since the last run. Instead of processing your entire dataset every time, incremental models append or update only the new rows, significantly reducing the time and resources required for your data transformations.

Even with all the benefits of incremental models as they exist today, there are limitations with this approach, such as:

• burden is on YOU to calculate what’s “new” - what has already been loaded, what needs to be loaded, etc.
• can be slow if you have many partitions to process (like when running in full-refresh mode) as it’s done in “one big” SQL statement - can time out, if it fails you end up needing to retry already successful partitions, etc.
• if you want to specifically name a partition for your incremental model to process, you have to add additional “hack”y logic, likely using vars
• data tests run on your entire model, rather than just the "new" data

In this project we're aiming to make incremental models easier to implement and more efficient to run.

P0s - Core

P0s - Core Framework

Beta Give feedback

1. [Feature] event time config #9490
   enhancement +1
   
2. [Feature] Accept and render EventTimeFilter in BaseRelation dbt-adapters#294
   +0
   
3. [Feature] --event-start-time and --event-end-time as CLI flags for specifying microbatch start and end times #10635
   enhancement +1
   
4. [Feature] Allow setting of batch_size as a top-level property of a model #10637
   enhancement +1
   
5. [Feature] Build WHERE filter for reading data of microbatch incremental builds from properties #10638
   enhancement +1
   
6. [Feature] Automatically set --event-start-time and --event-end-time values if not provided #10636
   enhancement +1
   
7. [Feature] Allow setting of a lookback as top level model property #10662
   enhancement +1
   
8. [Feature] the rendering of ref and sourceshould have microbatch WHERE filter when appropriate #10639
   enhancement +1
   
9. [Feature] Default microbatch strategy implementation + testing in dbt-adapters dbt-adapters#302
   +0
   
10. Materialize updates for microbatch incremental models in batches #10700
    +0
    
11. Add test utility to patch microbatch end time #10713
    +0
    
12. [dbt-snowflake] Microbatch strategy dbt-snowflake#1182
    +0
    
13. Support batch-level retries in dbt retry for microbatch models #10729
    +0
    
14. Support batch-level retries + serialize batch-level results in RunResults #10715
    +0
    
15. Require defining a begin config for microbatch incremental models #10701
    +0
    
16. [Improvement] Parse time validation of microbatch model and source configs #10709
    +0
    
17. https://github.com/dbt-labs/iglu.getdbt.com/issues/28
18. Write compiled code to one file per batch during execution of microbatch models #10714
    +0
    
19. Track incremental_strategy as part of model_run tracking event #10761
    +0
    
20. Ensure microbatch models respect model level full_refresh config during batch creation #10785
    +0
    
21. Ensure prior batch successes are propagated in RunResults during multiple dbt retry scenarios #10800
    +0
    
22. Replace DBT_EXPERIMENTAL_MICROBATCH env var with behaviour flag #10798
    user docs +1
    
23. Begin supplying a batch_id in the mode context for batches given to adapters #11025
    +0
    
24. Emit warning if node-level batch_concurrency is set to True while adapter does not support MicrobatchConcurrency #11122
    +0

P0s - Adapters

P0s - Adapters

Beta Give feedback

1. [dbt-postgres] Microbatch strategy implementation: merge dbt-postgres#149
   +0
   
2. [dbt-snowflake] Microbatch strategy dbt-snowflake#1182
   +0
   
3. [dbt-spark] Microbatch Incremental Strategy dbt-spark#1109
   +0
   
4. [dbt-bigquery] Microbatch Strategy dbt-bigquery#1354
   +0
   
5. [dbt-redshift] Microbatch strategy dbt-redshift#923
   +0
   
6. [dbt-athena] Microbatch Strategy dbt-athena#760
   +0
7. Introduce new Capability for MicrobatchConcurrency support dbt-adapters#359
   +0
   
8. [Microbatch] Update default make_temp_relation macro to incorporate a batch specific identifier if available dbt-adapters#360
   +0
   
9. Implement MicrobatchConcurrency for dbt-postgres
10. Enable MicrobatchConcurrency for dbt-snowflake dbt-snowflake#1260
    +0
    
11. Implement MIcrobatchConcurrency for dbt-bigquery
12. Implement MicrobatchConcurrency for dbt-redshift
13. Implement MicrobatchConcurrency for dbt-spark
14. Refactor dbt-postgres to use batch context object instead of __dbt_internal variables dbt-postgres#183
    +0
    
15. Refactor dbt-bigquery to use batch context object instead of __dbt_internal variables dbt-bigquery#1432
    +0
    
16. Refactor dbt-redshift to use batch context object instead of __dbt_internal variables dbt-redshift#966
    +0
    
17. Refactor dbt-spark to use batch context object instead of __dbt_internal variables dbt-spark#1156
    +0
    
18. Refactor dbt-snowflake to use batch context object instead of __dbt_internal variables dbt-snowflake#1272
    +0
    

Bugs

Beta bugs

Beta Give feedback

1. [Bug] Defaulting lookback to 0 results in consistently incomplete batches #10867
   bug microbatch user docs +3
   
2. [Bug] On first-run/full-refresh the incremental strategy of a model isn't validated dbt-adapters#330
   bug incremental +2
   
3. [Bug] canceling a run of a microbatch model should cause remaining batches to be skipped #10862
   bug microbatch +2
   
4. [Bug] Invalid where filter for latest batch when event_column is of type date #10868
   bug microbatch +2
   
5. [Bug] Microbatch error dbt-bigquery#1376
   bug +1
   
6. [Feature] Emit a warning if none of a microbatch model's ref/source inputs have an event_time defined #10926
   enhancement microbatch +2
   
7. [Bug] Microbatch: 'model' context variable is of ModelNode type instead of dict #10927
   +0
   
8. [Bug] Cartesian Join based deletion is causing performance problems when it hits a certain scale for microbatch models dbt-snowflake#1228
   bug +1
   
9. [Bug] Database errors in microbatches are missing "node_info" in structured logs #10840
   Impact: Exp bug incremental logging microbatch pre-release +6
   
10. [Bug] Microbatch: Calling ref() in a macro with a microbatch model name results in error #10928
    bug microbatch pre-release +3
    
11. [Bug] Microbatch timestamps incorrectly offset for system timezone in Snowflake dbt-snowflake#1256
    bug microbatch +2
    
12. [Bug] [Nit] Incorrect plural of partial success #10999
    bug microbatch +2
    
13. [Bug] Skipping microbatch model fails with typeError: object of type 'Field' has no len() #11021
    bug microbatch +2
    
14. [Bug] Microbatch does not work with date field types dbt-bigquery#1412
    bug +1
    
15. Run pre/post model hooks once per microbatch model #11094
    +0
16. Incremental microbatch dbt list --output JSON raises TypeError: Object of type datetime is not JSON serializable #11098
    +0

P1s

P1s

Beta Give feedback

1. Validate --event-time-start is earlier than --event-time-end #10786
   +0
   
2. Set a singular "current_time" for microbatch models per invocation #10819
   +0
   
3. [Feature] Make --event-end-time require --event-start-time and vice versa #10874
   enhancement microbatch +2
   
4. [Feature] microbatches should default run in parallel according to threads #10855
   enhancement microbatch +2
   
5. [Feature] automatically detect which batches can not be run in parallel #10853
   enhancement microbatch +2
   
6. [Microbatch] Leverage static partitioning in dbt-bigquery materialization strategy dbt-bigquery#1355
   +0
   
7. [Feature] Microbatch should allow copy_partitions dbt-bigquery#1414
   enhancement microbatch +2
   

P2s

P2s

Beta Give feedback

1. Make begin on microbatch incremental models optional by calculating min of mins #10702
   +0
   
2. Ensure internal nodes with resource_class properties only serialize keys defined in resource_class #11018
   +0
3. cold_storage configuration - I can limit how much old data we need to maintain as active or readily available

[MaartenN1234]

Just for my understanding: Is it right that this issue seeks to address technical (performance/load) issues in models that take just one single ref as source (or if it has other sources as well we assume them to be stale) ?

I am looking for ways to support incremental processing of multi-table join models (e.g. https://discourse.getdbt.com/t/template-for-complex-incremental-models/10054, but I've seen many more similar help requests on community forums). To be sure, such features will not be in scope right ?

[QMalcolm]

Just for my understanding: Is it right that this issue seeks to address technical (performance/load) issues in models that take just one single ref as source (or if it has other sources as well we assume them to be stale) ?

I am looking for ways to support incremental processing of multi-table join models (e.g. https://discourse.getdbt.com/t/template-for-complex-incremental-models/10054, but I've seen many more similar help requests on community forums). To be sure, such features will not be in scope right ?

@MaartenN1234 I'm not sure that I fully understand the question being asked. For my clarity, is the question whether this new functionality will support more than one input to an incremental model? If so, the answer is yes!

For example, say we turn the jaffle-shop customers model into an incremental microbatch model. It'd look like the following

{{ config(materialized='incremental', incremental_strategy='microbatch', unique_key='id', event_time='created_at', batch_size='day') }}

with

customers as (
    select * from {{ ref('stg_customers') }}
),

orders as (
    select * from {{ ref('orders') }}
),

customer_orders_summary as (
    select
        orders.customer_id,
        count(distinct orders.order_id) as count_lifetime_orders,
        count(distinct orders.order_id) > 1 as is_repeat_buyer,
        min(orders.ordered_at) as first_ordered_at,
        max(orders.ordered_at) as last_ordered_at,
        sum(orders.subtotal) as lifetime_spend_pretax,
        sum(orders.tax_paid) as lifetime_tax_paid,
        sum(orders.order_total) as lifetime_spend
    from orders
    group by 1
),

joined as (
    select
        customers.*,
        customer_orders_summary.count_lifetime_orders,
        customer_orders_summary.first_ordered_at,
        customer_orders_summary.last_ordered_at,
        customer_orders_summary.lifetime_spend_pretax,
        customer_orders_summary.lifetime_tax_paid,
        customer_orders_summary.lifetime_spend,
        case
            when customer_orders_summary.is_repeat_buyer then 'returning'
            else 'new'
        end as customer_type
    from customers

    left join customer_orders_summary
        on customers.customer_id = customer_orders_summary.customer_id
)

select * from joined

If the models orders and stg_customers both have an event_time defined (they don't need to be incremental themselves), then they will automatically be filtered and batched by the generated event time filters.

[MaartenN1234]

Just for my understanding: Is it right that this issue seeks to address technical (performance/load) issues in models that take just one single ref as source (or if it has other sources as well we assume them to be stale) ?
I am looking for ways to support incremental processing of multi-table join models (e.g. https://discourse.getdbt.com/t/template-for-complex-incremental-models/10054, but I've seen many more similar help requests on community forums). To be sure, such features will not be in scope right ?

@MaartenN1234 I'm not sure that I fully understand the question being asked. For my clarity, is the question whether this new functionality will support more than one input to an incremental model? If so, the answer is yes!

For example, say we turn the jaffle-shop customers model into an incremental microbatch model. It'd look like the following

{{ config(materialized='incremental', incremental_strategy='microbatch', unique_key='id', event_time='created_at', batch_size='day') }}

with

customers as (
    select * from {{ ref('stg_customers') }}
),

orders as (
    select * from {{ ref('orders') }}
),

customer_orders_summary as (
    select
        orders.customer_id,
        count(distinct orders.order_id) as count_lifetime_orders,
        count(distinct orders.order_id) > 1 as is_repeat_buyer,
        min(orders.ordered_at) as first_ordered_at,
        max(orders.ordered_at) as last_ordered_at,
        sum(orders.subtotal) as lifetime_spend_pretax,
        sum(orders.tax_paid) as lifetime_tax_paid,
        sum(orders.order_total) as lifetime_spend
    from orders
    group by 1
),

joined as (
    select
        customers.*,
        customer_orders_summary.count_lifetime_orders,
        customer_orders_summary.first_ordered_at,
        customer_orders_summary.last_ordered_at,
        customer_orders_summary.lifetime_spend_pretax,
        customer_orders_summary.lifetime_tax_paid,
        customer_orders_summary.lifetime_spend,
        case
            when customer_orders_summary.is_repeat_buyer then 'returning'
            else 'new'
        end as customer_type
    from customers

    left join customer_orders_summary
        on customers.customer_id = customer_orders_summary.customer_id
)

select * from joined

If the models orders and stg_customers both have an event_time defined (they don't need to be incremental themselves), then they will automatically be filtered and batched by the generated event time filters.

The critical requirement for me, is that matching rows (on the join condition) in both sources are not neccesarily created in the same batch. So when the filter is on the sources independently:
select * from {{ ref('stg_customers') }} where event_time > last_processed_event_time
and
select * from {{ ref('orders') }} where event_time > last_processed_event_time

stuff will be wrong (e.g. if we would load one more order, we would loose all previous from the aggregate or when the customer data is updated while no new orders for this client are to be processed the update will not be propagated).

To get it right, it should become somewhat like this:
select * from {{ ref('stg_customers') }} where event_time > last_processed_event_time or (customer_id IN ( select customer_id from {{ ref('orders') }} where event_time > last_processed_event_time))
and
select * from {{ ref('orders') }} where (customer_id IN ( select customer_id from {{ ref('stg_customers') }} where event_time > last_processed_event_time UNION ALL select customer_id from {{ ref('orders') }} where event_time > last_processed_event_time))

So one needs to incorporate the join clause and the aggregation into the change detection

[QMalcolm]

Sorry for accidentally closing this as completed last week. As penance, here is a photo of my cat Misu. He is very excited about microbatch models