wuchong commented on a change in pull request #10495:
[FLINK-15151][table-planner-blink] Use new type system in TableSource…
URL: https://github.com/apache/flink/pull/10495#discussion_r355963649
##########
File path:
flink-table/flink-table-planner-blink/src/main/scala/org/apache/flink/table/planner/sources/TableSourceUtil.scala
##########
@@ -371,30 +370,35 @@ object TableSourceUtil {
*
* @param fieldName The logical field to look up.
* @param tableSource The table source in which to look for the field.
- * @return The name, index, and type information of the physical field.
+ * @return The name, index, and logical type of the physical field.
*/
private def resolveInputField(
fieldName: String,
- tableSource: TableSource[_]): (String, Int, TypeInformation[_]) = {
+ tableSource: TableSource[_]): (String, Int, LogicalType) = {
- val returnType = fromDataTypeToTypeInfo(tableSource.getProducedDataType)
+ val returnType =
+ LogicalTypeDataTypeConverter.fromDataTypeToLogicalType(
+ tableSource.getProducedDataType)
/** Look up a field by name in a type information */
- def lookupField(fieldName: String, failMsg: String): (String, Int,
TypeInformation[_]) = {
- returnType match {
+ def lookupField(fieldName: String, failMsg: String): (String, Int,
LogicalType) = {
+
+ returnType.getTypeRoot match {
+ case LogicalTypeRoot.ROW =>
+ // composite type would convert to RowType in blink planner
+ val dt = returnType.asInstanceOf[RowType]
Review comment:
Can be simplify to
```scala
returnType match {
case dt: RowType =>
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
With regards,
Apache Git Services
