[FLINK-18665] Filesystem connector should use TableSchema exclude computed columns

• Type: Bug
• Status: Closed
• Priority: Critical
• Resolution: Fixed
• Affects Version/s: 1.11.1
• Fix Version/s: 1.11.2
• Component/s: Connectors / FileSystem, Table SQL / Ecosystem
• Labels: pull-request-available
• Assignee: Leonard Xu
• Reporter: Jark Wu
• Created: 21/Jul/20 15:06
• Updated: 23/Jul/20 05:35
• Resolved: 23/Jul/20 02:15

Description

This is reported in http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/How-to-use-a-nested-column-for-CREATE-TABLE-PARTITIONED-BY-td36796.html

create table navi (
  a STRING,
  location ROW<lastUpdateTime BIGINT, transId STRING>
) with (
  'connector' = 'filesystem',
  'path' = 'east-out',
  'format' = 'json'
)
CREATE TABLE output (
  `partition` AS location.transId
) PARTITIONED BY (`partition`)
WITH (
  'connector' = 'filesystem',
  'path' = 'east-out',
  'format' = 'json'
) LIKE navi (EXCLUDING ALL)
tEnv.sqlQuery("SELECT type, location FROM navi").executeInsert("output")

It throws the following exception

Exception in thread "main" org.apache.flink.table.api.ValidationException: The field count of logical schema of the table does not match with the field count of physical schema
. The logical schema: [STRING,ROW<`lastUpdateTime` BIGINT, `transId` STRING>]
The physical schema: [STRING,ROW<`lastUpdateTime` BIGINT, `transId` STRING>,STRING].

The reason is that FileSystemTableFactory#createTableSource should use schema excluded computed column, not the original catalog table schema.

[1]: https://github.com/apache/flink/blob/master/flink-table/flink-table-runtime-blink/src/main/java/org/apache/flink/table/filesystem/FileSystemTableFactory.java#L78

Issue Links

• links to: GitHub Pull Request #12953
• links to: GitHub Pull Request #12954