Skip to content

[FLINK-36647][transform] Support Timestampdiff and Timestampadd function in cdc pipeline transform #3698

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.

Sign up for GitHub

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

Merged
leonardBang merged 2 commits into from
Jan 16, 2025
Merged

[FLINK-36647][transform] Support Timestampdiff and Timestampadd function in cdc pipeline transform #3698

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
9f48142
[FLINK-36647][transform] Support Timestampdiff and Timestampadd funct…
aiwenmo Nov 7, 2024
cb5b2a0
[build][minor] Upgrade the max file length from 3k to 4k
leonardBang Jan 14, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions docs/content.zh/docs/core-concept/transform.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -157,6 +157,7 @@ Flink CDC uses [Calcite](https://calcite.apache.org/) to parse expressions and [
| CURRENT_TIMESTAMP | currentTimestamp() | Returns the current SQL timestamp in the local time zone, the return type is TIMESTAMP_LTZ(3). |
| NOW() | now() | Returns the current SQL timestamp in the local time zone, this is a synonym of CURRENT_TIMESTAMP. |
| DATE_FORMAT(timestamp, string) | dateFormat(timestamp, string) | Converts timestamp to a value of string in the format specified by the date format string. The format string is compatible with Java's SimpleDateFormat. |
| TIMESTAMPADD(timeintervalunit, interval, timepoint) | timestampadd(timeintervalunit, interval, timepoint) | Returns the timestamp of timepoint2 after timepoint added interval. 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. |
| 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. |
Expand Down
1 change: 1 addition & 0 deletions docs/content/docs/core-concept/transform.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -157,6 +157,7 @@ Flink CDC uses [Calcite](https://calcite.apache.org/) to parse expressions and [
| CURRENT_TIMESTAMP | currentTimestamp() | Returns the current SQL timestamp in the local time zone, the return type is TIMESTAMP_LTZ(3). |
| NOW() | now() | Returns the current SQL timestamp in the local time zone, this is a synonym of CURRENT_TIMESTAMP. |
| DATE_FORMAT(timestamp, string) | dateFormat(timestamp, string) | Converts timestamp to a value of string in the format specified by the date format string. The format string is compatible with Java's SimpleDateFormat. |
| TIMESTAMPADD(timeintervalunit, interval, timepoint) | timestampadd(timeintervalunit, interval, timepoint) | Returns the timestamp of timepoint2 after timepoint added interval. 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. |
| 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. |
Expand Down
90 changes: 90 additions & 0 deletions flink-cdc-common/src/main/java/org/apache/flink/cdc/common/utils/DateTimeUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@
import java.time.Instant;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.util.Calendar;
import java.util.Date;
import java.util.TimeZone;

Expand Down Expand Up @@ -198,4 +199,93 @@ public static String formatTimestampMillis(long ts, String format, TimeZone time
Date dateTime = new Date(ts);
return formatter.format(dateTime);
}

// --------------------------------------------------------------------------------------------
// Compare
// --------------------------------------------------------------------------------------------

public static Integer timestampDiff(
String timeIntervalUnit,
long fromDate,
String fromTimezone,
long toDate,
String toTimezone) {
Calendar from = Calendar.getInstance(TimeZone.getTimeZone(fromTimezone));
from.setTime(new Date(fromDate));
Calendar to = Calendar.getInstance(TimeZone.getTimeZone(toTimezone));
to.setTime(new Date(toDate));
long second = (to.getTimeInMillis() - from.getTimeInMillis()) / 1000;
switch (timeIntervalUnit) {
case "SECOND":
if (second > Integer.MAX_VALUE) {
return null;
}
return (int) second;
case "MINUTE":
if (second > Integer.MAX_VALUE) {
return null;
}
return (int) second / 60;
case "HOUR":
if (second > Integer.MAX_VALUE) {
return null;
}
return (int) second / 3600;
case "DAY":
if (second > Integer.MAX_VALUE) {
return null;
}
return (int) second / (24 * 3600);
case "MONTH":
return to.get(Calendar.YEAR) * 12
+ to.get(Calendar.MONTH)
- (from.get(Calendar.YEAR) * 12 + from.get(Calendar.MONTH));
case "YEAR":
return to.get(Calendar.YEAR) - from.get(Calendar.YEAR);
default:
throw new RuntimeException(
String.format(
"Unsupported timestamp interval unit %s. Supported units are: SECOND, MINUTE, HOUR, DAY, MONTH, YEAR",
timeIntervalUnit));
}
}

// --------------------------------------------------------------------------------------------
// Add
// --------------------------------------------------------------------------------------------

public static long timestampAdd(
String timeIntervalUnit, int interval, long timePoint, String timezone) {
Calendar calendar = Calendar.getInstance();
calendar.setTimeZone(TimeZone.getTimeZone(timezone));
calendar.setTime(new Date(timePoint));
int field;
switch (timeIntervalUnit) {
case "SECOND":
field = Calendar.SECOND;
break;
case "MINUTE":
field = Calendar.MINUTE;
break;
case "HOUR":
field = Calendar.HOUR;
break;
case "DAY":
field = Calendar.DATE;
break;
case "MONTH":
field = Calendar.MONTH;
break;
case "YEAR":
field = Calendar.YEAR;
break;
default:
throw new RuntimeException(
String.format(
"Unsupported timestamp interval unit %s. Supported units are: SECOND, MINUTE, HOUR, DAY, MONTH, YEAR",
timeIntervalUnit));
}
calendar.add(field, interval);
return calendar.getTimeInMillis();
}
}
168 changes: 105 additions & 63 deletions ...cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/SystemFunctionUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,8 +35,6 @@
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.Arrays;
import java.util.Calendar;
import java.util.Date;
import java.util.TimeZone;
import java.util.UUID;
import java.util.regex.Matcher;
Expand Down Expand Up @@ -130,83 +128,127 @@ public static TimestampData toTimestamp(String str, String format, String timezo
}
}

public static int timestampDiff(
String symbol,
// Be compatible with the existing definition of Function TIMESTAMP_DIFF
public static Integer timestampDiff(
String timeIntervalUnit,
LocalZonedTimestampData fromTimestamp,
LocalZonedTimestampData toTimestamp) {
return timestampDiff(
symbol, fromTimestamp.getEpochMillisecond(), toTimestamp.getEpochMillisecond());
}

public static int timestampDiff(
String symbol, TimestampData fromTimestamp, TimestampData toTimestamp) {
return timestampDiff(symbol, fromTimestamp.getMillisecond(), toTimestamp.getMillisecond());
}

public static int timestampDiff(
String symbol, TimestampData fromTimestamp, LocalZonedTimestampData toTimestamp) {
return timestampDiff(
symbol, fromTimestamp.getMillisecond(), toTimestamp.getEpochMillisecond());
LocalZonedTimestampData toTimestamp,
String timezone) {
if (fromTimestamp == null || toTimestamp == null) {
return null;
}
return DateTimeUtils.timestampDiff(
timeIntervalUnit,
fromTimestamp.getEpochMillisecond(),
timezone,
toTimestamp.getEpochMillisecond(),
timezone);
}

// Be compatible with the existing definition of Function TIMESTAMP_DIFF
public static Integer timestampDiff(
String timeIntervalUnit,
TimestampData fromTimestamp,
TimestampData toTimestamp,
String timezone) {
if (fromTimestamp == null || toTimestamp == null) {
return null;
}
return DateTimeUtils.timestampDiff(
timeIntervalUnit,
fromTimestamp.getMillisecond(),
"UTC",
toTimestamp.getMillisecond(),
"UTC");
}

// Be compatible with the existing definition of Function TIMESTAMP_DIFF
public static Integer timestampDiff(
String timeIntervalUnit,
TimestampData fromTimestamp,
LocalZonedTimestampData toTimestamp,
String timezone) {
if (fromTimestamp == null || toTimestamp == null) {
return null;
}
return DateTimeUtils.timestampDiff(
timeIntervalUnit,
fromTimestamp.getMillisecond(),
"UTC",
toTimestamp.getEpochMillisecond(),
timezone);
}

public static int timestampDiff(
String symbol, LocalZonedTimestampData fromTimestamp, TimestampData toTimestamp) {
return timestampDiff(
symbol, fromTimestamp.getEpochMillisecond(), toTimestamp.getMillisecond());
// Be compatible with the existing definition of Function TIMESTAMP_DIFF
public static Integer timestampDiff(
String timeIntervalUnit,
LocalZonedTimestampData fromTimestamp,
TimestampData toTimestamp,
String timezone) {
if (fromTimestamp == null || toTimestamp == null) {
return null;
}
return DateTimeUtils.timestampDiff(
timeIntervalUnit,
fromTimestamp.getEpochMillisecond(),
timezone,
toTimestamp.getMillisecond(),
"UTC");
}

public static int timestampDiff(
String symbol, ZonedTimestampData fromTimestamp, ZonedTimestampData toTimestamp) {
return timestampDiff(symbol, fromTimestamp.getMillisecond(), toTimestamp.getMillisecond());
public static Integer timestampdiff(
String timeIntervalUnit,
LocalZonedTimestampData fromTimestamp,
LocalZonedTimestampData toTimestamp,
String timezone) {
return timestampDiff(timeIntervalUnit, fromTimestamp, toTimestamp, timezone);
}

public static int timestampDiff(
String symbol, LocalZonedTimestampData fromTimestamp, ZonedTimestampData toTimestamp) {
return timestampDiff(
symbol, fromTimestamp.getEpochMillisecond(), toTimestamp.getMillisecond());
public static Integer timestampdiff(
String timeIntervalUnit,
TimestampData fromTimestamp,
TimestampData toTimestamp,
String timezone) {
return timestampDiff(timeIntervalUnit, fromTimestamp, toTimestamp, timezone);
}

public static int timestampDiff(
String symbol, ZonedTimestampData fromTimestamp, LocalZonedTimestampData toTimestamp) {
return timestampDiff(
symbol, fromTimestamp.getMillisecond(), toTimestamp.getEpochMillisecond());
public static Integer timestampdiff(
String timeIntervalUnit,
TimestampData fromTimestamp,
LocalZonedTimestampData toTimestamp,
String timezone) {
return timestampDiff(timeIntervalUnit, fromTimestamp, toTimestamp, timezone);
}

public static int timestampDiff(
String symbol, TimestampData fromTimestamp, ZonedTimestampData toTimestamp) {
return timestampDiff(symbol, fromTimestamp.getMillisecond(), toTimestamp.getMillisecond());
public static Integer timestampdiff(
String timeIntervalUnit,
LocalZonedTimestampData fromTimestamp,
TimestampData toTimestamp,
String timezone) {
return timestampDiff(timeIntervalUnit, fromTimestamp, toTimestamp, timezone);
}

public static int timestampDiff(
String symbol, ZonedTimestampData fromTimestamp, TimestampData toTimestamp) {
return timestampDiff(symbol, fromTimestamp.getMillisecond(), toTimestamp.getMillisecond());
public static LocalZonedTimestampData timestampadd(
String timeIntervalUnit,
Integer interval,
LocalZonedTimestampData timePoint,
String timezone) {
if (interval == null || timePoint == null) {
return null;
}
return LocalZonedTimestampData.fromEpochMillis(
DateTimeUtils.timestampAdd(
timeIntervalUnit, interval, timePoint.getEpochMillisecond(), timezone));
}

public static int timestampDiff(String symbol, long fromDate, long toDate) {
Calendar from = Calendar.getInstance();
from.setTime(new Date(fromDate));
Calendar to = Calendar.getInstance();
to.setTime(new Date(toDate));
Long second = (to.getTimeInMillis() - from.getTimeInMillis()) / 1000;
switch (symbol) {
case "SECOND":
return second.intValue();
case "MINUTE":
return second.intValue() / 60;
case "HOUR":
return second.intValue() / 3600;
case "DAY":
return second.intValue() / (24 * 3600);
case "MONTH":
return to.get(Calendar.YEAR) * 12
+ to.get(Calendar.MONDAY)
- (from.get(Calendar.YEAR) * 12 + from.get(Calendar.MONDAY));
case "YEAR":
return to.get(Calendar.YEAR) - from.get(Calendar.YEAR);
default:
LOG.error("Unsupported timestamp diff: {}", symbol);
throw new RuntimeException("Unsupported timestamp diff: " + symbol);
public static TimestampData timestampadd(
String timeIntervalUnit, Integer interval, TimestampData timePoint, String timezone) {
if (interval == null || timePoint == null) {
return null;
}
return TimestampData.fromMillis(
DateTimeUtils.timestampAdd(
timeIntervalUnit, interval, timePoint.getMillisecond(), "UTC"));
}

public static boolean betweenAsymmetric(String value, String minValue, String maxValue) {
Expand Down
Loading
Loading