etseidl commented on code in PR #10420: URL: https://github.com/apache/arrow-rs/pull/10420#discussion_r4106892227
########## parquet/src/file/metadata/dictionary.rs: ########## @@ -0,0 +1,383 @@ +// 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. + +//! Decoding a column chunk's dictionary page directly into an Arrow array, +//! independent of the row-by-row [`ArrayReader`] machinery. +//! +//! This is useful for callers that want the *set* of distinct values stored +//! in a dictionary-encoded column chunk without reading any data pages, e.g. +//! to prune a row group when the query predicate's literals are known not to +//! be in the dictionary. +//! +//! [`ArrayReader`]: crate::arrow::array_reader::ArrayReader + +use crate::arrow::{ByteArrayDecoderPlain, OffsetBuffer}; +use crate::basic::{ConvertedType, LogicalType, PageType, Type as PhysicalType}; +use crate::column::page::Page; +use crate::compression::{CodecOptions, create_codec}; +#[cfg(feature = "encryption")] +use crate::encryption::decrypt::CryptoContext; +use crate::errors::{ParquetError, Result}; +#[cfg(feature = "encryption")] +use crate::file::metadata::ColumnChunkMetaData; +use crate::file::metadata::ParquetMetaData; +use crate::file::serialized_reader::{ + SerializedPageReaderContext, decode_page, read_page_header_len_from_bytes, verify_page_size, +}; +use crate::schema::types::ColumnDescriptor; +use arrow_array::ArrayRef; +use arrow_schema::DataType as ArrowType; +use bytes::Bytes; +#[cfg(feature = "encryption")] +use std::sync::Arc; + +/// Decodes the dictionary page of a column chunk into an [`ArrayRef`]. +/// +/// `buffer` must contain the entire dictionary page, byte-for-byte, i.e. the +/// range `[dictionary_page_offset, data_page_offset)` of the column chunk. +/// +/// Only `BYTE_ARRAY` columns are currently supported; other physical types +/// return an error. The returned array never contains nulls: dictionary +/// pages only store the distinct non-null values, with nulls represented via +/// definition levels in the data pages. +/// +/// Note this only decodes whatever dictionary page is present -- it does +/// **not** verify that the entire column chunk is dictionary-encoded (i.e. +/// that every value in the chunk is drawn from this dictionary). Callers +/// that need that guarantee (for example, to use the dictionary as an exact +/// membership index) must check that themselves, e.g. via +/// [`crate::file::metadata::ColumnChunkMetaData::page_encoding_stats_mask`]. +pub(crate) fn decode_dictionary_page( + buffer: Bytes, + parquet_meta_data: &ParquetMetaData, + row_group_idx: usize, + column_idx: usize, +) -> Result<ArrayRef> { + let column_metadata = parquet_meta_data + .row_group(row_group_idx) + .column(column_idx); + let column_descriptor = column_metadata.column_descr(); + + if column_descriptor.physical_type() != PhysicalType::BYTE_ARRAY { + return Err(ParquetError::General(format!( + "decode_dictionary_page only supports BYTE_ARRAY columns, got {}", + column_descriptor.physical_type() + ))); + } + + // Dictionary pages are subject to the same modular encryption as data + // pages: both the page header and the page body may be ciphertext, so + // we must route through the same crypto-aware header/data path that + // `SerializedPageReader` uses rather than parsing the header directly. + let page_context = SerializedPageReaderContext { + read_stats: true, + #[cfg(feature = "encryption")] + crypto_context: dictionary_page_crypto_context( + parquet_meta_data, + column_metadata, + row_group_idx, + column_idx, + )?, + }; + + let (consumed, header) = + read_page_header_len_from_bytes(&page_context, buffer.as_ref(), 0, true)?; + if header.r#type != PageType::DICTIONARY_PAGE { + return Err(ParquetError::General(format!( + "Expected a dictionary page, found {:?}", + header.r#type + ))); + } + + // `compressed_page_size` comes from the (possibly maliciously crafted) + // file header; `verify_page_size` bounds-checks it against what we + // actually fetched before we slice, instead of trusting it blindly. + let remaining = (buffer.len() - consumed) as u64; + verify_page_size( + header.compressed_page_size, + header.uncompressed_page_size, + remaining, + )?; + let compressed_size = header.compressed_page_size as usize; + let page_buf = buffer.slice(consumed..consumed + compressed_size); + let page_buf = page_context.decrypt_page_data(page_buf, 0, true)?; + + let mut decompressor = create_codec(column_metadata.compression(), &CodecOptions::default())?; + let page = decode_page( + header, + page_buf, + column_descriptor.physical_type(), + decompressor.as_mut(), + )?; + let Page::DictionaryPage { + buf, num_values, .. + } = page + else { + return Err(ParquetError::General( + "Expected a dictionary page".to_string(), + )); + }; + let num_values = num_values as usize; + + // The dictionary page is always PLAIN-encoded, regardless of what the + // data pages' encoding is (RLE_DICTIONARY/PLAIN_DICTIONARY only describe + // how *data* pages reference the dictionary by index). + let is_utf8 = is_utf8(column_descriptor); + let mut decoder = ByteArrayDecoderPlain::new(buf, num_values, Some(num_values), is_utf8); + let mut offsets = OffsetBuffer::<i32>::with_capacity(num_values); + decoder.read(&mut offsets, usize::MAX)?; + + let arrow_type = if is_utf8 { + ArrowType::Utf8 + } else { + ArrowType::Binary + }; + Ok(offsets.into_array(None, arrow_type)) +} + +/// Builds the crypto context needed to decrypt the dictionary page of +/// `column_metadata`, or `None` if the file (or this column) isn't encrypted. +#[cfg(feature = "encryption")] +fn dictionary_page_crypto_context( + parquet_meta_data: &ParquetMetaData, + column_metadata: &ColumnChunkMetaData, + row_group_idx: usize, + column_idx: usize, +) -> Result<Option<Arc<CryptoContext>>> { + let Some(file_decryptor) = parquet_meta_data.file_decryptor() else { + return Ok(None); + }; + let Some(crypto_metadata) = column_metadata.crypto_metadata() else { + return Ok(None); + }; + let crypto_context = + CryptoContext::for_column(file_decryptor, crypto_metadata, row_group_idx, column_idx)? + .for_dictionary_page(); Review Comment: This is using the passed in `row_group_idx`, which is an index into the current `parquet_meta_data::row_groups`. But the row groups may have been filtered at this point, so the 0th row group may have been the 2nd in the original file. I think it would be better to use `parquet_meta_data.row_group(row_group_idx).ordinal()` here, and error if there is no ordinal in the metadata (it should be present when modular encryption is used, for exactly this purpose). ########## parquet/src/file/metadata/dictionary.rs: ########## @@ -0,0 +1,383 @@ +// 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. + +//! Decoding a column chunk's dictionary page directly into an Arrow array, +//! independent of the row-by-row [`ArrayReader`] machinery. +//! +//! This is useful for callers that want the *set* of distinct values stored +//! in a dictionary-encoded column chunk without reading any data pages, e.g. +//! to prune a row group when the query predicate's literals are known not to +//! be in the dictionary. +//! +//! [`ArrayReader`]: crate::arrow::array_reader::ArrayReader + +use crate::arrow::{ByteArrayDecoderPlain, OffsetBuffer}; +use crate::basic::{ConvertedType, LogicalType, PageType, Type as PhysicalType}; +use crate::column::page::Page; +use crate::compression::{CodecOptions, create_codec}; +#[cfg(feature = "encryption")] +use crate::encryption::decrypt::CryptoContext; +use crate::errors::{ParquetError, Result}; +#[cfg(feature = "encryption")] +use crate::file::metadata::ColumnChunkMetaData; +use crate::file::metadata::ParquetMetaData; +use crate::file::serialized_reader::{ + SerializedPageReaderContext, decode_page, read_page_header_len_from_bytes, verify_page_size, +}; +use crate::schema::types::ColumnDescriptor; +use arrow_array::ArrayRef; +use arrow_schema::DataType as ArrowType; +use bytes::Bytes; +#[cfg(feature = "encryption")] +use std::sync::Arc; + +/// Decodes the dictionary page of a column chunk into an [`ArrayRef`]. +/// +/// `buffer` must contain the entire dictionary page, byte-for-byte, i.e. the +/// range `[dictionary_page_offset, data_page_offset)` of the column chunk. +/// +/// Only `BYTE_ARRAY` columns are currently supported; other physical types +/// return an error. The returned array never contains nulls: dictionary +/// pages only store the distinct non-null values, with nulls represented via +/// definition levels in the data pages. +/// +/// Note this only decodes whatever dictionary page is present -- it does +/// **not** verify that the entire column chunk is dictionary-encoded (i.e. +/// that every value in the chunk is drawn from this dictionary). Callers +/// that need that guarantee (for example, to use the dictionary as an exact +/// membership index) must check that themselves, e.g. via +/// [`crate::file::metadata::ColumnChunkMetaData::page_encoding_stats_mask`]. +pub(crate) fn decode_dictionary_page( + buffer: Bytes, + parquet_meta_data: &ParquetMetaData, + row_group_idx: usize, + column_idx: usize, +) -> Result<ArrayRef> { + let column_metadata = parquet_meta_data + .row_group(row_group_idx) + .column(column_idx); + let column_descriptor = column_metadata.column_descr(); + + if column_descriptor.physical_type() != PhysicalType::BYTE_ARRAY { + return Err(ParquetError::General(format!( + "decode_dictionary_page only supports BYTE_ARRAY columns, got {}", + column_descriptor.physical_type() + ))); + } + + // Dictionary pages are subject to the same modular encryption as data + // pages: both the page header and the page body may be ciphertext, so + // we must route through the same crypto-aware header/data path that + // `SerializedPageReader` uses rather than parsing the header directly. + let page_context = SerializedPageReaderContext { + read_stats: true, + #[cfg(feature = "encryption")] + crypto_context: dictionary_page_crypto_context( + parquet_meta_data, + column_metadata, + row_group_idx, + column_idx, + )?, + }; + + let (consumed, header) = + read_page_header_len_from_bytes(&page_context, buffer.as_ref(), 0, true)?; + if header.r#type != PageType::DICTIONARY_PAGE { + return Err(ParquetError::General(format!( + "Expected a dictionary page, found {:?}", + header.r#type + ))); + } + + // `compressed_page_size` comes from the (possibly maliciously crafted) + // file header; `verify_page_size` bounds-checks it against what we + // actually fetched before we slice, instead of trusting it blindly. + let remaining = (buffer.len() - consumed) as u64; + verify_page_size( + header.compressed_page_size, + header.uncompressed_page_size, + remaining, + )?; + let compressed_size = header.compressed_page_size as usize; + let page_buf = buffer.slice(consumed..consumed + compressed_size); + let page_buf = page_context.decrypt_page_data(page_buf, 0, true)?; + + let mut decompressor = create_codec(column_metadata.compression(), &CodecOptions::default())?; + let page = decode_page( + header, + page_buf, + column_descriptor.physical_type(), + decompressor.as_mut(), + )?; + let Page::DictionaryPage { + buf, num_values, .. + } = page + else { + return Err(ParquetError::General( + "Expected a dictionary page".to_string(), + )); + }; + let num_values = num_values as usize; + + // The dictionary page is always PLAIN-encoded, regardless of what the + // data pages' encoding is (RLE_DICTIONARY/PLAIN_DICTIONARY only describe + // how *data* pages reference the dictionary by index). + let is_utf8 = is_utf8(column_descriptor); + let mut decoder = ByteArrayDecoderPlain::new(buf, num_values, Some(num_values), is_utf8); + let mut offsets = OffsetBuffer::<i32>::with_capacity(num_values); + decoder.read(&mut offsets, usize::MAX)?; Review Comment: Apparently `read` here can return when the input buffer is exhausted, but it still returns `to_read`, rather than the actual number of values read. You might want to do something like ```suggestion let num_read = decoder.read(&mut offsets, usize::MAX)?; if offsets.len() != num_values { return Err(general_err!("did not read entire dictionary")); } ``` ########## parquet/src/file/metadata/dictionary.rs: ########## @@ -0,0 +1,383 @@ +// 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. + +//! Decoding a column chunk's dictionary page directly into an Arrow array, +//! independent of the row-by-row [`ArrayReader`] machinery. +//! +//! This is useful for callers that want the *set* of distinct values stored +//! in a dictionary-encoded column chunk without reading any data pages, e.g. +//! to prune a row group when the query predicate's literals are known not to +//! be in the dictionary. +//! +//! [`ArrayReader`]: crate::arrow::array_reader::ArrayReader + +use crate::arrow::{ByteArrayDecoderPlain, OffsetBuffer}; +use crate::basic::{ConvertedType, LogicalType, PageType, Type as PhysicalType}; +use crate::column::page::Page; +use crate::compression::{CodecOptions, create_codec}; +#[cfg(feature = "encryption")] +use crate::encryption::decrypt::CryptoContext; +use crate::errors::{ParquetError, Result}; +#[cfg(feature = "encryption")] +use crate::file::metadata::ColumnChunkMetaData; +use crate::file::metadata::ParquetMetaData; +use crate::file::serialized_reader::{ + SerializedPageReaderContext, decode_page, read_page_header_len_from_bytes, verify_page_size, +}; +use crate::schema::types::ColumnDescriptor; +use arrow_array::ArrayRef; +use arrow_schema::DataType as ArrowType; +use bytes::Bytes; +#[cfg(feature = "encryption")] +use std::sync::Arc; + +/// Decodes the dictionary page of a column chunk into an [`ArrayRef`]. +/// +/// `buffer` must contain the entire dictionary page, byte-for-byte, i.e. the +/// range `[dictionary_page_offset, data_page_offset)` of the column chunk. +/// +/// Only `BYTE_ARRAY` columns are currently supported; other physical types +/// return an error. The returned array never contains nulls: dictionary +/// pages only store the distinct non-null values, with nulls represented via +/// definition levels in the data pages. +/// +/// Note this only decodes whatever dictionary page is present -- it does +/// **not** verify that the entire column chunk is dictionary-encoded (i.e. +/// that every value in the chunk is drawn from this dictionary). Callers +/// that need that guarantee (for example, to use the dictionary as an exact +/// membership index) must check that themselves, e.g. via +/// [`crate::file::metadata::ColumnChunkMetaData::page_encoding_stats_mask`]. +pub(crate) fn decode_dictionary_page( + buffer: Bytes, + parquet_meta_data: &ParquetMetaData, + row_group_idx: usize, + column_idx: usize, +) -> Result<ArrayRef> { Review Comment: This function will conditionally return binary or string arrays depending on the parquet logical type. If we want to later extend this to other physical types, we're now in the business of casting all of those to the appropriate arrow type, like we do in the array readers (e.g. https://github.com/apache/arrow-rs/blob/9e0ab287e9494891abe6f4cb45ce847886ede840/parquet/src/arrow/array_reader/primitive_array.rs#L162). This also doesn't handle decimals with a BYTE_ARRAY physical type. Maybe for now we can simply return a binary array, and leave it to the user to either cast to an appropriate arrow type, or instead cast probes to the physical type. Perhaps a later effort could add helpers for either approach (potentially reusing the current parquet->arrow conversion logic). ########## parquet/src/file/metadata/dictionary.rs: ########## @@ -0,0 +1,383 @@ +// 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. + +//! Decoding a column chunk's dictionary page directly into an Arrow array, +//! independent of the row-by-row [`ArrayReader`] machinery. +//! +//! This is useful for callers that want the *set* of distinct values stored +//! in a dictionary-encoded column chunk without reading any data pages, e.g. +//! to prune a row group when the query predicate's literals are known not to +//! be in the dictionary. +//! +//! [`ArrayReader`]: crate::arrow::array_reader::ArrayReader + +use crate::arrow::{ByteArrayDecoderPlain, OffsetBuffer}; +use crate::basic::{ConvertedType, LogicalType, PageType, Type as PhysicalType}; +use crate::column::page::Page; +use crate::compression::{CodecOptions, create_codec}; +#[cfg(feature = "encryption")] +use crate::encryption::decrypt::CryptoContext; +use crate::errors::{ParquetError, Result}; +#[cfg(feature = "encryption")] +use crate::file::metadata::ColumnChunkMetaData; +use crate::file::metadata::ParquetMetaData; +use crate::file::serialized_reader::{ + SerializedPageReaderContext, decode_page, read_page_header_len_from_bytes, verify_page_size, +}; +use crate::schema::types::ColumnDescriptor; +use arrow_array::ArrayRef; +use arrow_schema::DataType as ArrowType; +use bytes::Bytes; +#[cfg(feature = "encryption")] +use std::sync::Arc; + +/// Decodes the dictionary page of a column chunk into an [`ArrayRef`]. +/// +/// `buffer` must contain the entire dictionary page, byte-for-byte, i.e. the +/// range `[dictionary_page_offset, data_page_offset)` of the column chunk. +/// +/// Only `BYTE_ARRAY` columns are currently supported; other physical types +/// return an error. The returned array never contains nulls: dictionary +/// pages only store the distinct non-null values, with nulls represented via +/// definition levels in the data pages. +/// +/// Note this only decodes whatever dictionary page is present -- it does +/// **not** verify that the entire column chunk is dictionary-encoded (i.e. +/// that every value in the chunk is drawn from this dictionary). Callers +/// that need that guarantee (for example, to use the dictionary as an exact +/// membership index) must check that themselves, e.g. via +/// [`crate::file::metadata::ColumnChunkMetaData::page_encoding_stats_mask`]. +pub(crate) fn decode_dictionary_page( + buffer: Bytes, + parquet_meta_data: &ParquetMetaData, + row_group_idx: usize, + column_idx: usize, +) -> Result<ArrayRef> { + let column_metadata = parquet_meta_data + .row_group(row_group_idx) + .column(column_idx); + let column_descriptor = column_metadata.column_descr(); + + if column_descriptor.physical_type() != PhysicalType::BYTE_ARRAY { + return Err(ParquetError::General(format!( + "decode_dictionary_page only supports BYTE_ARRAY columns, got {}", + column_descriptor.physical_type() + ))); + } + + // Dictionary pages are subject to the same modular encryption as data + // pages: both the page header and the page body may be ciphertext, so + // we must route through the same crypto-aware header/data path that + // `SerializedPageReader` uses rather than parsing the header directly. + let page_context = SerializedPageReaderContext { + read_stats: true, + #[cfg(feature = "encryption")] + crypto_context: dictionary_page_crypto_context( + parquet_meta_data, + column_metadata, + row_group_idx, + column_idx, + )?, + }; + + let (consumed, header) = + read_page_header_len_from_bytes(&page_context, buffer.as_ref(), 0, true)?; + if header.r#type != PageType::DICTIONARY_PAGE { + return Err(ParquetError::General(format!( + "Expected a dictionary page, found {:?}", + header.r#type + ))); + } Review Comment: We should also check `header.dictionary_page_header.encoding` to ensure it's `PLAIN`. I think there's been talk of using other encodings (although maybe just replacing RLE) so it would be nice to future-proof this. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
