@alamb mentioned this sounds similar to Dynamic Filters, I assume this must be a known technique (or my analysis may be completely wrong 😆 ) but I don't know what it would be called.

There was a talk at CIDR this year that mentioned this:

Sponsor Talk 3: The Fine Art of Work Skipping
Stefan Mandl, Snowflake

It seems they wrote a blog about it too here: https://www.snowflake.com/en/engineering-blog/optimizing-top-k-aggregation-snowflake/

Nice to know I'm not totally off on the idea 😄

Nice to know I'm not totally off on the idea 😄

Not at all!

BTW I am pretty sure DuckDB is using this technique and why they are so much faster on ClickBench Q23:

• Make ClickBench Q23 Go Faster #15177

Does anyone have a handle on how we might implement this? I was thinking we’d need to add a method to exec operators called apply_filter but that basically sends down the additional filter and by default it gets forwarded to children until it hits an exec that knows what to do with it (eg DataSourceExec). But I’m not very clear beyond that.

Does anyone have a handle on how we might implement this? I was thinking we’d need to add a method to exec operators called apply_filter but that basically sends down the additional filter and by default it gets forwarded to children until it hits an exec that knows what to do with it (eg DataSourceExec). But I’m not very clear beyond that.

To begin with I would suggest:

1. Make a new PhysicalExpr named something like TopKRuntimeFilter
2. Add a physical optimizer pass that runs after all other passes (so the structure doesn't change) that finds TopK nodes and tries to find connected Scans the (start with some basic rules, don't try and go past joins, etc)
3. Add TopKRuntimeFilter to those scans

Then the trick will be to figure out how to share the TopKHeap created in the TopK operator

With the TopKRuntimeFilter

And then orchestrate concurrent access to it somehow

@alamb I implemented something like that in #15301

Thanks @adriangb -- I will try and review it asap (hopefully tomorrow afternoon or tomorrow)

We already have Statistics on PartitionedFile so we could potentially use Dynamic filters to prune based on those before opening the file

@adriangb and I had a discussion about #15301

here are some notes:

Usecases:

• TopK dynamic filter pushdown
  • Prune files with dynamic filter based on statistics
  • Prune row groups with dynamic filter based on statistics
  • Prune row pages with dynamic filter based on statistics
  • Apply during filtering when pushdown enabled
• Join SIPs

Pros / Cons

The pros for merging this PR are:

• We already have benchmarks that show some performance improvement
  The cons:
• It requires special implementation for any operators (like FileOpenenr) to take advantage of such filters. THis is not a blocker in my mind – but I do think implementing a PhysicalExpr is a cleaner design. As Adrian says, we can refactor it over time if/when PhysicalExpr gets more sophisticated
• We will get even more performance when filter_pushdown is enabled (again maybe this is just follow on work)

Nice to haves

• For a plan with multiple partitions (e.g. for 16 input partitions, we end up with 17 top heaps – one for each partition and then a global one), but this PR can only apply the per-partition top k value.
• It would be nice to somehow be able to use all the top values (aka pick the smallest one) when filtering.
• This PR takes a snapshot of the contents of the TopK heap when a file is opened and never changes it.
  • This is good for pruning as all the pruning (file, row group and page) happens on file opening
  • It is not as good for filter_pushdown when the values in the topK heap can change over the course of the query so using the snapshot means the dynamic filter doesn’t improve over time

I believe adrian is going to look into these – but I also think they could easily be done as a follow on PR

wrt waiting for filter pushdown to be enabled by default, I think we're just making our lives harder by coupling them, especially since we can already test them together under a feature flag.

I also would like to leverage this work to justify a lot of other optimizations:

• piping through file level stats and using those to prune with dynamic filters -> prune without even reading parquet metadata
• roughly ordering files by the sort order based on stats -> this optimization becomes even more efficient
• resolve merge conflicts and take next steps with per-file filters

This is unfortunately a blocker for all of that

I plan to spend a non trivial amount of time working on this with @adriangb this week