您好，请教一个问题：
背景：
运行环境-->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
