You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ja...@apache.org on 2020/05/14 02:43:49 UTC
[flink] branch master updated: [FLINK-17604][csv] Implement format
factory for CSV serialization and deseriazation schema of RowData type
This is an automated email from the ASF dual-hosted git repository.
jark 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 654a637 [FLINK-17604][csv] Implement format factory for CSV serialization and deseriazation schema of RowData type
654a637 is described below
commit 654a637edc154c2bd5cd8d3999be3dfd93d70a1c
Author: yuzhao.cyz <yu...@gmail.com>
AuthorDate: Wed May 13 11:09:32 2020 +0800
[FLINK-17604][csv] Implement format factory for CSV serialization and deseriazation schema of RowData type
This closes #12065
---
flink-formats/flink-csv/pom.xml | 9 +
.../apache/flink/formats/csv/CsvFormatFactory.java | 251 +++++++++++++++++++++
.../org/apache/flink/formats/csv/CsvOptions.java | 95 ++++++++
.../org.apache.flink.table.factories.Factory | 16 ++
.../flink/formats/csv/CsvFormatFactoryTest.java | 240 ++++++++++++++++++++
5 files changed, 611 insertions(+)
diff --git a/flink-formats/flink-csv/pom.xml b/flink-formats/flink-csv/pom.xml
index e0d3101..570ad24 100644
--- a/flink-formats/flink-csv/pom.xml
+++ b/flink-formats/flink-csv/pom.xml
@@ -92,6 +92,15 @@ under the License.
<type>test-jar</type>
<scope>test</scope>
</dependency>
+
+ <!-- CSV RowData format factory testing -->
+ <dependency>
+ <groupId>org.apache.flink</groupId>
+ <artifactId>flink-core</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ <type>test-jar</type>
+ </dependency>
</dependencies>
<profiles>
diff --git a/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvFormatFactory.java b/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvFormatFactory.java
new file mode 100644
index 0000000..99e4ac4
--- /dev/null
+++ b/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvFormatFactory.java
@@ -0,0 +1,251 @@
+/*
+ * 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.serialization.DeserializationSchema;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.api.common.typeinfo.TypeInformation;
+import org.apache.flink.configuration.ConfigOption;
+import org.apache.flink.configuration.ReadableConfig;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.connector.ChangelogMode;
+import org.apache.flink.table.connector.format.ScanFormat;
+import org.apache.flink.table.connector.format.SinkFormat;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.source.ScanTableSource;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.DeserializationFormatFactory;
+import org.apache.flink.table.factories.DynamicTableFactory;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.factories.SerializationFormatFactory;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.RowType;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.Set;
+
+import static org.apache.flink.formats.csv.CsvOptions.ALLOW_COMMENTS;
+import static org.apache.flink.formats.csv.CsvOptions.ARRAY_ELEMENT_DELIMITER;
+import static org.apache.flink.formats.csv.CsvOptions.DISABLE_QUOTE_CHARACTER;
+import static org.apache.flink.formats.csv.CsvOptions.ESCAPE_CHARACTER;
+import static org.apache.flink.formats.csv.CsvOptions.FIELD_DELIMITER;
+import static org.apache.flink.formats.csv.CsvOptions.IGNORE_PARSE_ERRORS;
+import static org.apache.flink.formats.csv.CsvOptions.LINE_DELIMITER;
+import static org.apache.flink.formats.csv.CsvOptions.NULL_LITERAL;
+import static org.apache.flink.formats.csv.CsvOptions.QUOTE_CHARACTER;
+
+/**
+ * Format factory for providing configured instances of CSV to RowData {@link SerializationSchema}
+ * and {@link DeserializationSchema}.
+ */
+public final class CsvFormatFactory implements
+ DeserializationFormatFactory,
+ SerializationFormatFactory {
+
+ public static final String IDENTIFIER = "csv";
+
+ @SuppressWarnings("unchecked")
+ @Override
+ public ScanFormat<DeserializationSchema<RowData>> createScanFormat(
+ DynamicTableFactory.Context context, ReadableConfig formatOptions) {
+ FactoryUtil.validateFactoryOptions(this, formatOptions);
+ validateFormatOptions(formatOptions);
+
+ return new ScanFormat<DeserializationSchema<RowData>>() {
+ @Override
+ public DeserializationSchema<RowData> createScanFormat(
+ ScanTableSource.Context scanContext,
+ DataType producedDataType) {
+ final RowType rowType = (RowType) producedDataType.getLogicalType();
+ final TypeInformation<RowData> rowDataTypeInfo =
+ (TypeInformation<RowData>) scanContext.createTypeInformation(producedDataType);
+ final CsvRowDataDeserializationSchema.Builder schemaBuilder =
+ new CsvRowDataDeserializationSchema.Builder(
+ rowType,
+ rowDataTypeInfo);
+ configureDeserializationSchema(formatOptions, schemaBuilder);
+ return schemaBuilder.build();
+ }
+
+ @Override
+ public ChangelogMode getChangelogMode() {
+ return ChangelogMode.insertOnly();
+ }
+ };
+ }
+
+ @Override
+ public SinkFormat<SerializationSchema<RowData>> createSinkFormat(
+ DynamicTableFactory.Context context,
+ ReadableConfig formatOptions) {
+ FactoryUtil.validateFactoryOptions(this, formatOptions);
+ validateFormatOptions(formatOptions);
+
+ return new SinkFormat<SerializationSchema<RowData>>() {
+ @Override
+ public SerializationSchema<RowData> createSinkFormat(
+ DynamicTableSink.Context context,
+ DataType consumedDataType) {
+ final RowType rowType = (RowType) consumedDataType.getLogicalType();
+ final CsvRowDataSerializationSchema.Builder schemaBuilder =
+ new CsvRowDataSerializationSchema.Builder(rowType);
+ configureSerializationSchema(formatOptions, schemaBuilder);
+ return schemaBuilder.build();
+ }
+
+ @Override
+ public ChangelogMode getChangelogMode() {
+ return ChangelogMode.insertOnly();
+ }
+ };
+ }
+
+ @Override
+ public String factoryIdentifier() {
+ return IDENTIFIER;
+ }
+
+ @Override
+ public Set<ConfigOption<?>> requiredOptions() {
+ return Collections.emptySet();
+ }
+
+ @Override
+ public Set<ConfigOption<?>> optionalOptions() {
+ Set<ConfigOption<?>> options = new HashSet<>();
+ options.add(FIELD_DELIMITER);
+ options.add(LINE_DELIMITER);
+ options.add(DISABLE_QUOTE_CHARACTER);
+ options.add(QUOTE_CHARACTER);
+ options.add(ALLOW_COMMENTS);
+ options.add(IGNORE_PARSE_ERRORS);
+ options.add(ARRAY_ELEMENT_DELIMITER);
+ options.add(ESCAPE_CHARACTER);
+ options.add(NULL_LITERAL);
+ return options;
+ }
+
+ // ------------------------------------------------------------------------
+ // Validation
+ // ------------------------------------------------------------------------
+
+ private static void validateFormatOptions(ReadableConfig tableOptions) {
+ final boolean hasQuoteCharacter = tableOptions.getOptional(QUOTE_CHARACTER).isPresent();
+ final boolean isDisabledQuoteCharacter = tableOptions.get(DISABLE_QUOTE_CHARACTER);
+ if (isDisabledQuoteCharacter && hasQuoteCharacter){
+ throw new ValidationException(
+ "Format cannot define a quote character and disabled quote character at the same time.");
+ }
+ // Validate the option value must be a single char.
+ validateCharacterVal(tableOptions, FIELD_DELIMITER);
+ validateCharacterVal(tableOptions, ARRAY_ELEMENT_DELIMITER);
+ validateCharacterVal(tableOptions, QUOTE_CHARACTER);
+ validateCharacterVal(tableOptions, ESCAPE_CHARACTER);
+
+ tableOptions.getOptional(LINE_DELIMITER).ifPresent(delimiter -> {
+ Set<String> allowedValues = new HashSet<>(Arrays.asList("\r", "\n", "\r\n", ""));
+ if (!allowedValues.contains(delimiter)) {
+ throw new ValidationException(
+ String.format("Invalid value for option '%s.%s'. Supported values are %s, but was: %s",
+ IDENTIFIER,
+ LINE_DELIMITER.key(),
+ "[\\r, \\n, \\r\\n, \"\"]",
+ delimiter));
+ }
+ });
+ }
+
+ /** Validates the option {@code option} value must be a Character. */
+ private static void validateCharacterVal(
+ ReadableConfig tableOptions,
+ ConfigOption<String> option) {
+ if (tableOptions.getOptional(option).isPresent()) {
+ if (tableOptions.get(option).length() != 1) {
+ throw new ValidationException(
+ String.format("Option '%s.%s' must be a string with single character, but was: %s",
+ IDENTIFIER,
+ option.key(),
+ tableOptions.get(option)));
+ }
+ }
+ }
+
+ // ------------------------------------------------------------------------
+ // Utilities
+ // ------------------------------------------------------------------------
+
+ private static void configureDeserializationSchema(
+ ReadableConfig formatOptions,
+ CsvRowDataDeserializationSchema.Builder schemaBuilder) {
+ formatOptions.getOptional(FIELD_DELIMITER)
+ .map(delimiter -> delimiter.charAt(0))
+ .ifPresent(schemaBuilder::setFieldDelimiter);
+
+ formatOptions.getOptional(QUOTE_CHARACTER)
+ .map(quote -> quote.charAt(0))
+ .ifPresent(schemaBuilder::setQuoteCharacter);
+
+ formatOptions.getOptional(ALLOW_COMMENTS)
+ .ifPresent(schemaBuilder::setAllowComments);
+
+ formatOptions.getOptional(IGNORE_PARSE_ERRORS)
+ .ifPresent(schemaBuilder::setIgnoreParseErrors);
+
+ formatOptions.getOptional(ARRAY_ELEMENT_DELIMITER)
+ .ifPresent(schemaBuilder::setArrayElementDelimiter);
+
+ formatOptions.getOptional(ESCAPE_CHARACTER)
+ .map(escape -> escape.charAt(0))
+ .ifPresent(schemaBuilder::setEscapeCharacter);
+
+ formatOptions.getOptional(NULL_LITERAL)
+ .ifPresent(schemaBuilder::setNullLiteral);
+ }
+
+ private static void configureSerializationSchema(
+ ReadableConfig formatOptions,
+ CsvRowDataSerializationSchema.Builder schemaBuilder) {
+ formatOptions.getOptional(FIELD_DELIMITER)
+ .map(delimiter -> delimiter.charAt(0))
+ .ifPresent(schemaBuilder::setFieldDelimiter);
+
+ formatOptions.getOptional(LINE_DELIMITER)
+ .ifPresent(schemaBuilder::setLineDelimiter);
+
+ if (formatOptions.get(DISABLE_QUOTE_CHARACTER)) {
+ schemaBuilder.disableQuoteCharacter();
+ } else {
+ formatOptions.getOptional(QUOTE_CHARACTER)
+ .map(quote -> quote.charAt(0))
+ .ifPresent(schemaBuilder::setQuoteCharacter);
+ }
+
+ formatOptions.getOptional(ARRAY_ELEMENT_DELIMITER)
+ .ifPresent(schemaBuilder::setArrayElementDelimiter);
+
+ formatOptions.getOptional(ESCAPE_CHARACTER)
+ .map(escape -> escape.charAt(0))
+ .ifPresent(schemaBuilder::setEscapeCharacter);
+
+ formatOptions.getOptional(NULL_LITERAL)
+ .ifPresent(schemaBuilder::setNullLiteral);
+ }
+}
diff --git a/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvOptions.java b/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvOptions.java
new file mode 100644
index 0000000..a6f1b2c
--- /dev/null
+++ b/flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvOptions.java
@@ -0,0 +1,95 @@
+/*
+ * 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.configuration.ConfigOption;
+import org.apache.flink.configuration.ConfigOptions;
+
+/** Options for CSV format. */
+public class CsvOptions {
+ private CsvOptions() {}
+
+ // ------------------------------------------------------------------------
+ // Options
+ // ------------------------------------------------------------------------
+
+ public static final ConfigOption<String> FIELD_DELIMITER = ConfigOptions
+ .key("field-delimiter")
+ .stringType()
+ .defaultValue(",")
+ .withDescription("Optional field delimiter character (',' by default)");
+
+ public static final ConfigOption<String> LINE_DELIMITER = ConfigOptions
+ .key("line-delimiter")
+ .stringType()
+ .defaultValue("\n")
+ .withDescription("Optional line delimiter (\"\\n\" by default, otherwise\n"
+ + "\"\\r\" or \"\\r\\n\" are allowed), unicode is supported if\n"
+ + "the delimiter is an invisible special character,\n"
+ + "e.g. U&'\\000D' is the unicode representation of carriage return \"\\r\""
+ + "e.g. U&'\\000A' is the unicode representation of line feed \"\\n\"");
+
+ public static final ConfigOption<Boolean> DISABLE_QUOTE_CHARACTER = ConfigOptions
+ .key("disable-quote-character")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Optional flag to disabled quote character for enclosing field values (false by default)\n"
+ + "if true, quote-character can not be set");
+
+ public static final ConfigOption<String> QUOTE_CHARACTER = ConfigOptions
+ .key("quote-character")
+ .stringType()
+ .defaultValue("\"")
+ .withDescription("Optional quote character for enclosing field values ('\"' by default)");
+
+ public static final ConfigOption<Boolean> ALLOW_COMMENTS = ConfigOptions
+ .key("allow-comments")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Optional flag to ignore comment lines that start with \"#\"\n"
+ + "(disabled by default);\n"
+ + "if enabled, make sure to also ignore parse errors to allow empty rows");
+
+ public static final ConfigOption<Boolean> IGNORE_PARSE_ERRORS = ConfigOptions
+ .key("ignore-parse-errors")
+ .booleanType()
+ .defaultValue(false)
+ .withDescription("Optional flag to skip fields and rows with parse errors instead of failing;\n"
+ + "fields are set to null in case of errors");
+
+ public static final ConfigOption<String> ARRAY_ELEMENT_DELIMITER = ConfigOptions
+ .key("array-element-delimiter")
+ .stringType()
+ .defaultValue(";")
+ .withDescription("Optional array element delimiter string for separating\n"
+ + "array and row element values (\";\" by default)");
+
+ public static final ConfigOption<String> ESCAPE_CHARACTER = ConfigOptions
+ .key("escape-character")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Optional escape character for escaping values (disabled by default)");
+
+ public static final ConfigOption<String> NULL_LITERAL = ConfigOptions
+ .key("null-literal")
+ .stringType()
+ .noDefaultValue()
+ .withDescription("Optional null literal string that is interpreted as a\n"
+ + "null value (disabled by default)");
+}
diff --git a/flink-formats/flink-csv/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory b/flink-formats/flink-csv/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
new file mode 100644
index 0000000..72657bf
--- /dev/null
+++ b/flink-formats/flink-csv/src/main/resources/META-INF/services/org.apache.flink.table.factories.Factory
@@ -0,0 +1,16 @@
+# 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.
+
+org.apache.flink.formats.csv.CsvFormatFactory
diff --git a/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvFormatFactoryTest.java b/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvFormatFactoryTest.java
new file mode 100644
index 0000000..343db00
--- /dev/null
+++ b/flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvFormatFactoryTest.java
@@ -0,0 +1,240 @@
+/*
+ * 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.serialization.DeserializationSchema;
+import org.apache.flink.api.common.serialization.SerializationSchema;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.table.api.DataTypes;
+import org.apache.flink.table.api.TableSchema;
+import org.apache.flink.table.api.ValidationException;
+import org.apache.flink.table.catalog.CatalogTableImpl;
+import org.apache.flink.table.catalog.ObjectIdentifier;
+import org.apache.flink.table.connector.sink.DynamicTableSink;
+import org.apache.flink.table.connector.source.DynamicTableSource;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.factories.FactoryUtil;
+import org.apache.flink.table.factories.TestDynamicTableFactory;
+import org.apache.flink.table.runtime.connector.source.ScanRuntimeProviderContext;
+import org.apache.flink.table.runtime.typeutils.RowDataTypeInfo;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.flink.util.TestLogger;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+
+import java.util.HashMap;
+import java.util.Map;
+import java.util.function.Consumer;
+
+import static org.apache.flink.util.CoreMatchers.containsCause;
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Tests for {@link CsvFormatFactory}.
+ */
+public class CsvFormatFactoryTest extends TestLogger {
+ @Rule
+ public ExpectedException thrown = ExpectedException.none();
+
+ private static final TableSchema SCHEMA = TableSchema.builder()
+ .field("a", DataTypes.STRING())
+ .field("b", DataTypes.INT())
+ .field("c", DataTypes.BOOLEAN())
+ .build();
+
+ private static final RowType ROW_TYPE = (RowType) SCHEMA.toRowDataType().getLogicalType();
+
+ @Test
+ public void testSeDeSchema() {
+ final CsvRowDataDeserializationSchema expectedDeser =
+ new CsvRowDataDeserializationSchema.Builder(ROW_TYPE, new RowDataTypeInfo(ROW_TYPE))
+ .setFieldDelimiter(';')
+ .setQuoteCharacter('\'')
+ .setAllowComments(true)
+ .setIgnoreParseErrors(true)
+ .setArrayElementDelimiter("|")
+ .setEscapeCharacter('\\')
+ .setNullLiteral("n/a")
+ .build();
+
+ final Map<String, String> options = getAllOptions();
+
+ final DynamicTableSource actualSource = createTableSource(options);
+ assert actualSource instanceof TestDynamicTableFactory.DynamicTableSourceMock;
+ TestDynamicTableFactory.DynamicTableSourceMock scanSourceMock =
+ (TestDynamicTableFactory.DynamicTableSourceMock) actualSource;
+
+ DeserializationSchema<RowData> actualDeser = scanSourceMock.sourceValueFormat
+ .createScanFormat(
+ ScanRuntimeProviderContext.INSTANCE,
+ SCHEMA.toRowDataType());
+
+ assertEquals(expectedDeser, actualDeser);
+
+ final CsvRowDataSerializationSchema expectedSer = new CsvRowDataSerializationSchema.Builder(ROW_TYPE)
+ .setFieldDelimiter(';')
+ .setLineDelimiter("\n")
+ .setQuoteCharacter('\'')
+ .setArrayElementDelimiter("|")
+ .setEscapeCharacter('\\')
+ .setNullLiteral("n/a")
+ .build();
+
+ final DynamicTableSink actualSink = createTableSink(options);
+ assert actualSink instanceof TestDynamicTableFactory.DynamicTableSinkMock;
+ TestDynamicTableFactory.DynamicTableSinkMock sinkMock =
+ (TestDynamicTableFactory.DynamicTableSinkMock) actualSink;
+
+ SerializationSchema<RowData> actualSer = sinkMock.sinkValueFormat
+ .createSinkFormat(
+ null,
+ SCHEMA.toRowDataType());
+
+ assertEquals(expectedSer, actualSer);
+ }
+
+ @Test
+ public void testDisableQuoteCharacter() {
+ final Map<String, String> options = getModifiedOptions(opts -> {
+ opts.put("csv.disable-quote-character", "true");
+ opts.remove("csv.quote-character");
+ });
+
+ final CsvRowDataSerializationSchema expectedSer = new CsvRowDataSerializationSchema.Builder(ROW_TYPE)
+ .setFieldDelimiter(';')
+ .setLineDelimiter("\n")
+ .setArrayElementDelimiter("|")
+ .setEscapeCharacter('\\')
+ .setNullLiteral("n/a")
+ .disableQuoteCharacter()
+ .build();
+
+ final DynamicTableSink actualSink = createTableSink(options);
+ assert actualSink instanceof TestDynamicTableFactory.DynamicTableSinkMock;
+ TestDynamicTableFactory.DynamicTableSinkMock sinkMock =
+ (TestDynamicTableFactory.DynamicTableSinkMock) actualSink;
+
+ SerializationSchema<RowData> actualSer = sinkMock.sinkValueFormat
+ .createSinkFormat(
+ null,
+ SCHEMA.toRowDataType());
+
+ assertEquals(expectedSer, actualSer);
+ }
+
+ @Test
+ public void testDisableQuoteCharacterException() {
+ thrown.expect(ValidationException.class);
+ thrown.expect(containsCause(new ValidationException("Format cannot define a quote character and disabled quote character at the same time.")));
+
+ final Map<String, String> options =
+ getModifiedOptions(opts -> opts.put("csv.disable-quote-character", "true"));
+
+ createTableSink(options);
+ }
+
+ @Test
+ public void testInvalidCharacterOption() {
+ thrown.expect(ValidationException.class);
+ thrown.expect(containsCause(new ValidationException("Option 'csv.quote-character' must be "
+ + "a string with single character, but was: abc")));
+
+ final Map<String, String> options =
+ getModifiedOptions(opts -> opts.put("csv.quote-character", "abc"));
+
+ createTableSink(options);
+ }
+
+ @Test
+ public void testInvalidLineDelimiter() {
+ thrown.expect(ValidationException.class);
+ thrown.expect(containsCause(new ValidationException("Invalid value for option 'csv.line-delimiter'. "
+ + "Supported values are [\\r, \\n, \\r\\n, \"\"], but was: abc")));
+
+ final Map<String, String> options =
+ getModifiedOptions(opts -> opts.put("csv.line-delimiter", "abc"));
+
+ createTableSink(options);
+ }
+
+ @Test
+ public void testInvalidIgnoreParseError() {
+ thrown.expect(ValidationException.class);
+ thrown.expect(containsCause(new IllegalArgumentException("Unrecognized option for boolean: abc. "
+ + "Expected either true or false(case insensitive)")));
+
+ final Map<String, String> options =
+ getModifiedOptions(opts -> opts.put("csv.ignore-parse-errors", "abc"));
+
+ createTableSink(options);
+ }
+
+ // ------------------------------------------------------------------------
+ // Utilities
+ // ------------------------------------------------------------------------
+
+ /**
+ * Returns the full options modified by the given consumer {@code optionModifier}.
+ *
+ * @param optionModifier Consumer to modify the options
+ */
+ private Map<String, String> getModifiedOptions(Consumer<Map<String, String>> optionModifier) {
+ Map<String, String> options = getAllOptions();
+ optionModifier.accept(options);
+ return options;
+ }
+
+ private Map<String, String> getAllOptions() {
+ final Map<String, String> options = new HashMap<>();
+ options.put("connector", TestDynamicTableFactory.IDENTIFIER);
+ options.put("target", "MyTarget");
+ options.put("buffer-size", "1000");
+
+ options.put("format", CsvFormatFactory.IDENTIFIER);
+ options.put("csv.field-delimiter", ";");
+ options.put("csv.line-delimiter", "\n");
+ options.put("csv.quote-character", "'");
+ options.put("csv.allow-comments", "true");
+ options.put("csv.ignore-parse-errors", "true");
+ options.put("csv.array-element-delimiter", "|");
+ options.put("csv.escape-character", "\\");
+ options.put("csv.null-literal", "n/a");
+ return options;
+ }
+
+ private static DynamicTableSource createTableSource(Map<String, String> options) {
+ return FactoryUtil.createTableSource(
+ null,
+ ObjectIdentifier.of("default", "default", "t1"),
+ new CatalogTableImpl(SCHEMA, options, "mock source"),
+ new Configuration(),
+ CsvFormatFactoryTest.class.getClassLoader());
+ }
+
+ private static DynamicTableSink createTableSink(Map<String, String> options) {
+ return FactoryUtil.createTableSink(
+ null,
+ ObjectIdentifier.of("default", "default", "t1"),
+ new CatalogTableImpl(SCHEMA, options, "mock sink"),
+ new Configuration(),
+ CsvFormatFactoryTest.class.getClassLoader());
+ }
+}