[
https://issues.apache.org/jira/browse/FLINK-5592?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15834743#comment-15834743
]
Jark Wu commented on FLINK-5592:
--------------------------------
In Table API, please use
{code}
// Java Table API
tEnv.scan("rows").select("person.get('age')")
// Scala Table API
tEnv.scan("rows").select('person.get("age"))
{code}
> Wrong number of RowSerializers with nested Rows in Collection mode
> ------------------------------------------------------------------
>
> Key: FLINK-5592
> URL: https://issues.apache.org/jira/browse/FLINK-5592
> Project: Flink
> Issue Type: Bug
> Reporter: Anton Solovev
> Priority: Minor
>
> {code}
> @Test
> def testNestedRowTypes(): Unit = {
> val env = ExecutionEnvironment.getExecutionEnvironment
> val tEnv = TableEnvironment.getTableEnvironment(env, config)
> tEnv.registerTableSource("rows", new MockSource)
> val table: Table = tEnv.scan("rows")
> val nestedTable: Table = tEnv.scan("rows").select('person)
> val collect: Seq[Row] = nestedTable.collect()
> print(collect)
> }
> class MockSource extends BatchTableSource[Row] {
> import org.apache.flink.api.java.ExecutionEnvironment
> import org.apache.flink.api.java.DataSet
> override def getDataSet(execEnv: ExecutionEnvironment): DataSet[Row] = {
> val data = List(
> Row.of(Row.of("data_1", "dob"), Row.of("info_4", "dub")),
> Row.of(Row.of("data_1", "dob"), Row.of("info_4", "dub")),
> Row.of(Row.of("data_1", "dob"), Row.of("info_4", "dub")))
> execEnv.fromCollection(data.asJava, getReturnType)
> }
> override def getReturnType: TypeInformation[Row] = {
> new RowTypeInfo(
> Array[TypeInformation[_]](
> new RowTypeInfo(
> Array[TypeInformation[_]](BasicTypeInfo.STRING_TYPE_INFO,
> BasicTypeInfo.STRING_TYPE_INFO),
> Array("name", "age"))),
> Array("person")
> )
> }
> }
> {code}
> throws {{java.lang.RuntimeException: Row arity of from does not match
> serializers}}
> stacktrace
> {code}
> at
> org.apache.flink.api.java.typeutils.runtime.RowSerializer.copy(RowSerializer.java:82)
> at
> org.apache.flink.api.java.typeutils.runtime.RowSerializer.copy(RowSerializer.java:36)
> at
> org.apache.flink.api.common.operators.GenericDataSourceBase.executeOnCollections(GenericDataSourceBase.java:234)
> at
> org.apache.flink.api.common.operators.CollectionExecutor.executeDataSource(CollectionExecutor.java:218)
> at
> org.apache.flink.api.common.operators.CollectionExecutor.execute(CollectionExecutor.java:154)
> at
> org.apache.flink.api.common.operators.CollectionExecutor.execute(CollectionExecutor.java:130)
> at
> org.apache.flink.api.common.operators.CollectionExecutor.executeDataSink(CollectionExecutor.java:181)
> at
> org.apache.flink.api.common.operators.CollectionExecutor.execute(CollectionExecutor.java:157)
> at
> org.apache.flink.api.common.operators.CollectionExecutor.execute(CollectionExecutor.java:130)
> at
> org.apache.flink.api.common.operators.CollectionExecutor.execute(CollectionExecutor.java:114)
> at
> org.apache.flink.api.java.CollectionEnvironment.execute(CollectionEnvironment.java:35)
> at
> org.apache.flink.test.util.CollectionTestEnvironment.execute(CollectionTestEnvironment.java:47)
> at
> org.apache.flink.test.util.CollectionTestEnvironment.execute(CollectionTestEnvironment.java:42)
> at
> org.apache.flink.api.scala.ExecutionEnvironment.execute(ExecutionEnvironment.scala:672)
> at org.apache.flink.api.scala.DataSet.collect(DataSet.scala:547)
> {code}
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
