[FLINK-13653] ResultStore should avoid using RowTypeInfo when creating a result

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 1.9.1, 1.10.0
• Component/s: Table SQL / Client
• Labels: pull-request-available
• Assignee: Rui Li
• Reporter: Rui Li
• Created: 08/Aug/19 11:27
• Updated: 11/Sep/19 18:35
• Resolved: 11/Sep/19 18:35

Description

Creating a RowTypeInfo from a TableSchema can lose type parameters. As a result, querying a Hive table with decimal column from SQL CLI will hit the following exception:

Caused by: org.apache.flink.table.api.ValidationException: Field types of query result and registered TableSink [default_catalog, default_database, default: select * from foo] do not match.
Query result schema: [x: BigDecimal]
TableSink schema:    [x: BigDecimal]
        at org.apache.flink.table.planner.sinks.TableSinkUtils$.validateSink(TableSinkUtils.scala:69)
        at org.apache.flink.table.planner.delegation.PlannerBase$$anonfun$2.apply(PlannerBase.scala:179)
        at org.apache.flink.table.planner.delegation.PlannerBase$$anonfun$2.apply(PlannerBase.scala:178)
        at scala.Option.map(Option.scala:146)
        at org.apache.flink.table.planner.delegation.PlannerBase.translateToRel(PlannerBase.scala:178)
        at org.apache.flink.table.planner.delegation.PlannerBase$$anonfun$1.apply(PlannerBase.scala:146)
        at org.apache.flink.table.planner.delegation.PlannerBase$$anonfun$1.apply(PlannerBase.scala:146)
        at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
        at scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
        at scala.collection.Iterator$class.foreach(Iterator.scala:891)
        at scala.collection.AbstractIterator.foreach(Iterator.scala:1334)
        at scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
        at scala.collection.AbstractIterable.foreach(Iterable.scala:54)
        at scala.collection.TraversableLike$class.map(TraversableLike.scala:234)
        at scala.collection.AbstractTraversable.map(Traversable.scala:104)
        at org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:146)
        at org.apache.flink.table.api.internal.TableEnvironmentImpl.translate(TableEnvironmentImpl.java:439)
        at org.apache.flink.table.api.internal.TableEnvironmentImpl.insertInto(TableEnvironmentImpl.java:327)
        at org.apache.flink.table.api.internal.TableImpl.insertInto(TableImpl.java:428)
        at org.apache.flink.table.client.gateway.local.LocalExecutor.lambda$executeQueryInternal$10(LocalExecutor.java:477)
        at org.apache.flink.table.client.gateway.local.ExecutionContext.wrapClassLoader(ExecutionContext.java:216)
        at org.apache.flink.table.client.gateway.local.LocalExecutor.executeQueryInternal(LocalExecutor.java:475)
        ... 8 more

Issue Links

• links to: GitHub Pull Request #9432