Mate Czagany created FLINK-30899: ------------------------------------ Summary: FileSystemTableSource with CSV format incorrectly selects fields if filtering for partition Key: FLINK-30899 URL: https://issues.apache.org/jira/browse/FLINK-30899 Project: Flink Issue Type: Bug Components: Connectors / FileSystem Affects Versions: 1.17.0 Reporter: Mate Czagany
In my testing it only affected csv and testcsv formats. I think it's caused by `FileSystemTableSource` calling `DeserializationFormatFactory#createRuntimeDecoder` with wrong `physicalDataType`. The files won't contain the partitioned field values, but in case of a projection pushdown (which can happen during planning phase if we filter the partition field by a constant value) the final `physicalDataType` passed to the deserializer by `FileSystemTableSource` will contain the partitioned fields as well. As described in `DecodingFormat`, every field in the `physicalDataType` parameter will have to be present in the serialized record. Example: {code:java} CREATE TABLE test_table ( f0 INT, f1 INT, f2 INT, f3 INT ) PARTITIONED BY (f0,f1) WITH ( 'connector' = 'filesystem', 'path' = 'file:///path/to/whatever', 'format' = 'csv' ) SELECT * FROM test_table WHERE f0 = 1; -- !!!! should be 1,4,7,10 !!!! +-------------+-------------+-------------+-------------+ | f0 | f1 | f2 | f3 | +-------------+-------------+-------------+-------------+ | 1 | 4 | 10 | 0 | +-------------+-------------+-------------+-------------+ SELECT * FROM test_table; +-------------+-------------+-------------+-------------+ | f0 | f1 | f2 | f3 | +-------------+-------------+-------------+-------------+ | 2 | 5 | 8 | 11 | | 1 | 4 | 7 | 10 | | 3 | 6 | 9 | 12 | +-------------+-------------+-------------+-------------+ SELECT * FROM test_table WHERE f0>0; +-------------+-------------+-------------+-------------+ | f0 | f1 | f2 | f3 | +-------------+-------------+-------------+-------------+ | 1 | 4 | 7 | 10 | | 3 | 6 | 9 | 12 | | 2 | 5 | 8 | 11 | +-------------+-------------+-------------+-------------+ SELECT * FROM test_table WHERE f0 = 1 AND f1 = 4; ... Caused by: java.lang.ArrayIndexOutOfBoundsException: Index -1 out of bounds for length 4 at org.apache.flink.types.parser.IntParser.parseField(IntParser.java:49) at org.apache.flink.types.parser.IntParser.parseField(IntParser.java:27) at org.apache.flink.types.parser.FieldParser.resetErrorStateAndParse(FieldParser.java:101) at org.apache.flink.formats.testcsv.TestCsvDeserializationSchema.deserialize(TestCsvDeserializationSchema.java:92) at org.apache.flink.formats.testcsv.TestCsvDeserializationSchema.deserialize(TestCsvDeserializationSchema.java:42) at org.apache.flink.api.common.serialization.DeserializationSchema.deserialize(DeserializationSchema.java:82) ... {code} At [https://github.com/apache/flink/blob/b1e70aebd3e248d68cf41a43db385ec9c9b6235a/flink-connectors/flink-connector-files/src/main/java/org/apache/flink/connector/file/table/FileSystemTableSource.java#L147] the `physicalRowDataType` will contain the partition fields as well, but `partitionKeysToExtract` will not contain it since `producedDataType` has been modified in the `applyProjection` method, so it will result in an empty projection. Then on line 154 we construct the final `physicalDataType`, but since `partitionKeysProjections` is empty, it will result with the same value as `physicalDataType` which contains the partition fields too. By changing {code:java} final Projection partitionKeysProjections = Projection.fromFieldNames(physicalDataType, partitionKeysToExtract);{code} to {code:java} final Projection partitionKeysProjections = Projection.fromFieldNames(physicalDataType, partitionKeys);{code} the issue can be solved. I have verified this solution with 1 and 2 partition keys, with and without metadata columns, with and without virtual columns. But I still need to test this change with other formats. If this solution seems correct and a committer could assign me to the JIRA I can start working on it -- This message was sent by Atlassian Jira (v8.20.10#820010)