You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Godfrey He (Jira)" <ji...@apache.org> on 2022/12/05 12:34:00 UTC
[jira] [Updated] (FLINK-28988) Incorrect result for filter after temporal join
[ https://issues.apache.org/jira/browse/FLINK-28988?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Godfrey He updated FLINK-28988:
-------------------------------
Fix Version/s: 1.17.0
> Incorrect result for filter after temporal join
> -----------------------------------------------
>
> Key: FLINK-28988
> URL: https://issues.apache.org/jira/browse/FLINK-28988
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / API
> Affects Versions: 1.15.1
> Reporter: Xuannan Su
> Assignee: Shuiqiang Chen
> Priority: Major
> Labels: pull-request-available
> Fix For: 1.17.0
>
>
> The following code can reproduce the case
>
> {code:java}
> public class TemporalJoinSQLExample1 {
> public static void main(String[] args) throws Exception {
> // set up the Java DataStream API
> final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
> // set up the Java Table API
> final StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
> final DataStreamSource<Tuple3<Integer, String, Instant>> ds =
> env.fromElements(
> new Tuple3<>(0, "online", Instant.ofEpochMilli(0)),
> new Tuple3<>(0, "offline", Instant.ofEpochMilli(10)),
> new Tuple3<>(0, "online", Instant.ofEpochMilli(20)));
> final Table table =
> tableEnv.fromDataStream(
> ds,
> Schema.newBuilder()
> .column("f0", DataTypes.INT())
> .column("f1", DataTypes.STRING())
> .column("f2", DataTypes.TIMESTAMP_LTZ(3))
> .watermark("f2", "f2 - INTERVAL '2' SECONDS")
> .build())
> .as("id", "state", "ts");
> tableEnv.createTemporaryView("source_table", table);
> final Table dedupeTable =
> tableEnv.sqlQuery(
> "SELECT * FROM ("
> + " SELECT *, ROW_NUMBER() OVER (PARTITION BY id ORDER BY ts DESC) AS row_num FROM source_table"
> + ") WHERE row_num = 1");
> tableEnv.createTemporaryView("versioned_table", dedupeTable);
> DataStreamSource<Tuple2<Integer, Instant>> event =
> env.fromElements(
> new Tuple2<>(0, Instant.ofEpochMilli(0)),
> new Tuple2<>(0, Instant.ofEpochMilli(5)),
> new Tuple2<>(0, Instant.ofEpochMilli(10)),
> new Tuple2<>(0, Instant.ofEpochMilli(15)),
> new Tuple2<>(0, Instant.ofEpochMilli(20)),
> new Tuple2<>(0, Instant.ofEpochMilli(25)));
> final Table eventTable =
> tableEnv.fromDataStream(
> event,
> Schema.newBuilder()
> .column("f0", DataTypes.INT())
> .column("f1", DataTypes.TIMESTAMP_LTZ(3))
> .watermark("f1", "f1 - INTERVAL '2' SECONDS")
> .build())
> .as("id", "ts");
> tableEnv.createTemporaryView("event_table", eventTable);
> final Table result =
> tableEnv.sqlQuery(
> "SELECT * FROM event_table"
> + " LEFT JOIN versioned_table FOR SYSTEM_TIME AS OF event_table.ts"
> + " ON event_table.id = versioned_table.id");
> result.execute().print();
> result.filter($("state").isEqual("online")).execute().print();
> }
> } {code}
>
> The result of temporal join is the following:
> |op| id| ts| id0| state| ts0| row_num|
> |+I| 0|1970-01-01 08:00:00.000| 0| online|1970-01-01 08:00:00.000| 1|
> |+I| 0|1970-01-01 08:00:00.005| 0| online|1970-01-01 08:00:00.000| 1|
> |+I| 0|1970-01-01 08:00:00.010| 0| offline|1970-01-01 08:00:00.010| 1|
> |+I| 0|1970-01-01 08:00:00.015| 0| offline|1970-01-01 08:00:00.010| 1|
> |+I| 0|1970-01-01 08:00:00.020| 0| online|1970-01-01 08:00:00.020| 1|
> |+I| 0|1970-01-01 08:00:00.025| 0| online|1970-01-01 08:00:00.020| 1|
>
> After filtering with predicate state = 'online', I expect only the two rows with state offline will be filtered out. But I got the following result:
> |op| id| ts| id0| state| ts0| row_num|
> |+I| 0|1970-01-01 08:00:00.020| 0| online|1970-01-01 08:00:00.020| 1|
> |+I| 0|1970-01-01 08:00:00.025| 0| online|1970-01-01 08:00:00.020| 1|
>
>
>
>
--
This message was sent by Atlassian Jira
(v8.20.10#820010)