-
Notifications
You must be signed in to change notification settings - Fork 2.1k
[FLINK-38079] Add Pipeline support for DateType and TimeType #4060
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: master
Are you sure you want to change the base?
Conversation
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Thanks for @proletarians' nice work, just left some trivial comments.
@@ -85,6 +93,37 @@ | |||
*/ | |||
@PublicEvolving | |||
public class SchemaMergingUtils { | |||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We may handle these changes in another PR and focus on supporting DATE
and TIME
formats here
void testCalculatedColumns(ValuesDataSink.SinkApi sinkApi) throws Exception { | ||
@ParameterizedTest(name = "API version: {0}, initializeMode: {1}") | ||
@MethodSource(value = "parameterProvider") | ||
void testCalculatedColumns(ValuesDataSink.SinkApi sinkApi, boolean initializeMode) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
What is initializeMode
? Seems it doesn't affect test cases at all.
@@ -141,6 +143,7 @@ public void testMySql8JsonDataTypesWithUseLegacyJsonFormat() throws Throwable { | |||
|
|||
@Test | |||
void testMysql57TimeDataTypes() throws Throwable { | |||
UniqueDatabase usedDd = fullTypesMySql57Database; |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Please double check if these changes are necessary, and revert these to keep this PR minimum.
@@ -48,6 +48,7 @@ public class DebeziumSchemaDataTypeInference implements SchemaDataTypeInference, | |||
|
|||
private static final long serialVersionUID = 1L; | |||
|
|||
public static final String DEBEZIUM_DATE_SCHEMA_NAME = "io.debezium.time.Date"; |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Seems redundant as it's an alias of io.debezium.time.Date.SCHEMA_NAME
?
if (o instanceof DateData) { | ||
writer.writeDate(pos, (DateData) o); | ||
} else { | ||
writer.writeInt(pos, (int) o); | ||
} | ||
break; |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We may add some comments here, clarifying that the writeInt
paths are kept for compatibility.
5a2b834
to
aa96a77
Compare
Pipeline currently treats DateType and TimeType as plain INT, so date/time columns lose precision (TIME only to milliseconds, DATE range limited) and cannot be used with built-in temporal functions.