Github user sunjincheng121 commented on a diff in the pull request:
https://github.com/apache/flink/pull/5206#discussion_r158603987
--- Diff:
flink-libraries/flink-table/src/test/scala/org/apache/flink/table/runtime/stream/sql/SqlITCase.scala
---
@@ -481,4 +484,34 @@ class SqlITCase extends StreamingWithStateTestBase {
assertEquals(expected.sorted, MemoryTableSinkUtil.results.sorted)
}
+ @Test
+ def testUDFWithLongVarargs(): Unit = {
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
+ val tEnv = TableEnvironment.getTableEnvironment(env)
+ StreamITCase.clear
+
+ tEnv.registerFunction("varudf", VarUDF)
+
+ val parameters = (0 until 255).map(_ => "c").mkString(",")
+ val sqlQuery = s"SELECT varudf($parameters) FROM T1"
+
+ val t1 =
StreamTestData.getSmall3TupleDataStream(env).toTable(tEnv).as('a, 'b, 'c)
+ tEnv.registerTable("T1", t1)
+
+ val result = tEnv.sqlQuery(sqlQuery).toAppendStream[Row]
+ result.addSink(new StreamITCase.StringSink[Row])
+ env.execute()
+
+ val expected = List(
+ "510",
+ "1275",
+ "2805")
+ assertEquals(expected.sorted, StreamITCase.testResults.sorted)
+ }
+
+}
+
+object VarUDF extends ScalarFunction {
--- End diff --
Suggest using exist test scalarFunction, such as :
`org.apache.flink.table.expressions.utils.userDefinedScalarFunctions #
Func15`
OR move the VarUDF into
`org.apache.flink.table.expressions.utils.userDefinedScalarFunctions`
---
