diff --git a/parquet/src/arrow/arrow_reader/metrics.rs b/parquet/src/arrow/arrow_reader/metrics.rs index b36d79586bb3..635aada68f5e 100644 --- a/parquet/src/arrow/arrow_reader/metrics.rs +++ b/parquet/src/arrow/arrow_reader/metrics.rs @@ -19,6 +19,9 @@ use std::sync::Arc; use std::sync::atomic::AtomicUsize; +use std::time::Duration; + +use crate::file::metrics::DecompressionMetrics; /// This enum represents the state of Arrow reader metrics collection. /// @@ -48,6 +51,67 @@ impl ArrowReaderMetrics { Self::Enabled(Arc::new(ArrowReaderMetricsInner::new())) } + /// The decompression counters. + /// + /// Read the four cells of the page kind by phase matrix through this, for + /// example the dictionary time paid while skipping, which is work done for + /// rows that were never returned. The totals are also forwarded directly by + /// [`Self::dictionary_decompress_time`] and its siblings. + /// + /// Also the handle to give to a page reader built by hand, via + /// [`SerializedPageReader::with_decompression_metrics`]. Disabled when + /// these metrics are, so nothing downstream reads a clock. + /// + /// [`SerializedPageReader::with_decompression_metrics`]: + /// crate::file::serialized_reader::SerializedPageReader::with_decompression_metrics + pub fn decompression(&self) -> DecompressionMetrics { + match self { + Self::Disabled => DecompressionMetrics::disabled(), + Self::Enabled(inner) => inner.decompression.clone(), + } + } + + /// Time spent decompressing dictionary pages, across every column read. + /// + /// A dictionary page is decompressed once per column chunk, however few of + /// its values are later decoded, so this is the cost a selective read pays + /// before it looks at anything. Decompression happens inside decoding and + /// inside skipping, so this is a share of the scan's time rather than a + /// stage to add beside it. + /// + /// Returns None if metrics are disabled. + pub fn dictionary_decompress_time(&self) -> Option { + self.decompression_view()?.dictionary_decompress_time() + } + + /// Time spent decompressing data pages, across every column read. + /// + /// Returns None if metrics are disabled. + pub fn data_decompress_time(&self) -> Option { + self.decompression_view()?.data_decompress_time() + } + + /// Uncompressed bytes produced from dictionary pages. + /// + /// Returns None if metrics are disabled. + pub fn dictionary_decompressed_bytes(&self) -> Option { + self.decompression_view()?.dictionary_decompressed_bytes() + } + + /// Uncompressed bytes produced from data pages. + /// + /// Returns None if metrics are disabled. + pub fn data_decompressed_bytes(&self) -> Option { + self.decompression_view()?.data_decompressed_bytes() + } + + fn decompression_view(&self) -> Option<&DecompressionMetrics> { + match self { + Self::Disabled => None, + Self::Enabled(inner) => Some(&inner.decompression), + } + } + /// Predicate Cache: number of records read directly from the inner reader /// /// This is the total number of records read from the inner reader (that is @@ -122,6 +186,8 @@ pub struct ArrowReaderMetricsInner { records_read_from_inner: AtomicUsize, /// Total number of records read from previously cached pages records_read_from_cache: AtomicUsize, + /// Decompression timings, shared with every page reader of this scan + decompression: DecompressionMetrics, } impl ArrowReaderMetricsInner { @@ -130,6 +196,7 @@ impl ArrowReaderMetricsInner { Self { records_read_from_inner: AtomicUsize::new(0), records_read_from_cache: AtomicUsize::new(0), + decompression: DecompressionMetrics::enabled(), } } } diff --git a/parquet/src/arrow/arrow_reader/mod.rs b/parquet/src/arrow/arrow_reader/mod.rs index 132236c7cdf9..8ddd5d08b409 100644 --- a/parquet/src/arrow/arrow_reader/mod.rs +++ b/parquet/src/arrow/arrow_reader/mod.rs @@ -1437,6 +1437,7 @@ impl ParquetRecordBatchReaderBuilder { reader: Arc::new(input.0), metadata, row_groups, + decompression: metrics.decompression(), }; let mut plan_builder = ReadPlanBuilder::new(batch_size) @@ -1480,6 +1481,8 @@ impl ParquetRecordBatchReaderBuilder { } struct ReaderRowGroups { + /// Shared with every column's page reader, so one read totals across columns + decompression: crate::file::metrics::DecompressionMetrics, reader: Arc, metadata: Arc, @@ -1502,6 +1505,7 @@ impl RowGroups for ReaderRowGroups { reader: self.reader.clone(), metadata: self.metadata.clone(), row_groups: self.row_groups.clone().into_iter(), + decompression: self.decompression.clone(), })) } @@ -1519,6 +1523,7 @@ impl RowGroups for ReaderRowGroups { } struct ReaderPageIterator { + decompression: crate::file::metrics::DecompressionMetrics, reader: Arc, column_idx: usize, row_groups: std::vec::IntoIter, @@ -1539,6 +1544,7 @@ impl ReaderPageIterator { let reader = self.reader.clone(); SerializedPageReader::new(reader, column_chunk_metadata, total_rows, page_locations)? + .with_decompression_metrics(self.decompression.clone()) .add_crypto_context( rg_idx, self.column_idx, diff --git a/parquet/src/arrow/in_memory_row_group.rs b/parquet/src/arrow/in_memory_row_group.rs index 45399b09469a..60034975e3ea 100644 --- a/parquet/src/arrow/in_memory_row_group.rs +++ b/parquet/src/arrow/in_memory_row_group.rs @@ -22,6 +22,7 @@ use crate::column::page::{PageIterator, PageReader}; use crate::errors::ParquetError; use crate::file::metadata::page_index::RowGroupPageIndex; use crate::file::metadata::{ParquetMetaData, RowGroupMetaData}; +use crate::file::metrics::DecompressionMetrics; use crate::file::page_index::offset_index::PageLocation; use crate::file::reader::{ChunkReader, Length, SerializedPageReader}; use bytes::{Buf, Bytes}; @@ -37,6 +38,8 @@ pub(crate) struct InMemoryRowGroup<'a> { pub(crate) row_count: usize, pub(crate) row_group_idx: usize, pub(crate) metadata: &'a ParquetMetaData, + /// Handed to every column's page reader, so one scan totals across columns + pub(crate) decompression: DecompressionMetrics, } /// What ranges to fetch for the columns in this row group @@ -50,6 +53,12 @@ pub(crate) struct FetchRanges { } impl InMemoryRowGroup<'_> { + /// Measures decompression in the page readers this row group hands out. + pub(crate) fn with_decompression_metrics(mut self, metrics: DecompressionMetrics) -> Self { + self.decompression = metrics; + self + } + /// Returns the byte ranges to fetch for the columns specified in /// `projection` and `selection`. /// @@ -318,7 +327,8 @@ impl RowGroups for InMemoryRowGroup<'_> { column_chunk_metadata, self.row_count, page_locations, - )?; + )? + .with_decompression_metrics(self.decompression.clone()); let page_reader = page_reader.add_crypto_context( self.row_group_idx, i, diff --git a/parquet/src/arrow/push_decoder/reader_builder/data.rs b/parquet/src/arrow/push_decoder/reader_builder/data.rs index ab628bf28581..f2c5bc3b88f6 100644 --- a/parquet/src/arrow/push_decoder/reader_builder/data.rs +++ b/parquet/src/arrow/push_decoder/reader_builder/data.rs @@ -22,6 +22,7 @@ use crate::arrow::arrow_reader::RowSelection; use crate::arrow::in_memory_row_group::{ColumnChunkData, FetchRanges, InMemoryRowGroup}; use crate::errors::ParquetError; use crate::file::metadata::ParquetMetaData; +use crate::file::metrics::DecompressionMetrics; use crate::file::reader::ChunkReader; use crate::util::push_buffers::PushBuffers; use bytes::Bytes; @@ -109,6 +110,7 @@ impl DataRequest { page_index, row_group_idx, metadata: parquet_metadata, + decompression: DecompressionMetrics::disabled(), }; in_memory_row_group.fill_column_chunks(projection, page_start_offsets, chunks); @@ -222,6 +224,7 @@ impl<'a> DataRequestBuilder<'a> { page_index, row_group_idx, metadata: parquet_metadata, + decompression: DecompressionMetrics::disabled(), }; let FetchRanges { diff --git a/parquet/src/arrow/push_decoder/reader_builder/mod.rs b/parquet/src/arrow/push_decoder/reader_builder/mod.rs index 02d1f7014e38..11cf1cae6a59 100644 --- a/parquet/src/arrow/push_decoder/reader_builder/mod.rs +++ b/parquet/src/arrow/push_decoder/reader_builder/mod.rs @@ -559,13 +559,15 @@ impl RowGroupReaderBuilder { let predicate = filter_info.current(); - let row_group = data_request.try_into_in_memory_row_group( - row_group_idx, - row_count, - &self.metadata, - predicate.projection(), - &mut self.buffers, - )?; + let row_group = data_request + .try_into_in_memory_row_group( + row_group_idx, + row_count, + &self.metadata, + predicate.projection(), + &mut self.buffers, + )? + .with_decompression_metrics(self.metrics.decompression()); let cache_options = filter_info.cache_builder().producer(); @@ -736,13 +738,15 @@ impl RowGroupReaderBuilder { budget, } = row_group_info; - let row_group = data_request.try_into_in_memory_row_group( - row_group_idx, - row_count, - &self.metadata, - &self.projection, - &mut self.buffers, - )?; + let row_group = data_request + .try_into_in_memory_row_group( + row_group_idx, + row_count, + &self.metadata, + &self.projection, + &mut self.buffers, + )? + .with_decompression_metrics(self.metrics.decompression()); let plan = plan_builder.build(); diff --git a/parquet/src/column/page.rs b/parquet/src/column/page.rs index b0e60b9f0dfa..8c9954af7d01 100644 --- a/parquet/src/column/page.rs +++ b/parquet/src/column/page.rs @@ -24,6 +24,7 @@ use crate::errors::{ParquetError, Result}; use crate::file::metadata::thrift::{ DataPageHeader, DataPageHeaderV2, DictionaryPageHeader, PageHeader, }; +use crate::file::metrics::DecompressionPhase; use crate::file::statistics::{Statistics, page_stats_to_thrift}; /// Parquet Page definition. @@ -404,6 +405,20 @@ pub trait PageReader: Iterator> + Send { /// column index information fn skip_next_page(&mut self) -> Result<()>; + /// Tell the reader whether the pages it decompresses next are being read + /// or skipped, so decompression cost can be attributed to the right one. + /// + /// Default is a no-op: an implementation that collects no metrics, and + /// every implementation outside this crate, needs no change. + fn set_decompression_phase(&mut self, _phase: DecompressionPhase) {} + + /// Tell the reader that values are about to be read from the page it + /// loaded last, so a page (and dictionary) decompressed while skipping + /// was not decompressed for nothing. + /// + /// Default is a no-op, like [`Self::set_decompression_phase`]. + fn current_page_read(&mut self) {} + /// Returns `true` if the next page can be assumed to contain the start of a new record /// /// Prior to parquet V2 the specification was ambiguous as to whether a single record diff --git a/parquet/src/column/reader.rs b/parquet/src/column/reader.rs index 9c2345b0977a..a24dc8d0d5b0 100644 --- a/parquet/src/column/reader.rs +++ b/parquet/src/column/reader.rs @@ -27,6 +27,7 @@ use crate::column::reader::decoder::{ }; use crate::data_type::*; use crate::errors::{ParquetError, Result}; +use crate::file::metrics::DecompressionPhase; use crate::schema::types::ColumnDescPtr; use crate::util::bit_util::{ceil, num_required_bits, read_num_bytes}; @@ -230,6 +231,14 @@ where where F: FnMut(&mut V::Buffer, usize, usize, Option<&D::Buffer>) -> Result<()>, { + // Pages decompressed from here on are being decoded, not walked past. + self.page_reader + .set_decompression_phase(DecompressionPhase::Read); + // A page loaded by an earlier skip that still has values is about to be + // read from, so its decompression was not wasted. + if self.num_decoded_values < self.num_buffered_values { + self.page_reader.current_page_read(); + } let mut total_records_read = 0; let mut total_levels_read = 0; let mut total_values_read = 0; @@ -310,6 +319,11 @@ where /// /// Returns the number of records skipped pub fn skip_records(&mut self, num_records: usize) -> Result { + // A page skipped whole costs nothing, but a page walked through to + // reach a selected row inside it is decompressed and then thrown away. + // Attribute that to skipping rather than to reading. + self.page_reader + .set_decompression_phase(DecompressionPhase::Skip); let mut remaining_records = num_records; while remaining_records != 0 { if self.num_buffered_values == self.num_decoded_values { diff --git a/parquet/src/file/metadata/dictionary.rs b/parquet/src/file/metadata/dictionary.rs index 809fbff355e7..4adf0fa7c8df 100644 --- a/parquet/src/file/metadata/dictionary.rs +++ b/parquet/src/file/metadata/dictionary.rs @@ -35,6 +35,7 @@ use crate::errors::{ParquetError, Result}; #[cfg(feature = "encryption")] use crate::file::metadata::ColumnChunkMetaData; use crate::file::metadata::ParquetMetaData; +use crate::file::metrics::DecompressionRecorder; use crate::file::serialized_reader::{ SerializedPageReaderContext, decode_page, read_page_header_len_from_bytes, verify_page_size, }; @@ -117,11 +118,13 @@ pub(crate) fn decode_dictionary_page( let page_buf = page_context.decrypt_page_data(page_buf, 0, true)?; let mut decompressor = create_codec(column_metadata.compression(), &CodecOptions::default())?; + // Reading a dictionary for metadata is not part of a scan, so it is not counted. let page = decode_page( header, page_buf, column_descriptor.physical_type(), decompressor.as_mut(), + &mut DecompressionRecorder::default(), )?; let Page::DictionaryPage { buf, diff --git a/parquet/src/file/metrics.rs b/parquet/src/file/metrics.rs new file mode 100644 index 000000000000..c1595f0ffdc3 --- /dev/null +++ b/parquet/src/file/metrics.rs @@ -0,0 +1,821 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! [`DecompressionMetrics`], for measuring what a reader spends on decompression. + +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::Duration; + +/// Whether a decompressed page was used to return values or only walked past. +/// +/// Both paths decompress: a page has to be turned back into plaintext before +/// it can be walked, whether its values are then decoded or thrown away. Only +/// a page skipped whole is free. Splitting the two says how much of the cost +/// was paid for rows that were never returned. +/// +/// The phase is decided by what happens to the page, not by which call loaded +/// it. A page first loaded while skipping counts as [`Self::Read`] once any of +/// its values is read, and as [`Self::Skip`] only if the reader moves past it +/// without reading. A dictionary page counts as [`Self::Read`] once any value of +/// its column chunk is read. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum DecompressionPhase { + /// At least one value of the page (or of the dictionary's column chunk) was + /// read. + #[default] + Read, + /// The page was decompressed and then walked past without reading any of + /// its values, so the work was thrown away. + Skip, +} + +/// Whether a page held the column chunk's dictionary or its values. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum PageKind { + /// The column chunk's dictionary page. + Dictionary, + /// A data page. + Data, +} + +/// Records how long a reader spends decompressing, split by page kind and by +/// whether the page was being read or skipped. +/// +/// Decompression sits inside decoding and inside skipping, so these numbers +/// are a share of those, not a separate stage to add alongside them. +/// +/// The four cells answer different questions. A dictionary page decompressed +/// while skipping is pure waste, paid for a column chunk whose values are +/// never decoded. A data page decompressed while skipping is the unavoidable +/// cost of walking to a selected row inside it, and points at page sizing +/// rather than at the dictionary. +/// +/// Collection is off by default. When [`DecompressionMetrics::Disabled`], no +/// clock is read and no counter is touched, so a reader that does not ask for +/// these numbers pays nothing for them. +/// +/// The counters live behind an `Arc`, so cloning shares them rather than +/// copying, and one handle can be given to every column's page reader. +/// +/// # Example +/// ``` +/// # use parquet::file::metrics::{DecompressionMetrics, DecompressionPhase, PageKind}; +/// let metrics = DecompressionMetrics::enabled(); +/// // ... hand a clone to a reader, read some data ... +/// if let Some(wasted) = +/// metrics.decompress_time(PageKind::Dictionary, DecompressionPhase::Skip) +/// { +/// println!("dictionary decompression paid for skipped rows: {wasted:?}"); +/// } +/// ``` +#[derive(Debug, Clone, Default)] +pub enum DecompressionMetrics { + /// Nothing is measured. This is the default. + #[default] + Disabled, + /// Measurements accumulate in the shared inner counters. + /// + /// Create this via [`DecompressionMetrics::enabled`]. + Enabled(Arc), +} + +impl DecompressionMetrics { + /// Creates a handle that measures nothing. + pub fn disabled() -> Self { + Self::Disabled + } + + /// Creates a handle that accumulates measurements. + pub fn enabled() -> Self { + Self::Enabled(Arc::new(DecompressionMetricsInner::default())) + } + + /// Whether this handle is collecting. + /// + /// Callers check this before reading a clock, so that the disabled path + /// costs one enum discriminant test and nothing else. + #[inline] + pub fn is_enabled(&self) -> bool { + matches!(self, Self::Enabled(_)) + } + + /// Time spent decompressing pages of `kind` during `phase`. + /// + /// `None` when disabled. + pub fn decompress_time(&self, kind: PageKind, phase: DecompressionPhase) -> Option { + self.cell(kind, phase) + .map(|c| Duration::from_nanos(c.time_ns.load(Ordering::Relaxed))) + } + + /// Uncompressed bytes produced from pages of `kind` during `phase`. + /// + /// Worth reading next to the time: a large dictionary that decompresses + /// quickly and a small one that does not are different problems. + /// + /// `None` when disabled. + pub fn decompress_bytes(&self, kind: PageKind, phase: DecompressionPhase) -> Option { + self.cell(kind, phase) + .map(|c| c.bytes.load(Ordering::Relaxed)) + } + + /// Number of pages of `kind` decompressed during `phase`. + /// + /// `None` when disabled. + pub fn pages_decompressed(&self, kind: PageKind, phase: DecompressionPhase) -> Option { + self.cell(kind, phase) + .map(|c| c.pages.load(Ordering::Relaxed)) + } + + /// Time spent decompressing dictionary pages, both phases together. + /// + /// `None` when disabled. + pub fn dictionary_decompress_time(&self) -> Option { + self.total_time(PageKind::Dictionary) + } + + /// Time spent decompressing data pages, both phases together. + /// + /// `None` when disabled. + pub fn data_decompress_time(&self) -> Option { + self.total_time(PageKind::Data) + } + + /// Uncompressed bytes produced from dictionary pages, both phases together. + /// + /// `None` when disabled. + pub fn dictionary_decompressed_bytes(&self) -> Option { + self.total_bytes(PageKind::Dictionary) + } + + /// Uncompressed bytes produced from data pages, both phases together. + /// + /// `None` when disabled. + pub fn data_decompressed_bytes(&self) -> Option { + self.total_bytes(PageKind::Data) + } + + fn total_time(&self, kind: PageKind) -> Option { + let read = self.decompress_time(kind, DecompressionPhase::Read)?; + let skip = self.decompress_time(kind, DecompressionPhase::Skip)?; + Some(read + skip) + } + + fn total_bytes(&self, kind: PageKind) -> Option { + let read = self.decompress_bytes(kind, DecompressionPhase::Read)?; + let skip = self.decompress_bytes(kind, DecompressionPhase::Skip)?; + Some(read + skip) + } + + fn cell(&self, kind: PageKind, phase: DecompressionPhase) -> Option<&PageCounters> { + match self { + Self::Disabled => None, + Self::Enabled(inner) => Some(match (kind, phase) { + (PageKind::Dictionary, DecompressionPhase::Read) => &inner.dictionary_read, + (PageKind::Dictionary, DecompressionPhase::Skip) => &inner.dictionary_skip, + (PageKind::Data, DecompressionPhase::Read) => &inner.data_read, + (PageKind::Data, DecompressionPhase::Skip) => &inner.data_skip, + }), + } + } + + /// Records one decompressed page. + /// + /// `elapsed` covers only the codec call, not the surrounding copies or the + /// size checks, so it stays comparable across codecs. + pub(crate) fn record( + &self, + kind: PageKind, + phase: DecompressionPhase, + elapsed: Duration, + bytes: usize, + ) { + let Some(cell) = self.cell(kind, phase) else { + return; + }; + cell.time_ns + .fetch_add(elapsed.as_nanos() as u64, Ordering::Relaxed); + cell.bytes.fetch_add(bytes as u64, Ordering::Relaxed); + cell.pages.fetch_add(1, Ordering::Relaxed); + } +} + +/// One page's decompression cost, held until it is known whether the page was +/// read or only walked past. +#[derive(Debug, Clone, Copy)] +struct Sample { + elapsed: Duration, + bytes: usize, +} + +/// Attributes one page reader's decompression to [`DecompressionPhase::Read`] +/// or [`DecompressionPhase::Skip`] by what happens to each page afterwards. +/// +/// A page decompressed while reading is recorded as read at once. A page +/// decompressed while skipping is held back: it becomes read if values are then +/// read from it ([`Self::current_page_read`]) and skip if the reader moves on to +/// the next page, or is dropped, first. The column chunk's dictionary is held +/// the same way and becomes read as soon as any value of the chunk is read. +#[derive(Debug, Default)] +pub(crate) struct DecompressionRecorder { + metrics: DecompressionMetrics, + phase: DecompressionPhase, + pending_dictionary: Option, + pending_data: Option, +} + +impl DecompressionRecorder { + /// A recorder that writes into `metrics`. + pub(crate) fn new(metrics: DecompressionMetrics) -> Self { + Self { + metrics, + ..Self::default() + } + } + + /// Whether decompression should be timed at all. + #[inline] + pub(crate) fn is_enabled(&self) -> bool { + self.metrics.is_enabled() + } + + /// Whether the pages loaded from now on are loaded to read or to skip. + pub(crate) fn set_phase(&mut self, phase: DecompressionPhase) { + self.phase = phase; + } + + /// Records one decompressed page of `kind`. + pub(crate) fn record(&mut self, kind: PageKind, elapsed: Duration, bytes: usize) { + let sample = Sample { elapsed, bytes }; + match (kind, self.phase) { + (PageKind::Data, DecompressionPhase::Read) => { + // The previous page was left without being read, and a value of + // this chunk is about to be read, so its dictionary is used. + self.flush_data(DecompressionPhase::Skip); + self.flush_dictionary(DecompressionPhase::Read); + self.commit(PageKind::Data, DecompressionPhase::Read, sample); + } + (PageKind::Data, DecompressionPhase::Skip) => { + self.flush_data(DecompressionPhase::Skip); + self.pending_data = Some(sample); + } + (PageKind::Dictionary, DecompressionPhase::Read) => { + self.flush_dictionary(DecompressionPhase::Skip); + self.commit(PageKind::Dictionary, DecompressionPhase::Read, sample); + } + (PageKind::Dictionary, DecompressionPhase::Skip) => { + self.flush_dictionary(DecompressionPhase::Skip); + self.pending_dictionary = Some(sample); + } + } + } + + /// Values are about to be read from the page currently loaded, so neither + /// that page nor the chunk's dictionary was decompressed for nothing. + pub(crate) fn current_page_read(&mut self) { + self.flush_data(DecompressionPhase::Read); + self.flush_dictionary(DecompressionPhase::Read); + } + + fn flush_data(&mut self, phase: DecompressionPhase) { + if let Some(sample) = self.pending_data.take() { + self.commit(PageKind::Data, phase, sample); + } + } + + fn flush_dictionary(&mut self, phase: DecompressionPhase) { + if let Some(sample) = self.pending_dictionary.take() { + self.commit(PageKind::Dictionary, phase, sample); + } + } + + fn commit(&self, kind: PageKind, phase: DecompressionPhase, sample: Sample) { + self.metrics + .record(kind, phase, sample.elapsed, sample.bytes); + } +} + +impl Drop for DecompressionRecorder { + /// Whatever is still held was walked past without being read. + fn drop(&mut self) { + self.flush_data(DecompressionPhase::Skip); + self.flush_dictionary(DecompressionPhase::Skip); + } +} + +/// One cell of the page kind by phase matrix. +#[derive(Debug, Default)] +struct PageCounters { + time_ns: AtomicU64, + bytes: AtomicU64, + pages: AtomicU64, +} + +/// The counters behind [`DecompressionMetrics`]. +/// +/// See [`DecompressionMetrics`] for the public interface. +#[derive(Debug, Default)] +pub struct DecompressionMetricsInner { + dictionary_read: PageCounters, + dictionary_skip: PageCounters, + data_read: PageCounters, + data_skip: PageCounters, +} + +#[cfg(test)] +mod tests { + use super::*; + + const DICT_READ: (PageKind, DecompressionPhase) = + (PageKind::Dictionary, DecompressionPhase::Read); + const DICT_SKIP: (PageKind, DecompressionPhase) = + (PageKind::Dictionary, DecompressionPhase::Skip); + const DATA_READ: (PageKind, DecompressionPhase) = (PageKind::Data, DecompressionPhase::Read); + + #[test] + fn disabled_reports_nothing_and_ignores_records() { + let metrics = DecompressionMetrics::disabled(); + assert!(!metrics.is_enabled()); + metrics.record( + PageKind::Dictionary, + DecompressionPhase::Read, + Duration::from_millis(5), + 1024, + ); + assert_eq!(metrics.decompress_time(DICT_READ.0, DICT_READ.1), None); + assert_eq!(metrics.decompress_bytes(DICT_READ.0, DICT_READ.1), None); + assert_eq!(metrics.pages_decompressed(DICT_READ.0, DICT_READ.1), None); + assert_eq!(metrics.dictionary_decompress_time(), None); + assert_eq!(metrics.data_decompressed_bytes(), None); + } + + #[test] + fn default_is_disabled() { + assert!(!DecompressionMetrics::default().is_enabled()); + assert_eq!(DecompressionPhase::default(), DecompressionPhase::Read); + } + + #[test] + fn the_four_cells_are_counted_apart() { + let metrics = DecompressionMetrics::enabled(); + metrics.record( + PageKind::Dictionary, + DecompressionPhase::Read, + Duration::from_micros(300), + 4096, + ); + metrics.record( + PageKind::Dictionary, + DecompressionPhase::Skip, + Duration::from_micros(700), + 8192, + ); + metrics.record( + PageKind::Data, + DecompressionPhase::Read, + Duration::from_micros(100), + 512, + ); + metrics.record( + PageKind::Data, + DecompressionPhase::Read, + Duration::from_micros(150), + 256, + ); + + assert_eq!( + metrics.decompress_time(DICT_READ.0, DICT_READ.1), + Some(Duration::from_micros(300)) + ); + assert_eq!( + metrics.decompress_time(DICT_SKIP.0, DICT_SKIP.1), + Some(Duration::from_micros(700)), + "a dictionary decompressed while skipping is the wasted cell" + ); + assert_eq!( + metrics.pages_decompressed(DATA_READ.0, DATA_READ.1), + Some(2) + ); + assert_eq!( + metrics.decompress_bytes(DATA_READ.0, DATA_READ.1), + Some(768) + ); + } + + #[test] + fn totals_add_the_two_phases() { + let metrics = DecompressionMetrics::enabled(); + metrics.record( + PageKind::Dictionary, + DecompressionPhase::Read, + Duration::from_micros(300), + 4096, + ); + metrics.record( + PageKind::Dictionary, + DecompressionPhase::Skip, + Duration::from_micros(700), + 8192, + ); + assert_eq!( + metrics.dictionary_decompress_time(), + Some(Duration::from_micros(1000)) + ); + assert_eq!(metrics.dictionary_decompressed_bytes(), Some(12288)); + } + + fn ms(n: u64) -> Duration { + Duration::from_millis(n) + } + + fn pages(m: &DecompressionMetrics, kind: PageKind, phase: DecompressionPhase) -> u64 { + m.pages_decompressed(kind, phase).unwrap() + } + + #[test] + fn a_page_loaded_by_a_skip_and_then_read_counts_as_read() { + let metrics = DecompressionMetrics::enabled(); + let mut r = DecompressionRecorder::new(metrics.clone()); + r.set_phase(DecompressionPhase::Skip); + r.record(PageKind::Dictionary, ms(1), 10); + r.record(PageKind::Data, ms(1), 100); + r.set_phase(DecompressionPhase::Read); + r.current_page_read(); + drop(r); + assert_eq!(pages(&metrics, PageKind::Data, DecompressionPhase::Read), 1); + assert_eq!(pages(&metrics, PageKind::Data, DecompressionPhase::Skip), 0); + assert_eq!( + pages(&metrics, PageKind::Dictionary, DecompressionPhase::Read), + 1 + ); + assert_eq!( + pages(&metrics, PageKind::Dictionary, DecompressionPhase::Skip), + 0 + ); + } + + #[test] + fn a_page_walked_past_counts_as_skip_and_a_later_read_saves_the_dictionary() { + let metrics = DecompressionMetrics::enabled(); + let mut r = DecompressionRecorder::new(metrics.clone()); + r.set_phase(DecompressionPhase::Skip); + r.record(PageKind::Dictionary, ms(1), 10); + r.record(PageKind::Data, ms(1), 100); // walked past + r.set_phase(DecompressionPhase::Read); + r.record(PageKind::Data, ms(1), 100); // loaded and read + drop(r); + assert_eq!(pages(&metrics, PageKind::Data, DecompressionPhase::Skip), 1); + assert_eq!(pages(&metrics, PageKind::Data, DecompressionPhase::Read), 1); + assert_eq!( + pages(&metrics, PageKind::Dictionary, DecompressionPhase::Read), + 1 + ); + } + + #[test] + fn a_chunk_that_is_only_skipped_counts_everything_as_skip() { + let metrics = DecompressionMetrics::enabled(); + let mut r = DecompressionRecorder::new(metrics.clone()); + r.set_phase(DecompressionPhase::Skip); + r.record(PageKind::Dictionary, ms(1), 10); + r.record(PageKind::Data, ms(1), 100); + r.record(PageKind::Data, ms(1), 100); + drop(r); + assert_eq!(pages(&metrics, PageKind::Data, DecompressionPhase::Skip), 2); + assert_eq!( + pages(&metrics, PageKind::Dictionary, DecompressionPhase::Skip), + 1 + ); + assert_eq!(pages(&metrics, PageKind::Data, DecompressionPhase::Read), 0); + assert_eq!( + pages(&metrics, PageKind::Dictionary, DecompressionPhase::Read), + 0 + ); + } + + #[test] + fn a_clone_shares_the_counters() { + let metrics = DecompressionMetrics::enabled(); + let handed_to_a_page_reader = metrics.clone(); + handed_to_a_page_reader.record( + PageKind::Data, + DecompressionPhase::Skip, + Duration::from_micros(10), + 64, + ); + assert_eq!(metrics.data_decompressed_bytes(), Some(64)); + } + + #[test] + fn counters_start_at_zero_rather_than_none_when_enabled() { + let metrics = DecompressionMetrics::enabled(); + assert_eq!( + metrics.dictionary_decompress_time(), + Some(Duration::ZERO), + "an enabled handle that saw no pages reports zero, not None" + ); + assert_eq!( + metrics.pages_decompressed(DATA_READ.0, DATA_READ.1), + Some(0) + ); + } +} + +#[cfg(all(test, feature = "arrow", feature = "zstd"))] +mod end_to_end_tests { + use std::sync::Arc; + + use arrow_array::types::Int32Type; + use arrow_array::{ArrayRef, Int64Array, ListArray, RecordBatch, StringArray}; + use bytes::Bytes; + + use super::*; + use crate::DecodeResult; + use crate::arrow::ArrowWriter; + use crate::arrow::arrow_reader::metrics::ArrowReaderMetrics; + use crate::arrow::arrow_reader::{ParquetRecordBatchReaderBuilder, RowSelection, RowSelector}; + use crate::arrow::push_decoder::ParquetPushDecoderBuilder; + use crate::basic::{Compression, ZstdLevel}; + use crate::file::metadata::ParquetMetaDataPushDecoder; + use crate::file::properties::{EnabledStatistics, WriterProperties}; + + const ROWS: usize = 40_000; + const DISTINCT: usize = 4_000; + + /// A zstd file with a dictionary big enough that decompressing it is + /// measurable: many distinct, long strings. + fn file() -> Bytes { + let strings: Vec = (0..ROWS) + .map(|i| format!("TICKER{:06}-{}", i % DISTINCT, "x".repeat(32))) + .collect(); + let ticker: ArrayRef = Arc::new(StringArray::from( + strings.iter().map(|s| s.as_str()).collect::>(), + )); + let value: ArrayRef = Arc::new(Int64Array::from( + (0..ROWS).map(|i| i as i64).collect::>(), + )); + let batch = RecordBatch::try_from_iter(vec![("ticker", ticker), ("value", value)]).unwrap(); + + let props = WriterProperties::builder() + .set_compression(Compression::ZSTD(ZstdLevel::try_new(3).unwrap())) + .set_dictionary_enabled(true) + .set_data_page_row_count_limit(2_000) + .set_write_batch_size(2_000) + .build(); + let mut out = Vec::new(); + let mut writer = ArrowWriter::try_new(&mut out, batch.schema(), Some(props)).unwrap(); + writer.write(&batch).unwrap(); + writer.close().unwrap(); + Bytes::from(out) + } + + fn read_all(data: Bytes, metrics: &ArrowReaderMetrics) -> usize { + let reader = ParquetRecordBatchReaderBuilder::try_new(data) + .unwrap() + .with_metrics(metrics.clone()) + .build() + .unwrap(); + reader.map(|b| b.unwrap().num_rows()).sum() + } + + #[test] + fn a_full_read_attributes_everything_to_the_read_phase() { + let metrics = ArrowReaderMetrics::enabled(); + let rows = read_all(file(), &metrics); + assert_eq!(rows, ROWS); + + let dict_read = metrics + .decompression() + .pages_decompressed(PageKind::Dictionary, DecompressionPhase::Read) + .unwrap(); + let data_read = metrics + .decompression() + .pages_decompressed(PageKind::Data, DecompressionPhase::Read) + .unwrap(); + let dict_skip = metrics + .decompression() + .pages_decompressed(PageKind::Dictionary, DecompressionPhase::Skip) + .unwrap(); + let data_skip = metrics + .decompression() + .pages_decompressed(PageKind::Data, DecompressionPhase::Skip) + .unwrap(); + + assert!(dict_read > 0, "the string column has a dictionary page"); + assert!(data_read > 0, "and several data pages"); + assert_eq!( + (dict_skip, data_skip), + (0, 0), + "nothing was skipped, so the skip cells stay empty" + ); + assert!( + metrics.dictionary_decompressed_bytes().unwrap() > 0, + "bytes are recorded alongside the time" + ); + } + + #[test] + fn skipping_one_row_then_reading_the_rest_counts_as_read() { + // The first page is loaded by the skip, but all its other values are + // returned, so none of its decompression was thrown away. + let metrics = ArrowReaderMetrics::enabled(); + let reader = ParquetRecordBatchReaderBuilder::try_new(file()) + .unwrap() + .with_metrics(metrics.clone()) + .with_row_selection(RowSelection::from(vec![ + RowSelector::skip(1), + RowSelector::select(ROWS - 1), + ])) + .build() + .unwrap(); + let rows: usize = reader.map(|b| b.unwrap().num_rows()).sum(); + assert_eq!(rows, ROWS - 1); + + let d = metrics.decompression(); + for kind in [PageKind::Dictionary, PageKind::Data] { + assert_eq!( + d.pages_decompressed(kind, DecompressionPhase::Skip), + Some(0), + "{kind:?}" + ); + assert!( + d.pages_decompressed(kind, DecompressionPhase::Read) + .unwrap() + > 0 + ); + } + } + + /// A zstd file with one list column and no offset index, so skipping rows + /// cannot jump over whole pages: every page on the way has to be + /// decompressed to walk its repetition levels. + fn list_file() -> Bytes { + let lists: Vec>>> = (0..ROWS) + .map(|i| Some((0..(i % 4) as i32).map(Some).collect())) + .collect(); + let tags: ArrayRef = Arc::new(ListArray::from_iter_primitive::(lists)); + let batch = RecordBatch::try_from_iter(vec![("tags", tags)]).unwrap(); + let props = WriterProperties::builder() + .set_compression(Compression::ZSTD(ZstdLevel::try_new(3).unwrap())) + .set_offset_index_disabled(true) + .set_statistics_enabled(EnabledStatistics::Chunk) + .set_data_page_row_count_limit(2_000) + .set_write_batch_size(2_000) + .build(); + let mut out = Vec::new(); + let mut writer = ArrowWriter::try_new(&mut out, batch.schema(), Some(props)).unwrap(); + writer.write(&batch).unwrap(); + writer.close().unwrap(); + Bytes::from(out) + } + + #[test] + fn a_selective_read_splits_pages_by_whether_any_value_was_read() { + // Keep one row in every 5000. A page holding a kept row is decompressed + // and read from; a page between kept rows is decompressed only to walk + // past it. + let mut selectors = Vec::new(); + let mut left = ROWS; + while left > 0 { + let skip = 4_999.min(left); + selectors.push(RowSelector::skip(skip)); + left -= skip; + if left == 0 { + break; + } + selectors.push(RowSelector::select(1)); + left -= 1; + } + + let metrics = ArrowReaderMetrics::enabled(); + let reader = ParquetRecordBatchReaderBuilder::try_new(list_file()) + .unwrap() + .with_metrics(metrics.clone()) + .with_row_selection(RowSelection::from(selectors)) + .build() + .unwrap(); + let rows: usize = reader.map(|b| b.unwrap().num_rows()).sum(); + assert!(rows > 0 && rows < ROWS / 100, "a few rows out of many"); + + let d = metrics.decompression(); + let read = d + .pages_decompressed(PageKind::Data, DecompressionPhase::Read) + .unwrap(); + let skip = d + .pages_decompressed(PageKind::Data, DecompressionPhase::Skip) + .unwrap(); + assert_eq!(read, rows as u64, "one page per kept row, each read from"); + assert!(skip > 0, "pages between kept rows were walked past"); + assert!( + d.decompress_bytes(PageKind::Data, DecompressionPhase::Skip) + .unwrap() + > 0 + ); + } + + /// Reads `data` through the push decoder, the path the async reader and + /// query engines built on it take. + fn read_all_push(data: Bytes, metrics: &ArrowReaderMetrics) -> usize { + let len = data.len() as u64; + let mut metadata_decoder = ParquetMetaDataPushDecoder::try_new(len).unwrap(); + metadata_decoder.push_range(0..len, data.clone()).unwrap(); + let DecodeResult::Data(metadata) = metadata_decoder.try_decode().unwrap() else { + panic!("metadata did not decode"); + }; + let mut decoder = ParquetPushDecoderBuilder::try_new_decoder(Arc::new(metadata)) + .unwrap() + .with_metrics(metrics.clone()) + .build() + .unwrap(); + let mut rows = 0; + loop { + match decoder.try_decode().unwrap() { + DecodeResult::NeedsData(ranges) => { + let bytes = ranges + .iter() + .map(|r| data.slice(r.start as usize..r.end as usize)) + .collect(); + decoder.push_ranges(ranges, bytes).unwrap(); + } + DecodeResult::Data(batch) => rows += batch.num_rows(), + DecodeResult::Finished => return rows, + } + } + } + + #[test] + fn the_push_decoder_records_what_the_sync_reader_records() { + let data = file(); + let sync = ArrowReaderMetrics::enabled(); + let push = ArrowReaderMetrics::enabled(); + assert_eq!(read_all(data.clone(), &sync), ROWS); + assert_eq!(read_all_push(data, &push), ROWS); + + // Bytes and page counts are deterministic; times are not compared. + for kind in [PageKind::Dictionary, PageKind::Data] { + let sync_bytes = sync + .decompression() + .decompress_bytes(kind, DecompressionPhase::Read) + .unwrap(); + assert!(sync_bytes > 0, "{kind:?}"); + assert_eq!( + push.decompression() + .decompress_bytes(kind, DecompressionPhase::Read), + Some(sync_bytes), + "{kind:?}" + ); + assert_eq!( + push.decompression() + .pages_decompressed(kind, DecompressionPhase::Read), + sync.decompression() + .pages_decompressed(kind, DecompressionPhase::Read), + "{kind:?}" + ); + } + assert!(push.dictionary_decompress_time().is_some()); + } + + #[test] + fn disabled_metrics_report_nothing_after_a_real_read() { + let metrics = ArrowReaderMetrics::disabled(); + let rows = read_all(file(), &metrics); + assert_eq!(rows, ROWS, "the read still works"); + assert_eq!(metrics.dictionary_decompress_time(), None); + assert_eq!(metrics.data_decompress_time(), None); + assert_eq!(metrics.dictionary_decompressed_bytes(), None); + assert_eq!( + metrics + .decompression() + .pages_decompressed(PageKind::Data, DecompressionPhase::Read), + None + ); + } + + #[test] + fn totals_are_the_two_phases_added() { + let metrics = ArrowReaderMetrics::enabled(); + read_all(file(), &metrics); + let d = metrics.decompression(); + assert_eq!( + metrics.data_decompressed_bytes().unwrap(), + d.decompress_bytes(PageKind::Data, DecompressionPhase::Read) + .unwrap() + + d.decompress_bytes(PageKind::Data, DecompressionPhase::Skip) + .unwrap() + ); + } +} diff --git a/parquet/src/file/mod.rs b/parquet/src/file/mod.rs index b741256c9b9e..d4cd3dcc5d39 100644 --- a/parquet/src/file/mod.rs +++ b/parquet/src/file/mod.rs @@ -100,6 +100,7 @@ #[cfg(feature = "encryption")] pub mod column_crypto_metadata; pub mod metadata; +pub mod metrics; pub mod page_index; pub mod properties; pub mod reader; diff --git a/parquet/src/file/serialized_reader.rs b/parquet/src/file/serialized_reader.rs index aab7bc8ab304..3aad4904492e 100644 --- a/parquet/src/file/serialized_reader.rs +++ b/parquet/src/file/serialized_reader.rs @@ -27,6 +27,9 @@ use crate::encryption::decrypt::{CryptoContext, read_and_decrypt}; use crate::errors::{ParquetError, Result}; use crate::file::metadata::page_index::RowGroupPageIndex; use crate::file::metadata::thrift::PageHeader; +use crate::file::metrics::{ + DecompressionMetrics, DecompressionPhase, DecompressionRecorder, PageKind, +}; use crate::file::page_index::offset_index::PageLocation; use crate::file::statistics; use crate::file::{ @@ -402,6 +405,7 @@ pub(crate) fn decode_page( buffer: Bytes, physical_type: Type, decompressor: Option<&mut Box>, + recorder: &mut DecompressionRecorder, ) -> Result { // Verify the 32-bit CRC checksum of the page #[cfg(feature = "crc")] @@ -460,7 +464,28 @@ pub(crate) fn decode_page( // see https://github.com/apache/parquet-format/blob/master/README.md#data-pages if decompressed_size > 0 { let compressed = &buffer[offset..]; - decompressor.decompress(compressed, &mut decompressed, Some(decompressed_size))?; + // Reading a clock per page is not free, so only do it when the + // caller asked for these numbers. Disabled is one enum test. + if recorder.is_enabled() { + let started = std::time::Instant::now(); + decompressor.decompress( + compressed, + &mut decompressed, + Some(decompressed_size), + )?; + let kind = if page_header.r#type == PageType::DICTIONARY_PAGE { + PageKind::Dictionary + } else { + PageKind::Data + }; + recorder.record(kind, started.elapsed(), decompressed_size); + } else { + decompressor.decompress( + compressed, + &mut decompressed, + Some(decompressed_size), + )?; + } } if decompressed.len() != uncompressed_page_size { @@ -585,9 +610,22 @@ pub struct SerializedPageReader { state: SerializedPageReaderState, context: SerializedPageReaderContext, + + /// Where decompression timings go, attributed to read or skip by what + /// the column reader does with each page. Disabled unless the caller asked. + decompression: DecompressionRecorder, } impl SerializedPageReader { + /// Collect decompression timings into `metrics`. + /// + /// Off by default: without this call the reader reads no clock while + /// decompressing. Pass a clone of one handle to every column's page reader + /// to total a whole read. + pub fn with_decompression_metrics(mut self, metrics: DecompressionMetrics) -> Self { + self.decompression = DecompressionRecorder::new(metrics); + self + } /// Creates a new serialized page reader from a chunk reader and metadata pub fn new( reader: Arc, @@ -687,6 +725,7 @@ impl SerializedPageReader { state, physical_type: meta.column_type(), context, + decompression: DecompressionRecorder::default(), }) } @@ -934,6 +973,14 @@ pub(crate) fn verify_page_size( } impl PageReader for SerializedPageReader { + fn set_decompression_phase(&mut self, phase: DecompressionPhase) { + self.decompression.set_phase(phase); + } + + fn current_page_read(&mut self) { + self.decompression.current_page_read(); + } + fn get_next_page(&mut self) -> Result> { loop { let page = match &mut self.state { @@ -988,6 +1035,7 @@ impl PageReader for SerializedPageReader { buffer, self.physical_type, self.decompressor.as_mut(), + &mut self.decompression, )?; if page.is_data_page() { *page_index += 1; @@ -1032,6 +1080,7 @@ impl PageReader for SerializedPageReader { bytes, self.physical_type, self.decompressor.as_mut(), + &mut self.decompression, )? } }; @@ -1234,7 +1283,14 @@ mod tests { }; let buffer = Bytes::new(); - let err = decode_page(page_header, buffer, Type::INT32, None).unwrap_err(); + let err = decode_page( + page_header, + buffer, + Type::INT32, + None, + &mut DecompressionRecorder::default(), + ) + .unwrap_err(); assert!( err.to_string() .contains("DataPage v2 header contains implausible values") @@ -1254,7 +1310,14 @@ mod tests { data_page_header_v2: None, }; let buffer = Bytes::new(); - let err = decode_page(page_header.clone(), buffer.clone(), Type::INT32, None).unwrap_err(); + let err = decode_page( + page_header.clone(), + buffer.clone(), + Type::INT32, + None, + &mut DecompressionRecorder::default(), + ) + .unwrap_err(); assert_eq!( err.to_string(), "Parquet error: Page type INDEX_PAGE is not supported" @@ -1270,7 +1333,14 @@ mod tests { is_compressed: None, statistics: None, }); - let err = decode_page(page_header, buffer, Type::INT32, None).unwrap_err(); + let err = decode_page( + page_header, + buffer, + Type::INT32, + None, + &mut DecompressionRecorder::default(), + ) + .unwrap_err(); assert!( err.to_string() .contains("DataPage v2 header contains implausible values")