Repository: apache/arrow-rs
Pull request #6479 (2024-09-28 13:40:20)

Add better documentation, examples and builer-style API to `ByteView`
# Which issue does this PR close?

<!--
We generally require a GitHub issue to be filed for all bug fixes and enhancements and this helps us generate change logs for our releases. You can link an issue to this PR using the GitHub syntax. For example `Closes #123` indicates that this PR will close issue #123.
-->

Closes https://github.com/apache/arrow-rs/issues/6478

# Rationale for this change
 
# Rationale for this change

While working with StringViewArray's downstream in DataFusion, I found it was not always clear that `ByteView` exists for creating/destructuring with the `u128`s so I wanted to write some additional documentation
I also found it quite awkward to write the examples (specifically setting the prefix) so I added some BuilderStyle APIs as well

This is what creating ByteViews looks like without the builder APIs :vomit:

```rust
let view = ByteView {
   length: 12,
   // convert 4 byte &str to u32
   prefix: u32::from_le_bytes(prefix.as_bytes().try_into().unwrap()),
   buffer_index: 3,
   offset: 42,
};
```

# What changes are included in this PR?
1. Add pointer to `GenericByteViewArray` from docs
2. Add examples of using `ByteView`
3. Add builder style APIs to `ByteView`

<!--
There is no need to duplicate the description in the issue here but it is sometimes worth providing a summary of the individual changes in this PR.
-->

# Are there any user-facing changes?

There are several new functions on `ByteView`

alamb: FYI @XiangpengHao 
alamb: Thank you for the review @tustvold 

Patch:
diff --git a/arrow-array/src/array/byte_view_array.rs b/arrow-array/src/array/byte_view_array.rs
index c53478d8b..b1b558057 100644
--- a/arrow-array/src/array/byte_view_array.rs
+++ b/arrow-array/src/array/byte_view_array.rs
@@ -44,8 +44,11 @@ use super::ByteArrayType;
 ///
 /// # See Also
 ///
-/// See [`StringViewArray`] for storing utf8 encoded string data and
-/// [`BinaryViewArray`] for storing bytes.
+/// * [`StringViewArray`] for storing utf8 encoded string data
+/// * [`BinaryViewArray`] for storing bytes
+/// * [`ByteView`] to interpret `u128`s layout of the views.
+///
+/// [`ByteView`]: arrow_data::ByteView
 ///
 /// # Notes
 ///
@@ -872,12 +875,9 @@ mod tests {
     #[should_panic(expected = "Invalid buffer index at 0: got index 3 but only has 1 buffers")]
     fn new_with_invalid_view_data() {
         let v = "large payload over 12 bytes";
-        let view = ByteView {
-            length: 13,
-            prefix: u32::from_le_bytes(v.as_bytes()[0..4].try_into().unwrap()),
-            buffer_index: 3,
-            offset: 1,
-        };
+        let view = ByteView::new(13, &v.as_bytes()[0..4])
+            .with_buffer_index(3)
+            .with_offset(1);
         let views = ScalarBuffer::from(vec![view.into()]);
         let buffers = vec![Buffer::from_slice_ref(v)];
         StringViewArray::new(views, buffers, None);
@@ -888,13 +888,12 @@ mod tests {
         expected = "Encountered non-UTF-8 data at index 0: invalid utf-8 sequence of 1 bytes from index 0"
     )]
     fn new_with_invalid_utf8_data() {
-        let v: Vec<u8> = vec![0xf0, 0x80, 0x80, 0x80];
-        let view = ByteView {
-            length: v.len() as u32,
-            prefix: u32::from_le_bytes(v[0..4].try_into().unwrap()),
-            buffer_index: 0,
-            offset: 0,
-        };
+        let v: Vec<u8> = vec![
+            // invalid UTF8
+            0xf0, 0x80, 0x80, 0x80, // more bytes to make it larger than 12
+            0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00,
+        ];
+        let view = ByteView::new(v.len() as u32, &v[0..4]);
         let views = ScalarBuffer::from(vec![view.into()]);
         let buffers = vec![Buffer::from_slice_ref(v)];
         StringViewArray::new(views, buffers, None);
diff --git a/arrow-data/src/byte_view.rs b/arrow-data/src/byte_view.rs
index 6f6d6d175..3b3ec6246 100644
--- a/arrow-data/src/byte_view.rs
+++ b/arrow-data/src/byte_view.rs
@@ -21,8 +21,40 @@ use arrow_schema::ArrowError;
 /// Helper to access views of [`GenericByteViewArray`] (`StringViewArray` and
 /// `BinaryViewArray`) where the length is greater than 12 bytes.
 ///
-/// See the documentation on [`GenericByteViewArray`] for more information on
-/// the layout of the views.
+/// See Also:
+/// * [`GenericByteViewArray`] for more information on the layout of the views.
+/// * [`validate_binary_view`] and [`validate_string_view`] to validate
+///
+/// # Example: Create a new u128 view
+///
+/// ```rust
+/// # use arrow_data::ByteView;;
+/// // Create a view for a string of length 20
+/// // first four bytes are "Rust"
+/// // stored in buffer 3
+/// // at offset 42
+/// let prefix = "Rust";
+/// let view = ByteView::new(20, prefix.as_bytes())
+///   .with_buffer_index(3)
+///   .with_offset(42);
+///
+/// // create the final u128
+/// let v = view.as_u128();
+/// assert_eq!(v, 0x2a000000037473755200000014);
+/// ```
+///
+/// # Example: decode a `u128` into its constituent fields
+/// ```rust
+/// # use arrow_data::ByteView;
+/// // Convert a u128 to a ByteView
+/// // See validate_{string,binary}_view functions to validate
+/// let v = ByteView::from(0x2a000000037473755200000014);
+///
+/// assert_eq!(v.length, 20);
+/// assert_eq!(v.prefix, 0x74737552);
+/// assert_eq!(v.buffer_index, 3);
+/// assert_eq!(v.offset, 42);
+/// ```
 ///
 /// [`GenericByteViewArray`]: https://docs.rs/arrow/latest/arrow/array/struct.GenericByteViewArray.html
 #[derive(Debug, Copy, Clone, Default)]
@@ -39,6 +71,42 @@ pub struct ByteView {
 }
 
 impl ByteView {
+    /// Construct a [`ByteView`] for data `length` of bytes with the specified prefix.
+    ///
+    /// See example on [`ByteView`] docs
+    ///
+    /// Notes:
+    /// * the length should always be greater than 12 (Data less than 12
+    ///   bytes is stored as an inline view)
+    /// * buffer and offset are set to `0`
+    ///
+    /// # Panics
+    /// If the prefix is not exactly 4 bytes
+    #[inline]
+    pub fn new(length: u32, prefix: &[u8]) -> Self {
+        debug_assert!(length > 12);
+        Self {
+            length,
+            prefix: u32::from_le_bytes(prefix.try_into().unwrap()),
+            buffer_index: 0,
+            offset: 0,
+        }
+    }
+
+    /// Set the [`Self::buffer_index`] field
+    #[inline]
+    pub fn with_buffer_index(mut self, buffer_index: u32) -> Self {
+        self.buffer_index = buffer_index;
+        self
+    }
+
+    /// Set the [`Self::offset`] field
+    #[inline]
+    pub fn with_offset(mut self, offset: u32) -> Self {
+        self.offset = offset;
+        self
+    }
+
     #[inline(always)]
     /// Convert `ByteView` to `u128` by concatenating the fields
     pub fn as_u128(self) -> u128 {
diff --git a/parquet/src/arrow/arrow_reader/statistics.rs b/parquet/src/arrow/arrow_reader/statistics.rs
index 4c0dac05c..8a7511be2 100644
--- a/parquet/src/arrow/arrow_reader/statistics.rs
+++ b/parquet/src/arrow/arrow_reader/statistics.rs
@@ -1159,7 +1159,7 @@ where
 ///
 /// # Schemas
 ///
-/// The converter ues the schema of the Parquet file and the Arrow schema to
+/// The converter uses the schema of the Parquet file and the Arrow schema to
 /// convert the underlying statistics value (stored as a parquet value) into the
 /// corresponding Arrow value. For example, Decimals are stored as binary in
 /// parquet files and this structure handles mapping them to the `i128`
@@ -1175,6 +1175,8 @@ pub struct StatisticsConverter<'a> {
     parquet_column_index: Option<usize>,
     /// The field (with data type) of the column in the Arrow schema
     arrow_field: &'a Field,
+    /// treat missing null_counts as 0 nulls
+    missing_null_counts_as_zero: bool,
 }
 
 impl<'a> StatisticsConverter<'a> {
@@ -1191,6 +1193,23 @@ impl<'a> StatisticsConverter<'a> {
         self.arrow_field
     }
 
+    /// Set the statistics converter to treat missing null counts as missing
+    ///
+    /// By default, the converter will treat missing null counts as though
+    /// the null count is known to be `0`.
+    ///
+    /// Note that parquet files written by parquet-rs currently do not store
+    /// null counts even when it is known there are zero nulls, and the reader
+    /// will return 0 for the null counts in that instance. This behavior may
+    /// change in a future release.
+    ///
+    /// Both parquet-java and parquet-cpp store null counts as 0 when there are
+    /// no nulls, and don't write unknown values to the null count field.
+    pub fn with_missing_null_counts_as_zero(mut self, missing_null_counts_as_zero: bool) -> Self {
+        self.missing_null_counts_as_zero = missing_null_counts_as_zero;
+        self
+    }
+
     /// Returns a [`UInt64Array`] with row counts for each row group
     ///
     /// # Return Value
@@ -1284,6 +1303,7 @@ impl<'a> StatisticsConverter<'a> {
         Ok(Self {
             parquet_column_index: parquet_index,
             arrow_field,
+            missing_null_counts_as_zero: true,
         })
     }
 
@@ -1382,7 +1402,15 @@ impl<'a> StatisticsConverter<'a> {
         let null_counts = metadatas
             .into_iter()
             .map(|x| x.column(parquet_index).statistics())
-            .map(|s| s.and_then(|s| s.null_count_opt()));
+            .map(|s| {
+                s.and_then(|s| {
+                    if self.missing_null_counts_as_zero {
+                        Some(s.null_count_opt().unwrap_or(0))
+                    } else {
+                        s.null_count_opt()
+                    }
+                })
+            });
         Ok(UInt64Array::from_iter(null_counts))
     }
 
@@ -1593,3 +1621,5 @@ impl<'a> StatisticsConverter<'a> {
         new_null_array(data_type, num_row_groups)
     }
 }
+
+// See tests in parquet/tests/arrow_reader/statistics.rs
diff --git a/parquet/src/file/metadata/reader.rs b/parquet/src/file/metadata/reader.rs
index 3fd2bd76f..88f53d02f 100644
--- a/parquet/src/file/metadata/reader.rs
+++ b/parquet/src/file/metadata/reader.rs
@@ -841,7 +841,7 @@ mod async_tests {
 
     struct MetadataFetchFn<F>(F);
 
-    impl<'a, F, Fut> MetadataFetch for &'a mut MetadataFetchFn<F>
+    impl<F, Fut> MetadataFetch for MetadataFetchFn<F>
     where
         F: FnMut(Range<usize>) -> Fut + Send,
         Fut: Future<Output = Result<Bytes>> + Send,
@@ -870,14 +870,14 @@ mod async_tests {
         let expected = expected.file_metadata().schema();
         let fetch_count = AtomicUsize::new(0);
 
-        let fetch = |range| {
+        let mut fetch = |range| {
             fetch_count.fetch_add(1, Ordering::SeqCst);
             futures::future::ready(read_range(&mut file, range))
         };
 
-        let mut f = MetadataFetchFn(fetch);
+        let input = MetadataFetchFn(&mut fetch);
         let actual = ParquetMetaDataReader::new()
-            .load_and_finish(&mut f, len)
+            .load_and_finish(input, len)
             .await
             .unwrap();
         assert_eq!(actual.file_metadata().schema(), expected);
@@ -885,9 +885,10 @@ mod async_tests {
 
         // Metadata hint too small - below footer size
         fetch_count.store(0, Ordering::SeqCst);
+        let input = MetadataFetchFn(&mut fetch);
         let actual = ParquetMetaDataReader::new()
             .with_prefetch_hint(Some(7))
-            .load_and_finish(&mut f, len)
+            .load_and_finish(input, len)
             .await
             .unwrap();
         assert_eq!(actual.file_metadata().schema(), expected);
@@ -895,9 +896,10 @@ mod async_tests {
 
         // Metadata hint too small
         fetch_count.store(0, Ordering::SeqCst);
+        let input = MetadataFetchFn(&mut fetch);
         let actual = ParquetMetaDataReader::new()
             .with_prefetch_hint(Some(10))
-            .load_and_finish(&mut f, len)
+            .load_and_finish(input, len)
             .await
             .unwrap();
         assert_eq!(actual.file_metadata().schema(), expected);
@@ -905,9 +907,10 @@ mod async_tests {
 
         // Metadata hint too large
         fetch_count.store(0, Ordering::SeqCst);
+        let input = MetadataFetchFn(&mut fetch);
         let actual = ParquetMetaDataReader::new()
             .with_prefetch_hint(Some(500))
-            .load_and_finish(&mut f, len)
+            .load_and_finish(input, len)
             .await
             .unwrap();
         assert_eq!(actual.file_metadata().schema(), expected);
@@ -915,23 +918,26 @@ mod async_tests {
 
         // Metadata hint exactly correct
         fetch_count.store(0, Ordering::SeqCst);
+        let input = MetadataFetchFn(&mut fetch);
         let actual = ParquetMetaDataReader::new()
             .with_prefetch_hint(Some(428))
-            .load_and_finish(&mut f, len)
+            .load_and_finish(input, len)
             .await
             .unwrap();
         assert_eq!(actual.file_metadata().schema(), expected);
         assert_eq!(fetch_count.load(Ordering::SeqCst), 1);
 
+        let input = MetadataFetchFn(&mut fetch);
         let err = ParquetMetaDataReader::new()
-            .load_and_finish(&mut f, 4)
+            .load_and_finish(input, 4)
             .await
             .unwrap_err()
             .to_string();
         assert_eq!(err, "EOF: file size of 4 is less than footer");
 
+        let input = MetadataFetchFn(&mut fetch);
         let err = ParquetMetaDataReader::new()
-            .load_and_finish(&mut f, 20)
+            .load_and_finish(input, 20)
             .await
             .unwrap_err()
             .to_string();
@@ -948,39 +954,42 @@ mod async_tests {
             futures::future::ready(read_range(&mut file, range))
         };
 
-        let mut f = MetadataFetchFn(&mut fetch);
+        let f = MetadataFetchFn(&mut fetch);
         let mut loader = ParquetMetaDataReader::new().with_page_indexes(true);
-        loader.try_load(&mut f, len).await.unwrap();
+        loader.try_load(f, len).await.unwrap();
         assert_eq!(fetch_count.load(Ordering::SeqCst), 3);
         let metadata = loader.finish().unwrap();
         assert!(metadata.offset_index().is_some() && metadata.column_index().is_some());
 
         // Prefetch just footer exactly
         fetch_count.store(0, Ordering::SeqCst);
+        let f = MetadataFetchFn(&mut fetch);
         let mut loader = ParquetMetaDataReader::new()
             .with_page_indexes(true)
             .with_prefetch_hint(Some(1729));
-        loader.try_load(&mut f, len).await.unwrap();
+        loader.try_load(f, len).await.unwrap();
         assert_eq!(fetch_count.load(Ordering::SeqCst), 2);
         let metadata = loader.finish().unwrap();
         assert!(metadata.offset_index().is_some() && metadata.column_index().is_some());
 
         // Prefetch more than footer but not enough
         fetch_count.store(0, Ordering::SeqCst);
+        let f = MetadataFetchFn(&mut fetch);
         let mut loader = ParquetMetaDataReader::new()
             .with_page_indexes(true)
             .with_prefetch_hint(Some(130649));
-        loader.try_load(&mut f, len).await.unwrap();
+        loader.try_load(f, len).await.unwrap();
         assert_eq!(fetch_count.load(Ordering::SeqCst), 2);
         let metadata = loader.finish().unwrap();
         assert!(metadata.offset_index().is_some() && metadata.column_index().is_some());
 
         // Prefetch exactly enough
         fetch_count.store(0, Ordering::SeqCst);
+        let f = MetadataFetchFn(&mut fetch);
         let metadata = ParquetMetaDataReader::new()
             .with_page_indexes(true)
             .with_prefetch_hint(Some(130650))
-            .load_and_finish(&mut f, len)
+            .load_and_finish(f, len)
             .await
             .unwrap();
         assert_eq!(fetch_count.load(Ordering::SeqCst), 1);
diff --git a/parquet/src/file/reader.rs b/parquet/src/file/reader.rs
index cff921b20..d8a61fafe 100644
--- a/parquet/src/file/reader.rs
+++ b/parquet/src/file/reader.rs
@@ -45,25 +45,38 @@ pub trait Length {
     fn len(&self) -> u64;
 }
 
-/// The ChunkReader trait generates readers of chunks of a source.
+/// Generates [`Read`]ers to read chunks of a Parquet data source.
 ///
-/// For more information see [`File::try_clone`]
+/// The Parquet reader uses [`ChunkReader`] to access Parquet data, allowing
+/// multiple decoders to read concurrently from different locations in the same file.
+///
+/// The trait provides:
+/// * random access (via [`Self::get_bytes`])
+/// * sequential (via [`Self::get_read`])
+///
+/// # Provided Implementations
+/// * [`File`] for reading from local file system
+/// * [`Bytes`] for reading from an in-memory buffer
+///
+/// User provided implementations can implement more sophisticated behaviors
+/// such as on-demand buffering or scan sharing.
 pub trait ChunkReader: Length + Send + Sync {
+    /// The concrete type of readers returned by this trait
     type T: Read;
 
-    /// Get a [`Read`] starting at the provided file offset
+    /// Get a [`Read`] instance starting at the provided file offset
     ///
-    /// Subsequent or concurrent calls to [`Self::get_read`] or [`Self::get_bytes`] may
-    /// side-effect on previously returned [`Self::T`]. Care should be taken to avoid this
-    ///
-    /// See [`File::try_clone`] for more information
+    /// Returned readers follow the model of [`File::try_clone`] where mutations
+    /// of one reader affect all readers. Thus subsequent or concurrent calls to
+    /// [`Self::get_read`] or [`Self::get_bytes`] may cause side-effects on
+    /// previously returned readers. Callers of `get_read` should take care
+    /// to avoid race conditions.
     fn get_read(&self, start: u64) -> Result<Self::T>;
 
-    /// Get a range as bytes
-    ///
-    /// Concurrent calls to [`Self::get_bytes`] may result in interleaved output
+    /// Get a range of data in memory as [`Bytes`]
     ///
-    /// See [`File::try_clone`] for more information
+    /// Similarly to [`Self::get_read`], this method may have side-effects on
+    /// previously returned readers.
     fn get_bytes(&self, start: u64, length: usize) -> Result<Bytes>;
 }
 


Test patch:
diff --git a/parquet/tests/arrow_reader/statistics.rs b/parquet/tests/arrow_reader/statistics.rs
index 384be83d3..29adbbb4f 100644
--- a/parquet/tests/arrow_reader/statistics.rs
+++ b/parquet/tests/arrow_reader/statistics.rs
@@ -22,6 +22,7 @@ use std::default::Default;
 use std::fs::File;
 use std::sync::Arc;
 
+use super::make_test_file_rg;
 use super::{struct_array, Scenario};
 use arrow::compute::kernels::cast_utils::Parser;
 use arrow::datatypes::{
@@ -37,16 +38,17 @@ use arrow_array::{
     TimestampMillisecondArray, TimestampNanosecondArray, TimestampSecondArray, UInt16Array,
     UInt32Array, UInt64Array, UInt8Array,
 };
-use arrow_schema::{DataType, Field, Schema, TimeUnit};
+use arrow_schema::{DataType, Field, Schema, SchemaRef, TimeUnit};
 use half::f16;
 use parquet::arrow::arrow_reader::statistics::StatisticsConverter;
 use parquet::arrow::arrow_reader::{
     ArrowReaderBuilder, ArrowReaderOptions, ParquetRecordBatchReaderBuilder,
 };
 use parquet::arrow::ArrowWriter;
+use parquet::file::metadata::{ColumnChunkMetaData, RowGroupMetaData};
 use parquet::file::properties::{EnabledStatistics, WriterProperties};
-
-use super::make_test_file_rg;
+use parquet::file::statistics::{Statistics, ValueStatistics};
+use parquet::schema::types::{SchemaDescPtr, SchemaDescriptor};
 
 #[derive(Debug, Default, Clone)]
 struct Int64Case {
@@ -2139,6 +2141,65 @@ async fn test_missing_statistics() {
     .run();
 }
 
+#[test]
+fn missing_null_counts_as_zero() {
+    let min = None;
+    let max = None;
+    let distinct_count = None;
+    let null_count = None; // NB: no null count
+    let is_min_max_deprecated = false;
+    let stats = Statistics::Boolean(ValueStatistics::new(
+        min,
+        max,
+        distinct_count,
+        null_count,
+        is_min_max_deprecated,
+    ));
+    let (arrow_schema, parquet_schema) = bool_arrow_and_parquet_schema();
+
+    let column_chunk = ColumnChunkMetaData::builder(parquet_schema.column(0))
+        .set_statistics(stats)
+        .build()
+        .unwrap();
+    let metadata = RowGroupMetaData::builder(parquet_schema.clone())
+        .set_column_metadata(vec![column_chunk])
+        .build()
+        .unwrap();
+
+    let converter = StatisticsConverter::try_new("b", &arrow_schema, &parquet_schema).unwrap();
+
+    // by default null count should be 0
+    assert_eq!(
+        converter.row_group_null_counts([&metadata]).unwrap(),
+        UInt64Array::from_iter(vec![Some(0)])
+    );
+
+    // if we disable missing null counts as zero flag null count will be None
+    let converter = converter.with_missing_null_counts_as_zero(false);
+    assert_eq!(
+        converter.row_group_null_counts([&metadata]).unwrap(),
+        UInt64Array::from_iter(vec![None])
+    );
+}
+
+/// return an Arrow schema and corresponding Parquet SchemaDescriptor for
+/// a schema with a single boolean column "b"
+fn bool_arrow_and_parquet_schema() -> (SchemaRef, SchemaDescPtr) {
+    let arrow_schema = Arc::new(Schema::new(vec![Field::new("b", DataType::Boolean, true)]));
+    use parquet::schema::types::Type as ParquetType;
+    let parquet_schema = ParquetType::group_type_builder("schema")
+        .with_fields(vec![Arc::new(
+            ParquetType::primitive_type_builder("a", parquet::basic::Type::INT32)
+                .build()
+                .unwrap(),
+        )])
+        .build()
+        .unwrap();
+
+    let parquet_schema = Arc::new(SchemaDescriptor::new(Arc::new(parquet_schema)));
+    (arrow_schema, parquet_schema)
+}
+
 /////// NEGATIVE TESTS ///////
 // column not found
 #[tokio::test]