-
Notifications
You must be signed in to change notification settings - Fork 555
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
fix(sink): switch remote sink chrono types to java.time.* #14413
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 the fix!
Also spotted some other references to the legacy classes:
connector-node/risingwave-connector-service/src/main/java/com/risingwave/connector/JsonDeserializer.java
connector-node/risingwave-sink-mock-flink/risingwave-sink-mock-flink-runtime/src/main/java/com/risingwave/mock/flink/runtime/RowDataImpl.java
connector-node/risingwave-sink-deltalake/src/main/java/com/risingwave/connector/DeltaLakeSink.java
connector-node/risingwave-source-cdc/src/main/java/com/risingwave/connector/cdc/debezium/converters/DatetimeTypeConverter.java
After cleaning these up, is it possible to add a static analysis rule to disallow usage of them? We have this in rust:
Lines 1 to 2 in df1b9de
disallowed-methods = [ | |
{ path = "std::iter::Iterator::zip", reason = "Please use `zip_eq_fast` if it's available. Otherwise use `zip_eq_debug`" }, |
case TIMESTAMP: | ||
return ((LocalDateTime) value).toInstant(ZoneOffset.UTC); |
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.
https://cassandra.apache.org/doc/stable/cassandra/cql/types.html#timestamps
Similar to ClickHouse, there is no naive datetime type in Cassandra. And when interpreting one without timezone:
The time zone may be omitted if desired ('2011-02-03 04:05:00'), and if so, the date will be interpreted as being in the time zone under which the coordinating Cassandra node is configured. There are however difficulties inherent in relying on the time zone configuration being as expected, so it is recommended that the time zone always be specified for timestamps when feasible.
To be fully correct, we either need to use the Cassandra coordinator node time zone (rather than hardcoding to UTC here), or reject sinking and require user to AT TIME ZONE 'xxx'
to get a timestamptz
before sink.
...risingwave-connector-test/src/test/java/com/risingwave/connector/sink/jdbc/JDBCSinkTest.java
Outdated
Show resolved
Hide resolved
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.
Would you mind adding e2e test to cover the case of #14198 in e2e_test/sink/remote/jdbc.load.slt
?
...risingwave-connector-test/src/test/java/com/risingwave/connector/sink/jdbc/JDBCSinkTest.java
Outdated
Show resolved
Hide resolved
...risingwave-connector-test/src/test/java/com/risingwave/connector/sink/jdbc/JDBCSinkTest.java
Outdated
Show resolved
Hide resolved
It appears that mysql jdbc will convert it to the default timezone? Maybe we can cast |
It is better to set the session/connection timezone to UTC when interacting with MySQL (and other systems as well). Local times are problematic and shall be avoided whenever possible. https://dev.mysql.com/doc/connector-j/en/connector-j-usagenotes-known-issues-limitations.html
|
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.
Rest LGTM! Thanks for the PR
@@ -3,17 +3,17 @@ | |||
query I | |||
select * from t_remote_0 order by id; | |||
---- | |||
1 Alex 28208 281620391 4986480304337356659 28162.0391 2.03 28162.0391 2023-03-20 10:18:30 |
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.
Any reason to change from a normal datetime to a strange one?
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.
private static LocalDate castDate(Object value) { | ||
try { | ||
Long days = castLong(value) - 1; | ||
return Date.valueOf(LocalDate.of(1, 1, 1).plusDays(days)); | ||
Long days = castLong(value); | ||
return LocalDate.ofEpochDay(days); |
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.
- old:
1
->0001-01-01
- new:
1
->1970-01-02
Look like related to #13486. Was the fix incomplete until this change also merged?
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.
Was the fix incomplete until this change also merged?
No.
It seems that there are no remote sinks using JsonDeserializer
currently? So I change the value to the number of days since epoch.
...de/risingwave-connector-service/src/main/java/com/risingwave/connector/JsonDeserializer.java
Outdated
Show resolved
Hide resolved
values = ((Timestamp) row.get(i)).getTime(); | ||
} else if (row.get(i) instanceof java.sql.Date) { | ||
if (row.get(i) instanceof LocalDateTime) { | ||
values = | ||
ChronoUnit.DAYS.between( | ||
LocalDate.ofEpochDay(0), | ||
((java.sql.Date) row.get(i)).toLocalDate()); | ||
((LocalDateTime) row.get(i)) | ||
.toInstant(ZoneOffset.UTC) | ||
.toEpochMilli(); |
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.
cc @xxhZs regarding #13420 #13677 I am not sure whether we need to test for all 4 LocalDate
/LocalTime
/LocalDateTime
/OffsetDateTime
out of Object
...
https://github.com/delta-io/delta/blob/master/PROTOCOL.md#timestamp-without-timezone-timestampntz
Also note that:
- RIsingWave
timestamp
is DeltaLakeTimestampNtz
(isAdjustedToUTC=false
) - RisingWave
timestamptz
is DeltaLakeTimestamp
(isAdjustedToUTC=true
)
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.
I figured this out: (Still possible to use diff --git a/java/tools/maven/checkstyle.xml b/java/tools/maven/checkstyle.xml
index 4bd0d510e0..f681633272 100644
--- a/java/tools/maven/checkstyle.xml
+++ b/java/tools/maven/checkstyle.xml
@@ -191,10 +191,24 @@ This file is based on the checkstyle file of Apache Beam.
<property name="illegalPkgs" value="io.netty"/>
<message key="import.illegal" value="{0}; Use flink-shaded-netty instead."/>
</module>
+ <!--
<module name="IllegalImport">
<property name="illegalPkgs" value="com.google.common"/>
<message key="import.illegal" value="{0}; Use flink-shaded-guava instead."/>
</module>
+ -->
+ <module name="IllegalImport">
+ <property name="illegalClasses" value="java.util.Date, java.sql.Date, java.sql.Timestamp, java.sql.Time"/>
+ <message key="import.illegal" value="{0}; Use java.time.* instead."/>
+ </module>
+ <module name="IllegalImport">
+ <property name="illegalClasses" value="java.text.DateFormat, java.text.SimpleDateFormat"/>
+ <message key="import.illegal" value="{0}; Use java.time.* instead."/>
+ </module>
+ <module name="IllegalImport">
+ <property name="illegalClasses" value="java.util.Calendar, java.util.GregorianCalendar"/>
+ <message key="import.illegal" value="{0}; Use java.time.* instead."/>
+ </module>
<module name="RedundantModifier">
<!-- Checks for redundant modifiers on various symbol definitions.
diff --git a/java/tools/maven/suppressions.xml b/java/tools/maven/suppressions.xml
index 5e19f59eae..11b4774359 100644
--- a/java/tools/maven/suppressions.xml
+++ b/java/tools/maven/suppressions.xml
@@ -30,5 +30,4 @@ under the License.
<suppress files=".*" checks="javadoc"/>
<suppress files=".*" checks="AvoidStarImport"/>
- <suppress files=".*" checks="IllegalImport"/>
</suppressions>
\ No newline at end of file |
f6892e9
Co-authored-by: xiangjinwu <17769960+xiangjinwu@users.noreply.github.com>
I hereby agree to the terms of the RisingWave Labs, Inc. Contributor License Agreement.
What's changed and what's your intention?
resolve #13684 and fix #14198
Checklist
./risedev check
(or alias,./risedev c
)Documentation
Release note
cassandra-sink
:Before:
timestamp
is allowed to sink intotimestamp
type in cassandra.After:
timestamp
cannot be sinked and has to be converted to timestamptz (via AT TIME ZONE) inside RisingWave before sinking out.