This is an automated email from the ASF dual-hosted git repository.
snuyanzin pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push:
new 20216e41992 [FLINK-39715][table] Sync SQL and table api behavior for
queries with sort
20216e41992 is described below
commit 20216e41992090ca3ee08f129335bde86c9c7c2d
Author: Sergey Nuyanzin <[email protected]>
AuthorDate: Tue Jul 7 22:23:17 2026 +0200
[FLINK-39715][table] Sync SQL and table api behavior for queries with sort
---------
Co-authored-by: Arvind Kandpal <[email protected]>
---
.../table/planner/calcite/FlinkRelBuilder.java | 12 ++-
.../planner/plan/QueryOperationConverter.java | 29 ++++++-
.../table/api/QueryOperationSqlSemanticTest.java | 3 +
.../api/QueryOperationSqlSerializationTest.java | 3 +
.../table/api/QueryOperationTestPrograms.java | 96 ++++++++++++++++++++++
.../table/planner/plan/batch/sql/SortTest.xml | 72 ++++++++++++++++
.../planner/plan/batch/table/AggregateTest.xml | 80 ++++++++++++++++++
.../planner/plan/batch/table/SetOperatorsTest.xml | 25 ++++++
.../table/planner/plan/batch/sql/SortTest.scala | 17 ++++
.../planner/plan/batch/table/AggregateTest.scala | 38 +++++++++
.../plan/batch/table/SetOperatorsTest.scala | 13 +++
11 files changed, 385 insertions(+), 3 deletions(-)
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkRelBuilder.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkRelBuilder.java
index 9045b7d9e38..c68cdfed8f8 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkRelBuilder.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/calcite/FlinkRelBuilder.java
@@ -242,8 +242,16 @@ public final class FlinkRelBuilder extends RelBuilder {
}
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
diff --git
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/QueryOperationConverter.java
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/QueryOperationConverter.java
index 2fb2b5b71fb..41b5ebe632a 100644
---
a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/QueryOperationConverter.java
+++
b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/QueryOperationConverter.java
@@ -108,6 +108,7 @@ import org.apache.calcite.plan.ViewExpanders;
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;
@@ -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)));
@@ -176,6 +189,14 @@ public class QueryOperationConverter extends
QueryOperationDefaultVisitor<RelNod
@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)
@@ -291,7 +312,13 @@ public class QueryOperationConverter extends
QueryOperationDefaultVisitor<RelNod
@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();
}
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlSemanticTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlSemanticTest.java
index 354ae8cd08f..26735396b58 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlSemanticTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlSemanticTest.java
@@ -44,6 +44,9 @@ public class QueryOperationSqlSemanticTest extends
SemanticTestBase {
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,
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlSerializationTest.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlSerializationTest.java
index e9525eae79e..c513c277cd5 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlSerializationTest.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationSqlSerializationTest.java
@@ -59,6 +59,9 @@ public class QueryOperationSqlSerializationTest implements
TableTestProgramRunne
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,
diff --git
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationTestPrograms.java
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationTestPrograms.java
index e8671f0da73..6436d5ef854 100644
---
a/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationTestPrograms.java
+++
b/flink-table/flink-table-planner/src/test/java/org/apache/flink/table/api/QueryOperationTestPrograms.java
@@ -427,6 +427,102 @@ public class QueryOperationTestPrograms {
+ " 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(
diff --git
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/SortTest.xml
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/SortTest.xml
index 1a6ee5d248c..5adc2c8422a 100644
---
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/SortTest.xml
+++
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/SortTest.xml
@@ -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>
diff --git
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/table/AggregateTest.xml
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/table/AggregateTest.xml
index 6f6ce9785d0..0a5581d6540 100644
---
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/table/AggregateTest.xml
+++
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/table/AggregateTest.xml
@@ -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>
diff --git
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/table/SetOperatorsTest.xml
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/table/SetOperatorsTest.xml
index 4fb765038f1..f531cda7fe6 100644
---
a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/table/SetOperatorsTest.xml
+++
b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/table/SetOperatorsTest.xml
@@ -69,6 +69,31 @@ Calc(select=[EXPR$0 AS a, b, EXPR$1 AS c])
: +- TableSourceScan(table=[[default_catalog, default_database,
left]], fields=[a, b, c])
+- Calc(select=[a, b, c], where=[(a > 0)])
+- TableSourceScan(table=[[default_catalog, default_database,
right]], fields=[a, b, c])
+]]>
+ </Resource>
+ </TestCase>
+ <TestCase name="testInSubQueryKeepsOuterOrderBy">
+ <Resource name="ast">
+ <![CDATA[
+LogicalSort(sort0=[$1], dir0=[ASC])
++- LogicalFilter(condition=[IN($0, {
+LogicalProject(a=[$0])
+ LogicalFilter(condition=[>($1, 10)])
+ LogicalTableScan(table=[[default_catalog, default_database, A]])
+})])
+ +- LogicalTableScan(table=[[default_catalog, default_database, A]])
+]]>
+ </Resource>
+ <Resource name="optimized exec plan">
+ <![CDATA[
+Sort(orderBy=[b ASC])
++- Exchange(distribution=[single])
+ +- HashJoin(joinType=[LeftSemiJoin], where=[(a = a0)], select=[a, b, c],
build=[right])
+ :- Exchange(distribution=[hash[a]])
+ : +- TableSourceScan(table=[[default_catalog, default_database, A]],
fields=[a, b, c])(reuse_id=[1])
+ +- Exchange(distribution=[hash[a]])
+ +- Calc(select=[a], where=[(b > 10)])
+ +- Reused(reference_id=[1])
]]>
</Resource>
</TestCase>
diff --git
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/SortTest.scala
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/SortTest.scala
index 518d86882b9..0a68d57c8ae 100644
---
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/SortTest.scala
+++
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/SortTest.scala
@@ -27,6 +27,8 @@ class SortTest extends TableTestBase {
private val util = batchTestUtil()
util.addTableSource[(Int, Long, String)]("MyTable", 'a, 'b, 'c)
+ // 2-column table so the plans below match the Table API AggregateTest
exactly.
+ util.addTableSource[(Int, String)]("MyTable2", 'a, 'b)
@Test
def testNonRangeSortOnSingleFieldWithoutForceLimit(): Unit = {
@@ -64,4 +66,19 @@ class SortTest extends TableTestBase {
// exec node does not support range sort yet, so we verify rel plan here
util.verifyRelPlan("SELECT * FROM MyTable ORDER BY a DESC")
}
+
+ @Test
+ def testOrderByThenGlobalAggregate(): Unit = {
+ util.verifyRelPlan("SELECT MAX(a) FROM (SELECT a, b FROM MyTable2 ORDER BY
b)")
+ }
+
+ @Test
+ def testOrderByWithFetchThenGlobalAggregate(): Unit = {
+ util.verifyRelPlan("SELECT MAX(a) FROM (SELECT a, b FROM MyTable2 ORDER BY
b LIMIT 10)")
+ }
+
+ @Test
+ def testOrderByExprWithFetchThenGlobalAggregate(): Unit = {
+ util.verifyRelPlan("SELECT MAX(a) FROM (SELECT a, b FROM MyTable2 ORDER BY
a + 1 LIMIT 10)")
+ }
}
diff --git
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/table/AggregateTest.scala
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/table/AggregateTest.scala
index 561eb11e1ba..7e02c411025 100644
---
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/table/AggregateTest.scala
+++
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/table/AggregateTest.scala
@@ -73,4 +73,42 @@ class AggregateTest extends TableTestBase {
util.verifyExecPlan(resultTable)
}
+
+ @Test
+ def testOrderByThenGlobalAggregate(): Unit = {
+ val util = batchTestUtil()
+ val sourceTable = util.addTableSource[(Int, String)]("MyTable", 'a, 'b)
+ val resultTable = sourceTable.orderBy('b.asc).select('a.max)
+
+ util.verifyRelPlan(resultTable)
+ }
+
+ @Test
+ def testOrderByWithFetchThenGlobalAggregate(): Unit = {
+ val util = batchTestUtil()
+ val sourceTable = util.addTableSource[(Int, String)]("MyTable", 'a, 'b)
+ val resultTable = sourceTable.orderBy('b.asc).fetch(10).select('a.max)
+
+ util.verifyRelPlan(resultTable)
+ }
+
+ @Test
+ def testOrderByExprWithFetchThenGlobalAggregate(): Unit = {
+ val util = batchTestUtil()
+ val sourceTable = util.addTableSource[(Int, String)]("MyTable", 'a, 'b)
+ val resultTable = sourceTable.orderBy(('a +
1).asc).fetch(10).select('a.max)
+
+ util.verifyRelPlan(resultTable)
+ }
+
+ @Test
+ def testGroupAggregateWithHaving(): Unit = {
+ // Filter on an aggregate result (HAVING semantics).
+ val util = batchTestUtil()
+ val sourceTable = util.addTableSource[(Int, String)]("MyTable", 'a, 'b)
+ val resultTable = sourceTable.groupBy('b).select('b,
'a.sum.as("s")).where('s > 3)
+
+ util.verifyRelPlan(resultTable)
+ }
+
}
diff --git
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/table/SetOperatorsTest.scala
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/table/SetOperatorsTest.scala
index 1ec48783dc4..b1d8f8d57c2 100644
---
a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/table/SetOperatorsTest.scala
+++
b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/table/SetOperatorsTest.scala
@@ -133,4 +133,17 @@ class SetOperatorsTest extends TableTestBase {
util.verifyExecPlan(result)
}
+
+ @Test
+ def testInSubQueryKeepsOuterOrderBy(): Unit = {
+ val util = batchTestUtil()
+ val t = util.addTableSource[(Int, Long, String)]("A", 'a, 'b, 'c)
+
+ // The IN subquery re-enters QueryOperationConverter, so the outer ORDER
BY (the actual
+ // root) must not be dropped as if it were a fetch-less subquery sort. See
FLINK-39715.
+ val elements = t.where('b > 10L).select('a)
+ val result = t.where('a.in(elements)).orderBy('b)
+
+ util.verifyExecPlan(result)
+ }
}
