This is an automated email from the ASF dual-hosted git repository. chesnay pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink.git
The following commit(s) were added to refs/heads/master by this push: new 1dee35a9966 [FLINK-28624][csv] Accept mapper/schema factories 1dee35a9966 is described below commit 1dee35a9966065593aa5dfc618f5fec17fb80890 Author: Chesnay Schepler <ches...@apache.org> AuthorDate: Tue Jul 26 12:51:16 2022 +0200 [FLINK-28624][csv] Accept mapper/schema factories --- .../docs/connectors/datastream/filesystem.md | 4 +- .../docs/connectors/datastream/formats/csv.md | 12 +-- .../docs/connectors/datastream/filesystem.md | 4 +- .../docs/connectors/datastream/formats/csv.md | 12 +-- .../flink/formats/csv/CsvFileFormatFactory.java | 4 +- .../apache/flink/formats/csv/CsvReaderFormat.java | 75 +++++++++----- .../flink/formats/csv/CsvReaderFormatTest.java | 115 +++++++++++++++++++++ .../flink/formats/csv/DataStreamCsvITCase.java | 12 ++- flink-python/pom.xml | 6 ++ flink-python/pyflink/datastream/formats/csv.py | 30 +----- .../flink/formats/csv/CsvReaderFormatFactory.java | 45 ++++++++ 11 files changed, 245 insertions(+), 74 deletions(-) diff --git a/docs/content.zh/docs/connectors/datastream/filesystem.md b/docs/content.zh/docs/connectors/datastream/filesystem.md index 3c4b0cea567..cf8e6e1dae2 100644 --- a/docs/content.zh/docs/connectors/datastream/filesystem.md +++ b/docs/content.zh/docs/connectors/datastream/filesystem.md @@ -161,8 +161,8 @@ FileSource<SomePojo> source = 如果需要对 CSV 模式或解析选项进行更细粒度的控制,可以使用 `CsvReaderFormat` 的更底层的 `forSchema` 静态工厂方法: ```java -CsvReaderFormat<T> forSchema(CsvMapper mapper, - CsvSchema schema, +CsvReaderFormat<T> forSchema(Supplier<CsvMapper> mapperFactory, + Function<CsvMapper, CsvSchema> schemaGenerator, TypeInformation<T> typeInformation) ``` diff --git a/docs/content.zh/docs/connectors/datastream/formats/csv.md b/docs/content.zh/docs/connectors/datastream/formats/csv.md index 870dde3488d..dbf8bd6be6c 100644 --- a/docs/content.zh/docs/connectors/datastream/formats/csv.md +++ b/docs/content.zh/docs/connectors/datastream/formats/csv.md @@ -60,8 +60,8 @@ Note: you might need to add `@JsonPropertyOrder({field1, field2, ...})` annotati If you need more fine-grained control over the CSV schema or the parsing options, use the more low-level `forSchema` static factory method of `CsvReaderFormat`: ```java -CsvReaderFormat<T> forSchema(CsvMapper mapper, - CsvSchema schema, +CsvReaderFormat<T> forSchema(Supplier<CsvMapper> mapperFactory, + Function<CsvMapper, CsvSchema> schemaGenerator, TypeInformation<T> typeInformation) ``` Below is an example of reading a POJO with a custom columns' separator: @@ -80,15 +80,14 @@ Below is an example of reading a POJO with a custom columns' separator: public long population; } -CsvMapper mapper = new CsvMapper(); -CsvSchema schema = +Function<CsvMapper, CsvSchema> schemaGenerator = mapper -> mapper.schemaFor(CityPojo.class).withoutQuoteChar().withColumnSeparator('|'); CsvReaderFormat<CityPojo> csvFormat = - CsvReaderFormat.forSchema(mapper, schema, TypeInformation.of(CityPojo.class)); + CsvReaderFormat.forSchema(() -> new CsvMapper(), schemaGenerator, TypeInformation.of(CityPojo.class)); FileSource<CityPojo> source = - FileSource.forRecordStreamFormat(csvFormat,Path.fromLocalFile(...)).build(); + FileSource.forRecordStreamFormat(csvFormat, Path.fromLocalFile(...)).build(); ``` The corresponding CSV file: ``` @@ -105,7 +104,6 @@ public static class ComplexPojo { CsvReaderFormat<ComplexPojo> csvFormat = CsvReaderFormat.forSchema( - new CsvMapper(), CsvSchema.builder() .addColumn( new CsvSchema.Column(0, "id", CsvSchema.ColumnType.NUMBER)) diff --git a/docs/content/docs/connectors/datastream/filesystem.md b/docs/content/docs/connectors/datastream/filesystem.md index f9f0f7ae05b..ce2a6c50824 100644 --- a/docs/content/docs/connectors/datastream/filesystem.md +++ b/docs/content/docs/connectors/datastream/filesystem.md @@ -157,8 +157,8 @@ The schema for CSV parsing, in this case, is automatically derived based on the If you need more fine-grained control over the CSV schema or the parsing options, use the more low-level `forSchema` static factory method of `CsvReaderFormat`: ```java -CsvReaderFormat<T> forSchema(CsvMapper mapper, - CsvSchema schema, +CsvReaderFormat<T> forSchema(Supplier<CsvMapper> mapperFactory, + Function<CsvMapper, CsvSchema> schemaGenerator, TypeInformation<T> typeInformation) ``` diff --git a/docs/content/docs/connectors/datastream/formats/csv.md b/docs/content/docs/connectors/datastream/formats/csv.md index 870dde3488d..dbf8bd6be6c 100644 --- a/docs/content/docs/connectors/datastream/formats/csv.md +++ b/docs/content/docs/connectors/datastream/formats/csv.md @@ -60,8 +60,8 @@ Note: you might need to add `@JsonPropertyOrder({field1, field2, ...})` annotati If you need more fine-grained control over the CSV schema or the parsing options, use the more low-level `forSchema` static factory method of `CsvReaderFormat`: ```java -CsvReaderFormat<T> forSchema(CsvMapper mapper, - CsvSchema schema, +CsvReaderFormat<T> forSchema(Supplier<CsvMapper> mapperFactory, + Function<CsvMapper, CsvSchema> schemaGenerator, TypeInformation<T> typeInformation) ``` Below is an example of reading a POJO with a custom columns' separator: @@ -80,15 +80,14 @@ Below is an example of reading a POJO with a custom columns' separator: public long population; } -CsvMapper mapper = new CsvMapper(); -CsvSchema schema = +Function<CsvMapper, CsvSchema> schemaGenerator = mapper -> mapper.schemaFor(CityPojo.class).withoutQuoteChar().withColumnSeparator('|'); CsvReaderFormat<CityPojo> csvFormat = - CsvReaderFormat.forSchema(mapper, schema, TypeInformation.of(CityPojo.class)); + CsvReaderFormat.forSchema(() -> new CsvMapper(), schemaGenerator, TypeInformation.of(CityPojo.class)); FileSource<CityPojo> source = - FileSource.forRecordStreamFormat(csvFormat,Path.fromLocalFile(...)).build(); + FileSource.forRecordStreamFormat(csvFormat, Path.fromLocalFile(...)).build(); ``` The corresponding CSV file: ``` @@ -105,7 +104,6 @@ public static class ComplexPojo { CsvReaderFormat<ComplexPojo> csvFormat = CsvReaderFormat.forSchema( - new CsvMapper(), CsvSchema.builder() .addColumn( new CsvSchema.Column(0, "id", CsvSchema.ColumnType.NUMBER)) diff --git a/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvFileFormatFactory.java b/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvFileFormatFactory.java index e59ec63794f..2054c182a01 100644 --- a/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvFileFormatFactory.java +++ b/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvFileFormatFactory.java @@ -138,8 +138,8 @@ public class CsvFileFormatFactory implements BulkReaderFormatFactory, BulkWriter .createRowConverter(projectedRowType, true); CsvReaderFormat<RowData> csvReaderFormat = new CsvReaderFormat<>( - new CsvMapper(), - schema, + () -> new CsvMapper(), + ignored -> schema, JsonNode.class, converter, context.createTypeInformation(projectedDataType), diff --git a/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvReaderFormat.java b/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvReaderFormat.java index 1ce00fd4d70..c2d14008cb9 100644 --- a/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvReaderFormat.java +++ b/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvReaderFormat.java @@ -25,6 +25,8 @@ import org.apache.flink.connector.file.src.reader.SimpleStreamFormat; import org.apache.flink.connector.file.src.reader.StreamFormat; import org.apache.flink.core.fs.FSDataInputStream; import org.apache.flink.formats.common.Converter; +import org.apache.flink.util.function.SerializableFunction; +import org.apache.flink.util.function.SerializableSupplier; import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.MappingIterator; import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.dataformat.csv.CsvMapper; @@ -33,6 +35,8 @@ import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.dataformat.csv.Csv import javax.annotation.Nullable; import java.io.IOException; +import java.util.function.Function; +import java.util.function.Supplier; import static org.apache.flink.util.Preconditions.checkNotNull; @@ -56,11 +60,11 @@ import static org.apache.flink.util.Preconditions.checkNotNull; * low-level {@code forSchema} static factory method based on the {@code Jackson} library utilities: * * <pre>{@code - * CsvMapper mapper = new CsvMapper(); - * CsvSchema schema = mapper.schemaFor(SomePojo.class) + * Function<CsvMapper, CsvSchema> schemaGenerator = + * mapper -> mapper.schemaFor(SomePojo.class) * .withColumnSeparator('|'); * CsvReaderFormat<SomePojo> csvFormat = - * CsvReaderFormat.forSchema(mapper,schema, TypeInformation.of(SomePojo.class)); + * CsvReaderFormat.forSchema(() -> new CsvMapper(), schemaGenerator, TypeInformation.of(SomePojo.class)); * FileSource<SomePojo> source = * FileSource.forRecordStreamFormat(csvFormat, Path.fromLocalFile(filesPath)).build(); * }</pre> @@ -72,8 +76,8 @@ public class CsvReaderFormat<T> extends SimpleStreamFormat<T> { private static final long serialVersionUID = 1L; - private final CsvMapper mapper; - private final CsvSchema schema; + private final SerializableSupplier<CsvMapper> mapperFactory; + private final SerializableFunction<CsvMapper, CsvSchema> schemaGenerator; private final Class<Object> rootType; private final Converter<Object, T, Void> converter; private final TypeInformation<T> typeInformation; @@ -81,14 +85,14 @@ public class CsvReaderFormat<T> extends SimpleStreamFormat<T> { @SuppressWarnings("unchecked") <R> CsvReaderFormat( - CsvMapper mapper, - CsvSchema schema, + SerializableSupplier<CsvMapper> mapperFactory, + SerializableFunction<CsvMapper, CsvSchema> schemaGenerator, Class<R> rootType, Converter<R, T, Void> converter, TypeInformation<T> typeInformation, boolean ignoreParseErrors) { - this.mapper = checkNotNull(mapper); - this.schema = checkNotNull(schema); + this.mapperFactory = checkNotNull(mapperFactory); + this.schemaGenerator = checkNotNull(schemaGenerator); this.rootType = (Class<Object>) checkNotNull(rootType); this.typeInformation = checkNotNull(typeInformation); this.converter = (Converter<Object, T, Void>) checkNotNull(converter); @@ -104,23 +108,45 @@ public class CsvReaderFormat<T> extends SimpleStreamFormat<T> { */ public static <T> CsvReaderFormat<T> forSchema( CsvSchema schema, TypeInformation<T> typeInformation) { - return forSchema(new CsvMapper(), schema, typeInformation); + return forSchema(() -> new CsvMapper(), ignored -> schema, typeInformation); } /** - * Builds a new {@code CsvReaderFormat} using a {@code CsvSchema} and a pre-created {@code - * CsvMapper}. + * @deprecated This method is limited to serializable {@link CsvMapper CsvMappers}, preventing + * the usage of certain Jackson modules (like the {@link + * org.apache.flink.shaded.jackson2.com.fasterxml.jackson.datatype.jsr310.JavaTimeModule + * Java 8 Date/Time Serializers}). Use {@link #forSchema(Supplier, Function, + * TypeInformation)} instead. + */ + @Deprecated + public static <T> CsvReaderFormat<T> forSchema( + CsvMapper mapper, CsvSchema schema, TypeInformation<T> typeInformation) { + return new CsvReaderFormat<>( + () -> mapper, + ignored -> schema, + typeInformation.getTypeClass(), + (value, context) -> value, + typeInformation, + false); + } + + /** + * Builds a new {@code CsvReaderFormat} using a {@code CsvSchema} generator and {@code + * CsvMapper} factory. * - * @param mapper The pre-created {@code CsvMapper}. - * @param schema The Jackson CSV schema configured for parsing specific CSV files. + * @param mapperFactory The factory creating the {@code CsvMapper}. + * @param schemaGenerator A generator that creates and configures the Jackson CSV schema for + * parsins specific CSV files, from a mapper created by the mapper factory. * @param typeInformation The Flink type descriptor of the returned elements. * @param <T> The type of the returned elements. */ public static <T> CsvReaderFormat<T> forSchema( - CsvMapper mapper, CsvSchema schema, TypeInformation<T> typeInformation) { + SerializableSupplier<CsvMapper> mapperFactory, + SerializableFunction<CsvMapper, CsvSchema> schemaGenerator, + TypeInformation<T> typeInformation) { return new CsvReaderFormat<>( - mapper, - schema, + mapperFactory, + schemaGenerator, typeInformation.getTypeClass(), (value, context) -> value, typeInformation, @@ -136,10 +162,9 @@ public class CsvReaderFormat<T> extends SimpleStreamFormat<T> { * @param <T> The type of the returned elements. */ public static <T> CsvReaderFormat<T> forPojo(Class<T> pojoType) { - CsvMapper mapper = new CsvMapper(); return forSchema( - mapper, - mapper.schemaFor(pojoType).withoutQuoteChar(), + () -> new CsvMapper(), + mapper -> mapper.schemaFor(pojoType).withoutQuoteChar(), TypeInformation.of(pojoType)); } @@ -149,8 +174,8 @@ public class CsvReaderFormat<T> extends SimpleStreamFormat<T> { */ public CsvReaderFormat<T> withIgnoreParseErrors() { return new CsvReaderFormat<T>( - this.mapper, - this.schema, + this.mapperFactory, + this.schemaGenerator, this.rootType, this.converter, this.typeInformation, @@ -160,8 +185,12 @@ public class CsvReaderFormat<T> extends SimpleStreamFormat<T> { @Override public StreamFormat.Reader<T> createReader(Configuration config, FSDataInputStream stream) throws IOException { + final CsvMapper csvMapper = mapperFactory.get(); return new Reader<>( - mapper.readerFor(rootType).with(schema).readValues(stream), + csvMapper + .readerFor(rootType) + .with(schemaGenerator.apply(csvMapper)) + .readValues(stream), converter, ignoreParseErrors); } diff --git a/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvReaderFormatTest.java b/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvReaderFormatTest.java new file mode 100644 index 00000000000..f6b04748f98 --- /dev/null +++ b/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvReaderFormatTest.java @@ -0,0 +1,115 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.formats.csv; + +import org.apache.flink.api.common.io.InputStreamFSInputWrapper; +import org.apache.flink.api.common.typeinfo.TypeInformation; +import org.apache.flink.configuration.Configuration; +import org.apache.flink.util.InstantiationUtil; + +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.dataformat.csv.CsvMapper; +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.dataformat.csv.CsvSchema; +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.datatype.jsr310.JavaTimeModule; + +import org.junit.jupiter.api.Test; + +import java.io.ByteArrayInputStream; +import java.io.IOException; +import java.util.concurrent.atomic.AtomicReference; + +import static org.assertj.core.api.Assertions.assertThat; + +class CsvReaderFormatTest { + + @Test + void testForPojoSerializability() throws IOException, ClassNotFoundException { + final CsvReaderFormat<Pojo> format = CsvReaderFormat.forPojo(Pojo.class); + + final byte[] bytes = InstantiationUtil.serializeObject(format); + InstantiationUtil.deserializeObject(bytes, CsvReaderFormatTest.class.getClassLoader()); + } + + @Test + void testForSchemaSerializability() throws IOException, ClassNotFoundException { + final CsvSchema schema = new CsvMapper().schemaFor(Pojo.class); + final CsvReaderFormat<Pojo> format = + CsvReaderFormat.forSchema(schema, TypeInformation.of(Pojo.class)); + + final byte[] bytes = InstantiationUtil.serializeObject(format); + InstantiationUtil.deserializeObject(bytes, CsvReaderFormatTest.class.getClassLoader()); + } + + @Test + void testForSchemaWithMapperSerializability() throws IOException, ClassNotFoundException { + final CsvReaderFormat<Pojo> format = + CsvReaderFormat.forSchema( + () -> new CsvMapper(), + mapper -> mapper.schemaFor(Pojo.class), + TypeInformation.of(Pojo.class)); + + final byte[] bytes = InstantiationUtil.serializeObject(format); + InstantiationUtil.deserializeObject(bytes, CsvReaderFormatTest.class.getClassLoader()); + } + + /** + * Verifies that we don't use eagerly use the mapper factory in the constructor to initialize + * some non-transient field. + */ + @Test + void testForSchemaWithMapperSerializabilityWithUnserializableMapper() + throws IOException, ClassNotFoundException { + final CsvReaderFormat<Pojo> format = + CsvReaderFormat.forSchema( + () -> { + final CsvMapper csvMapper = new CsvMapper(); + // this module is not serializable + csvMapper.registerModule(new JavaTimeModule()); + return csvMapper; + }, + mapper -> mapper.schemaFor(Pojo.class), + TypeInformation.of(Pojo.class)); + + final byte[] bytes = InstantiationUtil.serializeObject(format); + InstantiationUtil.deserializeObject(bytes, CsvReaderFormatTest.class.getClassLoader()); + } + + @Test + void testCreatedMapperPassedToSchemaFunction() throws IOException, ClassNotFoundException { + final CsvMapper csvMapper = new CsvMapper(); + + AtomicReference<CsvMapper> passedMapper = new AtomicReference<>(); + + final CsvReaderFormat<Pojo> format = + CsvReaderFormat.forSchema( + () -> csvMapper, + mapper -> { + passedMapper.set(csvMapper); + return mapper.schemaFor(Pojo.class); + }, + TypeInformation.of(Pojo.class)); + + format.createReader( + new Configuration(), + new InputStreamFSInputWrapper(new ByteArrayInputStream(new byte[0]))); + assertThat(passedMapper.get()).isSameAs(csvMapper); + } + + public static class Pojo { + public int x; + } +} diff --git a/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/DataStreamCsvITCase.java b/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/DataStreamCsvITCase.java index 97fd78459fc..ea073432116 100644 --- a/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/DataStreamCsvITCase.java +++ b/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/DataStreamCsvITCase.java @@ -161,12 +161,14 @@ public class DataStreamCsvITCase { public void testCsvReaderFormatFromSchema() throws Exception { writeFile(outDir, "data.csv", CSV_LINES_PIPE_SEPARATED); - CsvMapper mapper = new CsvMapper(); - CsvSchema schema = - mapper.schemaFor(CityPojo.class).withoutQuoteChar().withColumnSeparator('|'); - final CsvReaderFormat<CityPojo> csvFormat = - CsvReaderFormat.forSchema(mapper, schema, TypeInformation.of(CityPojo.class)); + CsvReaderFormat.forSchema( + () -> new CsvMapper(), + mapper -> + mapper.schemaFor(CityPojo.class) + .withoutQuoteChar() + .withColumnSeparator('|'), + TypeInformation.of(CityPojo.class)); final List<CityPojo> result = initializeSourceAndReadData(outDir, csvFormat); assertThat(Arrays.asList(POJOS)).isEqualTo(result); diff --git a/flink-python/pom.xml b/flink-python/pom.xml index 94133e2288b..fa5fe677285 100644 --- a/flink-python/pom.xml +++ b/flink-python/pom.xml @@ -77,6 +77,12 @@ under the License. <version>${project.version}</version> <scope>provided</scope> </dependency> + <dependency> + <groupId>org.apache.flink</groupId> + <artifactId>flink-csv</artifactId> + <version>${project.version}</version> + <scope>provided</scope> + </dependency> <!-- Beam dependencies --> diff --git a/flink-python/pyflink/datastream/formats/csv.py b/flink-python/pyflink/datastream/formats/csv.py index 82f405da191..2dc00f59d2d 100644 --- a/flink-python/pyflink/datastream/formats/csv.py +++ b/flink-python/pyflink/datastream/formats/csv.py @@ -17,11 +17,9 @@ ################################################################################ from typing import Optional, cast -from py4j.java_gateway import get_java_class from pyflink.datastream.connectors import StreamFormat from pyflink.java_gateway import get_gateway from pyflink.table.types import DataType, DataTypes, _to_java_data_type, RowType, NumericType -from pyflink.util.java_utils import to_jarray class CsvSchema(object): @@ -295,29 +293,9 @@ class CsvReaderFormat(StreamFormat): Builds a :class:`CsvReaderFormat` using `CsvSchema`. """ jvm = get_gateway().jvm - jackson = jvm.org.apache.flink.shaded.jackson2.com.fasterxml.jackson - constructor = get_java_class(jvm.org.apache.flink.formats.csv.CsvReaderFormat) \ - .getDeclaredConstructor( - to_jarray(jvm.Class, [ - get_java_class(jackson.dataformat.csv.CsvMapper), - get_java_class(jackson.dataformat.csv.CsvSchema), - get_java_class(jvm.Class), - get_java_class(jvm.org.apache.flink.formats.common.Converter), - get_java_class(jvm.org.apache.flink.api.common.typeinfo.TypeInformation), - get_java_class(jvm.boolean) - ]) - ) - constructor.setAccessible(True) - j_csv_format = constructor.newInstance( - to_jarray(jvm.Object, [ - jackson.dataformat.csv.CsvMapper(), + j_csv_format = jvm.org.apache.flink.formats.csv.CsvReaderFormatFactory \ + .createCsvReaderFormat( schema._j_schema, - get_java_class(jackson.databind.JsonNode), - jvm.org.apache.flink.formats.csv.CsvToRowDataConverters(False).createRowConverter( - _to_java_data_type(schema._data_type).getLogicalType(), True), - jvm.org.apache.flink.table.runtime.typeutils.InternalTypeInfo.of( - _to_java_data_type(schema._data_type).getLogicalType()), - False - ]) - ) + _to_java_data_type(schema._data_type) + ) return CsvReaderFormat(j_csv_format) diff --git a/flink-python/src/main/java/org/apache/flink/formats/csv/CsvReaderFormatFactory.java b/flink-python/src/main/java/org/apache/flink/formats/csv/CsvReaderFormatFactory.java new file mode 100644 index 00000000000..f5c845a6190 --- /dev/null +++ b/flink-python/src/main/java/org/apache/flink/formats/csv/CsvReaderFormatFactory.java @@ -0,0 +1,45 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.flink.formats.csv; + +import org.apache.flink.table.runtime.typeutils.InternalTypeInfo; +import org.apache.flink.table.types.DataType; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.table.types.logical.utils.LogicalTypeUtils; +import org.apache.flink.util.Preconditions; + +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.JsonNode; +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.dataformat.csv.CsvMapper; +import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.dataformat.csv.CsvSchema; + +/** Util for creating a {@link CsvReaderFormat}. */ +public class CsvReaderFormatFactory { + public static CsvReaderFormat createCsvReaderFormat(CsvSchema schema, DataType dataType) { + Preconditions.checkArgument(dataType.getLogicalType() instanceof RowType); + + return new CsvReaderFormat( + () -> new CsvMapper(), + ignored -> schema, + JsonNode.class, + new CsvToRowDataConverters(false) + .createRowConverter( + LogicalTypeUtils.toRowType(dataType.getLogicalType()), true), + InternalTypeInfo.of(dataType.getLogicalType()), + false); + } +}