Hi XiaChang,
I think this is a bug. Others have encountered similar problems before.
I have create a JIRA for it: https://issues.apache.org/jira/browse/FLINK-18750
<https://issues.apache.org/jira/browse/FLINK-18750>
Best,
Wei
> 在 2020年7月29日，15:00，XiaChang <13628620...@163.com> 写道：
>
> 您好，请教一个问题：
> 背景：
> 运行环境-->flink1.11.1
> source_table-->kafka输入表，kafka中每一个行为一个json数组。（[{},{}]）
> -- source
> CREATE TABLE source_table(
> __message STRING
> ) WITH(
> 'connector.type' = 'kafka',
> 'connector.version' = 'universal-xx',
> 'connector.properties.bootstrap.servers' = 'xxx',
> 'connector.topic' = 'xxx',
> 'connector.startup-mode' = 'latest-offset',
> 'connector.properties.group.id' = 'xxx',
> 'format.type' = 'json-xx',
> 'format.derive-schema' = 'true'
> );
>
>
> XX_JSON_TUPLE-->自定义udtf
> XX_ROW_VALUE-->自定义udf
>
>
> 可正常执行：
>
> INSERT INTO sink_table
> SELECT
> cast(XX_ROW_VALUE(msg, 1) as bigint) AS bbb
> ,((cast(XX_ROW_VALUE(msg, 0) as bigint))/1000) AS ddd
> ,cast(XX_ROW_VALUE(msg, 0) as bigint) AS aaa
> ,XX_ROW_VALUE(msg, 8) AS ccc
> FROM
> source_table, LATERAL TABLE(XX_JSON_TUPLE(__message, 'aaa','bbb','ccc')) AS
> T(msg)
>
>
> 不可正常运行：
> CREATE VIEW view_aaa AS
> SELECT
> cast(XX_ROW_VALUE(msg, 1) as bigint) AS bbb
> ,((cast(XX_ROW_VALUE(msg, 0) as bigint))/1000) AS ddd
> ,cast(XX_ROW_VALUE(msg, 0) as bigint) AS aaa
> ,XX_ROW_VALUE(msg, 8) AS ccc
> FROM
> source_table, LATERAL TABLE(XX_JSON_TUPLE(__message, 'aaa','bbb','ccc')) AS
> T(msg);
> INSERT INTO sink_table
> SELECT * FROM view_aaa;
>
>
>
>
> 相关错误堆栈信息：
> Caused by: org.apache.flink.table.api.ValidationException: SQL validation
> failed. From line 3, column 23 to line 3, column 33: Column '__message' not
> found in any table
> at
> org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org$apache$flink$table$planner$calcite$FlinkPlannerImpl$$validate(FlinkPlannerImpl.scala:146)
> at
> org.apache.flink.table.planner.calcite.FlinkPlannerImpl$ToRelContextImpl.expandView(FlinkPlannerImpl.scala:204)
> at org.apache.calcite.plan.ViewExpanders$1.expandView(ViewExpanders.java:52)
> at
> org.apache.flink.table.planner.catalog.SqlCatalogViewTable.convertToRel(SqlCatalogViewTable.java:58)
> at
> org.apache.flink.table.planner.plan.schema.ExpandingPreparingTable.expand(ExpandingPreparingTable.java:59)
> at
> org.apache.flink.table.planner.plan.schema.ExpandingPreparingTable.toRel(ExpandingPreparingTable.java:55)
> at
> org.apache.calcite.sql2rel.SqlToRelConverter.toRel(SqlToRelConverter.java:3492)
> at
> org.apache.calcite.sql2rel.SqlToRelConverter.convertIdentifier(SqlToRelConverter.java:2415)
> at
> org.apache.calcite.sql2rel.SqlToRelConverter.convertFrom(SqlToRelConverter.java:2102)
> at
> org.apache.calcite.sql2rel.SqlToRelConverter.convertFrom(SqlToRelConverter.java:2051)
> at
> org.apache.calcite.sql2rel.SqlToRelConverter.convertSelectImpl(SqlToRelConverter.java:661)
> at
> org.apache.calcite.sql2rel.SqlToRelConverter.convertSelect(SqlToRelConverter.java:642)
> at
> org.apache.calcite.sql2rel.SqlToRelConverter.convertQueryRecursive(SqlToRelConverter.java:3345)
> at
> org.apache.calcite.sql2rel.SqlToRelConverter.convertQuery(SqlToRelConverter.java:568)
> at
> org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org$apache$flink$table$planner$calcite$FlinkPlannerImpl$$rel(FlinkPlannerImpl.scala:164)
> at
> org.apache.flink.table.planner.calcite.FlinkPlannerImpl.rel(FlinkPlannerImpl.scala:151)
> at
> org.apache.flink.table.planner.operations.SqlToOperationConverter.toQueryOperation(SqlToOperationConverter.java:774)
> at
> org.apache.flink.table.planner.operations.SqlToOperationConverter.convertSqlQuery(SqlToOperationConverter.java:746)
> at
> org.apache.flink.table.planner.operations.SqlToOperationConverter.convert(SqlToOperationConverter.java:236)
> at
> org.apache.flink.table.planner.delegation.ParserImpl.parse(ParserImpl.java:78)
> at
> org.apache.flink.table.api.internal.TableEnvironmentImpl.sqlQuery(TableEnvironmentImpl.java:664)
> at
> com.douyu.ocean.xuanwu.flink.siddhi.SiddhiJob.lambda$callFrom$0(SiddhiJob.java:145)
> at java.util.ArrayList.forEach(ArrayList.java:1257)
> at com.douyu.ocean.xuanwu.flink.siddhi.SiddhiJob.callFrom(SiddhiJob.java:141)
> ... 4 common frames omitted
> Caused by: org.apache.calcite.runtime.CalciteContextException: From line 3,
> column 23 to line 3, column 33: Column '__message' not found in any table
> 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:457)
> at org.apache.calcite.sql.SqlUtil.newContextException(SqlUtil.java:839)
> at org.apache.calcite.sql.SqlUtil.newContextException(SqlUtil.java:824)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.newValidationError(SqlValidatorImpl.java:5089)
> at
> org.apache.calcite.sql.validate.DelegatingScope.fullyQualify(DelegatingScope.java:259)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl$DeriveTypeVisitor.visit(SqlValidatorImpl.java:5882)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl$DeriveTypeVisitor.visit(SqlValidatorImpl.java:5845)
> at org.apache.calcite.sql.SqlIdentifier.accept(SqlIdentifier.java:321)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.deriveTypeImpl(SqlValidatorImpl.java:1800)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.deriveType(SqlValidatorImpl.java:1785)
> at
> org.apache.calcite.sql.SqlCallBinding.getOperandType(SqlCallBinding.java:237)
> at
> org.apache.flink.table.planner.functions.utils.UserDefinedFunctionUtils$.$anonfun$getOperandType$1(UserDefinedFunctionUtils.scala:788)
> at
> org.apache.flink.table.planner.functions.utils.UserDefinedFunctionUtils$.$anonfun$getOperandType$1$adapted(UserDefinedFunctionUtils.scala:787)
> at scala.collection.TraversableLike.$anonfun$map$1(TraversableLike.scala:238)
> at scala.collection.immutable.Range.foreach(Range.scala:158)
> at scala.collection.TraversableLike.map(TraversableLike.scala:238)
> at scala.collection.TraversableLike.map$(TraversableLike.scala:231)
> at scala.collection.AbstractTraversable.map(Traversable.scala:108)
> at
> org.apache.flink.table.planner.functions.utils.UserDefinedFunctionUtils$.getOperandType(UserDefinedFunctionUtils.scala:787)
> at
> org.apache.flink.table.planner.functions.utils.TableSqlFunction$.inferOperandTypesInternal(TableSqlFunction.scala:119)
> at
> org.apache.flink.table.planner.functions.utils.TableSqlFunction$$anon$1.inferOperandTypes(TableSqlFunction.scala:107)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.inferUnknownTypes(SqlValidatorImpl.java:1980)
> at
> org.apache.calcite.sql.validate.ProcedureNamespace.validateImpl(ProcedureNamespace.java:56)
> at
> org.apache.calcite.sql.validate.AbstractNamespace.validate(AbstractNamespace.java:84)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.validateNamespace(SqlValidatorImpl.java:1110)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.validateQuery(SqlValidatorImpl.java:1084)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.validateFrom(SqlValidatorImpl.java:3256)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.validateFrom(SqlValidatorImpl.java:3238)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.validateJoin(SqlValidatorImpl.java:3303)
> at
> org.apache.flink.table.planner.calcite.FlinkCalciteSqlValidator.validateJoin(FlinkCalciteSqlValidator.java:86)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.validateFrom(SqlValidatorImpl.java:3247)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.validateSelect(SqlValidatorImpl.java:3510)
> at
> org.apache.calcite.sql.validate.SelectNamespace.validateImpl(SelectNamespace.java:60)
> at
> org.apache.calcite.sql.validate.AbstractNamespace.validate(AbstractNamespace.java:84)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.validateNamespace(SqlValidatorImpl.java:1110)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.validateQuery(SqlValidatorImpl.java:1084)
> at org.apache.calcite.sql.SqlSelect.validate(SqlSelect.java:232)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.validateScopedExpression(SqlValidatorImpl.java:1059)
> at
> org.apache.calcite.sql.validate.SqlValidatorImpl.validate(SqlValidatorImpl.java:766)
> at
> org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org$apache$flink$table$planner$calcite$FlinkPlannerImpl$$validate(FlinkPlannerImpl.scala:141)
> ... 27 common frames omitted
> Caused by: org.apache.calcite.sql.validate.SqlValidatorException: Column
> '__message' not found in any table
> 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:457)
> at org.apache.calcite.runtime.Resources$ExInst.ex(Resources.java:550)
> ... 66 common frames omitted
>
>
>
>
>
