From fb90cdc5c10ee7f6c3b61cf09f0543c7f64db413 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 20 Aug 2026 13:46:53 +0000 Subject: [PATCH 1/6] feat(parquet): add `bytes_processed` scan-completion metric MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `EXPLAIN ANALYZE` reports how many row groups a scan pruned, but nothing reports how much of a scan is done. `bytes_scanned` looks like it should serve — its natural denominator, the size of the files in the plan, is known up front — but it counts only the bytes fetched, so pruning and projection pushdown leave it understating progress by a factor that varies per query. `bytes_processed` completes that numerator: it counts the bytes a scan is finished with, whether it read them or proved it did not need them. Over the lifetime of a file it advances by exactly that file's size — or, for a file split into byte ranges for parallelism, by the size of the range — so `bytes_processed / total file bytes` is a completion fraction a progress reporter can use directly. Credit lands a row group at a time. Row groups pruned while opening the file (by range, statistics, bloom filter, page index or limit) are credited before the first batch is decoded, which is the progress a scan makes that `files_processed` cannot show; row groups dropped mid-scan by a dynamic filter are credited as they are dropped; the rest are credited as the decoder reaches them. Whatever is left over is credited when the file closes, so a scan cut short by a `LIMIT`, an early stop or an error still ends on exactly the size of its range. Costs one atomic add per row group. Crediting a row group's bytes progressively as its rows decode is left to a follow-up. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_017cmj1SUoV7Y9ADPbzLrzZ3 --- datafusion/core/tests/sql/explain_analyze.rs | 1 + datafusion/datasource-parquet/src/metrics.rs | 70 +++++ .../datasource-parquet/src/opener/mod.rs | 291 +++++++++++++++++- .../datasource-parquet/src/push_decoder.rs | 81 ++++- .../src/row_group_filter.rs | 27 +- .../dynamic_filter_pushdown_config.slt | 2 +- .../test_files/dynamic_row_group_pruning.slt | 2 +- .../test_files/explain_analyze.slt | 8 +- .../sqllogictest/test_files/limit_pruning.slt | 4 +- docs/source/user-guide/explain-usage.md | 1 + 10 files changed, 451 insertions(+), 36 deletions(-) diff --git a/datafusion/core/tests/sql/explain_analyze.rs b/datafusion/core/tests/sql/explain_analyze.rs index 4c8b8f9c01122..f8498cbedf311 100644 --- a/datafusion/core/tests/sql/explain_analyze.rs +++ b/datafusion/core/tests/sql/explain_analyze.rs @@ -889,6 +889,7 @@ async fn parquet_explain_analyze() { ); assert_contains!(&formatted, "output_rows_skew=0%"); assert_contains!(&formatted, "scan_efficiency_ratio=13.99%"); + assert_contains!(&formatted, "bytes_processed="); // The order of metrics is expected to be the same as the actual pruning order // (file-> row-group -> page) diff --git a/datafusion/datasource-parquet/src/metrics.rs b/datafusion/datasource-parquet/src/metrics.rs index a3573c8624792..7da8d2c2968f4 100644 --- a/datafusion/datasource-parquet/src/metrics.rs +++ b/datafusion/datasource-parquet/src/metrics.rs @@ -63,6 +63,23 @@ pub struct ParquetFileMetrics { pub row_groups_pruned_dynamic_filter: Count, /// Total number of bytes scanned pub bytes_scanned: Count, + /// Total number of bytes the scan is finished with, whether they were read + /// or skipped. + /// + /// Where [`Self::bytes_scanned`] counts only the bytes fetched from the + /// object store, this counts every byte the scan has resolved: the bytes it + /// read, plus the bytes of the row groups (and whole files) that pruning + /// proved cannot contribute. Over the lifetime of a file it therefore sums + /// to that file's size — or, for a file split into byte ranges for + /// parallelism, to the size of the range — so + /// `bytes_processed / total file bytes` is a scan completion fraction, + /// which `bytes_scanned` on its own is not: it understates progress by + /// however much pruning and projection pushdown saved. + /// + /// Credited at row-group granularity: a row group's bytes land when the + /// scan is done with it. Crediting a row group's bytes progressively as its + /// rows are decoded is left to a follow-up. + pub bytes_processed: Count, /// Total rows filtered out by predicates pushed into parquet scan pub pushdown_rows_pruned: Count, /// Total rows passed predicates pushed into parquet scan @@ -103,6 +120,52 @@ pub struct ParquetFileMetrics { pub predicate_cache_records: Gauge, } +/// Tracks how much of one file — or one byte range of a file — a scan has +/// finished with, crediting [`ParquetFileMetrics::bytes_processed`] as it goes. +/// +/// Every credit is clamped to the bytes left in the budget, and whatever is +/// left over is credited on drop. The counter therefore advances by exactly the +/// size of the range being scanned however the scan ends: normally, at a +/// `LIMIT`, when a dynamic filter proves the rest of the file irrelevant, or on +/// an error. That total is what makes the metric usable as a completion +/// fraction rather than just another counter. +/// +/// The clamp and the final top-up also absorb two small inexactnesses in +/// crediting by row group: a file is slightly larger than the sum of its row +/// groups (the footer, the page index and any padding belong to no row group), +/// and a row group is assigned to a byte range by the offset of its first page, +/// so a range's row groups do not add up to precisely its length. +#[derive(Debug)] +pub(crate) struct ByteProgress { + /// Bytes of the scanned range not yet credited. + remaining: u64, + bytes_processed: Count, +} + +impl ByteProgress { + /// Start tracking a range of `total` bytes. + pub(crate) fn new(total: u64, bytes_processed: Count) -> Self { + Self { + remaining: total, + bytes_processed, + } + } + + /// Record that the scan is finished with `bytes` more of the range. + pub(crate) fn credit(&mut self, bytes: u64) { + let bytes = bytes.min(self.remaining); + self.remaining -= bytes; + self.bytes_processed + .add(usize::try_from(bytes).unwrap_or(usize::MAX)); + } +} + +impl Drop for ByteProgress { + fn drop(&mut self) { + self.credit(self.remaining); + } +} + impl ParquetFileMetrics { /// Create new metrics pub fn new( @@ -144,6 +207,12 @@ impl ParquetFileMetrics { .with_category(MetricCategory::Bytes) .counter("bytes_scanned", partition); + let bytes_processed = builder + .clone() + .with_type(MetricType::Summary) + .with_category(MetricCategory::Bytes) + .counter("bytes_processed", partition); + let metadata_load_time = builder .clone() .with_type(MetricType::Summary) @@ -219,6 +288,7 @@ impl ParquetFileMetrics { row_groups_pruned_statistics, row_groups_pruned_dynamic_filter, bytes_scanned, + bytes_processed, pushdown_rows_pruned, pushdown_rows_matched, row_pushdown_eval_time, diff --git a/datafusion/datasource-parquet/src/opener/mod.rs b/datafusion/datasource-parquet/src/opener/mod.rs index b3ce024d66f1f..9c79f57676636 100644 --- a/datafusion/datasource-parquet/src/opener/mod.rs +++ b/datafusion/datasource-parquet/src/opener/mod.rs @@ -25,12 +25,13 @@ use self::early_stop::EarlyStoppingStream; use self::encryption::EncryptionContext; use crate::access_plan::PreparedAccessPlan; use crate::decoder_projection::DecoderProjection; +use crate::metrics::ByteProgress; use crate::page_filter::PagePruningAccessPlanFilter; use crate::push_decoder::{ DecoderBuilderConfig, PushDecoderStreamState, RgPlanEntry, RowGroupPruner, }; use crate::row_filter::RowFilterGenerator; -use crate::row_group_filter::RowGroupAccessPlanFilter; +use crate::row_group_filter::{RowGroupAccessPlanFilter, row_group_in_range}; use crate::{ BloomFilterStatistics, Int96Coercer, ParquetAccessPlan, ParquetFileMetrics, ParquetFileReaderFactory, ParquetRowSelection, ParquetVirtualColumn, @@ -890,6 +891,12 @@ impl PreparedParquetOpen { self.file_metrics .files_ranges_pruned_statistics .add_pruned(1); + // The scan is done with every byte of this file range without + // reading any of them. + self.file_metrics.bytes_processed.add( + usize::try_from(self.partitioned_file.effective_size()) + .unwrap_or(usize::MAX), + ); return Ok(None); } @@ -1480,7 +1487,10 @@ impl RowGroupsPrunedParquetOpen { .row_group_indexes .iter() .copied() - .map(|rg_index| RgPlanEntry { rg_index }) + .map(|rg_index| RgPlanEntry { + rg_index, + bytes: row_group_bytes(&rg_metadata[rg_index]), + }) .collect(); let mut builder = @@ -1497,6 +1507,34 @@ impl RowGroupsPrunedParquetOpen { (builder.build()?, rg_plan, has_row_selection) }; + // Track how much of this file range the scan has finished with. Credit + // up front every row group it will not read: those pruning removed, and + // — for a file split into ranges for parallelism — those belonging to + // another range. Without this the metric would sit at zero until the + // first row group finishes decoding, reporting no progress for a scan + // that may have just proved most of its work unnecessary. + let mut byte_progress = ByteProgress::new( + prepared.partitioned_file.effective_size(), + prepared.file_metrics.bytes_processed.clone(), + ); + let mut will_scan = vec![false; rg_metadata.len()]; + for entry in &rg_plan { + will_scan[entry.rg_index] = true; + } + let skipped_bytes: u64 = rg_metadata + .iter() + .enumerate() + .filter(|(rg_index, rg_meta)| { + !will_scan[*rg_index] + && prepared + .file_range + .as_ref() + .is_none_or(|range| row_group_in_range(rg_meta, range)) + }) + .map(|(_, rg_meta)| row_group_bytes(rg_meta)) + .sum(); + byte_progress.credit(skipped_bytes); + let predicate_cache_inner_records = prepared.file_metrics.predicate_cache_inner_records.clone(); let predicate_cache_records = @@ -1557,6 +1595,7 @@ impl RowGroupsPrunedParquetOpen { baseline_metrics: prepared.baseline_metrics, row_group_pruner, row_groups_pruned_dynamic, + byte_progress, } .into_stream(); @@ -1579,6 +1618,12 @@ impl RowGroupsPrunedParquetOpen { } } +/// The on-disk size of a row group, as credited to +/// [`ParquetFileMetrics::bytes_processed`]. +fn row_group_bytes(rg_meta: &RowGroupMetaData) -> u64 { + u64::try_from(rg_meta.compressed_size()).unwrap_or(0) +} + type ConstantColumns = HashMap; /// Extract constant column values from statistics, keyed by column name in the logical file schema. @@ -2406,6 +2451,248 @@ mod test { )) } + /// `bytes_processed` reports how much of a file range a scan has finished + /// with, so every one of these asserts the same invariant: by the time the + /// stream is dropped it has advanced by exactly the size of the range, + /// however the scan got there. + mod bytes_processed { + use super::*; + + /// Three batches of three rows, each forced into its own row group. + async fn write_three_row_groups(store: Arc) -> (SchemaRef, u64) { + let batches = vec![ + record_batch!(("a", Int32, vec![Some(1), Some(2), Some(3)])).unwrap(), + record_batch!(("a", Int32, vec![Some(4), Some(5), Some(6)])).unwrap(), + record_batch!(("a", Int32, vec![Some(7), Some(8), Some(9)])).unwrap(), + ]; + let schema = batches[0].schema(); + let props = WriterProperties::builder() + .set_max_row_group_row_count(Some(3)) + .build(); + let data_len = + write_parquet_batches(store, "test.parquet", batches, Some(props)).await; + (schema, u64::try_from(data_len).unwrap()) + } + + fn bytes_processed(metrics: &ExecutionPlanMetricsSet) -> u64 { + u64::try_from(counter_metric_value(metrics, "bytes_processed")).unwrap() + } + + #[tokio::test] + async fn scanning_a_whole_file_credits_all_of_it() { + let store = Arc::new(InMemory::new()) as Arc; + let (schema, data_len) = write_three_row_groups(Arc::clone(&store)).await; + let file = PartitionedFile::new("test.parquet".to_string(), data_len); + let metrics = ExecutionPlanMetricsSet::new(); + + let morselizer = ParquetMorselizerBuilder::new() + .with_store(Arc::clone(&store)) + .with_schema(Arc::clone(&schema)) + .with_projection_indices(&[0]) + .with_metrics(metrics.clone()) + .build(); + + let (_, rows) = + count_batches_and_rows(open_file(&morselizer, file).await.unwrap()).await; + + assert_eq!(rows, 9); + assert_eq!(bytes_processed(&metrics), data_len); + } + + /// Credit lands a row group at a time while the scan runs, rather than + /// all at once when the file closes — which is what makes this finer + /// grained than `files_processed`. + #[tokio::test] + async fn credit_advances_while_the_scan_runs() { + let store = Arc::new(InMemory::new()) as Arc; + let (schema, data_len) = write_three_row_groups(Arc::clone(&store)).await; + let file = PartitionedFile::new("test.parquet".to_string(), data_len); + let metrics = ExecutionPlanMetricsSet::new(); + + let morselizer = ParquetMorselizerBuilder::new() + .with_store(Arc::clone(&store)) + .with_schema(Arc::clone(&schema)) + .with_projection_indices(&[0]) + .with_metrics(metrics.clone()) + .build(); + + let mut stream = open_file(&morselizer, file).await.unwrap(); + assert_eq!( + bytes_processed(&metrics), + 0, + "nothing is pruned here, so nothing is credited before decoding", + ); + + let mut seen = 0; + let mut batches = 0; + while let Some(batch) = stream.next().await { + batch.unwrap(); + batches += 1; + let processed = bytes_processed(&metrics); + assert!( + processed > seen, + "batch {batches} must have advanced the credit past {seen}, \ + got {processed}", + ); + assert!( + processed < data_len, + "the whole file must not be credited while row groups remain", + ); + seen = processed; + } + + assert_eq!(batches, 3, "one batch per row group"); + drop(stream); + assert_eq!(bytes_processed(&metrics), data_len); + } + + /// The point of the metric: bytes pruning saved are credited when the + /// file is opened, not withheld until something is decoded. + #[tokio::test] + async fn row_group_pruning_is_credited_before_any_batch_is_read() { + let store = Arc::new(InMemory::new()) as Arc; + let (schema, data_len) = write_three_row_groups(Arc::clone(&store)).await; + let file = PartitionedFile::new("test.parquet".to_string(), data_len); + let metrics = ExecutionPlanMetricsSet::new(); + + // Only the first row group holds values below 4. + let predicate = logical2physical(&col("a").lt(lit(4)), &schema); + let morselizer = ParquetMorselizerBuilder::new() + .with_store(Arc::clone(&store)) + .with_schema(Arc::clone(&schema)) + .with_projection_indices(&[0]) + .with_predicate(predicate) + .with_row_group_stats_pruning(true) + .with_metrics(metrics.clone()) + .build(); + + let stream = open_file(&morselizer, file).await.unwrap(); + let pruned_at_open = bytes_processed(&metrics); + assert!( + pruned_at_open > 0, + "the two pruned row groups must be credited at open, before any \ + batch is decoded, since that is the progress `files_processed` misses", + ); + + let (_, rows) = count_batches_and_rows(stream).await; + assert_eq!(rows, 3); + assert_eq!(bytes_processed(&metrics), data_len); + } + + /// A file pruned by a dynamic filter never loads its footer, so its + /// whole size is credited from the plan-time file size. + #[tokio::test] + async fn a_file_pruned_before_open_credits_its_whole_size() { + let store = Arc::new(InMemory::new()) as Arc; + let (schema, data_len) = write_three_row_groups(Arc::clone(&store)).await; + let file = PartitionedFile::new("test.parquet".to_string(), data_len) + .with_statistics(Arc::new( + Statistics::new_unknown(&schema).add_column_statistics( + ColumnStatistics::new_unknown() + .with_min_value(Precision::Exact(ScalarValue::Int32(Some(1)))) + .with_max_value(Precision::Exact(ScalarValue::Int32(Some(9)))) + .with_null_count(Precision::Exact(0)), + ), + )); + let metrics = ExecutionPlanMetricsSet::new(); + + // No row in the file can match, so file-level pruning skips it + // without reading anything. + let predicate = + make_dynamic_expr(logical2physical(&col("a").gt(lit(100)), &schema)); + let morselizer = ParquetMorselizerBuilder::new() + .with_store(Arc::clone(&store)) + .with_schema(Arc::clone(&schema)) + .with_projection_indices(&[0]) + .with_predicate(predicate) + .with_metrics(metrics.clone()) + .build(); + + let (_, rows) = + count_batches_and_rows(open_file(&morselizer, file).await.unwrap()).await; + + assert_eq!(rows, 0); + assert_eq!( + counter_metric_value(&metrics, "bytes_scanned"), + 0, + "the file must have been pruned without being read", + ); + assert_eq!(bytes_processed(&metrics), data_len); + } + + /// A file split into byte ranges for parallelism: each range credits its + /// own size and no more, so the ranges add up to the file exactly rather + /// than each claiming all of it. + #[tokio::test] + async fn each_range_of_a_split_file_credits_only_its_own_bytes() { + let store = Arc::new(InMemory::new()) as Arc; + let (schema, data_len) = write_three_row_groups(Arc::clone(&store)).await; + + let split = i64::try_from(data_len).unwrap() / 2; + let ranges = [(0, split), (split, i64::try_from(data_len).unwrap())]; + + let mut total_processed = 0; + let mut total_rows = 0; + for (start, end) in ranges { + let file = PartitionedFile::new_with_range( + "test.parquet".to_string(), + data_len, + start, + end, + ); + let expected = file.effective_size(); + let metrics = ExecutionPlanMetricsSet::new(); + + let morselizer = ParquetMorselizerBuilder::new() + .with_store(Arc::clone(&store)) + .with_schema(Arc::clone(&schema)) + .with_projection_indices(&[0]) + .with_metrics(metrics.clone()) + .build(); + + let (_, rows) = + count_batches_and_rows(open_file(&morselizer, file).await.unwrap()) + .await; + + assert_eq!( + bytes_processed(&metrics), + expected, + "range [{start}, {end}) must credit exactly its own length", + ); + total_processed += expected; + total_rows += rows; + } + + assert_eq!(total_rows, 9, "the ranges together must scan every row"); + assert_eq!(total_processed, data_len); + } + + /// A scan that stops early still ends up crediting the whole range, so a + /// consumer dividing by the plan's byte total is not left permanently + /// short of 100%. + #[tokio::test] + async fn a_limit_that_ends_the_scan_early_still_credits_the_range() { + let store = Arc::new(InMemory::new()) as Arc; + let (schema, data_len) = write_three_row_groups(Arc::clone(&store)).await; + let file = PartitionedFile::new("test.parquet".to_string(), data_len); + let metrics = ExecutionPlanMetricsSet::new(); + + let morselizer = ParquetMorselizerBuilder::new() + .with_store(Arc::clone(&store)) + .with_schema(Arc::clone(&schema)) + .with_projection_indices(&[0]) + .with_limit(2) + .with_metrics(metrics.clone()) + .build(); + + let (_, rows) = + count_batches_and_rows(open_file(&morselizer, file).await.unwrap()).await; + + assert_eq!(rows, 2); + assert_eq!(bytes_processed(&metrics), data_len); + } + } + #[tokio::test] async fn test_prune_on_statistics() { let store = Arc::new(InMemory::new()) as Arc; diff --git a/datafusion/datasource-parquet/src/push_decoder.rs b/datafusion/datasource-parquet/src/push_decoder.rs index 74d8997198872..ad72ff28f9ef0 100644 --- a/datafusion/datasource-parquet/src/push_decoder.rs +++ b/datafusion/datasource-parquet/src/push_decoder.rs @@ -61,6 +61,7 @@ use datafusion_pruning::{PruningPredicate, PruningPredicateBuilder}; use crate::access_plan::PreparedAccessPlan; use crate::decoder_projection::DecoderProjection; +use crate::metrics::ByteProgress; use crate::row_group_filter::RowGroupPruningStatistics; /// Shared options applied to the [`ParquetPushDecoderBuilder`] for a file @@ -109,6 +110,11 @@ impl DecoderBuilderConfig<'_> { #[derive(Debug, Clone)] pub(crate) struct RgPlanEntry { pub(crate) rg_index: usize, + /// On-disk size of this row group, credited to + /// [`ParquetFileMetrics::bytes_processed`] once the scan is done with it. + /// + /// [`ParquetFileMetrics::bytes_processed`]: crate::ParquetFileMetrics::bytes_processed + pub(crate) bytes: u64, } /// Runtime row-group pruner driven by a dynamic predicate (e.g. the @@ -272,6 +278,10 @@ pub(crate) struct PushDecoderStreamState { pub(crate) row_group_pruner: Option, /// Count of row groups skipped at runtime by [`Self::row_group_pruner`]. pub(crate) row_groups_pruned_dynamic: Count, + /// How much of this file range the scan has finished with. Credited a row + /// group at a time as they are decoded or skipped, and topped up to the + /// full range when the stream is dropped. + pub(crate) byte_progress: ByteProgress, } impl PushDecoderStreamState { @@ -364,6 +374,7 @@ impl PushDecoderStreamState { if pruner.should_prune(&[entry.rg_index]) { pruned_count += 1; self.row_groups_pruned_dynamic.add(1); + self.byte_progress.credit(entry.bytes); } else { kept.push_back(entry); } @@ -417,7 +428,15 @@ impl PushDecoderStreamState { // Pop the RG this reader is for (we already filtered // pruned ones in step 2, so `rg_plan.front()` is the RG // the decoder is about to read). - self.rg_plan.pop_front(); + // + // Its bytes are credited here rather than once the reader + // is drained: the decoder has already fetched them, and a + // reader that is abandoned mid-row-group (`LIMIT`, early + // stop) would otherwise never credit them until the file + // closes. + if let Some(entry) = self.rg_plan.pop_front() { + self.byte_progress.credit(entry.bytes); + } self.active_reader = Some(reader); } Ok(DecodeResult::Finished) => return None, @@ -441,10 +460,18 @@ impl PushDecoderStreamState { .peek_next_row_group() .map_err(DataFusionError::from)? { - Some(actual) => Self::advance_rg_plan_to(&mut self.rg_plan, actual)?, + Some(actual) => Self::advance_rg_plan_to( + &mut self.rg_plan, + actual, + &mut self.byte_progress, + )?, // Decoder has nothing left to emit — drain our plan so the stream - // finishes cleanly. - None => self.rg_plan.clear(), + // finishes cleanly, crediting what it will now never read. + None => { + for entry in self.rg_plan.drain(..) { + self.byte_progress.credit(entry.bytes); + } + } } Ok(()) } @@ -460,12 +487,16 @@ impl PushDecoderStreamState { fn advance_rg_plan_to( rg_plan: &mut VecDeque, target: usize, + byte_progress: &mut ByteProgress, ) -> Result<()> { while let Some(front) = rg_plan.front() { if front.rg_index == target { return Ok(()); } - rg_plan.pop_front(); + // Popped here means arrow-rs finished this row group without + // handing back a reader, so the scan is done with its bytes. + let popped = rg_plan.pop_front().expect("front present"); + byte_progress.credit(popped.bytes); } internal_err!( "push decoder frontier RG {target} is not in rg_plan; \ @@ -667,28 +698,46 @@ mod tests { assert!(!pruner.should_prune(&[2])); } + /// A plan whose row group `i` is `100 * (i + 1)` bytes. + fn rg_plan(indexes: impl IntoIterator) -> VecDeque { + indexes + .into_iter() + .map(|rg_index| RgPlanEntry { + rg_index, + bytes: 100 * (rg_index as u64 + 1), + }) + .collect() + } + #[test] fn advance_rg_plan_to_pops_up_to_target() { - let mut plan: VecDeque = [0usize, 1, 2, 3] - .into_iter() - .map(|rg_index| RgPlanEntry { rg_index }) - .collect(); - PushDecoderStreamState::advance_rg_plan_to(&mut plan, 2).unwrap(); + let mut plan = rg_plan([0usize, 1, 2, 3]); + let bytes_processed = Count::new(); + let mut byte_progress = ByteProgress::new(1_000, Count::clone(&bytes_processed)); + + PushDecoderStreamState::advance_rg_plan_to(&mut plan, 2, &mut byte_progress) + .unwrap(); + assert_eq!( plan.iter().map(|e| e.rg_index).collect::>(), vec![2, 3], "must pop the entries before `target` and stop at it", ); + assert_eq!( + bytes_processed.value(), + 300, + "row groups finished without a reader must still credit their bytes", + ); } #[test] fn advance_rg_plan_to_errors_when_target_absent() { - let mut plan: VecDeque = [0usize, 1, 2] - .into_iter() - .map(|rg_index| RgPlanEntry { rg_index }) - .collect(); - let err = PushDecoderStreamState::advance_rg_plan_to(&mut plan, 5) - .expect_err("a target absent from the plan must be an internal error"); + let mut plan = rg_plan([0usize, 1, 2]); + let mut byte_progress = ByteProgress::new(1_000, Count::new()); + + let err = + PushDecoderStreamState::advance_rg_plan_to(&mut plan, 5, &mut byte_progress) + .expect_err("a target absent from the plan must be an internal error"); assert!( err.to_string().contains("diverged"), "expected a divergence internal error, got: {err}", diff --git a/datafusion/datasource-parquet/src/row_group_filter.rs b/datafusion/datasource-parquet/src/row_group_filter.rs index 9231d0232566b..df02355c0fe00 100644 --- a/datafusion/datasource-parquet/src/row_group_filter.rs +++ b/datafusion/datasource-parquet/src/row_group_filter.rs @@ -46,6 +46,22 @@ pub struct RowGroupAccessPlanFilter { access_plan: ParquetAccessPlan, } +/// Returns true if this row group belongs to `range`. +/// +/// A row group belongs to the range containing its first dictionary/data page, +/// so the ranges a file is split into for parallelism partition its row groups +/// with none shared and none left over. +/// +/// Note: don't use the location of metadata +/// +pub(crate) fn row_group_in_range(metadata: &RowGroupMetaData, range: &FileRange) -> bool { + let col = metadata.column(0); + let offset = col + .dictionary_page_offset() + .unwrap_or_else(|| col.data_page_offset()); + range.contains(offset) +} + impl RowGroupAccessPlanFilter { /// Create a new `RowGroupPlanBuilder` for pruning out the groups to scan /// based on metadata and statistics @@ -233,16 +249,7 @@ impl RowGroupAccessPlanFilter { continue; } - // Skip the row group if the first dictionary/data page are not - // within the range. - // - // note don't use the location of metadata - // - let col = metadata.column(0); - let offset = col - .dictionary_page_offset() - .unwrap_or_else(|| col.data_page_offset()); - if !range.contains(offset) { + if !row_group_in_range(metadata, range) { self.access_plan.skip(idx); } } diff --git a/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt b/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt index 41d259e88c0aa..2beeeeb569d53 100644 --- a/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt +++ b/datafusion/sqllogictest/test_files/dynamic_filter_pushdown_config.slt @@ -104,7 +104,7 @@ Plan with Metrics 03)----ProjectionExec: expr=[id@0 as id, value@1 as v, value@1 + id@0 as name], metrics=[output_rows=10, ] 04)------FilterExec: value@1 > 3, metrics=[output_rows=10, , selectivity=100% (10/10)] 05)--------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1, metrics=[output_rows=10, ] -06)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/test_data.parquet]]}, projection=[id, value], file_type=parquet, predicate=value@1 > 3 AND DynamicFilter [ value@1 IS NULL OR value@1 > 800 ], dynamic_rg_pruning=eligible, pruning_predicate=value_null_count@1 != row_count@2 AND value_max@0 > 3 AND (value_null_count@1 > 0 OR value_null_count@1 != row_count@2 AND value_max@0 > 800), required_guarantees=[], metrics=[output_rows=10, elapsed_compute=, output_bytes=80.0 B, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched -> 1 fully matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_pages_skipped_by_fully_matched=1, limit_pruned_row_groups=0 total → 0 matched, bytes_scanned=210, page_index_load_skipped=1, row_groups_pruned_dynamic_filter=0, metadata_load_time=, scan_efficiency_ratio=18.31% (210/1.15 K)] +06)----------DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_filter_pushdown_config/test_data.parquet]]}, projection=[id, value], file_type=parquet, predicate=value@1 > 3 AND DynamicFilter [ value@1 IS NULL OR value@1 > 800 ], dynamic_rg_pruning=eligible, pruning_predicate=value_null_count@1 != row_count@2 AND value_max@0 > 3 AND (value_null_count@1 > 0 OR value_null_count@1 != row_count@2 AND value_max@0 > 800), required_guarantees=[], metrics=[output_rows=10, elapsed_compute=, output_bytes=80.0 B, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=1 total → 1 matched -> 1 fully matched, row_groups_pruned_bloom_filter=1 total → 1 matched, page_index_pages_pruned=0 total → 0 matched, page_index_pages_skipped_by_fully_matched=1, limit_pruned_row_groups=0 total → 0 matched, bytes_processed=1.15 K, bytes_scanned=210, page_index_load_skipped=1, row_groups_pruned_dynamic_filter=0, metadata_load_time=, scan_efficiency_ratio=18.31% (210/1.15 K)] statement ok set datafusion.explain.analyze_level = dev; diff --git a/datafusion/sqllogictest/test_files/dynamic_row_group_pruning.slt b/datafusion/sqllogictest/test_files/dynamic_row_group_pruning.slt index c6700ebf0b97c..fddc14ca73d84 100644 --- a/datafusion/sqllogictest/test_files/dynamic_row_group_pruning.slt +++ b/datafusion/sqllogictest/test_files/dynamic_row_group_pruning.slt @@ -98,7 +98,7 @@ explain analyze select v from t order by v desc limit 3; ---- Plan with Metrics 01)SortExec: TopK(fetch=3), expr=[v@0 DESC], preserve_partitioning=[false], filter=[v@0 IS NULL OR v@0 > 12], metrics=[output_rows=3, elapsed_compute=, output_bytes=] -02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_row_group_pruning/data.parquet]]}, projection=[v], file_type=parquet, predicate=DynamicFilter [ v@0 IS NULL OR v@0 > 12 ], sort_order_for_reorder=[v@0 DESC], reverse_row_groups=true, dynamic_rg_pruning=eligible, pruning_predicate=v_null_count@0 > 0 OR v_null_count@0 != row_count@2 AND v_max@1 > 12, required_guarantees=[], metrics=[output_rows=3, elapsed_compute=, output_bytes=, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=5 total → 5 matched, row_groups_pruned_bloom_filter=5 total → 5 matched, page_index_pages_pruned=0 total → 0 matched, limit_pruned_row_groups=0 total → 0 matched, bytes_scanned=, row_groups_pruned_dynamic_filter=4, metadata_load_time=, scan_efficiency_ratio=] +02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/dynamic_row_group_pruning/data.parquet]]}, projection=[v], file_type=parquet, predicate=DynamicFilter [ v@0 IS NULL OR v@0 > 12 ], sort_order_for_reorder=[v@0 DESC], reverse_row_groups=true, dynamic_rg_pruning=eligible, pruning_predicate=v_null_count@0 > 0 OR v_null_count@0 != row_count@2 AND v_max@1 > 12, required_guarantees=[], metrics=[output_rows=3, elapsed_compute=, output_bytes=, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=5 total → 5 matched, row_groups_pruned_bloom_filter=5 total → 5 matched, page_index_pages_pruned=0 total → 0 matched, limit_pruned_row_groups=0 total → 0 matched, bytes_processed=, bytes_scanned=, row_groups_pruned_dynamic_filter=4, metadata_load_time=, scan_efficiency_ratio=] statement ok drop table t; diff --git a/datafusion/sqllogictest/test_files/explain_analyze.slt b/datafusion/sqllogictest/test_files/explain_analyze.slt index d64efe80ccae5..e083e8be33e95 100644 --- a/datafusion/sqllogictest/test_files/explain_analyze.slt +++ b/datafusion/sqllogictest/test_files/explain_analyze.slt @@ -262,7 +262,7 @@ explain analyze select * from cat_tracking where species > 'M' AND s >= 50 order ---- Plan with Metrics 01)SortExec: TopK(fetch=3), expr=[species@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[species@0 < Nlpine Sheep], metrics=[output_rows=3, output_bytes=] -02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND species_min@5 < Nlpine Sheep, required_guarantees=[], metrics=[output_rows=3, output_bytes=, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched -> 1 fully matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=2 total → 2 matched, page_index_pages_skipped_by_fully_matched=1, limit_pruned_row_groups=0 total → 0 matched, bytes_scanned=, scan_efficiency_ratio=] +02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND species_min@5 < Nlpine Sheep, required_guarantees=[], metrics=[output_rows=3, output_bytes=, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched -> 1 fully matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=2 total → 2 matched, page_index_pages_skipped_by_fully_matched=1, limit_pruned_row_groups=0 total → 0 matched, bytes_processed=, bytes_scanned=, scan_efficiency_ratio=] statement ok reset datafusion.explain.analyze_categories; @@ -277,7 +277,7 @@ explain analyze select * from cat_tracking where species > 'M' AND s >= 50 order ---- Plan with Metrics 01)SortExec: TopK(fetch=3), expr=[species@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[species@0 < Nlpine Sheep], metrics=[output_rows=3, output_bytes=] -02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND species_min@5 < Nlpine Sheep, required_guarantees=[], metrics=[output_rows=3, output_bytes=, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched -> 1 fully matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=2 total → 2 matched, page_index_pages_skipped_by_fully_matched=1, limit_pruned_row_groups=0 total → 0 matched, bytes_scanned=, scan_efficiency_ratio=] +02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND species_min@5 < Nlpine Sheep, required_guarantees=[], metrics=[output_rows=3, output_bytes=, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched -> 1 fully matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=2 total → 2 matched, page_index_pages_skipped_by_fully_matched=1, limit_pruned_row_groups=0 total → 0 matched, bytes_processed=, bytes_scanned=, scan_efficiency_ratio=] statement ok reset datafusion.explain.analyze_categories; @@ -568,7 +568,7 @@ EXPLAIN (ANALYZE, METRICS 'rows,bytes', LEVEL summary) select * from cat_trackin ---- Plan with Metrics 01)SortExec: TopK(fetch=3), expr=[species@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[species@0 < Nlpine Sheep], metrics=[output_rows=3, output_bytes=] -02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND species_min@5 < Nlpine Sheep, required_guarantees=[], metrics=[output_rows=3, output_bytes=, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched -> 1 fully matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=2 total → 2 matched, page_index_pages_skipped_by_fully_matched=1, limit_pruned_row_groups=0 total → 0 matched, bytes_scanned=, scan_efficiency_ratio=] +02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND species_min@5 < Nlpine Sheep, required_guarantees=[], metrics=[output_rows=3, output_bytes=, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched -> 1 fully matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=2 total → 2 matched, page_index_pages_skipped_by_fully_matched=1, limit_pruned_row_groups=0 total → 0 matched, bytes_processed=, bytes_scanned=, scan_efficiency_ratio=] # ---- (METRICS 'timing', LEVEL summary) — timing metrics only ---- @@ -588,7 +588,7 @@ EXPLAIN (ANALYZE, METRICS 'rows,bytes', TIMING off, LEVEL summary) select * from ---- Plan with Metrics 01)SortExec: TopK(fetch=3), expr=[species@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[species@0 < Nlpine Sheep], metrics=[output_rows=3, output_bytes=] -02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND species_min@5 < Nlpine Sheep, required_guarantees=[], metrics=[output_rows=3, output_bytes=, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched -> 1 fully matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=2 total → 2 matched, page_index_pages_skipped_by_fully_matched=1, limit_pruned_row_groups=0 total → 0 matched, bytes_scanned=, scan_efficiency_ratio=] +02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/explain_analyze/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND species_min@5 < Nlpine Sheep, required_guarantees=[], metrics=[output_rows=3, output_bytes=, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched -> 1 fully matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=2 total → 2 matched, page_index_pages_skipped_by_fully_matched=1, limit_pruned_row_groups=0 total → 0 matched, bytes_processed=, bytes_scanned=, scan_efficiency_ratio=] # ---- TIMING sugar: `METRICS 'rows', TIMING on` ↔ rows + timing ---- diff --git a/datafusion/sqllogictest/test_files/limit_pruning.slt b/datafusion/sqllogictest/test_files/limit_pruning.slt index 4ef0b5c74f3e7..5f07c91b822b1 100644 --- a/datafusion/sqllogictest/test_files/limit_pruning.slt +++ b/datafusion/sqllogictest/test_files/limit_pruning.slt @@ -63,7 +63,7 @@ set datafusion.explain.analyze_level = summary; query TT explain analyze select * from tracking_data where species > 'M' AND s >= 50 limit 3; ---- -Plan with Metrics DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/limit_pruning/data.parquet]]}, projection=[species, s], limit=3, file_type=parquet, predicate=species@0 > M AND s@1 >= 50, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50, required_guarantees=[], metrics=[output_rows=3, elapsed_compute=, output_bytes=, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched -> 1 fully matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=0 total → 0 matched, page_index_pages_skipped_by_fully_matched=1, limit_pruned_row_groups=2 total → 0 matched, bytes_scanned=, metadata_load_time=, scan_efficiency_ratio= (159/2.23 K)] +Plan with Metrics DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/limit_pruning/data.parquet]]}, projection=[species, s], limit=3, file_type=parquet, predicate=species@0 > M AND s@1 >= 50, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50, required_guarantees=[], metrics=[output_rows=3, elapsed_compute=, output_bytes=, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched -> 1 fully matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=0 total → 0 matched, page_index_pages_skipped_by_fully_matched=1, limit_pruned_row_groups=2 total → 0 matched, bytes_processed=, bytes_scanned=, metadata_load_time=, scan_efficiency_ratio= (159/2.23 K)] statement ok CREATE TABLE fully_matched_limit_source AS VALUES @@ -120,7 +120,7 @@ explain analyze select * from tracking_data where species > 'M' AND s >= 50 orde ---- Plan with Metrics 01)SortExec: TopK(fetch=3), expr=[species@0 ASC NULLS LAST], preserve_partitioning=[false], filter=[species@0 < Nlpine Sheep], metrics=[output_rows=3, elapsed_compute=, output_bytes=] -02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/limit_pruning/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND species_min@5 < Nlpine Sheep, required_guarantees=[], metrics=[output_rows=3, elapsed_compute=, output_bytes=, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched -> 1 fully matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=2 total → 2 matched, page_index_pages_skipped_by_fully_matched=1, limit_pruned_row_groups=0 total → 0 matched, bytes_scanned=, row_groups_pruned_dynamic_filter=0, metadata_load_time=, scan_efficiency_ratio= (/)] +02)--DataSourceExec: file_groups={1 group: [[WORKSPACE_ROOT/datafusion/sqllogictest/test_files/scratch/limit_pruning/data.parquet]]}, projection=[species, s], file_type=parquet, predicate=species@0 > M AND s@1 >= 50 AND DynamicFilter [ species@0 < Nlpine Sheep ], sort_order_for_reorder=[species@0 ASC NULLS LAST], dynamic_rg_pruning=eligible, pruning_predicate=species_null_count@1 != row_count@2 AND species_max@0 > M AND s_null_count@4 != row_count@2 AND s_max@3 >= 50 AND species_null_count@1 != row_count@2 AND species_min@5 < Nlpine Sheep, required_guarantees=[], metrics=[output_rows=3, elapsed_compute=, output_bytes=, files_ranges_pruned_statistics=1 total → 1 matched, row_groups_pruned_statistics=4 total → 3 matched -> 1 fully matched, row_groups_pruned_bloom_filter=3 total → 3 matched, page_index_pages_pruned=2 total → 2 matched, page_index_pages_skipped_by_fully_matched=1, limit_pruned_row_groups=0 total → 0 matched, bytes_processed=, bytes_scanned=, row_groups_pruned_dynamic_filter=0, metadata_load_time=, scan_efficiency_ratio= (/)] statement ok drop table tracking_data; diff --git a/docs/source/user-guide/explain-usage.md b/docs/source/user-guide/explain-usage.md index bc9dace297068..43ea6caac3a7a 100644 --- a/docs/source/user-guide/explain-usage.md +++ b/docs/source/user-guide/explain-usage.md @@ -207,6 +207,7 @@ Again, reading from bottom up: - `DataSourceExec` - `output_rows=99997497`: A total 99.9M rows were produced - `bytes_scanned=3703192723`: Of the 14GB file, 3.7GB were actually read (due to projection pushdown) + - `bytes_processed=14779976446`: All 14GB were accounted for: the 3.7GB read, plus the bytes of row groups that pruning ruled out. Comparing this against the total size of the files in the plan tells you how far along a scan is - `time_elapsed_opening=308.203002ms`: It took 300ms to open the file and prepare to read it - `time_elapsed_scanning_total=8.350342183s`: It took 8.3 seconds of CPU time (across 16 cores) to actually decode the parquet data - `FilterExec` From 5a65f43259000b38bb44fb07fc72c2f5b68e12c1 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 20 Aug 2026 15:04:56 +0000 Subject: [PATCH 2/6] fix(parquet): complete byte progress the moment a scan stops early MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two fixes from review of the `bytes_processed` metric. `EarlyStoppingStream` returned `None` once a dynamic filter proved the rest of a file irrelevant, but went on holding its inner stream. The decoder's drop is what credits the range's remaining bytes, so the file kept reading as partly unread after the scan had demonstrably finished with it — for however long the caller took to drop the wrapper. Hold the inner stream in an `Option` and release it when marking the stream done, which also frees the decoder's buffers at the point we stop reading rather than later. The same applies when the inner stream is simply exhausted. `ByteProgress` tracked its remaining budget as a `u64` while crediting a `Count`, which stores a `usize`. On a 32-bit target a credit above `usize::MAX` would decrement the budget in full while recording a saturated value, so the two could disagree and the metric would never reach the range size. Hold the budget at the counter's width instead, so an oversized range saturates once, at construction. `bytes_scanned` has the same ceiling; representing byte counters as `u64` would be a change to the core metric types, not to this scan. Both are covered by tests that fail without the corresponding fix. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_017cmj1SUoV7Y9ADPbzLrzZ3 --- datafusion/datasource-parquet/src/metrics.rs | 25 ++- .../src/opener/early_stop.rs | 172 +++++++++++++++++- .../datasource-parquet/src/opener/mod.rs | 9 +- 3 files changed, 189 insertions(+), 17 deletions(-) diff --git a/datafusion/datasource-parquet/src/metrics.rs b/datafusion/datasource-parquet/src/metrics.rs index 7da8d2c2968f4..425a539a8c6ff 100644 --- a/datafusion/datasource-parquet/src/metrics.rs +++ b/datafusion/datasource-parquet/src/metrics.rs @@ -135,10 +135,16 @@ pub struct ParquetFileMetrics { /// groups (the footer, the page index and any padding belong to no row group), /// and a row group is assigned to a byte range by the offset of its first page, /// so a range's row groups do not add up to precisely its length. +/// +/// The budget is held as a `usize` because [`Count`] is, so the two cannot +/// disagree: on a 32-bit target a range longer than `usize::MAX` saturates once, +/// here, rather than letting the remaining-byte arithmetic run ahead of what the +/// counter can record. [`ParquetFileMetrics::bytes_scanned`] has the same +/// ceiling. #[derive(Debug)] pub(crate) struct ByteProgress { /// Bytes of the scanned range not yet credited. - remaining: u64, + remaining: usize, bytes_processed: Count, } @@ -146,23 +152,30 @@ impl ByteProgress { /// Start tracking a range of `total` bytes. pub(crate) fn new(total: u64, bytes_processed: Count) -> Self { Self { - remaining: total, + remaining: saturating_usize(total), bytes_processed, } } /// Record that the scan is finished with `bytes` more of the range. pub(crate) fn credit(&mut self, bytes: u64) { - let bytes = bytes.min(self.remaining); + let bytes = saturating_usize(bytes).min(self.remaining); self.remaining -= bytes; - self.bytes_processed - .add(usize::try_from(bytes).unwrap_or(usize::MAX)); + self.bytes_processed.add(bytes); } } +/// Narrow a byte count to the width [`Count`] stores, saturating rather than +/// wrapping. Lossless on 64-bit targets. +pub(crate) fn saturating_usize(bytes: u64) -> usize { + usize::try_from(bytes).unwrap_or(usize::MAX) +} + impl Drop for ByteProgress { fn drop(&mut self) { - self.credit(self.remaining); + let remaining = self.remaining; + self.remaining = 0; + self.bytes_processed.add(remaining); } } diff --git a/datafusion/datasource-parquet/src/opener/early_stop.rs b/datafusion/datasource-parquet/src/opener/early_stop.rs index 75749d284068b..dc33c257dc7cb 100644 --- a/datafusion/datasource-parquet/src/opener/early_stop.rs +++ b/datafusion/datasource-parquet/src/opener/early_stop.rs @@ -38,8 +38,15 @@ pub(super) struct EarlyStoppingStream { done: bool, file_pruner: FilePruner, files_ranges_pruned_statistics: PruningMetrics, - /// The inner stream - inner: S, + /// The inner stream, dropped as soon as this stream is done with it. + /// + /// Held as an `Option` so finishing releases the decoder — and the buffers + /// and per-file metric state it owns — at the moment we stop reading, not + /// whenever the caller gets around to dropping this wrapper. Notably the + /// scan's byte-progress accounting is completed by that drop, so deferring + /// it would leave the file reading as partially scanned after the scan had + /// demonstrably finished with it. + inner: Option, } impl EarlyStoppingStream { @@ -50,11 +57,17 @@ impl EarlyStoppingStream { ) -> Self { Self { done: false, - inner: stream, + inner: Some(stream), file_pruner, files_ranges_pruned_statistics, } } + + /// Mark the stream finished and release the inner stream. + fn finish(&mut self) { + self.done = true; + self.inner = None; + } } impl EarlyStoppingStream @@ -70,7 +83,7 @@ where self.files_ranges_pruned_statistics.add_pruned(1); // Previously this file range has been counted as matched self.files_ranges_pruned_statistics.subtract_matched(1); - self.done = true; + self.finish(); Ok(None) } else { // Return the adapted batch @@ -92,10 +105,13 @@ where if self.done { return Poll::Ready(None); } - match ready!(self.inner.poll_next_unpin(cx)) { + let Some(inner) = self.inner.as_mut() else { + return Poll::Ready(None); + }; + match ready!(inner.poll_next_unpin(cx)) { None => { // input done - self.done = true; + self.finish(); Poll::Ready(None) } Some(input_batch) => { @@ -105,3 +121,147 @@ where } } } + +#[cfg(test)] +mod tests { + use super::*; + + use std::sync::Arc; + use std::sync::atomic::{AtomicBool, Ordering}; + + use arrow::array::{Int32Array, RecordBatch}; + use arrow::datatypes::{DataType, Field, Schema, SchemaRef}; + use datafusion_common::{ + ColumnStatistics, ScalarValue, Statistics, stats::Precision, + }; + use datafusion_datasource::PartitionedFile; + use datafusion_physical_expr::PhysicalExpr; + use datafusion_physical_expr::expressions::{ + BinaryExpr, Column, DynamicFilterPhysicalExpr, Literal, + }; + use datafusion_physical_plan::metrics::Count; + use futures::stream; + + /// An inner stream that records when it is dropped, standing in for the + /// decoder whose drop completes the scan's byte accounting. + struct DropRecordingStream { + inner: S, + dropped: Arc, + } + + impl Stream for DropRecordingStream { + type Item = S::Item; + + fn poll_next( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll> { + self.inner.poll_next_unpin(cx) + } + } + + impl Drop for DropRecordingStream { + fn drop(&mut self) { + self.dropped.store(true, Ordering::Relaxed); + } + } + + fn schema() -> SchemaRef { + Arc::new(Schema::new(vec![Field::new("a", DataType::Int32, false)])) + } + + /// A file whose only column holds values 1..=9, so a predicate demanding + /// larger values prunes it outright. + fn file_with_stats() -> PartitionedFile { + // Built field by field rather than from `Statistics::new_unknown`, which + // already seeds one entry per column, so that column 0 carries these + // bounds rather than an unknown placeholder. + let statistics = Statistics { + num_rows: Precision::Absent, + total_byte_size: Precision::Absent, + column_statistics: vec![ + ColumnStatistics::new_unknown() + .with_min_value(Precision::Exact(ScalarValue::Int32(Some(1)))) + .with_max_value(Precision::Exact(ScalarValue::Int32(Some(9)))) + .with_null_count(Precision::Exact(0)), + ], + }; + PartitionedFile::new("test.parquet".to_string(), 1_000) + .with_statistics(Arc::new(statistics)) + } + + fn pruning_filter(schema: &SchemaRef) -> FilePruner { + let expr: Arc = Arc::new(BinaryExpr::new( + Arc::new(Column::new("a", 0)), + datafusion_expr::Operator::Gt, + Arc::new(Literal::new(ScalarValue::Int32(Some(100)))), + )); + let dynamic: Arc = Arc::new(DynamicFilterPhysicalExpr::new( + expr.children().into_iter().map(Arc::clone).collect(), + expr, + )); + FilePruner::try_new(dynamic, schema, &file_with_stats(), Count::new()) + .expect("file has statistics, so a pruner can be built") + } + + fn batch(schema: &SchemaRef) -> RecordBatch { + RecordBatch::try_new( + Arc::clone(schema), + vec![Arc::new(Int32Array::from(vec![1]))], + ) + .unwrap() + } + + /// Stopping early must release the inner stream there and then. The decoder's + /// drop is what completes this file's byte-progress accounting, so holding it + /// until the caller drops the wrapper would leave the scan reporting a file it + /// has finished with as still partly unread. + #[tokio::test] + async fn stopping_early_releases_the_inner_stream() { + let schema = schema(); + let dropped = Arc::new(AtomicBool::new(false)); + let inner = DropRecordingStream { + inner: stream::iter(vec![Ok(batch(&schema)), Ok(batch(&schema))]), + dropped: Arc::clone(&dropped), + }; + + let mut early_stopping = EarlyStoppingStream::new( + inner, + pruning_filter(&schema), + PruningMetrics::new(), + ); + + assert!( + early_stopping.next().await.is_none(), + "the filter prunes every row, so the first batch must end the stream", + ); + assert!( + dropped.load(Ordering::Relaxed), + "the inner stream must be released when the scan stops, not when the \ + wrapper is eventually dropped", + ); + } + + /// The same must hold when the inner stream simply runs out. + #[tokio::test] + async fn exhausting_the_inner_stream_releases_it() { + let schema = schema(); + let dropped = Arc::new(AtomicBool::new(false)); + let inner = DropRecordingStream { + inner: stream::iter(Vec::>::new()), + dropped: Arc::clone(&dropped), + }; + + let mut early_stopping = EarlyStoppingStream::new( + inner, + pruning_filter(&schema), + PruningMetrics::new(), + ); + + assert!(early_stopping.next().await.is_none()); + assert!( + dropped.load(Ordering::Relaxed), + "an exhausted inner stream must be released too", + ); + } +} diff --git a/datafusion/datasource-parquet/src/opener/mod.rs b/datafusion/datasource-parquet/src/opener/mod.rs index 9c79f57676636..abc31d13addd2 100644 --- a/datafusion/datasource-parquet/src/opener/mod.rs +++ b/datafusion/datasource-parquet/src/opener/mod.rs @@ -25,7 +25,7 @@ use self::early_stop::EarlyStoppingStream; use self::encryption::EncryptionContext; use crate::access_plan::PreparedAccessPlan; use crate::decoder_projection::DecoderProjection; -use crate::metrics::ByteProgress; +use crate::metrics::{ByteProgress, saturating_usize}; use crate::page_filter::PagePruningAccessPlanFilter; use crate::push_decoder::{ DecoderBuilderConfig, PushDecoderStreamState, RgPlanEntry, RowGroupPruner, @@ -893,10 +893,9 @@ impl PreparedParquetOpen { .add_pruned(1); // The scan is done with every byte of this file range without // reading any of them. - self.file_metrics.bytes_processed.add( - usize::try_from(self.partitioned_file.effective_size()) - .unwrap_or(usize::MAX), - ); + self.file_metrics + .bytes_processed + .add(saturating_usize(self.partitioned_file.effective_size())); return Ok(None); } From 096167e52e7f6f5ab80decda967f5f96eddd27f8 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 20 Aug 2026 16:16:43 +0000 Subject: [PATCH 3/6] refactor(parquet): derive skipped bytes by subtraction MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review suggestion: the row groups left in `rg_plan` are a subset of the ones this range owns, because the plan they came from had `prune_by_range` applied. Subtracting the planned bytes from the range's total therefore leaves exactly the skipped ones, with no need to build a lookup of which row groups the plan kept. Also pin the property the in-range filter exists for: with nothing pruned a split file's ranges must credit nothing at open, since every row group a range owns is one it will read and the rest belong to its sibling. The existing assertions could not catch a range crediting its sibling's row groups — the drop-time top-up brings the total back to the range size either way. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_017cmj1SUoV7Y9ADPbzLrzZ3 --- .../datasource-parquet/src/opener/mod.rs | 40 +++++++++++-------- 1 file changed, 23 insertions(+), 17 deletions(-) diff --git a/datafusion/datasource-parquet/src/opener/mod.rs b/datafusion/datasource-parquet/src/opener/mod.rs index abc31d13addd2..baf4169f04ce8 100644 --- a/datafusion/datasource-parquet/src/opener/mod.rs +++ b/datafusion/datasource-parquet/src/opener/mod.rs @@ -1516,23 +1516,22 @@ impl RowGroupsPrunedParquetOpen { prepared.partitioned_file.effective_size(), prepared.file_metrics.bytes_processed.clone(), ); - let mut will_scan = vec![false; rg_metadata.len()]; - for entry in &rg_plan { - will_scan[entry.rg_index] = true; - } - let skipped_bytes: u64 = rg_metadata + // Every row group still in `rg_plan` is one this range owns, since the + // plan it was built from had `prune_by_range` applied. The planned row + // groups are therefore a subset of the in-range ones, and subtracting + // leaves exactly those the scan will skip. + let in_range_bytes: u64 = rg_metadata .iter() - .enumerate() - .filter(|(rg_index, rg_meta)| { - !will_scan[*rg_index] - && prepared - .file_range - .as_ref() - .is_none_or(|range| row_group_in_range(rg_meta, range)) + .filter(|rg_meta| { + prepared + .file_range + .as_ref() + .is_none_or(|range| row_group_in_range(rg_meta, range)) }) - .map(|(_, rg_meta)| row_group_bytes(rg_meta)) + .map(row_group_bytes) .sum(); - byte_progress.credit(skipped_bytes); + let planned_bytes: u64 = rg_plan.iter().map(|entry| entry.bytes).sum(); + byte_progress.credit(in_range_bytes.saturating_sub(planned_bytes)); let predicate_cache_inner_records = prepared.file_metrics.predicate_cache_inner_records.clone(); @@ -2649,9 +2648,16 @@ mod test { .with_metrics(metrics.clone()) .build(); - let (_, rows) = - count_batches_and_rows(open_file(&morselizer, file).await.unwrap()) - .await; + let stream = open_file(&morselizer, file).await.unwrap(); + assert_eq!( + bytes_processed(&metrics), + 0, + "nothing is pruned here, so range [{start}, {end}) must credit \ + nothing at open: every row group it owns is one it will read, \ + and the row groups it does not own belong to the other range", + ); + + let (_, rows) = count_batches_and_rows(stream).await; assert_eq!( bytes_processed(&metrics), From 37a72363386c2838980eed50fca1b7444b812029 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 20 Aug 2026 16:28:50 +0000 Subject: [PATCH 4/6] fix(parquet): credit byte progress for files that fail to open MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The progress guard was created alongside the decoder, so a file whose metadata load, filter preparation or bloom-filter load failed never credited its bytes: the guard that tops up the remainder on drop did not exist yet. Under `OnError::Skip` the scan carries on past that file, and `bytes_processed` is left permanently short of the plan's byte total — so a consumer dividing by it never reaches 100%. `files_processed` already counts a file that failed to open as processed, and its byte-granular counterpart should agree. Move the guard onto the state that travels through the whole open, so every early return drops it and credits the range. That also folds the file-level pruning path into the same mechanism: pruning before open now credits by dropping the guard rather than through a separate `add`, so there is one way bytes are accounted for rather than two. Covered by a test that opens a file which is not parquet at all, failing well before a stream exists; it fails if the guard is built alongside the decoder as before. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_017cmj1SUoV7Y9ADPbzLrzZ3 --- datafusion/datasource-parquet/src/metrics.rs | 8 ++- .../datasource-parquet/src/opener/mod.rs | 64 ++++++++++++++++--- 2 files changed, 59 insertions(+), 13 deletions(-) diff --git a/datafusion/datasource-parquet/src/metrics.rs b/datafusion/datasource-parquet/src/metrics.rs index 425a539a8c6ff..354bf073c944a 100644 --- a/datafusion/datasource-parquet/src/metrics.rs +++ b/datafusion/datasource-parquet/src/metrics.rs @@ -127,8 +127,10 @@ pub struct ParquetFileMetrics { /// left over is credited on drop. The counter therefore advances by exactly the /// size of the range being scanned however the scan ends: normally, at a /// `LIMIT`, when a dynamic filter proves the rest of the file irrelevant, or on -/// an error. That total is what makes the metric usable as a completion -/// fraction rather than just another counter. +/// an error — including one that stops the file being opened at all, which is +/// why the guard is created before the fallible stages of opening rather than +/// alongside the decoder. That total is what makes the metric usable as a +/// completion fraction rather than just another counter. /// /// The clamp and the final top-up also absorb two small inexactnesses in /// crediting by row group: a file is slightly larger than the sum of its row @@ -167,7 +169,7 @@ impl ByteProgress { /// Narrow a byte count to the width [`Count`] stores, saturating rather than /// wrapping. Lossless on 64-bit targets. -pub(crate) fn saturating_usize(bytes: u64) -> usize { +fn saturating_usize(bytes: u64) -> usize { usize::try_from(bytes).unwrap_or(usize::MAX) } diff --git a/datafusion/datasource-parquet/src/opener/mod.rs b/datafusion/datasource-parquet/src/opener/mod.rs index baf4169f04ce8..2c4caf4292e9c 100644 --- a/datafusion/datasource-parquet/src/opener/mod.rs +++ b/datafusion/datasource-parquet/src/opener/mod.rs @@ -25,7 +25,7 @@ use self::early_stop::EarlyStoppingStream; use self::encryption::EncryptionContext; use crate::access_plan::PreparedAccessPlan; use crate::decoder_projection::DecoderProjection; -use crate::metrics::{ByteProgress, saturating_usize}; +use crate::metrics::ByteProgress; use crate::page_filter::PagePruningAccessPlanFilter; use crate::push_decoder::{ DecoderBuilderConfig, PushDecoderStreamState, RgPlanEntry, RowGroupPruner, @@ -424,6 +424,16 @@ impl fmt::Debug for ParquetOpenState { struct PreparedParquetOpen { partition_index: usize, partitioned_file: PartitionedFile, + /// Tracks how much of this file range the scan has finished with. + /// + /// Held here, rather than built when the stream is, so that it covers the + /// fallible stages of opening a file too: loading metadata, preparing + /// filters, loading bloom filters. Any of those failing drops this state, + /// and the guard credits the range on the way out — matching + /// `files_processed`, which counts a file that failed to open as processed + /// (see `FileStreamScanState`). Otherwise a scan that skipped over a bad + /// file could never reach 100%. + byte_progress: ByteProgress, file_range: Option, extensions: datafusion_datasource::FileExtensions, file_name: String, @@ -822,8 +832,14 @@ impl ParquetMorselizer { ) }); + let byte_progress = ByteProgress::new( + partitioned_file.effective_size(), + file_metrics.bytes_processed.clone(), + ); + Ok(PreparedParquetOpen { partition_index: self.partition_index, + byte_progress, partitioned_file, file_range, extensions, @@ -891,11 +907,8 @@ impl PreparedParquetOpen { self.file_metrics .files_ranges_pruned_statistics .add_pruned(1); - // The scan is done with every byte of this file range without - // reading any of them. - self.file_metrics - .bytes_processed - .add(saturating_usize(self.partitioned_file.effective_size())); + // Dropping `self` here credits the whole range: the scan is done + // with every byte of it without having read any. return Ok(None); } @@ -1512,10 +1525,7 @@ impl RowGroupsPrunedParquetOpen { // another range. Without this the metric would sit at zero until the // first row group finishes decoding, reporting no progress for a scan // that may have just proved most of its work unnecessary. - let mut byte_progress = ByteProgress::new( - prepared.partitioned_file.effective_size(), - prepared.file_metrics.bytes_processed.clone(), - ); + let mut byte_progress = prepared.byte_progress; // Every row group still in `rg_plan` is one this range owns, since the // plan it was built from had `prune_by_range` applied. The planned row // groups are therefore a subset of the in-range ones, and subtracting @@ -2672,6 +2682,40 @@ mod test { assert_eq!(total_processed, data_len); } + /// A file that cannot be opened at all is still a file the scan is done + /// with. `files_processed` counts one under `OnError::Skip`, so its + /// byte-granular counterpart has to agree — otherwise a scan that + /// skipped a corrupt file could never reach 100%. + #[tokio::test] + async fn a_file_that_fails_to_open_still_credits_its_range() { + let store = Arc::new(InMemory::new()) as Arc; + let (schema, _) = write_three_row_groups(Arc::clone(&store)).await; + + // Not a parquet file at all, so reading the footer fails long + // before a stream — and so before a `ByteProgress` would exist if + // one were only built alongside the decoder. + let garbage = vec![b'x'; 512]; + let data_len = u64::try_from(garbage.len()).unwrap(); + store + .put(&Path::from("corrupt.parquet"), garbage.into()) + .await + .unwrap(); + + let file = PartitionedFile::new("corrupt.parquet".to_string(), data_len); + let metrics = ExecutionPlanMetricsSet::new(); + + let morselizer = ParquetMorselizerBuilder::new() + .with_store(Arc::clone(&store)) + .with_schema(schema) + .with_projection_indices(&[0]) + .with_metrics(metrics.clone()) + .build(); + + let opened = open_file(&morselizer, file).await; + assert!(opened.is_err(), "a non-parquet file must fail to open"); + assert_eq!(bytes_processed(&metrics), data_len); + } + /// A scan that stops early still ends up crediting the whole range, so a /// consumer dividing by the plan's byte total is not left permanently /// short of 100%. From d37ca45a4901982635693284a5553df9bca25074 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 20 Aug 2026 17:01:18 +0000 Subject: [PATCH 5/6] refactor(parquet): build the bytes_processed counter on demand Adding a `pub` field to `ParquetFileMetrics` is a semver-breaking change for anyone constructing it with a struct literal, which cargo-semver-checks reports and reviewers have flagged. The field also turned out to earn nothing: once file-level pruning started crediting through the progress guard, the only thing left reading it was the guard's construction. Build the counter on demand instead, next to the other metrics in this file that are registered where they are used rather than held on the struct. The public API is unchanged, so the semver break goes away, and future metrics of this kind need not widen it either. `EXPLAIN ANALYZE` output is unaffected: the counter keeps its name, type, category and filename label, so it renders in the same position. The sqllogictest expectations are unchanged and still pass. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_017cmj1SUoV7Y9ADPbzLrzZ3 --- datafusion/datasource-parquet/src/metrics.rs | 57 +++++++++++-------- .../datasource-parquet/src/opener/mod.rs | 6 +- 2 files changed, 38 insertions(+), 25 deletions(-) diff --git a/datafusion/datasource-parquet/src/metrics.rs b/datafusion/datasource-parquet/src/metrics.rs index 354bf073c944a..1be7be55ba603 100644 --- a/datafusion/datasource-parquet/src/metrics.rs +++ b/datafusion/datasource-parquet/src/metrics.rs @@ -63,23 +63,6 @@ pub struct ParquetFileMetrics { pub row_groups_pruned_dynamic_filter: Count, /// Total number of bytes scanned pub bytes_scanned: Count, - /// Total number of bytes the scan is finished with, whether they were read - /// or skipped. - /// - /// Where [`Self::bytes_scanned`] counts only the bytes fetched from the - /// object store, this counts every byte the scan has resolved: the bytes it - /// read, plus the bytes of the row groups (and whole files) that pruning - /// proved cannot contribute. Over the lifetime of a file it therefore sums - /// to that file's size — or, for a file split into byte ranges for - /// parallelism, to the size of the range — so - /// `bytes_processed / total file bytes` is a scan completion fraction, - /// which `bytes_scanned` on its own is not: it understates progress by - /// however much pruning and projection pushdown saved. - /// - /// Credited at row-group granularity: a row group's bytes land when the - /// scan is done with it. Crediting a row group's bytes progressively as its - /// rows are decoded is left to a follow-up. - pub bytes_processed: Count, /// Total rows filtered out by predicates pushed into parquet scan pub pushdown_rows_pruned: Count, /// Total rows passed predicates pushed into parquet scan @@ -222,12 +205,6 @@ impl ParquetFileMetrics { .with_category(MetricCategory::Bytes) .counter("bytes_scanned", partition); - let bytes_processed = builder - .clone() - .with_type(MetricType::Summary) - .with_category(MetricCategory::Bytes) - .counter("bytes_processed", partition); - let metadata_load_time = builder .clone() .with_type(MetricType::Summary) @@ -303,7 +280,6 @@ impl ParquetFileMetrics { row_groups_pruned_statistics, row_groups_pruned_dynamic_filter, bytes_scanned, - bytes_processed, pushdown_rows_pruned, pushdown_rows_matched, row_pushdown_eval_time, @@ -319,6 +295,39 @@ impl ParquetFileMetrics { } } + /// The `bytes_processed` counter for one file: the total number of bytes the + /// scan is finished with, whether they were read or skipped. + /// + /// Where [`Self::bytes_scanned`] counts only the bytes fetched from the + /// object store, this counts every byte the scan has resolved: the bytes it + /// read, plus the bytes of the row groups (and whole files) that pruning + /// proved cannot contribute. Over the lifetime of a file it therefore sums + /// to that file's size — or, for a file split into byte ranges for + /// parallelism, to the size of the range — so + /// `bytes_processed / total file bytes` is a scan completion fraction, + /// which `bytes_scanned` on its own is not: it understates progress by + /// however much pruning and projection pushdown saved. + /// + /// Credited at row-group granularity: a row group's bytes land when the + /// scan is done with it. Crediting a row group's bytes progressively as its + /// rows are decoded is left to a follow-up. + /// + /// Built on demand rather than held on [`ParquetFileMetrics`] because only + /// [`ByteProgress`] ever touches it, and a public field would make every + /// future metric added here a breaking change for anyone constructing the + /// struct with a literal. + pub(crate) fn bytes_processed_counter( + metrics: &ExecutionPlanMetricsSet, + partition: usize, + filename: &str, + ) -> Count { + MetricBuilder::new(metrics) + .with_new_label("filename", filename.to_string()) + .with_type(MetricType::Summary) + .with_category(MetricCategory::Bytes) + .counter("bytes_processed", partition) + } + /// Record pages whose page-index pruning was skipped because the containing /// row group was fully matched by row-group statistics. /// diff --git a/datafusion/datasource-parquet/src/opener/mod.rs b/datafusion/datasource-parquet/src/opener/mod.rs index 2c4caf4292e9c..e24ffc2a5c006 100644 --- a/datafusion/datasource-parquet/src/opener/mod.rs +++ b/datafusion/datasource-parquet/src/opener/mod.rs @@ -834,7 +834,11 @@ impl ParquetMorselizer { let byte_progress = ByteProgress::new( partitioned_file.effective_size(), - file_metrics.bytes_processed.clone(), + ParquetFileMetrics::bytes_processed_counter( + &self.metrics, + self.partition_index, + &file_name, + ), ); Ok(PreparedParquetOpen { From b8abaaf8a3336a49bc34a80fd905c5085135fbfc Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 21 Aug 2026 10:14:28 -0500 Subject: [PATCH 6/6] docs(parquet): give bytes_processed a documented home Building the counter on demand rather than storing it on `ParquetFileMetrics` left three intra-doc links pointing at a field that no longer exists, which fails rustdoc under `-D warnings`. Repointing them at the struct alone would lose the information, because there is now no Rust item that carries the metric's contract: consumers reach it by name through the plan's metrics set, and nothing said so anywhere rustdoc renders. So the fix is to document it on `ParquetFileMetrics` itself -- why it is not a field, how to read it, and what its sum means -- and point the three links there. Also records the caveat the metric's obvious use invites: it counts work resolved, not time spent. Pruned bytes are credited as soon as they are proved unnecessary, and proving that is nearly free, so the ratio is the right numerator for a progress bar and only a rough one for predicting how much longer a query will run. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_017cmj1SUoV7Y9ADPbzLrzZ3 --- datafusion/datasource-parquet/src/metrics.rs | 27 ++++++++++++++++++- .../datasource-parquet/src/opener/mod.rs | 4 +-- .../datasource-parquet/src/push_decoder.rs | 6 ++--- 3 files changed, 31 insertions(+), 6 deletions(-) diff --git a/datafusion/datasource-parquet/src/metrics.rs b/datafusion/datasource-parquet/src/metrics.rs index 1be7be55ba603..ad605ccaf6b9a 100644 --- a/datafusion/datasource-parquet/src/metrics.rs +++ b/datafusion/datasource-parquet/src/metrics.rs @@ -27,6 +27,31 @@ use datafusion_physical_plan::metrics::{ /// This component is a subject to **change** in near future and is exposed for low level integrations /// through [`ParquetFileReaderFactory`]. /// +/// # The `bytes_processed` metric +/// +/// Not every metric the parquet scan reports is a field on this struct. `bytes_processed` — the +/// number of bytes the scan is finished with, whether it read them or proved by pruning that it +/// did not need them — is registered straight onto the plan's metrics set, because only the +/// internal progress guard ever touches it and a public field here would make every future +/// metric a breaking change for anyone building this struct with a literal. +/// +/// Read it the way `EXPLAIN ANALYZE` does, by name off the plan's metrics: +/// +/// ```no_run +/// # use datafusion_physical_plan::ExecutionPlan; +/// # fn completion_fraction(plan: &dyn ExecutionPlan, total_file_bytes: u64) -> Option { +/// let processed = plan.metrics()?.sum_by_name("bytes_processed")?.as_usize(); +/// Some(processed as f64 / total_file_bytes as f64) +/// # } +/// ``` +/// +/// Over the life of a file — or of one byte range of a file split for parallelism — it advances +/// by exactly that file's size, which is what makes the ratio above a completion fraction rather +/// than just another counter. Note that it measures work *resolved*, not time spent: bytes that +/// pruning removes are credited the moment they are proved unnecessary, and proving that is +/// nearly free. It is the right numerator for a progress bar, and only a rough one for +/// predicting how much longer a query will take. +/// /// [`ParquetFileReaderFactory`]: super::ParquetFileReaderFactory #[derive(Debug, Clone)] pub struct ParquetFileMetrics { @@ -104,7 +129,7 @@ pub struct ParquetFileMetrics { } /// Tracks how much of one file — or one byte range of a file — a scan has -/// finished with, crediting [`ParquetFileMetrics::bytes_processed`] as it goes. +/// finished with, crediting [`ParquetFileMetrics`]'s `bytes_processed` as it goes. /// /// Every credit is clamped to the bytes left in the budget, and whatever is /// left over is credited on drop. The counter therefore advances by exactly the diff --git a/datafusion/datasource-parquet/src/opener/mod.rs b/datafusion/datasource-parquet/src/opener/mod.rs index e24ffc2a5c006..2e30fcc43038e 100644 --- a/datafusion/datasource-parquet/src/opener/mod.rs +++ b/datafusion/datasource-parquet/src/opener/mod.rs @@ -1630,8 +1630,8 @@ impl RowGroupsPrunedParquetOpen { } } -/// The on-disk size of a row group, as credited to -/// [`ParquetFileMetrics::bytes_processed`]. +/// The on-disk size of a row group, as credited to the `bytes_processed` metric documented on +/// [`ParquetFileMetrics`]. fn row_group_bytes(rg_meta: &RowGroupMetaData) -> u64 { u64::try_from(rg_meta.compressed_size()).unwrap_or(0) } diff --git a/datafusion/datasource-parquet/src/push_decoder.rs b/datafusion/datasource-parquet/src/push_decoder.rs index ad72ff28f9ef0..e7acefcdd4ba2 100644 --- a/datafusion/datasource-parquet/src/push_decoder.rs +++ b/datafusion/datasource-parquet/src/push_decoder.rs @@ -110,10 +110,10 @@ impl DecoderBuilderConfig<'_> { #[derive(Debug, Clone)] pub(crate) struct RgPlanEntry { pub(crate) rg_index: usize, - /// On-disk size of this row group, credited to - /// [`ParquetFileMetrics::bytes_processed`] once the scan is done with it. + /// On-disk size of this row group, credited to the `bytes_processed` metric documented on + /// [`ParquetFileMetrics`] once the scan is done with it. /// - /// [`ParquetFileMetrics::bytes_processed`]: crate::ParquetFileMetrics::bytes_processed + /// [`ParquetFileMetrics`]: crate::ParquetFileMetrics pub(crate) bytes: u64, }