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..1be7be55ba603 100644 --- a/datafusion/datasource-parquet/src/metrics.rs +++ b/datafusion/datasource-parquet/src/metrics.rs @@ -103,6 +103,67 @@ 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 — 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 +/// 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: usize, + bytes_processed: Count, +} + +impl ByteProgress { + /// Start tracking a range of `total` bytes. + pub(crate) fn new(total: u64, bytes_processed: Count) -> Self { + Self { + 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 = saturating_usize(bytes).min(self.remaining); + self.remaining -= bytes; + self.bytes_processed.add(bytes); + } +} + +/// Narrow a byte count to the width [`Count`] stores, saturating rather than +/// wrapping. Lossless on 64-bit targets. +fn saturating_usize(bytes: u64) -> usize { + usize::try_from(bytes).unwrap_or(usize::MAX) +} + +impl Drop for ByteProgress { + fn drop(&mut self) { + let remaining = self.remaining; + self.remaining = 0; + self.bytes_processed.add(remaining); + } +} + impl ParquetFileMetrics { /// Create new metrics pub fn new( @@ -234,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/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 b3ce024d66f1f..e24ffc2a5c006 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, @@ -423,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, @@ -821,8 +832,18 @@ impl ParquetMorselizer { ) }); + let byte_progress = ByteProgress::new( + partitioned_file.effective_size(), + ParquetFileMetrics::bytes_processed_counter( + &self.metrics, + self.partition_index, + &file_name, + ), + ); + Ok(PreparedParquetOpen { partition_index: self.partition_index, + byte_progress, partitioned_file, file_range, extensions, @@ -890,6 +911,8 @@ impl PreparedParquetOpen { self.file_metrics .files_ranges_pruned_statistics .add_pruned(1); + // Dropping `self` here credits the whole range: the scan is done + // with every byte of it without having read any. return Ok(None); } @@ -1480,7 +1503,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 +1523,30 @@ 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 = 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 + // leaves exactly those the scan will skip. + let in_range_bytes: u64 = rg_metadata + .iter() + .filter(|rg_meta| { + prepared + .file_range + .as_ref() + .is_none_or(|range| row_group_in_range(rg_meta, range)) + }) + .map(row_group_bytes) + .sum(); + 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(); let predicate_cache_records = @@ -1557,6 +1607,7 @@ impl RowGroupsPrunedParquetOpen { baseline_metrics: prepared.baseline_metrics, row_group_pruner, row_groups_pruned_dynamic, + byte_progress, } .into_stream(); @@ -1579,6 +1630,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 +2463,289 @@ 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 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), + 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 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%. + #[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`