You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ja...@apache.org on 2022/08/06 12:47:15 UTC
[flink] branch master updated (01d31f6ed30 -> 8ac7e9724a6)
This is an automated email from the ASF dual-hosted git repository.
jark pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from 01d31f6ed30 [FLINK-27217][hive] Improve default partition test cases
new 3e1018e0cbc [FLINK-27619][sql] Support NTILE Function
new 8ac7e9724a6 [FLINK-27619][sql] fix use wrong constant for over window.
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
docs/data/sql_functions.yml | 5 ++
docs/data/sql_functions_zh.yml | 5 ++
.../planner/expressions/ExpressionBuilder.java | 5 ++
...eDistAggFunction.java => NTILEAggFunction.java} | 54 +++++++++++----
.../functions/sql/FlinkSqlOperatorTable.java | 1 +
.../exec/batch/BatchExecOverAggregateBase.java | 5 +-
.../batch/BatchPhysicalOverAggregateRule.scala | 80 +++++++++++++++++++---
.../planner/plan/utils/AggFunctionFactory.scala | 11 +++
.../plan/batch/sql/agg/OverAggregateTest.xml | 29 ++++++++
.../plan/batch/sql/agg/OverAggregateTest.scala | 12 ++++
.../runtime/batch/sql/OverAggregateITCase.scala | 48 +++++++++++++
.../operators/over/frame/InsensitiveOverFrame.java | 2 +-
12 files changed, 231 insertions(+), 26 deletions(-)
copy flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/{CumeDistAggFunction.java => NTILEAggFunction.java} (53%)
[flink] 01/02: [FLINK-27619][sql] Support NTILE Function
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 3e1018e0cbc6d28d59f2d0658c6a94b044484f67
Author: luoyuxia <lu...@alumni.sjtu.edu.cn>
AuthorDate: Tue Jun 28 20:58:06 2022 +0800
[FLINK-27619][sql] Support NTILE Function
---
docs/data/sql_functions.yml | 5 +
docs/data/sql_functions_zh.yml | 5 +
.../planner/expressions/ExpressionBuilder.java | 5 +
.../functions/aggfunctions/NTILEAggFunction.java | 112 +++++++++++++++++++++
.../functions/sql/FlinkSqlOperatorTable.java | 1 +
.../exec/batch/BatchExecOverAggregateBase.java | 5 +-
.../planner/plan/utils/AggFunctionFactory.scala | 11 ++
.../runtime/batch/sql/OverAggregateITCase.scala | 48 +++++++++
.../operators/over/frame/InsensitiveOverFrame.java | 2 +-
9 files changed, 192 insertions(+), 2 deletions(-)
diff --git a/docs/data/sql_functions.yml b/docs/data/sql_functions.yml
index 646beaae6f2..2a7ba09c844 100644
--- a/docs/data/sql_functions.yml
+++ b/docs/data/sql_functions.yml
@@ -1009,6 +1009,11 @@ aggregate:
description: Return the cumulative distribution of a value in a group of values. The result is the number of rows preceding or equal to the current row in the ordering of the partition divided by the number of rows in the window partition.
- sql: PERCENT_RANK()
description: Return the percentage ranking of a value in a group of values. The result is the rank value minus one, divided by the number of rows in the parition minus one. If the partition only contains one row, the function will return 0.
+ - sql: NTILE(n)
+ description: |
+ Divides the rows for each window partition into `n` buckets ranging from 1 to at most `n`.
+ If the number of rows in the window partition doesn't divide evenly into the number of buckets, then the remainder values are distributed one per bucket, starting with the first bucket.
+ For example, with 6 rows and 4 buckets, the bucket values would be as follows: 1 1 2 2 3 4
catalog:
- sql: CURRENT_DATABASE()
diff --git a/docs/data/sql_functions_zh.yml b/docs/data/sql_functions_zh.yml
index e43efd90282..ea47e7e0fe0 100644
--- a/docs/data/sql_functions_zh.yml
+++ b/docs/data/sql_functions_zh.yml
@@ -1100,6 +1100,11 @@ aggregate:
description: 返回值在一组值的累积分布。结果是小于或等于当前行的值的行数除以窗口分区的总行数。
- sql: PERCENT_RANK()
description: 返回值在一组值的百分比排名。结果是当前行在窗口分区中的排名减 1，然后除以窗口分区的总行数减 1。如果窗口分区的总行数为 1，则该函数返回 0。
+ - sql: NTILE(n)
+ description: |
+ 将窗口分区中的所有数据按照顺序划分为 n 个分组，返回分配给各行数据的分组编号（从 1 开始）。
+ 如果不能均匀划分为 n 个分组，则从第 1 个分组开始，为每一分组分配一个剩余值。
+ 比如某个窗口分区有 6 行数据，划分为 4 个分组，则各行的分组编号为：1，1，2，2，3，4。
catalog:
- sql: CURRENT_DATABASE()
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/ExpressionBuilder.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/ExpressionBuilder.java
index 867ce097d66..213b8b51cef 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/ExpressionBuilder.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/expressions/ExpressionBuilder.java
@@ -40,6 +40,7 @@ import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.GREATE
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.IF;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.IS_NULL;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.LESS_THAN;
+import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.LESS_THAN_OR_EQUAL;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.MINUS;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.MOD;
import static org.apache.flink.table.functions.BuiltInFunctionDefinitions.NOT;
@@ -143,6 +144,10 @@ public class ExpressionBuilder {
return call(EQUALS, input1, input2);
}
+ public static UnresolvedCallExpression lessThanOrEqual(Expression input1, Expression input2) {
+ return call(LESS_THAN_OR_EQUAL, input1, input2);
+ }
+
public static UnresolvedCallExpression lessThan(Expression input1, Expression input2) {
return call(LESS_THAN, input1, input2);
}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/NTILEAggFunction.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/NTILEAggFunction.java
new file mode 100644
index 00000000000..e133c0fc675
--- /dev/null
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/aggfunctions/NTILEAggFunction.java
@@ -0,0 +1,112 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.table.planner.functions.aggfunctions;
+
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.TableException;
+import org.apache.flink.table.expressions.Expression;
+import org.apache.flink.table.expressions.UnresolvedReferenceExpression;
+import org.apache.flink.table.types.DataType;
+
+import static org.apache.flink.table.expressions.ApiExpressionUtils.unresolvedRef;
+import static org.apache.flink.table.expressions.ApiExpressionUtils.valueLiteral;
+import static org.apache.flink.table.planner.expressions.ExpressionBuilder.div;
+import static org.apache.flink.table.planner.expressions.ExpressionBuilder.greaterThan;
+import static org.apache.flink.table.planner.expressions.ExpressionBuilder.ifThenElse;
+import static org.apache.flink.table.planner.expressions.ExpressionBuilder.lessThanOrEqual;
+import static org.apache.flink.table.planner.expressions.ExpressionBuilder.literal;
+import static org.apache.flink.table.planner.expressions.ExpressionBuilder.mod;
+import static org.apache.flink.table.planner.expressions.ExpressionBuilder.plus;
+
+/** built-in NTILE agg function. */
+public class NTILEAggFunction extends DeclarativeAggregateFunction
+ implements SizeBasedWindowFunction {
+
+ private final Expression bucketSize;
+ private final Expression bucketsWithPadding;
+ private final UnresolvedReferenceExpression rowNumber = unresolvedRef("rowNumber");
+ private final UnresolvedReferenceExpression bucket = unresolvedRef("bucket");
+ private final UnresolvedReferenceExpression bucketThreshold = unresolvedRef("bucketThreshold");
+
+ public NTILEAggFunction() {
+ this.bucketSize = div(windowSizeAttribute(), operand(0));
+ this.bucketsWithPadding = mod(windowSizeAttribute(), operand(0));
+ }
+
+ @Override
+ public int operandCount() {
+ return 1;
+ }
+
+ @Override
+ public UnresolvedReferenceExpression[] aggBufferAttributes() {
+ return new UnresolvedReferenceExpression[] {rowNumber, bucket, bucketThreshold};
+ }
+
+ @Override
+ public DataType[] getAggBufferTypes() {
+ return new DataType[] {DataTypes.BIGINT(), DataTypes.BIGINT(), DataTypes.BIGINT()};
+ }
+
+ @Override
+ public DataType getResultType() {
+ return DataTypes.BIGINT();
+ }
+
+ @Override
+ public Expression[] initialValuesExpressions() {
+ return new Expression[] {valueLiteral(0L), valueLiteral(0L), valueLiteral(0L)};
+ }
+
+ @Override
+ public Expression[] accumulateExpressions() {
+ return new Expression[] {
+ /* rowNumber = */ plus(rowNumber, literal(1L)),
+ /* bucket = */ plus(bucket, bucketOverflowThen(literal(1L))),
+ /* bucketThreshold = */ plus(
+ bucketThreshold,
+ bucketOverflowThen(
+ plus(
+ bucketSize,
+ ifThenElse(
+ lessThanOrEqual(bucket, bucketsWithPadding),
+ valueLiteral(1),
+ valueLiteral(0)))))
+ };
+ }
+
+ private Expression bucketOverflowThen(Expression e) {
+ return ifThenElse(greaterThan(rowNumber, bucketThreshold), e, valueLiteral(0L));
+ }
+
+ @Override
+ public Expression[] retractExpressions() {
+ throw new TableException("This function does not support retraction.");
+ }
+
+ @Override
+ public Expression[] mergeExpressions() {
+ throw new TableException("This function does not support merge.");
+ }
+
+ @Override
+ public Expression getValueExpression() {
+ return bucket;
+ }
+}
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/FlinkSqlOperatorTable.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/FlinkSqlOperatorTable.java
index 0161b3c2140..cfdf34e2e79 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/FlinkSqlOperatorTable.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/functions/sql/FlinkSqlOperatorTable.java
@@ -1140,6 +1140,7 @@ public class FlinkSqlOperatorTable extends ReflectiveSqlOperatorTable {
public static final SqlAggFunction ROW_NUMBER = SqlStdOperatorTable.ROW_NUMBER;
public static final SqlAggFunction CUME_DIST = SqlStdOperatorTable.CUME_DIST;
public static final SqlAggFunction PERCENT_RANK = SqlStdOperatorTable.PERCENT_RANK;
+ public static final SqlAggFunction NTILE = SqlStdOperatorTable.NTILE;
public static final SqlAggFunction LEAD = SqlStdOperatorTable.LEAD;
public static final SqlAggFunction LAG = SqlStdOperatorTable.LAG;
diff --git a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecOverAggregateBase.java b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecOverAggregateBase.java
index 437ab6a80d9..b639980610f 100644
--- a/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecOverAggregateBase.java
+++ b/flink-table/flink-table-planner/src/main/java/org/apache/flink/table/planner/plan/nodes/exec/batch/BatchExecOverAggregateBase.java
@@ -115,6 +115,9 @@ public abstract class BatchExecOverAggregateBase extends ExecNodeBase<RowData>
/** Infer the over window mode based on given group info. */
protected OverWindowMode inferGroupMode(GroupSpec group) {
AggregateCall aggCall = group.getAggCalls().get(0);
+ if (aggCall.getAggregation().getKind() == NTILE) {
+ return OverWindowMode.INSENSITIVE;
+ }
if (aggCall.getAggregation().allowsFraming()) {
if (group.isRows()) {
return OverWindowMode.ROW;
@@ -147,7 +150,7 @@ public abstract class BatchExecOverAggregateBase extends ExecNodeBase<RowData>
OFFSET,
/**
* The INSENSITIVE mode does not care the window framing without LEAD/LAG agg function, for
- * example RANK/DENSE_RANK/PERCENT_RANK/ROW_NUMBER.
+ * example RANK/DENSE_RANK/PERCENT_RANK/ROW_NUMBER/NTILE.
*/
INSENSITIVE
}
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/AggFunctionFactory.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/AggFunctionFactory.scala
index 1def38cec15..a9599ce5d0b 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/AggFunctionFactory.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/utils/AggFunctionFactory.scala
@@ -112,6 +112,13 @@ class AggFunctionFactory(
throw new TableException("PERCENT_RANK Function is not supported in stream mode.")
}
+ case _: SqlNtileAggFunction =>
+ if (isBounded) {
+ createNTILEAggFUnction(argTypes)
+ } else {
+ throw new TableException("NTILE Function is not supported in stream mode.")
+ }
+
case func: SqlLeadLagAggFunction =>
if (isBounded) {
createBatchLeadLagAggFunction(argTypes, index)
@@ -535,6 +542,10 @@ class AggFunctionFactory(
new PercentRankAggFunction(argTypes)
}
+ private def createNTILEAggFUnction(argTypes: Array[LogicalType]): UserDefinedFunction = {
+ new NTILEAggFunction
+ }
+
private def createFirstValueAggFunction(
argTypes: Array[LogicalType],
index: Int): UserDefinedFunction = {
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/OverAggregateITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/OverAggregateITCase.scala
index 748f0a82379..979fa83e8e3 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/OverAggregateITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/batch/sql/OverAggregateITCase.scala
@@ -2855,6 +2855,54 @@ class OverAggregateITCase extends BatchTestBase {
)
)
}
+
+ @Test
+ def testNTILE(): Unit = {
+ checkResult(
+ "SELECT f, NTILE(4) over (order by e)," +
+ " NTILE(3) over (partition by d order by e desc) FROM Table5",
+ Seq(
+ row(0, 1, 1),
+ row(1, 1, 2),
+ row(2, 1, 1),
+ row(3, 1, 3),
+ row(4, 2, 2),
+ row(5, 2, 1),
+ row(6, 2, 3),
+ row(7, 2, 2),
+ row(8, 3, 1),
+ row(9, 3, 1),
+ row(10, 3, 3),
+ row(11, 3, 2),
+ row(12, 4, 2),
+ row(13, 4, 1),
+ row(14, 4, 1)
+ )
+ )
+
+ // test values of order-key containing duplicates
+ checkResult(
+ "SELECT d, NTILE(4) over (order by d)" +
+ " FROM Table5",
+ Seq(
+ row(5, 3),
+ row(5, 3),
+ row(5, 4),
+ row(5, 4),
+ row(5, 4),
+ row(4, 2),
+ row(4, 2),
+ row(4, 3),
+ row(4, 3),
+ row(3, 1),
+ row(3, 2),
+ row(3, 2),
+ row(2, 1),
+ row(2, 1),
+ row(1, 1)
+ )
+ )
+ }
}
/** The initial accumulator for count aggregate function */
diff --git a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/over/frame/InsensitiveOverFrame.java b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/over/frame/InsensitiveOverFrame.java
index 0f4383e9b78..2c8ba39ac14 100644
--- a/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/over/frame/InsensitiveOverFrame.java
+++ b/flink-table/flink-table-runtime/src/main/java/org/apache/flink/table/runtime/operators/over/frame/InsensitiveOverFrame.java
@@ -27,7 +27,7 @@ import org.apache.flink.table.runtime.util.ResettableExternalBuffer;
/**
* The insensitive window frame calculates the statements which shouldn't care the window frame, for
- * example RANK/DENSE_RANK/PERCENT_RANK/CUME_DIST/ROW_NUMBER.
+ * example RANK/DENSE_RANK/PERCENT_RANK/NTILE/ROW_NUMBER.
*/
public class InsensitiveOverFrame implements OverWindowFrame {
[flink] 02/02: [FLINK-27619][sql] fix use wrong constant for over window.
Posted by ja...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jark pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 8ac7e9724a60d9f90ede81b8e1bb13b116623ff3
Author: luoyuxia <lu...@alumni.sjtu.edu.cn>
AuthorDate: Sun Jul 24 12:34:02 2022 +0800
[FLINK-27619][sql] fix use wrong constant for over window.
---
.../batch/BatchPhysicalOverAggregateRule.scala | 80 +++++++++++++++++++---
.../plan/batch/sql/agg/OverAggregateTest.xml | 29 ++++++++
.../plan/batch/sql/agg/OverAggregateTest.scala | 12 ++++
3 files changed, 111 insertions(+), 10 deletions(-)
diff --git a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalOverAggregateRule.scala b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalOverAggregateRule.scala
index 0449fe849bc..6d20ce229ec 100644
--- a/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalOverAggregateRule.scala
+++ b/flink-table/flink-table-planner/src/main/scala/org/apache/flink/table/planner/plan/rules/physical/batch/BatchPhysicalOverAggregateRule.scala
@@ -29,14 +29,18 @@ import org.apache.flink.table.planner.plan.utils.PythonUtil.isPythonAggregate
import org.apache.flink.table.planner.typeutils.RowTypeUtils
import org.apache.flink.table.planner.utils.ShortcutUtils
-import org.apache.calcite.plan.{RelOptCluster, RelOptRule, RelOptRuleCall}
+import org.apache.calcite.plan.{RelOptCluster, RelOptRule, RelOptRuleCall, RelOptUtil}
import org.apache.calcite.plan.RelOptRule._
import org.apache.calcite.rel._
import org.apache.calcite.rel.`type`.RelDataType
import org.apache.calcite.rel.core.{AggregateCall, Window}
import org.apache.calcite.rel.core.Window.Group
+import org.apache.calcite.rex.{RexInputRef, RexNode, RexShuttle}
+import org.apache.calcite.sql.SqlAggFunction
import org.apache.calcite.tools.ValidationException
+import java.util
+
import scala.collection.JavaConversions._
import scala.collection.JavaConverters._
import scala.collection.mutable.ArrayBuffer
@@ -55,15 +59,10 @@ class BatchPhysicalOverAggregateRule
val logicWindow: FlinkLogicalOverAggregate = call.rel(0)
var input: RelNode = call.rel(1)
var inputRowType = logicWindow.getInput.getRowType
+ val originInputSize = inputRowType.getFieldCount
val typeFactory = logicWindow.getCluster.getTypeFactory.asInstanceOf[FlinkTypeFactory]
val constants = logicWindow.constants.asScala
- val constantTypes = constants.map(c => FlinkTypeFactory.toLogicalType(c.getType))
- val inputNamesWithConstants = inputRowType.getFieldNames ++ constants.indices.map(i => s"TMP$i")
- val inputTypesWithConstants = inputRowType.getFieldList
- .map(i => FlinkTypeFactory.toLogicalType(i.getType)) ++ constantTypes
- val inputTypeWithConstants =
- typeFactory.buildRelNodeRowType(inputNamesWithConstants, inputTypesWithConstants)
var overWindowAgg: BatchPhysicalOverAggregateBase = null
@@ -92,12 +91,16 @@ class BatchPhysicalOverAggregateRule
val newInput = RelOptRule.convert(input, requiredTrait)
- val groupToAggCallToAggFunction = groupBuffer.map {
- group =>
+ val groupToAggCallToAggFunction = groupBuffer.zipWithIndex.map {
+ case (_, idx) =>
+ // we may need to adjust the arg index of AggregateCall in the group
+ // for the input's size may change
+ adjustGroup(groupBuffer, idx, originInputSize, newInput.getRowType.getFieldCount)
+ val group = groupBuffer.get(idx)
val aggregateCalls = group.getAggregateCalls(logicWindow)
val (_, _, aggregates) = AggregateUtil.transformToBatchAggregateFunctions(
ShortcutUtils.unwrapTypeFactory(input),
- FlinkTypeFactory.toLogicalRowType(inputTypeWithConstants),
+ FlinkTypeFactory.toLogicalRowType(generateInputTypeWithConstants()),
aggregateCalls,
sortSpec.getFieldIndices)
val aggCallToAggFunction = aggregateCalls.zip(aggregates)
@@ -152,6 +155,15 @@ class BatchPhysicalOverAggregateRule
inputRowType = outputRowType
}
+ def generateInputTypeWithConstants(): RelDataType = {
+ val constantTypes = constants.map(c => FlinkTypeFactory.toLogicalType(c.getType))
+ val inputNamesWithConstants = inputRowType.getFieldNames ++
+ constants.indices.map(i => s"TMP$i")
+ val inputTypesWithConstants = inputRowType.getFieldList
+ .map(i => FlinkTypeFactory.toLogicalType(i.getType)) ++ constantTypes
+ typeFactory.buildRelNodeRowType(inputNamesWithConstants, inputTypesWithConstants)
+ }
+
logicWindow.groups.foreach {
group =>
validate(group)
@@ -202,6 +214,54 @@ class BatchPhysicalOverAggregateRule
typeFactory.createStructType(inputTypeList ++ aggTypes, inputNameList ++ aggNames)
}
+ private def adjustGroup(
+ groupBuffer: ArrayBuffer[Window.Group],
+ groupIdx: Int,
+ originInputSize: Int,
+ newInputSize: Int): Unit = {
+ val inputSizeDiff = newInputSize - originInputSize
+ if (inputSizeDiff > 0) {
+ // the input's size of this group has increased, adjust the arg index of agg call
+ // in the group to make sure the arg index still refers to the origin value
+ var hasAdjust = false
+ val indexAdjustment = new RexShuttle() {
+ override def visitInputRef(inputRef: RexInputRef): RexNode = {
+ if (inputRef.getIndex >= originInputSize) {
+ hasAdjust = true
+ new RexInputRef(inputRef.getIndex + inputSizeDiff, inputRef.getType)
+ } else {
+ inputRef
+ }
+ }
+ }
+ val group = groupBuffer.get(groupIdx)
+ val newAggCalls = new util.ArrayList[Window.RexWinAggCall]()
+ group.aggCalls.forEach(
+ aggCall => {
+ val newOperands = indexAdjustment.visitList(aggCall.operands);
+ newAggCalls.add(
+ new Window.RexWinAggCall(
+ aggCall.getOperator.asInstanceOf[SqlAggFunction],
+ aggCall.getType,
+ newOperands,
+ aggCall.ordinal,
+ aggCall.distinct,
+ aggCall.ignoreNulls))
+ })
+ if (hasAdjust) {
+ groupBuffer.set(
+ groupIdx,
+ new Group(
+ group.keys,
+ group.isRows,
+ group.lowerBound,
+ group.upperBound,
+ group.orderKeys,
+ newAggCalls))
+ }
+ }
+ }
+
// SPARK/PostgreSQL don't support distinct on over(), and Hive only support distinct without
// window frame. Because it is complicated for Distinct on over().
private def validate(group: Group): Unit = {
diff --git a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/agg/OverAggregateTest.xml b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/agg/OverAggregateTest.xml
index a3c3538a66d..c58b35754ae 100644
--- a/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/agg/OverAggregateTest.xml
+++ b/flink-table/flink-table-planner/src/test/resources/org/apache/flink/table/planner/plan/batch/sql/agg/OverAggregateTest.xml
@@ -418,6 +418,35 @@ Calc(select=[CASE((w0$o0 > 0), w0$o1, null:INTEGER) AS EXPR$0, w0$o2 AS EXPR$1])
]]>
</Resource>
</TestCase>
+ <TestCase name="testOverWindowWithConstants3">
+ <Resource name="sql">
+ <![CDATA[
+SELECT
+ COUNT(2) OVER (ORDER BY a),
+ COUNT(1) OVER (PARTITION BY c ORDER BY a)
+FROM MyTable
+ ]]>
+ </Resource>
+ <Resource name="ast">
+ <![CDATA[
+LogicalProject(EXPR$0=[COUNT(2) OVER (ORDER BY $0 NULLS FIRST)], EXPR$1=[COUNT(1) OVER (PARTITION BY $2 ORDER BY $0 NULLS FIRST)])
++- LogicalTableScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]])
+]]>
+ </Resource>
+ <Resource name="optimized exec plan">
+ <![CDATA[
+Calc(select=[w0$o0 AS $0, w1$o0 AS $1])
++- OverAggregate(partitionBy=[c], orderBy=[a ASC], window#0=[COUNT(1) AS w1$o0 RANG BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], select=[a, c, w0$o0, w1$o0])
+ +- Sort(orderBy=[c ASC, a ASC])
+ +- Exchange(distribution=[hash[c]])
+ +- OverAggregate(orderBy=[a ASC], window#0=[COUNT(2) AS w0$o0 RANG BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], select=[a, c, w0$o0])
+ +- Sort(orderBy=[a ASC])
+ +- Exchange(distribution=[single])
+ +- Calc(select=[a, c])
+ +- LegacyTableSourceScan(table=[[default_catalog, default_database, MyTable, source: [TestTableSource(a, b, c)]]], fields=[a, b, c])
+]]>
+ </Resource>
+ </TestCase>
<TestCase name="testSamePartitionKeysWithDiffOrderKeys1">
<Resource name="sql">
<![CDATA[
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/agg/OverAggregateTest.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/agg/OverAggregateTest.scala
index 35401b41367..d0753274fc0 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/agg/OverAggregateTest.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/plan/batch/sql/agg/OverAggregateTest.scala
@@ -308,6 +308,18 @@ class OverAggregateTest extends TableTestBase {
util.verifyExecPlan(sqlQuery)
}
+ @Test
+ def testOverWindowWithConstants3(): Unit = {
+ val sqlQuery =
+ """
+ |SELECT
+ | COUNT(2) OVER (ORDER BY a),
+ | COUNT(1) OVER (PARTITION BY c ORDER BY a)
+ |FROM MyTable
+ """.stripMargin
+ util.verifyExecPlan(sqlQuery)
+ }
+
@Test(expected = classOf[RuntimeException])
def testDistinct(): Unit = {
val sqlQuery =