alamb commented on code in PR #12090:
URL: https://github.com/apache/arrow-rs/pull/12090#discussion_r4517932469
##########
parquet/src/file/metadata/reader.rs:
##########
@@ -801,6 +805,77 @@ impl ParquetMetaDataReader {
self.load_page_index_with_remainder_impl(fetch, None).await
}
+ /// Reads the Bloom filter for one physical leaf column in a row group.
+ ///
+ /// The metadata and reader must refer to the same file. `column_idx` is a
+ /// physical Parquet leaf index, not a top-level Arrow field index.
+ ///
+ /// Returns `Ok(None)` without I/O when the column chunk has no Bloom
filter.
+ /// Invalid coordinates, invalid Bloom metadata, malformed filters, and
read
+ /// failures return an error. The filter is returned separately; the
supplied
+ /// metadata is neither modified nor reloaded.
+ ///
+ /// Prefer reading filters after other pruning has narrowed the row groups
and
Review Comment:
I recommend we also note that reading BloomFilters typically requires more
I/O as they are not stored inline in the footer
##########
parquet/src/file/metadata/reader.rs:
##########
@@ -801,6 +805,77 @@ impl ParquetMetaDataReader {
self.load_page_index_with_remainder_impl(fetch, None).await
}
+ /// Reads the Bloom filter for one physical leaf column in a row group.
+ ///
+ /// The metadata and reader must refer to the same file. `column_idx` is a
+ /// physical Parquet leaf index, not a top-level Arrow field index.
+ ///
+ /// Returns `Ok(None)` without I/O when the column chunk has no Bloom
filter.
+ /// Invalid coordinates, invalid Bloom metadata, malformed filters, and
read
Review Comment:
I am not sure we need to spell out all the possible errors -- like "if there
is an error, it is returned" is already kind of redundant
##########
parquet/src/file/metadata/reader.rs:
##########
@@ -801,6 +805,77 @@ impl ParquetMetaDataReader {
self.load_page_index_with_remainder_impl(fetch, None).await
}
+ /// Reads the Bloom filter for one physical leaf column in a row group.
+ ///
+ /// The metadata and reader must refer to the same file. `column_idx` is a
+ /// physical Parquet leaf index, not a top-level Arrow field index.
+ ///
+ /// Returns `Ok(None)` without I/O when the column chunk has no Bloom
filter.
+ /// Invalid coordinates, invalid Bloom metadata, malformed filters, and
read
+ /// failures return an error. The filter is returned separately; the
supplied
+ /// metadata is neither modified nor reloaded.
+ ///
+ /// Prefer reading filters after other pruning has narrowed the row groups
and
+ /// columns of interest. Footer and page index reads never load Bloom
filters.
+ ///
+ /// # Example
+ ///
+ /// Read footer metadata first, then request a filter only when needed:
+ ///
+ /// ```rust
+ /// use bytes::Bytes;
+ /// use parquet::errors::Result;
+ /// use parquet::file::metadata::ParquetMetaDataReader;
+ /// # use std::sync::Arc;
+ /// # use parquet::data_type::Int32Type;
+ /// # use parquet::file::properties::WriterProperties;
+ /// # use parquet::file::writer::SerializedFileWriter;
+ /// # use parquet::schema::parser::parse_message_type;
+ /// # fn main() -> Result<()> {
+ /// # let schema = Arc::new(parse_message_type("message schema { REQUIRED
INT32 value; }")?);
+ /// # let properties =
Arc::new(WriterProperties::builder().set_bloom_filter_enabled(true).build());
+ /// # let mut writer = SerializedFileWriter::new(Vec::new(), schema,
properties)?;
+ /// # let mut row_group = writer.next_row_group()?;
+ /// # let mut column = row_group.next_column()?.unwrap();
+ /// # column.typed::<Int32Type>().write_batch(&[42], None, None)?;
+ /// # column.close()?;
+ /// # row_group.close()?;
+ /// # let data = Bytes::from(writer.into_inner()?);
+ /// let metadata = ParquetMetaDataReader::new().parse_and_finish(&data)?;
+ /// let filter = ParquetMetaDataReader::read_column_bloom_filter(&data,
&metadata, 0, 0)?;
+ /// assert!(filter.unwrap().check(&42_i32));
+ /// # Ok(())
+ /// # }
+ /// ```
+ pub fn read_column_bloom_filter<R: ChunkReader>(
+ reader: &R,
+ metadata: &ParquetMetaData,
+ row_group_idx: usize,
+ column_idx: usize,
+ ) -> Result<Option<Sbbf>> {
+ let column = bloom_filter_column(metadata, row_group_idx, column_idx)?;
+ Sbbf::read_from_column_chunk(column, reader)
+ }
+
+ /// Asynchronous version of [`Self::read_column_bloom_filter`].
+ ///
+ /// Accepts a [`MetadataFetch`] byte source without requiring an Arrow
reader
+ /// builder or an [`crate::arrow::async_reader::AsyncFileReader`]
implementation.
+ /// Existing async file readers can be passed by mutable reference and
reused
+ /// for subsequent reads. For a boxed trait-object reader, pass `&mut
reader`
Review Comment:
> For a boxed trait-object reader, pass `&mut reader`
Is that specific advice for this function? It seems like general advice for
any boxed trait (and thus we don't need to document anything special for
Parquet rust)
##########
parquet/src/file/metadata/reader.rs:
##########
@@ -801,6 +805,77 @@ impl ParquetMetaDataReader {
self.load_page_index_with_remainder_impl(fetch, None).await
}
+ /// Reads the Bloom filter for one physical leaf column in a row group.
+ ///
+ /// The metadata and reader must refer to the same file. `column_idx` is a
+ /// physical Parquet leaf index, not a top-level Arrow field index.
+ ///
+ /// Returns `Ok(None)` without I/O when the column chunk has no Bloom
filter.
+ /// Invalid coordinates, invalid Bloom metadata, malformed filters, and
read
+ /// failures return an error. The filter is returned separately; the
supplied
+ /// metadata is neither modified nor reloaded.
+ ///
+ /// Prefer reading filters after other pruning has narrowed the row groups
and
+ /// columns of interest. Footer and page index reads never load Bloom
filters.
+ ///
+ /// # Example
+ ///
+ /// Read footer metadata first, then request a filter only when needed:
+ ///
+ /// ```rust
+ /// use bytes::Bytes;
+ /// use parquet::errors::Result;
+ /// use parquet::file::metadata::ParquetMetaDataReader;
+ /// # use std::sync::Arc;
+ /// # use parquet::data_type::Int32Type;
+ /// # use parquet::file::properties::WriterProperties;
+ /// # use parquet::file::writer::SerializedFileWriter;
+ /// # use parquet::schema::parser::parse_message_type;
+ /// # fn main() -> Result<()> {
+ /// # let schema = Arc::new(parse_message_type("message schema { REQUIRED
INT32 value; }")?);
+ /// # let properties =
Arc::new(WriterProperties::builder().set_bloom_filter_enabled(true).build());
+ /// # let mut writer = SerializedFileWriter::new(Vec::new(), schema,
properties)?;
+ /// # let mut row_group = writer.next_row_group()?;
+ /// # let mut column = row_group.next_column()?.unwrap();
+ /// # column.typed::<Int32Type>().write_batch(&[42], None, None)?;
+ /// # column.close()?;
+ /// # row_group.close()?;
+ /// # let data = Bytes::from(writer.into_inner()?);
+ /// let metadata = ParquetMetaDataReader::new().parse_and_finish(&data)?;
+ /// let filter = ParquetMetaDataReader::read_column_bloom_filter(&data,
&metadata, 0, 0)?;
Review Comment:
❤️
##########
parquet/src/file/metadata/reader.rs:
##########
@@ -1233,6 +1308,25 @@ fn parse_page_index_data(
}
}
+fn bloom_filter_column(
Review Comment:
Should this be a method on `ParquetMetadata` ? It seems like it is a very
reasonable accessor
##########
parquet/src/file/metadata/reader.rs:
##########
@@ -801,6 +805,77 @@ impl ParquetMetaDataReader {
self.load_page_index_with_remainder_impl(fetch, None).await
}
+ /// Reads the Bloom filter for one physical leaf column in a row group.
+ ///
+ /// The metadata and reader must refer to the same file. `column_idx` is a
+ /// physical Parquet leaf index, not a top-level Arrow field index.
+ ///
+ /// Returns `Ok(None)` without I/O when the column chunk has no Bloom
filter.
+ /// Invalid coordinates, invalid Bloom metadata, malformed filters, and
read
+ /// failures return an error. The filter is returned separately; the
supplied
+ /// metadata is neither modified nor reloaded.
+ ///
+ /// Prefer reading filters after other pruning has narrowed the row groups
and
+ /// columns of interest. Footer and page index reads never load Bloom
filters.
+ ///
+ /// # Example
+ ///
+ /// Read footer metadata first, then request a filter only when needed:
+ ///
+ /// ```rust
+ /// use bytes::Bytes;
+ /// use parquet::errors::Result;
+ /// use parquet::file::metadata::ParquetMetaDataReader;
+ /// # use std::sync::Arc;
+ /// # use parquet::data_type::Int32Type;
+ /// # use parquet::file::properties::WriterProperties;
+ /// # use parquet::file::writer::SerializedFileWriter;
+ /// # use parquet::schema::parser::parse_message_type;
+ /// # fn main() -> Result<()> {
+ /// # let schema = Arc::new(parse_message_type("message schema { REQUIRED
INT32 value; }")?);
+ /// # let properties =
Arc::new(WriterProperties::builder().set_bloom_filter_enabled(true).build());
+ /// # let mut writer = SerializedFileWriter::new(Vec::new(), schema,
properties)?;
+ /// # let mut row_group = writer.next_row_group()?;
+ /// # let mut column = row_group.next_column()?.unwrap();
+ /// # column.typed::<Int32Type>().write_batch(&[42], None, None)?;
+ /// # column.close()?;
+ /// # row_group.close()?;
+ /// # let data = Bytes::from(writer.into_inner()?);
+ /// let metadata = ParquetMetaDataReader::new().parse_and_finish(&data)?;
+ /// let filter = ParquetMetaDataReader::read_column_bloom_filter(&data,
&metadata, 0, 0)?;
+ /// assert!(filter.unwrap().check(&42_i32));
+ /// # Ok(())
+ /// # }
+ /// ```
+ pub fn read_column_bloom_filter<R: ChunkReader>(
+ reader: &R,
+ metadata: &ParquetMetaData,
+ row_group_idx: usize,
+ column_idx: usize,
+ ) -> Result<Option<Sbbf>> {
+ let column = bloom_filter_column(metadata, row_group_idx, column_idx)?;
+ Sbbf::read_from_column_chunk(column, reader)
+ }
+
+ /// Asynchronous version of [`Self::read_column_bloom_filter`].
+ ///
+ /// Accepts a [`MetadataFetch`] byte source without requiring an Arrow
reader
Review Comment:
is "without requiring an Arrow reader builder or an
[`crate::arrow::async_reader::AsyncFileReader`] implementation." an important
detail 🤔
It seems like that is obvious from the signature
##########
parquet/tests/arrow_reader/bloom_filter/metadata.rs:
##########
@@ -0,0 +1,280 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: See the NOTICE file distributed with this work for
additional information regarding copyright ownership
+// SPDX-FileContributor: Licensed to the Apache Software Foundation (ASF)
under one or more contributor license agreements
+
+use super::{RecordingReader, column};
+use bytes::Bytes;
+use parquet::bloom_filter::Sbbf;
+use parquet::file::metadata::{
+ FileMetaData, ParquetMetaData, ParquetMetaDataReader, RowGroupMetaData,
+};
+use parquet::schema::{parser::parse_message_type, types::SchemaDescriptor};
+use std::sync::Arc;
+
+fn metadata(offset: Option<i64>, length: Option<i32>) -> ParquetMetaData {
+ let schema = Arc::new(SchemaDescriptor::new(Arc::new(
+ parse_message_type("message schema { REQUIRED BYTE_ARRAY value;
}").unwrap(),
+ )));
+ let row_group = RowGroupMetaData::builder(schema.clone())
+ .set_column_metadata(vec![column(offset, length)])
+ .build()
+ .unwrap();
+ ParquetMetaData::new(
+ FileMetaData::new(1, 0, None, None, schema, None),
+ vec![row_group],
+ )
+}
+
+#[test]
+fn sync_metadata_bloom_absence_and_invalid_requests_do_not_read() {
+ let reader = RecordingReader::default();
+ let absent = metadata(None, Some(-1));
+ assert!(
+ ParquetMetaDataReader::read_column_bloom_filter(&reader, &absent, 0, 0)
+ .unwrap()
+ .is_none()
+ );
+ for (row_group, column) in [(1, 0), (0, 1), (usize::MAX, 0), (0,
usize::MAX)] {
+ assert!(
+ ParquetMetaDataReader::read_column_bloom_filter(&reader, &absent,
row_group, column)
+ .is_err()
+ );
+ }
+ for invalid in [metadata(Some(-1), None), metadata(Some(0), Some(0))] {
+ assert!(ParquetMetaDataReader::read_column_bloom_filter(&reader,
&invalid, 0, 0).is_err());
+ }
+ assert!(reader.requests().is_empty());
+}
+
+#[test]
+fn sync_metadata_bloom_selects_requested_row_group() {
+ let mut original = Sbbf::new_with_num_of_bytes(32);
+ original.insert("present");
+ let mut bytes = Vec::new();
+ original.write(&mut bytes).unwrap();
+ let bytes = Bytes::from(bytes);
+ let absent = metadata(None, None);
+ let present = metadata(Some(9), Some(bytes.len() as i32));
+ let metadata = ParquetMetaData::new(
+ absent.file_metadata().clone(),
+ vec![absent.row_group(0).clone(), present.row_group(0).clone()],
+ );
+ let reader = RecordingReader::new([bytes.clone()]);
+ assert!(
+ ParquetMetaDataReader::read_column_bloom_filter(&reader, &metadata, 0,
0)
+ .unwrap()
+ .is_none()
+ );
+ assert!(reader.requests().is_empty());
+ let filter = ParquetMetaDataReader::read_column_bloom_filter(&reader,
&metadata, 1, 0)
+ .unwrap()
+ .unwrap();
+ assert!(filter.check("present"));
+ assert_eq!(reader.requests(), vec![9..9 + bytes.len() as u64]);
+}
+
+#[cfg(feature = "async")]
+mod r#async {
+ use super::super::encode;
+ use super::*;
+ use futures::future::{BoxFuture, ready};
+ use parquet::arrow::async_reader::{AsyncFileReader, MetadataFetch};
+ use parquet::errors::Result;
+ use std::ops::Range;
+
+ // A metadata-only byte source: no AsyncFileReader or metadata getter
needed.
+ struct FetchOnly<'a>(&'a RecordingReader);
+
+ impl MetadataFetch for FetchOnly<'_> {
+ fn fetch(&mut self, range: Range<u64>) -> BoxFuture<'_, Result<Bytes>>
{
+ Box::pin(ready(self.0.read(range)))
+ }
+ }
+
+ #[tokio::test]
Review Comment:
seems like a lot of boiler plate in this test -- I wonder if we can use
fixtures or something to reduce the replication
##########
parquet/tests/arrow_reader/bloom_filter/metadata.rs:
##########
@@ -0,0 +1,280 @@
+// SPDX-License-Identifier: Apache-2.0
+// SPDX-FileCopyrightText: See the NOTICE file distributed with this work for
additional information regarding copyright ownership
+// SPDX-FileContributor: Licensed to the Apache Software Foundation (ASF)
under one or more contributor license agreements
+
+use super::{RecordingReader, column};
+use bytes::Bytes;
+use parquet::bloom_filter::Sbbf;
+use parquet::file::metadata::{
+ FileMetaData, ParquetMetaData, ParquetMetaDataReader, RowGroupMetaData,
+};
+use parquet::schema::{parser::parse_message_type, types::SchemaDescriptor};
+use std::sync::Arc;
+
+fn metadata(offset: Option<i64>, length: Option<i32>) -> ParquetMetaData {
+ let schema = Arc::new(SchemaDescriptor::new(Arc::new(
+ parse_message_type("message schema { REQUIRED BYTE_ARRAY value;
}").unwrap(),
+ )));
+ let row_group = RowGroupMetaData::builder(schema.clone())
+ .set_column_metadata(vec![column(offset, length)])
+ .build()
+ .unwrap();
+ ParquetMetaData::new(
+ FileMetaData::new(1, 0, None, None, schema, None),
+ vec![row_group],
+ )
+}
+
+#[test]
+fn sync_metadata_bloom_absence_and_invalid_requests_do_not_read() {
+ let reader = RecordingReader::default();
+ let absent = metadata(None, Some(-1));
+ assert!(
+ ParquetMetaDataReader::read_column_bloom_filter(&reader, &absent, 0, 0)
+ .unwrap()
+ .is_none()
+ );
+ for (row_group, column) in [(1, 0), (0, 1), (usize::MAX, 0), (0,
usize::MAX)] {
+ assert!(
+ ParquetMetaDataReader::read_column_bloom_filter(&reader, &absent,
row_group, column)
+ .is_err()
+ );
+ }
+ for invalid in [metadata(Some(-1), None), metadata(Some(0), Some(0))] {
+ assert!(ParquetMetaDataReader::read_column_bloom_filter(&reader,
&invalid, 0, 0).is_err());
+ }
+ assert!(reader.requests().is_empty());
+}
+
+#[test]
+fn sync_metadata_bloom_selects_requested_row_group() {
+ let mut original = Sbbf::new_with_num_of_bytes(32);
+ original.insert("present");
+ let mut bytes = Vec::new();
+ original.write(&mut bytes).unwrap();
+ let bytes = Bytes::from(bytes);
+ let absent = metadata(None, None);
+ let present = metadata(Some(9), Some(bytes.len() as i32));
+ let metadata = ParquetMetaData::new(
+ absent.file_metadata().clone(),
+ vec![absent.row_group(0).clone(), present.row_group(0).clone()],
+ );
+ let reader = RecordingReader::new([bytes.clone()]);
+ assert!(
+ ParquetMetaDataReader::read_column_bloom_filter(&reader, &metadata, 0,
0)
+ .unwrap()
+ .is_none()
+ );
+ assert!(reader.requests().is_empty());
+ let filter = ParquetMetaDataReader::read_column_bloom_filter(&reader,
&metadata, 1, 0)
+ .unwrap()
+ .unwrap();
+ assert!(filter.check("present"));
+ assert_eq!(reader.requests(), vec![9..9 + bytes.len() as u64]);
+}
+
+#[cfg(feature = "async")]
+mod r#async {
+ use super::super::encode;
Review Comment:
There is a burrito place near us that sells "super super burritos" which
this made me think of
--
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]
