kinolaev commented on issue #11648: URL: https://github.com/apache/iceberg/issues/11648#issuecomment-5387856932
I ran a more realistic benchmark with a local S3 server and want to share the results. **TL;DR** - **Disabling the cache was the wrong fix.** The real bug is `spark.sql.iceberg.executor-cache.max-total-size` being too small (causing repeated eviction and full re-decode) plus unnecessary per-row rehashing - #15714 fixes the rehashing part (~1.8x faster: ~11.09-11.51s → ~6.04-6.50s per delete file). - **Size the cache correctly instead**: in the worked example below (128 data files, a too-small cache), processing time drops from ~25.6 minutes to ~12 seconds. - **Independent tip:** lowering `write.delete.parquet.row-group-size-bytes` (default 128mb) makes uncached selective reads up to 17x faster, and raises the cache break-even point from ~1 lookup at the 64mb default up to ~18 lookups at 2mb row groups (per the example). All benchmark code is published in [a separate branch](https://github.com/kinolaev/iceberg/tree/position-delete-benchmarks). The benchmarks were run on a MacBook Pro (M1 Pro CPU, 16gb RAM). The tested position delete file has 67108864 rows (64\*1024\*1024), and its size varies between 63630623 and 66101686 bytes. I picked that size because 64mb is the default value for [write.delete.target-file-size-bytes](https://iceberg.apache.org/docs/latest/configuration/#write-properties). The time it takes to read deleted positions for a specific data file depends on how many distinct data files the delete file references: | file_path count | read timing (seconds) | |---:|---:| | 4 | 7.706 | | 8 | 6.909 | | 16 | 6.325 | | 32 | 6.206 | | 64 | 6.181 | | 128 | 6.296 | | 256 | 5.792 | | 512 | 6.091 | The time for caching doesn't depend on how many data files are referenced; it's 11.09-11.51 seconds on the main branch and 6.04-6.50 seconds on the PR #15714 branch. So, the PR makes the caching process only 1.8 times faster, but that's enough to make it as fast as reading position deletes for just a single data file. In other words, disabling the cache for position deletes is a bad idea. From what I can tell, the issue has two root causes: 1. `spark.sql.iceberg.executor-cache.max-total-size` was too small 2. hashes were computed for every row in a position delete file Because the cache was too small, entries got evicted before they could be reused. For example, say you have 128 data files and two 65mb position delete files, each referencing 64 data files. If you're lucky and the first 64 data files you read all happen to be referenced by the same delete file, you'll only need ~12x2 seconds (or ~6x2 with the PR) to process the delete files. But if you're unlucky - the even-numbered data files are referenced by the first delete file and the odd-numbered ones by the second - you'll run into this issue: 1. the first delete file is loaded, ~67 million hashes are computed, and the result `CharSequenceMap<PositionDeleteIndex>` is cached 2. the result of the previous step is evicted, the second delete file is loaded, ~67 million more hashes are computed, and the new result is cached Then the same steps repeat 64 times. If you have enough RAM for 128 `CharSequenceMap<PositionDeleteIndex>` instances (each 8mb to 3.4gb depending on how sparse the deletes are), you'll probably need only ~12x128=25.6 minutes to process the delete files (or ~6x128=12.8 minutes with the PR). But if you have to gc each map before creating the next one, you get to the point the issue author described as "slow means it almost never ends." With the PR applied and the cache size set correctly (130mb in this example), processing the delete files would take only ~12 seconds - significantly less than the ~6*128=768 seconds it takes without the cache. Conclusion: hardcoding `spark.sql.iceberg.executor-cache.max-total-size=false` slowed down compaction both for tables with equality delete files (the data files x delete files request count problem) and for tables with position delete files. The root cause is simply a misconfigured cache size. @davseitsev, I'd really appreciate it if you could measure `select sum()` (or a similar query that has to apply all the position deletes) on your tables before compaction, with `spark.sql.iceberg.executor-cache.max-total-size=true`, across different cache sizes. @nastra, @aokolnychyi, @anuragmantri, @singhpk234, please take a look at this issue and PR #15714 - this is critical for anyone compacting tables with equality deletes using Spark. --- This part is about selective-read performance. It's unrelated to the cache problem, but might be interesting if you want to disable the cache for position delete files. The easy way to improve the performance of reading position delete files is to lower `write.delete.parquet.row-group-size-bytes`: | file_path count \ row group size | 1 | 2 | 4 | 8 | 16 | 32 | 64 | |---:|---:|---:|---:|---:|---:|---:|---:| | 4 | 6.490 | 3.514 | 3.507 | 3.896 | 4.366 | 5.458 | 7.706 | | 8 | 3.274 | 1.808 | 1.949 | 2.284 | 3.577 | 4.539 | 6.909 | | 16 | 1.664 | 0.986 | 1.134 | 1.637 | 2.201 | 3.586 | 6.325 | | 32 | 0.809 | 0.563 | 0.750 | 1.435 | 1.817 | 3.364 | 6.206 | | 64 | 0.389 | 0.361 | 0.545 | 0.935 | 1.696 | 3.202 | 6.181 | | 128 | 0.221 | 0.256 | 0.464 | 0.959 | 1.756 | 3.533 | 6.296 | The default value is 128mb, but since `write.delete.target-file-size-bytes=64mb`, in practice all position deletes end up in a single 64mb row group. Just lowering the size to 2mb in the example would make the uncached read 6.181/0.361=17 times faster. Row group size only affects caching performance at the 1mb setting. The table below shows cache vs. selective-read timing at different file_path counts and row group sizes: | file_path count \ row group size | 1 | 2 | 4 | 8 | 16 | 32 | 64 | |---:|---:|---:|---:|---:|---:|---:|---:| | 4 | 3.31x | 1.85x | 1.78x | 1.60x | 1.43x | 1.14x | 0.81x | | 8 | 6.56x | 3.59x | 3.20x | 2.73x | 1.74x | 1.37x | 0.90x | | 16 | 12.92x | 6.58x | 5.49x | 3.81x | 2.83x | 1.74x | 0.99x | | 32 | 26.57x | 11.52x | 8.31x | 4.34x | 3.43x | 1.85x | 1.00x | | 64 | 55.25x | 17.97x | 11.43x | 6.66x | 3.67x | 1.95x | 1.01x | | 128 | 97.26x | 25.34x | 13.43x | 6.50x | 3.55x | 1.76x | 0.99x | So, with a 2mb row group size and 64 referenced files (as in the example), caching pays off only past ~18 lookups. --- This part is for the especially curious - for those who think anything over a second is slow for 64 megarows. If we add page pruning, filter before materialization, skip decoding file_path pages where min=max, and parallelize row group processing, caching becomes 2.3-7.5 times faster than on the PR branch. Selective read becomes 1.4-76.8 times faster, never crossing the 1-second mark. I still think there's room for improvement, though :) <details> <summary>Full benchmark results</summary> | row group size | file_path count | read | read speedup | cache | cache speedup | cache vs read | |---:|---:|---:|---:|---:|---:|---:| | 4 | 4 | 0.528 | 6.64x | 0.836 | 7.45x | 1.58x | | 4 | 8 | 0.464 | 4.20x | 0.836 | 7.45x | 1.80x | | 4 | 16 | 0.441 | 2.57x | 0.836 | 7.45x | 1.90x | | 4 | 32 | 0.351 | 2.14x | 0.836 | 7.45x | 2.38x | | 4 | 64 | 0.343 | 1.59x | 0.836 | 7.45x | 2.44x | | 4 | 128 | 0.335 | 1.39x | 0.836 | 7.45x | 2.49x | | 8 | 4 | 0.601 | 6.48x | 0.862 | 7.23x | 1.43x | | 8 | 8 | 0.454 | 5.03x | 0.862 | 7.23x | 1.90x | | 8 | 16 | 0.294 | 5.57x | 0.862 | 7.23x | 2.93x | | 8 | 32 | 0.218 | 6.58x | 0.862 | 7.23x | 3.95x | | 8 | 64 | 0.184 | 5.08x | 0.862 | 7.23x | 4.68x | | 8 | 128 | 0.168 | 5.71x | 0.862 | 7.23x | 5.13x | | 16 | 4 | 0.764 | 5.71x | 1.033 | 6.03x | 1.35x | | 16 | 8 | 0.430 | 8.32x | 1.033 | 6.03x | 2.40x | | 16 | 16 | 0.247 | 8.91x | 1.033 | 6.03x | 4.18x | | 16 | 32 | 0.159 | 11.43x | 1.033 | 6.03x | 6.49x | | 16 | 64 | 0.119 | 14.25x | 1.033 | 6.03x | 8.68x | | 16 | 128 | 0.096 | 18.29x | 1.033 | 6.03x | 10.76x | | 32 | 4 | 0.724 | 7.54x | 1.610 | 3.87x | 2.22x | | 32 | 8 | 0.400 | 11.35x | 1.610 | 3.87x | 4.03x | | 32 | 16 | 0.216 | 16.60x | 1.610 | 3.87x | 7.45x | | 32 | 32 | 0.137 | 24.55x | 1.610 | 3.87x | 11.75x | | 32 | 64 | 0.092 | 34.80x | 1.610 | 3.87x | 17.50x | | 32 | 128 | 0.075 | 47.11x | 1.610 | 3.87x | 21.47x | | 64 | 4 | 0.739 | 10.43x | 2.691 | 2.32x | 3.64x | | 64 | 8 | 0.420 | 16.45x | 2.691 | 2.32x | 6.41x | | 64 | 16 | 0.230 | 27.50x | 2.691 | 2.32x | 11.70x | | 64 | 32 | 0.143 | 43.40x | 2.691 | 2.32x | 18.82x | | 64 | 64 | 0.098 | 63.07x | 2.691 | 2.32x | 27.46x | | 64 | 128 | 0.082 | 76.78x | 2.691 | 2.32x | 32.82x | </details> -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
