You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "aiwenmo (via GitHub)" <gi...@apache.org> on 2024/12/26 12:48:58 UTC
[PR] [FLINK-36865][cdc] Provide UNIX_TIMESTAMP series functions in YAML pipeline [flink-cdc]
aiwenmo opened a new pull request, #3819:
URL: https://github.com/apache/flink-cdc/pull/3819
(no comment)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36865][cdc] Provide UNIX_TIMESTAMP series functions in YAML pipeline [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on code in PR #3819:
URL: https://github.com/apache/flink-cdc/pull/3819#discussion_r1900682875
##########
flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorTest.java:
##########
@@ -780,6 +795,76 @@ void testTimestampTransform() throws Exception {
transformFunctionEventEventOperatorTestHarness.close();
}
+ @Test
+ void testUnixTimestampTransform() throws Exception {
+ PostTransformOperator transform =
+ PostTransformOperator.newBuilder()
+ .addTransform(
+ UNIX_TIMESTAMP_TABLEID.identifier(),
+ "col1, FROM_UNIXTIME(44) as from_unixtime,"
Review Comment:
One suit of test case may not be sufficient. Still suggest something like https://github.com/apache/flink-cdc/pull/3698#discussion_r1899936303
Also, separating normal cases and malformed strings would be nice.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36865][cdc] Provide UNIX_TIMESTAMP series functions in YAML pipeline [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on code in PR #3819:
URL: https://github.com/apache/flink-cdc/pull/3819#discussion_r1900670113
##########
flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/DateTimeUtils.java:
##########
@@ -128,6 +138,61 @@ private static int ymdToJulian(int year, int month, int day) {
return day + (153 * m + 2) / 5 + 365 * y + y / 4 - y / 100 + y / 400 - 32045;
}
+ // --------------------------------------------------------------------------------------------
+ // UNIX TIME
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Convert unix timestamp (seconds since '1970-01-01 00:00:00' UTC) to datetime string in the
+ * "yyyy-MM-dd HH:mm:ss" format.
+ */
+ public static String formatUnixTimestamp(long unixtime, TimeZone tz) {
+ return formatUnixTimestamp(unixtime, TIMESTAMP_FORMAT_STRING, tz);
+ }
+
+ /**
+ * Convert unix timestamp (seconds since '1970-01-01 00:00:00' UTC) to datetime string in the
+ * given format.
+ */
+ public static String formatUnixTimestamp(long unixtime, String format, TimeZone tz) {
+ SimpleDateFormat formatter = FORMATTER_CACHE.get(format);
+ formatter.setTimeZone(tz);
+ Date date = new Date(unixtime * 1000);
+ try {
+ return formatter.format(date);
+ } catch (Exception e) {
+ LOG.error("Exception when formatting.", e);
+ return null;
+ }
+ }
+
+ /** Returns the value of the timestamp to seconds since '1970-01-01 00:00:00' UTC. */
+ public static long unixTimestamp(long ts) {
+ return ts / 1000;
+ }
Review Comment:
This method seems meaningless and undocumented.
##########
flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/DateTimeUtils.java:
##########
@@ -128,6 +138,61 @@ private static int ymdToJulian(int year, int month, int day) {
return day + (153 * m + 2) / 5 + 365 * y + y / 4 - y / 100 + y / 400 - 32045;
}
+ // --------------------------------------------------------------------------------------------
+ // UNIX TIME
+ // --------------------------------------------------------------------------------------------
+
+ /**
+ * Convert unix timestamp (seconds since '1970-01-01 00:00:00' UTC) to datetime string in the
+ * "yyyy-MM-dd HH:mm:ss" format.
+ */
+ public static String formatUnixTimestamp(long unixtime, TimeZone tz) {
+ return formatUnixTimestamp(unixtime, TIMESTAMP_FORMAT_STRING, tz);
+ }
+
+ /**
+ * Convert unix timestamp (seconds since '1970-01-01 00:00:00' UTC) to datetime string in the
+ * given format.
+ */
+ public static String formatUnixTimestamp(long unixtime, String format, TimeZone tz) {
Review Comment:
Call it `unixTime` / `timeZone` in arguments.
##########
docs/content.zh/docs/core-concept/transform.md:
##########
@@ -160,6 +160,9 @@ Flink CDC uses [Calcite](https://calcite.apache.org/) to parse expressions and [
| TIMESTAMPDIFF(timepointunit, timepoint1, timepoint2) | timestampDiff(timepointunit, timepoint1, timepoint2) | Returns the (signed) number of timepointunit between timepoint1 and timepoint2. The unit for the interval is given by the first argument, which should be one of the following values: SECOND, MINUTE, HOUR, DAY, MONTH, or YEAR. |
| TO_DATE(string1[, string2]) | toDate(string1[, string2]) | Converts a date string string1 with format string2 (by default 'yyyy-MM-dd') to a date. |
| TO_TIMESTAMP(string1[, string2]) | toTimestamp(string1[, string2]) | Converts date time string string1 with format string2 (by default: 'yyyy-MM-dd HH:mm:ss') to a timestamp, without time zone. |
+| FROM_UNIXTIME(numeric[, string]) | fromUnixtime(NUMERIC[, STRING]) | Returns a representation of the numeric argument as a value in string format (default is ‘yyyy-MM-dd HH:mm:ss’). numeric is an internal timestamp value representing seconds since ‘1970-01-01 00:00:00’ UTC, such as produced by the UNIX_TIMESTAMP() function. The return value is expressed in the session time zone (specified in TableConfig). E.g., FROM_UNIXTIME(44) returns ‘1970-01-01 00:00:44’ if in UTC time zone, but returns ‘1970-01-01 09:00:44’ if in ‘Asia/Tokyo’ time zone. |
+| UNIX_TIMESTAMP() | unixTimestamp() | Gets current Unix timestamp in seconds. This function is not deterministic which means the value would be recalculated for each record. |
+| UNIX_TIMESTAMP(string1[, string2]) | unixTimestamp(STRING1[, STRING2]) | Converts a date time string string1 with format string2 (by default: yyyy-MM-dd HH:mm:ss if not specified) to Unix timestamp (in seconds), using the specified timezone in table config.If a time zone is specified in the date time string and parsed by UTC+X format such as “yyyy-MM-dd HH:mm:ss.SSS X”, this function will use the specified timezone in the date time string instead of the timezone in table config. If the date time string can not be parsed, the default value Long.MIN_VALUE(-9223372036854775808) will be returned.|
Review Comment:
```suggestion
| UNIX_TIMESTAMP(string1[, string2]) | unixTimestamp(STRING1[, STRING2]) | Converts a date time string string1 with format string2 (by default: yyyy-MM-dd HH:mm:ss if not specified) to Unix timestamp (in seconds), using the specified timezone in table config.
If a time zone is specified in the date time string and parsed by UTC+X format such as “yyyy-MM-dd HH:mm:ss.SSS X”, this function will use the specified timezone in the date time string instead of the timezone in table config. If the date time string can not be parsed, the default value Long.MIN_VALUE(-9223372036854775808) will be returned.|
```
There should be a newline.
##########
flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/DateTimeUtils.java:
##########
@@ -109,7 +119,7 @@ private static long internalParseTimestampMillis(String dateStr, String format,
} catch (ParseException e) {
LOG.error(
String.format(
- "Exception when parsing datetime string '%s' in format '%s'",
+ "Exception when parsing datetime string '%s' in format '%s', the default value Long.MIN_VALUE(-9223372036854775808) will be returned.",
Review Comment:
`SystemFunctionUtils` (and `DateTimeUtils`) are evaluated inside Janino engine. Will these error messages ever be printed out?
##########
flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorTest.java:
##########
Review Comment:
Consider adding cases in `TransformE2eITCase#testTemporalFunctions`
##########
flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorTest.java:
##########
@@ -780,6 +795,76 @@ void testTimestampTransform() throws Exception {
transformFunctionEventEventOperatorTestHarness.close();
}
+ @Test
+ void testUnixTimestampTransform() throws Exception {
Review Comment:
Please make this test parameterized in multiple timezones.
See MySqlTimezoneITCase for an example.
##########
flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformParserTest.java:
##########
@@ -252,9 +263,9 @@ public void testTranslateFilterToJaninoExpression() {
"TO_DATE(dt, 'yyyy-MM-dd')", "toDate(dt, \"yyyy-MM-dd\", __time_zone__)");
testFilterExpression("TO_TIMESTAMP(dt)", "toTimestamp(dt, __time_zone__)");
testFilterExpression("TIMESTAMP_DIFF('DAY', dt1, dt2)", "timestampDiff(\"DAY\", dt1, dt2)");
- testFilterExpression("IF(a>b,a,b)", "a > b ? a : b");
- testFilterExpression("NULLIF(a,b)", "nullif(a, b)");
- testFilterExpression("COALESCE(a,b,c)", "coalesce(a, b, c)");
+ testFilterExpression("IF(a>b, a, b)", "a > b ? a : b");
+ testFilterExpression("NULLIF(a, b)", "nullif(a, b)");
+ testFilterExpression("COALESCE(a, b, c)", "coalesce(a, b, c)");
Review Comment:
Irrelevant change can be split into another PR.
##########
flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorTest.java:
##########
@@ -780,6 +795,76 @@ void testTimestampTransform() throws Exception {
transformFunctionEventEventOperatorTestHarness.close();
}
+ @Test
+ void testUnixTimestampTransform() throws Exception {
+ PostTransformOperator transform =
+ PostTransformOperator.newBuilder()
+ .addTransform(
+ UNIX_TIMESTAMP_TABLEID.identifier(),
+ "col1, FROM_UNIXTIME(44) as from_unixtime,"
Review Comment:
One suit of test case may not be sufficient. Still suggest something like https://github.com/apache/flink-cdc/pull/3698#discussion_r1899936303
##########
flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorTest.java:
##########
@@ -780,6 +795,76 @@ void testTimestampTransform() throws Exception {
transformFunctionEventEventOperatorTestHarness.close();
}
+ @Test
+ void testUnixTimestampTransform() throws Exception {
+ PostTransformOperator transform =
+ PostTransformOperator.newBuilder()
+ .addTransform(
+ UNIX_TIMESTAMP_TABLEID.identifier(),
+ "col1, FROM_UNIXTIME(44) as from_unixtime,"
+ + " FROM_UNIXTIME(44, 'yyyy/MM/dd HH:mm:ss') as from_unixtime_format,"
+ + " IF(UNIX_TIMESTAMP() = UNIX_TIMESTAMP(DATE_FORMAT(LOCALTIMESTAMP, 'yyyy-MM-dd HH:mm:ss')), 1, 0) as current_unix_timestamp,"
+ + " UNIX_TIMESTAMP('1970-01-01 08:00:01') as unix_timestamp,"
+ + " UNIX_TIMESTAMP('1970-01-01 08:00:01.001 +0800', 'yyyy-MM-dd HH:mm:ss.SSS X') as unix_timestamp_format_tz,"
+ + " UNIX_TIMESTAMP('1970-01-01 08:00:01.001 +0800', 'yyyy-MM-dd HH:mm:ss.SSS') as unix_timestamp_format,"
+ + " UNIX_TIMESTAMP('1970-01-01 08:00:01.001', 'yyyy-MM-dd HH:mm:ss.SSS X') as unix_timestamp_format_error",
+ null)
+ // .addTimezone("UTC")
Review Comment:
Remove this
##########
flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorTest.java:
##########
@@ -780,6 +795,76 @@ void testTimestampTransform() throws Exception {
transformFunctionEventEventOperatorTestHarness.close();
}
+ @Test
+ void testUnixTimestampTransform() throws Exception {
+ PostTransformOperator transform =
+ PostTransformOperator.newBuilder()
+ .addTransform(
+ UNIX_TIMESTAMP_TABLEID.identifier(),
+ "col1, FROM_UNIXTIME(44) as from_unixtime,"
+ + " FROM_UNIXTIME(44, 'yyyy/MM/dd HH:mm:ss') as from_unixtime_format,"
+ + " IF(UNIX_TIMESTAMP() = UNIX_TIMESTAMP(DATE_FORMAT(LOCALTIMESTAMP, 'yyyy-MM-dd HH:mm:ss')), 1, 0) as current_unix_timestamp,"
+ + " UNIX_TIMESTAMP('1970-01-01 08:00:01') as unix_timestamp,"
+ + " UNIX_TIMESTAMP('1970-01-01 08:00:01.001 +0800', 'yyyy-MM-dd HH:mm:ss.SSS X') as unix_timestamp_format_tz,"
+ + " UNIX_TIMESTAMP('1970-01-01 08:00:01.001 +0800', 'yyyy-MM-dd HH:mm:ss.SSS') as unix_timestamp_format,"
+ + " UNIX_TIMESTAMP('1970-01-01 08:00:01.001', 'yyyy-MM-dd HH:mm:ss.SSS X') as unix_timestamp_format_error",
+ null)
+ // .addTimezone("UTC")
+ .addTimezone("Europe/Berlin")
+ .build();
+ RegularEventOperatorTestHarness<PostTransformOperator, Event>
+ transformFunctionEventEventOperatorTestHarness =
+ RegularEventOperatorTestHarness.with(transform, 1);
+ // Initialization
+ transformFunctionEventEventOperatorTestHarness.open();
+ // Create table
+ CreateTableEvent createTableEvent =
+ new CreateTableEvent(UNIX_TIMESTAMP_TABLEID, UNIX_TIMESTAMP_SCHEMA);
+ BinaryRecordDataGenerator recordDataGenerator =
+ new BinaryRecordDataGenerator(((RowType) UNIX_TIMESTAMP_SCHEMA.toRowDataType()));
+ // Insert
+ DataChangeEvent insertEvent =
+ DataChangeEvent.insertEvent(
+ UNIX_TIMESTAMP_TABLEID,
+ recordDataGenerator.generate(
+ new Object[] {
+ new BinaryStringData("1"),
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null
+ }));
Review Comment:
I wonder what's the point to use the same schema in the source and sink? These fields are all `null` in sources anyway, and all test data are hard-encoded in transform rules... Please consider a clearer way to organize test cases in the future.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36865][cdc] Provide UNIX_TIMESTAMP series functions in YAML pipeline [flink-cdc]
Posted by "aiwenmo (via GitHub)" <gi...@apache.org>.
aiwenmo commented on PR #3819:
URL: https://github.com/apache/flink-cdc/pull/3819#issuecomment-2577646924
PTAL @yuxiqian
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36865][cdc] Provide UNIX_TIMESTAMP series functions in YAML pipeline [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on code in PR #3819:
URL: https://github.com/apache/flink-cdc/pull/3819#discussion_r1908120223
##########
flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorTest.java:
##########
@@ -780,6 +814,320 @@ void testTimestampTransform() throws Exception {
transformFunctionEventEventOperatorTestHarness.close();
}
+ @Test
+ void testFromUnixTimeTransform() throws Exception {
+ // In UTC, from_unix_time(0s) ==> 1970-01-01 00:00:00
+ testFromUnixTimeTransformWithTimeZone("UTC", 0L, "1970-01-01 00:00:00");
+ // In UTC, from_unix_time(44s) ==> 1970-01-01 00:00:44
+ testFromUnixTimeTransformWithTimeZone("UTC", 44L, "1970-01-01 00:00:44");
+ // In Berlin, the time zone is +1:00, from_unix_time(44s) ==> 1970-01-01 01:00:44
+ testFromUnixTimeTransformWithTimeZone("Europe/Berlin", 44L, "1970-01-01 01:00:44");
+ // In Shanghai, the time zone is +8:00, from_unix_time(44s) ==> 1970-01-01 08:00:44
+ testFromUnixTimeTransformWithTimeZone("Asia/Shanghai", 44L, "1970-01-01 08:00:44");
+ }
+
+ private void testFromUnixTimeTransformWithTimeZone(
+ String timeZone, Long seconds, String unixTimeStr) throws Exception {
+ PostTransformOperator transform =
+ PostTransformOperator.newBuilder()
+ .addTransform(
+ FROM_UNIX_TIME_TABLEID.identifier(),
+ "col1, FROM_UNIXTIME(seconds) as from_unix_time,"
+ + " FROM_UNIXTIME(seconds, format_str) as from_unix_time_format",
+ null)
+ .addTimezone(timeZone)
+ .build();
+ RegularEventOperatorTestHarness<PostTransformOperator, Event>
+ transformFunctionEventEventOperatorTestHarness =
+ RegularEventOperatorTestHarness.with(transform, 1);
+ // Initialization
+ transformFunctionEventEventOperatorTestHarness.open();
+ // Create table
+ CreateTableEvent createTableEvent =
+ new CreateTableEvent(FROM_UNIX_TIME_TABLEID, FROM_UNIX_TIME_SCHEMA);
+ BinaryRecordDataGenerator recordDataGenerator =
+ new BinaryRecordDataGenerator(((RowType) FROM_UNIX_TIME_SCHEMA.toRowDataType()));
+ BinaryRecordDataGenerator expectedRecordDataGenerator =
+ new BinaryRecordDataGenerator(
+ ((RowType) EXPECTED_FROM_UNIX_TIME_SCHEMA.toRowDataType()));
+ DataChangeEvent insertEvent =
+ DataChangeEvent.insertEvent(
+ FROM_UNIX_TIME_TABLEID,
+ recordDataGenerator.generate(
+ new Object[] {
+ new BinaryStringData("1"),
+ seconds,
+ new BinaryStringData("yyyy-MM-dd HH:mm:ss")
+ }));
+ DataChangeEvent insertEventExpect =
+ DataChangeEvent.insertEvent(
+ FROM_UNIX_TIME_TABLEID,
+ expectedRecordDataGenerator.generate(
+ new Object[] {
+ new BinaryStringData("1"),
+ new BinaryStringData(unixTimeStr),
+ new BinaryStringData(unixTimeStr)
+ }));
+ transform.processElement(new StreamRecord<>(createTableEvent));
+ Assertions.assertThat(
+ transformFunctionEventEventOperatorTestHarness.getOutputRecords().poll())
+ .isEqualTo(
+ new StreamRecord<>(
+ new CreateTableEvent(
+ FROM_UNIX_TIME_TABLEID, EXPECTED_FROM_UNIX_TIME_SCHEMA)));
+ transform.processElement(new StreamRecord<>(insertEvent));
+ Assertions.assertThat(
+ transformFunctionEventEventOperatorTestHarness.getOutputRecords().poll())
+ .isEqualTo(new StreamRecord<>(insertEventExpect));
+ transformFunctionEventEventOperatorTestHarness.close();
+ }
+
+ /*
+ Converts a date time string string1 with format string2 (by default: yyyy-MM-dd HH:mm:ss if not specified) to Unix timestamp (in seconds),
+ using the specified timezone in table config.
+
+ If a time zone is specified in the date time string and parsed by UTC+X format such as “yyyy-MM-dd HH:mm:ss.SSS X”,
+ this function will use the specified timezone in the date time string instead of the timezone in table config. If the date time string can not be parsed,
+ the default value Long.MIN_VALUE(-9223372036854775808) will be returned.
+ */
+ @Test
+ void testUnixTimestampTransformInBerlin() throws Exception {
+ PostTransformOperator transform =
+ PostTransformOperator.newBuilder()
+ .addTransform(
+ UNIX_TIMESTAMP_TABLEID.identifier(),
+ "col1,"
+ + " UNIX_TIMESTAMP(date_time_str) as unix_timestamp,"
+ + " UNIX_TIMESTAMP(date_time_str, unix_timestamp_format) as unix_timestamp_format",
+ null)
+ .addTimezone("Europe/Berlin")
+ .build();
+ RegularEventOperatorTestHarness<PostTransformOperator, Event>
+ transformFunctionEventEventOperatorTestHarness =
+ RegularEventOperatorTestHarness.with(transform, 1);
+ // Initialization
+ transformFunctionEventEventOperatorTestHarness.open();
+ // Create table
+ CreateTableEvent createTableEvent =
+ new CreateTableEvent(UNIX_TIMESTAMP_TABLEID, UNIX_TIMESTAMP_SCHEMA);
+ BinaryRecordDataGenerator recordDataGenerator =
+ new BinaryRecordDataGenerator(((RowType) UNIX_TIMESTAMP_SCHEMA.toRowDataType()));
+ BinaryRecordDataGenerator expectedRecordDataGenerator =
+ new BinaryRecordDataGenerator(
+ ((RowType) EXPECTED_UNIX_TIMESTAMP_SCHEMA.toRowDataType()));
+ transform.processElement(new StreamRecord<>(createTableEvent));
+ Assertions.assertThat(
+ transformFunctionEventEventOperatorTestHarness.getOutputRecords().poll())
+ .isEqualTo(
+ new StreamRecord<>(
+ new CreateTableEvent(
+ UNIX_TIMESTAMP_TABLEID, EXPECTED_UNIX_TIMESTAMP_SCHEMA)));
+
+ // In Berlin, "1970-01-01 08:00:01.001" formated by "yyyy-MM-dd HH:mm:ss.SSS" ==> 25201L
Review Comment:
nit: formated -> formatted
##########
flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorTest.java:
##########
@@ -780,6 +814,320 @@ void testTimestampTransform() throws Exception {
transformFunctionEventEventOperatorTestHarness.close();
}
+ @Test
+ void testFromUnixTimeTransform() throws Exception {
+ // In UTC, from_unix_time(0s) ==> 1970-01-01 00:00:00
+ testFromUnixTimeTransformWithTimeZone("UTC", 0L, "1970-01-01 00:00:00");
+ // In UTC, from_unix_time(44s) ==> 1970-01-01 00:00:44
+ testFromUnixTimeTransformWithTimeZone("UTC", 44L, "1970-01-01 00:00:44");
+ // In Berlin, the time zone is +1:00, from_unix_time(44s) ==> 1970-01-01 01:00:44
+ testFromUnixTimeTransformWithTimeZone("Europe/Berlin", 44L, "1970-01-01 01:00:44");
+ // In Shanghai, the time zone is +8:00, from_unix_time(44s) ==> 1970-01-01 08:00:44
+ testFromUnixTimeTransformWithTimeZone("Asia/Shanghai", 44L, "1970-01-01 08:00:44");
+ }
+
+ private void testFromUnixTimeTransformWithTimeZone(
+ String timeZone, Long seconds, String unixTimeStr) throws Exception {
+ PostTransformOperator transform =
+ PostTransformOperator.newBuilder()
+ .addTransform(
+ FROM_UNIX_TIME_TABLEID.identifier(),
+ "col1, FROM_UNIXTIME(seconds) as from_unix_time,"
+ + " FROM_UNIXTIME(seconds, format_str) as from_unix_time_format",
+ null)
+ .addTimezone(timeZone)
+ .build();
+ RegularEventOperatorTestHarness<PostTransformOperator, Event>
+ transformFunctionEventEventOperatorTestHarness =
+ RegularEventOperatorTestHarness.with(transform, 1);
+ // Initialization
+ transformFunctionEventEventOperatorTestHarness.open();
+ // Create table
+ CreateTableEvent createTableEvent =
+ new CreateTableEvent(FROM_UNIX_TIME_TABLEID, FROM_UNIX_TIME_SCHEMA);
+ BinaryRecordDataGenerator recordDataGenerator =
+ new BinaryRecordDataGenerator(((RowType) FROM_UNIX_TIME_SCHEMA.toRowDataType()));
+ BinaryRecordDataGenerator expectedRecordDataGenerator =
+ new BinaryRecordDataGenerator(
+ ((RowType) EXPECTED_FROM_UNIX_TIME_SCHEMA.toRowDataType()));
+ DataChangeEvent insertEvent =
+ DataChangeEvent.insertEvent(
+ FROM_UNIX_TIME_TABLEID,
+ recordDataGenerator.generate(
+ new Object[] {
+ new BinaryStringData("1"),
+ seconds,
+ new BinaryStringData("yyyy-MM-dd HH:mm:ss")
+ }));
+ DataChangeEvent insertEventExpect =
+ DataChangeEvent.insertEvent(
+ FROM_UNIX_TIME_TABLEID,
+ expectedRecordDataGenerator.generate(
+ new Object[] {
+ new BinaryStringData("1"),
+ new BinaryStringData(unixTimeStr),
+ new BinaryStringData(unixTimeStr)
+ }));
+ transform.processElement(new StreamRecord<>(createTableEvent));
+ Assertions.assertThat(
+ transformFunctionEventEventOperatorTestHarness.getOutputRecords().poll())
+ .isEqualTo(
+ new StreamRecord<>(
+ new CreateTableEvent(
+ FROM_UNIX_TIME_TABLEID, EXPECTED_FROM_UNIX_TIME_SCHEMA)));
+ transform.processElement(new StreamRecord<>(insertEvent));
+ Assertions.assertThat(
+ transformFunctionEventEventOperatorTestHarness.getOutputRecords().poll())
+ .isEqualTo(new StreamRecord<>(insertEventExpect));
+ transformFunctionEventEventOperatorTestHarness.close();
+ }
+
+ /*
+ Converts a date time string string1 with format string2 (by default: yyyy-MM-dd HH:mm:ss if not specified) to Unix timestamp (in seconds),
+ using the specified timezone in table config.
+
+ If a time zone is specified in the date time string and parsed by UTC+X format such as “yyyy-MM-dd HH:mm:ss.SSS X”,
+ this function will use the specified timezone in the date time string instead of the timezone in table config. If the date time string can not be parsed,
+ the default value Long.MIN_VALUE(-9223372036854775808) will be returned.
+ */
+ @Test
+ void testUnixTimestampTransformInBerlin() throws Exception {
+ PostTransformOperator transform =
+ PostTransformOperator.newBuilder()
+ .addTransform(
+ UNIX_TIMESTAMP_TABLEID.identifier(),
+ "col1,"
+ + " UNIX_TIMESTAMP(date_time_str) as unix_timestamp,"
+ + " UNIX_TIMESTAMP(date_time_str, unix_timestamp_format) as unix_timestamp_format",
+ null)
+ .addTimezone("Europe/Berlin")
+ .build();
+ RegularEventOperatorTestHarness<PostTransformOperator, Event>
+ transformFunctionEventEventOperatorTestHarness =
+ RegularEventOperatorTestHarness.with(transform, 1);
+ // Initialization
+ transformFunctionEventEventOperatorTestHarness.open();
+ // Create table
+ CreateTableEvent createTableEvent =
+ new CreateTableEvent(UNIX_TIMESTAMP_TABLEID, UNIX_TIMESTAMP_SCHEMA);
+ BinaryRecordDataGenerator recordDataGenerator =
+ new BinaryRecordDataGenerator(((RowType) UNIX_TIMESTAMP_SCHEMA.toRowDataType()));
+ BinaryRecordDataGenerator expectedRecordDataGenerator =
+ new BinaryRecordDataGenerator(
+ ((RowType) EXPECTED_UNIX_TIMESTAMP_SCHEMA.toRowDataType()));
+ transform.processElement(new StreamRecord<>(createTableEvent));
+ Assertions.assertThat(
+ transformFunctionEventEventOperatorTestHarness.getOutputRecords().poll())
+ .isEqualTo(
+ new StreamRecord<>(
+ new CreateTableEvent(
+ UNIX_TIMESTAMP_TABLEID, EXPECTED_UNIX_TIMESTAMP_SCHEMA)));
+
+ // In Berlin, "1970-01-01 08:00:01.001" formated by "yyyy-MM-dd HH:mm:ss.SSS" ==> 25201L
+ DataChangeEvent insertEvent1 =
+ DataChangeEvent.insertEvent(
+ UNIX_TIMESTAMP_TABLEID,
+ recordDataGenerator.generate(
+ new Object[] {
+ new BinaryStringData("1"),
+ new BinaryStringData("1970-01-01 08:00:01.001"),
+ new BinaryStringData("yyyy-MM-dd HH:mm:ss.SSS")
+ }));
+ DataChangeEvent insertEventExpect1 =
+ DataChangeEvent.insertEvent(
+ UNIX_TIMESTAMP_TABLEID,
+ expectedRecordDataGenerator.generate(
+ new Object[] {new BinaryStringData("1"), 25201L, 25201L}));
+ transform.processElement(new StreamRecord<>(insertEvent1));
+ Assertions.assertThat(
+ transformFunctionEventEventOperatorTestHarness.getOutputRecords().poll())
+ .isEqualTo(new StreamRecord<>(insertEventExpect1));
+
+ // In Berlin, "1970-01-01 08:00:01.001 +0800" formated by "yyyy-MM-dd HH:mm:ss.SSS X" ==> 1L
+ DataChangeEvent insertEvent2 =
+ DataChangeEvent.insertEvent(
+ UNIX_TIMESTAMP_TABLEID,
+ recordDataGenerator.generate(
+ new Object[] {
+ new BinaryStringData("2"),
+ new BinaryStringData("1970-01-01 08:00:01.001 +0800"),
+ new BinaryStringData("yyyy-MM-dd HH:mm:ss.SSS X")
+ }));
+ DataChangeEvent insertEventExpect2 =
+ DataChangeEvent.insertEvent(
+ UNIX_TIMESTAMP_TABLEID,
+ expectedRecordDataGenerator.generate(
+ new Object[] {new BinaryStringData("2"), 25201L, 1L}));
+ transform.processElement(new StreamRecord<>(insertEvent2));
+ Assertions.assertThat(
+ transformFunctionEventEventOperatorTestHarness.getOutputRecords().poll())
+ .isEqualTo(new StreamRecord<>(insertEventExpect2));
+
+ // In Berlin, "1970-01-01 08:00:01.001 +0800" formated by "yyyy-MM-dd HH:mm:ss.SSS" ==>
+ // 25201L
+ DataChangeEvent insertEvent3 =
+ DataChangeEvent.insertEvent(
+ UNIX_TIMESTAMP_TABLEID,
+ recordDataGenerator.generate(
+ new Object[] {
+ new BinaryStringData("3"),
+ new BinaryStringData("1970-01-01 08:00:01.001 +0800"),
+ new BinaryStringData("yyyy-MM-dd HH:mm:ss.SSS")
+ }));
+ DataChangeEvent insertEventExpect3 =
+ DataChangeEvent.insertEvent(
+ UNIX_TIMESTAMP_TABLEID,
+ expectedRecordDataGenerator.generate(
+ new Object[] {new BinaryStringData("3"), 25201L, 25201L}));
+ transform.processElement(new StreamRecord<>(insertEvent3));
+ Assertions.assertThat(
+ transformFunctionEventEventOperatorTestHarness.getOutputRecords().poll())
+ .isEqualTo(new StreamRecord<>(insertEventExpect3));
+
+ // In Berlin, "1970-01-01 08:00:01.001" formated by "yyyy-MM-dd HH:mm:ss.SSS X" ==>
+ // -9223372036854775808L
+ DataChangeEvent insertEvent4 =
+ DataChangeEvent.insertEvent(
+ UNIX_TIMESTAMP_TABLEID,
+ recordDataGenerator.generate(
+ new Object[] {
+ new BinaryStringData("4"),
+ new BinaryStringData("1970-01-01 08:00:01.001"),
+ new BinaryStringData("yyyy-MM-dd HH:mm:ss.SSS X")
+ }));
+ DataChangeEvent insertEventExpect4 =
+ DataChangeEvent.insertEvent(
+ UNIX_TIMESTAMP_TABLEID,
+ expectedRecordDataGenerator.generate(
+ new Object[] {
+ new BinaryStringData("4"), 25201L, -9223372036854775808L
+ }));
+ transform.processElement(new StreamRecord<>(insertEvent4));
+ Assertions.assertThat(
+ transformFunctionEventEventOperatorTestHarness.getOutputRecords().poll())
+ .isEqualTo(new StreamRecord<>(insertEventExpect4));
+ transformFunctionEventEventOperatorTestHarness.close();
+ }
+
+ @Test
+ void testUnixTimestampTransformInShanghai() throws Exception {
+ PostTransformOperator transform =
+ PostTransformOperator.newBuilder()
+ .addTransform(
+ UNIX_TIMESTAMP_TABLEID.identifier(),
+ "col1,"
+ + " UNIX_TIMESTAMP(date_time_str) as unix_timestamp,"
+ + " UNIX_TIMESTAMP(date_time_str, unix_timestamp_format) as unix_timestamp_format",
+ null)
+ .addTimezone("Asia/Shanghai")
+ .build();
+ RegularEventOperatorTestHarness<PostTransformOperator, Event>
+ transformFunctionEventEventOperatorTestHarness =
+ RegularEventOperatorTestHarness.with(transform, 1);
+ // Initialization
+ transformFunctionEventEventOperatorTestHarness.open();
+ // Create table
+ CreateTableEvent createTableEvent =
+ new CreateTableEvent(UNIX_TIMESTAMP_TABLEID, UNIX_TIMESTAMP_SCHEMA);
+ BinaryRecordDataGenerator recordDataGenerator =
+ new BinaryRecordDataGenerator(((RowType) UNIX_TIMESTAMP_SCHEMA.toRowDataType()));
+ BinaryRecordDataGenerator expectedRecordDataGenerator =
+ new BinaryRecordDataGenerator(
+ ((RowType) EXPECTED_UNIX_TIMESTAMP_SCHEMA.toRowDataType()));
+ transform.processElement(new StreamRecord<>(createTableEvent));
+ Assertions.assertThat(
+ transformFunctionEventEventOperatorTestHarness.getOutputRecords().poll())
+ .isEqualTo(
+ new StreamRecord<>(
+ new CreateTableEvent(
+ UNIX_TIMESTAMP_TABLEID, EXPECTED_UNIX_TIMESTAMP_SCHEMA)));
+
+ // In Shanghai, "1970-01-01 08:00:01.001" formated by "yyyy-MM-dd HH:mm:ss.SSS" ==> 1L
+ DataChangeEvent insertEvent1 =
+ DataChangeEvent.insertEvent(
+ UNIX_TIMESTAMP_TABLEID,
+ recordDataGenerator.generate(
+ new Object[] {
+ new BinaryStringData("1"),
+ new BinaryStringData("1970-01-01 08:00:01.001"),
+ new BinaryStringData("yyyy-MM-dd HH:mm:ss.SSS")
+ }));
+ DataChangeEvent insertEventExpect1 =
+ DataChangeEvent.insertEvent(
+ UNIX_TIMESTAMP_TABLEID,
+ expectedRecordDataGenerator.generate(
+ new Object[] {new BinaryStringData("1"), 1L, 1L}));
+ transform.processElement(new StreamRecord<>(insertEvent1));
+ Assertions.assertThat(
+ transformFunctionEventEventOperatorTestHarness.getOutputRecords().poll())
+ .isEqualTo(new StreamRecord<>(insertEventExpect1));
+
+ // In Shanghai, "1970-01-01 08:00:01.001 +0800" formated by "yyyy-MM-dd HH:mm:ss.SSS X" ==>
+ // 1L
Review Comment:
Since Asia/Shanghai is equivalent to +08:00, maybe we can specify another time zone to make our test more robust?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36865][cdc] Provide UNIX_TIMESTAMP series functions in YAML pipeline [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on code in PR #3819:
URL: https://github.com/apache/flink-cdc/pull/3819#discussion_r1900673509
##########
flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/DateTimeUtils.java:
##########
@@ -109,7 +119,7 @@ private static long internalParseTimestampMillis(String dateStr, String format,
} catch (ParseException e) {
LOG.error(
String.format(
- "Exception when parsing datetime string '%s' in format '%s'",
+ "Exception when parsing datetime string '%s' in format '%s', the default value Long.MIN_VALUE(-9223372036854775808) will be returned.",
Review Comment:
`SystemFunctionUtils` (and `DateTimeUtils`) are evaluated inside Janino engine. Will these error messages ever be printed out?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36865][cdc] Provide UNIX_TIMESTAMP series functions in YAML pipeline [flink-cdc]
Posted by "yuxiqian (via GitHub)" <gi...@apache.org>.
yuxiqian commented on code in PR #3819:
URL: https://github.com/apache/flink-cdc/pull/3819#discussion_r1900684276
##########
flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/operators/transform/PostTransformOperatorTest.java:
##########
@@ -780,6 +795,76 @@ void testTimestampTransform() throws Exception {
transformFunctionEventEventOperatorTestHarness.close();
}
+ @Test
+ void testUnixTimestampTransform() throws Exception {
+ PostTransformOperator transform =
+ PostTransformOperator.newBuilder()
+ .addTransform(
+ UNIX_TIMESTAMP_TABLEID.identifier(),
+ "col1, FROM_UNIXTIME(44) as from_unixtime,"
+ + " FROM_UNIXTIME(44, 'yyyy/MM/dd HH:mm:ss') as from_unixtime_format,"
+ + " IF(UNIX_TIMESTAMP() = UNIX_TIMESTAMP(DATE_FORMAT(LOCALTIMESTAMP, 'yyyy-MM-dd HH:mm:ss')), 1, 0) as current_unix_timestamp,"
+ + " UNIX_TIMESTAMP('1970-01-01 08:00:01') as unix_timestamp,"
+ + " UNIX_TIMESTAMP('1970-01-01 08:00:01.001 +0800', 'yyyy-MM-dd HH:mm:ss.SSS X') as unix_timestamp_format_tz,"
+ + " UNIX_TIMESTAMP('1970-01-01 08:00:01.001 +0800', 'yyyy-MM-dd HH:mm:ss.SSS') as unix_timestamp_format,"
+ + " UNIX_TIMESTAMP('1970-01-01 08:00:01.001', 'yyyy-MM-dd HH:mm:ss.SSS X') as unix_timestamp_format_error",
+ null)
+ // .addTimezone("UTC")
+ .addTimezone("Europe/Berlin")
+ .build();
+ RegularEventOperatorTestHarness<PostTransformOperator, Event>
+ transformFunctionEventEventOperatorTestHarness =
+ RegularEventOperatorTestHarness.with(transform, 1);
+ // Initialization
+ transformFunctionEventEventOperatorTestHarness.open();
+ // Create table
+ CreateTableEvent createTableEvent =
+ new CreateTableEvent(UNIX_TIMESTAMP_TABLEID, UNIX_TIMESTAMP_SCHEMA);
+ BinaryRecordDataGenerator recordDataGenerator =
+ new BinaryRecordDataGenerator(((RowType) UNIX_TIMESTAMP_SCHEMA.toRowDataType()));
+ // Insert
+ DataChangeEvent insertEvent =
+ DataChangeEvent.insertEvent(
+ UNIX_TIMESTAMP_TABLEID,
+ recordDataGenerator.generate(
+ new Object[] {
+ new BinaryStringData("1"),
+ null,
+ null,
+ null,
+ null,
+ null,
+ null,
+ null
+ }));
Review Comment:
I wonder what's the point to use the same schema in the source and sink? These fields are all `null` in sources anyway, and all test data are hard-encoded in transform rules.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36865][cdc] Provide UNIX_TIMESTAMP series functions in YAML pipeline [flink-cdc]
Posted by "aiwenmo (via GitHub)" <gi...@apache.org>.
aiwenmo commented on PR #3819:
URL: https://github.com/apache/flink-cdc/pull/3819#issuecomment-2562662899
PTAL @lvyanquan @leonardBang
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [FLINK-36865][cdc] Provide UNIX_TIMESTAMP series functions in YAML pipeline [flink-cdc]
Posted by "leonardBang (via GitHub)" <gi...@apache.org>.
leonardBang merged PR #3819:
URL: https://github.com/apache/flink-cdc/pull/3819
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@flink.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org