[jira] [Updated] (FLINK-36626) Flink SQL temporal lookup JOINs behavior change from Flink 1.15 to Flink 1.20

["Eduardo Breijo (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/FLINK-36626?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Eduardo Breijo updated FLINK-36626:
-----------------------------------
    Summary: Flink SQL temporal lookup JOINs behavior change from Flink 1.15 to Flink 1.20  (was: Flink SQL temporal lookup JOINs behavior change from Flink 1.15 to Flink 1.20+)

> Flink SQL temporal lookup JOINs behavior change from Flink 1.15 to Flink 1.20
> -----------------------------------------------------------------------------
>
>                 Key: FLINK-36626
>                 URL: https://issues.apache.org/jira/browse/FLINK-36626
>             Project: Flink
>          Issue Type: Bug
>          Components: Table SQL / API
>    Affects Versions: 1.18.1, 1.20.0
>         Environment: AWS Managed Apache Flink 
>            Reporter: Eduardo Breijo
>            Priority: Critical
>         Attachments: Tasks Execution Plan.txt
>
>
> There is a behavior change I found when migrating to Flink 1.18+ from Flink 1.15 in regards to Flink SQL temporal lookup joins that I haven't been able to pin point and is causing the query below to output different results.
> *Flink SQL Query:*
> ~WITH assets_setpoint AS (~
>   ~SELECT~
>     ~asset_id,~
>     ~TUMBLE_START(`timestamp`, INTERVAL '1' MINUTES) AS start_timestamp,~
>     ~TUMBLE_END(`timestamp`, INTERVAL '1' MINUTES) AS end_timestamp,~
>     ~LAST_VALUE(`value`) AS `value`~
>   ~FROM asset_readings~
>   ~JOIN metric FOR SYSTEM_TIME AS OF `proctime`~
>   ~ON metric.metric_id = asset_readings.metric_id~
>   ~WHERE metric.alias IN ('Parameters.Main.SupplyAirTempSetpoint')~
>   ~GROUP BY TUMBLE(`timestamp`, INTERVAL '1' MINUTES), asset_id~
> ~),~
> ~assets_supply_air_temp AS (~
>   ~-- CAST needed to perform regular joins instead of temporal joins in the outer query~
>   ~SELECT CAST(asset_readings.`timestamp` AS TIMESTAMP) AS `timestamp`,~
>       ~asset_readings.asset_id,~
>       ~asset_readings.`value` AS `value`~
>   ~FROM asset_readings~
>   ~-- Metrics temporal lookup inner join~
>   ~JOIN metric FOR SYSTEM_TIME AS OF `proctime`~
>   ~ON metric.metric_id = asset_readings.metric_id~
>   ~-- Assets to ignore for this computed metric definition temporal lookup left join~
>   ~LEFT JOIN asset_to_ignore_per_computed_metric_definition FOR SYSTEM_TIME AS OF `proctime`~
>   ~ON asset_to_ignore_per_computed_metric_definition.computed_metric_definition_id = :computedMetricDefinitionId~
>   ~AND asset_to_ignore_per_computed_metric_definition.asset_id = asset_readings.asset_id~
>   ~WHERE metric.alias IN ('Sensors.InsideAir.Supply.Temp.SupplyAirTemp')~
>   ~-- Filter assets not present in the asset to ignore for this computed metric definition table~
>   ~AND asset_to_ignore_per_computed_metric_definition.asset_id IS NULL~
> ~)~
> ~SELECT~
> ~assets_supply_air_temp.`timestamp`,~
> ~assets_supply_air_temp.asset_id,~
> ~assets_supply_air_temp.`value` - assets_setpoint.`value` AS `value`~
> ~FROM assets_supply_air_temp~
> ~INNER JOIN assets_setpoint~
> ~ON assets_setpoint.asset_id = assets_supply_air_temp.asset_id~
> ~AND assets_supply_air_temp.`timestamp` BETWEEN assets_setpoint.start_timestamp AND assets_setpoint.end_timestamp~
>  
> *Schema:*
> ~{+}{-}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{-}{{-}}{-}{{-}}{-}{{-}}{-}{{-}}{-}{-}{-}{+}-------{-}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{+}{{+}}{{-}}{{-}}{{-}}{-}{{-}}{{-}}{{-}}{{-}}{{-}}{-}{{-}}{{-}}{+}{{+}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{-}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{-}----------\{+}~
> ~|      name |                        type |  null | key |        extras |          watermark |~
> ~{+}{-}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{-}{{-}}{-}{{-}}{-}{{-}}{-}{{-}}{-}{-}{-}{+}-------{-}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{+}{{+}}{{-}}{{-}}{{-}}{-}{{-}}{{-}}{{-}}{{-}}{{-}}{-}{{-}}{{-}}{+}{{+}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{-}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{-}----------\{+}~
> ~| timestamp |  TIMESTAMP_LTZ(3) *ROWTIME* |  TRUE |     |               | SOURCE_WATERMARK() |~
> ~|  asset_id |                      BIGINT |  TRUE |     |               |                    |~
> ~| metric_id |                         INT |  TRUE |     |               |                    |~
> ~|     value |                      DOUBLE |  TRUE |     |               |                    |~
> ~|  metadata |         MAP<STRING, STRING> |  TRUE |     |               |                    |~
> ~|  proctime | TIMESTAMP_LTZ(3) *PROCTIME* | FALSE |     | AS PROCTIME() |                    |~
> ~{+}{-}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{-}{{-}}{-}{{-}}{-}{{-}}{-}{{-}}{-}{-}{-}{+}-------{-}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{+}{{+}}{{-}}{{-}}{{-}}{-}{{-}}{{-}}{{-}}{{-}}{{-}}{-}{{-}}{{-}}{+}{{+}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{-}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{{-}}{-}----------\{+}~
> ~6 rows in set~
> ~+------------------------------------------------+~
> ~|                                     table name |~
> ~+------------------------------------------------+~
> ~|                                 asset_readings |~
> ~|              asset_relationship_parent_to_unit |~
> ~| asset_to_ignore_per_computed_metric_definition |~
> ~|                                         metric |~
> ~+------------------------------------------------+~
> Results:
>  * On Flink 1.15 the difference ({~}assets_supply_air_temp.`value` - assets_setpoint.`value` AS `value`{~}) of the assets_supply_air_temp and assets_setpoint is computed correctly for every value of the assets_supply_air_temp (note that this subquery does not perform any window-based grouping, so it is just raw data)
>  * On Flink 1.18+, for the same query, this difference always results in 0
>  * On Flink 1.18+, updating the query to use regular join against the metric lookup table (removing {~}FOR SYSTEM_TIME AS OF `proctime`{~}) makes the query to output the correct value, however this causes a performance hit as the assets_readings table is built from a Kinesis data stream and the metric table can change over time.
>  * Please see the attached "Task Execution Plan.txt" file to see the difference in temporal joins between Flink 1.15 and Flink 1.20
>  
> I have tried updating the query using different formats with temporal joins but I have not found a workaround and I don't know why this is happening. Attached you will find a file with the different SQL formats I have tried with no luck.
> Any help would be appreciated
>  



--
This message was sent by Atlassian Jira
(v8.20.10#820010)