Skip to content

[FLINK-39715][table] Sync SQL and table api behavior for queries with sort #28612

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
snuyanzin wants to merge 2 commits into
base: master
Choose a base branch
from
+385 −3
Open

[FLINK-39715][table] Sync SQL and table api behavior for queries with sort #28612

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
49730e6
[FLINK-39715][table] Sync SQL and table api behavior for queries with…
snuyanzin Jul 2, 2026
285cbc7
Address feedback
snuyanzin Jul 4, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 10 additions & 2 deletions ...k-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkRelBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -242,8 +242,16 @@ public RelBuilder watermark(int rowtimeFieldIndex, RexNode watermarkExpr) {
}

public RelBuilder queryOperation(QueryOperation queryOperation) {
final RelNode relNode = queryOperation.accept(toRelNodeConverter);
return push(relNode);
// Only the outermost call has a root (its top-level ORDER BY is kept); subquery
// re-entries have none, so their fetch-less sorts are dropped. Restore afterwards.
final QueryOperation previousRoot = toRelNodeConverter.getRootOperation();
toRelNodeConverter.setRootOperation(previousRoot == null ? queryOperation : null);
try {
final RelNode relNode = queryOperation.accept(toRelNodeConverter);
return push(relNode);
} finally {
toRelNodeConverter.setRootOperation(previousRoot);
}
}

@Override
Expand Down
29 changes: 28 additions & 1 deletion ...le-planner/src/main/java/org/apache/flink/table/planner/plan/QueryOperationConverter.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -108,6 +108,7 @@
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.core.CorrelationId;
import org.apache.calcite.rel.core.JoinRelType;
import org.apache.calcite.rel.core.Sort;
import org.apache.calcite.rel.logical.LogicalTableFunctionScan;
import org.apache.calcite.rel.logical.LogicalTableScan;
import org.apache.calcite.rel.logical.LogicalValues;
Expand Down Expand Up @@ -151,12 +152,24 @@ public class QueryOperationConverter extends QueryOperationDefaultVisitor<RelNod
private final JoinExpressionVisitor joinExpressionVisitor = new JoinExpressionVisitor();
private final boolean isBatchMode;

// Root of the operation tree currently being converted; used to keep a top-level ORDER BY
// while dropping fetch-less sorts elsewhere (mirrors SqlToRelConverter#removeSortInSubQuery).
private QueryOperation rootOperation;

public QueryOperationConverter(FlinkRelBuilder relBuilder, boolean isBatchMode) {
this.relBuilder = relBuilder;
this.expressionConverter = new ExpressionConverter(relBuilder);
this.isBatchMode = isBatchMode;
}

public void setRootOperation(QueryOperation rootOperation) {
this.rootOperation = rootOperation;
}

public QueryOperation getRootOperation() {
return rootOperation;
}

@Override
public RelNode defaultMethod(QueryOperation other) {
other.getChildren().forEach(child -> relBuilder.push(child.accept(this)));
Expand All @@ -176,6 +189,14 @@ public RelNode visit(ProjectQueryOperation projection) {

@Override
public RelNode visit(AggregateQueryOperation aggregate) {
// Project the input off a sort (as SqlToRelConverter does) so the sort's collation is
// not required on the aggregate output, avoiding an IOOBE in rules like e.g.
// FlinkExpandConversionRule.
final RelNode input = relBuilder.peek();
if (input instanceof Sort) {
relBuilder.project(relBuilder.fields(), input.getRowType().getFieldNames(), true);
}

List<AggCall> aggregations =
aggregate.getAggregateExpressions().stream()
.map(this::getAggCall)
Expand Down Expand Up @@ -291,7 +312,13 @@ public RelNode visit(DistinctQueryOperation distinct) {

@Override
public RelNode visit(SortQueryOperation sort) {
List<RexNode> rexNodes = convertToRexNodes(sort.getOrder());
// A non-root sort with neither FETCH nor a non-zero OFFSET has no observable effect,
// so drop it (mirrors SqlToRelConverter#removeSortInSubQuery; -1 means "unset").
final boolean isRoot = sort == rootOperation;
if (!isRoot && sort.getFetch() < 0 && sort.getOffset() <= 0) {
return relBuilder.build();
}
final List<RexNode> rexNodes = convertToRexNodes(sort.getOrder());
return relBuilder.sortLimit(sort.getOffset(), sort.getFetch(), rexNodes).build();
}

Expand Down
3 changes: 3 additions & 0 deletions ...table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlSemanticTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,9 @@ public List<TableTestProgram> programs() {
QueryOperationTestPrograms.DISTINCT_QUERY_OPERATION,
QueryOperationTestPrograms.JOIN_QUERY_OPERATION,
QueryOperationTestPrograms.ORDER_BY_QUERY_OPERATION,
QueryOperationTestPrograms.ORDER_BY_AGGREGATE_QUERY_OPERATION,
QueryOperationTestPrograms.ORDER_BY_NO_FETCH_AGGREGATE_QUERY_OPERATION,
QueryOperationTestPrograms.AGGREGATE_HAVING_QUERY_OPERATION,
QueryOperationTestPrograms.LIMIT_QUERY_OPERATION,
QueryOperationTestPrograms.WINDOW_AGGREGATE_QUERY_OPERATION,
QueryOperationTestPrograms.UNION_ALL_QUERY_OPERATION,
Expand Down
3 changes: 3 additions & 0 deletions ...-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlSerializationTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,9 @@ public List<TableTestProgram> programs() {
QueryOperationTestPrograms.DISTINCT_QUERY_OPERATION,
QueryOperationTestPrograms.JOIN_QUERY_OPERATION,
QueryOperationTestPrograms.ORDER_BY_QUERY_OPERATION,
QueryOperationTestPrograms.ORDER_BY_AGGREGATE_QUERY_OPERATION,
QueryOperationTestPrograms.ORDER_BY_NO_FETCH_AGGREGATE_QUERY_OPERATION,
QueryOperationTestPrograms.AGGREGATE_HAVING_QUERY_OPERATION,

@Timm0 Timm0 Jul 3, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could we also add these tests to QueryOperationSqlSemanticTest?

QueryOperationTestPrograms.LIMIT_QUERY_OPERATION,
QueryOperationTestPrograms.WINDOW_AGGREGATE_QUERY_OPERATION,
QueryOperationTestPrograms.UNION_ALL_QUERY_OPERATION,
Expand Down
96 changes: 96 additions & 0 deletions ...nk-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationTestPrograms.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -427,6 +427,102 @@ private static Instant dayOfSeconds(int second) {
+ " OFFSET 1 ROWS FETCH NEXT 2 ROWS ONLY")
.build();

static final TableTestProgram ORDER_BY_AGGREGATE_QUERY_OPERATION =
TableTestProgram.of("order-by-aggregate-query-operation", "verifies sql serialization")
.setupTableSource(
SourceTestStep.newBuilder("s")
.addSchema("a bigint", "b string")
.producedValues(
Row.of(1L, "a"), Row.of(2L, "b"), Row.of(3L, "c"))
.build())
.setupTableSink(
SinkTestStep.newBuilder("sink")
.addSchema("a bigint")
.consumedValues(
Row.ofKind(RowKind.INSERT, 1L),
Row.ofKind(RowKind.UPDATE_BEFORE, 1L),
Row.ofKind(RowKind.UPDATE_AFTER, 2L))
.build())
.runTableApi(
t -> t.from("s").orderBy($("b")).fetch(2).select($("a").max()), "sink")
.runSql(
"SELECT `$$T_PROJECT`.`EXPR$0` FROM (\n"
+ " SELECT (MAX(`$$T_AGG`.`a`)) AS `EXPR$0` FROM (\n"
+ " SELECT `$$T_SORT`.`a`, `$$T_SORT`.`b` FROM (\n"
+ " SELECT `$$T_SOURCE`.`a`, `$$T_SOURCE`.`b` FROM "
+ "`default_catalog`.`default_database`.`s` $$T_SOURCE\n"
+ " ) $$T_SORT ORDER BY `$$T_SORT`.`b` ASC OFFSET 0 ROWS"
+ " FETCH NEXT 2 ROWS ONLY\n"
+ " ) $$T_AGG\n"
+ " GROUP BY 1\n"
+ ") $$T_PROJECT")
.build();

static final TableTestProgram ORDER_BY_NO_FETCH_AGGREGATE_QUERY_OPERATION =
TableTestProgram.of(
"order-by-no-fetch-aggregate-query-operation",
"verifies sql serialization")
.setupTableSource(
SourceTestStep.newBuilder("s")
.addSchema("a bigint", "b string")
.producedValues(
Row.of(1L, "a"), Row.of(2L, "b"), Row.of(3L, "c"))
.build())
.setupTableSink(
SinkTestStep.newBuilder("sink")
.addSchema("a bigint")
.consumedValues(
Row.ofKind(RowKind.INSERT, 1L),
Row.ofKind(RowKind.UPDATE_BEFORE, 1L),
Row.ofKind(RowKind.UPDATE_AFTER, 2L),
Row.ofKind(RowKind.UPDATE_BEFORE, 2L),
Row.ofKind(RowKind.UPDATE_AFTER, 3L))
.build())
.runTableApi(t -> t.from("s").orderBy($("b")).select($("a").max()), "sink")
.runSql(
"SELECT `$$T_PROJECT`.`EXPR$0` FROM (\n"
+ " SELECT (MAX(`$$T_AGG`.`a`)) AS `EXPR$0` FROM (\n"
+ " SELECT `$$T_SORT`.`a`, `$$T_SORT`.`b` FROM (\n"
+ " SELECT `$$T_SOURCE`.`a`, `$$T_SOURCE`.`b` FROM "
+ "`default_catalog`.`default_database`.`s` $$T_SOURCE\n"
+ " ) $$T_SORT ORDER BY `$$T_SORT`.`b` ASC\n"
+ " ) $$T_AGG\n"
+ " GROUP BY 1\n"
+ ") $$T_PROJECT")
.build();

static final TableTestProgram AGGREGATE_HAVING_QUERY_OPERATION =
TableTestProgram.of("aggregate-having-query-operation", "verifies sql serialization")
.setupTableSource(
SourceTestStep.newBuilder("s")
.addSchema("a bigint", "b string")
.producedValues(
Row.of(1L, "a"), Row.of(2L, "b"), Row.of(3L, "b"))
.build())
.setupTableSink(
SinkTestStep.newBuilder("sink")
.addSchema("b string", "s bigint")
.consumedValues(Row.ofKind(RowKind.UPDATE_AFTER, "b", 5L))
.build())
.runTableApi(
t ->
t.from("s")
.groupBy($("b"))
.select($("b"), $("a").sum().as("s"))
.where($("s").isGreater(3L)),
"sink")
.runSql(
"SELECT `$$T_FILTER`.`b`, `$$T_FILTER`.`s` FROM (\n"
+ " SELECT `$$T_PROJECT`.`b`, `$$T_PROJECT`.`EXPR$0` AS `s` FROM (\n"
+ " SELECT `$$T_AGG`.`b`, (SUM(`$$T_AGG`.`a`)) AS `EXPR$0` FROM (\n"
+ " SELECT `$$T_SOURCE`.`a`, `$$T_SOURCE`.`b` FROM "
+ "`default_catalog`.`default_database`.`s` $$T_SOURCE\n"
+ " ) $$T_AGG\n"
+ " GROUP BY `$$T_AGG`.`b`\n"
+ " ) $$T_PROJECT\n"
+ ") $$T_FILTER WHERE `$$T_FILTER`.`s` > CAST(3 AS BIGINT)")
.build();

static final TableTestProgram LIMIT_QUERY_OPERATION =
TableTestProgram.of("limit-query-operation", "verifies sql serialization")
.setupTableSource(
Expand Down
72 changes: 72 additions & 0 deletions ...ble-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/SortTest.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -71,6 +71,78 @@ SortLimit(orderBy=[a DESC], offset=[0], fetch=[200], global=[true])
+- Exchange(distribution=[single])
+- SortLimit(orderBy=[a DESC], offset=[0], fetch=[200], global=[false])
+- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c])
]]>
</Resource>
</TestCase>
<TestCase name="testOrderByExprWithFetchThenGlobalAggregate">
<Resource name="sql">
<![CDATA[SELECT MAX(a) FROM (SELECT a, b FROM MyTable2 ORDER BY a + 1 LIMIT 10)]]>
</Resource>
<Resource name="ast">
<![CDATA[
LogicalAggregate(group=[{}], EXPR$0=[MAX($0)])
+- LogicalProject(a=[$0])
+- LogicalProject(a=[$0], b=[$1])
+- LogicalSort(sort0=[$2], dir0=[ASC-nulls-first], fetch=[10])
+- LogicalProject(a=[$0], b=[$1], EXPR$2=[+($0, 1)])
+- LogicalTableScan(table=[[default_catalog, default_database, MyTable2]])
]]>
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
HashAggregate(isMerge=[false], select=[MAX(a) AS EXPR$0])
+- Calc(select=[a])
+- SortLimit(orderBy=[EXPR$2 ASC], offset=[0], fetch=[10], global=[true])
+- Exchange(distribution=[single])
+- SortLimit(orderBy=[EXPR$2 ASC], offset=[0], fetch=[10], global=[false])
+- Calc(select=[a, b, +(a, 1) AS EXPR$2])
+- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a, b])
]]>
</Resource>
</TestCase>
<TestCase name="testOrderByThenGlobalAggregate">
<Resource name="sql">
<![CDATA[SELECT MAX(a) FROM (SELECT a, b FROM MyTable2 ORDER BY b)]]>
</Resource>
<Resource name="ast">
<![CDATA[
LogicalAggregate(group=[{}], EXPR$0=[MAX($0)])
+- LogicalProject(a=[$0])
+- LogicalProject(a=[$0], b=[$1])
+- LogicalTableScan(table=[[default_catalog, default_database, MyTable2]])
]]>
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
HashAggregate(isMerge=[true], select=[Final_MAX(max$0) AS EXPR$0])
+- Exchange(distribution=[single])
+- LocalHashAggregate(select=[Partial_MAX(a) AS max$0])
+- Calc(select=[a])
+- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a, b])
]]>
</Resource>
</TestCase>
<TestCase name="testOrderByWithFetchThenGlobalAggregate">
<Resource name="sql">
<![CDATA[SELECT MAX(a) FROM (SELECT a, b FROM MyTable2 ORDER BY b LIMIT 10)]]>
</Resource>
<Resource name="ast">
<![CDATA[
LogicalAggregate(group=[{}], EXPR$0=[MAX($0)])
+- LogicalProject(a=[$0])
+- LogicalSort(sort0=[$1], dir0=[ASC-nulls-first], fetch=[10])
+- LogicalProject(a=[$0], b=[$1])
+- LogicalTableScan(table=[[default_catalog, default_database, MyTable2]])
]]>
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
HashAggregate(isMerge=[false], select=[MAX(a) AS EXPR$0])
+- Calc(select=[a])
+- SortLimit(orderBy=[b ASC], offset=[0], fetch=[10], global=[true])
+- Exchange(distribution=[single])
+- SortLimit(orderBy=[b ASC], offset=[0], fetch=[10], global=[false])
+- TableSourceScan(table=[[default_catalog, default_database, MyTable2]], fields=[a, b])
]]>
</Resource>
</TestCase>
Expand Down
80 changes: 80 additions & 0 deletions ...nner/src/test/resources/org/apache/flink/table/planner/plan/batch/table/AggregateTest.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,6 +91,86 @@ Calc(select=[CAST(1 AS INTEGER) AS a, EXPR$0, EXPR$1, EXPR$2])
+- LocalHashAggregate(groupBy=[a], select=[a, Partial_AVG(a) AS (sum$0, count$1), Partial_SUM(b) AS sum$2, Partial_COUNT(c) AS count$3])
+- Calc(select=[a, b, c], where=[(a = 1)])
+- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b, c])
]]>
</Resource>
</TestCase>
<TestCase name="testGroupAggregateWithHaving">
<Resource name="ast">
<![CDATA[
LogicalFilter(condition=[>($1, 3)])
+- LogicalProject(b=[$0], s=[$1])
+- LogicalAggregate(group=[{1}], EXPR$0=[SUM($0)])
+- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
]]>
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
Calc(select=[b, EXPR$0], where=[>(EXPR$0, 3)])
+- HashAggregate(isMerge=[true], groupBy=[b], select=[b, Final_SUM(sum$0) AS EXPR$0])
+- Exchange(distribution=[hash[b]])
+- LocalHashAggregate(groupBy=[b], select=[b, Partial_SUM(a) AS sum$0])
+- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b])
]]>
</Resource>
</TestCase>
<TestCase name="testOrderByExprWithFetchThenGlobalAggregate">
<Resource name="ast">
<![CDATA[
LogicalProject(EXPR$0=[$0])
+- LogicalAggregate(group=[{}], EXPR$0=[MAX($0)])
+- LogicalProject(a=[$0])
+- LogicalSort(sort0=[$2], dir0=[ASC], fetch=[10])
+- LogicalProject(a=[$0], b=[$1], $f2=[+($0, 1)])
+- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
]]>
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
HashAggregate(isMerge=[false], select=[MAX(a) AS EXPR$0])
+- Calc(select=[a])
+- SortLimit(orderBy=[$f2 ASC], offset=[0], fetch=[10], global=[true])
+- Exchange(distribution=[single])
+- SortLimit(orderBy=[$f2 ASC], offset=[0], fetch=[10], global=[false])
+- Calc(select=[a, b, +(a, 1) AS $f2])
+- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b])
]]>
</Resource>
</TestCase>
<TestCase name="testOrderByThenGlobalAggregate">
<Resource name="ast">
<![CDATA[
LogicalProject(EXPR$0=[$0])
+- LogicalAggregate(group=[{}], EXPR$0=[MAX($0)])
+- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
]]>
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
HashAggregate(isMerge=[true], select=[Final_MAX(max$0) AS EXPR$0])
+- Exchange(distribution=[single])
+- LocalHashAggregate(select=[Partial_MAX(a) AS max$0])
+- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b])
]]>
</Resource>
</TestCase>
<TestCase name="testOrderByWithFetchThenGlobalAggregate">
<Resource name="ast">
<![CDATA[
LogicalProject(EXPR$0=[$0])
+- LogicalAggregate(group=[{}], EXPR$0=[MAX($0)])
+- LogicalProject(a=[$0])
+- LogicalSort(sort0=[$1], dir0=[ASC], fetch=[10])
+- LogicalTableScan(table=[[default_catalog, default_database, MyTable]])
]]>
</Resource>
<Resource name="optimized rel plan">
<![CDATA[
HashAggregate(isMerge=[false], select=[MAX(a) AS EXPR$0])
+- Calc(select=[a])
+- SortLimit(orderBy=[b ASC], offset=[0], fetch=[10], global=[true])
+- Exchange(distribution=[single])
+- SortLimit(orderBy=[b ASC], offset=[0], fetch=[10], global=[false])
+- TableSourceScan(table=[[default_catalog, default_database, MyTable]], fields=[a, b])
]]>
</Resource>
</TestCase>
Expand Down
Loading