[FLINK-19872][csv] Fix CSV format is unable to parse millisecond for TIME type #13834

pyscala · 2020-10-29T06:42:52Z

What is the purpose of the change

Support to parse millisecond for TIME type in CSV format

Brief change log

Support to parse millisecond for TIME type in CSV format

Verifying this change

This change added tests and can be verified as follows:
testSerializeDeserializeForTime()

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): ( no)
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? (no)

flinkbot · 2020-10-29T06:44:58Z

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 9a45aa4 (Thu Oct 29 06:44:57 UTC 2020)

Warnings:

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

flinkbot · 2020-10-29T07:04:01Z

CI report:

6dd7a16 Azure: SUCCESS

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

pyscala · 2020-10-29T11:21:06Z

CC @wuchong

wuchong

Thanks for the contribution. I left some comments.

wuchong · 2020-10-30T04:36:58Z

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

+				throw new IllegalArgumentException("Csv does not support TIME type " +
+					"with precision: " + precision + ", it only supports precision 0 or 3.");
+			}
+		};


final int precision = timeType.getPrecision(); if (precision > 3) { throw new IllegalArgumentException("CSV format does not support TIME type " + "with precision: " + precision + ", it only supports precision 0 ~ 3."); } // get number of milliseconds of the day return jsonNode -> { LocalTime localTime = LocalTime.parse(jsonNode.asText()); if (precision == 0) { return localTime.toSecondOfDay() * 1000L; } else { return (int) (localTime.toNanoOfDay() / 1000_000L); } };

Throws exception in compile phase instead during runtime.
Besides, we can also support precision 0~3.

I will optimize.

wuchong · 2020-10-30T04:38:14Z

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

@@ -128,7 +128,7 @@ private static RowFieldConverter createRowFieldConverter(LogicalType fieldType)
 			case DATE:
 				return (csvMapper, container, row, pos) -> convertDate(row.getInt(pos), container);
 			case TIME_WITHOUT_TIME_ZONE:
-				return (csvMapper, container, row, pos) -> convertTime(row.getInt(pos), container);
+				return (csvMapper, container, row, pos) -> convertTime(row.getLong(pos), container);


We should still use getInt, because we always use int to represent number of milliseconds of the day in Flink SQL internally. See Javadoc of RowData.

Will overflow when supporting nanosecond precision in the future

That's why we don't support precision > 3 for now.
When the planner supports precision > 3 in the future, we can update csv format to use the higer precision data structure (maybe long or otheres).

wuchong · 2020-10-30T04:44:06Z

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

+			"12:12:12.232",
+			"12:12:12.232",
+			(deserSchema) -> deserSchema.setNullLiteral("null"),
+			(serSchema) -> serSchema.setNullLiteral("null"),


Why not reuse the testNullableField method?

For example, the input is "12:12:12.232", but the precision is 2, then the result should be "12:12:12.23". The existing test named testNullableField must be modified to meet the needs of this test. But the change affects other test cases. So i add a new test methed named testFieldForTime.

For such case, I think you can use the test method only validate the deserialization.

testField( DataType fieldType, String csvValue, Object value, Consumer<CsvRowDataDeserializationSchema.Builder> deserializationConfig, String fieldDelimiter)

Sorry for my carelessness, I will optimize it.

@wuchong
the following code
Row actualRow = (Row) DataFormatConverters.getConverterForDataType(dataType).toExternal(deserializedRow);
in methed
testField( DataType fieldType, String csvValue, Object value, Consumer<CsvRowDataDeserializationSchema.Builder> deserializationConfig, String fieldDelimiter)
will change TIME(n) (0<=n<=3) to TIME(0) .Then the test will fail.
I still use testFieldForTime().

This works in my environment:

testField( TIME(3), "12:12:12.232421", LocalTime.parse("12:12:12.232"), (deserSchema) -> deserSchema.setNullLiteral("null"), ",");

wuchong · 2020-10-30T04:44:43Z

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

+			(serSchema) -> serSchema.setNullLiteral("null"),
+			",");
+		testFieldForTime(
+			TIME(0),


Add more tests for TIME(1), TIME(2).

wuchong · 2020-10-30T04:45:45Z

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

+				",");
+		} catch (Exception e) {
+			actualMessage = e.getCause().getMessage();
+		}
+		assertEquals(expectedMessage, actualMessage);


Suggested change

",");

} catch (Exception e) {

actualMessage = e.getCause().getMessage();

}

assertEquals(expectedMessage, actualMessage);

",");

fail("Exception should be thrown.");

} catch (Exception e) {

assertEquals(expectedMessage, e.getCause().getMessage());

}

wuchong

Thanks @pyscala , I think the pull request is in a good shape now.

I only left some comments to improve the tests.

wuchong · 2020-11-09T09:26:41Z

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

@@ -320,6 +339,34 @@ private void testField(
 		assertEquals(expectedCsv, new String(serializedRow));
 	}

+	private void testFieldForTime(


This is never used.

i will clean up

wuchong · 2020-11-09T09:27:15Z

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

+		testField(TIME(0), "12:12:12", LocalTime.parse("12:12:12") , deserConfig , ";");
+		testField(TIME(0), "12:12:12.45", LocalTime.parse("12:12:12") , deserConfig , ";");
+		int precision = 5;
+		String expectedMessage = String.format("Csv does not support TIME type with precision: %d, it only supports precision 0 ~ 3.", precision);


Why not hard code the exception message?

wuchong · 2020-11-09T09:31:40Z

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

+			LocalTime localTime = LocalTime.parse(jsonNode.asText());
+			int mills = (int) (localTime.toNanoOfDay() / 1000_000L);
+			if (precision == 2) {
+				mills = mills / 10 * 10;


Could you add a comment "this is for rounding off values out of precision"?

wuchong · 2020-11-09T09:32:44Z

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

+		int precision = 5;
+		String expectedMessage = String.format("Csv does not support TIME type with precision: %d, it only supports precision 0 ~ 3.", precision);
+		try {
+			testField(TIME(0), "12:12:12.45", LocalTime.parse("12:12:12") , deserConfig , ";");


This will not fail. Because it is not TIME(5).

You should add fail(); if you want to verify it should fail.

wuchong · 2020-11-09T09:35:30Z

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

+		testField(TIME(3), "12:12:12.232", LocalTime.parse("12:12:12.232") , deserConfig , ";");
+		testField(TIME(3), "12:12:12.232342", LocalTime.parse("12:12:12.232") , deserConfig , ";");
+		testField(TIME(3), "12:12:12.23", LocalTime.parse("12:12:12.23") , deserConfig , ";");
+		testField(TIME(2), "12:12:12.23", LocalTime.parse("12:12:12.23") , deserConfig , ";");
+		testField(TIME(2), "12:12:12.232312", LocalTime.parse("12:12:12.23") , deserConfig , ";");
+		testField(TIME(2), "12:12:12.2", LocalTime.parse("12:12:12.2") , deserConfig , ";");
+		testField(TIME(1), "12:12:12.2", LocalTime.parse("12:12:12.2") , deserConfig , ";");
+		testField(TIME(1), "12:12:12.2235", LocalTime.parse("12:12:12.2") , deserConfig , ";");
+		testField(TIME(1), "12:12:12", LocalTime.parse("12:12:12") , deserConfig , ";");
+		testField(TIME(0), "12:12:12", LocalTime.parse("12:12:12") , deserConfig , ";");
+		testField(TIME(0), "12:12:12.45", LocalTime.parse("12:12:12") , deserConfig , ";");


All of these only test the deserialization. We should also add tests for serialization. I think we can use testNullableField for this purpose.

Besides, we can rename method testField(fieldType, csvValue, value, deserializationConfig, fieldDelimiter) to testFieldSerialization to be more specific.

rename method testField(fieldType, csvValue, value, deserializationConfig, fieldDelimiter) to testFieldSerialization ? or testFieldDeserialization ?

The reason for not using testNullableField can be seen in our previous dialogue.

Rename to testFieldDeserialization.

We only use testNullableField to test conversion without precision loss.

wuchong

Thanks for the contribution @pyscala . LGTM.

…TIME type This closes apache#13834

wuchong · 2020-11-12T07:50:22Z

Merging...

…TIME type This closes apache#13834

rmetzger added review=description? component=Formats component=TableSQL/Ecosystem labels Oct 29, 2020

wuchong requested changes Oct 30, 2020

View reviewed changes

pyscala requested a review from wuchong November 2, 2020 02:38

wuchong reviewed Nov 9, 2020

View reviewed changes

wuchong approved these changes Nov 12, 2020

View reviewed changes

[FLINK-19872][csv] Fix CSV format is unable to parse millisecond for …

6dd7a16

…TIME type This closes apache#13834

wuchong force-pushed the FLINK-19872 branch from 63f6644 to 6dd7a16 Compare November 12, 2020 03:16

wuchong changed the title ~~[FLINK-19872][Formats/Csv] Support to parse millisecond for TIME type in CSV format~~ [FLINK-19872][csv] Fix CSV format is unable to parse millisecond for TIME type Nov 12, 2020

wuchong merged commit e86b409 into apache:master Nov 12, 2020

rkhachatryan pushed a commit to rkhachatryan/flink that referenced this pull request Nov 12, 2020

[FLINK-19872][csv] Fix CSV format is unable to parse millisecond for …

bbf0ea5

…TIME type This closes apache#13834

[FLINK-19872][csv] Fix CSV format is unable to parse millisecond for TIME type #13834

[FLINK-19872][csv] Fix CSV format is unable to parse millisecond for TIME type #13834

Conversation

pyscala commented Oct 29, 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 Oct 29, 2020

Automated Checks

Review Progress

flinkbot commented Oct 29, 2020 • edited

CI report:

pyscala commented Oct 29, 2020

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

pyscala Oct 30, 2020 • edited

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

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

Choose a reason for hiding this comment

pyscala Nov 11, 2020 • edited

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

wuchong left a comment

Choose a reason for hiding this comment

wuchong commented Nov 12, 2020

flinkbot commented Oct 29, 2020 •

edited

pyscala Oct 30, 2020 •

edited

pyscala Nov 11, 2020 •

edited