What is the recommended way to deal with late-arriving data in source tables of incremental models?

[plaflamme]

Assuming a large source table with the following data:

received_at	event	value
2023-12-10T00:00:00	viewed_page	abcd
2023-12-10T00:10:00	viewed_page	efgh

And an incremental model:

config {
  type: "incremental"
}

pre_operations {
  DECLARE event_timestamp_checkpoint DEFAULT (
    ${
      when(
        incremental(),
        `SELECT MAX(received_at) FROM ${self()}`,
      )
    }
  )
}

SELECT * FROM ${ref("source")}
WHERE received_at > event_timestamp_checkpoint

Then the next day, the table looks like this:

received_at	event	value
2023-12-10T00:00:00	viewed_page	abcd
2023-12-10T00:10:00	viewed_page	efgh
2023-12-10T00:05:00	viewed_page	late
2023-12-11T00:01:00	viewed_page	new_day

If we don't do anything in particular, the incremental model will miss the row that was inserted "before" the latest in the target table.

If we use a "lookback" value in the FROM clause (say max(received_at) - interval 1 day), we'll end up with duplicated data.

We could use the unique_key approach, but this isn't always applicable and is potentially prohibitively expensive on large tables / partitions.

What is the recommended approach to deal with this kind of late-arriving data?

[BenBirt]

I can only think of one other option than the ones you've already listed: you could store a data_arrival_timestamp (as well as received_at), which always increases, regardless of any event timestamp. Then do your incremental selection based on that timestamp, and it's always safe.

[plaflamme]

Right, but this requires changes to the source table which may not be possible.

BigQuery's MERGE operation can do this, see here for an example.

I believe this is an example of how this could work:

-- existing incremental table
create or replace table events as (
  select TIMESTAMP('2023-12-09T00:00:00') as received_at, "viewed_page" as event, "early" as value
  union all
  select TIMESTAMP('2023-12-10T00:00:00') as received_at, "viewed_page" as event, "abcd" as value
  union all
  select TIMESTAMP('2023-12-10T00:10:00') as received_at, "viewed_page" as event, "efgh" as value
);
-- events selected from the source, which includes the late arriving event this time
create or replace table new_events as (
  select TIMESTAMP('2023-12-10T00:00:00') as received_at, "viewed_page" as event, "abcd" as value
  union all
  select TIMESTAMP('2023-12-10T00:10:00') as received_at, "viewed_page" as event, "efgh" as value
  union all
  select TIMESTAMP('2023-12-10T00:05:00') as received_at, "viewed_page" as event, "late" as value
  union all
  select TIMESTAMP('2023-12-11T00:00:00') as received_at, "viewed_page" as event, "new_day" as value
);
MERGE events T
USING new_events S
ON FALSE
WHEN NOT MATCHED AND received_at >= TIMESTAMP('2023-12-10T00:00:00')THEN
  INSERT(received_at,event,value) VALUES(received_at,event,value)
WHEN NOT MATCHED BY SOURCE AND received_at >= TIMESTAMP('2023-12-10T00:00:00') THEN
  DELETE;

select * from events order by received_at

[BenBirt]

Dataform does a MERGE operation if you specify uniqueKey (https://cloud.google.com/dataform/docs/incremental-tables#merge_rows_in_an_incremental_table) - but I thought you'd discounted uniqueKey in your original question (?).

Or perhaps I am missing something else?

[plaflamme]

Right, but notice the ON FALSE statement here. If I'm not mistaken, specifying a unique key would make that ON T.unique_key = S.unique_key.

This is what the BigQuery documentation states:

If the merge_condition is FALSE, the query optimizer avoids using a JOIN. This optimization is referred to as a constant false predicate. A constant false predicate is useful when you perform an atomic DELETE on the target plus an INSERT from a source (DELETE with INSERT is also known as a REPLACE operation).

Which is precisely what we'd like to do: delete all rows after a timestamp from the target and insert all rows after the same timestamp from the source.

Here's the documentation for both when clauses:

The not_matched_by_target_clause defines how to insert into the target table if a row from source table does not match any row in the target table.

which is this part (insert all rows from source that are after timestamp):

WHEN NOT MATCHED AND received_at >= TIMESTAMP('2023-12-10T00:00:00')THEN
  INSERT(received_at,event,value) VALUES(received_at,event,value)

The not_matched_by_source_clause defines how to update or delete a row in the target table if that row does not match any row in the source table.

which is this part (delete all rows from target that are after timestamp):

WHEN NOT MATCHED BY SOURCE AND received_at >= TIMESTAMP('2023-12-10T00:00:00') THEN
  DELETE;

[BenBirt]

OK, understood, so you do want the MERGE operation, but you want to avoid the join, and you know that is safe because you want a strict delete / insert.

Unfortunately Dataform doesn't have built-in support for this today. For now, the workaround would be to encapsulate all incremental logic in an operations script.

[BenBirt]

I filed a feature request here: #1638