feat: serde for TIME type #7664
Conversation
| final ConnectDataTranslator connectToKsqlTranslator = new ConnectDataTranslator(Time.SCHEMA); | ||
|
|
||
| // When: | ||
| final Object row = connectToKsqlTranslator.toKsqlRow(Time.SCHEMA, new Date(100L)); |
There was a problem hiding this comment.
This is a Date. Why is the result a Time?
There was a problem hiding this comment.
The Connect time types (Time, Date, Timestamp) store data as java.util.Date. ConnectDataTranslator converts data from Connect to KSQL.
| case TIMESTAMP: | ||
| return handleTimestamp((Timestamp) value); | ||
| default: | ||
| return value; | ||
| } | ||
| } | ||
|
|
||
| private static Integer handleTime(final Time value) { | ||
| // Return milliseconds | ||
| return value == null ? null : (int) value.getTime(); |
There was a problem hiding this comment.
Why isn't using long? I see that getTime() returns long. In fact, I just noticed that we are using INT32 for times and you're returning longValue() in the connector converter. Is there a problem with using long and int in different contexts?
There was a problem hiding this comment.
Avro and Protobuf store Time as int, so we're serializing everything to int for consistency. The longValue() in the connect converter eventually gets converted to a java.sql.Time object, so it doesn't cause problems, but I've updated that and others to use int so that its clearer.
| @@ -163,6 +169,14 @@ private static Object enforceFieldType( | |||
| } | |||
| } | |||
|
|
|||
| private static Object handleInt(final JsonValueContext context) { | |||
| if (context.schema.name() == Time.LOGICAL_NAME) { | |||
| return JsonSerdeUtils.toTime(context.val); | |||
There was a problem hiding this comment.
I see that inside JsonSerdeUtils.toTime is obtained the number as long, but here we sai handleInt. What is the correct type we need to handle?
There was a problem hiding this comment.
see previous comment
| return new Time(object.asLong()); | ||
| return new Time(object.asInt()); | ||
| } | ||
| if (object instanceof TextNode) { | ||
| try { | ||
| return new Time(Long.parseLong(object.textValue())); |
There was a problem hiding this comment.
We should handle overflow/underflow issues in these two cases. First, are we sure that we cannot handle a Time numeric value that is higher/lower than the INT type? If so, then we need to make sure the Json object does not have a huge value that can cause an overflow.
There was a problem hiding this comment.
Added a check for this
| @@ -131,7 +131,7 @@ private static Parser decimalParser(final SqlType sqlType) { | |||
| } | |||
|
|
|||
| private static Parser timeParser(final SqlType sqlType) { | |||
| return v -> new Time(Long.parseLong(v)); | |||
| return v -> new Time(Integer.parseInt(v)); | |||
There was a problem hiding this comment.
Same question about overflow if the type has a value higher than the maximum INT.
There was a problem hiding this comment.
Added a check for this
| @@ -227,7 +227,7 @@ private Object toKsqlValue( | |||
| } | |||
| case INT32: | |||
| if (schema.name() == Time.LOGICAL_NAME) { | |||
| return new java.sql.Time(((Number) convertedValue).longValue()); | |||
| return new java.sql.Time(((Number) convertedValue).intValue()); | |||
There was a problem hiding this comment.
I assumed that being the number INT32, then intValue won't return an overflowed/underflowed value if it has a higher/lower number, right?
There was a problem hiding this comment.
Yeah, Connect does this check.
Description
Implement serde for TIME. TIME gets serialized to an integer number of milliseconds, and deserialized to
java.sql.timeTesting done
Added unit tests
Reviewer checklist