You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by "Eduardo Breijo (Jira)" <ji...@apache.org> on 2024/10/29 20:05:00 UTC
[jira] [Updated] (FLINK-36626) Flink SQL JOINs behavior change from Flink 1.15 to Flink 1.18+
[ https://issues.apache.org/jira/browse/FLINK-36626?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Eduardo Breijo updated FLINK-36626:
-----------------------------------
Description:
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~
*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+ 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
was:
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
> 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.18.1, 1.20.0
> Environment: AWS Managed Apache Flink
> Reporter: Eduardo Breijo
> Priority: Critical
> 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~
> *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+ 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)