[FLINK-40449][table] Reject non-temporal streaming sort during planning by MartijnVisser · Pull Request #29003 · apache/flink

MartijnVisser · 2026-08-21T09:21:21Z

What is the purpose of the change

A streaming query that sorts on a non-time attribute (e.g. SELECT a FROM t ORDER BY c, where c is not an ascending time attribute) is not supported. Today that rejection is thrown from StreamExecSort.translateToPlanInternal, i.e. during execution-plan translation, so it only surfaces when the job is (re)submitted rather than during optimization. As a result COMPILE PLAN FOR '' succeeds and the statement fails only later; in SQL-gateway/REST flows this can lead to a restart loop before the deterministic error reaches the user.

This change rejects the sort during optimization instead, so it is reported by optimize() / COMPILE PLAN before job submission. It also replaces the previous message (which was inaccurate for a descending time attribute, it claimed "non-time-attribute" even when the column was a time attribute) with two targeted messages.

Brief change log

StreamPhysicalSortRule.convert() rejects a non-temporal streaming sort during the physical optimization phase (consistent with e.g. StreamPhysicalOverAggregateRule), reusing SortUtil.getFirstSortField.
Two error messages live in SortUtil and are shared by the rule and the exec node: one for a non-time-attribute primary key, one for a time attribute sorted DESC.
The StreamExecSort.translateToPlanInternal check is kept as a backstop for compiled plans loaded via loadPlan, which bypasses optimize().

Verifying this change

This change added tests and can be verified as follows:

Added SortValidationTest (parameterized over the non-temporal sort shapes, plus a COMPILE PLAN case) asserting the rejection now happens at planning time.
Adapted existing tests whose ORDER BY became invalid in streaming: those that sorted only incidentally still assert their feature (TableSinkTest.testDistribution, GroupingSetsTest.testFromBlogspot); the rest assert the rejection (GroupingSetsTest.testRollupPlusOrderBy/testGroupingInOrderByClause, streaming PartialInsertTest.testPartialInsertWithOrderBy, SortITCase.testDisableSortNonTemporalField).
mvn verify passes for flink-table-planner, including SortITCase end-to-end.

Note: because the exception is thrown from a rule during the Volcano phase, the top-level message is wrapped by FlinkVolcanoProgram ("Sql optimization: …") with the original text as the root cause, that's the same shape as other rule-phase rejections (GROUPING SETS, OVER aggregates).

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 (CompiledPlan JSON schema for StreamExecSort is unchanged)
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

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

Yes (please specify the tool below)

Generated-by: Claude Code (Claude Opus 4.8)

flinkbot · 2026-08-21T09:28:39Z

CI report:

3e939ff Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run azure re-run the last Azure build

fhueske

Thanks for the fix @MartijnVisser!
The changes look good.

Left just two minor comments.

Thanks, Fabian

fhueske · 2026-08-21T10:11:08Z

+   * Error message when the primary streaming sort key is a time attribute but sorted descending.
+   */
+  def sortKeyTimeAttributeMustBeAscendingMessage(column: String): String =
+    s"Streaming ORDER BY on time attribute '$column' must be sorted in ascending order; DESC is " +


Suggested change

s"Streaming ORDER BY on time attribute '$column' must be sorted in ascending order; DESC is " +

s"Streaming ORDER BY on time attribute '$column' must be sorted in ascending order; descending order is " +

Spell out descending as ascending is spelled out as well?

Yes, good one. Fixed!

fhueske · 2026-08-21T10:38:03Z

+ * be rejected during optimization (so it surfaces at {@code COMPILE PLAN} / planning time), not
+ * deferred to execution-plan translation. Valid temporal sorts are covered by {@code SortTest}.
+ */
+class SortValidationTest extends TableTestBase {


add a test asserting that the plan check are only done if the TABLE_EXEC_NON_TEMPORAL_SORT_ENABLED flag isn't set?

Makes sense, I've added that too

snuyanzin · 2026-08-22T16:55:05Z

+ * be rejected during optimization (so it surfaces at {@code COMPILE PLAN} / planning time), not
+ * deferred to execution-plan translation. Valid temporal sorts are covered by {@code SortTest}.
+ */
+class SortValidationTest extends TableTestBase {


How about using new way semantic tests instead of old fashion approach ?

Good question — I looked into migrating this to the semantic test framework (SemanticTestBase / TableTestProgram), but it does not fit this particular test:

runFailingSql only accepts ValidationException or TableRuntimeException (hard Preconditions check in FailingSqlTestStep), while this check throws TableException — the same type the previous execution-time check threw, so I would rather not change the user-facing exception type in this PR.

Three of the assertions here have no equivalent test step: the table.exec.non-temporal-sort.enabled escape hatch (asserting no exception), the COMPILE PLAN check via compilePlanSql, and asserting the message names the column and its type.

Since *ValidationTest classes extending TableTestBase are still the established pattern for planning-time rejections (e.g. the batch SortValidationTest, AggregateValidationTest), I would prefer to keep this as-is. Happy to revisit in a follow-up if we relax the exception-type restriction in FailingSqlTestStep.

snuyanzin · 2026-08-22T16:57:05Z

+    static Stream<Arguments> nonTemporalSorts() {
+        return Stream.of(
+                // primary sort key is not a time attribute -> message A
+                Arguments.of("SELECT a FROM MyTable ORDER BY c", MESSAGE),


Do we have tests with sorting by ordinals?
Probably same for sorting by alias?

Based on FlinkSqlConformance they should be valid

Good point. Ordinals and aliases are expanded by the SQL validator before optimization, so they hit the same rule — but explicit coverage is cheap. I have added negative cases for ORDER BY <ordinal>, ORDER BY <alias>, and ORDER BY <time-attribute alias> DESC, plus a positive case in SortTest showing that an alias of rowtime still routes to the temporal sort.

A streaming sort on a non-time attribute is unsupported, but the check lived in StreamExecSort.translateToPlanInternal, so it only fired at job (re)submission, not during optimization: COMPILE PLAN succeeded and the query failed later. Reject it in StreamPhysicalSortRule.convert() during optimization, with two targeted messages (non-time-attribute key, and descending time attribute) built in SortUtil. The StreamExecSort check is kept as a backstop for compiled plans loaded via loadPlan, which bypasses optimization. Existing tests whose ORDER BY became invalid in streaming are adapted: those that sorted only incidentally keep asserting their feature (TableSinkTest.testDistribution, GroupingSetsTest.testFromBlogspot), the rest assert the rejection, and SortValidationTest covers the new behavior. Generated-by: Claude Code (Claude Opus 4.8)

fhueske reviewed Aug 21, 2026

View reviewed changes

MartijnVisser force-pushed the FLINK-40449-non-temporal-sort-planning branch from 74fb43b to c2d671a Compare August 21, 2026 11:44

MartijnVisser requested a review from fhueske August 21, 2026 14:07

snuyanzin reviewed Aug 22, 2026

View reviewed changes

MartijnVisser force-pushed the FLINK-40449-non-temporal-sort-planning branch from c2d671a to 3e939ff Compare August 24, 2026 08:04

MartijnVisser requested a review from snuyanzin August 24, 2026 08:15

Navigation Menu

Uh oh!

[FLINK-40449][table] Reject non-temporal streaming sort during planning - #29003

[FLINK-40449][table] Reject non-temporal streaming sort during planning#29003
MartijnVisser wants to merge 1 commit into
apache:masterfrom
MartijnVisser:FLINK-40449-non-temporal-sort-planning

MartijnVisser commented Aug 21, 2026

Uh oh!

flinkbot commented Aug 21, 2026 •

edited

Loading

Uh oh!

fhueske left a comment •

edited

Loading

Uh oh!

fhueske Aug 21, 2026

Uh oh!

MartijnVisser Aug 21, 2026

Uh oh!

fhueske Aug 21, 2026

Uh oh!

MartijnVisser Aug 21, 2026

Uh oh!

snuyanzin Aug 22, 2026

Uh oh!

MartijnVisser Aug 24, 2026

Uh oh!

snuyanzin Aug 22, 2026

Uh oh!

MartijnVisser Aug 24, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

	s"Streaming ORDER BY on time attribute '$column' must be sorted in ascending order; DESC is " +
	s"Streaming ORDER BY on time attribute '$column' must be sorted in ascending order; descending order is " +

Uh oh!

Conversation

MartijnVisser commented Aug 21, 2026

Uh oh!

flinkbot commented Aug 21, 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 • edited Loading Uh oh! There was an error while loading. Please reload this page.

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!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

flinkbot commented Aug 21, 2026 •

edited

Loading

fhueske left a comment •

edited

Loading