refactor(parquet): replace magic 8 literals with named constants

HippoBaro · HippoBaro · commit c891c35848f5 · 2026-04-08T17:10:12.000-04:00
The literal `8` appeared in two distinct roles throughout `RleEncoder`,
`RleDecoder`, and their tests. Replacing each with a named constant
makes the intent explicit and prevents the two meanings from being
confused.

`BIT_PACK_GROUP_SIZE = 8`
  The Parquet RLE/bit-packing hybrid format always bit-packs values in
  multiples of this count (spec: "we always bit-pack a multiple of 8
  values at a time"). Every occurrence related to the staging buffer
  size, the repeat-count threshold that triggers the RLE decision, and
  the group-count arithmetic in bit-packed headers now uses this name.

`u8::BITS` (= 8, from std)
  Used wherever a bit-count is divided by 8 to obtain a byte-count
  (e.g. `ceil(bit_width, u8::BITS as usize)`). This is a bits-per-byte
  conversion, a fundamentally different concept from the packing-group
  size.

No behaviour change.

Signed-off-by: Hippolyte Barraud &lt;hippolyte.barraud@datadoghq.com&gt;
diff --git a/parquet/src/arrow/arrow_writer/levels.rs b/parquet/src/arrow/arrow_writer/levels.rs
@@ -41,6 +41,7 @@
 //! \[1\] [parquet-format#nested-encoding](https://github.com/apache/parquet-format#nested-encoding)
 
 use crate::column::chunker::CdcChunk;
+use crate::column::writer::{LevelDataRef, ValueSelectionRef};
 use crate::errors::{ParquetError, Result};
 use arrow_array::cast::AsArray;
 use arrow_array::{Array, ArrayRef, OffsetSizeTrait};
@@ -822,6 +823,17 @@ impl LevelData {
         }
     }
 
+    pub(crate) fn as_ref(&self) -> LevelDataRef<'_> {
+        match self {
+            Self::Absent => LevelDataRef::Absent,
+            Self::Materialized(values) => LevelDataRef::Materialized(values),
+            Self::Uniform { value, count } => LevelDataRef::Uniform {
+                value: *value,
+                count: *count,
+            },
+        }
+    }
+
     pub(crate) fn slice(&self, offset: usize, len: usize) -> Self {
         match self {
             Self::Absent => Self::Absent,
@@ -919,6 +931,17 @@ impl PartialEq for ValueSelection {
 impl Eq for ValueSelection {}
 
 impl ValueSelection {
+    pub(crate) fn as_ref(&self) -> ValueSelectionRef<'_> {
+        match self {
+            Self::Empty => ValueSelectionRef::Empty,
+            Self::Dense { offset, len } => ValueSelectionRef::Dense {
+                offset: *offset,
+                len: *len,
+            },
+            Self::Sparse(indices) => ValueSelectionRef::Sparse(indices),
+        }
+    }
+
     fn from_indices(indices: Vec<usize>) -> Self {
         match (indices.first().copied(), indices.last().copied()) {
             (None, _) => Self::Empty,
diff --git a/parquet/src/arrow/arrow_writer/mod.rs b/parquet/src/arrow/arrow_writer/mod.rs
@@ -41,8 +41,7 @@ use crate::column::page::{CompressedPage, PageWriteSpec, PageWriter};
 use crate::column::page_encryption::PageEncryptor;
 use crate::column::writer::encoder::ColumnValueEncoder;
 use crate::column::writer::{
-    ColumnCloseResult, ColumnWriter, GenericColumnWriter, LevelDataRef, ValueSelectionRef,
-    get_column_writer,
+    ColumnCloseResult, ColumnWriter, GenericColumnWriter, ValueSelectionRef, get_column_writer,
 };
 use crate::data_type::{ByteArray, FixedLenByteArray};
 #[cfg(feature = "encryption")]
@@ -54,7 +53,7 @@ use crate::file::reader::{ChunkReader, Length};
 use crate::file::writer::{SerializedFileWriter, SerializedRowGroupWriter};
 use crate::parquet_thrift::{ThriftCompactOutputProtocol, WriteThrift};
 use crate::schema::types::{ColumnDescPtr, SchemaDescPtr, SchemaDescriptor};
-use levels::{ArrayLevels, LevelData, ValueSelection, calculate_array_levels};
+use levels::{ArrayLevels, calculate_array_levels};
 
 mod byte_array;
 mod levels;
@@ -888,28 +887,6 @@ enum ArrowColumnWriterImpl {
     Column(ColumnWriter<'static>),
 }
 
-fn level_data_ref(levels: &LevelData) -> LevelDataRef<'_> {
-    match levels {
-        LevelData::Absent => LevelDataRef::Absent,
-        LevelData::Materialized(values) => LevelDataRef::Materialized(values),
-        LevelData::Uniform { value, count } => LevelDataRef::Uniform {
-            value: *value,
-            count: *count,
-        },
-    }
-}
-
-fn value_selection_ref(selection: &ValueSelection) -> ValueSelectionRef<'_> {
-    match selection {
-        ValueSelection::Empty => ValueSelectionRef::Empty,
-        ValueSelection::Dense { offset, len } => ValueSelectionRef::Dense {
-            offset: *offset,
-            len: *len,
-        },
-        ValueSelection::Sparse(indices) => ValueSelectionRef::Sparse(indices),
-    }
-}
-
 impl ArrowColumnWriter {
     /// Write an [`ArrowLeafColumn`]
     pub fn write(&mut self, col: &ArrowLeafColumn) -> Result<()> {
@@ -924,8 +901,8 @@ impl ArrowColumnWriter {
     ) -> Result<()> {
         let levels = &col.0;
         let chunks = chunker.get_arrow_chunks(
-            level_data_ref(levels.def_level_data()),
-            level_data_ref(levels.rep_level_data()),
+            levels.def_level_data().as_ref(),
+            levels.rep_level_data().as_ref(),
             levels.array(),
         )?;
 
@@ -1396,8 +1373,8 @@ fn write_leaf(
                     offset: 0,
                     len: values.len(),
                 },
-                level_data_ref(levels.def_level_data()),
-                level_data_ref(levels.rep_level_data()),
+                levels.def_level_data().as_ref(),
+                levels.rep_level_data().as_ref(),
                 None,
                 None,
                 None,
@@ -1558,8 +1535,8 @@ fn write_leaf(
                     offset: 0,
                     len: bytes.len(),
                 },
-                level_data_ref(levels.def_level_data()),
-                level_data_ref(levels.rep_level_data()),
+                levels.def_level_data().as_ref(),
+                levels.rep_level_data().as_ref(),
                 None,
                 None,
                 None,
@@ -1575,9 +1552,9 @@ fn write_primitive<E: ColumnValueEncoder>(
 ) -> Result<usize> {
     writer.write_batch_internal(
         values,
-        value_selection_ref(levels.value_selection()),
-        level_data_ref(levels.def_level_data()),
-        level_data_ref(levels.rep_level_data()),
+        levels.value_selection().as_ref(),
+        levels.def_level_data().as_ref(),
+        levels.rep_level_data().as_ref(),
         None,
         None,
         None,
@@ -1591,9 +1568,9 @@ fn write_byte_array(
 ) -> Result<usize> {
     writer.write_batch_internal(
         values,
-        value_selection_ref(levels.value_selection()),
-        level_data_ref(levels.def_level_data()),
-        level_data_ref(levels.rep_level_data()),
+        levels.value_selection().as_ref(),
+        levels.def_level_data().as_ref(),
+        levels.rep_level_data().as_ref(),
         None,
         None,
         None,
diff --git a/parquet/src/encodings/rle.rs b/parquet/src/encodings/rle.rs
@@ -41,7 +41,12 @@ use bytes::Bytes;
 use crate::errors::{ParquetError, Result};
 use crate::util::bit_util::{self, BitReader, BitWriter, FromBitpacked};
 
-/// Maximum groups of 8 values per bit-packed run. Current value is 64.
+/// Number of values in one bit-packed group. The Parquet RLE/bit-packing hybrid
+/// format always bit-packs values in multiples of this count (see the format spec:
+/// "we always bit-pack a multiple of 8 values at a time").
+const BIT_PACK_GROUP_SIZE: usize = 8;
+
+/// Maximum groups of `BIT_PACK_GROUP_SIZE` values per bit-packed run. Current value is 64.
 const MAX_GROUPS_PER_BIT_PACKED_RUN: usize = 1 << 6;
 
 /// A RLE/Bit-Packing hybrid encoder.
@@ -54,9 +59,9 @@ pub struct RleEncoder {
     bit_writer: BitWriter,
 
     // Buffered values for bit-packed runs.
-    buffered_values: [u64; 8],
+    buffered_values: [u64; BIT_PACK_GROUP_SIZE],
 
-    // Number of current buffered values. Must be less than 8.
+    // Number of current buffered values. Must be less than BIT_PACK_GROUP_SIZE.
     num_buffered_values: usize,
 
     // The current (also last) value that was written and the count of how many
@@ -89,7 +94,7 @@ impl RleEncoder {
         RleEncoder {
             bit_width,
             bit_writer,
-            buffered_values: [0; 8],
+            buffered_values: [0; BIT_PACK_GROUP_SIZE],
             num_buffered_values: 0,
             current_value: 0,
             repeat_count: 0,
@@ -101,10 +106,10 @@ impl RleEncoder {
     /// Returns the maximum buffer size to encode `num_values` values with
     /// `bit_width`.
     pub fn max_buffer_size(bit_width: u8, num_values: usize) -> usize {
-        // The maximum size occurs with the shortest possible runs of 8
-        let num_runs = bit_util::ceil(num_values, 8);
+        // The maximum size occurs with the shortest possible runs of BIT_PACK_GROUP_SIZE
+        let num_runs = bit_util::ceil(num_values, BIT_PACK_GROUP_SIZE);
 
-        // The number of bytes in a run of 8
+        // The number of bytes in a run of BIT_PACK_GROUP_SIZE
         let bytes_per_run = bit_width as usize;
 
         // The maximum size if stored as shortest possible bit packed runs of 8
@@ -114,7 +119,8 @@ impl RleEncoder {
         let rle_len_prefix = 1;
 
         // The length of an RLE run of 8
-        let min_rle_run_size = rle_len_prefix + bit_util::ceil(bit_width as usize, 8);
+        let min_rle_run_size =
+            rle_len_prefix + bit_util::ceil(bit_width as usize, u8::BITS as usize);
 
         // The maximum size if stored as shortest possible RLE runs of 8
         let rle_max_size = num_runs * min_rle_run_size;
@@ -123,7 +129,7 @@ impl RleEncoder {
     }
 
     /// Returns `true` if the encoder is currently in RLE accumulation mode
-    /// for the given value (i.e., `repeat_count >= 8` and `current_value == value`).
+    /// for the given value (i.e., `repeat_count >= BIT_PACK_GROUP_SIZE` and `current_value == value`).
     ///
     /// The encoder enters accumulation mode as soon as the 8th consecutive identical
     /// value has been seen: at that point `flush_buffered_values` has committed the
@@ -132,7 +138,7 @@ impl RleEncoder {
     /// repetitions in O(1) once this returns `true`.
     #[inline]
     pub fn is_accumulating(&self, value: u64) -> bool {
-        self.repeat_count >= 8 && self.current_value == value
+        self.repeat_count >= BIT_PACK_GROUP_SIZE && self.current_value == value
     }
 
     /// Extends the current RLE run by `count` additional repetitions.
@@ -142,23 +148,24 @@ impl RleEncoder {
     /// returns `true` for the same value before calling this method.
     #[inline]
     pub fn extend_run(&mut self, count: usize) {
-        debug_assert!(self.repeat_count >= 8);
+        debug_assert!(self.repeat_count >= BIT_PACK_GROUP_SIZE);
         self.repeat_count += count;
     }
 
     /// Encodes `value`, which must be representable with `bit_width` bits.
     #[inline]
     pub fn put(&mut self, value: u64) {
-        // This function buffers 8 values at a time. After seeing 8 values, it
-        // decides whether the current run should be encoded in bit-packed or RLE.
+        // This function buffers BIT_PACK_GROUP_SIZE values at a time. After seeing that
+        // many values, it decides whether the current run should be encoded in bit-packed
+        // or RLE.
         if self.current_value == value {
             self.repeat_count += 1;
-            if self.repeat_count > 8 {
+            if self.repeat_count > BIT_PACK_GROUP_SIZE {
                 // A continuation of last value. No need to buffer.
                 return;
             }
         } else {
-            if self.repeat_count >= 8 {
+            if self.repeat_count >= BIT_PACK_GROUP_SIZE {
                 // The current RLE run has ended and we've gathered enough. Flush first.
                 debug_assert_eq!(self.bit_packed_count, 0);
                 self.flush_rle_run();
@@ -169,9 +176,9 @@ impl RleEncoder {
 
         self.buffered_values[self.num_buffered_values] = value;
         self.num_buffered_values += 1;
-        if self.num_buffered_values == 8 {
+        if self.num_buffered_values == BIT_PACK_GROUP_SIZE {
             // Buffered values are full. Flush them.
-            debug_assert_eq!(self.bit_packed_count % 8, 0);
+            debug_assert_eq!(self.bit_packed_count % BIT_PACK_GROUP_SIZE, 0);
             self.flush_buffered_values();
         }
     }
@@ -243,9 +250,9 @@ impl RleEncoder {
             if self.repeat_count > 0 && all_repeat {
                 self.flush_rle_run();
             } else {
-                // Buffer the last group of bit-packed values to 8 by padding with 0s.
+                // Buffer the last group of bit-packed values to BIT_PACK_GROUP_SIZE by padding with 0s.
                 if self.num_buffered_values > 0 {
-                    while self.num_buffered_values < 8 {
+                    while self.num_buffered_values < BIT_PACK_GROUP_SIZE {
                         self.buffered_values[self.num_buffered_values] = 0;
                         self.num_buffered_values += 1;
                     }
@@ -263,7 +270,7 @@ impl RleEncoder {
         self.bit_writer.put_vlq_int(indicator_value as u64);
         self.bit_writer.put_aligned(
             self.current_value,
-            bit_util::ceil(self.bit_width as usize, 8),
+            bit_util::ceil(self.bit_width as usize, u8::BITS as usize),
         );
         self.num_buffered_values = 0;
         self.repeat_count = 0;
@@ -281,7 +288,7 @@ impl RleEncoder {
         self.num_buffered_values = 0;
         if update_indicator_byte {
             // Write the indicator byte to the reserved position in `bit_writer`
-            let num_groups = self.bit_packed_count / 8;
+            let num_groups = self.bit_packed_count / BIT_PACK_GROUP_SIZE;
             let indicator_byte = ((num_groups << 1) | 1) as u8;
             self.bit_writer
                 .put_aligned_offset(indicator_byte, 1, self.indicator_byte_pos as usize);
@@ -291,19 +298,19 @@ impl RleEncoder {
     }
 
     fn flush_buffered_values(&mut self) {
-        if self.repeat_count >= 8 {
+        if self.repeat_count >= BIT_PACK_GROUP_SIZE {
             self.num_buffered_values = 0;
             if self.bit_packed_count > 0 {
                 // In this case we choose RLE encoding. Flush the current buffered values
                 // as bit-packed encoding.
-                debug_assert_eq!(self.bit_packed_count % 8, 0);
+                debug_assert_eq!(self.bit_packed_count % BIT_PACK_GROUP_SIZE, 0);
                 self.flush_bit_packed_run(true)
             }
             return;
         }
 
         self.bit_packed_count += self.num_buffered_values;
-        let num_groups = self.bit_packed_count / 8;
+        let num_groups = self.bit_packed_count / BIT_PACK_GROUP_SIZE;
         if num_groups + 1 >= MAX_GROUPS_PER_BIT_PACKED_RUN {
             // We've reached the maximum value that can be hold in a single bit-packed
             // run.
@@ -584,10 +591,10 @@ impl RleDecoder {
                 return Ok(false);
             }
             if indicator_value & 1 == 1 {
-                self.bit_packed_left = ((indicator_value >> 1) * 8) as u32;
+                self.bit_packed_left = ((indicator_value >> 1) * BIT_PACK_GROUP_SIZE as i64) as u32;
             } else {
                 self.rle_left = (indicator_value >> 1) as u32;
-                let value_width = bit_util::ceil(self.bit_width as usize, 8);
+                let value_width = bit_util::ceil(self.bit_width as usize, u8::BITS as usize);
                 self.current_value = bit_reader.get_aligned::<u64>(value_width);
                 self.current_value.ok_or_else(|| {
                     general_err!("parquet_data_error: not enough data for RLE decoding")
@@ -869,7 +876,7 @@ mod tests {
                 &values[..],
                 width as u8,
                 None,
-                2 * (1 + bit_util::ceil(width as i64, 8) as i32),
+                2 * (1 + bit_util::ceil(width as i64, u8::BITS as i64) as i32),
             );
         }
 
@@ -879,9 +886,12 @@ mod tests {
         for i in 0..101 {
             values.push(i % 2);
         }
-        let num_groups = bit_util::ceil(100, 8) as u8;
+        let num_groups = bit_util::ceil(100, BIT_PACK_GROUP_SIZE) as u8;
         expected_buffer.push((num_groups << 1) | 1);
-        expected_buffer.resize(expected_buffer.len() + 100 / 8, 0b10101010);
+        expected_buffer.resize(
+            expected_buffer.len() + 100 / BIT_PACK_GROUP_SIZE,
+            0b10101010,
+        );
 
         // For the last 4 0 and 1's, padded with 0.
         expected_buffer.push(0b00001010);
@@ -892,12 +902,12 @@ mod tests {
             1 + num_groups as i32,
         );
         for width in 2..MAX_WIDTH + 1 {
-            let num_values = bit_util::ceil(100, 8) * 8;
+            let num_values = bit_util::ceil(100, BIT_PACK_GROUP_SIZE) * BIT_PACK_GROUP_SIZE;
             validate_rle(
                 &values,
                 width as u8,
                 None,
-                1 + bit_util::ceil(width as i64 * num_values, 8) as i32,
+                1 + bit_util::ceil(width as i64 * num_values as i64, u8::BITS as i64) as i32,
             );
         }
     }
@@ -1058,7 +1068,7 @@ mod tests {
         let num_values = 2002;
 
         // bit-packed header
-        let run_bytes = ceil(num_values * bit_width, 8) as u64;
+        let run_bytes = ceil(num_values * bit_width, u8::BITS as usize) as u64;
         writer.put_vlq_int((run_bytes << 1) | 1);
         for _ in 0..run_bytes {
             writer.put_aligned(0xFF_u8, 1);
