You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by li...@apache.org on 2024/09/18 03:11:43 UTC
(flink) branch master updated: [FLINK-36283][table-planner] Fix udaf incorrectly accessing the built-in function PERCENTILE's implementation with the same name
This is an automated email from the ASF dual-hosted git repository.
lincoln 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 cfd532d5520 [FLINK-36283][table-planner] Fix udaf incorrectly accessing the built-in function PERCENTILE's implementation with the same name
cfd532d5520 is described below
commit cfd532d552008fb10d158707da9e5538d5938a05
Author: dylanhz <53...@users.noreply.github.com>
AuthorDate: Wed Sep 18 11:11:36 2024 +0800
[FLINK-36283][table-planner] Fix udaf incorrectly accessing the built-in function PERCENTILE's implementation with the same name
This closes #25335
---
.../planner/plan/utils/AggFunctionFactory.scala | 8 ++-----
.../runtime/stream/sql/AggregateITCase.scala | 28 ++++++++++++++++++++++
.../utils/UserDefinedFunctionTestUtils.scala | 17 +++++++++++++
3 files changed, 47 insertions(+), 6 deletions(-)
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 5284aafd395..32ae59c513f 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
@@ -62,10 +62,6 @@ class AggFunctionFactory(
/** The entry point to create an aggregate function from the given [[AggregateCall]]. */
def createAggFunction(call: AggregateCall, index: Int): UserDefinedFunction = {
- object BuiltInAggFunctionName {
- val PERCENTILE: String = BuiltInFunctionDefinitions.PERCENTILE.getName
- }
-
val argTypes: Array[LogicalType] = call.getArgList
.map(inputRowType.getChildren.get(_))
.toArray
@@ -170,9 +166,9 @@ class AggFunctionFactory(
udagg.makeFunction(constants.toArray, argTypes)
case bridge: BridgingSqlAggFunction =>
- bridge.getName match {
+ bridge.getDefinition match {
// built-in imperativeFunction
- case BuiltInAggFunctionName.PERCENTILE =>
+ case BuiltInFunctionDefinitions.PERCENTILE =>
createPercentileAggFunction(argTypes)
// DeclarativeAggregateFunction & UDF
case _ =>
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/AggregateITCase.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/AggregateITCase.scala
index 9b95408eba3..647ca88f9ce 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/AggregateITCase.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/stream/sql/AggregateITCase.scala
@@ -2049,4 +2049,32 @@ class AggregateITCase(aggMode: AggMode, miniBatch: MiniBatchMode, backend: State
}
}
}
+
+ @TestTemplate
+ def testAggFunctionPriority(): Unit = {
+ // reported in FLINK-36283
+ val sql =
+ """
+ |SELECT
+ | c,
+ | PERCENTILE(b, 0.5) AS `swo`
+ |FROM MyTable
+ |GROUP BY c
+ """.stripMargin
+
+ val t = failingDataSource(data).toTable(tEnv, 'a, 'b, 'c)
+ tEnv.createTemporaryView("MyTable", t)
+
+ // create a UDAF to cover built-in agg function with the same name
+ tEnv.createTemporarySystemFunction("PERCENTILE", classOf[FakePercentile])
+
+ val sink = new TestingRetractSink
+ tEnv.sqlQuery(sql).toRetractStream[Row].addSink(sink).setParallelism(1)
+ env.execute()
+
+ val expected = List("Hello,21.0", "Hello World,35.0")
+ assertThat(sink.getRetractResults.sorted).isEqualTo(expected.sorted)
+
+ tEnv.dropTemporarySystemFunction("PERCENTILE")
+ }
}
diff --git a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/utils/UserDefinedFunctionTestUtils.scala b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/utils/UserDefinedFunctionTestUtils.scala
index bac0e9edfad..96e49926ff4 100644
--- a/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/utils/UserDefinedFunctionTestUtils.scala
+++ b/flink-table/flink-table-planner/src/test/scala/org/apache/flink/table/planner/runtime/utils/UserDefinedFunctionTestUtils.scala
@@ -190,6 +190,23 @@ object UserDefinedFunctionTestUtils {
}
+ class FakePercentile extends AggregateFunction[Double, Tuple1[Double]] {
+ def accumulate(acc: Tuple1[Double], value: Int, percentage: Double): Unit = acc.f0 += value
+
+ override def createAccumulator: Tuple1[Double] = Tuple1.of(0d)
+
+ override def getValue(acc: Tuple1[Double]): Double = acc.f0
+
+ override def getTypeInference(typeFactory: DataTypeFactory): TypeInference = {
+ TypeInference.newBuilder
+ .typedArguments(DataTypes.INT(), DataTypes.DOUBLE())
+ .accumulatorTypeStrategy(TypeStrategies.explicit(
+ DataTypes.STRUCTURED(classOf[Tuple1[Double]], DataTypes.FIELD("f0", DataTypes.DOUBLE()))))
+ .outputTypeStrategy(TypeStrategies.explicit(DataTypes.DOUBLE()))
+ .build
+ }
+ }
+
// ------------------------------------------------------------------------------------
// ScalarFunctions
// ------------------------------------------------------------------------------------