首先,我的Flink版本为1.16.0 为了方便理解,我以Kafka作为案例来描述: 我有以下两个表: CREATE TABLE orders( user_id BIGINT, name STRING, timestamp TIMESTAMP(3) METADATA VIRTUAL )WITH( 'connector'='kafka', 'topic'='orders', 'properties.group.id' = 'test_join_tempral', 'scan.startup.mode'='earliest-offset', 'properties.bootstrap.servers'='localhost:9092', 'format'='json', 'json.ignore-parse-errors' = 'true' ); CREATE TABLE kafka_sink( user_id BIGINT, name STRING, timestamp TIMESTAMP(3) METADATA FROM 'timestamp' )WITH( 'connector'='kafka', 'topic'='kafka_sink', 'properties.group.id' = 'test_join_tempral', 'scan.startup.mode'='earliest-offset', 'properties.bootstrap.servers'='localhost:9092', 'format'='json', 'json.ignore-parse-errors' = 'true' );
正常情况: Flink SQL> insert into kafka_sink select user_id,name,`timestamp` from orders; [INFO] Submitting SQL update statement to the cluster... [INFO] SQL update statement has been successfully submitted to the cluster: Job ID: e419ae9d2cad4c3c2a2c1150c1a86653 异常情况: Flink SQL> insert into kafka_sink(user_id,name,`timestamp`) select user_id,name,`timestamp` from orders; [ERROR] Could not execute SQL statement. Reason: org.apache.calcite.sql.validate.SqlValidatorException: Unknown target column 'timestamp' 很奇怪,为什么指定列名就不行了呢?而且还是识别不到”ts”列,kafka_sink schema如下: Flink SQL> describe kafka_sink; +-----------+--------------+------+-----+---------------------------+-----------+ | name | type | null | key | extras | watermark | +-----------+--------------+------+-----+---------------------------+-----------+ | user_id | BIGINT | TRUE | | | | | name | STRING | TRUE | | | | | timestamp | TIMESTAMP(3) | TRUE | | METADATA FROM 'timestamp' | | +-----------+--------------+------+-----+---------------------------+-----------+ 恳请解答!