[FLINK-39616][docs] Document ORDER BY usage with FROM_CHANGELOG by raminqaf · Pull Request #28166 · apache/flink

raminqaf · 2026-05-15T08:44:29Z

What is the purpose of the change

CDC streams often deliver events out of order. The PTF ORDER BY clause introduced in FLINK-39256 aligns naturally with FROM_CHANGELOG: partitioning by the primary key and ordering by event time reorders out-of-order CDC events per key before the operations are applied.

Adds a new "Ordering CDC events with ORDER BY" subsection to the FROM_CHANGELOG documentation with the SQL example and the requirements (watermarked time attribute, ASC on the first order column, set semantics via PARTITION BY). Extends the Table API examples with the equivalent partitionBy(...).orderBy(...).fromChangelog() pattern.

Adds a corresponding example in the PartitionedTable#fromChangelog Javadoc.

Adds a cross-reference from the PTF ORDER BY guide to FROM_CHANGELOG so users discover the pattern from either side.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (no)
The serializers: (no)
The runtime per-record code paths (performance sensitive): (no)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: (no)
The S3 file system connector: (no)

Documentation

Does this pull request introduce a new feature? (no)
If yes, how is the feature documented? (docs / JavaDocs)

Was generative AI tooling used to co-author this PR?

Yes (please specify the tool below)

Generated-by: Opus 4.7

flinkbot · 2026-05-15T08:49:42Z

CI report:

cfce65f Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

fhueske

Thanks for the PR @raminqaf!

Left a few comments with improvement suggestions.

Cheers, Fabian

fhueske · 2026-05-15T17:50:15Z

 less than or equal to the watermark are delivered to the eval() method in sorted order. Late
 events (arriving after the watermark) are dropped to maintain the ordering guarantee.

+This pattern is particularly useful for the built-in [`FROM_CHANGELOG`]({{< ref "docs/sql/reference/queries/changelog" >}}#from_changelog) PTF, which converts an append-only CDC stream into a changelog. Combining `PARTITION BY` on the primary key with `ORDER BY` on the event-time column reorders out-of-order CDC events per key before the operations are applied.


Suggested change

This pattern is particularly useful for the built-in [`FROM_CHANGELOG`]({{< ref "docs/sql/reference/queries/changelog" >}}#from_changelog) PTF, which converts an append-only CDC stream into a changelog. Combining `PARTITION BY` on the primary key with `ORDER BY` on the event-time column reorders out-of-order CDC events per key before the operations are applied.

This pattern is particularly useful for the built-in [`FROM_CHANGELOG`]({{< ref "docs/sql/reference/queries/changelog" >}}#from_changelog) PTF, which converts an append-only CDC stream into a changelog. Combining `PARTITION BY` on the primary key with `ORDER BY` on the event-time column reorders out-of-order CDC events per key before the conversion operations are applied.

fhueske · 2026-05-15T17:54:15Z

+
+CDC streams can deliver events out of order. For example, a key's `UPDATE_AFTER` may arrive before its matching `UPDATE_BEFORE` when events are partitioned across upstream brokers. If the source itself does not guarantee ordering, applying such a changelog directly produces incorrect state.
+
+`FROM_CHANGELOG` accepts an [ORDER BY clause]({{< ref "docs/dev/table/functions/ptfs" >}}#ordering) that sorts events within each partition before they are processed. The framework buffers events per partition and flushes them to the function in sorted order once the watermark advances. Late events (arriving after the watermark) are dropped.


Explain what happens with a record that matches a dropped record?
For example what happens to an UPDATE_AFTER that wasn't late if its matching UPDATE_BEFORE is dropped?

Added a section where I describe the behavior in retract/upsert mode

fhueske · 2026-05-15T17:56:23Z

+  op STRING,
+  name STRING,
+  event_time TIMESTAMP_LTZ(3),
+  WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND


Suggested change

WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND

WATERMARK FOR event_time AS event_time - INTERVAL '5' MINUTE

to keep it aligned with the example below?

Thanks for pointing this out. Addressed

fhueske · 2026-05-15T17:59:22Z

+| Incoming row | Current watermark | Outcome |
+|---|---|---|
+| `+I[id: 6, op: 'INSERT', event_time: '10:05']` | `10:00` | Buffered. Emitted later when the watermark passes `10:05`. |
+| `+I[id: 5, op: 'INSERT', event_time: '09:55']` | `10:00` | Dropped. Timestamp is below the current watermark. |


Suggested change

| `+I[id: 5, op: 'INSERT', event_time: '09:55']` | `10:00` | Dropped. Timestamp is below the current watermark. |

| `+I[id: 5, op: 'INSERT', event_time: '09:57']` | `10:00` | Dropped. Timestamp is below the current watermark. |

Otherwise, readers might think that the 5 min difference between 9:55 and 10:00 would have any relevance.

fhueske · 2026-05-15T18:01:30Z

+|---|---|---|
+| `+I[id: 6, op: 'INSERT', event_time: '10:05']` | `10:00` | Buffered. Emitted later when the watermark passes `10:05`. |
+| `+I[id: 5, op: 'INSERT', event_time: '09:55']` | `10:00` | Dropped. Timestamp is below the current watermark. |
+| `+I[id: 7, op: 'INSERT', event_time: '10:11']` | `10:06` | Record `id=6` is emitted; this row is buffered until the watermark passes `10:11`. |


to be precise, the watermark would only be incremented after the record was processed.
So "Current Watermark" shouldn't be incremented to 10:06 yet.
This happens when a new WM message is received and processed by the PTF.

Goode one! Updated the table and made it more precise and clear. Hope it is good now!

CDC streams often deliver events out of order. The PTF ORDER BY clause introduced in FLINK-39256 aligns naturally with FROM_CHANGELOG: partitioning by the primary key and ordering by event time reorders out-of-order CDC events per key before the operations are applied. Adds a new "Ordering CDC events with ORDER BY" subsection to the FROM_CHANGELOG documentation with the SQL example and the requirements (watermarked time attribute, ASC on the first order column, set semantics via PARTITION BY). Extends the Table API examples with the equivalent partitionBy(...).orderBy(...).fromChangelog() pattern. Adds a corresponding example in the PartitionedTable#fromChangelog Javadoc. Adds a cross-reference from the PTF ORDER BY guide to FROM_CHANGELOG so users discover the pattern from either side.

featzhang · 2026-05-16T08:55:04Z

LGTM

raminqaf force-pushed the FLINK-39616 branch from 1a08ae0 to ca8d886 Compare May 15, 2026 14:36

fhueske reviewed May 15, 2026

View reviewed changes

raminqaf and others added 2 commits May 16, 2026 09:58

[FLINK-39661[docs] Address feedbacks

cfce65f

raminqaf force-pushed the FLINK-39616 branch from ca8d886 to cfce65f Compare May 16, 2026 08:18

featzhang approved these changes May 16, 2026

View reviewed changes

github-actions Bot added the community-reviewed PR has been reviewed by the community. label May 16, 2026

	This pattern is particularly useful for the built-in [`FROM_CHANGELOG`]({{< ref "docs/sql/reference/queries/changelog" >}}#from_changelog) PTF, which converts an append-only CDC stream into a changelog. Combining `PARTITION BY` on the primary key with `ORDER BY` on the event-time column reorders out-of-order CDC events per key before the operations are applied.
	This pattern is particularly useful for the built-in [`FROM_CHANGELOG`]({{< ref "docs/sql/reference/queries/changelog" >}}#from_changelog) PTF, which converts an append-only CDC stream into a changelog. Combining `PARTITION BY` on the primary key with `ORDER BY` on the event-time column reorders out-of-order CDC events per key before the conversion operations are applied.


		CDC streams can deliver events out of order. For example, a key's `UPDATE_AFTER` may arrive before its matching `UPDATE_BEFORE` when events are partitioned across upstream brokers. If the source itself does not guarantee ordering, applying such a changelog directly produces incorrect state.

		`FROM_CHANGELOG` accepts an [ORDER BY clause]({{< ref "docs/dev/table/functions/ptfs" >}}#ordering) that sorts events within each partition before they are processed. The framework buffers events per partition and flushes them to the function in sorted order once the watermark advances. Late events (arriving after the watermark) are dropped.

	WATERMARK FOR event_time AS event_time - INTERVAL '5' SECOND
	WATERMARK FOR event_time AS event_time - INTERVAL '5' MINUTE

	\| `+I[id: 5, op: 'INSERT', event_time: '09:55']` \| `10:00` \| Dropped. Timestamp is below the current watermark. \|
	\| `+I[id: 5, op: 'INSERT', event_time: '09:57']` \| `10:00` \| Dropped. Timestamp is below the current watermark. \|

Conversation

raminqaf commented May 15, 2026

What is the purpose of the change

Does this pull request potentially affect one of the following parts:

Documentation

Was generative AI tooling used to co-author this PR?

Uh oh!

flinkbot commented May 15, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

CI report:

Uh oh!

fhueske left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

featzhang commented May 16, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

flinkbot commented May 15, 2026 •

edited

Loading