Skip to content

Commit

Permalink
bug fix
Browse files Browse the repository at this point in the history
  • Loading branch information
@pyscala
pyscala committed Nov 12, 2020
1 parent dc0a2a2 commit 63f6644
Show file tree
Hide file tree
Showing 2 changed files with 36 additions and 46 deletions.
1 change: 1 addition & 0 deletions ...-formats/flink-csv/src/main/java/org/apache/flink/formats/csv/CsvToRowDataConverters.java
View file
Expand Up @@ -233,6 +233,7 @@ private CsvToRowDataConverter convertToTime(TimeType timeType) {
return jsonNode -> {
LocalTime localTime = LocalTime.parse(jsonNode.asText());
int mills = (int) (localTime.toNanoOfDay() / 1000_000L);
// this is for rounding off values out of precision
if (precision == 2) {
mills = mills / 10 * 10;
} else if (precision == 1) {
Expand Down
81 changes: 35 additions & 46 deletions ...rmats/flink-csv/src/test/java/org/apache/flink/formats/csv/CsvRowDataSerDeSchemaTest.java
View file
Expand Up @@ -108,6 +108,22 @@ public void testSerializeDeserialize() throws Exception {
BYTES(),
"awML",
new byte[] {107, 3, 11});
testNullableField(
TIME(3),
"12:12:12.232",
LocalTime.parse("12:12:12.232"));
testNullableField(
TIME(2),
"12:12:12.23",
LocalTime.parse("12:12:12.23"));
testNullableField(
TIME(1),
"12:12:12.2",
LocalTime.parse("12:12:12.2"));
testNullableField(
TIME(0),
"12:12:12",
LocalTime.parse("12:12:12"));
}

@Test
Expand All @@ -125,13 +141,13 @@ public void testSerializeDeserializeCustomizedProperties() throws Exception {
.setArrayElementDelimiter(":")
.setFieldDelimiter(';');

testField(STRING(), "123*'4**", "123'4*", deserConfig, ";");
testField(STRING(), "'123''4**'", "123'4*", serConfig, deserConfig, ";");
testField(STRING(), "'a;b*'c'", "a;b'c", deserConfig, ";");
testFieldDeserialization(STRING(), "123*'4**", "123'4*", deserConfig, ";");
testField(STRING(), "'123''4**'", "'123''4**'", serConfig, deserConfig, ";");
testFieldDeserialization(STRING(), "'a;b*'c'", "a;b'c", deserConfig, ";");
testField(STRING(), "'a;b''c'", "a;b'c", serConfig, deserConfig, ";");
testField(INT(), " 12 ", 12, deserConfig, ";");
testFieldDeserialization(INT(), " 12 ", 12, deserConfig, ";");
testField(INT(), "12", 12, serConfig, deserConfig, ";");
testField(
testFieldDeserialization(
ROW(FIELD("f0", STRING()), FIELD("f1", STRING())),
"1:hello", Row.of("1", "hello"),
deserConfig,
Expand All @@ -151,21 +167,22 @@ public void testSerializeDeserializeCustomizedProperties() throws Exception {
deserConfig,
";");
testField(STRING(), "null", "null", serConfig, deserConfig, ";"); // string because null literal has not been set
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 , ";");
testFieldDeserialization(TIME(3), "12:12:12.232", LocalTime.parse("12:12:12.232"), deserConfig, ";");
testFieldDeserialization(TIME(3), "12:12:12.232342", LocalTime.parse("12:12:12.232"), deserConfig, ";");
testFieldDeserialization(TIME(3), "12:12:12.23", LocalTime.parse("12:12:12.23"), deserConfig, ";");
testFieldDeserialization(TIME(2), "12:12:12.23", LocalTime.parse("12:12:12.23"), deserConfig, ";");
testFieldDeserialization(TIME(2), "12:12:12.232312", LocalTime.parse("12:12:12.23"), deserConfig, ";");
testFieldDeserialization(TIME(2), "12:12:12.2", LocalTime.parse("12:12:12.2"), deserConfig, ";");
testFieldDeserialization(TIME(1), "12:12:12.2", LocalTime.parse("12:12:12.2"), deserConfig, ";");
testFieldDeserialization(TIME(1), "12:12:12.2235", LocalTime.parse("12:12:12.2"), deserConfig, ";");
testFieldDeserialization(TIME(1), "12:12:12", LocalTime.parse("12:12:12"), deserConfig, ";");
testFieldDeserialization(TIME(0), "12:12:12", LocalTime.parse("12:12:12"), deserConfig, ";");
testFieldDeserialization(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);
try {
testField(TIME(0), "12:12:12.45", LocalTime.parse("12:12:12") , deserConfig , ";");
testFieldDeserialization(TIME(precision), "12:12:12.45", LocalTime.parse("12:12:12"), deserConfig, ";");
fail();
} catch (Exception e) {
assertEquals(expectedMessage, e.getMessage());
}
Expand Down Expand Up @@ -339,36 +356,8 @@ private void testField(
assertEquals(expectedCsv, new String(serializedRow));
}

private void testFieldForTime(
DataType fieldType,
String csvValue,
String expectValue,
Consumer<CsvRowDataSerializationSchema.Builder> serializationConfig,
Consumer<CsvRowDataDeserializationSchema.Builder> deserializationConfig,
String fieldDelimiter) throws Exception {
RowType rowType = (RowType) ROW(
FIELD("f0", STRING()),
FIELD("f1", fieldType),
FIELD("f2", STRING())
).getLogicalType();
String inputCsv = "BEGIN" + fieldDelimiter + csvValue + fieldDelimiter + "END\n";
String expectedCsv = "BEGIN" + fieldDelimiter + expectValue + fieldDelimiter + "END\n";

// deserialization
CsvRowDataDeserializationSchema.Builder deserSchemaBuilder =
new CsvRowDataDeserializationSchema.Builder(rowType, InternalTypeInfo.of(rowType));
deserializationConfig.accept(deserSchemaBuilder);
RowData deserializedRow = deserialize(deserSchemaBuilder, inputCsv);

// serialization
CsvRowDataSerializationSchema.Builder serSchemaBuilder = new CsvRowDataSerializationSchema.Builder(rowType);
serializationConfig.accept(serSchemaBuilder);
byte[] serializedRow = serialize(serSchemaBuilder, deserializedRow);
assertEquals(expectedCsv, new String(serializedRow));
}

@SuppressWarnings("unchecked")
private void testField(
private void testFieldDeserialization(
DataType fieldType,
String csvValue,
Object value,
Expand Down

0 comments on commit 63f6644

Please sign in to comment.