(DEPRECATED) Apache Flink User Mailing List archive.

question about concating an array in flink sql

Classic

List

Threaded

2 messages Options

vtygoss

question about concating an array in flink sql

Hi,

I have below use case

I want concat an array<string> using comma separator, but got exception “Cannot apply 'CONCAT_WS' to arguments of type 'CONCAT_WS(<CHAR(1)>, <VARCHAR(2147483647) ARRAY>)'. Supported form(s): 'CONCAT_WS(<STRING>)’”。

How to concat an array in flink sql? please help to offer some advices.

Regards

```

[test case code]

val senv = StreamExecutionEnvironment.getExecutionEnvironment
val tenv = StreamTableEnvironment.create(senv)
import org.apache.flink.api.scala._
val dStream = senv.fromCollection(Array((1,Array("1a","1b","1c")),(2,Array("2d","2e","2f"))))
import org.apache.flink.table.api._
val table = tenv.fromDataStream(dStream, $("id"),$("names"))
tenv.createTemporaryView("test",table)
tenv.executeSql("select id, concat_ws(',',names) from test").print()

```

Exception in thread "main" org.apache.flink.table.api.ValidationException: SQL validation failed. From line 1, column 12 to line 1, column 31: Cannot apply 'CONCAT_WS' to arguments of type 'CONCAT_WS(<CHAR(1)>, <VARCHAR(2147483647) ARRAY>)'. Supported form(s): 'CONCAT_WS(<STRING>)'

at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org$apache$flink$table$planner$calcite$FlinkPlannerImpl$$validate(FlinkPlannerImpl.scala:152)

at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.validate(FlinkPlannerImpl.scala:111)

at org.apache.flink.table.planner.operations.SqlToOperationConverter.convert(SqlToOperationConverter.java:193)

at org.apache.flink.table.planner.delegation.ParserImpl.parse(ParserImpl.java:78)

at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeSql(TableEnvironmentImpl.java:659)

at com.test.TestPk$.main(TestPk.scala:25)

at com.test.TestPk.main(TestPk.scala)

Caused by: org.apache.calcite.runtime.CalciteContextException: From line 1, column 12 to line 1, column 31: Cannot apply 'CONCAT_WS' to arguments of type 'CONCAT_WS(<CHAR(1)>, <VARCHAR(2147483647) ARRAY>)'. Supported form(s): 'CONCAT_WS(<STRING>)'

at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)

at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)

at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)

at java.lang.reflect.Constructor.newInstance(Constructor.java:423)

at org.apache.calcite.runtime.Resources$ExInstWithCause.ex(Resources.java:467)

at org.apache.calcite.sql.SqlUtil.newContextException(SqlUtil.java:883)

at org.apache.calcite.sql.SqlUtil.newContextException(SqlUtil.java:868)

at org.apache.calcite.sql.validate.SqlValidatorImpl.newValidationError(SqlValidatorImpl.java:5043)

at org.apache.calcite.sql.SqlCallBinding.newValidationSignatureError(SqlCallBinding.java:389)

at org.apache.calcite.sql.type.CompositeOperandTypeChecker.checkOperandTypes(CompositeOperandTypeChecker.java:262)

at org.apache.calcite.sql.SqlOperator.checkOperandTypes(SqlOperator.java:679)

at org.apache.calcite.sql.SqlOperator.validateOperands(SqlOperator.java:444)

at org.apache.calcite.sql.SqlFunction.deriveType(SqlFunction.java:330)

JING ZHANG

Re: question about concating an array in flink sql

Hi vtygoss,

If the length of names is fixed, please try this 'select id, concat_ws(',',names[1],names[2],names[3]) from test', and note begin with 1 instead of 0.

Else maybe you need to define a custom UDF which receives two arguments, first is string as separator, second is a string array as contents. Please check[1] for details.

[1] https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/dev/table/functions/udfs/

Best regards,

JING ZHANG

vtygoss <[hidden email]> 于2021年6月10日周四下午5:25写道：

Hi,

I have below use case

I want concat an array<string> using comma separator, but got exception “Cannot apply 'CONCAT_WS' to arguments of type 'CONCAT_WS(<CHAR(1)>, <VARCHAR(2147483647) ARRAY>)'. Supported form(s): 'CONCAT_WS(<STRING>)’”。

How to concat an array in flink sql? please help to offer some advices.

Regards

```
[test case code]
val senv = StreamExecutionEnvironment.getExecutionEnvironment
val tenv = StreamTableEnvironment.create(senv)
import org.apache.flink.api.scala._
val dStream = senv.fromCollection(Array((1,Array("1a","1b","1c")),(2,Array("2d","2e","2f"))))
import org.apache.flink.table.api._
val table = tenv.fromDataStream(dStream, $("id"),$("names"))
tenv.createTemporaryView("test",table)
tenv.executeSql("select id, concat_ws(',',names) from test").print()
```

Exception in thread "main" org.apache.flink.table.api.ValidationException: SQL validation failed. From line 1, column 12 to line 1, column 31: Cannot apply 'CONCAT_WS' to arguments of type 'CONCAT_WS(<CHAR(1)>, <VARCHAR(2147483647) ARRAY>)'. Supported form(s): 'CONCAT_WS(<STRING>)'
at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org$apache$flink$table$planner$calcite$FlinkPlannerImpl$$validate(FlinkPlannerImpl.scala:152)
at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.validate(FlinkPlannerImpl.scala:111)
at org.apache.flink.table.planner.operations.SqlToOperationConverter.convert(SqlToOperationConverter.java:193)
at org.apache.flink.table.planner.delegation.ParserImpl.parse(ParserImpl.java:78)
at org.apache.flink.table.api.internal.TableEnvironmentImpl.executeSql(TableEnvironmentImpl.java:659)
at com.test.TestPk$.main(TestPk.scala:25)
at com.test.TestPk.main(TestPk.scala)
Caused by: org.apache.calcite.runtime.CalciteContextException: From line 1, column 12 to line 1, column 31: Cannot apply 'CONCAT_WS' to arguments of type 'CONCAT_WS(<CHAR(1)>, <VARCHAR(2147483647) ARRAY>)'. Supported form(s): 'CONCAT_WS(<STRING>)'
at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
at java.lang.reflect.Constructor.newInstance(Constructor.java:423)
at org.apache.calcite.runtime.Resources$ExInstWithCause.ex(Resources.java:467)
at org.apache.calcite.sql.SqlUtil.newContextException(SqlUtil.java:883)
at org.apache.calcite.sql.SqlUtil.newContextException(SqlUtil.java:868)
at org.apache.calcite.sql.validate.SqlValidatorImpl.newValidationError(SqlValidatorImpl.java:5043)
at org.apache.calcite.sql.SqlCallBinding.newValidationSignatureError(SqlCallBinding.java:389)
at org.apache.calcite.sql.type.CompositeOperandTypeChecker.checkOperandTypes(CompositeOperandTypeChecker.java:262)
at org.apache.calcite.sql.SqlOperator.checkOperandTypes(SqlOperator.java:679)
at org.apache.calcite.sql.SqlOperator.validateOperands(SqlOperator.java:444)
at org.apache.calcite.sql.SqlFunction.deriveType(SqlFunction.java:330)