support both filter + limit in ScanBuilder

[haohuaijin]

recently i do a benchmark for parquet vs vortex, i find the below case parquet is 2x faster than vortex:

the file is already order by time desc, the sql like below

select * from table where xxx order by time desc limit 10

datafusion will generate the plan like below

SortPreservingMergeExec limit=10
    DataSourceExec limit=10

but current vortex do not support filter + limit together, result in vortex read more records than parquet.

vortex/vortex-layout/src/scan/scan_builder.rs

Lines 243 to 245 in dcd7097

	if self.filter.is_some() && self.limit.is_some() {
	vortex_bail!("Vortex doesn't support scans with both a filter and a limit")
	}

vortex/vortex-datafusion/src/persistent/opener.rs

Lines 355 to 359 in dcd7097

	if let Some(limit) = limit
	&& filter.is_none()
	{
	scan_builder = scan_builder.with_limit(limit);
	}

[haohuaijin]

another context, if i enable pushdown_filter for parquet format, the above query will 5x slower than disable pushdown_filter in parquet format, see apache/arrow-rs#9765

[AdamGS]

This is definitely a missing piece, let me see if I can dig something up, been on the back of my head for a long time now. Do you have a sense of how many rows more Vortex returns to the upstream DataFusion ExecutionNode? It should also just polling once it has enough rows.

[haohuaijin]

weird, only 138 rows, my target partition is 16, and i have a lot of data that match the where condition.
look like my assumption was wrong🤔

metrics=[output_rows=138, elapsed_compute=16ns, output_bytes=139.5 MB, output_batches=28, batches_split=0, file_open_errors=0, file_scan_errors=0, time_elapsed_opening=30.80ms, time_elapsed_processing=85.88ms, time_elapsed_scanning_total=1.86s, time_elapsed_scanning_until_data=1.79s]

[haohuaijin]

this is the metrics from parquet(pushdown filter disable)

metrics=[output_rows=229.4 K, elapsed_compute=16ns, output_bytes=230.3 MB, output_batches=28, files_ranges_pruned_statistics=45 total → 24 matched, row_groups_pruned_statistics=47 total → 40 matched, row_groups_pruned_bloom_filter=40 total → 40 matched, page_index_rows_pruned=34.14 M total → 27.70 M matched, batches_split=0, bytes_scanned=482.7 M, file_open_errors=0, file_scan_errors=0, num_predicate_creation_errors=0, predicate_cache_inner_records=0, predicate_cache_records=0, predicate_evaluation_errors=0, pushdown_rows_matched=0, pushdown_rows_pruned=0, bloom_filter_eval_time=1.61ms, metadata_load_time=248.72ms, page_index_eval_time=2.00ms, row_pushdown_eval_time=90ns, statistics_eval_time=1.11ms, time_elapsed_opening=99.31ms, time_elapsed_processing=501.75ms, time_elapsed_scanning_total=651.19ms, time_elapsed_scanning_until_data=584.90ms, scan_efficiency_ratio=840% (482.7 M/57.28 M)]

parquet(pushdown filter enable)

metrics=[output_rows=150, elapsed_compute=16ns, output_bytes=128.1 MB, output_batches=15, files_ranges_pruned_statistics=45 total → 24 matched, row_groups_pruned_statistics=47 total → 40 matched, row_groups_pruned_bloom_filter=40 total → 40 matched, page_index_rows_pruned=34.14 M total → 27.70 M matched, batches_split=0, bytes_scanned=285.3 M, file_open_errors=0, file_scan_errors=0, num_predicate_creation_errors=0, predicate_cache_inner_records=33.12 M, predicate_cache_records=7.45 M, predicate_evaluation_errors=0, pushdown_rows_matched=28.48 K, pushdown_rows_pruned=11.42 M, bloom_filter_eval_time=508.94µs, metadata_load_time=330.02ms, page_index_eval_time=1.50ms, row_pushdown_eval_time=215.08ms, statistics_eval_time=528.53µs, time_elapsed_opening=180.87ms, time_elapsed_processing=3.59s, time_elapsed_scanning_total=3.94s, time_elapsed_scanning_until_data=3.94s, scan_efficiency_ratio=430% (285.3 M/67.04 M)]

[AdamGS]

Vortex still potentially plans a lot of work in that case, which might be the source of the higher time_elapsed_scanning_total and time_elapsed_scanning_until_data, I'll take a deeper look into this. Any change this is data you can share? Might make it easier to reproduce.

[haohuaijin]

the original data i not able to share to you, but i will try generate some data that can reproduce the result.

[haohuaijin]

i shared the files in vortex slack to you

[AdamGS]

Also just to make sure I understand the relation ship performance, is it:

1. Vortex is slowest
2. Default parquet is 50% (2x faster)
3. Parquet with pushdown is 20% (5x faster)

[haohuaijin]

1. parquet with pushdown is slowest(datafusion 53)
2. correct
3. Parquet without pushdown is 200% (2x slower) (datafusion 53)