[jira] [Commented] (FLINK-8897) Rowtime materialization causes "mismatched type" AssertionError

["ASF GitHub Bot (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/FLINK-8897?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16671585#comment-16671585 ] 

ASF GitHub Bot commented on FLINK-8897:
---------------------------------------

twalthr opened a new pull request #6987: [FLINK-8897] [table] Fix rowtime materialization issues for filters and joins
URL: https://github.com/apache/flink/pull/6987
 
 
   ## What is the purpose of the change
   
   This PR solves the `"mismatched type" AssertionError` caused by invalid time indicator materialization. FLINK-6232 removed the materialization of filters for windowed joins, however, this led to mismatches between input types and expected input types of `RexNodes`. This PR modifies the window join rule to match against a newly materialized expression (`CAST(rowtime)` and `PROCTIME(proctime)`) instead.
   
   Additionally, this PR simplifies the general time attribute logic for joins and gives better error messages for current limitations (namely no rowtime attributes in projections or condiitions). We should fix these limitation by some major refactoring as part of FLINK-10211.
   
   
   ## Brief change log
   
   - Materialize conditions of filters and joins again and match against newly materialized expressions
   - Simplify code and improve error message
   
   
   ## Verifying this change
   
   - Change is covered by existing tests.
   - New tests in `JoinValidationTest`
   - New test in `TimeAttributesITCase`
   
   ## Does this pull request potentially affect one of the following parts:
   
     - Dependencies (does it add or upgrade a dependency): no
     - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no
     - The serializers: no
     - The runtime per-record code paths (performance sensitive): no
     - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Yarn/Mesos, ZooKeeper: no
     - The S3 file system connector: no
   
   ## Documentation
   
     - Does this pull request introduce a new feature? no
     - If yes, how is the feature documented? not applicable
   

----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
users@infra.apache.org


> 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
>            Assignee: Timo Walther
>            Priority: Major
>              Labels: pull-request-available
>             Fix For: 1.5.6
>
>
> 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)