How does Materialize deal with slowly changing data (facts that are never complete?)

[ruf-io]

Let's say I have an order entity that might always change in any moment from one status to another, and never become fully finalized.

For example, imagine there is a status field on the order entity that typically goes from pending to complete, but every once in a while a customer calls to complain about an old order and the status gets updated to disputed. Say you have a materialized view that filters for all disputed order to calculate the SUM of disputed order revenue.

I've read elsewhere that Materialize is strongly consistent and other stream processing systems are eventually consistent.

1. Does this mean that Materialize would withhold output until it is sure a record will never change again?
2. If I'm calculating a SUM and the disputed status comes through on a really old order, where does Materialize go to find the amount for that order?

To put it another way, can Materialize deal with data that isn't just an immutable stream of events (e.g. clicks, impressions)

(xPosting this question from a community slack discussion for future reference.)

[ruf-io]

TL;DR: The example described is a great use-case for Materialize (because it is unbounded change on bounded sets of data), and it handles it well. But that shouldn't be too surprising because this is not the type of "late-arriving data" that typically causes issues with streaming systems.

This answer can be broken into a few parts:

1. How would Materialize handle the exact scenario described above.
2. How does Materialize generally handle time
3. Are there other kinds of late-arriving data issues that are more difficult to handle?

1. How would Materialize handle the exact scenario described above?

This is very straightforward. As a human interpreting the data we might classify it as a "late arriving," but to a stream processor it's just more change events.

To recap: You have an order entity that typically changes from pending to complete, but every once in a while a customer lodges a complaint long after it has completed, causing the status to go to disputed. You want to create a Materialized view that shows the SUM of disputed revenue.

Originating from a database

If the order entity is originating from a database, Materialize is pulling in CDC change events from kafka that look roughly like:

{
  order_id: 1234,
  time: 1614747515,
  "before" : {
    "state" : "complete",
    <... EVERY OTHER COLUMN ...>
   },
   "after" : {
     "state" : "disputed",
     <... EVERY OTHER COLUMN ...>
   }
}

In a properly functioning setup*, Materialize will treat this like an update because it already has records on order_id=1234, and Materialize will do the minimum amount of work to update materialized views that change when the state changes from complete to disputed.

*Properly functioning = Tools like Debezium avoid potential issues created by starting after a database has state and activities by starting with a big snapshot->replay of "current state" of the database before switching to streaming of live changes.

Originating from a stream

If your events are originating from a stream the answer is mostly the same with the caveat that if you need Materialize to have access to the entire history of your stream, you'll either need:

1. Infinite data retention in kafka (I think only confluent provides this right now)
2. Use kafka log compaction in a way that allows you to know the state of the order leading in to events currently retained in your topic, and materialize the view using UPSERT ENVELOPE
3. Soon - Use an S3 source to supplement (See note below)

NOTE: S3 Sources coming soon
For scenarios where you are only keeping the tip of your data in Kafka, but a full history in S3, we are building a feature that allows you to use S3 as a source in order to give Materialize access to the full history. See #4914

2. How does Materialize generally handle time

"Time" in materialize is something you get to define, either explicitly by specifying it in your CDC input, or implicitly by letting us pick a timestamp. If you let us do it data are never "late" (like wizards), and if you define times yourself we will hold back results until we are certain they are correct (part of the CDC input are statements about the completion of times; like transaction boundaries in logical replication).

3. Are there other kinds of late-arriving data issues that are more difficult to handle?

TBD

For additional reference, see: Eventual consistency isnt for streaming