alamb commented on code in PR #27110:
URL: https://github.com/apache/datafusion/pull/27110#discussion_r4482184368
##########
datafusion/core/tests/parquet/io/bloom_filter_io.rs:
##########
@@ -19,34 +37,78 @@ use super::ParquetIoTest;
#[tokio::test]
async fn test_parquet_bloom_filter_io_matching() {
let test = bloom_filter_test().await;
+ let traces = test.capture_traces();
// Loads the Bloom filters, finds a match, then reads the data pages and
// produces a batch.
- assert_snapshot!(test.query("SELECT * FROM t WHERE a = 10 AND b =
100").await, @r"
- --- Query: SELECT * FROM t WHERE a = 10 AND b = 100 (1 rows) ---
- Read Parquet footer (offsets 0..678)
- Read OffsetIndex for column 'a' (rg 0) (offsets 192..213)
- Read Bloom filter for column 'a' (rg 0) (offsets 98..145)
- Read Bloom filter for column 'b' (rg 0) (offsets 145..192)
- Read data page for column 'a' (rg 0) (offsets 4..51)
- Read data page for column 'b' (rg 0) (offsets 51..98)
- Produce batch (1 rows)
+ test.query("SELECT * FROM t WHERE a = 10 AND b = 100")
+ .await
+ .unwrap();
+ assert_snapshot!(traces.render(), @"
+ Query (sql=SELECT * FROM t WHERE a = 10 AND b = 100, rows=1)
+ ObjectStore::head (location=table.parquet)
+ Metadata cache hit (has_page_index=false)
+ Load page index into cached metadata
+ Read Parquet metadata (range=192..213)
+ ObjectStore::get (location=table.parquet)
+ Read byte range (range=192..213, parquet=OffsetIndex for
column 'a' (rg 0))
Review Comment:
A nit is I would probably combine this into one line
```
ObjectStore::get (location=table.parquet)
Read byte range (range=192..213, parquet=OffsetIndex for
column 'a' (rg 0))
```
into
```
ObjectStore::get (location=table.parquet) (range=192..213,
parquet=OffsetIndex for column 'a' (rg 0))
```
##########
datafusion/core/Cargo.toml:
##########
@@ -167,6 +167,8 @@ bytes = { workspace = true }
env_logger = { workspace = true }
glob = { workspace = true }
insta = { workspace = true }
+tracing = { workspace = true }
Review Comment:
this is my biggest hesitation -- to add this new tracing dependency (I know
it is just a dev-dependency, but still I feel like it would be easy to
accidentally turn into a real dependnecy, and not everyone uses the tracing
crate
##########
datafusion/core/tests/parquet/io/filter_pushdown_io.rs:
##########
@@ -0,0 +1,142 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: See the NOTICE file distributed with this work for
additional information regarding copyright ownership
+// SPDX-FileContributor: Licensed to the Apache Software Foundation (ASF)
under one or more contributor license agreements
+
+//! Tests tracking I/O events with and without Parquet filter pushdown
+//! (`datafusion.execution.parquet.pushdown_filters`, see
+//! [`ParquetOptions::pushdown_filters`]).
+//!
+//! Both tests run `val = 7`, which statistics cannot prune, but which only
+//! matches a row in row group 1:
+//!
+//! ```text
+//! row group 0 row group 1 row group 2
+//! ┌──────┬───────┐ ┌──────┬───────┐ ┌──────┬───────┐
+//! │ id │ val │ │ id │ val │ │ id │ val │
+//! ├──────┼───────┤ ├──────┼───────┤ ├──────┼───────┤
+//! │ 1 │ 0 │ │ 4 │ 0 │ │ 7 │ 0 │
+//! │ 2 │ 5 │ │ 5 │ 7 * │ │ 8 │ 5 │
+//! │ 3 │ 10 │ │ 6 │ 10 │ │ 9 │ 10 │
+//! └──────┴───────┘ └──────┴───────┘ └──────┴───────┘
+//! val: min 0 max 10 val: min 0 max 10 val: min 0 max 10
+//!
+//! * the only row matching `val = 7`
+//! ```
+//!
+//! [`ParquetOptions::pushdown_filters`]:
datafusion::config::ParquetOptions::pushdown_filters
+
+use std::sync::Arc;
+
+use arrow::array::{ArrayRef, Int32Array, RecordBatch};
+use bytes::Bytes;
+use datafusion::prelude::SessionConfig;
+use insta::assert_snapshot;
+use parquet::arrow::ArrowWriter;
+use parquet::file::properties::WriterProperties;
+
+use super::ParquetIoTest;
+
+/// Without filter pushdown, the scan fetches every column of every row group
+#[tokio::test]
+async fn test_filter_pushdown_io_disabled() {
+ let test = filter_pushdown_test(false).await;
+ let traces = test.capture_traces();
+ test.query("SELECT * FROM t WHERE val = 7").await.unwrap();
+ assert_snapshot!(traces.render(), @"
+ Query (sql=SELECT * FROM t WHERE val = 7, rows=1)
+ ObjectStore::head (location=table.parquet)
+ Metadata cache hit (has_page_index=true)
+ Open Parquet file (file=table.parquet, size=1402)
+ Load Parquet metadata
+ Metadata cache hit (has_page_index=true)
+ Prune row groups
+ Row groups remaining (row_groups=3, row_groups_remaining=3,
pruning_predicate=true)
+ Load page index
+ Load Bloom filters
+ Prune row groups with Bloom filters
+ Row groups remaining (row_groups_remaining=3)
+ Prune pages with page index
+ Row groups remaining (row_groups_remaining=3)
+ Fetch column chunk data (ranges=2)
+ ObjectStore::get_ranges (location=table.parquet, ranges=2)
+ Read byte range (range=4..51, parquet=dictionary page and data
page 0 for column 'id' (rg 0))
+ Read byte range (range=51..98, parquet=dictionary page and
data page 0 for column 'val' (rg 0))
+ Fetch column chunk data (ranges=2)
+ ObjectStore::get_ranges (location=table.parquet, ranges=2)
+ Read byte range (range=98..145, parquet=dictionary page and
data page 0 for column 'id' (rg 1))
+ Read byte range (range=145..192, parquet=dictionary page and
data page 0 for column 'val' (rg 1))
+ Fetch column chunk data (ranges=2)
+ ObjectStore::get_ranges (location=table.parquet, ranges=2)
+ Read byte range (range=192..239, parquet=dictionary page and
data page 0 for column 'id' (rg 2))
+ Read byte range (range=239..286, parquet=dictionary page and
data page 0 for column 'val' (rg 2))
+ Produce batch (rows=1)
+ ");
+}
+
+/// **With** filter pushdown, the scan first fetches and evaluates the filter
+/// column (`val`) for each row group, and then fetches the remaining columns
+/// (`id`) only for the row groups with matching rows (row group 1).
+#[tokio::test]
+async fn test_filter_pushdown_io_enabled() {
+ let test = filter_pushdown_test(true).await;
+ let traces = test.capture_traces();
+ test.query("SELECT * FROM t WHERE val = 7").await.unwrap();
+ assert_snapshot!(traces.render(), @"
+ Query (sql=SELECT * FROM t WHERE val = 7, rows=1)
+ ObjectStore::head (location=table.parquet)
+ Metadata cache hit (has_page_index=true)
+ Open Parquet file (file=table.parquet, size=1402)
+ Load Parquet metadata
+ Metadata cache hit (has_page_index=true)
Review Comment:
this is cool to show the metadata cache hit
##########
datafusion/datasource-parquet/src/opener/mod.rs:
##########
@@ -636,7 +637,12 @@ impl ParquetOpenState {
let Some(prepared) = (*prepared).prune_file()? else {
return Ok(ParquetOpenState::Done);
};
- Ok(ParquetOpenState::LoadMetadata(prepared.load().boxed()))
+ Ok(ParquetOpenState::LoadMetadata(
+ prepared
+ .load()
+ .instrument(tracing::debug_span!("Load Parquet
metadata"))
Review Comment:
I really like the idea of instrumenting the parquet opener, but I don't like
tying the instrumentation to the tracing crate
##########
datafusion/datasource-parquet/Cargo.toml:
##########
@@ -43,6 +43,7 @@ object_store = { workspace = true }
parking_lot = { workspace = true }
parquet = { workspace = true }
tokio = { workspace = true }
+tracing = { workspace = true }
Review Comment:
I think this is a bad idea to pick up the tracing dependency
--
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]
