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/06/08 02:05:02 UTC
[flink] 02/03: [hotfix][avro] Fix AvroRowSerializationSchema
doesn't support TIMESTAMP 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
commit a479aee9131bebc67c1eb512542eb7c8f1e54233
Author: Jark Wu <ja...@apache.org>
AuthorDate: Thu Jun 4 20:02:13 2020 +0800
[hotfix][avro] Fix AvroRowSerializationSchema doesn't support TIMESTAMP type
---
.../apache/flink/formats/avro/AvroRowSerializationSchema.java | 9 +++++++++
1 file changed, 9 insertions(+)
diff --git a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowSerializationSchema.java b/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowSerializationSchema.java
index d4c73197..30f9754 100644
--- a/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowSerializationSchema.java
+++ b/flink-formats/flink-avro/src/main/java/org/apache/flink/formats/avro/AvroRowSerializationSchema.java
@@ -48,6 +48,9 @@ import java.nio.ByteBuffer;
import java.sql.Date;
import java.sql.Time;
import java.sql.Timestamp;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -242,14 +245,20 @@ public class AvroRowSerializationSchema implements SerializationSchema<Row> {
// check for logical types
if (object instanceof Date) {
return convertFromDate(schema, (Date) object);
+ } else if (object instanceof LocalDate) {
+ return convertFromDate(schema, Date.valueOf((LocalDate) object));
} else if (object instanceof Time) {
return convertFromTime(schema, (Time) object);
+ } else if (object instanceof LocalTime) {
+ return convertFromTime(schema, Time.valueOf((LocalTime) object));
}
return object;
case LONG:
// check for logical type
if (object instanceof Timestamp) {
return convertFromTimestamp(schema, (Timestamp) object);
+ } else if (object instanceof LocalDateTime) {
+ return convertFromTimestamp(schema, Timestamp.valueOf((LocalDateTime) object));
}
return object;
case FLOAT: