Realtime Compute for Apache Flink:Handle out-of-order changelog events

Changelog and stream types

In relational databases such as MySQL, the binary log (binlog) captures every INSERT, UPDATE, and DELETE operation. Flink SQL uses a similar mechanism called the changelog to track data changes and enable incremental processing across streaming pipelines.

A changelog stream falls into one of two categories:

Not all operators can consume update streams. Over aggregation and interval join operators only accept append-only streams as input.

Changelog event types

Flink SQL uses four event types, based on the RowKind enum in the Apache Flink API:

Flink keeps UPDATE_BEFORE (-U) and UPDATE_AFTER (+U) as separate event types rather than combining them into a composite UPDATE event for two reasons:

• Uniform structure: Both events share the same row structure, differentiated only by the RowKind property. A composite event type would require heterogeneous structures or special alignment between INSERT and DELETE events.

• Distributed shuffling: In parallel pipelines, join and aggregation operations shuffle data across tasks. Composite UPDATE events would still need to be split into separate events during shuffling to maintain correctness — so keeping them separate from the start simplifies the model.

How it works

SinkUpsertMaterializer is an intermediate operator that Flink inserts to resolve ordering issues. It was introduced to address FLINK-20374.

To understand why SinkUpsertMaterializer is needed, it helps to understand upsert keys. An upsert key is a column (or set of columns) that preserves the sort order of a unique key through a SQL operation. When upsert keys exist, the downstream operator receives update events in the correct order. When a data shuffling operation breaks unique key ordering — as the JOIN on level does in this example — the upsert key becomes empty.

In this example, rows from s1 are shuffled by level, so the Join output contains rows with the same s1.id value but in arbitrary order. The unique keys are (s1.id), (s1.id, s1.level), and (s1.id, s2.id), but the upsert key is empty. Additionally, the primary key of the sink table (id) does not match the upsert key in the Join output. SinkUpsertMaterializer bridges this gap.

Out-of-order changelog events follow specific rules: for a specific upsert key (or for all columns if the upsert key is empty), ADD events (+I and +U) always occur before the corresponding RETRACT events (-D and -U). A pair of changelog events with the same upsert key are processed by the same task even when data shuffling occurs. These ordering guarantees are what SinkUpsertMaterializer relies on to reconstruct correct results.

The operator works as follows:

1. Maintains a list of RowData values in state, keyed by the deduced upsert key (or by the entire row if the upsert key is empty).

2. On an ADD event (+I or +U): adds or updates the row in state.

3. On a RETRACT event (-U or -D): removes the row from state.

4. Generates correct changelog events based on the primary key of the sink table.

The following diagram shows how SinkUpsertMaterializer handles Cases 2 and 3 from the example above:

• Case 2: When -U (id=1, level=10, attr='a1') arrives last, SinkUpsertMaterializer removes that row from state and generates an UPDATE event based on the second-to-last row. The final result is (id=1, level=20, attr='b1').

• Case 3: When +U (id=1, level=20, attr='b1') arrives, the operator passes it downstream. When -U (id=1, level=10, attr='a1') arrives later, the operator removes the corresponding row from state without emitting an event. The final result is again (id=1, level=20, attr='b1').

For the source code, see SinkUpsertMaterializer (Flink release-1.17).

When SinkUpsertMaterializer is triggered

Flink adds the SinkUpsertMaterializer operator in the following scenarios:

• The sink table has a primary key but the incoming data does not satisfy the UNIQUE constraint. Common causes include:

  • Defining a primary key on the sink table when the source table has no primary key.

  • Excluding the source primary key column when writing to the sink, or mapping a non-primary-key column of the source to the sink primary key.

  • Reducing the precision of a primary key column through type conversion or group aggregation (for example, casting from BIGINT to INT).

  • Transforming the primary key column, such as concatenating multiple columns into one:

    CREATE TABLE students (
      student_id   BIGINT NOT NULL,
      student_name STRING NOT NULL,
      course_id    BIGINT NOT NULL,
      score        DOUBLE NOT NULL,
      PRIMARY KEY(student_id) NOT ENFORCED
    ) WITH (...);
    
    CREATE TABLE performance_report (
      student_info STRING NOT NULL PRIMARY KEY NOT ENFORCED,
      avg_score    DOUBLE NOT NULL
    ) WITH (...);
    
    CREATE TEMPORARY VIEW v AS
    SELECT student_id, student_name, AVG(score) AS avg_score
    FROM students
    GROUP BY student_id, student_name;
    
    -- The concatenated result no longer satisfies the UNIQUE constraint
    -- but is used as the primary key of the sink table.
    INSERT INTO performance_report
    SELECT
      CONCAT('id:', student_id, ',name:', student_name) AS student_info,
      avg_score
    FROM v;

• A data shuffling operation disrupts the sort order of a unique key before writing to the sink table. This is the scenario in the join example above: the JOIN on level shuffles rows from s1, breaking the sort order of the id primary key.

• The table.exec.sink.upsert-materialize parameter is set to force.

Avoid triggering SinkUpsertMaterializer

• Make the partition key used for deduplication or group aggregation match the primary key of the sink table.

• If a single parallelism fits your dataset and you want to avoid out-of-order events, set parallelism to 1 and disable SinkUpsertMaterializer by setting table.exec.sink.upsert-materialize to none.

• If an operator chain exists between the Sink operator and an upstream stateful operator (such as a deduplication or group aggregation operator), and no data accuracy issues occurred with VVR versions earlier than 6.0, migrate the deployment to VVR 6.0 or later. Set table.exec.sink.upsert-materialize to none and keep other configurations unchanged. For migration steps, see Upgrade the engine version of deployments.

When you must use SinkUpsertMaterializer

• Do not write columns produced by non-deterministic functions (such as CURRENT_TIMESTAMP or NOW()) to the sink table. When the upsert key is unavailable, SinkUpsertMaterializer compares entire rows, and non-deterministic values prevent historical rows from being matched and removed, causing state to grow without bound.

• If the operator state grows large enough to affect performance, increase the deployment parallelism. See Configure resources for a deployment.

Known issues

SinkUpsertMaterializer can cause unbounded state growth in the following situations:

• No state TTL, TTL is too long, or TTL is too short: Without a configured time-to-live (TTL), state accumulates indefinitely. An excessively short TTL can also cause problems: if the interval between a DELETE event and its corresponding ADD event exceeds the configured TTL, Flink retains the row in state as dirty data (see FLINK-29225) and produces the following log message:

  int index = findremoveFirst(values, row);
  if (index == -1) {
      LOG.info(STATE_CLEARED_WARN_MSG);
      return;
  }

  Configure TTL based on your business requirements. See Configure a deployment. Realtime Compute for Apache Flink with VVR 8.0.7 or later supports per-operator TTL configuration to reduce resource consumption for large-state deployments. See Configure the parallelism, chaining strategy, and TTL of an operator.

• Non-deterministic columns with no upsert key: If the update stream arriving at SinkUpsertMaterializer has no deducible upsert key and includes columns from non-deterministic functions, historical rows cannot be matched by value and are never deleted, causing continuous state growth.