[
https://issues.apache.org/jira/browse/FLINK-26760?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17516747#comment-17516747
]
Francesco Guardiani commented on FLINK-26760:
---------------------------------------------
Ok just checked the TPC-DS spec:
> Comment: The data generated by dsdgen includes some international characters.
> Examples of international
characters are Ô and É. The database must preserve these characters during
loading and processing by using a
character encoding such as ISO/IEC 8859-1 that includes these characters.
from https://www.tpc.org/tpc_documents_current_versions/pdf/tpc-ds_v3.2.0.pdf
> The new CSV source (file system source + CSV format) does not support reading
> files whose file encoding is not UTF-8
> --------------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-26760
> URL: https://issues.apache.org/jira/browse/FLINK-26760
> Project: Flink
> Issue Type: Improvement
> Components: Connectors / FileSystem, Formats (JSON, Avro, Parquet,
> ORC, SequenceFile)
> Affects Versions: 1.15.0, 1.13.6, 1.14.4
> Reporter: Lijie Wang
> Assignee: Arvid Heise
> Priority: Major
> Fix For: 1.15.0
>
> Attachments: PerformanceTest.java, example.csv
>
>
> The new CSV source (file system source + CSV format) does not support reading
> files whose file encoding is not UTF-8, but the legacy {{CsvTableSource}}
> supports it.
> We provide an {{*example.csv*}} whose file encoding is {{{}ISO-8599-1{}}}.
> When reading it with the legacy {{{}CsvTableSource{}}}, it executes correctly:
> {code:java}
> @Test
> public void testLegacyCsvSource() {
> EnvironmentSettings environmentSettings =
> EnvironmentSettings.inBatchMode();
> TableEnvironment tEnv = TableEnvironment.create(environmentSettings);
> CsvTableSource.Builder builder = CsvTableSource.builder();
> CsvTableSource source =
> builder.path("example.csv")
> .emptyColumnAsNull()
> .lineDelimiter("\n")
> .fieldDelimiter("|")
> .field("name", DataTypes.STRING())
> .build();
> ConnectorCatalogTable catalogTable =
> ConnectorCatalogTable.source(source, true);
> tEnv.getCatalog(tEnv.getCurrentCatalog())
> .ifPresent(
> catalog -> {
> try {
> catalog.createTable(
> new
> ObjectPath(tEnv.getCurrentDatabase(), "example"),
> catalogTable,
> false);
> } catch (Exception e) {
> throw new RuntimeException(e);
> }
> });
> tEnv.executeSql("select count(name) from example").print();
> }
> {code}
>
> When reading it with the new CSV source (file system source + CSV format), it
> throws the following error:
> {code:java}
> @Test
> public void testNewCsvSource() {
> EnvironmentSettings environmentSettings =
> EnvironmentSettings.inBatchMode();
> TableEnvironment tEnv = TableEnvironment.create(environmentSettings);
> String ddl =
> "create table example ("
> + " name string"
> + ") with ("
> + " 'connector' = 'filesystem',"
> + " 'path' = 'example.csv',"
> + " 'format' = 'csv',"
> + " 'csv.array-element-delimiter' = '\n',"
> + " 'csv.field-delimiter' = '|',"
> + " 'csv.null-literal' = ''"
> + ")";
> tEnv.executeSql(ddl);
> tEnv.executeSql("select count(name) from example").print();
> }
> {code}
> {code:java}
> Caused by: java.lang.RuntimeException: One or more fetchers have encountered
> exception
> at
> org.apache.flink.connector.base.source.reader.fetcher.SplitFetcherManager.checkErrors(SplitFetcherManager.java:225)
> at
> org.apache.flink.connector.base.source.reader.SourceReaderBase.getNextFetch(SourceReaderBase.java:169)
> at
> org.apache.flink.connector.base.source.reader.SourceReaderBase.pollNext(SourceReaderBase.java:130)
> at
> org.apache.flink.streaming.api.operators.SourceOperator.emitNext(SourceOperator.java:385)
> at
> org.apache.flink.streaming.runtime.io.StreamTaskSourceInput.emitNext(StreamTaskSourceInput.java:68)
> at
> org.apache.flink.streaming.runtime.io.StreamOneInputProcessor.processInput(StreamOneInputProcessor.java:65)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(StreamTask.java:519)
> at
> org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor.runMailboxLoop(MailboxProcessor.java:203)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(StreamTask.java:804)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:753)
> at
> org.apache.flink.runtime.taskmanager.Task.runWithSystemExitMonitoring(Task.java:948)
> at
> org.apache.flink.runtime.taskmanager.Task.restoreAndInvoke(Task.java:927)
> at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:741)
> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:563)
> at java.lang.Thread.run(Thread.java:748)
> Caused by: java.lang.RuntimeException: SplitFetcher thread 0 received
> unexpected exception while polling the records
> at
> org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:150)
> at
> org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.run(SplitFetcher.java:105)
> at
> java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511)
> at java.util.concurrent.FutureTask.run(FutureTask.java:266)
> at
> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
> at
> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
> ... 1 more
> Caused by: java.lang.RuntimeException: Invalid UTF-8 middle byte 0x54 (at
> char #9, byte #8): check content encoding, does not look like UTF-8
> at
> org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.MappingIterator._handleIOException(MappingIterator.java:417)
> at
> org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.MappingIterator.next(MappingIterator.java:203)
> at
> org.apache.flink.formats.csv.CsvReaderFormat$Reader.read(CsvReaderFormat.java:196)
> at
> org.apache.flink.connector.file.src.impl.StreamFormatAdapter$Reader.readBatch(StreamFormatAdapter.java:214)
> at
> org.apache.flink.connector.file.src.impl.FileSourceSplitReader.fetch(FileSourceSplitReader.java:67)
> at
> org.apache.flink.connector.base.source.reader.fetcher.FetchTask.run(FetchTask.java:58)
> at
> org.apache.flink.connector.base.source.reader.fetcher.SplitFetcher.runOnce(SplitFetcher.java:142)
> ... 6 more
> Caused by: java.io.CharConversionException: Invalid UTF-8 middle byte 0x54
> (at char #9, byte #8): check content encoding, does not look like UTF-8
> at
> org.apache.flink.shaded.jackson2.com.fasterxml.jackson.dataformat.csv.impl.UTF8Reader.reportInvalidOther(UTF8Reader.java:507)
> at
> org.apache.flink.shaded.jackson2.com.fasterxml.jackson.dataformat.csv.impl.UTF8Reader.reportDeferredInvalid(UTF8Reader.java:518)
> at
> org.apache.flink.shaded.jackson2.com.fasterxml.jackson.dataformat.csv.impl.UTF8Reader.read(UTF8Reader.java:172)
> at
> org.apache.flink.shaded.jackson2.com.fasterxml.jackson.dataformat.csv.impl.CsvDecoder.loadMore(CsvDecoder.java:443)
> at
> org.apache.flink.shaded.jackson2.com.fasterxml.jackson.dataformat.csv.impl.CsvDecoder.nextString(CsvDecoder.java:679)
> at
> org.apache.flink.shaded.jackson2.com.fasterxml.jackson.dataformat.csv.CsvParser._handleNextEntry(CsvParser.java:915)
> at
> org.apache.flink.shaded.jackson2.com.fasterxml.jackson.dataformat.csv.CsvParser.nextFieldName(CsvParser.java:727)
> at
> org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.deser.std.BaseNodeDeserializer.deserializeObject(JsonNodeDeserializer.java:268)
> at
> org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.deser.std.JsonNodeDeserializer.deserialize(JsonNodeDeserializer.java:69)
> at
> org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.deser.std.JsonNodeDeserializer.deserialize(JsonNodeDeserializer.java:16)
> at
> org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.MappingIterator.nextValue(MappingIterator.java:280)
> at
> org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.MappingIterator.next(MappingIterator.java:199)
> ... 11 more
> {code}
>
--
This message was sent by Atlassian Jira
(v8.20.1#820001)
