Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
adriangb merged PR #16433: URL: https://github.com/apache/datafusion/pull/16433 -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
adriangb commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-3215218818 Benchmarks look good, several faster queries, no queries really slower! I'll merge this in the next couple hours if you don't first Andrew. -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
alamb commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-3215208209 🤖: Benchmark completed Details ``` Comparing HEAD and topk-filters Benchmark clickbench_extended.json ┏━━┳━┳━━┳━━━┓ ┃ Query┃HEAD ┃ topk-filters ┃Change ┃ ┡━━╇━╇━━╇━━━┩ │ QQuery 0 │ 2639.26 ms │ 2604.08 ms │ no change │ │ QQuery 1 │ 1219.97 ms │ 1240.23 ms │ no change │ │ QQuery 2 │ 2324.31 ms │ 2392.51 ms │ no change │ │ QQuery 3 │ 1207.99 ms │ 1172.77 ms │ no change │ │ QQuery 4 │ 2236.03 ms │ 2231.24 ms │ no change │ │ QQuery 5 │ 27457.81 ms │ 27345.17 ms │ no change │ │ QQuery 6 │ 4125.85 ms │ 4209.18 ms │ no change │ │ QQuery 7 │ 3599.32 ms │ 3503.08 ms │ no change │ └──┴─┴──┴───┘ ┏━┳┓ ┃ Benchmark Summary ┃┃ ┡━╇┩ │ Total Time (HEAD) │ 44810.53ms │ │ Total Time (topk-filters) │ 44698.26ms │ │ Average Time (HEAD) │ 5601.32ms │ │ Average Time (topk-filters) │ 5587.28ms │ │ Queries Faster │ 0 │ │ Queries Slower │ 0 │ │ Queries with No Change │ 8 │ │ Queries with Failure│ 0 │ └─┴┘ Benchmark clickbench_partitioned.json ┏━━┳━┳━━┳━━━┓ ┃ Query┃HEAD ┃ topk-filters ┃Change ┃ ┡━━╇━╇━━╇━━━┩ │ QQuery 0 │ 2.12 ms │ 2.15 ms │ no change │ │ QQuery 1 │50.61 ms │ 51.14 ms │ no change │ │ QQuery 2 │ 135.87 ms │137.82 ms │ no change │ │ QQuery 3 │ 162.54 ms │162.41 ms │ no change │ │ QQuery 4 │ 1055.31 ms │ 1027.29 ms │ no change │ │ QQuery 5 │ 1475.94 ms │ 1472.47 ms │ no change │ │ QQuery 6 │ 2.13 ms │ 2.17 ms │ no change │ │ QQuery 7 │54.79 ms │ 54.63 ms │ no change │ │ QQuery 8 │ 1472.25 ms │ 1422.87 ms │ no change │ │ QQuery 9 │ 1837.78 ms │ 1812.93 ms │ no change │ │ QQuery 10│ 369.82 ms │385.13 ms │ no change │ │ QQuery 11│ 428.04 ms │435.19 ms │ no change │ │ QQuery 12│ 1347.17 ms │ 1365.75 ms │ no change │ │ QQuery 13│ 2098.07 ms │ 2115.15 ms │ no change │ │ QQuery 14│ 1237.31 ms │ 1239.03 ms │ no change │ │ QQuery 15│ 1173.37 ms │ 1173.14 ms │ no change │ │ QQuery 16│ 2612.56 ms │ 2607.94 ms │ no change │ │ QQuery 17│ 2597.59 ms │ 2607.24 ms │ no change │ │ QQuery 18│ 5156.33 ms │ 4834.22 ms │ +1.07x faster │ │ QQuery 19│ 126.62 ms │128.06 ms │ no change │ │ QQuery 20│ 1961.14 ms │ 2007.25 ms │ no change │ │ QQuery 21│ 2279.44 ms │ 2322.19 ms │ no change │ │ QQuery 22│ 3932.07 ms │ 3939.04 ms │ no change │ │ QQuery 23│ 15969.33 ms │ 12635.10 ms │ +1.26x faster │ │ QQuery 24│ 265.46 ms │208.07 ms │ +1.28x faster │ │ QQuery 25│ 492.34 ms │505.17 ms │ no change │ │ QQuery 26│ 259.62 ms │229.17 ms │ +1.13x faster │ │ QQuery 27│ 2872.01 ms │ 2888.04 ms │ no change │ │ QQuery 28│ 23296.75 ms │ 23107.00 ms │ no change │ │ QQuery 29│ 984.75 ms │ 1005.99 ms │ no change │ │ QQuery 30│ 1328.94 ms │ 1289.57 ms │ no change │ │ QQuery 31│ 1326.20 ms │ 1330.12 ms │ no change │ │ QQuery 32│ 4487.03 ms │ 4331.85 ms │ no change │ │ QQuery 33│ 5750.40 ms │ 5628.25 ms │ no change │ │ QQuery 34│ 5811.68 ms │ 5931.34 ms │ no change │ │ QQuery 35│ 2014.76 ms │ 2008.45 ms │ no change │ │ QQuery 36│ 122.81 ms │120.62 ms │ no change │ │ QQuery 37│53.34 ms │ 53.77 ms │ no change │ │ QQuery 38│ 122.13 ms │123.16 ms │ no change │ │ QQuery 39│ 194.75 ms │197.72 ms │ no change │ │ QQuery 40│43.97 ms │ 43.22 ms │ no change │ │ QQuery 41│40.40 ms │ 39.76 ms │ no change │ │ QQuery 42│32.38 ms │ 31.56 ms │ no change │ └──┴─┴──┴───┘ ┏━┳┓ ┃ Benchmark Summary ┃┃ ┡━╇┩ │ Total Time (HEAD) │ 97037.93ms │ │ Total Time (topk-filters) │ 93013.16ms │ │ Average Tim
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
alamb commented on code in PR #16433:
URL: https://github.com/apache/datafusion/pull/16433#discussion_r2294333805
##
datafusion/datasource-parquet/src/opener.rs:
##
@@ -409,12 +409,58 @@ impl FileOpener for ParquetOpener {
.with_row_groups(row_group_indexes)
.build()?;
-let adapted = stream
-.map_err(|e| ArrowError::ExternalError(Box::new(e)))
-.map(move |maybe_batch| {
-maybe_batch
-.and_then(|b|
schema_mapping.map_batch(b).map_err(Into::into))
-});
+// Create a stateful stream that can check pruning after each batch
+let adapted = {
Review Comment:
- https://github.com/apache/datafusion/pull/17293
--
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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
alamb commented on code in PR #16433:
URL: https://github.com/apache/datafusion/pull/16433#discussion_r2294236933
##
datafusion/physical-plan/src/topk/mod.rs:
##
@@ -319,19 +343,89 @@ impl TopK {
/// (a > 2 OR (a = 2 AND b < 3))
/// ```
fn update_filter(&mut self) -> Result<()> {
-let Some(filter) = &self.filter else {
+// If the heap doesn't have k elements yet, we can't create thresholds
+let Some(max_row) = self.heap.max() else {
return Ok(());
};
-let Some(thresholds) = self.heap.get_threshold_values(&self.expr)?
else {
+
+let new_threshold_row = &max_row.row;
+
+// Extract scalar values BEFORE acquiring lock to reduce critical
section
+let thresholds = match self.heap.get_threshold_values(&self.expr)? {
Review Comment:
I think we can move this down after the check for update too:
- https://github.com/pydantic/datafusion/pull/37
##
datafusion/datasource-parquet/src/opener.rs:
##
@@ -409,12 +409,58 @@ impl FileOpener for ParquetOpener {
.with_row_groups(row_group_indexes)
.build()?;
-let adapted = stream
-.map_err(|e| ArrowError::ExternalError(Box::new(e)))
-.map(move |maybe_batch| {
-maybe_batch
-.and_then(|b|
schema_mapping.map_batch(b).map_err(Into::into))
-});
+// Create a stateful stream that can check pruning after each batch
+let adapted = {
Review Comment:
I found this code somewhat 🤯 (and this function is already 100s of lines
long) I spent some time refactoring it into its own stream for readability and
I also understand it better now. I'll put up a follow on PR to extract this
logic -- no need to do it in this one
--
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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
alamb commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-3215116380 🤖: Benchmark completed Details ``` Comparing HEAD and topk-filters Benchmark run_topk_tpch.json ┏━━┳━━━┳━━┳━━━┓ ┃ Query┃ HEAD ┃ topk-filters ┃Change ┃ ┡━━╇━━━╇━━╇━━━┩ │ Q1 │ 36.60 ms │ 8.76 ms │ +4.18x faster │ │ Q2 │ 45.45 ms │ 45.55 ms │ no change │ │ Q3 │ 124.01 ms │125.24 ms │ no change │ │ Q4 │ 51.42 ms │ 48.09 ms │ +1.07x faster │ │ Q5 │ 35.31 ms │ 38.12 ms │ 1.08x slower │ │ Q6 │ 66.22 ms │ 67.99 ms │ no change │ │ Q7 │ 162.63 ms │166.41 ms │ no change │ │ Q8 │ 106.50 ms │108.96 ms │ no change │ │ Q9 │ 135.16 ms │135.68 ms │ no change │ │ Q10 │ 223.96 ms │221.17 ms │ no change │ │ Q11 │ 113.44 ms │117.74 ms │ no change │ └──┴───┴──┴───┘ ┏━┳━━━┓ ┃ Benchmark Summary ┃ ┃ ┡━╇━━━┩ │ Total Time (HEAD) │ 1100.71ms │ │ Total Time (topk-filters) │ 1083.72ms │ │ Average Time (HEAD) │ 100.06ms │ │ Average Time (topk-filters) │ 98.52ms │ │ Queries Faster │ 2 │ │ Queries Slower │ 1 │ │ Queries with No Change │ 8 │ │ Queries with Failure│ 0 │ └─┴───┘ ``` -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
alamb commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-3215116778 🤖 `./gh_compare_branch.sh` [Benchmark Script](https://github.com/alamb/datafusion-benchmarking/blob/main/gh_compare_branch.sh) Running Linux aal-dev 6.14.0-1014-gcp #15~24.04.1-Ubuntu SMP Fri Jul 25 23:26:08 UTC 2025 x86_64 x86_64 x86_64 GNU/Linux Comparing topk-filters (d4e2312d2c3b28db15d194bb279577f495f84029) to f363e382661a4f45dad2912e9988f1703e46939b [diff](https://github.com/apache/datafusion/compare/f363e382661a4f45dad2912e9988f1703e46939b..d4e2312d2c3b28db15d194bb279577f495f84029) using: tpch_mem clickbench_partitioned clickbench_extended Results will be posted here when complete -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
alamb commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-3215086665 🤖 `./gh_compare_branch.sh` [Benchmark Script](https://github.com/alamb/datafusion-benchmarking/blob/main/gh_compare_branch.sh) Running Linux aal-dev 6.14.0-1014-gcp #15~24.04.1-Ubuntu SMP Fri Jul 25 23:26:08 UTC 2025 x86_64 x86_64 x86_64 GNU/Linux Comparing topk-filters (d4e2312d2c3b28db15d194bb279577f495f84029) to f363e382661a4f45dad2912e9988f1703e46939b [diff](https://github.com/apache/datafusion/compare/f363e382661a4f45dad2912e9988f1703e46939b..d4e2312d2c3b28db15d194bb279577f495f84029) using: topk_tpch Results will be posted here when complete -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
alamb commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-3215086584 🤖: Benchmark completed Details ``` Comparing HEAD and topk-filters Benchmark tpch_mem_sf1.json ┏━━┳━━━┳━━┳━━━┓ ┃ Query┃ HEAD ┃ topk-filters ┃Change ┃ ┡━━╇━━━╇━━╇━━━┩ │ QQuery 1 │ 169.63 ms │165.07 ms │ no change │ │ QQuery 2 │ 27.55 ms │ 27.70 ms │ no change │ │ QQuery 3 │ 44.86 ms │ 45.11 ms │ no change │ │ QQuery 4 │ 26.77 ms │ 26.41 ms │ no change │ │ QQuery 5 │ 72.18 ms │ 73.65 ms │ no change │ │ QQuery 6 │ 19.67 ms │ 19.50 ms │ no change │ │ QQuery 7 │ 144.25 ms │139.83 ms │ no change │ │ QQuery 8 │ 30.48 ms │ 33.56 ms │ 1.10x slower │ │ QQuery 9 │ 83.26 ms │ 82.54 ms │ no change │ │ QQuery 10│ 57.63 ms │ 58.02 ms │ no change │ │ QQuery 11│ 42.63 ms │ 40.83 ms │ no change │ │ QQuery 12│ 49.97 ms │ 50.12 ms │ no change │ │ QQuery 13│ 44.87 ms │ 46.68 ms │ no change │ │ QQuery 14│ 13.69 ms │ 13.90 ms │ no change │ │ QQuery 15│ 23.59 ms │ 23.69 ms │ no change │ │ QQuery 16│ 23.76 ms │ 23.45 ms │ no change │ │ QQuery 17│ 141.23 ms │146.02 ms │ no change │ │ QQuery 18│ 318.80 ms │313.81 ms │ no change │ │ QQuery 19│ 48.11 ms │ 34.80 ms │ +1.38x faster │ │ QQuery 20│ 60.59 ms │ 47.85 ms │ +1.27x faster │ │ QQuery 21│ 224.40 ms │216.34 ms │ no change │ │ QQuery 22│ 20.00 ms │ 19.36 ms │ no change │ └──┴───┴──┴───┘ ┏━┳━━━┓ ┃ Benchmark Summary ┃ ┃ ┡━╇━━━┩ │ Total Time (HEAD) │ 1687.90ms │ │ Total Time (topk-filters) │ 1648.24ms │ │ Average Time (HEAD) │ 76.72ms │ │ Average Time (topk-filters) │ 74.92ms │ │ Queries Faster │ 2 │ │ Queries Slower │ 1 │ │ Queries with No Change │19 │ │ Queries with Failure│ 0 │ └─┴───┘ ``` -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
alamb commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-3215012818 Looking at it now -- I kicked off the benchmarks again after making some changes to my gcp machine that hopefully will make the results more consistent -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
alamb commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-3215011892 🤖 `./gh_compare_branch.sh` [Benchmark Script](https://github.com/alamb/datafusion-benchmarking/blob/main/gh_compare_branch.sh) Running Linux aal-dev 6.14.0-1014-gcp #15~24.04.1-Ubuntu SMP Fri Jul 25 23:26:08 UTC 2025 x86_64 x86_64 x86_64 GNU/Linux Comparing topk-filters (0b1cb583b781e99e0163c6caa62b8f735bafdcf8) to f363e382661a4f45dad2912e9988f1703e46939b [diff](https://github.com/apache/datafusion/compare/f363e382661a4f45dad2912e9988f1703e46939b..0b1cb583b781e99e0163c6caa62b8f735bafdcf8) using: tpch_mem Results will be posted here when complete -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
adriangb commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-3214807633 @alamb I think you've reviewed but not approved, I thought you had approved. Can you take another look at this PR when you get a chance? I think I've addressed all of the feedback. -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
adriangb commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-3210820667 I've rebased this on `main` and plan to merge it once CI passes -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
adriangb commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-3203042277 @alamb I think most of the feedback has been addressed, thank you for your review and improvements -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
alamb commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-3202553698 🤖: Benchmark completed Details ``` Comparing HEAD and topk-filters Benchmark clickbench_extended.json ┏━━┳━┳━━┳━━━┓ ┃ Query┃HEAD ┃ topk-filters ┃Change ┃ ┡━━╇━╇━━╇━━━┩ │ QQuery 0 │ 1941.87 ms │ 1911.66 ms │ no change │ │ QQuery 1 │ 745.79 ms │645.87 ms │ +1.15x faster │ │ QQuery 2 │ 1448.99 ms │ 1286.31 ms │ +1.13x faster │ │ QQuery 3 │ 620.31 ms │614.76 ms │ no change │ │ QQuery 4 │ 1293.11 ms │ 1282.47 ms │ no change │ │ QQuery 5 │ 13859.62 ms │ 13799.54 ms │ no change │ │ QQuery 6 │ 2359.44 ms │ 2349.22 ms │ no change │ │ QQuery 7 │ 1800.51 ms │ 1811.91 ms │ no change │ └──┴─┴──┴───┘ ┏━┳┓ ┃ Benchmark Summary ┃┃ ┡━╇┩ │ Total Time (HEAD) │ 24069.65ms │ │ Total Time (topk-filters) │ 23701.73ms │ │ Average Time (HEAD) │ 3008.71ms │ │ Average Time (topk-filters) │ 2962.72ms │ │ Queries Faster │ 2 │ │ Queries Slower │ 0 │ │ Queries with No Change │ 6 │ │ Queries with Failure│ 0 │ └─┴┘ Benchmark clickbench_partitioned.json ┏━━┳━┳━━┳━━━┓ ┃ Query┃HEAD ┃ topk-filters ┃Change ┃ ┡━━╇━╇━━╇━━━┩ │ QQuery 0 │ 2.47 ms │ 2.23 ms │ +1.11x faster │ │ QQuery 1 │29.14 ms │ 29.27 ms │ no change │ │ QQuery 2 │75.23 ms │ 76.03 ms │ no change │ │ QQuery 3 │95.06 ms │ 89.60 ms │ +1.06x faster │ │ QQuery 4 │ 595.70 ms │594.73 ms │ no change │ │ QQuery 5 │ 862.32 ms │877.24 ms │ no change │ │ QQuery 6 │ 2.31 ms │ 2.18 ms │ +1.06x faster │ │ QQuery 7 │32.52 ms │ 33.61 ms │ no change │ │ QQuery 8 │ 848.36 ms │863.85 ms │ no change │ │ QQuery 9 │ 1168.82 ms │ 1152.74 ms │ no change │ │ QQuery 10│ 229.56 ms │225.21 ms │ no change │ │ QQuery 11│ 254.54 ms │249.47 ms │ no change │ │ QQuery 12│ 882.94 ms │836.94 ms │ +1.05x faster │ │ QQuery 13│ 1234.12 ms │ 1189.38 ms │ no change │ │ QQuery 14│ 805.61 ms │789.22 ms │ no change │ │ QQuery 15│ 783.32 ms │783.07 ms │ no change │ │ QQuery 16│ 1593.42 ms │ 1584.35 ms │ no change │ │ QQuery 17│ 1586.06 ms │ 1578.98 ms │ no change │ │ QQuery 18│ 2818.83 ms │ 2850.55 ms │ no change │ │ QQuery 19│82.25 ms │ 79.62 ms │ no change │ │ QQuery 20│ 1204.54 ms │ 1212.60 ms │ no change │ │ QQuery 21│ 1379.57 ms │ 1350.79 ms │ no change │ │ QQuery 22│ 2278.24 ms │ 2259.20 ms │ no change │ │ QQuery 23│ 7743.86 ms │ 6870.55 ms │ +1.13x faster │ │ QQuery 24│ 416.53 ms │405.25 ms │ no change │ │ QQuery 25│ 294.09 ms │288.28 ms │ no change │ │ QQuery 26│ 416.90 ms │405.81 ms │ no change │ │ QQuery 27│ 1617.30 ms │ 1592.63 ms │ no change │ │ QQuery 28│ 11970.14 ms │ 11759.08 ms │ no change │ │ QQuery 29│ 508.33 ms │520.47 ms │ no change │ │ QQuery 30│ 787.97 ms │783.59 ms │ no change │ │ QQuery 31│ 787.57 ms │773.39 ms │ no change │ │ QQuery 32│ 2368.78 ms │ 2381.14 ms │ no change │ │ QQuery 33│ 3177.45 ms │ 3187.93 ms │ no change │ │ QQuery 34│ 3188.98 ms │ 3216.57 ms │ no change │ │ QQuery 35│ 1215.92 ms │ 1294.23 ms │ 1.06x slower │ │ QQuery 36│ 129.68 ms │128.39 ms │ no change │ │ QQuery 37│56.08 ms │ 56.33 ms │ no change │ │ QQuery 38│ 124.60 ms │127.86 ms │ no change │ │ QQuery 39│ 198.97 ms │209.04 ms │ 1.05x slower │ │ QQuery 40│41.21 ms │ 42.17 ms │ no change │ │ QQuery 41│38.62 ms │ 39.69 ms │ no change │ │ QQuery 42│31.94 ms │ 33.52 ms │ no change │ └──┴─┴──┴───┘ ┏━┳┓ ┃ Benchmark Summary ┃┃ ┡━╇┩ │ Total Time (HEAD) │ 53959.86ms │ │ Total Time
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
adriangb commented on code in PR #16433:
URL: https://github.com/apache/datafusion/pull/16433#discussion_r2286514826
##
datafusion/physical-plan/src/topk/mod.rs:
##
@@ -121,13 +122,37 @@ pub struct TopK {
/// Common sort prefix between the input and the sort expressions to allow
early exit optimization
common_sort_prefix: Arc<[PhysicalSortExpr]>,
/// Filter matching the state of the `TopK` heap used for dynamic filter
pushdown
-filter: Option>,
+filter: TopKDynamicFilters,
/// If true, indicates that all rows of subsequent batches are guaranteed
/// to be greater (by byte order, after row conversion) than the top K,
/// which means the top K won't change and the computation can be finished
early.
pub(crate) finished: bool,
}
+#[derive(Debug, Clone)]
+pub struct TopKDynamicFilters {
+/// The current *global* threshold for the dynamic filter.
+/// This is shared across all partitions and is updated by any of them.
+/// Stored as row bytes for efficient comparison.
+threshold_row: Arc>>,
+/// The expression used to evaluate the dynamic filter
+expr: Arc,
Review Comment:
Yep good call! I think otherwise we're bound to run into bugs /
synchronization issues. Did it in bcc0dcdc4.
--
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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
adriangb commented on code in PR #16433:
URL: https://github.com/apache/datafusion/pull/16433#discussion_r2286494820
##
datafusion/physical-plan/src/topk/mod.rs:
##
@@ -319,19 +342,84 @@ impl TopK {
/// (a > 2 OR (a = 2 AND b < 3))
/// ```
fn update_filter(&mut self) -> Result<()> {
-let Some(filter) = &self.filter else {
+// If the heap doesn't have k elements yet, we can't create thresholds
+let Some(max_row) = self.heap.max() else {
return Ok(());
};
-let Some(thresholds) = self.heap.get_threshold_values(&self.expr)?
else {
-return Ok(());
+
+let new_threshold_row = &max_row.row;
+
+// Extract scalar values BEFORE acquiring lock to reduce critical
section
+let thresholds = match self.heap.get_threshold_values(&self.expr)? {
+Some(t) => t,
+None => return Ok(()),
};
+// Extract filter expression reference before entering critical section
+let filter_expr = Arc::clone(&self.filter.expr);
+
+// Check if we need to update the threshold (lock-free read)
+let current_threshold = self.filter.threshold_row.load();
+let needs_update = match current_threshold.as_ref() {
+Some(current_row) => {
+// new < current means new threshold is more selective
+current_row.as_slice().cmp(new_threshold_row) ==
Ordering::Greater
+}
+None => true, // No current threshold, so we need to set one
+};
+
+// Only proceed if we need to update
+if needs_update {
+// Build the filter expression OUTSIDE any synchronization
+let predicate = Self::build_filter_expression(&self.expr,
thresholds)?;
+let new_threshold_arc = Arc::new(new_threshold_row.to_vec());
+
+// Atomically update the threshold using compare-and-swap
+let old_threshold = self.filter.threshold_row.compare_and_swap(
+¤t_threshold,
+Some(Arc::clone(&new_threshold_arc)),
+);
+
+// Only update filter if we successfully updated the threshold
+// (or if there was no previous threshold and we're the first)
+let should_update_filter =
+match (old_threshold.as_ref(), current_threshold.as_ref()) {
+// We successfully swapped
+(Some(old), Some(expected)) if Arc::ptr_eq(old, expected)
=> true,
+// We were the first to set it
+(None, None) => true,
+// Another thread updated before us, check if our
threshold is still better
+(Some(actual_old), _) => {
+actual_old.as_slice().cmp(new_threshold_row) ==
Ordering::Greater
Review Comment:
I think you resolved this in your commits, just confirming
--
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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
adriangb commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-3202471933 > 🤖: Benchmark completed > > Details Hmm quite a bit worse than https://github.com/apache/datafusion/pull/16433#issuecomment-3201715769. I still think we should move forward with this sans ArcSwap and then we can look at that in isolation, otherwise it's too hard to know what's responsible for performance. -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
alamb commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-3202450803 🤖 `./gh_compare_branch.sh` [Benchmark Script](https://github.com/alamb/datafusion-benchmarking/blob/main/gh_compare_branch.sh) Running Linux aal-dev 6.11.0-1016-gcp #16~24.04.1-Ubuntu SMP Wed May 28 02:40:52 UTC 2025 x86_64 x86_64 x86_64 GNU/Linux Comparing topk-filters (ffe68565b55ba640f9c5197a4f38106ab0460807) to 8f15991f33bf6aca9d4da8958141b59d196b2ed6 [diff](https://github.com/apache/datafusion/compare/8f15991f33bf6aca9d4da8958141b59d196b2ed6..ffe68565b55ba640f9c5197a4f38106ab0460807) using: tpch_mem clickbench_partitioned clickbench_extended Results will be posted here when complete -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
alamb commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-3202450580 🤖: Benchmark completed Details ``` Comparing HEAD and topk-filters Benchmark run_topk_tpch.json ┏━━┳━━━┳━━┳━━━┓ ┃ Query┃ HEAD ┃ topk-filters ┃Change ┃ ┡━━╇━━━╇━━╇━━━┩ │ Q1 │ 24.01 ms │ 8.64 ms │ +2.78x faster │ │ Q2 │ 30.33 ms │ 30.83 ms │ no change │ │ Q3 │ 88.89 ms │ 98.89 ms │ 1.11x slower │ │ Q4 │ 34.17 ms │ 39.35 ms │ 1.15x slower │ │ Q5 │ 23.42 ms │ 27.51 ms │ 1.17x slower │ │ Q6 │ 48.61 ms │ 48.06 ms │ no change │ │ Q7 │ 121.79 ms │121.96 ms │ no change │ │ Q8 │ 90.35 ms │ 69.09 ms │ +1.31x faster │ │ Q9 │ 93.10 ms │ 91.78 ms │ no change │ │ Q10 │ 163.82 ms │170.23 ms │ no change │ │ Q11 │ 78.22 ms │ 13.29 ms │ +5.89x faster │ └──┴───┴──┴───┘ ┏━┳━━┓ ┃ Benchmark Summary ┃ ┃ ┡━╇━━┩ │ Total Time (HEAD) │ 796.69ms │ │ Total Time (topk-filters) │ 719.62ms │ │ Average Time (HEAD) │ 72.43ms │ │ Average Time (topk-filters) │ 65.42ms │ │ Queries Faster │3 │ │ Queries Slower │3 │ │ Queries with No Change │5 │ │ Queries with Failure│0 │ └─┴──┘ ``` -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
alamb commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-3202449463 🤖 `./gh_compare_branch.sh` [Benchmark Script](https://github.com/alamb/datafusion-benchmarking/blob/main/gh_compare_branch.sh) Running Linux aal-dev 6.11.0-1016-gcp #16~24.04.1-Ubuntu SMP Wed May 28 02:40:52 UTC 2025 x86_64 x86_64 x86_64 GNU/Linux Comparing topk-filters (ffe68565b55ba640f9c5197a4f38106ab0460807) to 8f15991f33bf6aca9d4da8958141b59d196b2ed6 [diff](https://github.com/apache/datafusion/compare/8f15991f33bf6aca9d4da8958141b59d196b2ed6..ffe68565b55ba640f9c5197a4f38106ab0460807) using: topk_tpch Results will be posted here when complete -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
alamb commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-3202449210 🤖: Benchmark completed Details ``` Comparing HEAD and topk-filters Benchmark tpch_mem_sf1.json ┏━━┳━━━┳━━┳━━━┓ ┃ Query┃ HEAD ┃ topk-filters ┃Change ┃ ┡━━╇━━━╇━━╇━━━┩ │ QQuery 1 │ 98.51 ms │ 98.91 ms │ no change │ │ QQuery 2 │ 20.91 ms │ 20.89 ms │ no change │ │ QQuery 3 │ 31.83 ms │ 32.16 ms │ no change │ │ QQuery 4 │ 18.38 ms │ 18.19 ms │ no change │ │ QQuery 5 │ 49.14 ms │ 49.23 ms │ no change │ │ QQuery 6 │ 11.96 ms │ 11.79 ms │ no change │ │ QQuery 7 │ 88.37 ms │ 85.17 ms │ no change │ │ QQuery 8 │ 24.71 ms │ 23.62 ms │ no change │ │ QQuery 9 │ 52.90 ms │ 53.36 ms │ no change │ │ QQuery 10│ 40.20 ms │ 39.90 ms │ no change │ │ QQuery 11│ 39.34 ms │ 38.67 ms │ no change │ │ QQuery 12│ 29.89 ms │ 30.23 ms │ no change │ │ QQuery 13│ 25.64 ms │ 26.43 ms │ no change │ │ QQuery 14│ 9.81 ms │ 9.54 ms │ no change │ │ QQuery 15│ 19.13 ms │ 18.48 ms │ no change │ │ QQuery 16│ 17.07 ms │ 17.31 ms │ no change │ │ QQuery 17│ 94.55 ms │ 93.25 ms │ no change │ │ QQuery 18│ 189.44 ms │188.32 ms │ no change │ │ QQuery 19│ 23.73 ms │ 24.38 ms │ no change │ │ QQuery 20│ 31.93 ms │ 30.80 ms │ no change │ │ QQuery 21│ 141.87 ms │139.60 ms │ no change │ │ QQuery 22│ 13.79 ms │ 14.01 ms │ no change │ └──┴───┴──┴───┘ ┏━┳━━━┓ ┃ Benchmark Summary ┃ ┃ ┡━╇━━━┩ │ Total Time (HEAD) │ 1073.08ms │ │ Total Time (topk-filters) │ 1064.25ms │ │ Average Time (HEAD) │ 48.78ms │ │ Average Time (topk-filters) │ 48.37ms │ │ Queries Faster │ 0 │ │ Queries Slower │ 0 │ │ Queries with No Change │22 │ │ Queries with Failure│ 0 │ └─┴───┘ ``` -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
alamb commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-3202448125 🤖 `./gh_compare_branch.sh` [Benchmark Script](https://github.com/alamb/datafusion-benchmarking/blob/main/gh_compare_branch.sh) Running Linux aal-dev 6.11.0-1016-gcp #16~24.04.1-Ubuntu SMP Wed May 28 02:40:52 UTC 2025 x86_64 x86_64 x86_64 GNU/Linux Comparing topk-filters (ffe68565b55ba640f9c5197a4f38106ab0460807) to 8f15991f33bf6aca9d4da8958141b59d196b2ed6 [diff](https://github.com/apache/datafusion/compare/8f15991f33bf6aca9d4da8958141b59d196b2ed6..ffe68565b55ba640f9c5197a4f38106ab0460807) using: tpch_mem Results will be posted here when complete -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
alamb commented on code in PR #16433:
URL: https://github.com/apache/datafusion/pull/16433#discussion_r2286474109
##
datafusion/physical-expr/src/expressions/dynamic_filters.rs:
##
@@ -137,7 +139,7 @@ impl DynamicFilterPhysicalExpr {
Self {
children,
remapped_children: None, // Initially no remapped children
-inner: Arc::new(RwLock::new(Inner::new(inner))),
+inner: Arc::new(ArcSwap::from_pointee(Inner::new(inner))),
Review Comment:
I maybe have been burned one too many times by random maintainers on the
internet deciding to disappear, etc and I am trying to keep the dependency load
down for DataFusion. I realize I may be over sensitive on this topic
--
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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
adriangb commented on code in PR #16433:
URL: https://github.com/apache/datafusion/pull/16433#discussion_r2286456332
##
datafusion/physical-plan/src/topk/mod.rs:
##
@@ -121,13 +122,37 @@ pub struct TopK {
/// Common sort prefix between the input and the sort expressions to allow
early exit optimization
common_sort_prefix: Arc<[PhysicalSortExpr]>,
/// Filter matching the state of the `TopK` heap used for dynamic filter
pushdown
-filter: Option>,
+filter: TopKDynamicFilters,
/// If true, indicates that all rows of subsequent batches are guaranteed
/// to be greater (by byte order, after row conversion) than the top K,
/// which means the top K won't change and the computation can be finished
early.
pub(crate) finished: bool,
}
+#[derive(Debug, Clone)]
+pub struct TopKDynamicFilters {
Review Comment:
Yes because it gets passed into `TopK::try_new` which is currently public.
Maybe `TopK` should be `pub(crate)` then we could make the whole thing
`pub(crate)` but as it stands that's a breaking change.
--
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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
adriangb commented on code in PR #16433:
URL: https://github.com/apache/datafusion/pull/16433#discussion_r2286456332
##
datafusion/physical-plan/src/topk/mod.rs:
##
@@ -121,13 +122,37 @@ pub struct TopK {
/// Common sort prefix between the input and the sort expressions to allow
early exit optimization
common_sort_prefix: Arc<[PhysicalSortExpr]>,
/// Filter matching the state of the `TopK` heap used for dynamic filter
pushdown
-filter: Option>,
+filter: TopKDynamicFilters,
/// If true, indicates that all rows of subsequent batches are guaranteed
/// to be greater (by byte order, after row conversion) than the top K,
/// which means the top K won't change and the computation can be finished
early.
pub(crate) finished: bool,
}
+#[derive(Debug, Clone)]
+pub struct TopKDynamicFilters {
Review Comment:
Yes because it gets passed into `TopK::try_new`
--
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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
adriangb commented on code in PR #16433:
URL: https://github.com/apache/datafusion/pull/16433#discussion_r2286448656
##
datafusion/physical-expr/src/expressions/dynamic_filters.rs:
##
@@ -137,7 +139,7 @@ impl DynamicFilterPhysicalExpr {
Self {
children,
remapped_children: None, // Initially no remapped children
-inner: Arc::new(RwLock::new(Inner::new(inner))),
+inner: Arc::new(ArcSwap::from_pointee(Inner::new(inner))),
Review Comment:
> it seems to have plenty of dependencies
https://crates.io/crates/arc-swap/1.7.1/dependencies
I see no required dependencies. Only dev deps.
> And its last release was over a year ago:
https://crates.io/crates/arc-swap/1.7.1
Personally I don't see any issue with this: I'd look at the ratio of open
issues / obvious bugs vs. releases. If there's no features to add and no major
bugs having few releases can be a good thing 😃
--
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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
alamb commented on code in PR #16433:
URL: https://github.com/apache/datafusion/pull/16433#discussion_r2286140587
##
datafusion/physical-plan/src/topk/mod.rs:
##
@@ -214,41 +239,39 @@ impl TopK {
let mut selected_rows = None;
-if let Some(filter) = self.filter.as_ref() {
-// If a filter is provided, update it with the new rows
-let filter = filter.current()?;
-let filtered = filter.evaluate(&batch)?;
-let num_rows = batch.num_rows();
-let array = filtered.into_array(num_rows)?;
-let mut filter = array.as_boolean().clone();
-let true_count = filter.true_count();
-if true_count == 0 {
-// nothing to filter, so no need to update
-return Ok(());
+// If a filter is provided, update it with the new rows
+let filter = self.filter.expr.current()?;
+let filtered = filter.evaluate(&batch)?;
Review Comment:
A minor nit for a follow on would be to move this code into its own function
like `TopK::update_dynamic_filter` or something to make it easier to understand
##
datafusion/physical-expr/src/expressions/dynamic_filters.rs:
##
@@ -137,7 +139,7 @@ impl DynamicFilterPhysicalExpr {
Self {
children,
remapped_children: None, // Initially no remapped children
-inner: Arc::new(RwLock::new(Inner::new(inner))),
+inner: Arc::new(ArcSwap::from_pointee(Inner::new(inner))),
Review Comment:
I made a PR to show what this looks like without arc_swap
- https://github.com/apache/datafusion/pull/17245
##
datafusion/physical-plan/src/topk/mod.rs:
##
@@ -121,13 +122,37 @@ pub struct TopK {
/// Common sort prefix between the input and the sort expressions to allow
early exit optimization
common_sort_prefix: Arc<[PhysicalSortExpr]>,
/// Filter matching the state of the `TopK` heap used for dynamic filter
pushdown
-filter: Option>,
+filter: TopKDynamicFilters,
/// If true, indicates that all rows of subsequent batches are guaranteed
/// to be greater (by byte order, after row conversion) than the top K,
/// which means the top K won't change and the computation can be finished
early.
pub(crate) finished: bool,
}
+#[derive(Debug, Clone)]
+pub struct TopKDynamicFilters {
Review Comment:
does it need to be pub?
##
datafusion/physical-plan/src/topk/mod.rs:
##
@@ -319,19 +342,84 @@ impl TopK {
/// (a > 2 OR (a = 2 AND b < 3))
/// ```
fn update_filter(&mut self) -> Result<()> {
-let Some(filter) = &self.filter else {
+// If the heap doesn't have k elements yet, we can't create thresholds
+let Some(max_row) = self.heap.max() else {
return Ok(());
};
-let Some(thresholds) = self.heap.get_threshold_values(&self.expr)?
else {
-return Ok(());
+
+let new_threshold_row = &max_row.row;
+
+// Extract scalar values BEFORE acquiring lock to reduce critical
section
+let thresholds = match self.heap.get_threshold_values(&self.expr)? {
+Some(t) => t,
+None => return Ok(()),
};
+// Extract filter expression reference before entering critical section
+let filter_expr = Arc::clone(&self.filter.expr);
+
+// Check if we need to update the threshold (lock-free read)
+let current_threshold = self.filter.threshold_row.load();
+let needs_update = match current_threshold.as_ref() {
+Some(current_row) => {
+// new < current means new threshold is more selective
+current_row.as_slice().cmp(new_threshold_row) ==
Ordering::Greater
+}
+None => true, // No current threshold, so we need to set one
+};
+
+// Only proceed if we need to update
+if needs_update {
+// Build the filter expression OUTSIDE any synchronization
+let predicate = Self::build_filter_expression(&self.expr,
thresholds)?;
+let new_threshold_arc = Arc::new(new_threshold_row.to_vec());
+
+// Atomically update the threshold using compare-and-swap
+let old_threshold = self.filter.threshold_row.compare_and_swap(
+¤t_threshold,
+Some(Arc::clone(&new_threshold_arc)),
+);
+
+// Only update filter if we successfully updated the threshold
+// (or if there was no previous threshold and we're the first)
+let should_update_filter =
+match (old_threshold.as_ref(), current_threshold.as_ref()) {
+// We successfully swapped
+(Some(old), Some(expected)) if Arc::ptr_eq(old, expected)
=> true,
+// We were the first to set it
+(None, None) => true,
+
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
alamb commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-3201878567 🌶️ 🌶️ 🌶️ -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
adriangb commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-3201759854 > Q11 │ 80.40 ms │ 11.82 ms │ +6.80x faster  -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
alamb commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-3201715769 🤖: Benchmark completed Details ``` Comparing HEAD and topk-filters Benchmark run_topk_tpch.json ┏━━┳━━━┳━━┳━━━┓ ┃ Query┃ HEAD ┃ topk-filters ┃Change ┃ ┡━━╇━━━╇━━╇━━━┩ │ Q1 │ 24.21 ms │ 7.05 ms │ +3.43x faster │ │ Q2 │ 31.06 ms │ 30.96 ms │ no change │ │ Q3 │ 100.83 ms │ 92.74 ms │ +1.09x faster │ │ Q4 │ 40.93 ms │ 33.97 ms │ +1.20x faster │ │ Q5 │ 23.65 ms │ 27.52 ms │ 1.16x slower │ │ Q6 │ 45.75 ms │ 49.21 ms │ 1.08x slower │ │ Q7 │ 119.32 ms │125.34 ms │ 1.05x slower │ │ Q8 │ 93.13 ms │ 71.32 ms │ +1.31x faster │ │ Q9 │ 94.83 ms │ 92.51 ms │ no change │ │ Q10 │ 166.33 ms │160.07 ms │ no change │ │ Q11 │ 80.40 ms │ 11.82 ms │ +6.80x faster │ └──┴───┴──┴───┘ ┏━┳━━┓ ┃ Benchmark Summary ┃ ┃ ┡━╇━━┩ │ Total Time (HEAD) │ 820.45ms │ │ Total Time (topk-filters) │ 702.52ms │ │ Average Time (HEAD) │ 74.59ms │ │ Average Time (topk-filters) │ 63.87ms │ │ Queries Faster │5 │ │ Queries Slower │3 │ │ Queries with No Change │3 │ │ Queries with Failure│0 │ └─┴──┘ ``` -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
alamb commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-3201681965 Given the nice performance improvements this seems to provide, I plan to review this PR carefully in the next day or two -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
alamb commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-3201680033 🤖 `./gh_compare_branch.sh` [Benchmark Script](https://github.com/alamb/datafusion-benchmarking/blob/main/gh_compare_branch.sh) Running Linux aal-dev 6.11.0-1016-gcp #16~24.04.1-Ubuntu SMP Wed May 28 02:40:52 UTC 2025 x86_64 x86_64 x86_64 GNU/Linux Comparing topk-filters (3945aba663062debc24757c857a8bd679ad8b223) to 8f15991f33bf6aca9d4da8958141b59d196b2ed6 [diff](https://github.com/apache/datafusion/compare/8f15991f33bf6aca9d4da8958141b59d196b2ed6..3945aba663062debc24757c857a8bd679ad8b223) using: topk_tpch Results will be posted here when complete -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
adriangb commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-3197984660 > QQuery 23│ 7751.39 ms │ 6855.54 ms │ +1.13x faster > QQuery 23│ 7690.27 ms │ 6835.65 ms │ +1.13x faster I'm guessing this is mostly real 😄 -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
alamb commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-3197955377 🤖: Benchmark completed Details ``` Comparing HEAD and topk-filters Benchmark clickbench_extended.json ┏━━┳━┳━━┳━━━┓ ┃ Query┃HEAD ┃ topk-filters ┃Change ┃ ┡━━╇━╇━━╇━━━┩ │ QQuery 0 │ 2053.78 ms │ 1974.31 ms │ no change │ │ QQuery 1 │ 735.78 ms │660.02 ms │ +1.11x faster │ │ QQuery 2 │ 1460.75 ms │ 1307.73 ms │ +1.12x faster │ │ QQuery 3 │ 623.71 ms │617.23 ms │ no change │ │ QQuery 4 │ 1285.31 ms │ 1298.10 ms │ no change │ │ QQuery 5 │ 13863.88 ms │ 14128.38 ms │ no change │ │ QQuery 6 │ 2383.77 ms │ 2341.30 ms │ no change │ │ QQuery 7 │ 1951.39 ms │ 1953.50 ms │ no change │ └──┴─┴──┴───┘ ┏━┳┓ ┃ Benchmark Summary ┃┃ ┡━╇┩ │ Total Time (HEAD) │ 24358.37ms │ │ Total Time (topk-filters) │ 24280.57ms │ │ Average Time (HEAD) │ 3044.80ms │ │ Average Time (topk-filters) │ 3035.07ms │ │ Queries Faster │ 2 │ │ Queries Slower │ 0 │ │ Queries with No Change │ 6 │ │ Queries with Failure│ 0 │ └─┴┘ Benchmark clickbench_partitioned.json ┏━━┳━┳━━┳━━━┓ ┃ Query┃HEAD ┃ topk-filters ┃Change ┃ ┡━━╇━╇━━╇━━━┩ │ QQuery 0 │ 2.30 ms │ 2.32 ms │ no change │ │ QQuery 1 │29.71 ms │ 28.77 ms │ no change │ │ QQuery 2 │77.53 ms │ 76.14 ms │ no change │ │ QQuery 3 │94.53 ms │ 90.05 ms │ no change │ │ QQuery 4 │ 597.57 ms │609.85 ms │ no change │ │ QQuery 5 │ 880.91 ms │881.20 ms │ no change │ │ QQuery 6 │ 2.22 ms │ 2.35 ms │ 1.05x slower │ │ QQuery 7 │31.73 ms │ 33.48 ms │ 1.06x slower │ │ QQuery 8 │ 861.09 ms │883.18 ms │ no change │ │ QQuery 9 │ 1167.22 ms │ 1144.09 ms │ no change │ │ QQuery 10│ 233.16 ms │227.48 ms │ no change │ │ QQuery 11│ 263.75 ms │259.39 ms │ no change │ │ QQuery 12│ 865.53 ms │865.13 ms │ no change │ │ QQuery 13│ 1180.94 ms │ 1231.48 ms │ no change │ │ QQuery 14│ 809.24 ms │809.07 ms │ no change │ │ QQuery 15│ 775.27 ms │777.78 ms │ no change │ │ QQuery 16│ 1619.25 ms │ 1639.14 ms │ no change │ │ QQuery 17│ 1608.47 ms │ 1643.10 ms │ no change │ │ QQuery 18│ 2875.25 ms │ 3002.07 ms │ no change │ │ QQuery 19│79.14 ms │ 81.14 ms │ no change │ │ QQuery 20│ 1227.30 ms │ 1234.70 ms │ no change │ │ QQuery 21│ 1377.70 ms │ 1356.98 ms │ no change │ │ QQuery 22│ 2268.17 ms │ 2238.49 ms │ no change │ │ QQuery 23│ 7690.27 ms │ 6835.65 ms │ +1.13x faster │ │ QQuery 24│ 426.90 ms │417.85 ms │ no change │ │ QQuery 25│ 296.39 ms │294.12 ms │ no change │ │ QQuery 26│ 428.93 ms │417.70 ms │ no change │ │ QQuery 27│ 1618.21 ms │ 1573.12 ms │ no change │ │ QQuery 28│ 12002.03 ms │ 12016.65 ms │ no change │ │ QQuery 29│ 520.45 ms │523.19 ms │ no change │ │ QQuery 30│ 774.86 ms │761.23 ms │ no change │ │ QQuery 31│ 794.38 ms │783.65 ms │ no change │ │ QQuery 32│ 2461.37 ms │ 2452.40 ms │ no change │ │ QQuery 33│ 3218.56 ms │ 3214.72 ms │ no change │ │ QQuery 34│ 3200.35 ms │ 3223.54 ms │ no change │ │ QQuery 35│ 1267.63 ms │ 1276.62 ms │ no change │ │ QQuery 36│ 125.57 ms │121.93 ms │ no change │ │ QQuery 37│53.62 ms │ 53.02 ms │ no change │ │ QQuery 38│ 127.07 ms │124.19 ms │ no change │ │ QQuery 39│ 206.77 ms │199.55 ms │ no change │ │ QQuery 40│41.71 ms │ 41.96 ms │ no change │ │ QQuery 41│39.44 ms │ 38.61 ms │ no change │ │ QQuery 42│32.99 ms │ 32.51 ms │ no change │ └──┴─┴──┴───┘ ┏━┳┓ ┃ Benchmark Summary ┃┃ ┡━╇┩ │ Total Time (HEAD) │ 54255.46ms │ │ Total Time
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
Dandandan commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-3197945496 Really nice 🚀 -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
adriangb commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-3197906730 @alamb I feel like the interesting benchmark to run is `topk_tpch` -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
alamb commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-3197868163 🤖 `./gh_compare_branch.sh` [Benchmark Script](https://github.com/alamb/datafusion-benchmarking/blob/main/gh_compare_branch.sh) Running Linux aal-dev 6.11.0-1016-gcp #16~24.04.1-Ubuntu SMP Wed May 28 02:40:52 UTC 2025 x86_64 x86_64 x86_64 GNU/Linux Comparing topk-filters (3945aba663062debc24757c857a8bd679ad8b223) to 8f15991f33bf6aca9d4da8958141b59d196b2ed6 [diff](https://github.com/apache/datafusion/compare/8f15991f33bf6aca9d4da8958141b59d196b2ed6..3945aba663062debc24757c857a8bd679ad8b223) using: tpch_mem clickbench_partitioned clickbench_extended Results will be posted here when complete -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
adriangb commented on code in PR #16433:
URL: https://github.com/apache/datafusion/pull/16433#discussion_r2282817078
##
datafusion/physical-expr/src/expressions/dynamic_filters.rs:
##
@@ -137,7 +139,7 @@ impl DynamicFilterPhysicalExpr {
Self {
children,
remapped_children: None, // Initially no remapped children
-inner: Arc::new(RwLock::new(Inner::new(inner))),
+inner: Arc::new(ArcSwap::from_pointee(Inner::new(inner))),
Review Comment:
Well I stumbled upon it because we originally had some performance
regressions and I suspected they might be due to hitting the `Arc>`s a
lot heavier. It seems that `ArcSwap` is the solution to that but I haven't
benchmarked to confirm the performance impact / if it's that or other
refactors. From a micro benchmark sitauation it makes sense. I can make a
branch / another PR without ArcSwap and we can see if there's a measurable
change at the macro level, but I expect it may be hard to measure.
That said it's a well maintained dep from a reputable maintainer, has a lot
of usage and has no dependencies of its own.
--
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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
adriangb commented on code in PR #16433:
URL: https://github.com/apache/datafusion/pull/16433#discussion_r2282817078
##
datafusion/physical-expr/src/expressions/dynamic_filters.rs:
##
@@ -137,7 +139,7 @@ impl DynamicFilterPhysicalExpr {
Self {
children,
remapped_children: None, // Initially no remapped children
-inner: Arc::new(RwLock::new(Inner::new(inner))),
+inner: Arc::new(ArcSwap::from_pointee(Inner::new(inner))),
Review Comment:
Well I stumbled upon it because we originally had some performance
regressions and I suspected they might be due to hitting the `Arc>`s a
lot heavier. It seems that `ArcSwap` is the solution to that but I haven't
benchmarked to confirm the performance impact / if it's that or other
refactors. From a micro benchmark sitauation it makes sense. I can make a
branch / another PR without ArcSwap and we can see if there's a measurable
change at the macro level.
That said it's a well maintained dep from a reputable maintainer, has a lot
of usage and has no dependencies of its own.
--
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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
alamb commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-3197557763 🤖: Benchmark completed Details ``` Comparing HEAD and topk-filters Benchmark clickbench_extended.json ┏━━┳━┳━━┳━━━┓ ┃ Query┃HEAD ┃ topk-filters ┃Change ┃ ┡━━╇━╇━━╇━━━┩ │ QQuery 0 │ 2098.75 ms │ 2062.77 ms │ no change │ │ QQuery 1 │ 736.19 ms │636.32 ms │ +1.16x faster │ │ QQuery 2 │ 1447.86 ms │ 1301.32 ms │ +1.11x faster │ │ QQuery 3 │ 640.39 ms │605.76 ms │ +1.06x faster │ │ QQuery 4 │ 1297.18 ms │ 1346.75 ms │ no change │ │ QQuery 5 │ 13751.05 ms │ 14220.36 ms │ no change │ │ QQuery 6 │ 2400.54 ms │ 2335.19 ms │ no change │ │ QQuery 7 │ 1933.90 ms │ 2000.55 ms │ no change │ └──┴─┴──┴───┘ ┏━┳┓ ┃ Benchmark Summary ┃┃ ┡━╇┩ │ Total Time (HEAD) │ 24305.85ms │ │ Total Time (topk-filters) │ 24509.01ms │ │ Average Time (HEAD) │ 3038.23ms │ │ Average Time (topk-filters) │ 3063.63ms │ │ Queries Faster │ 3 │ │ Queries Slower │ 0 │ │ Queries with No Change │ 5 │ │ Queries with Failure│ 0 │ └─┴┘ Benchmark clickbench_partitioned.json ┏━━┳━┳━━┳━━━┓ ┃ Query┃HEAD ┃ topk-filters ┃Change ┃ ┡━━╇━╇━━╇━━━┩ │ QQuery 0 │ 2.29 ms │ 2.43 ms │ 1.06x slower │ │ QQuery 1 │28.83 ms │ 29.37 ms │ no change │ │ QQuery 2 │75.39 ms │ 76.90 ms │ no change │ │ QQuery 3 │88.02 ms │ 90.54 ms │ no change │ │ QQuery 4 │ 585.22 ms │605.52 ms │ no change │ │ QQuery 5 │ 849.36 ms │886.24 ms │ no change │ │ QQuery 6 │ 2.32 ms │ 2.36 ms │ no change │ │ QQuery 7 │31.99 ms │ 33.28 ms │ no change │ │ QQuery 8 │ 830.49 ms │866.81 ms │ no change │ │ QQuery 9 │ 1110.42 ms │ 1160.29 ms │ no change │ │ QQuery 10│ 221.99 ms │221.95 ms │ no change │ │ QQuery 11│ 255.21 ms │260.43 ms │ no change │ │ QQuery 12│ 826.88 ms │862.45 ms │ no change │ │ QQuery 13│ 1186.96 ms │ 1159.18 ms │ no change │ │ QQuery 14│ 767.58 ms │799.01 ms │ no change │ │ QQuery 15│ 717.27 ms │794.54 ms │ 1.11x slower │ │ QQuery 16│ 1552.61 ms │ 1604.86 ms │ no change │ │ QQuery 17│ 1542.14 ms │ 1589.89 ms │ no change │ │ QQuery 18│ 2895.45 ms │ 2868.94 ms │ no change │ │ QQuery 19│79.79 ms │ 81.57 ms │ no change │ │ QQuery 20│ 1278.75 ms │ 1211.54 ms │ +1.06x faster │ │ QQuery 21│ 1374.67 ms │ 1346.29 ms │ no change │ │ QQuery 22│ 2277.06 ms │ 2246.93 ms │ no change │ │ QQuery 23│ 7751.39 ms │ 6855.54 ms │ +1.13x faster │ │ QQuery 24│ 423.18 ms │406.23 ms │ no change │ │ QQuery 25│ 294.80 ms │289.58 ms │ no change │ │ QQuery 26│ 421.41 ms │404.45 ms │ no change │ │ QQuery 27│ 1609.35 ms │ 1548.29 ms │ no change │ │ QQuery 28│ 12027.18 ms │ 11975.67 ms │ no change │ │ QQuery 29│ 499.72 ms │516.83 ms │ no change │ │ QQuery 30│ 762.44 ms │768.76 ms │ no change │ │ QQuery 31│ 788.02 ms │800.18 ms │ no change │ │ QQuery 32│ 2404.71 ms │ 2486.31 ms │ no change │ │ QQuery 33│ 3305.56 ms │ 3284.68 ms │ no change │ │ QQuery 34│ 3362.30 ms │ 3402.04 ms │ no change │ │ QQuery 35│ 1299.53 ms │ 1264.01 ms │ no change │ │ QQuery 36│ 133.03 ms │130.08 ms │ no change │ │ QQuery 37│54.36 ms │ 56.59 ms │ no change │ │ QQuery 38│ 124.72 ms │124.72 ms │ no change │ │ QQuery 39│ 210.60 ms │203.82 ms │ no change │ │ QQuery 40│46.04 ms │ 42.45 ms │ +1.08x faster │ │ QQuery 41│39.97 ms │ 40.83 ms │ no change │ │ QQuery 42│33.44 ms │ 33.46 ms │ no change │ └──┴─┴──┴───┘ ┏━┳┓ ┃ Benchmark Summary ┃┃ ┡━╇┩ │ Total Time (HEAD) │ 54172.42ms │ │ Total Time
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
adriangb commented on code in PR #16433:
URL: https://github.com/apache/datafusion/pull/16433#discussion_r2282817078
##
datafusion/physical-expr/src/expressions/dynamic_filters.rs:
##
@@ -137,7 +139,7 @@ impl DynamicFilterPhysicalExpr {
Self {
children,
remapped_children: None, // Initially no remapped children
-inner: Arc::new(RwLock::new(Inner::new(inner))),
+inner: Arc::new(ArcSwap::from_pointee(Inner::new(inner))),
Review Comment:
Well I stumbled upon it because we originally had some performance
regressions and I suspected they might be due to hitting the `Arc>`s a
lot heavier. It seems that `ArcSwap` is the solution to that but I haven't
benchmarked to confirm the performance impact / if it's that or other
refactors. From a micro benchmark sitauation it makes sense. I can make a
branch / another PR without ArcSwap and we can confirm from there if we want.
--
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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
alamb commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-3197411815 🤖 `./gh_compare_branch.sh` [Benchmark Script](https://github.com/alamb/datafusion-benchmarking/blob/main/gh_compare_branch.sh) Running Linux aal-dev 6.11.0-1016-gcp #16~24.04.1-Ubuntu SMP Wed May 28 02:40:52 UTC 2025 x86_64 x86_64 x86_64 GNU/Linux Comparing topk-filters (b7ac320c0d9fcc72f2bb3750cf19ea7be12ff320) to 8f15991f33bf6aca9d4da8958141b59d196b2ed6 [diff](https://github.com/apache/datafusion/compare/8f15991f33bf6aca9d4da8958141b59d196b2ed6..b7ac320c0d9fcc72f2bb3750cf19ea7be12ff320) using: tpch_mem clickbench_partitioned clickbench_extended Results will be posted here when complete -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
alamb commented on code in PR #16433:
URL: https://github.com/apache/datafusion/pull/16433#discussion_r2282739532
##
datafusion/physical-expr/src/expressions/dynamic_filters.rs:
##
@@ -137,7 +139,7 @@ impl DynamicFilterPhysicalExpr {
Self {
children,
remapped_children: None, // Initially no remapped children
-inner: Arc::new(RwLock::new(Inner::new(inner))),
+inner: Arc::new(ArcSwap::from_pointee(Inner::new(inner))),
Review Comment:
do we have evidence that using ArcSwap is needed for performance?
I think we could use `parking_lot` mutexes if we wanted to make the code
cleaner (not check for Lock poisioning)
##
datafusion/physical-plan/src/sorts/sort.rs:
##
@@ -1158,7 +1153,10 @@ impl ExecutionPlan for SortExec {
context.session_config().batch_size(),
context.runtime_env(),
&self.metrics_set,
-self.filter.clone(),
+self.filter
+.as_ref()
+.expect("Filter should be set when fetch is Some")
+.clone(),
Review Comment:
I recommend turning this into an internal error so that if someone does hit
this for some reason the symptom is less severe
--
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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
adriangb commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-3193850159 Okay I've re-run after 310100b and am now seeing only improvements across the board. @Dandandan I've requested another review from you since I think this may be in a good place now. ``` Comparing main and topk-filters Benchmark run_topk_tpch.json ┏━━┳━━┳━━┳━━━┓ ┃ Query┃ main ┃ topk-filters ┃Change ┃ ┡━━╇━━╇━━╇━━━┩ │ Q1 │ 11.22 ms │ 9.41 ms │ +1.19x faster │ │ Q2 │ 13.90 ms │ 11.78 ms │ +1.18x faster │ │ Q3 │ 33.71 ms │ 29.55 ms │ +1.14x faster │ │ Q4 │ 13.81 ms │ 12.89 ms │ +1.07x faster │ │ Q5 │ 8.65 ms │ 7.04 ms │ +1.23x faster │ │ Q6 │ 14.67 ms │ 14.65 ms │ no change │ │ Q7 │ 45.67 ms │ 38.13 ms │ +1.20x faster │ │ Q8 │ 27.73 ms │ 25.61 ms │ +1.08x faster │ │ Q9 │ 35.25 ms │ 34.09 ms │ no change │ │ Q10 │ 56.27 ms │ 52.07 ms │ +1.08x faster │ │ Q11 │ 32.06 ms │ 30.42 ms │ +1.05x faster │ └──┴──┴──┴───┘ ┏━┳━━┓ ┃ Benchmark Summary ┃ ┃ ┡━╇━━┩ │ Total Time (main) │ 292.93ms │ │ Total Time (topk-filters) │ 265.65ms │ │ Average Time (main) │ 26.63ms │ │ Average Time (topk-filters) │ 24.15ms │ │ Queries Faster │9 │ │ Queries Slower │0 │ │ Queries with No Change │2 │ │ Queries with Failure│0 │ └─┴──┘ ``` -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
adriangb commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-3184720872 ClickBench results for partitioned datasets look much better: ``` ┏━━┳┳━━┳━━━┓ ┃ Query┃ main ┃ topk-filters ┃Change ┃ ┡━━╇╇━━╇━━━┩ │ QQuery 0 │0.92 ms │ 1.15 ms │ 1.25x slower │ │ QQuery 1 │ 12.45 ms │ 11.01 ms │ +1.13x faster │ │ QQuery 2 │ 42.41 ms │ 35.19 ms │ +1.21x faster │ │ QQuery 3 │ 40.63 ms │ 33.55 ms │ +1.21x faster │ │ QQuery 4 │ 337.67 ms │267.36 ms │ +1.26x faster │ │ QQuery 5 │ 434.44 ms │362.25 ms │ +1.20x faster │ │ QQuery 6 │1.21 ms │ 1.05 ms │ +1.15x faster │ │ QQuery 7 │ 14.20 ms │ 11.94 ms │ +1.19x faster │ │ QQuery 8 │ 413.55 ms │317.13 ms │ +1.30x faster │ │ QQuery 9 │ 601.85 ms │480.06 ms │ +1.25x faster │ │ QQuery 10│ 100.43 ms │ 84.75 ms │ +1.18x faster │ │ QQuery 11│ 108.91 ms │ 91.61 ms │ +1.19x faster │ │ QQuery 12│ 419.13 ms │323.82 ms │ +1.29x faster │ │ QQuery 13│ 570.70 ms │481.27 ms │ +1.19x faster │ │ QQuery 14│ 428.36 ms │310.63 ms │ +1.38x faster │ │ QQuery 15│ 426.88 ms │309.22 ms │ +1.38x faster │ │ QQuery 16│ 1251.45 ms │729.69 ms │ +1.72x faster │ │ QQuery 17│ 1175.59 ms │738.35 ms │ +1.59x faster │ │ QQuery 18│ 2670.84 ms │ 1945.33 ms │ +1.37x faster │ │ QQuery 19│ 41.71 ms │ 32.77 ms │ +1.27x faster │ │ QQuery 20│ 1002.98 ms │738.13 ms │ +1.36x faster │ │ QQuery 21│ 969.39 ms │763.04 ms │ +1.27x faster │ │ QQuery 22│ 1506.05 ms │ 1214.86 ms │ +1.24x faster │ │ QQuery 23│ 4418.24 ms │ 4008.72 ms │ +1.10x faster │ │ QQuery 24│ 198.37 ms │196.65 ms │ no change │ │ QQuery 25│ 150.97 ms │150.72 ms │ no change │ │ QQuery 26│ 200.45 ms │191.65 ms │ no change │ │ QQuery 27│ 1007.57 ms │893.24 ms │ +1.13x faster │ │ QQuery 28│ 7214.47 ms │ 7065.58 ms │ no change │ │ QQuery 29│ 356.81 ms │327.34 ms │ +1.09x faster │ │ QQuery 30│ 323.66 ms │310.25 ms │ no change │ │ QQuery 31│ 366.94 ms │346.37 ms │ +1.06x faster │ │ QQuery 32│ 2028.11 ms │ 1692.80 ms │ +1.20x faster │ │ QQuery 33│ 2286.38 ms │ 1775.46 ms │ +1.29x faster │ │ QQuery 34│ 1978.34 ms │ 2305.40 ms │ 1.17x slower │ │ QQuery 35│ 555.06 ms │552.82 ms │ no change │ │ QQuery 36│ 61.16 ms │ 62.54 ms │ no change │ │ QQuery 37│ 24.59 ms │ 23.99 ms │ no change │ │ QQuery 38│ 61.70 ms │ 63.87 ms │ no change │ │ QQuery 39│ 102.67 ms │103.98 ms │ no change │ │ QQuery 40│ 17.16 ms │ 18.43 ms │ 1.07x slower │ │ QQuery 41│ 16.16 ms │ 17.05 ms │ 1.06x slower │ │ QQuery 42│ 12.72 ms │ 13.14 ms │ no change │ └──┴┴──┴───┘ ``` -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
adriangb commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-3184612966 I tried some stuff. Still not an obvious improvement: ``` /bench.sh compare main topk-filters Comparing main and topk-filters Benchmark run_topk_tpch.json ┏━━┳━━┳━━┳━━━┓ ┃ Query┃ main ┃ topk-filters ┃Change ┃ ┡━━╇━━╇━━╇━━━┩ │ Q1 │ 12.46 ms │ 10.55 ms │ +1.18x faster │ │ Q2 │ 14.02 ms │ 12.06 ms │ +1.16x faster │ │ Q3 │ 29.39 ms │ 30.10 ms │ no change │ │ Q4 │ 13.74 ms │ 12.79 ms │ +1.07x faster │ │ Q5 │ 7.44 ms │ 7.09 ms │ no change │ │ Q6 │ 15.87 ms │ 13.79 ms │ +1.15x faster │ │ Q7 │ 37.22 ms │ 44.24 ms │ 1.19x slower │ │ Q8 │ 27.51 ms │ 25.17 ms │ +1.09x faster │ │ Q9 │ 33.04 ms │ 31.28 ms │ +1.06x faster │ │ Q10 │ 55.26 ms │ 58.96 ms │ 1.07x slower │ │ Q11 │ 28.74 ms │ 29.75 ms │ no change │ └──┴──┴──┴───┘ ┏━┳━━┓ ┃ Benchmark Summary ┃ ┃ ┡━╇━━┩ │ Total Time (main) │ 274.68ms │ │ Total Time (topk-filters) │ 275.79ms │ │ Average Time (main) │ 24.97ms │ │ Average Time (topk-filters) │ 25.07ms │ │ Queries Faster │6 │ │ Queries Slower │2 │ │ Queries with No Change │3 │ │ Queries with Failure│0 │ └─┴──┘ ``` -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
Dandandan commented on code in PR #16433:
URL: https://github.com/apache/datafusion/pull/16433#discussion_r2186109216
##
datafusion/physical-plan/src/topk/mod.rs:
##
@@ -319,19 +342,75 @@ impl TopK {
/// (a > 2 OR (a = 2 AND b < 3))
/// ```
fn update_filter(&mut self) -> Result<()> {
-let Some(filter) = &self.filter else {
+// If the heap doesn't have k elements yet, we can't create thresholds
+let Some(max_row) = self.heap.max() else {
return Ok(());
};
-let Some(thresholds) = self.heap.get_threshold_values(&self.expr)?
else {
-return Ok(());
+
+let new_threshold_row = &max_row.row;
+
+// Extract filter expression reference before entering critical section
+let filter_expr = Arc::clone(&self.filter.expr);
+
+// Check if we need to update and do both threshold and filter update
atomically
+{
+let mut threshold_guard = self.filter.threshold_row.write();
+if let Some(current_row) = threshold_guard.as_ref() {
+match current_row.as_slice().cmp(new_threshold_row) {
+Ordering::Greater => {
+// new < current, so new threshold is more selective
Review Comment:
I guess we should try changing it to the global heap thing - write updates
to the shared/global heap and update the filter based on global heap.
--
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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
Dandandan commented on code in PR #16433:
URL: https://github.com/apache/datafusion/pull/16433#discussion_r2186107501
##
datafusion/physical-plan/src/topk/mod.rs:
##
@@ -319,19 +342,75 @@ impl TopK {
/// (a > 2 OR (a = 2 AND b < 3))
/// ```
fn update_filter(&mut self) -> Result<()> {
-let Some(filter) = &self.filter else {
+// If the heap doesn't have k elements yet, we can't create thresholds
+let Some(max_row) = self.heap.max() else {
return Ok(());
};
-let Some(thresholds) = self.heap.get_threshold_values(&self.expr)?
else {
-return Ok(());
+
+let new_threshold_row = &max_row.row;
+
+// Extract filter expression reference before entering critical section
+let filter_expr = Arc::clone(&self.filter.expr);
+
+// Check if we need to update and do both threshold and filter update
atomically
+{
+let mut threshold_guard = self.filter.threshold_row.write();
+if let Some(current_row) = threshold_guard.as_ref() {
+match current_row.as_slice().cmp(new_threshold_row) {
+Ordering::Greater => {
+// new < current, so new threshold is more selective
Review Comment:
The current solution seems roughly on par with main, I don't observe a
speedup.
--
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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
Dandandan commented on code in PR #16433:
URL: https://github.com/apache/datafusion/pull/16433#discussion_r2186106170
##
datafusion/physical-plan/src/topk/mod.rs:
##
@@ -319,19 +342,75 @@ impl TopK {
/// (a > 2 OR (a = 2 AND b < 3))
/// ```
fn update_filter(&mut self) -> Result<()> {
-let Some(filter) = &self.filter else {
+// If the heap doesn't have k elements yet, we can't create thresholds
+let Some(max_row) = self.heap.max() else {
return Ok(());
};
-let Some(thresholds) = self.heap.get_threshold_values(&self.expr)?
else {
-return Ok(());
+
+let new_threshold_row = &max_row.row;
+
+// Extract filter expression reference before entering critical section
+let filter_expr = Arc::clone(&self.filter.expr);
+
+// Check if we need to update and do both threshold and filter update
atomically
+{
+let mut threshold_guard = self.filter.threshold_row.write();
+if let Some(current_row) = threshold_guard.as_ref() {
+match current_row.as_slice().cmp(new_threshold_row) {
+Ordering::Greater => {
+// new < current, so new threshold is more selective
Review Comment:
I think this was wrong before @adriangb - in the heap lower means more
selective
--
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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
Dandandan commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-3037193167 I am taking a look now, see if I can find a thing -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
adriangb commented on code in PR #16433:
URL: https://github.com/apache/datafusion/pull/16433#discussion_r2185669579
##
datafusion/physical-plan/src/topk/mod.rs:
##
@@ -214,41 +238,39 @@ impl TopK {
let mut selected_rows = None;
-if let Some(filter) = self.filter.as_ref() {
-// If a filter is provided, update it with the new rows
-let filter = filter.current()?;
-let filtered = filter.evaluate(&batch)?;
-let num_rows = batch.num_rows();
-let array = filtered.into_array(num_rows)?;
-let mut filter = array.as_boolean().clone();
-let true_count = filter.true_count();
-if true_count == 0 {
-// nothing to filter, so no need to update
-return Ok(());
+// If a filter is provided, update it with the new rows
Review Comment:
Agreed
--
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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
Dandandan commented on code in PR #16433:
URL: https://github.com/apache/datafusion/pull/16433#discussion_r2185350825
##
datafusion/physical-plan/src/topk/mod.rs:
##
@@ -214,41 +238,39 @@ impl TopK {
let mut selected_rows = None;
-if let Some(filter) = self.filter.as_ref() {
-// If a filter is provided, update it with the new rows
-let filter = filter.current()?;
-let filtered = filter.evaluate(&batch)?;
-let num_rows = batch.num_rows();
-let array = filtered.into_array(num_rows)?;
-let mut filter = array.as_boolean().clone();
-let true_count = filter.true_count();
-if true_count == 0 {
-// nothing to filter, so no need to update
-return Ok(());
+// If a filter is provided, update it with the new rows
Review Comment:
Ah yeah I understand... maybe we should check if it is unset / true as
additional "fast path".
But I think that's not the source of the slow down.
##
datafusion/physical-plan/src/topk/mod.rs:
##
@@ -214,41 +238,39 @@ impl TopK {
let mut selected_rows = None;
-if let Some(filter) = self.filter.as_ref() {
-// If a filter is provided, update it with the new rows
-let filter = filter.current()?;
-let filtered = filter.evaluate(&batch)?;
-let num_rows = batch.num_rows();
-let array = filtered.into_array(num_rows)?;
-let mut filter = array.as_boolean().clone();
-let true_count = filter.true_count();
-if true_count == 0 {
-// nothing to filter, so no need to update
-return Ok(());
+// If a filter is provided, update it with the new rows
Review Comment:
Ah yeah I understand... maybe we should check if it is unset / returning
constant true as additional "fast path".
But I think that's not the source of the slow down.
--
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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
adriangb commented on code in PR #16433:
URL: https://github.com/apache/datafusion/pull/16433#discussion_r2185318951
##
datafusion/physical-plan/src/topk/mod.rs:
##
@@ -214,41 +238,39 @@ impl TopK {
let mut selected_rows = None;
-if let Some(filter) = self.filter.as_ref() {
-// If a filter is provided, update it with the new rows
-let filter = filter.current()?;
-let filtered = filter.evaluate(&batch)?;
-let num_rows = batch.num_rows();
-let array = filtered.into_array(num_rows)?;
-let mut filter = array.as_boolean().clone();
-let true_count = filter.true_count();
-if true_count == 0 {
-// nothing to filter, so no need to update
-return Ok(());
+// If a filter is provided, update it with the new rows
Review Comment:
I think Some(filter) was always passed into the constructor
--
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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
Dandandan commented on code in PR #16433:
URL: https://github.com/apache/datafusion/pull/16433#discussion_r2185314221
##
datafusion/physical-plan/src/topk/mod.rs:
##
@@ -214,41 +238,39 @@ impl TopK {
let mut selected_rows = None;
-if let Some(filter) = self.filter.as_ref() {
-// If a filter is provided, update it with the new rows
-let filter = filter.current()?;
-let filtered = filter.evaluate(&batch)?;
-let num_rows = batch.num_rows();
-let array = filtered.into_array(num_rows)?;
-let mut filter = array.as_boolean().clone();
-let true_count = filter.true_count();
-if true_count == 0 {
-// nothing to filter, so no need to update
-return Ok(());
+// If a filter is provided, update it with the new rows
Review Comment:
Hm.. it was used, as in the initial phase of topk there might not be
initialized (heap not yet filled) for larger values for limit.
--
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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
adriangb commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-3036131650 I did a bench run, confounding results: ``` ┏━━┳━━┳━━┳━━━┓ ┃ Query┃ main ┃ topk-filters ┃Change ┃ ┡━━╇━━╇━━╇━━━┩ │ Q1 │ 9.96 ms │ 10.70 ms │ 1.07x slower │ │ Q2 │ 12.63 ms │ 15.30 ms │ 1.21x slower │ │ Q3 │ 33.45 ms │ 35.22 ms │ 1.05x slower │ │ Q4 │ 15.22 ms │ 16.82 ms │ 1.11x slower │ │ Q5 │ 9.87 ms │ 8.85 ms │ +1.12x faster │ │ Q6 │ 16.54 ms │ 22.16 ms │ 1.34x slower │ │ Q7 │ 43.41 ms │ 44.00 ms │ no change │ │ Q8 │ 22.91 ms │ 34.56 ms │ 1.51x slower │ │ Q9 │ 38.68 ms │ 39.60 ms │ no change │ │ Q10 │ 65.73 ms │ 73.24 ms │ 1.11x slower │ │ Q11 │ 34.78 ms │ 37.14 ms │ 1.07x slower │ └──┴──┴──┴───┘ ┏━┳━━┓ ┃ Benchmark Summary ┃ ┃ ┡━╇━━┩ │ Total Time (main) │ 303.17ms │ │ Total Time (topk-filters) │ 337.58ms │ │ Average Time (main) │ 27.56ms │ │ Average Time (topk-filters) │ 30.69ms │ │ Queries Faster │1 │ │ Queries Slower │8 │ │ Queries with No Change │2 │ │ Queries with Failure│0 │ └─┴──┘ ``` Mostly... slower -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
adriangb commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-2989360115 > how would you compute the shared heap on the fly? I was thinking you'd compute the top K of the top K * partitions on the fly. But maybe your proposal makes more sense. -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
Dandandan commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-2989336247 > We could try a shared heap. It might work? I guess it will be a sort of balance between lock contention and better selectivity. Maybe we can balance it by having distinct heaps for writes with no locks but read only references to all of them so that when we do reads we compute on the fly the "combined" heap? Then we don't need any locks. The cost is that computations on the heap are larger but as long as `k ~ constant` then it should be fine. how would you compute the shared heap on the fly? I was thinking something similar: each write to a own heap, only write the ones that updated the local heap to the shared heap (limiting the lock access time). -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
adriangb commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-2988328335 We could try a shared heap. It might work? I guess it will be a sort of balance between lock contention and better selectivity. Maybe we can balance it by having distinct heaps for writes with no locks but read only references to all of them so that when we do reads we compute on the fly the "combined" heap? Then we don't need any locks. The cost is that computations on the heap are larger but as long as `k ~ constant` then it should be fine. -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
Dandandan commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-2988289302 > > Hm my earlier benchmarks didn't seem correct. not sure where the earlier run came from 🤔 > > What do the current ones show? Not much improvement? Yes about the same compared to main. -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
adriangb commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-2987967895 > Hm my earlier benchmarks didn't seem correct. not sure where the earlier run came from 🤔 What do the current ones show? Not much improvement? -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
adriangb commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-2987966449 > I am wondering if we can somehow synchronize the values in the heap efficiently in order to make the filter for a topk(n*partitions) as efficient as topk(n) 🤔 I think the only way to do that would be to make a global heap and put a lock on it, similar to what we do with the filters? -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
adriangb commented on code in PR #16433:
URL: https://github.com/apache/datafusion/pull/16433#discussion_r2156917768
##
datafusion/physical-plan/src/topk/mod.rs:
##
@@ -214,41 +238,39 @@ impl TopK {
let mut selected_rows = None;
-if let Some(filter) = self.filter.as_ref() {
-// If a filter is provided, update it with the new rows
-let filter = filter.current()?;
-let filtered = filter.evaluate(&batch)?;
-let num_rows = batch.num_rows();
-let array = filtered.into_array(num_rows)?;
-let mut filter = array.as_boolean().clone();
-let true_count = filter.true_count();
-if true_count == 0 {
-// nothing to filter, so no need to update
-return Ok(());
+// If a filter is provided, update it with the new rows
Review Comment:
Yeah I don't think that was useful /being used as a code path?
--
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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
Dandandan commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-2987413335 (Removed a non-working proposal). I am wondering if we can somehow synchronize the values in the heap efficiently in order to make the filter for a `topk(n*partitions)` as efficient as `topk(n)` 🤔 -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
Dandandan commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-2987392519 WDYT @adriangb ? -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
Dandandan commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-2987381689 Synchronizing the max between all partitions will help a bit - but I wonder if we shouldn't just synchronize it based on the true TopK in SortPreservingMergeExec ? -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
Dandandan commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-2987370039 In hindsight I think actually another fact that we don't see topk being as effective with more partitions is that spreading them over partitions will essentially make `topk(n)` into a `topk(n*partitions)`. So a limit of `10` will be `200` with 20 partitions. -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
Dandandan commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-2986650188 Hm my earlier benchmarks didn't seem correct. not sure where the earlier run came from 🤔 -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
Dandandan commented on code in PR #16433:
URL: https://github.com/apache/datafusion/pull/16433#discussion_r2156139567
##
datafusion/physical-plan/src/topk/mod.rs:
##
@@ -319,13 +341,87 @@ impl TopK {
/// (a > 2 OR (a = 2 AND b < 3))
/// ```
fn update_filter(&mut self) -> Result<()> {
-let Some(filter) = &self.filter else {
-return Ok(());
-};
let Some(thresholds) = self.heap.get_threshold_values(&self.expr)?
else {
return Ok(());
};
+// Are the new thresholds more selective than our existing ones?
+let should_update = {
+let mut current = self.filter.thresholds.write();
Review Comment:
I think this would be a good idea anyway to simplify the code.
--
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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
Dandandan commented on code in PR #16433:
URL: https://github.com/apache/datafusion/pull/16433#discussion_r2156133514
##
datafusion/physical-plan/src/topk/mod.rs:
##
@@ -319,13 +341,87 @@ impl TopK {
/// (a > 2 OR (a = 2 AND b < 3))
/// ```
fn update_filter(&mut self) -> Result<()> {
-let Some(filter) = &self.filter else {
-return Ok(());
-};
let Some(thresholds) = self.heap.get_threshold_values(&self.expr)?
else {
return Ok(());
};
+// Are the new thresholds more selective than our existing ones?
+let should_update = {
+let mut current = self.filter.thresholds.write();
Review Comment:
Perhaps this lock takes too long + overhead of doing this for all updates +
all partitions quite often?
We could also store a `Row` instead of `thresholds` to make comparison much
quicker (should also be able to avoid allocations)?
--
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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
Dandandan commented on code in PR #16433:
URL: https://github.com/apache/datafusion/pull/16433#discussion_r2156106525
##
datafusion/physical-plan/src/topk/mod.rs:
##
@@ -214,41 +238,39 @@ impl TopK {
let mut selected_rows = None;
-if let Some(filter) = self.filter.as_ref() {
-// If a filter is provided, update it with the new rows
-let filter = filter.current()?;
-let filtered = filter.evaluate(&batch)?;
-let num_rows = batch.num_rows();
-let array = filtered.into_array(num_rows)?;
-let mut filter = array.as_boolean().clone();
-let true_count = filter.true_count();
-if true_count == 0 {
-// nothing to filter, so no need to update
-return Ok(());
+// If a filter is provided, update it with the new rows
Review Comment:
`if let Some(filter)` condition is gone?
This provided the "unitialized filter" case before (that just runs the
normal topk without any filtering overhead).
--
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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
adriangb commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-2985967718 Seems like a bug in my implementation right? I'd be surprised if the update checks I added are that heavy compared to other work... -- 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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
Dandandan commented on PR #16433: URL: https://github.com/apache/datafusion/pull/16433#issuecomment-2985955904 It seems in some cases it's faster: ``` ┏━━┳━┳━━┳━━━┓ ┃ Query┃ topk-dynamic-filter ┃ topk-filters ┃Change ┃ ┡━━╇━╇━━╇━━━┩ │ Q1 │16.17 ms │ 16.89 ms │ no change │ │ Q2 │27.60 ms │ 20.22 ms │ +1.36x faster │ │ Q3 │55.41 ms │ 54.45 ms │ no change │ │ Q4 │22.26 ms │ 22.07 ms │ no change │ │ Q5 │11.98 ms │ 12.10 ms │ no change │ │ Q6 │26.25 ms │ 27.26 ms │ no change │ │ Q7 │71.56 ms │ 67.86 ms │ +1.05x faster │ │ Q8 │90.02 ms │ 46.15 ms │ +1.95x faster │ │ Q9 │61.64 ms │ 58.91 ms │ no change │ │ Q10 │ 101.13 ms │ 98.24 ms │ no change │ │ Q11 │58.43 ms │ 56.85 ms │ no change │ └──┴─┴──┴───┘ ``` And in other cases it's mixed / slower: ```┏━━┳━┳━━┳━━━┓ ┃ Query┃ topk-dynamic-filter ┃ topk-filters ┃Change ┃ ┡━━╇━╇━━╇━━━┩ │ QQuery 0 │14.00 ms │ 13.07 ms │ +1.07x faster │ │ QQuery 1 │20.55 ms │ 18.81 ms │ +1.09x faster │ │ QQuery 2 │56.58 ms │ 56.33 ms │ no change │ │ QQuery 3 │56.30 ms │ 54.94 ms │ no change │ │ QQuery 4 │ 468.00 ms │516.71 ms │ 1.10x slower │ │ QQuery 5 │ 626.85 ms │659.29 ms │ 1.05x slower │ │ QQuery 6 │15.31 ms │ 15.26 ms │ no change │ │ QQuery 7 │30.44 ms │ 21.48 ms │ +1.42x faster │ │ QQuery 8 │ 547.48 ms │569.67 ms │ no change │ │ QQuery 9 │ 744.50 ms │778.67 ms │ no change │ │ QQuery 10│ 182.53 ms │176.03 ms │ no change │ │ QQuery 11│ 217.10 ms │190.23 ms │ +1.14x faster │ │ QQuery 12│ 832.33 ms │715.38 ms │ +1.16x faster │ │ QQuery 13│ 1102.37 ms │914.96 ms │ +1.20x faster │ │ QQuery 14│ 823.97 ms │576.94 ms │ +1.43x faster │ │ QQuery 15│ 574.53 ms │496.51 ms │ +1.16x faster │ │ QQuery 16│ 1324.12 ms │ 1150.68 ms │ +1.15x faster │ │ QQuery 17│ 1324.29 ms │ 1245.55 ms │ +1.06x faster │ │ QQuery 18│ 2603.07 ms │ 2248.86 ms │ +1.16x faster │ │ QQuery 19│48.17 ms │ 49.60 ms │ no change │ │ QQuery 20│ 978.10 ms │964.90 ms │ no change │ │ QQuery 21│ 1021.47 ms │ 1050.51 ms │ no change │ │ QQuery 22│ 1958.07 ms │ 1660.81 ms │ +1.18x faster │ │ QQuery 23│ 760.66 ms │ 5353.24 ms │ 7.04x slower │ │ QQuery 24│ 525.78 ms │342.39 ms │ +1.54x faster │ │ QQuery 25│ 480.36 ms │278.59 ms │ +1.72x faster │ │ QQuery 26│ 542.13 ms │342.32 ms │ +1.58x faster │ │ QQuery 27│ 1826.05 ms │ 1260.73 ms │ +1.45x faster │ │ QQuery 28│ 11094.31 ms │ 10685.64 ms │ no change │ │ QQuery 29│ 427.18 ms │432.16 ms │ no change │ │ QQuery 30│ 826.43 ms │539.39 ms │ +1.53x faster │ │ QQuery 31│ 789.89 ms │553.49 ms │ +1.43x faster │ │ QQuery 32│ 2580.93 ms │ 2339.06 ms │ +1.10x faster │ │ QQuery 33│ 2660.40 ms │ 2761.14 ms │ no change │ │ QQuery 34│ 2948.08 ms │ 2923.44 ms │ no change │ │ QQuery 35│ 886.56 ms │810.35 ms │ +1.09x faster │ │ QQuery 36│18.59 ms │ 83.97 ms │ 4.52x slower │ │ QQuery 37│18.67 ms │ 35.52 ms │ 1.90x slower │ │ QQuery 38│18.10 ms │ 81.86 ms │ 4.52x slower │ │ QQuery 39│17.89 ms │135.04 ms │ 7.55x slower │ │ QQuery 40│18.30 ms │ 27.70 ms │ 1.51x slower │ │ QQuery 41│18.15 ms │ 26.07 ms │ 1.44x slower │ │ QQuery 42│17.78 ms │ 22.56 ms │ 1.27x slower │ └──┴─┴──┴───┘ ┏┳┓ ``` -- 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
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
Dandandan commented on code in PR #16433:
URL: https://github.com/apache/datafusion/pull/16433#discussion_r2154078467
##
datafusion/physical-plan/src/topk/mod.rs:
##
@@ -319,13 +341,88 @@ impl TopK {
/// (a > 2 OR (a = 2 AND b < 3))
/// ```
fn update_filter(&mut self) -> Result<()> {
-let Some(filter) = &self.filter else {
-return Ok(());
-};
let Some(thresholds) = self.heap.get_threshold_values(&self.expr)?
else {
return Ok(());
};
+// Are the new thresholds more selective than our existing ones?
+let should_update = {
+if let Some(current) = self.filter.thresholds.write().as_mut() {
+assert!(current.len() == thresholds.len());
+// Check if new thresholds are more selective than current ones
+let mut more_selective = false;
+for ((current_value, new_value), sort_expr) in
+current.iter().zip(thresholds.iter()).zip(self.expr.iter())
+{
+// Handle null cases
+let (current_is_null, new_is_null) =
+(current_value.is_null(), new_value.is_null());
Review Comment:
Can't we use `ScalarValue::partial_cmp`?
--
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]
Re: [PR] Only update TopK dynamic filters if the new ones are more selective [datafusion]
adriangb commented on code in PR #16433:
URL: https://github.com/apache/datafusion/pull/16433#discussion_r2153281386
##
datafusion/physical-plan/src/sorts/sort.rs:
##
@@ -846,8 +846,10 @@ pub struct SortExec {
common_sort_prefix: Vec,
/// Cache holding plan properties like equivalences, output partitioning
etc.
cache: PlanProperties,
-/// Filter matching the state of the sort for dynamic filter pushdown
-filter: Option>,
+/// Filter matching the state of the sort for dynamic filter pushdown.
+/// If `fetch` is `Some`, this will also be set and a TopK operator may be
used.
+/// If `fetch` is `None`, this will be `None`.
+filter: Option,
Review Comment:
I feel like there's some further refactoring that could happen here, e.g.
split up SortExec, leaving for another day.
##
datafusion/physical-plan/src/sorts/sort.rs:
##
@@ -1158,7 +1153,10 @@ impl ExecutionPlan for SortExec {
context.session_config().batch_size(),
context.runtime_env(),
&self.metrics_set,
-self.filter.clone(),
+self.filter
+.as_ref()
+.expect("Filter should be set when fetch is Some")
+.clone(),
Review Comment:
I refactored so that the `TopK` struct always expects this parameter which
better reflects the reality of execution. But since it's strangely tied to the
`fetch` param I am doing an `expect` assertion here. It should never fail at
runtime.
##
datafusion/physical-plan/src/sorts/sort.rs:
##
@@ -935,11 +940,6 @@ impl SortExec {
}
}
-pub fn with_filter(mut self, filter: Arc) ->
Self {
-self.filter = Some(filter);
-self
-}
Review Comment:
This was unused and had slipped through the cracks. I can make a new PR to
just remove these methods.
##
datafusion/physical-plan/src/topk/mod.rs:
##
@@ -319,13 +341,88 @@ impl TopK {
/// (a > 2 OR (a = 2 AND b < 3))
/// ```
fn update_filter(&mut self) -> Result<()> {
-let Some(filter) = &self.filter else {
-return Ok(());
-};
let Some(thresholds) = self.heap.get_threshold_values(&self.expr)?
else {
return Ok(());
};
+// Are the new thresholds more selective than our existing ones?
+let should_update = {
+if let Some(current) = self.filter.thresholds.write().as_mut() {
+assert!(current.len() == thresholds.len());
+// Check if new thresholds are more selective than current ones
+let mut more_selective = false;
+for ((current_value, new_value), sort_expr) in
+current.iter().zip(thresholds.iter()).zip(self.expr.iter())
+{
+// Handle null cases
+let (current_is_null, new_is_null) =
+(current_value.is_null(), new_value.is_null());
+
+match (current_is_null, new_is_null) {
+(true, true) => {
+// Both null, continue checking next values
+}
+(true, false) => {
+// Current is null, new is not null
+// For nulls_first: null < non-null, so new value
is less selective
+// For nulls_last: null > non-null, so new value
is more selective
+more_selective = !sort_expr.options.nulls_first;
+break;
+}
+(false, true) => {
+// Current is not null, new is null
+// For nulls_first: non-null > null, so new value
is more selective
+// For nulls_last: non-null < null, so new value
is less selective
+more_selective = sort_expr.options.nulls_first;
+break;
+}
+(false, false) => {
+// Neither is null, compare values
+match current_value.partial_cmp(new_value) {
+Some(ordering) => {
+match ordering {
+Ordering::Equal => {
+// Continue checking next values
+}
+Ordering::Less => {
+// For descending sort: new >
current means more selective
+// For ascending sort: new >
current means less selective
+more_selective =
sort_expr.
