https://issues.apache.org/jira/browse/FLINK-32718 is there any way to get the update on the ticket? It has been a month since a create the bug ticket.

I don’t see necessarily why it’s considered a bug. Checkpoints are internal to Flink, and they depend on things like compaction etc. Besides that, filing a bug doesn’t mean that we can solve them all. We have a couple of thousand open tickets 😅

understood. I really wanted to understand why a union all will change the complete behaviour of the checkpointing. Thanks.

What are your sources?

Kafka

You might want to use explain statement to get the physical plan and see the difference between the two queries

I am attaching the Physical Plan. Let me know if you want the execution plan as well. == Optimized Physical Plan == Union(all=[true], union=[rill_server_timestamp, key, value]) :- Calc(select=[CAST(PROCTIME_MATERIALIZE(PROCTIME()) AS TIMESTAMP(3) NOT NULL) AS rill_server_timestamp, ||(_UTF-16LE'lmSpeedKmph_avg_storeId_30m#', CAST(store_id0 AS VARCHAR(2147483647) CHARACTER SET "UTF-16LE")) AS key, ROUND($f1, 4) AS value]) : +- GroupWindowAggregate(groupBy=[store_id0], window=[SlidingGroupWindow('w$, proctime0, 1800000, 300000)], select=[store_id0, AVG($f2) AS $f1]) : +- Exchange(distribution=[hash[store_id0]]) : +- Calc(select=[proctime0, store_id AS store_id0, /(/(CAST(metadata_json.lastMileDistance AS DOUBLE), 1000:DOUBLE), /(CAST(-(updated_at0, updated_at) AS DOUBLE), 3.6E6:DOUBLE)) AS $f2]) : +- IntervalJoin(joinType=[InnerJoin], windowBounds=[isRowTime=false, leftLowerBound=-5700000, leftUpperBound=300000, leftTimeIndex=2, rightTimeIndex=4], where=[AND(=(job_id0, job_id),>=(proctime, -(proctime0, 5700000:INTERVAL MINUTE)), <=(proctime, +(proctime0, 300000:INTERVAL MINUTE)))], select=[job_id, updated_at, proctime, job_id0, updated_at0, store_id, metadata_json, proctime0]) : :- Exchange(distribution=[hash[job_id]]) : : +- Calc(select=[job_id, updated_at, PROCTIME() AS proctime], where=[AND(=(status, _UTF-16LE'DELIVERY_PICKEDUP':VARCHAR(2147483647) CHARACTER SET "UTF-16LE"), =(type, _UTF-16LE'INSTA':VARCHAR(2147483647) CHARACTER SET "UTF-16LE"))]) : : +- TableSourceScan(table=[[anonymous_datastream_source$10]], fields=[event_id, uuid, time_stamp, schema_version, _server_time_stamp, id, job_id, event_type, update_time, update_json, status, updated_at, type, structured, store_id, deTriggered, metadata_json, proctime]) : +- Exchange(distribution=[hash[job_id]]) : +- Calc(select=[job_id, updated_at, store_id, metadata_json, PROCTIME() AS proctime], where=[AND(=(status, _UTF-16LE'DELIVERY_DELIVERED':VARCHAR(2147483647) CHARACTER SET "UTF-16LE"), =(type, _UTF-16LE'INSTA':VARCHAR(2147483647) CHARACTER SET "UTF-16LE"))]) : +- TableSourceScan(table=[[anonymous_datastream_source$10]], fields=[event_id, uuid, time_stamp, schema_version, _server_time_stamp, id, job_id, event_type, update_time, update_json, status, updated_at, type, structured, store_id, deTriggered, metadata_json, proctime]) +- Calc(select=[CAST(PROCTIME_MATERIALIZE(PROCTIME()) AS TIMESTAMP(3) NOT NULL) AS rill_server_timestamp, ||(_UTF-16LE'lmSpeedKmph_avg_cxGh6_30m#', $f1) AS key, ROUND($f1_0, 4) AS value]) +- GroupWindowAggregate(groupBy=[$f1], window=[SlidingGroupWindow('w$, proctime0, 1800000, 300000)], select=[$f1, AVG($f2) AS $f1_0]) +- Exchange(distribution=[hash[$f1]]) +- Calc(select=[proctime0, SUBSTRING(GetGeoHash(metadata_json.address.lat, metadata_json.address.lng), 1, 6) AS $f1, /(/(CAST(metadata_json.lastMileDistance AS DOUBLE), 1000:DOUBLE), /(CAST(-(updated_at0, updated_at) AS DOUBLE), 3.6E6:DOUBLE)) AS $f2]) +- IntervalJoin(joinType=[InnerJoin], windowBounds=[isRowTime=false, leftLowerBound=-5700000, leftUpperBound=300000, leftTimeIndex=2, rightTimeIndex=3], where=[AND(=(job_id0, job_id),>=(proctime, -(proctime0, 5700000:INTERVAL MINUTE)), <=(proctime, +(proctime0, 300000:INTERVAL MINUTE)))], select=[job_id, updated_at, proctime, job_id0, updated_at0, metadata_json, proctime0]) :- Exchange(distribution=[hash[job_id]]) : +- Calc(select=[job_id, updated_at, PROCTIME() AS proctime], where=[AND(=(status, _UTF-16LE'DELIVERY_PICKEDUP':VARCHAR(2147483647) CHARACTER SET "UTF-16LE"), =(type, _UTF-16LE'INSTA':VARCHAR(2147483647) CHARACTER SET "UTF-16LE"))]) : +- TableSourceScan(table=[[anonymous_datastream_source$10]], fields=[event_id, uuid, time_stamp, schema_version, _server_time_stamp, id, job_id, event_type, update_time, update_json, status, updated_at, type, structured, store_id, deTriggered, metadata_json, proctime]) +- Exchange(distribution=[hash[job_id]]) +- Calc(select=[job_id, updated_at, metadata_json, PROCTIME() AS proctime], where=[AND(=(status, _UTF-16LE'DELIVERY_DELIVERED':VARCHAR(2147483647) CHARACTER SET "UTF-16LE"), =(type, _UTF-16LE'INSTA':VARCHAR(2147483647) CHARACTER SET "UTF-16LE"))]) +- TableSourceScan(table=[[anonymous_datastream_source$10]], fields=[event_id, uuid, time_stamp, schema_version, _server_time_stamp, id, job_id, event_type, update_time, update_json, status, updated_at, type, structured, store_id, deTriggered, metadata_json, proctime]) without union all query: == Optimized Physical Plan == Calc(select=[CAST(PROCTIME_MATERIALIZE(PROCTIME()) AS TIMESTAMP(3) NOT NULL) AS rill_server_timestamp, ||(_UTF-16LE'lmSpeedKmph_avg_cxGh6_30m#', $f1) AS key, ROUND($f1_0, 4) AS value]) +- GroupWindowAggregate(groupBy=[$f1], window=[SlidingGroupWindow('w$, proctime0, 1800000, 300000)], select=[$f1, AVG($f2) AS $f1_0]) +- Exchange(distribution=[hash[$f1]]) +- Calc(select=[proctime0, SUBSTRING(GetGeoHash(metadata_json.address.lat, metadata_json.address.lng), 1, 6) AS $f1, /(/(CAST(metadata_json.lastMileDistance AS DOUBLE), 1000:DOUBLE), /(CAST(-(updated_at0, updated_at) AS DOUBLE), 3.6E6:DOUBLE)) AS $f2]) +- IntervalJoin(joinType=[InnerJoin], windowBounds=[isRowTime=false, leftLowerBound=-5700000, leftUpperBound=300000, leftTimeIndex=2, rightTimeIndex=3], where=[AND(=(job_id0, job_id), >=(proctime, -(proctime0, 5700000:INTERVAL MINUTE)), <=(proctime, +(proctime0, 300000:INTERVAL MINUTE)))], select=[job_id, updated_at, proctime, job_id0, updated_at0, metadata_json, proctime0]) :- Exchange(distribution=[hash[job_id]]) : +- Calc(select=[job_id, updated_at, PROCTIME() AS proctime], where=[AND(=(status, _UTF-16LE'DELIVERY_PICKEDUP':VARCHAR(2147483647) CHARACTER SET "UTF-16LE"), =(type, _UTF-16LE'INSTA':VARCHAR(2147483647) CHARACTER SET "UTF-16LE"))]) : +- TableSourceScan(table=[[anonymous_datastream_source$11]], fields=[event_id, uuid, time_stamp, schema_version, _server_time_stamp, id, job_id, event_type, update_time, update_json, status, updated_at, type, structured, store_id, deTriggered, metadata_json, proctime]) +- Exchange(distribution=[hash[job_id]]) +- Calc(select=[job_id, updated_at, metadata_json, PROCTIME() AS proctime], where=[AND(=(status, _UTF-16LE'DELIVERY_DELIVERED':VARCHAR(2147483647) CHARACTER SET "UTF-16LE"), =(type, _UTF-16LE'INSTA':VARCHAR(2147483647) CHARACTER SET "UTF-16LE"))]) +- TableSourceScan(table=[[anonymous_datastream_source$11]], fields=[event_id, uuid, time_stamp, schema_version, _server_time_stamp, id, job_id, event_type, update_time, update_json, status, updated_at, type, structured, store_id, deTriggered, metadata_json, proctime])