[ https://issues.apache.org/jira/browse/FLINK-8897?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Xingcan Cui updated FLINK-8897: ------------------------------- Description: As raised in [this thread|https://lists.apache.org/thread.html/e2ea38aa7ae224d7481145334955d84243690e9aad10d58310bdb8e7@%3Cuser.flink.apache.org%3E], the query created by the following code will throw a calcite "mismatch type" ({{Timestamp(3)}} and {{TimeIndicator}}) exception. {code:java} String sql1 = "select id, eventTs as t1, count(*) over (partition by id order by eventTs rows between 100 preceding and current row) as cnt1 from myTable1"; String sql2 = "select distinct id as r_id, eventTs as t2, count(*) over (partition by id order by eventTs rows between 50 preceding and current row) as cnt2 from myTable2"; Table left = tableEnv.sqlQuery(sql1); Table right = tableEnv.sqlQuery(sql2); left.join(right).where("id === r_id && t1 === t2").select("id, t1").writeToSink(...) {code} The logical plan is as follows. {code} LogicalProject(id=[$0], t1=[$1]) LogicalFilter(condition=[AND(=($0, $3), =($1, $4))]) LogicalJoin(condition=[true], joinType=[inner]) LogicalAggregate(group=[{0, 1, 2}]) LogicalWindow(window#0=[window(partition {0} order by [1] rows between $2 PRECEDING and CURRENT ROW aggs [COUNT()])]) LogicalProject(id=[$0], eventTs=[$3]) LogicalTableScan(table=[[_DataStreamTable_0]]) LogicalAggregate(group=[{0, 1, 2}]) LogicalWindow(window#0=[window(partition {0} order by [1] rows between $2 PRECEDING and CURRENT ROW aggs [COUNT()])]) LogicalProject(id=[$0], eventTs=[$3]) LogicalTableScan(table=[[_DataStreamTable_0]]) {code} That is because the the rowtime field after an aggregation will be materialized while the {{RexInputRef}} type for the filter's operands ({{t1 === t2}}) is still {{TimeIndicator}}. We should make them unified. was: As raised in [this thread|https://lists.apache.org/thread.html/e2ea38aa7ae224d7481145334955d84243690e9aad10d58310bdb8e7@%3Cuser.flink.apache.org%3E], the query created by the following code will throw a calcite "mismatch type" ({{Timestamp(3)}} and {{TimeIndicator}}) exception. {code:java} String sql1 = "select id, eventTs as t1, count(*) over (partition by id order by eventTs rows between 100 preceding and current row) as cnt1 from myTable1"; String sql2 = "select distinct id as r_id, eventTs as t2, count(*) over (partition by id order by eventTs rows between 50 preceding and current row) as cnt2 from myTable2"; Table left = tableEnv.sqlQuery(sql1); Table right = tableEnv.sqlQuery(sql2); left.join(right).where("id === r_id && t1 === t2").select("id, t1").writeToSink(...) {code} That is because the the rowtime field after an aggregation will be materialized while the {{RexInputRef}} type for the filter's operands ({{t1 === t2}}) is still {{TimeIndicator}}. We should make them unified. > Rowtime materialization causes "mismatched type" AssertionError > --------------------------------------------------------------- > > Key: FLINK-8897 > URL: https://issues.apache.org/jira/browse/FLINK-8897 > Project: Flink > Issue Type: Bug > Components: Table API & SQL > Reporter: Xingcan Cui > Priority: Major > > As raised in [this > thread|https://lists.apache.org/thread.html/e2ea38aa7ae224d7481145334955d84243690e9aad10d58310bdb8e7@%3Cuser.flink.apache.org%3E], > the query created by the following code will throw a calcite "mismatch type" > ({{Timestamp(3)}} and {{TimeIndicator}}) exception. > {code:java} > String sql1 = "select id, eventTs as t1, count(*) over (partition by id order > by eventTs rows between 100 preceding and current row) as cnt1 from myTable1"; > String sql2 = "select distinct id as r_id, eventTs as t2, count(*) over > (partition by id order by eventTs rows between 50 preceding and current row) > as cnt2 from myTable2"; > Table left = tableEnv.sqlQuery(sql1); > Table right = tableEnv.sqlQuery(sql2); > left.join(right).where("id === r_id && t1 === t2").select("id, > t1").writeToSink(...) > {code} > The logical plan is as follows. > {code} > LogicalProject(id=[$0], t1=[$1]) > LogicalFilter(condition=[AND(=($0, $3), =($1, $4))]) > LogicalJoin(condition=[true], joinType=[inner]) > LogicalAggregate(group=[{0, 1, 2}]) > LogicalWindow(window#0=[window(partition {0} order by [1] rows > between $2 PRECEDING and CURRENT ROW aggs [COUNT()])]) > LogicalProject(id=[$0], eventTs=[$3]) > LogicalTableScan(table=[[_DataStreamTable_0]]) > LogicalAggregate(group=[{0, 1, 2}]) > LogicalWindow(window#0=[window(partition {0} order by [1] rows > between $2 PRECEDING and CURRENT ROW aggs [COUNT()])]) > LogicalProject(id=[$0], eventTs=[$3]) > LogicalTableScan(table=[[_DataStreamTable_0]]) > {code} > That is because the the rowtime field after an aggregation will be > materialized while the {{RexInputRef}} type for the filter's operands ({{t1 > === t2}}) is still {{TimeIndicator}}. We should make them unified. -- This message was sent by Atlassian JIRA (v7.6.3#76005)