[FLINK-17604] Implement format factory for CSV serialization and #12065

danny0405 · 2020-05-11T03:11:53Z

deseriazation schema of RowData type

What is the purpose of the change

Implements format factory for CSV RowData Se/De schema so that they can use in SQL.

Brief change log

A new CsvFormatFactory, the options are in style of what FLIP-122 has concluded.

Verifying this change

Added UTs.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): no
The public API, i.e., is any changed class annotated with @Public(Evolving): yes
The serializers: no
The runtime per-record code paths (performance sensitive): no
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: no
The S3 file system connector: no

Documentation

Does this pull request introduce a new feature? yes
If yes, how is the feature documented? not documented

flinkbot · 2020-05-11T03:15:34Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit a2a08d5 (Mon May 11 03:15:34 UTC 2020)

Warnings:

1 pom.xml files were touched: Check for build and licensing issues.
No documentation files were touched! Remember to keep the Flink docs up to date!

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

JingsongLi · 2020-05-11T03:15:41Z

flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvFormatFactory.java

+ // Options
+ // ------------------------------------------------------------------------
+
+ private static final ConfigOption<Character> FIELD_DELIMITER = ConfigOptions


Can you extract these config options to a CsvOptions? Because file system also requires these.

Would do, thanks for the suggestions ~

flinkbot · 2020-05-11T03:29:23Z

CI report:

96a476e Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build
@flinkbot run azure re-run the last Azure build

…riazation schema of RowData type

wuchong

Thanks, I left some comments.

wuchong · 2020-05-13T03:50:05Z

flink-formats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvFormatFactoryTest.java

+ @Test
+ public void testInvalidCharacterOption() {
+ thrown.expect(ValidationException.class);
+ thrown.expect(containsCause(new ValidationException("Option [csv.quote-character] must be a Character.")));


Suggested change

thrown.expect(containsCause(new ValidationException("Option [csv.quote-character] must be a Character.")));

thrown.expect(containsCause(new ValidationException("Option 'csv.quote-character' must be a Character.")));

wuchong · 2020-05-13T03:50:32Z

flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvOptions.java

+ public static final ConfigOption<String> ARRAY_ELEMENT_DELIMITER = ConfigOptions
+ .key("array-element-delimiter")
+ .stringType()
+ .noDefaultValue();


default value is ;.

wuchong · 2020-05-13T03:52:36Z

flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvFormatFactory.java

+ DynamicTableFactory.Context context,
+ ReadableConfig formatOptions) {
+ validateFormatOptions(formatOptions);
+ return new SinkFormat<SerializationSchema<RowData>>() {


Add a empty line before this.

wuchong · 2020-05-13T03:56:01Z

flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvOptions.java

+ public static final ConfigOption<String> FIELD_DELIMITER = ConfigOptions
+ .key("field-delimiter")
+ .stringType()
+ .defaultValue(",");


Could you add .withDescription(...) to each option? We may auto generate the documentation in the near future. You can refer description in the documentation.

wuchong · 2020-05-13T03:57:05Z

flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvFormatFactory.java

+ 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 Character.", IDENTIFIER, option.key()));


Suggested change

throw new ValidationException(String.format("Option [%s.%s] must be a Character.", IDENTIFIER, option.key()));

throw new ValidationException(String.format("Option '%s.%s' must be a single character, but was '%s'.", IDENTIFIER, option.key(), tableOptions.get(option)));

Use 'csv.xxx' to be more align with DDL options. And use character instead of Character because this is not in Java language.

wuchong · 2020-05-13T04:00:40Z

flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvFormatFactory.java

+ }
+ validateCharacterVal(tableOptions, FIELD_DELIMITER);
+ validateCharacterVal(tableOptions, QUOTE_CHARACTER);
+ validateCharacterVal(tableOptions, ESCAPE_CHARACTER);


ARRAY_ELEMENT_DELIMITER is also a single character in CsvValidator.

wuchong · 2020-05-13T04:04:10Z

flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvFormatFactory.java

+ // Validation
+ // ------------------------------------------------------------------------
+
+ private static void validateFormatOptions(ReadableConfig tableOptions) {


We should also check LINE_DELIMITER is one of "\r", "\n", "\r\n".

wuchong · 2020-05-13T04:07:14Z

flink-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvFormatFactory.java

+ @SuppressWarnings("unchecked")
+ @Override
+ public ScanFormat<DeserializationSchema<RowData>> createScanFormat(
+ DynamicTableFactory.Context context, ReadableConfig formatOptions) {


Call FactoryUtil.validateFactoryOptions(this, formatOptions); at the beginning. It will checks all the options.
Please also add a test for verify invalid "csv.ignore-parse-errors" = "abc" option.

Added to FactoryUtil.

Added in the factory instead.

wuchong

Thanks for the updating. LGTM.

wuchong · 2020-05-14T02:25:13Z

The failed tests are not related to this one. Will merge it.

wuchong · 2020-05-14T02:36:10Z

The failed python tests is caused by FLINK-17596.

rmetzger added the review=description? label May 11, 2020

JingsongLi reviewed May 11, 2020

View reviewed changes

rmetzger added the component=Formats label May 11, 2020

danny0405 force-pushed the csv-factory branch 2 times, most recently from ef39423 to fb244ad Compare May 12, 2020 07:20

[FLINK-17604] Implement format factory for CSV serialization and dese…

22f3b5f

…riazation schema of RowData type

danny0405 force-pushed the csv-factory branch from fb244ad to 22f3b5f Compare May 13, 2020 03:26

wuchong requested changes May 13, 2020

View reviewed changes

Fix Jark's review comment address

96a476e

danny0405 force-pushed the csv-factory branch from 139ce1d to 96a476e Compare May 13, 2020 10:19

wuchong approved these changes May 13, 2020

View reviewed changes

wuchong closed this in 654a637 May 14, 2020

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-17604] Implement format factory for CSV serialization and #12065

[FLINK-17604] Implement format factory for CSV serialization and #12065

danny0405 commented May 11, 2020

flinkbot commented May 11, 2020

JingsongLi May 11, 2020

danny0405 May 12, 2020

flinkbot commented May 11, 2020 •

edited

Loading

wuchong left a comment

wuchong May 13, 2020

wuchong May 13, 2020

wuchong May 13, 2020

wuchong May 13, 2020

wuchong May 13, 2020

wuchong May 13, 2020

wuchong May 13, 2020

wuchong May 13, 2020

danny0405 May 13, 2020 •

edited

Loading

danny0405 May 13, 2020

wuchong left a comment

wuchong commented May 14, 2020

wuchong commented May 14, 2020

	thrown.expect(containsCause(new ValidationException("Option [csv.quote-character] must be a Character.")));
	thrown.expect(containsCause(new ValidationException("Option 'csv.quote-character' must be a Character.")));

	throw new ValidationException(String.format("Option [%s.%s] must be a Character.", IDENTIFIER, option.key()));
	throw new ValidationException(String.format("Option '%s.%s' must be a single character, but was '%s'.", IDENTIFIER, option.key(), tableOptions.get(option)));

[FLINK-17604] Implement format factory for CSV serialization and #12065

[FLINK-17604] Implement format factory for CSV serialization and #12065

Conversation

danny0405 commented May 11, 2020

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

flinkbot commented May 11, 2020

Automated Checks

Review Progress

Choose a reason for hiding this comment

Choose a reason for hiding this comment

flinkbot commented May 11, 2020 • edited Loading

CI report:

wuchong left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

danny0405 May 13, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wuchong left a comment

Choose a reason for hiding this comment

wuchong commented May 14, 2020

wuchong commented May 14, 2020

flinkbot commented May 11, 2020 •

edited

Loading

danny0405 May 13, 2020 •

edited

Loading