[
https://issues.apache.org/jira/browse/FLINK-36636?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17894484#comment-17894484
]
Wenkai Qi commented on FLINK-36636:
-----------------------------------
{code:java}
@Test
void testTimestampCompare() throws Exception {
// TIMESTAMP_DIFF('SECOND',TO_TIMESTAMP('2024-01-01 00:00:00') ,
TO_TIMESTAMP('2024-08-01 00:00:00')) > 0
PostTransformOperator transform =
PostTransformOperator.newBuilder()
.addTransform(
TIMESTAMP_TABLEID.identifier(),
"col1, IF(LOCALTIME = CURRENT_TIME, 1, 0) as
time_equal,"
+ " IF( TO_TIMESTAMP('2024-01-01 00:00:00')
< TO_TIMESTAMP('2024-08-01 00:00:00'), 1, 0) as timestamp_equal,"
+ " IF(TO_DATE(DATE_FORMAT(LOCALTIMESTAMP,
'yyyy-MM-dd')) = CURRENT_DATE, 1, 0) as date_equal",
"LOCALTIMESTAMP = CAST(CURRENT_TIMESTAMP AS
TIMESTAMP)")
.addTimezone("UTC")
.build();
EventOperatorTestHarness<PostTransformOperator, Event>
transformFunctionEventEventOperatorTestHarness =
new EventOperatorTestHarness<>(transform, 1);
// Initialization
transformFunctionEventEventOperatorTestHarness.open();
// Create table
CreateTableEvent createTableEvent =
new CreateTableEvent(TIMESTAMP_TABLEID, TIMESTAMP_SCHEMA);
BinaryRecordDataGenerator recordDataGenerator =
new BinaryRecordDataGenerator(((RowType)
TIMESTAMP_SCHEMA.toRowDataType()));
// Insert
DataChangeEvent insertEvent =
DataChangeEvent.insertEvent(
TIMESTAMP_TABLEID,
recordDataGenerator.generate(
new Object[] {new BinaryStringData("1"), null,
null, null}));
DataChangeEvent insertEventExpect =
DataChangeEvent.insertEvent(
TIMESTAMP_TABLEID,
recordDataGenerator.generate(
new Object[] {new BinaryStringData("1"), 1, 1, 1}));
transform.processElement(new StreamRecord<>(createTableEvent));
Assertions.assertThat(
transformFunctionEventEventOperatorTestHarness.getOutputRecords().poll())
.isEqualTo(
new StreamRecord<>(
new CreateTableEvent(TIMESTAMP_TABLEID,
TIMESTAMP_SCHEMA)));
transform.processElement(new StreamRecord<>(insertEvent));
Assertions.assertThat(
transformFunctionEventEventOperatorTestHarness.getOutputRecords().poll())
.isEqualTo(new StreamRecord<>(insertEventExpect));
}
{code}
> Timestamp data can not be compared in cdc pipeline transform
> ------------------------------------------------------------
>
> Key: FLINK-36636
> URL: https://issues.apache.org/jira/browse/FLINK-36636
> Project: Flink
> Issue Type: Bug
> Components: Flink CDC
> Affects Versions: cdc-3.2.0
> Reporter: Wenkai Qi
> Priority: Major
> Original Estimate: 168h
> Remaining Estimate: 168h
>
> Timestamp data can not be compared in cdc pipeline transform.
> When projecting or filtering code containing timestamp comparisons, the
> program will encounter errors.
> {code:java}
> TO_TIMESTAMP('2024-01-01 00:00:00') < TO_TIMESTAMP('2024-08-01 00:00:00')
> {code}
> Error:
> {code:java}
> org.apache.flink.api.common.InvalidProgramException: Expression cannot be
> compiled. This is a bug. Please file an issue.
> Expression: import static
> org.apache.flink.cdc.runtime.functions.SystemFunctionUtils.*;toTimestamp("2024-01-01
> 00:00:00", __time_zone__) < toTimestamp("2024-08-01 00:00:00",
> __time_zone__) ? 1 : 0
> ...
> ...
> Caused by: org.codehaus.commons.compiler.CompileException: Line 1, Column
> 176: Operator "<" not allowed on reference operands {code}
> There is currently a method that can be temporarily replaced, but I believe
> it is not in line with user habits.
> {code:java}
> TIMESTAMP_DIFF('SECOND',TO_TIMESTAMP('2024-01-01 00:00:00') ,
> TO_TIMESTAMP('2024-08-01 00:00:00')) > 0 {code}
> Do we need to support comparing timestamps through relational operators, such
> as
> {code:java}
> > >= < <= {code}
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
