Performance regression on Iceberg Metadata queries on `$files` table

[dotjdk]

It seems that #25677 has indeed improved performance of queries on $files tables, but only for the first 1 to 2 executions after a cluster restart. The first 1-2 executions takes less than a second, but after that it consistently takes 1 minute for the same query.

The cumulative user memory in the first execution is 11-12 MB but 1.72GB in subsequent executions. In stage 0 CPU time is rougly the same, but blocked time goes from 3 minutes to 6 hours.

We can reproduce this behavior consistently after restarting the cluster. Downgrading to 476 eliminates the issue, and the query consistently completes in about 1.5 seconds every time.

[raunaqmorarka]

Can you provide output of EXPLAIN ANALYZE VERBOSE for the fast and slow runs ?
cc: @tbaeg

[dotjdk]

@raunaqmorarka, @tbaeg

Here you go.

Fast Execution:

Trino version: 477
Queued: 7.01ms, Analysis: 485.44ms, Planning: 194.80ms, Execution: 2.04s
Fragment 1 [HASH]
    CPU: 1.49s, Scheduled: 3.04s, Blocked 7.32m (Input: 3.67m, Output: 0.00ns), Input: 6010 rows (228.90kB); per task: avg.: 1001.67 std.dev.: 29.53, Output: 2695 rows (76.32kB)
    Peak Memory: 108.25kB, Tasks count: 6; per task: max: 4.88MB
    Output buffer active time: 88.12ms, buffer utilization distribution (%): {p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.04, p75=0.05, p90=0.05, p95=0.06, p99=0.06, min=0.00, max=0.06}
    Task output distribution: {count=6.00, p01=12.15kB, p05=12.15kB, p10=12.15kB, p25=12.21kB, p50=12.83kB, p75=13.06kB, p90=13.34kB, p95=13.34kB, p99=13.34kB, max=13.34kB}
    Task input distribution: {count=6.00, p01=36.68kB, p05=36.68kB, p10=36.68kB, p25=36.91kB, p50=38.54kB, p75=38.58kB, p90=40.03kB, p95=40.03kB, p99=40.03kB, max=40.03kB}
    Output layout: [partition, count, avg]
    Output partitioning: SINGLE []
    Aggregate[type = FINAL, keys = [partition]]
    │   Layout: [partition:row(timestamp_day date, redacted_key_bucket_8 integer), count:bigint, avg:double]
    │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: 0B}
    │   CPU: 799.00ms (0.48%), Scheduled: 1.95s (0.28%), Blocked: 0.00ns (0.00%), Output: 2695 rows (76.32kB)
    │   metrics:
    │     'Accumulator update CPU time' = {duration=163.02ms}
    │     'CPU time distribution (s)' = {count=192, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.01, p95=0.01, p99=0.02, min=0.00, max=0.02}
    │     'Group by hash update CPU time' = {duration=428.77ms}
    │     'Input rows distribution' = {count=192, p01=11.00, p05=15.00, p10=20.00, p25=26.00, p50=30.98, p75=37.00, p90=43.00, p95=46.00, p99=53.00, min=7.00, max=56.00}
    │     'Input rows processed without partial aggregation enabled' = LongCount{total=0}
    │     'Scheduled time distribution (s)' = {count=192, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.01, p90=0.03, p95=0.03, p99=0.04, min=0.00, max=0.04}
    │   Input avg.: 31.30 rows, Input std.dev.: 28.67%
    │   count := count(count_0)
    │   avg := avg(avg_1)
    └─ LocalExchange[partitioning = HASH, arguments = [partition::row(timestamp_day date, redacted_key_bucket_8 integer)]]
       │   Layout: [partition:row(timestamp_day date, redacted_key_bucket_8 integer), count_0:bigint, avg_1:row(double, bigint)]
       │   Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B}
       │   CPU: 264.00ms (0.16%), Scheduled: 330.00ms (0.05%), Blocked: 3.65m (49.89%), Output: 6010 rows (228.90kB)
       │   metrics:
       │     'CPU time distribution (s)' = {count=192, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.00, min=0.00, max=0.00}
       │     'Input rows distribution' = {count=192, p01=4.00, p05=7.00, p10=9.00, p25=15.11, p50=24.08, p75=41.25, p90=62.60, p95=78.00, p99=115.00, min=2.00, max=115.00}
       │     'Scheduled time distribution (s)' = {count=192, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.01, min=0.00, max=0.02}
       │   Input avg.: 31.30 rows, Input std.dev.: 71.56%
       └─ RemoteSource[sourceFragmentIds = [2]]
              Layout: [partition:row(timestamp_day date, redacted_key_bucket_8 integer), count_0:bigint, avg_1:row(double, bigint)]
              CPU: 182.00ms (0.11%), Scheduled: 401.00ms (0.06%), Blocked: 3.67m (50.11%), Output: 6010 rows (228.90kB)
              metrics:
                'Blocked time distribution (s)' = {count=192, p01=1.10, p05=1.12, p10=1.12, p25=1.14, p50=1.14, p75=1.16, p90=1.17, p95=1.18, p99=1.18, min=1.09, max=1.18}
                'CPU time distribution (s)' = {count=192, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.00, min=0.00, max=0.00}
                'Input rows distribution' = {count=192, p01=4.00, p05=7.00, p10=9.00, p25=15.11, p50=24.08, p75=41.25, p90=62.60, p95=78.00, p99=115.00, min=2.00, max=115.00}
                'Scheduled time distribution (s)' = {count=192, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.01, min=0.00, max=0.07}
              Input avg.: 31.30 rows, Input std.dev.: 71.56%

Fragment 2 [SOURCE]
    CPU: 2.74m, Scheduled: 11.55m, Blocked 0.00ns (Input: 0.00ns, Output: 0.00ns), Input: 6185 rows (120.80kB); per task: avg.: 1030.83 std.dev.: 201.12, Output: 6010 rows (274.68kB)
    Peak Memory: 6.03MB, Tasks count: 6; per task: max: 1.01MB
    Output buffer active time: 1.36s, buffer utilization distribution (%): {p01=0.00, p05=0.00, p10=0.00, p25=0.01, p50=0.01, p75=0.03, p90=0.05, p95=0.07, p99=0.17, min=0.00, max=0.17}
    Task output distribution: {count=6.00, p01=33.63kB, p05=33.63kB, p10=33.63kB, p25=38.25kB, p50=47.35kB, p75=52.26kB, p90=60.53kB, p95=60.53kB, p99=60.53kB, max=60.53kB}
    Task input distribution: {count=6.00, p01=14.79kB, p05=14.79kB, p10=14.79kB, p25=16.64kB, p50=21.50kB, p75=22.70kB, p90=26.54kB, p95=26.54kB, p99=26.54kB, max=26.54kB}
    Output layout: [partition, count_0, avg_1]
    Output partitioning: HASH [partition]
    Aggregate[type = PARTIAL, keys = [partition]]
    │   Layout: [partition:row(timestamp_day date, redacted_key_bucket_8 integer), count_0:bigint, avg_1:row(double, bigint)]
    │   CPU: 1.84s (1.11%), Scheduled: 5.21s (0.75%), Blocked: 0.00ns (0.00%), Output: 6010 rows (274.68kB)
    │   metrics:
    │     'Accumulator update CPU time' = {duration=133.73ms}
    │     'CPU time distribution (s)' = {count=693, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.01, p95=0.01, p99=0.01, min=0.00, max=0.02}
    │     'Group by hash update CPU time' = {duration=223.53ms}
    │     'Input rows distribution' = {count=693, p01=1.00, p05=1.00, p10=1.00, p25=1.04, p50=2.00, p75=7.11, p90=15.93, p95=32.63, p99=111.00, min=0.00, max=400.00}
    │     'Input rows processed without partial aggregation enabled' = LongCount{total=0}
    │     'Scheduled time distribution (s)' = {count=693, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.02, p95=0.04, p99=0.04, min=0.00, max=0.04}
    │   Input avg.: 8.92 rows, Input std.dev.: 291.04%
    │   count_0 := count(*)
    │   avg_1 := avg(file_size_in_bytes)
    └─ TableScan[table = redacted_conn.redacted_table$files]
           Layout: [partition:row(timestamp_day date, redacted_key_bucket_8 integer), file_size_in_bytes:bigint]
           Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B}
           CPU: 2.71m (98.14%), Scheduled: 11.48m (98.87%), Blocked: 0.00ns (0.00%), Output: 6185 rows (120.80kB)
           metrics:
             'CPU time distribution (s)' = {count=693, p01=0.16, p05=0.17, p10=0.18, p25=0.20, p50=0.22, p75=0.28, p90=0.30, p95=0.31, p99=0.34, min=0.14, max=0.35}
             'Input rows distribution' = {count=693, p01=1.00, p05=1.00, p10=1.00, p25=1.02, p50=2.00, p75=7.20, p90=16.10, p95=33.60, p99=111.00, min=0.00, max=400.00}
             'Scheduled time distribution (s)' = {count=693, p01=0.82, p05=0.87, p10=0.89, p25=0.94, p50=1.00, p75=1.05, p90=1.07, p95=1.09, p99=1.14, min=0.80, max=1.15}
           Input avg.: 8.92 rows, Input std.dev.: 291.04%
           partition := partition
           file_size_in_bytes := file_size_in_bytes
           Input: 6185 rows (120.80kB), Physical input: 6.36MB, Physical input time: 2.14s, Splits: 693, Splits generation wait time: 0.17ms

Slow Execution:

Trino version: 477
Queued: 224.21us, Analysis: 293.74ms, Planning: 19.48ms, Execution: 1.01m
Fragment 1 [HASH]
    CPU: 631.85ms, Scheduled: 749.65ms, Blocked 6.42h (Input: 3.22h, Output: 0.00ns), Input: 6010 rows (228.90kB); per task: avg.: 1001.67 std.dev.: 29.53, Output: 2695 rows (76.32kB)
    Peak Memory: 29.27MB, Tasks count: 6; per task: max: 4.88MB
    Output buffer active time: 35.19ms, buffer utilization distribution (%): {p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.02, p75=0.05, p90=0.06, p95=0.06, p99=0.06, min=0.00, max=0.06}
    Task output distribution: {count=6.00, p01=12.15kB, p05=12.15kB, p10=12.15kB, p25=12.21kB, p50=12.83kB, p75=13.06kB, p90=13.34kB, p95=13.34kB, p99=13.34kB, max=13.34kB}
    Task input distribution: {count=6.00, p01=36.68kB, p05=36.68kB, p10=36.68kB, p25=36.91kB, p50=38.54kB, p75=38.58kB, p90=40.03kB, p95=40.03kB, p99=40.03kB, max=40.03kB}
    Output layout: [partition, count, avg]
    Output partitioning: SINGLE []
    Aggregate[type = FINAL, keys = [partition]]
    │   Layout: [partition:row(timestamp_day date, redacted_key_bucket_8 integer), count:bigint, avg:double]
    │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: 0B}
    │   CPU: 310.00ms (2.28%), Scheduled: 327.00ms (0.04%), Blocked: 0.00ns (0.00%), Output: 2695 rows (76.32kB)
    │   metrics:
    │     'Accumulator update CPU time' = {duration=21.22ms}
    │     'CPU time distribution (s)' = {count=192, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.00, min=0.00, max=0.00}
    │     'Group by hash update CPU time' = {duration=91.86ms}
    │     'Input rows distribution' = {count=192, p01=11.00, p05=15.00, p10=20.00, p25=26.00, p50=31.10, p75=36.90, p90=43.00, p95=46.00, p99=53.00, min=7.00, max=56.00}
    │     'Input rows processed without partial aggregation enabled' = LongCount{total=0}
    │     'Scheduled time distribution (s)' = {count=192, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.00, min=0.00, max=0.00}
    │   Input avg.: 31.30 rows, Input std.dev.: 28.67%
    │   count := count(count_0)
    │   avg := avg(avg_1)
    └─ LocalExchange[partitioning = HASH, arguments = [partition::row(timestamp_day date, redacted_key_bucket_8 integer)]]
       │   Layout: [partition:row(timestamp_day date, redacted_key_bucket_8 integer), count_0:bigint, avg_1:row(double, bigint)]
       │   Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B}
       │   CPU: 120.00ms (0.88%), Scheduled: 151.00ms (0.02%), Blocked: 3.22h (50.00%), Output: 6010 rows (228.90kB)
       │   metrics:
       │     'CPU time distribution (s)' = {count=192, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.00, min=0.00, max=0.00}
       │     'Input rows distribution' = {count=192, p01=8.00, p05=12.00, p10=14.00, p25=19.00, p50=25.25, p75=36.13, p90=55.00, p95=76.00, p99=105.00, min=8.00, max=114.00}
       │     'Scheduled time distribution (s)' = {count=192, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.01, min=0.00, max=0.01}
       │   Input avg.: 31.30 rows, Input std.dev.: 63.12%
       └─ RemoteSource[sourceFragmentIds = [2]]
              Layout: [partition:row(timestamp_day date, redacted_key_bucket_8 integer), count_0:bigint, avg_1:row(double, bigint)]
              CPU: 90.00ms (0.66%), Scheduled: 121.00ms (0.02%), Blocked: 3.22h (50.00%), Output: 6010 rows (228.90kB)
              metrics:
                'Blocked time distribution (s)' = {count=192, p01=60.32, p05=60.32, p10=60.32, p25=60.32, p50=60.33, p75=60.33, p90=60.33, p95=60.33, p99=60.33, min=60.32, max=60.33}
                'CPU time distribution (s)' = {count=192, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.00, min=0.00, max=0.00}
                'Input rows distribution' = {count=192, p01=8.00, p05=12.00, p10=14.00, p25=19.00, p50=25.25, p75=36.13, p90=55.00, p95=76.00, p99=105.00, min=8.00, max=114.00}
                'Scheduled time distribution (s)' = {count=192, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.01, min=0.00, max=0.01}
              Input avg.: 31.30 rows, Input std.dev.: 63.12%

Fragment 2 [SOURCE]
    CPU: 13.08s, Scheduled: 13.09m, Blocked 0.00ns (Input: 0.00ns, Output: 0.00ns), Input: 6185 rows (120.80kB); per task: avg.: 1030.83 std.dev.: 80.14, Output: 6010 rows (274.68kB)
    Peak Memory: 5.77MB, Tasks count: 6; per task: max: 1.01MB
    Output buffer active time: 4.02m, buffer utilization distribution (%): {p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.00, min=0.00, max=0.06}
    Task output distribution: {count=6.00, p01=40.06kB, p05=40.06kB, p10=40.06kB, p25=44.25kB, p50=45.71kB, p75=47.41kB, p90=51.66kB, p95=51.66kB, p99=51.66kB, max=51.66kB}
    Task input distribution: {count=6.00, p01=17.48kB, p05=17.48kB, p10=17.48kB, p25=19.24kB, p50=20.82kB, p75=20.96kB, p90=22.50kB, p95=22.50kB, p99=22.50kB, max=22.50kB}
    Output layout: [partition, count_0, avg_1]
    Output partitioning: HASH [partition]
    Aggregate[type = PARTIAL, keys = [partition]]
    │   Layout: [partition:row(timestamp_day date, redacted_key_bucket_8 integer), count_0:bigint, avg_1:row(double, bigint)]
    │   CPU: 358.00ms (2.64%), Scheduled: 423.00ms (0.05%), Blocked: 0.00ns (0.00%), Output: 6010 rows (274.68kB)
    │   metrics:
    │     'Accumulator update CPU time' = {duration=13.05ms}
    │     'CPU time distribution (s)' = {count=693, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.00, min=0.00, max=0.00}
    │     'Group by hash update CPU time' = {duration=21.45ms}
    │     'Input rows distribution' = {count=693, p01=0.46, p05=1.00, p10=1.00, p25=1.00, p50=2.00, p75=7.18, p90=16.20, p95=33.65, p99=111.00, min=0.00, max=400.00}
    │     'Input rows processed without partial aggregation enabled' = LongCount{total=0}
    │     'Scheduled time distribution (s)' = {count=693, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.00, min=0.00, max=0.01}
    │   Input avg.: 8.92 rows, Input std.dev.: 291.04%
    │   count_0 := count(*)
    │   avg_1 := avg(file_size_in_bytes)
    └─ TableScan[table = redacted_conn.redacted_table$files]
           Layout: [partition:row(timestamp_day date, redacted_key_bucket_8 integer), file_size_in_bytes:bigint]
           Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B}
           CPU: 12.70s (93.53%), Scheduled: 13.09m (99.87%), Blocked: 0.00ns (0.00%), Output: 6185 rows (120.80kB)
           metrics:
             'CPU time distribution (s)' = {count=693, p01=0.01, p05=0.01, p10=0.01, p25=0.01, p50=0.02, p75=0.02, p90=0.03, p95=0.03, p99=0.04, min=0.01, max=0.06}
             'Input rows distribution' = {count=693, p01=0.46, p05=1.00, p10=1.00, p25=1.00, p50=2.00, p75=7.18, p90=16.20, p95=33.65, p99=111.00, min=0.00, max=400.00}
             'Scheduled time distribution (s)' = {count=693, p01=0.13, p05=0.22, p10=0.25, p25=0.30, p50=0.35, p75=0.43, p90=0.46, p95=0.47, p99=60.13, min=0.07, max=60.31}
           Input avg.: 8.92 rows, Input std.dev.: 291.04%
           partition := partition
           file_size_in_bytes := file_size_in_bytes
           Input: 6185 rows (120.80kB), Physical input: 6.36MB, Physical input time: 103.85ms, Splits: 693, Splits generation wait time: 0.02ms

[tbaeg]

I am not able to reproduce. Do repeated simple select queries have the same behaviors? Can you provide an EXPLAIN ANALYZE output for the query below as well?

SELECT * FROM "your_table$files" LIMIT 10;

[dotjdk]

No, a simple select runs consistently in under 1 second. I did test other GROUP BY queries on the standard table to rule out the changes to GROUP BY in 477, and I didn't notice similar behavior there either.

Trino version: 477
Queued: 65.63us, Analysis: 205.04ms, Planning: 3.11ms, Execution: 377.57ms
Fragment 1 [SINGLE]
    CPU: 1.64ms, Scheduled: 1.71ms, Blocked 5.44s (Input: 5.28s, Output: 0.00ns), Input: 10 rows (37.23kB); per task: avg.: 10.00 std.dev.: 0.00, Output: 10 rows (37.23kB)
    Peak Memory: 0B, Tasks count: 1; per task: max: 2.60kB
    Output buffer active time: 119.71ms, buffer utilization distribution (%): {p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.02, min=0.00, max=0.06}
    Task output distribution: {count=1.00, p01=37.23kB, p05=37.23kB, p10=37.23kB, p25=37.23kB, p50=37.23kB, p75=37.23kB, p90=37.23kB, p95=37.23kB, p99=37.23kB, max=37.23kB}
    Task input distribution: {count=1.00, p01=37.23kB, p05=37.23kB, p10=37.23kB, p25=37.23kB, p50=37.23kB, p75=37.23kB, p90=37.23kB, p95=37.23kB, p99=37.23kB, max=37.23kB}
    Output layout: [content, file_path, file_format, spec_id, partition, record_count, file_size_in_bytes, column_sizes, value_counts, null_value_counts, nan_value_counts, lower_bounds, upper_bounds, key_metadata, split_offsets, equality_ids, sort_order_id, readable_metrics]
    Output partitioning: SINGLE []
    Limit[count = 10]
    │   Layout: [content:integer, file_path:varchar, file_format:varchar, spec_id:integer, partition:row(timestamp_day date, redacted_key_bucket_8 integer), record_count:bigint, file_size_in_bytes:bigint, column_sizes:map(integer, bigint), value_counts:map(integer, bigint), null_value_counts:map(integer, bigint), nan_value_counts:map(integer, bigint), lower_bounds:map(integer, varchar), upper_bounds:map(integer, varchar), key_metadata:varbinary, split_offsets:array(bigint), equality_ids:array(integer), sort_order_id:integer, readable_metrics:json]
    │   Estimates: {rows: 10 (7.30kB), cpu: 7.30k, memory: 0B, network: 0B}
    │   CPU: 0.00ns (0.00%), Scheduled: 0.00ns (0.00%), Blocked: 0.00ns (0.00%), Output: 10 rows (37.23kB)
    │   metrics:
    │     'CPU time distribution (s)' = {count=1, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.00, min=0.00, max=0.00}
    │     'Input rows distribution' = {count=1, p01=10.00, p05=10.00, p10=10.00, p25=10.00, p50=10.00, p75=10.00, p90=10.00, p95=10.00, p99=10.00, min=10.00, max=10.00}
    │     'Scheduled time distribution (s)' = {count=1, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.00, min=0.00, max=0.00}
    │   Input avg.: 10.00 rows, Input std.dev.: 0.00%
    └─ LocalExchange[partitioning = SINGLE]
       │   Layout: [content:integer, file_path:varchar, file_format:varchar, spec_id:integer, partition:row(timestamp_day date, redacted_key_bucket_8 integer), record_count:bigint, file_size_in_bytes:bigint, column_sizes:map(integer, bigint), value_counts:map(integer, bigint), null_value_counts:map(integer, bigint), nan_value_counts:map(integer, bigint), lower_bounds:map(integer, varchar), upper_bounds:map(integer, varchar), key_metadata:varbinary, split_offsets:array(bigint), equality_ids:array(integer), sort_order_id:integer, readable_metrics:json]
       │   Estimates: {rows: 10 (7.30kB), cpu: 0, memory: 0B, network: 0B}
       │   CPU: 0.00ns (0.00%), Scheduled: 0.00ns (0.00%), Blocked: 164.00ms (3.01%), Output: 10 rows (37.23kB)
       │   metrics:
       │     'CPU time distribution (s)' = {count=32, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.00, min=0.00, max=0.00}
       │     'Input rows distribution' = {count=32, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=1.00, p95=2.00, p99=4.00, min=0.00, max=4.00}
       │     'Scheduled time distribution (s)' = {count=32, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.00, min=0.00, max=0.00}
       │   Input avg.: 0.31 rows, Input std.dev.: 258.46%
       └─ RemoteSource[sourceFragmentIds = [2]]
              Layout: [content:integer, file_path:varchar, file_format:varchar, spec_id:integer, partition:row(timestamp_day date, redacted_key_bucket_8 integer), record_count:bigint, file_size_in_bytes:bigint, column_sizes:map(integer, bigint), value_counts:map(integer, bigint), null_value_counts:map(integer, bigint), nan_value_counts:map(integer, bigint), lower_bounds:map(integer, varchar), upper_bounds:map(integer, varchar), key_metadata:varbinary, split_offsets:array(bigint), equality_ids:array(integer), sort_order_id:integer, readable_metrics:json]
              CPU: 1.00ms (3.33%), Scheduled: 1.00ms (0.05%), Blocked: 5.28s (96.99%), Output: 10 rows (37.23kB)
              metrics:
                'Blocked time distribution (s)' = {count=32, p01=0.16, p05=0.16, p10=0.16, p25=0.16, p50=0.16, p75=0.16, p90=0.17, p95=0.17, p99=0.17, min=0.16, max=0.17}
                'CPU time distribution (s)' = {count=32, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.00, min=0.00, max=0.00}
                'Input rows distribution' = {count=32, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=1.00, p95=2.00, p99=4.00, min=0.00, max=4.00}
                'Scheduled time distribution (s)' = {count=32, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.00, min=0.00, max=0.00}
              Input avg.: 0.31 rows, Input std.dev.: 258.46%

Fragment 2 [SOURCE]
    CPU: 851.76ms, Scheduled: 2.65m, Blocked 0.00ns (Input: 0.00ns, Output: 0.00ns), Input: 56 rows (216.66kB); per task: avg.: 9.33 std.dev.: 14.76, Output: 45 rows (173.55kB)
    Peak Memory: 75.80kB, Tasks count: 6; per task: max: 18.21kB
    Output buffer active time: 788.96ms, buffer utilization distribution (%): {p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.00, min=0.00, max=0.33}
    Task output distribution: {count=6.00, p01=0B, p05=0B, p10=0B, p25=0B, p50=6.16kB, p75=31.20kB, p90=132.31kB, p95=132.31kB, p99=132.31kB, max=132.31kB}
    Task input distribution: {count=6.00, p01=0B, p05=0B, p10=0B, p25=0B, p50=6.16kB, p75=46.87kB, p90=159.75kB, p95=159.75kB, p99=159.75kB, max=159.75kB}
    Amount of input data processed by the workers for this stage might be skewed
    Output layout: [content, file_path, file_format, spec_id, partition, record_count, file_size_in_bytes, column_sizes, value_counts, null_value_counts, nan_value_counts, lower_bounds, upper_bounds, key_metadata, split_offsets, equality_ids, sort_order_id, readable_metrics]
    Output partitioning: SINGLE []
    LimitPartial[count = 10]
    │   Layout: [content:integer, file_path:varchar, file_format:varchar, spec_id:integer, partition:row(timestamp_day date, redacted_key_bucket_8 integer), record_count:bigint, file_size_in_bytes:bigint, column_sizes:map(integer, bigint), value_counts:map(integer, bigint), null_value_counts:map(integer, bigint), nan_value_counts:map(integer, bigint), lower_bounds:map(integer, varchar), upper_bounds:map(integer, varchar), key_metadata:varbinary, split_offsets:array(bigint), equality_ids:array(integer), sort_order_id:integer, readable_metrics:json]
    │   Estimates: {rows: 10 (7.30kB), cpu: 7.30k, memory: 0B, network: 0B}
    │   CPU: 0.00ns (0.00%), Scheduled: 0.00ns (0.00%), Blocked: 0.00ns (0.00%), Output: 45 rows (173.55kB)
    │   metrics:
    │     'CPU time distribution (s)' = {count=692, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.00, min=0.00, max=0.00}
    │     'Input rows distribution' = {count=692, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=2.00, min=0.00, max=17.00}
    │     'Scheduled time distribution (s)' = {count=692, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.00, min=0.00, max=0.00}
    │   Input avg.: 0.08 rows, Input std.dev.: 1035.01%
    └─ TableScan[table = redacted_conniceberg.redacted_tablefiles]
           Layout: [content:integer, file_path:varchar, file_format:varchar, spec_id:integer, partition:row(timestamp_day date, redacted_key_bucket_8 integer), record_count:bigint, file_size_in_bytes:bigint, column_sizes:map(integer, bigint), value_counts:map(integer, bigint), null_value_counts:map(integer, bigint), nan_value_counts:map(integer, bigint), lower_bounds:map(integer, varchar), upper_bounds:map(integer, varchar), key_metadata:varbinary, split_offsets:array(bigint), equality_ids:array(integer), sort_order_id:integer, readable_metrics:json]
           Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B}
           CPU: 29.00ms (96.67%), Scheduled: 2.05s (99.95%), Blocked: 0.00ns (0.00%), Output: 56 rows (216.66kB)
           metrics:
             'CPU time distribution (s)' = {count=692, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.00, min=0.00, max=0.00}
             'Input rows distribution' = {count=692, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=2.00, min=0.00, max=17.00}
             'Scheduled time distribution (s)' = {count=692, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.16, min=0.00, max=0.16}
           Input avg.: 0.08 rows, Input std.dev.: 1035.01%
           split_offsets := split_offsets
           sort_order_id := sort_order_id
           upper_bounds := upper_bounds
           equality_ids := equality_ids
           file_size_in_bytes := file_size_in_bytes
           file_format := file_format
           spec_id := spec_id
           value_counts := value_counts
           nan_value_counts := nan_value_counts
           null_value_counts := null_value_counts
           key_metadata := key_metadata
           content := content
           readable_metrics := readable_metrics
           partition := partition
           lower_bounds := lower_bounds
           file_path := file_path
           record_count := record_count
           column_sizes := column_sizes
           Input: 56 rows (216.66kB), Physical input: 123.26kB, Physical input time: 3.61ms, Splits: 692, Splits generation wait time: 0.01ms

[dotjdk]

(See update in next post before this one)

I don't know if it is helpful, but I just tested select count(distinct partition) from "my_table$files" and that one is consistently taking 1 minute exactly. Even after a fresh restart of the cluster.

Usually, I can do 2 fast GROUP BY after a restart before those goes to 1 minute. This time I did

• one count (distinct) (1 minute)
• one group by (< 1 sec)
• one group by (1 minute)

The table has 2695 partitions: day(timestamp), bucket(8, redacted_key), 6201 files in total.

Here's the explain for the count(distinct partition):

Trino version: 477
Queued: 148.00us, Analysis: 228.98ms, Planning: 11.64ms, Execution: 1.01m
Fragment 1 [SINGLE]
    CPU: 5.52ms, Scheduled: 11.56ms, Blocked 33.20m (Input: 32.20m, Output: 0.00ns), Input: 192 rows (1.69kB); per task: avg.: 192.00 std.dev.: 0.00, Output: 1 row (9B)
    Peak Memory: 2.60kB, Tasks count: 1; per task: max: 2.60kB
    Output buffer active time: 1.64ms, buffer utilization distribution (%): {p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.00, min=0.00, max=0.00}
    Task output distribution: {count=1.00, p01=9B, p05=9B, p10=9B, p25=9B, p50=9B, p75=9B, p90=9B, p95=9B, p99=9B, max=9B}
    Task input distribution: {count=1.00, p01=1.69kB, p05=1.69kB, p10=1.69kB, p25=1.69kB, p50=1.69kB, p75=1.69kB, p90=1.69kB, p95=1.69kB, p99=1.69kB, max=1.69kB}
    Output layout: [count]
    Output partitioning: SINGLE []
    Aggregate[type = FINAL]
    │   Layout: [count:bigint]
    │   Estimates: {rows: 1 (9B), cpu: ?, memory: 9B, network: 0B}
    │   CPU: 1.00ms (0.02%), Scheduled: 1.00ms (0.00%), Blocked: 0.00ns (0.00%), Output: 1 row (9B)
    │   metrics:
    │     'Accumulator update CPU time' = {duration=391.23us}
    │     'CPU time distribution (s)' = {count=1, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.00, min=0.00, max=0.00}
    │     'Group by hash update CPU time' = {duration=0.00ns}
    │     'Input rows distribution' = {count=1, p01=192.00, p05=192.00, p10=192.00, p25=192.00, p50=192.00, p75=192.00, p90=192.00, p95=192.00, p99=192.00, min=192.00, max=192.00}
    │     'Input rows processed without partial aggregation enabled' = LongCount{total=0}
    │     'Scheduled time distribution (s)' = {count=1, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.00, min=0.00, max=0.00}
    │   Input avg.: 192.00 rows, Input std.dev.: 0.00%
    │   count := count(count_0)
    └─ LocalExchange[partitioning = SINGLE]
       │   Layout: [count_0:bigint]
       │   Estimates: {rows: ? (?), cpu: 0, memory: 0B, network: 0B}
       │   CPU: 0.00ns (0.00%), Scheduled: 0.00ns (0.00%), Blocked: 1.01m (0.24%), Output: 192 rows (1.69kB)
       │   metrics:
       │     'CPU time distribution (s)' = {count=32, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.00, min=0.00, max=0.00}
       │     'Input rows distribution' = {count=32, p01=0.00, p05=0.00, p10=1.00, p25=1.00, p50=2.00, p75=5.00, p90=21.00, p95=26.00, p99=29.00, min=0.00, max=29.00}
       │     'Scheduled time distribution (s)' = {count=32, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.00, min=0.00, max=0.00}
       │   Input avg.: 6.00 rows, Input std.dev.: 136.93%
       └─ RemoteSource[sourceFragmentIds = [2]]
              Layout: [count_0:bigint]
              CPU: 3.00ms (0.05%), Scheduled: 8.00ms (0.00%), Blocked: 32.20m (7.67%), Output: 192 rows (1.69kB)
              metrics:
                'Blocked time distribution (s)' = {count=32, p01=60.37, p05=60.37, p10=60.37, p25=60.37, p50=60.37, p75=60.37, p90=60.37, p95=60.37, p99=60.37, min=60.37, max=60.37}
                'CPU time distribution (s)' = {count=32, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.00, min=0.00, max=0.00}
                'Input rows distribution' = {count=32, p01=0.00, p05=0.00, p10=1.00, p25=1.00, p50=2.00, p75=5.00, p90=21.00, p95=26.00, p99=29.00, min=0.00, max=29.00}
                'Scheduled time distribution (s)' = {count=32, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.00, min=0.00, max=0.00}
              Input avg.: 6.00 rows, Input std.dev.: 136.93%

Fragment 2 [HASH]
    CPU: 329.31ms, Scheduled: 500.00ms, Blocked 6.42h (Input: 3.22h, Output: 0.00ns), Input: 6026 rows (64.73kB); per task: avg.: 1004.33 std.dev.: 29.79, Output: 192 rows (1.69kB)
    Peak Memory: 20.23MB, Tasks count: 6; per task: max: 3.37MB
    Output buffer active time: 13.27ms, buffer utilization distribution (%): {p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.01, p90=0.01, p95=0.01, p99=0.01, min=0.00, max=0.01}
    Task output distribution: {count=6.00, p01=288B, p05=288B, p10=288B, p25=288B, p50=288B, p75=288B, p90=288B, p95=288B, p99=288B, max=288B}
    Task input distribution: {count=6.00, p01=10.37kB, p05=10.37kB, p10=10.37kB, p25=10.43kB, p50=10.91kB, p75=10.92kB, p90=11.31kB, p95=11.31kB, p99=11.31kB, max=11.31kB}
    Output layout: [count_0]
    Output partitioning: SINGLE []
    Aggregate[type = PARTIAL]
    │   Layout: [count_0:bigint]
    │   CPU: 45.00ms (0.76%), Scheduled: 57.00ms (0.01%), Blocked: 0.00ns (0.00%), Output: 192 rows (1.69kB)
    │   metrics:
    │     'Accumulator update CPU time' = {duration=5.79ms}
    │     'CPU time distribution (s)' = {count=192, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.00, min=0.00, max=0.00}
    │     'Group by hash update CPU time' = {duration=0.00ns}
    │     'Input rows distribution' = {count=192, p01=4.00, p05=8.00, p10=9.00, p25=12.00, p50=14.00, p75=16.00, p90=19.00, p95=20.00, p99=25.00, min=4.00, max=26.00}
    │     'Input rows processed without partial aggregation enabled' = LongCount{total=0}
    │     'Scheduled time distribution (s)' = {count=192, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.00, min=0.00, max=0.00}
    │   Input avg.: 14.04 rows, Input std.dev.: 26.35%
    │   count_0 := count(partition)
    └─ Aggregate[type = FINAL, keys = [partition]]
       │   Layout: [partition:row(timestamp_day date, redacted_key_bucket_8 integer)]
       │   Estimates: {rows: ? (?), cpu: ?, memory: ?, network: 0B}
       │   CPU: 146.00ms (2.47%), Scheduled: 252.00ms (0.03%), Blocked: 0.00ns (0.00%), Output: 2695 rows (28.95kB)
       │   metrics:
       │     'Accumulator update CPU time' = {duration=0.00ns}
       │     'CPU time distribution (s)' = {count=192, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.00, min=0.00, max=0.00}
       │     'Group by hash update CPU time' = {duration=183.83ms}
       │     'Input rows distribution' = {count=192, p01=11.00, p05=15.00, p10=20.00, p25=26.00, p50=31.00, p75=37.00, p90=43.80, p95=46.00, p99=55.00, min=7.00, max=56.00}
       │     'Input rows processed without partial aggregation enabled' = LongCount{total=0}
       │     'Scheduled time distribution (s)' = {count=192, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.01, min=0.00, max=0.01}
       │   Input avg.: 31.39 rows, Input std.dev.: 28.74%
       └─ LocalExchange[partitioning = HASH, arguments = [partition::row(timestamp_day date, redacted_key_bucket_8 integer)]]
          │   Layout: [partition:row(timestamp_day date, redacted_key_bucket_8 integer)]
          │   Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B}
          │   CPU: 63.00ms (1.06%), Scheduled: 77.00ms (0.01%), Blocked: 3.22h (46.04%), Output: 6026 rows (64.73kB)
          │   metrics:
          │     'CPU time distribution (s)' = {count=192, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.00, min=0.00, max=0.00}
          │     'Input rows distribution' = {count=192, p01=9.00, p05=12.00, p10=15.00, p25=17.79, p50=26.56, p75=37.71, p90=54.70, p95=74.00, p99=116.00, min=9.00, max=124.00}
          │     'Scheduled time distribution (s)' = {count=192, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.00, min=0.00, max=0.00}
          │   Input avg.: 31.39 rows, Input std.dev.: 62.34%
          └─ RemoteSource[sourceFragmentIds = [3]]
                 Layout: [partition:row(timestamp_day date, redacted_key_bucket_8 integer)]
                 CPU: 37.00ms (0.62%), Scheduled: 53.00ms (0.01%), Blocked: 3.22h (46.05%), Output: 6026 rows (64.73kB)
                 metrics:
                   'Blocked time distribution (s)' = {count=192, p01=60.36, p05=60.37, p10=60.37, p25=60.37, p50=60.37, p75=60.37, p90=60.37, p95=60.37, p99=60.37, min=60.36, max=60.37}
                   'CPU time distribution (s)' = {count=192, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.00, min=0.00, max=0.00}
                   'Input rows distribution' = {count=192, p01=9.00, p05=12.00, p10=15.00, p25=17.79, p50=26.56, p75=37.71, p90=54.70, p95=74.00, p99=116.00, min=9.00, max=124.00}
                   'Scheduled time distribution (s)' = {count=192, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.00, min=0.00, max=0.00}
                 Input avg.: 31.39 rows, Input std.dev.: 62.34%

Fragment 3 [SOURCE]
    CPU: 5.62s, Scheduled: 15.63m, Blocked 0.00ns (Input: 0.00ns, Output: 0.00ns), Input: 6201 rows (66.61kB); per task: avg.: 1033.50 std.dev.: 288.80, Output: 6026 rows (87.10kB)
    Peak Memory: 2.61MB, Tasks count: 6; per task: max: 455.27kB
    Output buffer active time: 5.01m, buffer utilization distribution (%): {p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.00, min=0.00, max=0.02}
    Task output distribution: {count=6.00, p01=8.41kB, p05=8.41kB, p10=8.41kB, p25=11.80kB, p50=14.36kB, p75=17.79kB, p90=21.50kB, p95=21.50kB, p99=21.50kB, max=21.50kB}
    Task input distribution: {count=6.00, p01=6.45kB, p05=6.45kB, p10=6.45kB, p25=8.99kB, p50=10.95kB, p75=13.44kB, p90=16.19kB, p95=16.19kB, p99=16.19kB, max=16.19kB}
    Output layout: [partition]
    Output partitioning: HASH [partition]
    Aggregate[type = PARTIAL, keys = [partition]]
    │   Layout: [partition:row(timestamp_day date, redacted_key_bucket_8 integer)]
    │   CPU: 102.00ms (1.72%), Scheduled: 146.00ms (0.02%), Blocked: 0.00ns (0.00%), Output: 6026 rows (87.10kB)
    │   metrics:
    │     'Accumulator update CPU time' = {duration=0.00ns}
    │     'CPU time distribution (s)' = {count=695, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.00, min=0.00, max=0.00}
    │     'Group by hash update CPU time' = {duration=14.81ms}
    │     'Input rows distribution' = {count=695, p01=0.95, p05=1.00, p10=1.00, p25=1.03, p50=2.00, p75=7.18, p90=16.16, p95=33.05, p99=111.00, min=0.00, max=400.00}
    │     'Input rows processed without partial aggregation enabled' = LongCount{total=0}
    │     'Scheduled time distribution (s)' = {count=695, p01=0.00, p05=0.00, p10=0.00, p25=0.00, p50=0.00, p75=0.00, p90=0.00, p95=0.00, p99=0.00, min=0.00, max=0.01}
    │   Input avg.: 8.92 rows, Input std.dev.: 290.71%
    └─ TableScan[table = redacted_conniceberg.redacted_tablefiles]
           Layout: [partition:row(timestamp_day date, redacted_key_bucket_8 integer)]
           Estimates: {rows: ? (?), cpu: ?, memory: 0B, network: 0B}
           CPU: 5.53s (93.30%), Scheduled: 15.63m (99.94%), Blocked: 0.00ns (0.00%), Output: 6201 rows (66.61kB)
           metrics:
             'CPU time distribution (s)' = {count=695, p01=0.00, p05=0.00, p10=0.00, p25=0.01, p50=0.01, p75=0.01, p90=0.01, p95=0.01, p99=0.02, min=0.00, max=0.02}
             'Input rows distribution' = {count=695, p01=0.95, p05=1.00, p10=1.00, p25=1.03, p50=2.00, p75=7.18, p90=16.16, p95=33.05, p99=111.00, min=0.00, max=400.00}
             'Scheduled time distribution (s)' = {count=695, p01=0.22, p05=0.25, p10=0.29, p25=0.34, p50=0.41, p75=0.48, p90=0.49, p95=0.52, p99=60.32, min=0.08, max=60.35}
           Input avg.: 8.92 rows, Input std.dev.: 290.71%
           partition := partition
           Input: 6201 rows (66.61kB), Physical input: 6.38MB, Physical input time: 125.92ms, Splits: 695, Splits generation wait time: 0.01ms

[dotjdk]

So, I did a count(*) from manifests, and Trino returned 691, which seemed odd to me as we continuously rewrite the recent files and the manifests from spark. I then did a manual rewrite_manifests from spark, which said 5 manifests deleted, 1 added. Puzzled to say the least

After this, Trino reported 3 manifests (there are frequent appends to the table) and both the GROUP BY and the count(distinct) queries are consistently fast

[dotjdk]

I will check again later this weekend or Monday when I am back at the office to see if a similar issue has come up again. I think you can close this issue for now, unless you feel there's still something to investigate regarding the performance on the $files optimization. I am not sure it is related to that anymore, even though the fast -> slow behavior after a cluster restart before still seems odd to me.

[ebyhr]

cc: @tbaeg @raunaqmorarka

[dotjdk]

Just an update after the weekend:

Manifest count in Trino is still aligned with Spark, and the metadata queries are lightning fast.