This is an automated email from the ASF dual-hosted git repository.
lvyanquan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
The following commit(s) were added to refs/heads/master by this push:
new ff26ee22f [FLINK-40312] [runtime] Support EXTRACT, date-part
functions, and interval arithmetic in YAML Transform (#4495)
ff26ee22f is described below
commit ff26ee22f7a89aafdb926d23f207c489d930b623
Author: haruki <[email protected]>
AuthorDate: Mon Aug 10 12:56:18 2026 +0800
[FLINK-40312] [runtime] Support EXTRACT, date-part functions, and interval
arithmetic in YAML Transform (#4495)
Co-authored-by: 春栖 <[email protected]>
---
docs/content.zh/docs/core-concept/transform.md | 5 +
docs/content/docs/core-concept/transform.md | 5 +
.../src/test/resources/specs/temporal.yaml | 44 +++++++
.../runtime/functions/impl/TemporalFunctions.java | 94 ++++++++++++++
.../flink/cdc/runtime/parser/JaninoCompiler.java | 126 +++++++++++++++++++
.../parser/metadata/TransformSqlOperatorTable.java | 11 ++
.../functions/impl/TemporalFunctionsTest.java | 138 +++++++++++++++++++++
.../cdc/runtime/parser/TransformParserTest.java | 97 ++++++++++++++-
8 files changed, 516 insertions(+), 4 deletions(-)
diff --git a/docs/content.zh/docs/core-concept/transform.md
b/docs/content.zh/docs/core-concept/transform.md
index 726769254..d6daab547 100644
--- a/docs/content.zh/docs/core-concept/transform.md
+++ b/docs/content.zh/docs/core-concept/transform.md
@@ -216,6 +216,11 @@ Flink CDC 使用 [Calcite](https://calcite.apache.org/)
来解析表达式并且
| CURRENT_DATE | currentDate()
| 返回本地时区的当前 SQL 日期。
|
| CURRENT_TIMESTAMP | currentTimestamp()
| 返回本地时区的当前 SQL 时间戳，返回类型为 TIMESTAMP_LTZ(3)。
|
| NOW() | now()
| 返回本地时区的当前 SQL 时间戳，是 CURRENT_TIMESTAMP 的同义词。
|
+| EXTRACT(timeintervalunit FROM temporal) |
extract(timeintervalunit, temporal) | 从时间值中提取指定部分。支持
YEAR、QUARTER、MONTH、WEEK、DAY、DOY、DOW、HOUR、MINUTE 和 SECOND。
|
+| YEAR / QUARTER / MONTH / WEEK(temporal) |
extract(timeintervalunit, temporal) | 从时间值中提取年、季度、月或 ISO 周。
|
+| DAYOFYEAR / DAYOFMONTH / DAYOFWEEK(temporal) |
extract(timeintervalunit, temporal) |
从时间值中提取一年中的第几天、一月中的第几天或一周中的第几天。DAYOFWEEK 以星期日为 1。
|
+| HOUR / MINUTE / SECOND(temporal) |
extract(timeintervalunit, temporal) | 从时间值中提取小时、分钟或秒。
|
+| temporal +/- INTERVAL 'value' unit | temporalPlusMonths /
temporalPlusMillis | 对 DATE、TIME、TIMESTAMP 和 TIMESTAMP_LTZ
值加减年-月或日-时间隔字面量。支持 `INTERVAL + temporal`，以及 `INTERVAL '1-2' YEAR TO MONTH` 和
`INTERVAL '1 02:03:04' DAY TO SECOND` 等复合间隔。
|
| DATE_FORMAT(timestamp, string) | dateFormat(timestamp,
string) | 将时间戳转换为指定日期格式字符串的值。格式字符串与 Java 的
SimpleDateFormat 兼容。
|
| DATE_FORMAT(date, string) | dateFormat(date,
string) | 将给定日期转换为指定格式字符串的值。格式字符串与 Java 的
SimpleDateFormat 兼容。
|
| DATE_FORMAT(time, string) | dateFormat(time,
string) | 将给定时间转换为指定格式字符串的值。格式字符串与 Java 的
SimpleDateFormat 兼容。
|
diff --git a/docs/content/docs/core-concept/transform.md
b/docs/content/docs/core-concept/transform.md
index f60f4d7c7..28bda0e34 100644
--- a/docs/content/docs/core-concept/transform.md
+++ b/docs/content/docs/core-concept/transform.md
@@ -217,6 +217,11 @@ Logical functions follow SQL three-valued logic for
nullable BOOLEAN values. `AN
| CURRENT_DATE | currentDate()
| Returns the current SQL date in the local time
zone.
[...]
| 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.
[...]
+| EXTRACT(timeintervalunit FROM temporal) |
extract(timeintervalunit, temporal) | Extracts a time interval
unit from a temporal value. Supported units are YEAR, QUARTER, MONTH, WEEK,
DAY, DOY, DOW, HOUR, MINUTE, and SECOND.
[...]
+| YEAR / QUARTER / MONTH / WEEK(temporal) |
extract(timeintervalunit, temporal) | Extracts the year,
quarter, month, or ISO week from a temporal value.
[...]
+| DAYOFYEAR / DAYOFMONTH / DAYOFWEEK(temporal) |
extract(timeintervalunit, temporal) | Extracts the day of
year, day of month, or day of week from a temporal value. DAYOFWEEK uses Sunday
as 1.
[...]
+| HOUR / MINUTE / SECOND(temporal) |
extract(timeintervalunit, temporal) | Extracts the hour,
minute, or second from a temporal value.
[...]
+| temporal +/- INTERVAL 'value' unit | temporalPlusMonths /
temporalPlusMillis | Adds or subtracts a year-month or day-time
interval literal for DATE, TIME, TIMESTAMP, and TIMESTAMP_LTZ values. `INTERVAL
+ temporal` and compound intervals such as `INTERVAL '1-2' YEAR TO MONTH` and
`INTERVAL '1 02:03:04' DAY TO SECOND` are supported.
[...]
| DATE_FORMAT(timestamp, string) | dateFormat(timestamp,
string) | Converts timestamp to a value of string in the
format specified by the format string. The format string is compatible with
Java's SimpleDateFormat.
[...]
| DATE_FORMAT(date, string) | dateFormat(date,
string) | Converts given date to a value of string
in the format specified by the format string. The format string is compatible
with Java's SimpleDateFormat.
[...]
| DATE_FORMAT(time, string) | dateFormat(time,
string) | Converts given time to a value of string
in the format specified by the format string. The format string is compatible
with Java's SimpleDateFormat.
[...]
diff --git a/flink-cdc-composer/src/test/resources/specs/temporal.yaml
b/flink-cdc-composer/src/test/resources/specs/temporal.yaml
index 3e6125296..8420ccaf7 100644
--- a/flink-cdc-composer/src/test/resources/specs/temporal.yaml
+++ b/flink-cdc-composer/src/test/resources/specs/temporal.yaml
@@ -276,3 +276,47 @@
DataChangeEvent{tableId=foo.bar.baz, before=[1970-01-26, 1970-01-27,
1970-01-28, 1970-01-26, 1970-01-27, 1970-01-28, 2001-01-18, 2015-08-09,
2015-08-09], after=[], op=DELETE, meta=()}
DataChangeEvent{tableId=foo.bar.baz, before=[], after=[null, null, null,
null, null, null, null, 2015-08-09, 2015-08-09], op=INSERT, meta=()}
DataChangeEvent{tableId=foo.bar.baz, before=[null, null, null, null, null,
null, null, 2015-08-09, 2015-08-09], after=[], op=DELETE, meta=()}
+- do: EXTRACT and Date Part Functions
+ projection: |-
+ id_
+ YEAR(date_) AS year_
+ QUARTER(date_) AS quarter_
+ MONTH(timestamp_9_) AS month_
+ WEEK(date_) AS week_
+ DAYOFYEAR(date_) AS day_of_year_
+ DAYOFMONTH(date_) AS day_of_month_
+ DAYOFWEEK(date_) AS day_of_week_
+ HOUR(time_0_) AS hour_
+ MINUTE(time_0_) AS minute_
+ SECOND(time_0_) AS second_
+ EXTRACT(DAY FROM timestamp_0_) AS extract_day_
+ EXTRACT(HOUR FROM timestamp_ltz_0_) AS extract_ltz_hour_
+ EXTRACT(YEAR FROM TO_TIMESTAMP_LTZ('2023-12-31 16:30:00.000')) AS
extract_ltz_year_
+ filter: YEAR(date_) >= 2000 OR date_ IS NULL
+ time-zone: Asia/Shanghai
+ expect: |-
+ CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT
NULL 'Identifier',`year_` BIGINT,`quarter_` BIGINT,`month_` BIGINT,`week_`
BIGINT,`day_of_year_` BIGINT,`day_of_month_` BIGINT,`day_of_week_`
BIGINT,`hour_` BIGINT,`minute_` BIGINT,`second_` BIGINT,`extract_day_`
BIGINT,`extract_ltz_hour_` BIGINT,`extract_ltz_year_` BIGINT}, primaryKeys=,
options=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, 2000, 4, 1, 52,
366, 31, 1, 19, 43, 17, 2, 18, 2024], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[1, 2000, 4, 1, 52, 366, 31,
1, 19, 43, 17, 2, 18, 2024], after=[-1, 2001, 1, 1, 1, 1, 1, 2, 12, 34, 45, 9,
16, 2024], op=UPDATE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[-1, 2001, 1, 1, 1, 1, 1, 2,
12, 34, 45, 9, 16, 2024], after=[], op=DELETE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, null, null,
null, null, null, null, null, null, null, null, null, null, 2024], op=INSERT,
meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[0, null, null, null, null,
null, null, null, null, null, null, null, null, 2024], after=[], op=DELETE,
meta=()}
+- do: Interval Literal Arithmetic
+ projection: |-
+ id_
+ date_ + INTERVAL '1' DAY AS date_plus_day_
+ date_ - INTERVAL '1' MONTH AS date_minus_month_
+ time_0_ + INTERVAL '6' HOUR AS time_plus_hour_
+ timestamp_0_ + INTERVAL '1 02:03:04' DAY TO SECOND AS
timestamp_plus_day_time_
+ timestamp_0_ - INTERVAL '1-2' YEAR TO MONTH AS timestamp_minus_year_month_
+ INTERVAL '3' DAY + timestamp_ltz_0_ AS ltz_plus_day_
+ timestamp_ltz_0_ + INTERVAL '1' MONTH AS ltz_plus_month_
+ timestamp_0_ + INTERVAL '1' DAY - INTERVAL '2' HOUR AS nested_interval_
+ filter: timestamp_0_ + INTERVAL '1' DAY > timestamp_0_ OR timestamp_0_ IS
NULL
+ expect: |-
+ CreateTableEvent{tableId=foo.bar.baz, schema=columns={`id_` BIGINT NOT
NULL 'Identifier',`date_plus_day_` DATE,`date_minus_month_`
DATE,`time_plus_hour_` TIME(0),`timestamp_plus_day_time_`
TIMESTAMP(0),`timestamp_minus_year_month_` TIMESTAMP(0),`ltz_plus_day_`
TIMESTAMP_LTZ(0),`ltz_plus_month_` TIMESTAMP_LTZ(0),`nested_interval_`
TIMESTAMP(0)}, primaryKeys=, options=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[1, 2001-01-01,
2000-11-30, 01:43:17, 1970-01-03T12:20:40.789123456,
1968-11-02T10:17:36.789123456, 1970-01-05T10:17:36.789123456,
1970-02-02T10:17:36.789123456, 1970-01-03T08:17:36.789123456], op=INSERT,
meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[1, 2001-01-01, 2000-11-30,
01:43:17, 1970-01-03T12:20:40.789123456, 1968-11-02T10:17:36.789123456,
1970-01-05T10:17:36.789123456, 1970-02-02T10:17:36.789123456,
1970-01-03T08:17:36.789123456], after=[-1, 2001-01-02, 2000-12-01, 18:34:45,
1970-01-10T11:00:40.789723456, 1968-11-09T08:57:36.789723456,
1970-01-12T08:57:36.789723456, 1970-02-09T08:57:36.789723456,
1970-01-10T06:57:36.789723456], op=UPDATE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[-1, 2001-01-02, 2000-12-01,
18:34:45, 1970-01-10T11:00:40.789723456, 1968-11-09T08:57:36.789723456,
1970-01-12T08:57:36.789723456, 1970-02-09T08:57:36.789723456,
1970-01-10T06:57:36.789723456], after=[], op=DELETE, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[], after=[0, null, null,
null, null, null, null, null, null], op=INSERT, meta=()}
+ DataChangeEvent{tableId=foo.bar.baz, before=[0, null, null, null, null,
null, null, null, null], after=[], op=DELETE, meta=()}
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/TemporalFunctions.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/TemporalFunctions.java
index de31a81fc..1deeae0c0 100644
---
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/TemporalFunctions.java
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/functions/impl/TemporalFunctions.java
@@ -32,6 +32,11 @@ import java.time.LocalTime;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
+import java.time.temporal.ChronoField;
+import java.time.temporal.ChronoUnit;
+import java.time.temporal.IsoFields;
+import java.time.temporal.TemporalAccessor;
+import java.time.temporal.TemporalField;
import java.util.TimeZone;
/** Temporal built-in functions. */
@@ -39,6 +44,7 @@ public class TemporalFunctions {
private static final Logger LOG =
LoggerFactory.getLogger(TemporalFunctions.class);
+ private static final long MILLIS_PER_DAY = 24L * 60 * 60 * 1000;
private static final String DEFAULT_DATE_FORMAT = "yyyy-MM-dd";
private static final String DEFAULT_DATETIME_FORMAT = "yyyy-MM-dd
HH:mm:ss";
@@ -71,6 +77,94 @@ public class TemporalFunctions {
return localtimestamp(epochTime, timezone).toLocalDate();
}
+ public static Long extract(String unit, TemporalAccessor temporal, String
timezone) {
+ if (temporal == null) {
+ return null;
+ }
+ if (temporal instanceof Instant) {
+ temporal = ((Instant) temporal).atZone(ZoneId.of(timezone));
+ }
+ switch (unit) {
+ case "YEAR":
+ return getTemporalField(temporal, ChronoField.YEAR, unit);
+ case "QUARTER":
+ return getTemporalField(temporal, IsoFields.QUARTER_OF_YEAR,
unit);
+ case "MONTH":
+ return getTemporalField(temporal, ChronoField.MONTH_OF_YEAR,
unit);
+ case "WEEK":
+ return getTemporalField(temporal,
IsoFields.WEEK_OF_WEEK_BASED_YEAR, unit);
+ case "DAY":
+ return getTemporalField(temporal, ChronoField.DAY_OF_MONTH,
unit);
+ case "DOY":
+ return getTemporalField(temporal, ChronoField.DAY_OF_YEAR,
unit);
+ case "DOW":
+ // SQL DOW starts with Sunday as 1, while ISO starts with
Monday as 1.
+ return getTemporalField(temporal, ChronoField.DAY_OF_WEEK,
unit) % 7 + 1;
+ case "HOUR":
+ return getTemporalField(temporal, ChronoField.HOUR_OF_DAY,
unit);
+ case "MINUTE":
+ return getTemporalField(temporal, ChronoField.MINUTE_OF_HOUR,
unit);
+ case "SECOND":
+ return getTemporalField(temporal,
ChronoField.SECOND_OF_MINUTE, unit);
+ default:
+ throw new IllegalArgumentException("Unsupported EXTRACT unit:
" + unit);
+ }
+ }
+
+ private static long getTemporalField(
+ TemporalAccessor temporal, TemporalField field, String unit) {
+ if (temporal instanceof LocalDate
+ && (field == ChronoField.HOUR_OF_DAY
+ || field == ChronoField.MINUTE_OF_HOUR
+ || field == ChronoField.SECOND_OF_MINUTE)) {
+ return 0L;
+ }
+ if (!temporal.isSupported(field)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "EXTRACT unit %s cannot be applied to %s",
+ unit, temporal.getClass().getSimpleName()));
+ }
+ return temporal.getLong(field);
+ }
+
+ public static LocalDate temporalPlusMonths(LocalDate temporal, long
months) {
+ return temporal == null ? null : temporal.plusMonths(months);
+ }
+
+ public static LocalTime temporalPlusMonths(LocalTime temporal, long
months) {
+ // Year-month intervals do not affect TIME values.
+ return temporal;
+ }
+
+ public static LocalDateTime temporalPlusMonths(LocalDateTime temporal,
long months) {
+ return temporal == null ? null : temporal.plusMonths(months);
+ }
+
+ public static Instant temporalPlusMonths(Instant temporal, long months) {
+ // TIMESTAMP_LTZ month arithmetic follows UTC calendar semantics.
+ return temporal == null
+ ? null
+ :
temporal.atZone(ZoneId.of("UTC")).plusMonths(months).toInstant();
+ }
+
+ public static LocalDate temporalPlusMillis(LocalDate temporal, long
millis) {
+ // DATE arithmetic uses only the whole-day portion of a day-time
interval.
+ return temporal == null ? null : temporal.plusDays(millis /
MILLIS_PER_DAY);
+ }
+
+ public static LocalTime temporalPlusMillis(LocalTime temporal, long
millis) {
+ return temporal == null ? null : temporal.plus(millis,
ChronoUnit.MILLIS);
+ }
+
+ public static LocalDateTime temporalPlusMillis(LocalDateTime temporal,
long millis) {
+ return temporal == null ? null : temporal.plus(millis,
ChronoUnit.MILLIS);
+ }
+
+ public static Instant temporalPlusMillis(Instant temporal, long millis) {
+ return temporal == null ? null : temporal.plusMillis(millis);
+ }
+
public static String fromUnixtime(Integer seconds, String timezone) {
if (seconds == null) {
return null;
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java
index 53e81a5dc..6f8d0e464 100644
---
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/JaninoCompiler.java
@@ -35,11 +35,15 @@ import org.apache.calcite.sql.SqlBasicTypeNameSpec;
import org.apache.calcite.sql.SqlCharStringLiteral;
import org.apache.calcite.sql.SqlDataTypeSpec;
import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlIntervalLiteral;
+import org.apache.calcite.sql.SqlIntervalQualifier;
+import org.apache.calcite.sql.SqlKind;
import org.apache.calcite.sql.SqlLiteral;
import org.apache.calcite.sql.SqlNode;
import org.apache.calcite.sql.SqlNodeList;
import org.apache.calcite.sql.SqlNumericLiteral;
import org.apache.calcite.sql.fun.SqlCase;
+import org.apache.calcite.sql.parser.SqlParserUtil;
import org.apache.calcite.sql.type.SqlTypeName;
import org.apache.calcite.util.NlsString;
import org.codehaus.commons.compiler.CompileException;
@@ -54,6 +58,7 @@ import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.Set;
import java.util.stream.Collectors;
/**
@@ -64,6 +69,24 @@ import java.util.stream.Collectors;
*/
public class JaninoCompiler {
+ private static final Map<String, String> DATE_PART_FUNCTION_UNITS =
+ Map.ofEntries(
+ Map.entry("YEAR", "YEAR"),
+ Map.entry("QUARTER", "QUARTER"),
+ Map.entry("MONTH", "MONTH"),
+ Map.entry("WEEK", "WEEK"),
+ Map.entry("DAYOFYEAR", "DOY"),
+ Map.entry("DAYOFMONTH", "DAY"),
+ Map.entry("DAYOFWEEK", "DOW"),
+ Map.entry("HOUR", "HOUR"),
+ Map.entry("MINUTE", "MINUTE"),
+ Map.entry("SECOND", "SECOND"));
+
+ private static final Set<String> SUPPORTED_EXTRACT_UNITS =
+ Set.of(
+ "YEAR", "QUARTER", "MONTH", "WEEK", "DAY", "DOY", "DOW",
"HOUR", "MINUTE",
+ "SECOND");
+
private static final List<SqlTypeName> SQL_TYPE_NAME_IGNORE =
Arrays.asList(SqlTypeName.SYMBOL);
private static final List<String> TIMEZONE_FREE_TEMPORAL_FUNCTIONS =
Arrays.asList("CURRENT_TIMESTAMP", "NOW");
@@ -200,6 +223,17 @@ public class JaninoCompiler {
}
private static Java.Rvalue translateSqlBasicCall(Context context,
SqlBasicCall sqlBasicCall) {
+ String functionName =
sqlBasicCall.getOperator().getName().toUpperCase();
+ if (isIntervalArithmetic(sqlBasicCall)) {
+ return generateIntervalArithmeticOperation(context, sqlBasicCall);
+ }
+ if (sqlBasicCall.getKind() == SqlKind.EXTRACT) {
+ return generateExtractOperation(context, sqlBasicCall);
+ }
+ if (DATE_PART_FUNCTION_UNITS.containsKey(functionName)) {
+ return generateDatePartFunctionOperation(context, sqlBasicCall,
functionName);
+ }
+
List<SqlNode> operandList = sqlBasicCall.getOperandList();
List<Java.Rvalue> atoms = new ArrayList<>();
for (SqlNode sqlNode : operandList) {
@@ -219,6 +253,98 @@ public class JaninoCompiler {
return sqlBasicCallToJaninoRvalue(context, sqlBasicCall,
atoms.toArray(new Java.Rvalue[0]));
}
+ private static boolean isIntervalArithmetic(SqlBasicCall sqlBasicCall) {
+ if (sqlBasicCall.getKind() != SqlKind.PLUS && sqlBasicCall.getKind()
!= SqlKind.MINUS) {
+ return false;
+ }
+ return sqlBasicCall.getOperandList().stream()
+ .anyMatch(operand -> operand instanceof SqlIntervalLiteral);
+ }
+
+ private static Java.Rvalue generateIntervalArithmeticOperation(
+ Context context, SqlBasicCall sqlBasicCall) {
+ List<SqlNode> operands = sqlBasicCall.getOperandList();
+ if (operands.size() != 2) {
+ throw new ParseException("Unrecognized interval arithmetic: " +
sqlBasicCall);
+ }
+
+ boolean intervalOnLeft = operands.get(0) instanceof SqlIntervalLiteral;
+ boolean intervalOnRight = operands.get(1) instanceof
SqlIntervalLiteral;
+ if (intervalOnLeft == intervalOnRight
+ || (intervalOnLeft && sqlBasicCall.getKind() ==
SqlKind.MINUS)) {
+ throw new ParseException("Unsupported interval arithmetic: " +
sqlBasicCall);
+ }
+
+ SqlIntervalLiteral intervalLiteral =
+ (SqlIntervalLiteral) operands.get(intervalOnLeft ? 0 : 1);
+ SqlNode temporalOperand = operands.get(intervalOnLeft ? 1 : 0);
+ Java.Rvalue temporal = translateSqlNodeToJaninoRvalue(context,
temporalOperand);
+ if (temporal == null) {
+ throw new ParseException("Unrecognized temporal expression: " +
temporalOperand);
+ }
+
+ SqlIntervalLiteral.IntervalValue intervalValue =
+
intervalLiteral.getValueAs(SqlIntervalLiteral.IntervalValue.class);
+ boolean yearMonth = intervalValue.getIntervalQualifier().isYearMonth();
+ long amount =
+ yearMonth
+ ? SqlParserUtil.intervalToMonths(intervalValue)
+ : SqlParserUtil.intervalToMillis(intervalValue);
+ if (sqlBasicCall.getKind() == SqlKind.MINUS) {
+ amount = Math.negateExact(amount);
+ }
+
+ return generateFunctionOperation(
+ yearMonth ? "temporalPlusMonths" : "temporalPlusMillis",
+ new Java.Rvalue[] {
+ temporal,
+ new Java.AmbiguousName(
+ Location.NOWHERE, new String[]
{Long.toString(amount) + "L"})
+ });
+ }
+
+ private static Java.Rvalue generateDatePartFunctionOperation(
+ Context context, SqlBasicCall sqlBasicCall, String functionName) {
+ if (sqlBasicCall.getOperandList().size() != 1) {
+ throw new ParseException("Unrecognized expression: " +
sqlBasicCall);
+ }
+ return generateExtractOperation(
+ context,
+ DATE_PART_FUNCTION_UNITS.get(functionName),
+ sqlBasicCall.getOperandList().get(0));
+ }
+
+ private static Java.Rvalue generateExtractOperation(
+ Context context, SqlBasicCall sqlBasicCall) {
+ List<SqlNode> operands = sqlBasicCall.getOperandList();
+ if (operands.size() != 2 || !(operands.get(0) instanceof
SqlIntervalQualifier)) {
+ throw new ParseException("Unrecognized expression: " +
sqlBasicCall);
+ }
+ SqlIntervalQualifier qualifier = (SqlIntervalQualifier)
operands.get(0);
+ if (!qualifier.isSingleDatetimeField()) {
+ throw new ParseException("Unsupported EXTRACT unit: " + qualifier);
+ }
+ return generateExtractOperation(context,
qualifier.getStartUnit().name(), operands.get(1));
+ }
+
+ private static Java.Rvalue generateExtractOperation(
+ Context context, String unit, SqlNode temporalOperand) {
+ if (!SUPPORTED_EXTRACT_UNITS.contains(unit)) {
+ throw new ParseException("Unsupported EXTRACT unit: " + unit);
+ }
+ Java.Rvalue temporal = translateSqlNodeToJaninoRvalue(context,
temporalOperand);
+ if (temporal == null) {
+ throw new ParseException("Unrecognized temporal expression: " +
temporalOperand);
+ }
+ return generateFunctionOperation(
+ "extract",
+ new Java.Rvalue[] {
+ new Java.AmbiguousName(Location.NOWHERE, new String[]
{"\"" + unit + "\""}),
+ temporal,
+ new Java.AmbiguousName(Location.NOWHERE, new String[]
{DEFAULT_TIME_ZONE})
+ });
+ }
+
private static Java.Rvalue translateSqlCase(Context context, SqlCase
sqlCase) {
SqlNodeList whenOperands = sqlCase.getWhenOperands();
SqlNodeList thenOperands = sqlCase.getThenOperands();
diff --git
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/metadata/TransformSqlOperatorTable.java
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/metadata/TransformSqlOperatorTable.java
index 4850815cc..096e3fa2b 100644
---
a/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/metadata/TransformSqlOperatorTable.java
+++
b/flink-cdc-runtime/src/main/java/org/apache/flink/cdc/runtime/parser/metadata/TransformSqlOperatorTable.java
@@ -443,6 +443,17 @@ public class TransformSqlOperatorTable extends
ReflectiveSqlOperatorTable {
// ------------------
// Temporal Functions
// ------------------
+ public static final SqlFunction EXTRACT = SqlStdOperatorTable.EXTRACT;
+ public static final SqlFunction YEAR = SqlStdOperatorTable.YEAR;
+ public static final SqlFunction QUARTER = SqlStdOperatorTable.QUARTER;
+ public static final SqlFunction MONTH = SqlStdOperatorTable.MONTH;
+ public static final SqlFunction WEEK = SqlStdOperatorTable.WEEK;
+ public static final SqlFunction DAYOFYEAR = SqlStdOperatorTable.DAYOFYEAR;
+ public static final SqlFunction DAYOFMONTH =
SqlStdOperatorTable.DAYOFMONTH;
+ public static final SqlFunction DAYOFWEEK = SqlStdOperatorTable.DAYOFWEEK;
+ public static final SqlFunction HOUR = SqlStdOperatorTable.HOUR;
+ public static final SqlFunction MINUTE = SqlStdOperatorTable.MINUTE;
+ public static final SqlFunction SECOND = SqlStdOperatorTable.SECOND;
public static final SqlFunction LOCALTIME = SqlStdOperatorTable.LOCALTIME;
public static final SqlFunction LOCALTIMESTAMP =
new BuiltInTimestampFunction("LOCALTIMESTAMP",
SqlTypeName.TIMESTAMP, 3);
diff --git
a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/TemporalFunctionsTest.java
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/TemporalFunctionsTest.java
new file mode 100644
index 000000000..ad6d29bf9
--- /dev/null
+++
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/functions/impl/TemporalFunctionsTest.java
@@ -0,0 +1,138 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.cdc.runtime.functions.impl;
+
+import org.junit.jupiter.api.Test;
+
+import java.time.Instant;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.ZoneId;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+/** Tests for {@link TemporalFunctions}. */
+class TemporalFunctionsTest {
+
+ @Test
+ void testExtractDateParts() {
+ LocalDate date = LocalDate.of(2024, 2, 29);
+
+ assertThat(TemporalFunctions.extract("YEAR", date,
"UTC")).isEqualTo(2024L);
+ assertThat(TemporalFunctions.extract("QUARTER", date,
"UTC")).isEqualTo(1L);
+ assertThat(TemporalFunctions.extract("MONTH", date,
"UTC")).isEqualTo(2L);
+ assertThat(TemporalFunctions.extract("WEEK", date,
"UTC")).isEqualTo(9L);
+ assertThat(TemporalFunctions.extract("DAY", date,
"UTC")).isEqualTo(29L);
+ assertThat(TemporalFunctions.extract("DOY", date,
"UTC")).isEqualTo(60L);
+ assertThat(TemporalFunctions.extract("DOW", date,
"UTC")).isEqualTo(5L);
+ assertThat(TemporalFunctions.extract("HOUR", date, "UTC")).isZero();
+ assertThat(TemporalFunctions.extract("MINUTE", date, "UTC")).isZero();
+ assertThat(TemporalFunctions.extract("SECOND", date, "UTC")).isZero();
+ }
+
+ @Test
+ void testExtractTimeParts() {
+ LocalTime time = LocalTime.of(23, 58, 57);
+
+ assertThat(TemporalFunctions.extract("HOUR", time,
"UTC")).isEqualTo(23L);
+ assertThat(TemporalFunctions.extract("MINUTE", time,
"UTC")).isEqualTo(58L);
+ assertThat(TemporalFunctions.extract("SECOND", time,
"UTC")).isEqualTo(57L);
+ }
+
+ @Test
+ void testExtractTimestampParts() {
+ LocalDateTime timestamp = LocalDateTime.of(2024, 12, 31, 23, 58, 57);
+
+ assertThat(TemporalFunctions.extract("YEAR", timestamp,
"UTC")).isEqualTo(2024L);
+ assertThat(TemporalFunctions.extract("WEEK", timestamp,
"UTC")).isEqualTo(1L);
+ assertThat(TemporalFunctions.extract("HOUR", timestamp,
"UTC")).isEqualTo(23L);
+ }
+
+ @Test
+ void testExtractTimestampLtzUsesPipelineTimeZone() {
+ Instant timestamp = Instant.parse("2023-12-31T16:30:00Z");
+
+ assertThat(TemporalFunctions.extract("YEAR", timestamp,
"UTC")).isEqualTo(2023L);
+ assertThat(TemporalFunctions.extract("YEAR", timestamp,
"Asia/Shanghai")).isEqualTo(2024L);
+ assertThat(TemporalFunctions.extract("HOUR", timestamp,
"Asia/Shanghai")).isEqualTo(0L);
+ }
+
+ @Test
+ void testExtractTimestampTzUsesValueTimeZone() {
+ assertThat(
+ TemporalFunctions.extract(
+ "HOUR",
+ Instant.parse("2024-01-01T00:00:00Z")
+ .atZone(ZoneId.of("Asia/Shanghai")),
+ "UTC"))
+ .isEqualTo(8L);
+ }
+
+ @Test
+ void testExtractNullAndUnsupportedParts() {
+ assertThat(TemporalFunctions.extract("YEAR", null, "UTC")).isNull();
+ assertThatThrownBy(() -> TemporalFunctions.extract("YEAR",
LocalTime.NOON, "UTC"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("EXTRACT unit YEAR cannot be applied to
LocalTime");
+ assertThatThrownBy(
+ () -> TemporalFunctions.extract("MILLISECOND",
LocalDateTime.now(), "UTC"))
+ .isInstanceOf(IllegalArgumentException.class)
+ .hasMessage("Unsupported EXTRACT unit: MILLISECOND");
+ }
+
+ @Test
+ void testTemporalPlusMonths() {
+ assertThat(TemporalFunctions.temporalPlusMonths(LocalDate.of(2024, 1,
31), 1))
+ .isEqualTo(LocalDate.of(2024, 2, 29));
+ assertThat(
+ TemporalFunctions.temporalPlusMonths(
+ LocalDateTime.of(2024, 1, 31, 12, 34, 56), 13))
+ .isEqualTo(LocalDateTime.of(2025, 2, 28, 12, 34, 56));
+
assertThat(TemporalFunctions.temporalPlusMonths(Instant.parse("2024-01-31T12:34:56Z"),
-1))
+ .isEqualTo(Instant.parse("2023-12-31T12:34:56Z"));
+ assertThat(TemporalFunctions.temporalPlusMonths(LocalTime.NOON, 12))
+ .isEqualTo(LocalTime.NOON);
+ assertThat(TemporalFunctions.temporalPlusMonths((LocalDate) null,
1)).isNull();
+ }
+
+ @Test
+ void testTemporalPlusMillis() {
+ assertThat(
+ TemporalFunctions.temporalPlusMillis(
+ LocalDate.of(2024, 2, 28), 24L * 60 * 60 *
1000))
+ .isEqualTo(LocalDate.of(2024, 2, 29));
+ assertThat(
+ TemporalFunctions.temporalPlusMillis(
+ LocalDate.of(2024, 2, 28), 12L * 60 * 60 *
1000))
+ .isEqualTo(LocalDate.of(2024, 2, 28));
+ assertThat(TemporalFunctions.temporalPlusMillis(LocalTime.of(23, 30),
2L * 60 * 60 * 1000))
+ .isEqualTo(LocalTime.of(1, 30));
+ assertThat(
+ TemporalFunctions.temporalPlusMillis(
+ LocalDateTime.of(2024, 2, 28, 23, 30), 2L * 60
* 60 * 1000))
+ .isEqualTo(LocalDateTime.of(2024, 2, 29, 1, 30));
+ assertThat(
+ TemporalFunctions.temporalPlusMillis(
+ Instant.parse("2024-02-28T23:30:00Z"), 2L * 60
* 60 * 1000))
+ .isEqualTo(Instant.parse("2024-02-29T01:30:00Z"));
+ assertThat(TemporalFunctions.temporalPlusMillis((Instant) null,
1)).isNull();
+ }
+}
diff --git
a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformParserTest.java
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformParserTest.java
index c46525cd7..ab9ed2467 100644
---
a/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformParserTest.java
+++
b/flink-cdc-runtime/src/test/java/org/apache/flink/cdc/runtime/parser/TransformParserTest.java
@@ -59,6 +59,72 @@ class TransformParserTest {
.primaryKey("id")
.build();
+ @Test
+ void testGenerateDatePartFunctionProjectionColumns() {
+ List<ProjectionColumn> projectionColumns =
+ TransformParser.generateProjectionColumns(
+ "YEAR(date_col) AS year_col, "
+ + "HOUR(time_col) AS hour_col, "
+ + "EXTRACT(DAY FROM timestamp_col) AS day_col,
"
+ + "EXTRACT(HOUR FROM timestamp_ltz_col) AS
ltz_hour_col",
+ Arrays.asList(
+ Column.physicalColumn("date_col",
DataTypes.DATE()),
+ Column.physicalColumn("time_col",
DataTypes.TIME()),
+ Column.physicalColumn("timestamp_col",
DataTypes.TIMESTAMP()),
+ Column.physicalColumn(
+ "timestamp_ltz_col",
DataTypes.TIMESTAMP_LTZ())),
+ Collections.emptyList(),
+ new SupportedMetadataColumn[0]);
+
+ Assertions.assertThat(projectionColumns)
+ .extracting(ProjectionColumn::getDataType)
+ .containsExactly(
+ DataTypes.BIGINT(),
+ DataTypes.BIGINT(),
+ DataTypes.BIGINT(),
+ DataTypes.BIGINT());
+ Assertions.assertThat(projectionColumns)
+ .extracting(ProjectionColumn::getScriptExpression)
+ .containsExactly(
+ "extract(\"YEAR\", $0, __time_zone__)",
+ "extract(\"HOUR\", $0, __time_zone__)",
+ "extract(\"DAY\", $0, __time_zone__)",
+ "extract(\"HOUR\", $0, __time_zone__)");
+ }
+
+ @Test
+ void testGenerateIntervalArithmeticProjectionColumns() {
+ List<ProjectionColumn> projectionColumns =
+ TransformParser.generateProjectionColumns(
+ "date_col + INTERVAL '1' DAY AS date_col, "
+ + "time_col + INTERVAL '2:03' HOUR TO MINUTE
AS time_col, "
+ + "timestamp_col - INTERVAL '1-2' YEAR TO
MONTH AS timestamp_col, "
+ + "INTERVAL '3' DAY + timestamp_ltz_col AS
timestamp_ltz_col",
+ Arrays.asList(
+ Column.physicalColumn("date_col",
DataTypes.DATE()),
+ Column.physicalColumn("time_col",
DataTypes.TIME()),
+ Column.physicalColumn("timestamp_col",
DataTypes.TIMESTAMP()),
+ Column.physicalColumn(
+ "timestamp_ltz_col",
DataTypes.TIMESTAMP_LTZ())),
+ Collections.emptyList(),
+ new SupportedMetadataColumn[0]);
+
+ Assertions.assertThat(projectionColumns)
+ .extracting(ProjectionColumn::getDataType)
+ .containsExactly(
+ DataTypes.DATE(),
+ DataTypes.TIME(),
+ DataTypes.TIMESTAMP(3),
+ DataTypes.TIMESTAMP_LTZ(3));
+ Assertions.assertThat(projectionColumns)
+ .extracting(ProjectionColumn::getScriptExpression)
+ .containsExactly(
+ "temporalPlusMillis($0, 86400000L)",
+ "temporalPlusMillis($0, 7380000L)",
+ "temporalPlusMonths($0, -14L)",
+ "temporalPlusMillis($0, 259200000L)");
+ }
+
@Test
void testCalciteParser() {
SqlSelect parse =
@@ -302,10 +368,18 @@ class TransformParserTest {
testFilterExpression(
"UNIX_TIMESTAMP('1970-01-01 08:00:01.001 +0800', 'yyyy-MM-dd
HH:mm:ss.SSS X')",
"unixTimestamp(\"1970-01-01 08:00:01.001 +0800\", \"yyyy-MM-dd
HH:mm:ss.SSS X\", __epoch_time__, __time_zone__)");
- testFilterExpression("YEAR(dt)", "year(dt)");
- testFilterExpression("QUARTER(dt)", "quarter(dt)");
- testFilterExpression("MONTH(dt)", "month(dt)");
- testFilterExpression("WEEK(dt)", "week(dt)");
+ testFilterExpression("YEAR(dt)", "extract(\"YEAR\", dt,
__time_zone__)");
+ testFilterExpression("QUARTER(dt)", "extract(\"QUARTER\", dt,
__time_zone__)");
+ testFilterExpression("MONTH(dt)", "extract(\"MONTH\", dt,
__time_zone__)");
+ testFilterExpression("WEEK(dt)", "extract(\"WEEK\", dt,
__time_zone__)");
+ testFilterExpression("DAYOFYEAR(dt)", "extract(\"DOY\", dt,
__time_zone__)");
+ testFilterExpression("DAYOFMONTH(dt)", "extract(\"DAY\", dt,
__time_zone__)");
+ testFilterExpression("DAYOFWEEK(dt)", "extract(\"DOW\", dt,
__time_zone__)");
+ testFilterExpression("HOUR(dt)", "extract(\"HOUR\", dt,
__time_zone__)");
+ testFilterExpression("MINUTE(dt)", "extract(\"MINUTE\", dt,
__time_zone__)");
+ testFilterExpression("SECOND(dt)", "extract(\"SECOND\", dt,
__time_zone__)");
+ testFilterExpression("EXTRACT(YEAR FROM dt)", "extract(\"YEAR\", dt,
__time_zone__)");
+ testFilterExpression("EXTRACT(DOY FROM dt)", "extract(\"DOY\", dt,
__time_zone__)");
testFilterExpression(
"DATE_FORMAT(dt,'yyyy-MM-dd')", "dateFormat(dt,
\"yyyy-MM-dd\", __time_zone__)");
testFilterExpression("TO_DATE(dt, 'yyyy-MM-dd')", "toDate(dt,
\"yyyy-MM-dd\")");
@@ -412,6 +486,11 @@ class TransformParserTest {
testFilterExpression("id * 2", "id * 2");
testFilterExpression("id / 2", "id / 2");
testFilterExpression("id % 2", "id % 2");
+ testFilterExpression("dt + INTERVAL '1' DAY", "temporalPlusMillis(dt,
86400000L)");
+ testFilterExpression("dt - INTERVAL '1-2' YEAR TO MONTH",
"temporalPlusMonths(dt, -14L)");
+ testFilterExpression(
+ "INTERVAL '2:03' HOUR TO MINUTE + dt", "temporalPlusMillis(dt,
7380000L)");
+ testFilterExpression("dt + INTERVAL '-1' SECOND",
"temporalPlusMillis(dt, -1000L)");
testFilterExpression("a < b", "lessThan(a, b)");
testFilterExpression("a <= b", "lessThanOrEqual(a, b)");
testFilterExpression("a > b", "greaterThan(a, b)");
@@ -661,6 +740,16 @@ class TransformParserTest {
@Test
public void testTranslateFilterToJaninoExpressionError() {
+ Assertions.assertThatThrownBy(
+ () ->
+
TransformParser.translateFilterExpressionToJaninoExpression(
+ "INTERVAL '1' DAY - dt",
+ Collections.emptyList(),
+ Collections.emptyList(),
+ new SupportedMetadataColumn[0],
+ Collections.emptyMap()))
+ .isExactlyInstanceOf(ParseException.class)
+ .hasMessageStartingWith("Unsupported interval arithmetic:");
Assertions.assertThatThrownBy(
() -> {
TransformParser.translateFilterExpressionToJaninoExpression(
