parts_active_at_ingest_finish.sql

WITH
    'default' AS db_name,
    'pypi' AS table_name,
    (
        SELECT event_time
        FROM clusterAllReplicas(default, system.query_log)
        WHERE
            has(tables, db_name || '.' || table_name)
            AND length(tables) = 1
            AND is_initial_query
            AND query_kind = 'Insert'
            AND type = 'QueryFinish'
        ORDER BY event_time_microseconds DESC
        LIMIT 1
    ) AS query_end_time
SELECT
    max(event_time) as _event_time,
    argMax(value, event_time) as parts
FROM clusterAllReplicas(default, system.asynchronous_metric_log)
WHERE
    toStartOfInterval(event_time, toIntervalSecond(60)) = toStartOfInterval(query_end_time, toIntervalSecond(60))
    AND metric = 'MaxPartCountForPartition'
SETTINGS skip_unavailable_shards = 1;