Enable Parquet Row and Page Filtering by default (WIP)

[alamb]

Draft until

• Expose parquet reader settings using normal DataFusion ConfigOptions #3822 is merged
• We have completed testing / validation

Which issue does this PR close?

Closes #3463
closes #4085
re #3462

Rationale for this change

This PR turns on parquet scan predicate pushdown (see #3462) by default -- I am putting it up early as part of the testing process (so we can work through any issues it may uncover)

This feature promises to be one of the most significant performance improvements for DataFusion reading from parquet in a while. All the hard work was done by @Ted-Jiang @thinkharderdev and @tustvold

What changes are included in this PR?

Enable pushing filters into the scan directly

Note this feature can be disabled by setting the datafusion.execution.parquet.pushdown_filters configuration setting to false.

Are there any user-facing changes?

Hopefully faster performance

[alamb]

A small update here is that when I ran the tpch benchmarks against the default parquet files created by the benchmark I did not see any improvement. Also, there was some sort of error with the page index code which I need to track down

[alamb]

Specifically made the parquet files like this:

RUSTFLAGS="-C target-cpu=native" cargo run --release --bin tpch -- convert --input ~/tpch_data/data_SF1 --output ~/tpch_data/parquet_data_SF1 --format=parquet

And then ran

RUSTFLAGS="-C target-cpu=native" cargo run --release --bin tpch -- benchmark datafusion --iterations 3 --path ~/tpch_data/parquet_data_SF1 --format parquet --batch-size 4096          

    Finished release [optimized] target(s) in 0.28s
     Running `target/release/tpch benchmark datafusion --iterations 3 --path /home/alamb/tpch_data/parquet_data_SF1 --format parquet --batch-size 4096`
Running benchmarks with the following options: DataFusionBenchmarkOpt { query: None, debug: false, iterations: 3, partitions: 2, batch_size: 4096, path: "/home/alamb/tpch_data/parquet_data_SF1", file_format: "parquet", mem_table: false, output_path: None, disable_statistics: false, enable_scheduler: false }
Query 1 iteration 0 took 1511.2 ms and returned 4 rows
Query 1 iteration 1 took 1372.2 ms and returned 4 rows
Query 1 iteration 2 took 1419.7 ms and returned 4 rows
Query 1 avg time: 1434.38 ms
thread 'tokio-runtime-worker' panicked at 'called `Option::unwrap()` on a `None` value', datafusion/core/src/physical_plan/file_format/parquet/page_filter.rs:129:27
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace
Error: ArrowError(ExternalError(ArrowError(ExternalError("Arrow error: External error: Execution error: Arrow error: External error: Arrow error: External error: Execution error: Arrow error: External error: Execution error: Join Error: task 218 panicked"))))
alamb@aal-dev:~/arrow-datafusion$ 

FYI @Ted-Jiang -- haven't had a chance to file this as a ticket or look more carefully into it

[Ted-Jiang]

Specifically made the parquet files like this:

RUSTFLAGS="-C target-cpu=native" cargo run --release --bin tpch -- convert --input ~/tpch_data/data_SF1 --output ~/tpch_data/parquet_data_SF1 --format=parquet

And then ran

RUSTFLAGS="-C target-cpu=native" cargo run --release --bin tpch -- benchmark datafusion --iterations 3 --path ~/tpch_data/parquet_data_SF1 --format parquet --batch-size 4096          

    Finished release [optimized] target(s) in 0.28s
     Running `target/release/tpch benchmark datafusion --iterations 3 --path /home/alamb/tpch_data/parquet_data_SF1 --format parquet --batch-size 4096`
Running benchmarks with the following options: DataFusionBenchmarkOpt { query: None, debug: false, iterations: 3, partitions: 2, batch_size: 4096, path: "/home/alamb/tpch_data/parquet_data_SF1", file_format: "parquet", mem_table: false, output_path: None, disable_statistics: false, enable_scheduler: false }
Query 1 iteration 0 took 1511.2 ms and returned 4 rows
Query 1 iteration 1 took 1372.2 ms and returned 4 rows
Query 1 iteration 2 took 1419.7 ms and returned 4 rows
Query 1 avg time: 1434.38 ms
thread 'tokio-runtime-worker' panicked at 'called `Option::unwrap()` on a `None` value', datafusion/core/src/physical_plan/file_format/parquet/page_filter.rs:129:27
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace
Error: ArrowError(ExternalError(ArrowError(ExternalError("Arrow error: External error: Execution error: Arrow error: External error: Arrow error: External error: Execution error: Arrow error: External error: Execution error: Join Error: task 218 panicked"))))
alamb@aal-dev:~/arrow-datafusion$ 

FYI @Ted-Jiang -- haven't had a chance to file this as a ticket or look more carefully into it

Thanks for testing this, i will try to figure it out tomorrow.

[Ted-Jiang]

@alamb i think it fixed by #4387
run

(venv) yangjiang@LM-SHC-15009782 benchmarks % OPT_PARQUET_ENABLE_PAGE_INDEX=true  cargo run --release --bin tpch -- benchmark datafusion --iterations 3 --path ~/tpch-parquet  --format parquet --batch-size 4096                 

without error