[EPIC] Support limit pruning

[xudong963]

Let's see an example:

SELECT * FROM t1 WHERE ticker = 'ABT' AND window_start <= (1761364799999999999 AT TIME ZONE 'America/New_York') AND window_start >= (1698465600000000000 AT TIME ZONE 'America/New_York')  LIMIT 1

After the kinds of pruning inside DF, we can get some matched row groups.

Assume that there are still four row groups remaining after row group pruning by ticker = 'ABT' AND window_start <= (1761364799999999999 AT TIME ZONE 'America/New_York') AND window_start >= (1698465600000000000 AT TIME ZONE 'America/New_York') and related statistics in row group metadata. And one (row group 3) of the four is fully matched with the predicates, and the others are not, which are only partially matched.

The worst case is that we need to scan rg1, rg2, then find a matched row in rg3. But suppose we know the rg3 is fully matched (currently, DF only supports partial-matched). In that case, we can only check it if the total row count across all the rg3 meets or exceeds the threshold defined by k in the LIMIT clause, then we definitely can reduce the scanned data and skip other rows.

If there are no fully matched rg, we can extend limit pruning to the page level, to find the fully matched pages.

---

I plan to support row group level limit pruning first, then explore the page level

[alamb]

FWIW if the limit is pushed into the parquet reader, it will internally skip reading future row groups once the limit is reached. Here is some of the relevant code

https://github.com/apache/arrow-rs/blob/ed9efe78e4cc958cc96707557818e754419debb0/parquet/src/arrow/arrow_reader/read_plan.rs#L254-L294

So in other words, unless you are able to skip additional files / IO than otherwise I suspect applying a limit in DataFusion to skip row groups might not improve performance much

And one (row group 3) of the four is fully matched with the predicates, and the others are not, which are only partially matched.

What does "fully matches" /"partially matched" mean in this case? Does that mean all the rows in the row groups would be filtered?

[xudong963]

FWIW if the limit is pushed into the parquet reader, it will internally skip reading future row groups once the limit is reached. Here is some of the relevant code

If there's a filter, I think we still need to do row group pruning, then for the matched row groups, do row filters and get the limit rows.

What does "fully matches" /"partially matched" mean in this case? Does that mean all the rows in the row groups would be filtered?

For a row group, currently, we define it in row group pruning as pruned or matched. Here, I mean the matched row group can be further subdivided into partially matched and fully matched.

Then we leverage the fully matched row groups to return the limit k, we can reduce the cost of fetching partially row groups and doing row filter for them.

[alamb]

FWIW if the limit is pushed into the parquet reader, it will internally skip reading future row groups once the limit is reached. Here is some of the relevant code

If there's a filter, I think we still need to do row group pruning, then for the matched row groups, do row filters and get the limit rows.

If there is a filter applied in the scan (via pushdown_filters), the parquet reader will stop (and not fetch any more row groups) once the limit is hit.

https://github.com/apache/arrow-rs/blob/ed9efe78e4cc958cc96707557818e754419debb0/parquet/src/arrow/push_decoder/reader_builder/mod.rs#L504-L518

I am probably not understanding what you are proposing. I'll try and read the PR

What does "fully matches" /"partially matched" mean in this case? Does that mean all the rows in the row groups would be filtered?

So fully matches means there are no rows that are filtered out -- make sensee

[xudong963]

the parquet reader will stop (and not fetch any more row groups) once the limit is hit.

There is a picture from snowflake, we can set the partition in the pic as row group.

The filter is WHERE species LIKE'Alpine%' AND s >= 50

rg1 -> pruned, rg2, rg4 -> partially matched, rg3 fully matched.

We need to read rg2, rg3, then get the satisfied rows from rg3. But if we have limit pruning and a fully matched rg definition, we can directly fetch limit rows from rg3.

[alamb]

We need to read rg2, rg3, then get the satisfied rows from rg3. But if we have limit pruning and a fully matched rg definition, we can directly fetch limit rows from rg3.

Ah, I get it now -- that is quite interesting . Thank you @xudong963 that is a very clever idea

[2010YOUY01]

the parquet reader will stop (and not fetch any more row groups) once the limit is hit.

There is a picture from snowflake, we can set the partition in the pic as row group.

The filter is WHERE species LIKE'Alpine%' AND s >= 50

rg1 -> pruned, rg2, rg4 -> partially matched, rg3 fully matched.

We need to read rg2, rg3, then get the satisfied rows from rg3. But if we have limit pruning and a fully matched rg definition, we can directly fetch limit rows from rg3.

I also want to double check I get it correctly, did you mean:

Now we have a query

SELECT * 
FROM t 
WHERE species LIKE'Alpine%' AND s >= 50
LIMIT 3

And without this limit pruning feature, the plan might look like

-- Executing with `target_partition=3`, and we already got partition 1 skipped because it's stat shows 'not matched'
DataSourceExec (partition 2)
DataSourceExec (partition 3)
DataSourceExec (partition 4)

From the stat, we can ensure partition 3 will output 3 rows satisfying the LIMIT 3, so we can safely skip reading partition 2 and 4.

[xudong963]

@2010YOUY01 The limit pruning is at the row group/page level. We can see the partition in the above pic as a row group or a page in Parquet.

Without this limit pruning feature, we need to scan the partition2, partition 4 isn't needed due to the current early stop mentioned by @alamb .

After the feature, the partition2 can be skipped, what we need to do is give the partition3(rg3) a RowSelector(limit) then return the expected results,