[
https://issues.apache.org/jira/browse/FLINK-32718?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17766730#comment-17766730
]
Neha Aggarwal edited comment on FLINK-32718 at 9/19/23 10:10 AM:
-----------------------------------------------------------------
[~qingyue] I understand that the checkpoint size will be larger for the job
where there is "union all" but it should not keep on increasing and should show
the same pattern as the individual jobs do. For the individual queries, the
checkpoint size remains bounded and as soon as i put union all, checkpoint size
keeps on increasing.
Please look at the screenshots attached for more information.
was (Author: JIRAUSER299078):
[~qingyue] I understand that the checkpoint size will be larger for the job
where there is "union all" but it should not keep on increasing and should show
the same pattern as the individual jobs do. For the individual queries, the
checkpoint size remains bounded and as soon as i put union all, checkpoint size
keeps on increasing.
> 'UNION ALL' with a 'GROUP BY' condition in Flink is causing the checkpoint
> to become unbounded
> -----------------------------------------------------------------------------------------------
>
> Key: FLINK-32718
> URL: https://issues.apache.org/jira/browse/FLINK-32718
> Project: Flink
> Issue Type: Bug
> Components: Table SQL / API, Table SQL / Planner
> Affects Versions: 1.16.1
> Environment: Apache Flink 1.16.1
> Reporter: Neha Aggarwal
> Assignee: Jayme Howard
> Priority: Major
> Attachments: Screenshot 2023-07-31 at 10.37.11 AM.png, Screenshot
> 2023-07-31 at 10.40.30 AM.png, Screenshot 2023-07-31 at 10.40.38 AM.png,
> Screenshot 2023-07-31 at 10.41.07 AM.png, Screenshot 2023-07-31 at 10.41.26
> AM.png, Screenshot 2023-07-31 at 10.41.35 AM.png, Screenshot 2023-07-31 at
> 10.41.48 AM.png
>
>
> I am having trouble understanding why adding a 'UNION ALL' with a 'GROUP BY'
> condition in Flink is causing the checkpoint to become unbounded. I've
> attached two Flink SQL queries where the only difference is the addition of
> one more 'UNION ALL'. Surprisingly, as soon as we add the 'UNION ALL' part to
> the query, the checkpoint size keeps increasing, which is not expected. We
> anticipated it would follow the same pattern of increasing and decreasing
> based on the throughput, as it does without the 'UNION ALL'.
> When I analyze the breakdown of the checkpoint size, I find that the
> 'Interval Join' operator has the largest size. I suspect there might be a bug
> causing this difference in checkpoint behavior due to the addition of the
> 'GROUP BY' in the query.
> We are using Flink version 1.16.1 with RocksDB as the backend, and
> incremental checkpointing is enabled.
> PS: Interestingly, when I take periodic savepoints for both jobs, their
> patterns are similar, and the savepoint size is smaller than the checkpoint
> in the job with the 'UNION ALL'. Attaching the queries in the comment below.
> With Union ALL:
> select
> *
> from
> (
> with p2d as (
> select
> delivered.job_id,
> delivered.store_id,
> substring(
> GetGeoHash(
> CAST(delivered.metadata_json.address.lat as double),
> CAST(delivered.metadata_json.address.lng as double)
> )
> from
> 1 for 6
> ) AS cx_gh6,
> (
> (
> cast(
> delivered.metadata_json.lastMileDistance as double
> ) / cast(1000 as double)
> ) /(
> cast(
> delivered.updated_at - pickedup.updated_at as double
> ) / cast(60 * 60 * 1000 as double)
> )
> ) as lm_speed_kmph,
> delivered.proctime as proctime
> from
> awz_s3_OrderLogsEvent pickedup
> inner join awz_s3_OrderLogsEvent delivered on delivered.job_id =
> pickedup.job_id
> and delivered.status = 'DELIVERY_DELIVERED'
> and delivered.type = 'INSTA'
> and pickedup.proctime between delivered.proctime - interval '95'
> minute
> and delivered.proctime + interval '5' minute
> where
> pickedup.status = 'DELIVERY_PICKEDUP'
> and pickedup.type = 'INSTA'
> )
> select
> 'lmSpeedKmph_avg_storeId_30m#' || cast(store_id as varchar) as key,
> round(Avg(lm_speed_kmph), 4) AS `value`
> from
> p2d
> group by
> HOP(
> proctime,
> interval '5' minute,
> interval '30' minute
> ),
> store_id
> union all
> select
> 'lmSpeedKmph_avg_cxGh6_30m#' || cast(cx_gh6 as varchar) as key,
> round(Avg(lm_speed_kmph), 4) AS `value`
> from
> p2d
> group by
> HOP(
> proctime,
> interval '5' minute,
> interval '30' minute
> ),
> cx_gh6
> )
>
> Without Union all which is showing expected behaviour:
> select
> *
> from
> (
> with p2d as (
> select
> delivered.job_id,
> delivered.store_id,
> substring(
> GetGeoHash(
> CAST(delivered.metadata_json.address.lat as double),
> CAST(delivered.metadata_json.address.lng as double)
> )
> from
> 1 for 6
> ) AS cx_gh6,
> (
> (
> cast(
> delivered.metadata_json.lastMileDistance as double
> ) / cast(1000 as double)
> ) /(
> cast(
> delivered.updated_at - pickedup.updated_at as double
> ) / cast(60 * 60 * 1000 as double)
> )
> ) as lm_speed_kmph,
> delivered.proctime as proctime
> from
> awz_s3_OrderLogsEvent pickedup
> inner join awz_s3_OrderLogsEvent delivered on delivered.job_id =
> pickedup.job_id
> and delivered.status = 'DELIVERY_DELIVERED'
> and delivered.type = 'INSTA'
> and pickedup.proctime between delivered.proctime - interval '95'
> minute
> and delivered.proctime + interval '5' minute
> where
> pickedup.status = 'DELIVERY_PICKEDUP'
> and pickedup.type = 'INSTA'
> )
> select
> 'lmSpeedKmph_avg_cxGh6_30m#' || cast(cx_gh6 as varchar) as key,
> round(Avg(lm_speed_kmph), 4) AS `value`
> from
> p2d
> group by
> HOP(
> proctime,
> interval '5' minute,
> interval '30' minute
> ),
> cx_gh6
> )
>
>
> !Screenshot 2023-07-31 at 10.37.11 AM.png|width=1113,height=597!
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
