[jira] [Created] (FLINK-36626) Flink SQL JOINs behavior change from Flink 1.15 to Flink 1.18+

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

Eduardo Breijo created FLINK-36626:
--------------------------------------

             Summary: Flink SQL JOINs behavior change from Flink 1.15 to Flink 1.18+
                 Key: FLINK-36626
                 URL: https://issues.apache.org/jira/browse/FLINK-36626
             Project: Flink
          Issue Type: Bug
          Components: Table SQL / API
    Affects Versions: 1.20.0, 1.18.1
         Environment: AWS Managed Apache Flink 
            Reporter: Eduardo Breijo
         Attachments: Flink-SQL-query.txt

There is a behavior change I found when migrating to Flink 1.18+ from Flink 1.15 in regards to Flink SQL joins that I haven't been able to pin point and is causing the query 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~
  ~)~
  ~SELECT~
  ~assets_supply_air_temp.`timestamp`,~
  ~assets_supply_air_temp.asset_id,~
  ~assets_supply_air_temp.`value` - assets_setpoint.`value` AS `value`~
  ~FROM (~
    ~SELECT asset_readings.`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~
  ~) AS assets_supply_air_temp~
  ~INNER JOIN assets_setpoint~
  ~ON assets_setpoint.asset_id = assets_supply_air_temp.asset_id~
  ~WHERE assets_supply_air_temp.`timestamp` BETWEEN assets_setpoint.start_timestamp AND assets_setpoint.end_timestamp~

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+ this difference always results in 0

 

I have tried updating the query using different formats 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)