You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/08/21 15:37:11 UTC
(datafusion) branch gh-readonly-queue/main/pr-24524-5f0ba13c804ea8835766e68ba0ed3f3c8ab54a93 created (now a6e2d3f7a4)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-24524-5f0ba13c804ea8835766e68ba0ed3f3c8ab54a93
in repository https://gitbox.apache.org/repos/asf/datafusion.git
at a6e2d3f7a4 feat(parquet): add `bytes_processed` scan-completion metric (#24524)
This branch includes the following new commits:
new a6e2d3f7a4 feat(parquet): add `bytes_processed` scan-completion metric (#24524)
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org
(datafusion) 01/01: feat(parquet): add `bytes_processed` scan-completion metric (#24524)
Posted by gi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-24524-5f0ba13c804ea8835766e68ba0ed3f3c8ab54a93
in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit a6e2d3f7a43b8007817889875b662fa0463468c7
Author: Adrian Garcia Badaracco <17...@users.noreply.github.com>
AuthorDate: Fri Aug 21 15:36:46 2026 +0000
feat(parquet): add `bytes_processed` scan-completion metric (#24524)
## Which issue does this PR close?
- No issue filed yet. Happy to open one describing the problem before
review if that is preferred.
## Rationale for this change
Nothing today reports how far along a scan is. That is what a progress
bar, a query-progress API, or a watchdog that cancels runaway queries
needs, and both existing signals fail at it in opposite directions:
- **`bytes_scanned`** looks like the right metric, because its natural
denominator — the size of the files in the plan — is known before the
query runs. But it counts only the bytes actually *fetched*. A scan that
prunes most of its row groups, or projects 3 of 50 columns, reports a
small fraction of the file even once it has finished, so `bytes_scanned
/ total file bytes` understates progress by a factor that varies per
query and is not knowable up front.
- **`files_processed`** only moves when a whole file completes. With
high scan parallelism it reads 0% through the entire first wave of
files, however much work is in flight, and then jumps.
The gap is not the unit, it is the numerator: a scan is *done with* a
byte once it has either read it or proved it does not need it, and
`bytes_scanned` counts only the first half.
## What changes are included in this PR?
A new `bytes_processed` counter on `ParquetFileMetrics`, alongside
`bytes_scanned` and sharing its metric type, category and per-file
label:
```
bytes_scanned — bytes fetched from the object store
bytes_processed — bytes the scan is finished with, whether read or pruned
```
Its contract is one invariant:
> Over the lifetime of a file — or, for a file split into byte ranges
for parallelism, a range — `bytes_processed` advances by exactly
`effective_size()`, monotonically.
so `bytes_processed / total file bytes` is a completion fraction that
needs no statistics and no second metric to normalise against.
Credit lands a row group at a time:
| event | credit |
| --- | --- |
| file/range pruned before open (file statistics, dynamic filter) |
`effective_size()` |
| row groups the final access plan skips — range, statistics, bloom
filter, limit, page index | `compressed_size(rg)`, at open |
| row group dropped mid-scan by a dynamic filter | `compressed_size(rg)`
|
| row group reached by the decoder | `compressed_size(rg)` |
| file closed for any reason (finished, `LIMIT`, early stop, error) |
the remainder |
Two details worth a reviewer's attention:
- **`ByteProgress`** (in `metrics.rs`) clamps every credit to the bytes
left in the range and credits the remainder on `Drop`. The clamp absorbs
the two inexactnesses of crediting by row group — a file is slightly
larger than the sum of its row groups, and a row group is assigned to a
byte range by its first page's offset — and `Drop` is what makes the
invariant hold under `LIMIT`, `EarlyStoppingStream` and errors without
instrumenting every exit path.
- **`row_group_in_range`** is extracted out of
`RowGroupAccessPlanFilter::prune_by_range` so the open-time skip pass
decides which row groups a range owns by the same rule, instead of
duplicating the offset logic. Without it, one range of a split file
would credit its siblings' row groups against its own budget and jump
straight to 100% at open.
Folding every open-time pruning stage into a single pass over the
*final* access plan is what keeps this free of double counting: range,
statistics, bloom, limit and page-index pruning have all been applied by
then, so there is no need to instrument five sites and reason about
their overlap.
Cost is one atomic add per row group. Crediting a row group's bytes
progressively as its rows decode — proportionally by rows resolved,
trued up at the boundary — is deliberately left to a follow-up; it does
not change this metric's contract.
## Are these changes tested?
Yes. Seven new tests in `opener::test::bytes_processed`, all asserting
the invariant, for: a plain scan, a scan pruning row groups, a file
pruned before open by a dynamic filter (also asserting `bytes_scanned ==
0`), a byte-split file (each range credits exactly its own length, and
the ranges sum to the file), a `LIMIT` that ends the scan early, a file
that errors before it can be opened at all, and one that steps through
the stream batch by batch to pin that credit advances *during* the scan
rather than all at once on close.
The two carrying the real claims were mutation-checked: removing the
mid-scan credit fails `credit_advances_while_the_scan_runs`, and
removing the open-time credit fails
`row_group_pruning_is_credited_before_any_batch_is_read`. The other five
would also pass a trivial credit-everything-on-close implementation, so
those two are the ones doing the work.
`push_decoder`'s existing `advance_rg_plan_to_pops_up_to_target` and
`advance_rg_plan_to_errors_when_target_absent` are extended to assert
that a row group arrow-rs finishes without handing back a reader still
credits its bytes.
Also run: the full `datafusion-datasource-parquet` suite (233 passed),
the `explain_analyze` core tests, the four affected sqllogictest files,
`cargo fmt`, `cargo clippy --all-targets --all-features -- -D warnings`,
and rustdoc with `-D warnings`.
## Are there any user-facing changes?
- `EXPLAIN ANALYZE` on a parquet scan gains a `bytes_processed=` entry,
next to `bytes_scanned`. Four sqllogictest files are updated; the diff
there is exactly eight added `bytes_processed=` fragments and nothing
else — existing `<slt:ignore>` markers and pinned values are preserved.
- A bullet in `docs/source/user-guide/explain-usage.md` documenting the
metric next to `bytes_scanned`.
- No Rust API change: the diff adds and removes no public item.
`bytes_processed` is built on demand rather than stored on
`ParquetFileMetrics`, so the struct is untouched and code constructing
it with a struct literal still compiles. Consumers read the counter out
of the plan's `MetricsSet` by name, exactly as they read `bytes_scanned`
— which is the intended way to drive a progress bar or a query watchdog
from it.
---
🤖 Generated with [Claude Code](https://claude.com/claude-code)
https://claude.ai/code/session_017cmj1SUoV7Y9ADPbzLrzZ3
---
_Generated by [Claude
Code](https://claude.ai/code/session_017cmj1SUoV7Y9ADPbzLrzZ3)_
---------
Co-authored-by: Claude <no...@anthropic.com>
---
datafusion/core/tests/sql/explain_analyze.rs | 1 +
datafusion/datasource-parquet/src/metrics.rs | 119 +++++++
.../datasource-parquet/src/opener/early_stop.rs | 172 ++++++++++-
datafusion/datasource-parquet/src/opener/mod.rs | 344 ++++++++++++++++++++-
datafusion/datasource-parquet/src/push_decoder.rs | 81 ++++-
.../datasource-parquet/src/row_group_filter.rs | 27 +-
.../test_files/dynamic_filter_pushdown_config.slt | 2 +-
.../test_files/dynamic_row_group_pruning.slt | 2 +-
.../sqllogictest/test_files/explain_analyze.slt | 8 +-
.../sqllogictest/test_files/limit_pruning.slt | 4 +-
docs/source/user-guide/explain-usage.md | 1 +
11 files changed, 719 insertions(+), 42 deletions(-)
diff --git a/datafusion/core/tests/sql/explain_analyze.rs b/datafusion/core/tests/sql/explain_analyze.rs
index 4c8b8f9c01..f8498cbedf 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 a3573c8624..ad605ccaf6 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<f64> {
+/// 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 {
@@ -103,6 +128,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`]'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
+/// 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 +320,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 75749d2840..dc33c257dc 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<S> {
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<S>,
}
impl<S> EarlyStoppingStream<S> {
@@ -50,11 +57,17 @@ impl<S> EarlyStoppingStream<S> {
) -> 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<S> EarlyStoppingStream<S>
@@ -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<S> {
+ inner: S,
+ dropped: Arc<AtomicBool>,
+ }
+
+ impl<S: Stream + Unpin> Stream for DropRecordingStream<S> {
+ type Item = S::Item;
+
+ fn poll_next(
+ mut self: Pin<&mut Self>,
+ cx: &mut Context<'_>,
+ ) -> Poll<Option<Self::Item>> {
+ self.inner.poll_next_unpin(cx)
+ }
+ }
+
+ impl<S> Drop for DropRecordingStream<S> {
+ 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<dyn PhysicalExpr> = 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<dyn PhysicalExpr> = 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::<Result<RecordBatch>>::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 b3ce024d66..2e30fcc430 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<datafusion_datasource::FileRange>,
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 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)
+}
+
type ConstantColumns = HashMap<String, ScalarValue>;
/// 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<dyn ObjectStore>) -> (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<dyn ObjectStore>;
+ 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<dyn ObjectStore>;
+ 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<dyn ObjectStore>;
+ 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<dyn ObjectStore>;
+ 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<dyn ObjectStore>;
+ 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<dyn ObjectStore>;
+ 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<dyn ObjectStore>;
+ 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<dyn ObjectStore>;
diff --git a/datafusion/datasource-parquet/src/push_decoder.rs b/datafusion/datasource-parquet/src/push_decoder.rs
index 74d8997198..e7acefcdd4 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 the `bytes_processed` metric documented on
+ /// [`ParquetFileMetrics`] once the scan is done with it.
+ ///
+ /// [`ParquetFileMetrics`]: crate::ParquetFileMetrics
+ 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<RowGroupPruner>,
/// 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<RgPlanEntry>,
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<Item = usize>) -> VecDeque<RgPlanEntry> {
+ 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<RgPlanEntry> = [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<_>>(),
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<RgPlanEntry> = [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 9231d02325..df02355c0f 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
+/// <https://github.com/apache/datafusion/issues/5995>
+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
- // <https://github.com/apache/datafusion/issues/5995>
- 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 41d259e88c..2beeeeb569 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, <slt:ignore>]
04)------FilterExec: value@1 > 3, metrics=[output_rows=10, <slt:ignore>, selectivity=100% (10/10)]
05)--------RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1, metrics=[output_rows=10, <slt:ignore>]
-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_guarante [...]
+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_guarante [...]
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 c6700ebf0b..fddc14ca73 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=<slt:ignore>, output_bytes=<slt:ignore>]
-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=<slt:ignore>, out [...]
+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=<slt:ignore>, out [...]
statement ok
drop table t;
diff --git a/datafusion/sqllogictest/test_files/explain_analyze.slt b/datafusion/sqllogictest/test_files/explain_analyze.slt
index d64efe80cc..e083e8be33 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=<slt:ignore>]
-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_c [...]
+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_c [...]
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=<slt:ignore>]
-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_c [...]
+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_c [...]
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=<slt:ignore>]
-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_c [...]
+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_c [...]
# ---- (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=<slt:ignore>]
-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_c [...]
+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_c [...]
# ---- 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 4ef0b5c74f..5f07c91b82 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=<slt:ignore>, output_bytes=<slt:ignore>, files_ranges_pruned [...]
+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=<slt:ignore>, output_bytes=<slt:ignore>, files_ranges_pruned [...]
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=<slt:ignore>, output_bytes=<slt:ignore>]
-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_cou [...]
+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_cou [...]
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 bc9dace297..43ea6caac3 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`
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org