diff --git a/crates/ourios-parquet/src/audit_reader.rs b/crates/ourios-parquet/src/audit_reader.rs index 418b7336f..f37560b5c 100644 --- a/crates/ourios-parquet/src/audit_reader.rs +++ b/crates/ourios-parquet/src/audit_reader.rs @@ -25,6 +25,9 @@ //! non-breaking for readers, and folds defined over named kinds //! ignore unknown rows by construction. +// The shared accessor family (epic #745 wave 2). +#[allow(clippy::wildcard_imports)] +use crate::decode::*; use std::fmt; use std::fs::File; use std::io; @@ -32,7 +35,7 @@ use std::path::{Path, PathBuf}; use std::time::{Duration, SystemTime}; use arrow_array::cast::AsArray; -use arrow_array::types::{Int32Type, TimestampNanosecondType, UInt8Type, UInt32Type, UInt64Type}; +use arrow_array::types::Int32Type; use arrow_array::{Array, RecordBatch, StructArray}; use ourios_core::alias::ActorId; use ourios_core::audit::{AuditEvent, AuditPayload, ParamType, SlotExpansion, TemplateChange}; @@ -116,18 +119,7 @@ impl AuditReader { let builder = ParquetRecordBatchReaderBuilder::try_new(reader).map_err(AuditReaderError::Parquet)?; - let file_schema = builder.schema(); - for expected_field in crate::audit_schema().fields() { - if !expected_field.is_nullable() - && file_schema - .column_with_name(expected_field.name()) - .is_none() - { - return Err(AuditReaderError::MissingRequiredColumn { - name: expected_field.name().clone(), - }); - } - } + require_baseline_columns(builder.schema(), &crate::audit_schema())?; let inner = builder.build().map_err(AuditReaderError::Parquet)?; @@ -329,7 +321,11 @@ fn batch_to_audit_events( let new_template = optional_string(batch, audit_columns::NEW_TEMPLATE)?.unwrap_or_default(); let positions_widened_lists = decode_positions_column(batch, row_offset)?; let slots_expanded_lists = decode_slots_column(batch, row_offset)?; - let triggering_line_hash = optional_fixed_bytes16(batch, audit_columns::TRIGGERING_LINE_HASH)?; + // The shared accessor models an absent column as `None`; this + // decode's index-guarded readers want the audit reader's historical + // empty-vec shape. + let triggering_line_hash = + optional_fixed_bytes16(batch, audit_columns::TRIGGERING_LINE_HASH)?.unwrap_or_default(); let triggering_line_sample = optional_string(batch, audit_columns::TRIGGERING_LINE_SAMPLE)?.unwrap_or_default(); let reason = optional_string(batch, audit_columns::REASON)?.unwrap_or_default(); @@ -1084,296 +1080,6 @@ fn decode_param_type(ord: i32) -> ParamType { // variants — the §3.9 "internal index convention" CodeRabbit // flagged on the first round. -fn required_string( - batch: &RecordBatch, - name: &'static str, - row_offset: usize, -) -> Result, AuditReaderError> { - let col = required_column(batch, name)?; - let arr = crate::reader::StrCol::try_new(col).ok_or_else(|| AuditReaderError::Conversion { - column: name, - detail: format!( - "expected Utf8/Utf8View string array, got {:?}", - col.data_type() - ), - })?; - let mut out = Vec::with_capacity(arr.len()); - for i in 0..arr.len() { - let Some(v) = arr.get(i) else { - return Err(AuditReaderError::Conversion { - column: name, - detail: format!("row {}: null on a REQUIRED column", row_offset + i), - }); - }; - out.push(v.to_string()); - } - Ok(out) -} - -fn required_u8( - batch: &RecordBatch, - name: &'static str, - row_offset: usize, -) -> Result, AuditReaderError> { - let col = required_column(batch, name)?; - let arr = col - .as_primitive_opt::() - .ok_or_else(|| AuditReaderError::Conversion { - column: name, - detail: format!("expected UInt8Array, got {:?}", col.data_type()), - })?; - materialize_required_primitive(arr, name, row_offset) -} - -fn required_timestamp( - batch: &RecordBatch, - name: &'static str, - row_offset: usize, -) -> Result, AuditReaderError> { - let col = required_column(batch, name)?; - let arr = col - .as_primitive_opt::() - .ok_or_else(|| AuditReaderError::Conversion { - column: name, - detail: format!( - "expected TimestampNanosecondArray, got {:?}", - col.data_type() - ), - })?; - let mut out = Vec::with_capacity(arr.len()); - for i in 0..arr.len() { - if arr.is_null(i) { - return Err(AuditReaderError::Conversion { - column: name, - detail: format!("row {}: null on a REQUIRED column", row_offset + i), - }); - } - out.push(arr.value(i)); - } - Ok(out) -} - -fn materialize_required_primitive( - arr: &arrow_array::PrimitiveArray, - name: &'static str, - row_offset: usize, -) -> Result, AuditReaderError> { - if arr.null_count() == 0 { - return Ok(arr.values().to_vec()); - } - for i in 0..arr.len() { - if arr.is_null(i) { - return Err(AuditReaderError::Conversion { - column: name, - detail: format!("row {}: null on a REQUIRED column", row_offset + i), - }); - } - } - Ok(arr.values().to_vec()) -} - -fn required_column<'a>( - batch: &'a RecordBatch, - name: &'static str, -) -> Result<&'a dyn Array, AuditReaderError> { - let idx = - batch - .schema() - .index_of(name) - .map_err(|_| AuditReaderError::MissingRequiredColumn { - name: name.to_string(), - })?; - Ok(batch.column(idx).as_ref()) -} - -fn optional_string( - batch: &RecordBatch, - name: &'static str, -) -> Result>>, AuditReaderError> { - let Some(col) = optional_column(batch, name) else { - return Ok(None); - }; - let arr = crate::reader::StrCol::try_new(col).ok_or_else(|| AuditReaderError::Conversion { - column: name, - detail: format!( - "expected Utf8/Utf8View string array, got {:?}", - col.data_type() - ), - })?; - let mut out = Vec::with_capacity(arr.len()); - for i in 0..arr.len() { - out.push(arr.get(i).map(str::to_string)); - } - Ok(Some(out)) -} - -fn optional_column<'a>(batch: &'a RecordBatch, name: &'static str) -> Option<&'a dyn Array> { - let idx = batch.schema().index_of(name).ok()?; - Some(batch.column(idx).as_ref()) -} - -/// Per-row `Option` for a nullable `UInt64` column (the §3.7 -/// `template_id` / `compaction_generation` / `compaction_rows` -/// columns). -fn optional_u64( - batch: &RecordBatch, - name: &'static str, -) -> Result>, AuditReaderError> { - let Some(col) = optional_column(batch, name) else { - return Ok(Vec::new()); - }; - let arr = col - .as_primitive_opt::() - .ok_or_else(|| AuditReaderError::Conversion { - column: name, - detail: format!("expected UInt64Array, got {:?}", col.data_type()), - })?; - Ok((0..arr.len()) - .map(|i| (!arr.is_null(i)).then(|| arr.value(i))) - .collect()) -} - -/// Per-row `Option` for a nullable `UInt32` column. -fn optional_u32( - batch: &RecordBatch, - name: &'static str, -) -> Result>, AuditReaderError> { - let Some(col) = optional_column(batch, name) else { - return Ok(Vec::new()); - }; - let arr = col - .as_primitive_opt::() - .ok_or_else(|| AuditReaderError::Conversion { - column: name, - detail: format!("expected UInt32Array, got {:?}", col.data_type()), - })?; - Ok((0..arr.len()) - .map(|i| (!arr.is_null(i)).then(|| arr.value(i))) - .collect()) -} - -/// Per-row `Option<[u8; 16]>` for the nullable `triggering_line_hash` -/// `FixedSizeBinary(16)` column. -fn optional_fixed_bytes16( - batch: &RecordBatch, - name: &'static str, -) -> Result>, AuditReaderError> { - let Some(col) = optional_column(batch, name) else { - return Ok(Vec::new()); - }; - let arr = col - .as_fixed_size_binary_opt() - .ok_or_else(|| AuditReaderError::Conversion { - column: name, - detail: format!("expected FixedSizeBinaryArray, got {:?}", col.data_type()), - })?; - let mut out = Vec::with_capacity(arr.len()); - for i in 0..arr.len() { - if arr.is_null(i) { - out.push(None); - } else { - let mut buf = [0u8; 16]; - buf.copy_from_slice(arr.value(i)); - out.push(Some(buf)); - } - } - Ok(out) -} - -/// Per-row `Option>` for the nullable -/// `compaction_input_files` `LIST` column. The element field -/// is non-nullable, so a NULL element is a corrupt row. -fn optional_string_list( - batch: &RecordBatch, - name: &'static str, -) -> Result>>, AuditReaderError> { - let Some(col) = optional_column(batch, name) else { - return Ok(Vec::new()); - }; - let list = col - .as_list_opt::() - .ok_or_else(|| AuditReaderError::Conversion { - column: name, - detail: "column is not a LIST as declared".to_string(), - })?; - let mut out = Vec::with_capacity(list.len()); - for row_idx in 0..list.len() { - if list.is_null(row_idx) { - out.push(None); - continue; - } - let elements = list.value(row_idx); - let strs = crate::reader::StrCol::try_new(elements.as_ref()).ok_or_else(|| { - AuditReaderError::Conversion { - column: name, - detail: "list element is not Utf8/Utf8View".to_string(), - } - })?; - let mut row = Vec::with_capacity(strs.len()); - for i in 0..strs.len() { - let Some(v) = strs.get(i) else { - return Err(AuditReaderError::Conversion { - column: name, - detail: format!( - "batch row {row_idx} element {i}: NULL but the element field is non-nullable", - ), - }); - }; - row.push(v.to_string()); - } - out.push(Some(row)); - } - Ok(out) -} - -/// Per-row `Option>` for the nullable `alias_member_ids` -/// `LIST` column. NULL list ⇒ `None` (not an alias row); -/// empty list ⇒ `Some(vec![])` — the §3.7 empty-vs-NULL distinction. -/// The element field is non-nullable, so a NULL element is a corrupt -/// row. -fn optional_u64_list( - batch: &RecordBatch, - name: &'static str, -) -> Result>>, AuditReaderError> { - let Some(col) = optional_column(batch, name) else { - return Ok(Vec::new()); - }; - let list = col - .as_list_opt::() - .ok_or_else(|| AuditReaderError::Conversion { - column: name, - detail: "column is not a LIST as declared".to_string(), - })?; - let mut out = Vec::with_capacity(list.len()); - for row_idx in 0..list.len() { - if list.is_null(row_idx) { - out.push(None); - continue; - } - let elements = list.value(row_idx); - let ids = elements.as_primitive_opt::().ok_or_else(|| { - AuditReaderError::Conversion { - column: name, - detail: "list element is not UInt64".to_string(), - } - })?; - let mut row = Vec::with_capacity(ids.len()); - for i in 0..ids.len() { - if ids.is_null(i) { - return Err(AuditReaderError::Conversion { - column: name, - detail: format!( - "batch row {row_idx} element {i}: NULL but the element field is non-nullable", - ), - }); - } - row.push(ids.value(i)); - } - out.push(Some(row)); - } - Ok(out) -} - #[cfg(test)] mod tests { //! Colocated unit tests for the audit-reader paths that the diff --git a/crates/ourios-parquet/src/decode.rs b/crates/ourios-parquet/src/decode.rs new file mode 100644 index 000000000..bf2b00e2d --- /dev/null +++ b/crates/ourios-parquet/src/decode.rs @@ -0,0 +1,643 @@ +//! The shared column-accessor family both parquet readers decode +//! through (epic #745 wave 2): one implementation of every +//! `required_*` / `optional_*` accessor, the view-tolerant +//! [`StrCol`]/[`BinCol`] wrappers, and the §3.8/§3.9 baseline-column +//! check — generic over a [`DecodeError`] so `ReaderError` and +//! `AuditReaderError` keep their own variants and messages. Before +//! this module the two readers carried line-for-line twins, and a fix +//! landing on one twin only produced a live divergence (the audit +//! `Utf8View` gap, #746). + +use arrow_array::cast::AsArray; +use arrow_array::types::{Float32Type, TimestampNanosecondType, UInt8Type, UInt32Type, UInt64Type}; +use arrow_array::{Array, BinaryViewArray, RecordBatch, StringViewArray}; + +use crate::audit_reader::AuditReaderError; +use crate::reader::ReaderError; + +/// A string column in either arrow representation. The parquet reader +/// yields plain `Utf8`; `DataFusion` (the querier's scan path) yields +/// `Utf8View` by default — one decoder serves both (RFC 0021 §3.1). +/// `pub(crate)` so the audit reader's accessors share it rather than +/// re-growing the plain-`Utf8`-only shape this replaced (epic #745 +/// wave 0: the audit path had exactly that divergence). +pub(crate) enum StrCol<'a> { + Plain(&'a arrow_array::StringArray), + View(&'a StringViewArray), +} + +impl<'a> StrCol<'a> { + pub(crate) fn try_new(col: &'a dyn Array) -> Option { + if let Some(a) = col.as_string_opt::() { + return Some(Self::Plain(a)); + } + col.as_string_view_opt().map(Self::View) + } + + pub(crate) fn len(&self) -> usize { + match self { + Self::Plain(a) => a.len(), + Self::View(a) => a.len(), + } + } + + pub(crate) fn get(&self, i: usize) -> Option<&str> { + match self { + Self::Plain(a) => (!a.is_null(i)).then(|| a.value(i)), + Self::View(a) => (!a.is_null(i)).then(|| a.value(i)), + } + } +} + +/// [`StrCol`]'s binary counterpart (`Binary` | `BinaryView`). +pub(crate) enum BinCol<'a> { + Plain(&'a arrow_array::BinaryArray), + View(&'a BinaryViewArray), +} + +impl<'a> BinCol<'a> { + pub(crate) fn try_new(col: &'a dyn Array) -> Option { + if let Some(a) = col.as_binary_opt::() { + return Some(Self::Plain(a)); + } + col.as_binary_view_opt().map(Self::View) + } + + pub(crate) fn len(&self) -> usize { + match self { + Self::Plain(a) => a.len(), + Self::View(a) => a.len(), + } + } + + pub(crate) fn get(&self, i: usize) -> Option<&[u8]> { + match self { + Self::Plain(a) => (!a.is_null(i)).then(|| a.value(i)), + Self::View(a) => (!a.is_null(i)).then(|| a.value(i)), + } + } +} + +/// RFC 0005 §3.9 (the reader contract; §3.8 gives the policy): +/// every REQUIRED (non-nullable) column of +/// `expected` must be present in the file's schema, else a hard +/// error. The one copy of the baseline-schema invariant — every +/// constructor of both readers (data and audit) enforces it here. +pub(crate) fn require_baseline_columns( + file_schema: &arrow_schema::Schema, + expected: &arrow_schema::Schema, +) -> Result<(), E> { + for expected_field in expected.fields() { + if !expected_field.is_nullable() + && file_schema + .column_with_name(expected_field.name()) + .is_none() + { + return Err(E::missing_required(expected_field.name())); + } + } + Ok(()) +} + +/// The constructor surface a reader error must offer the shared +/// accessors. Message text is built here once, so the two readers +/// cannot drift apart on diagnostics again. +pub(crate) trait DecodeError: Sized { + /// A §3.8/§3.9 baseline REQUIRED column absent from the file. + fn missing_required(name: &str) -> Self; + /// A column present but shape-mismatched (type, null-on-required, + /// list nesting). + fn conversion(column: &'static str, detail: String) -> Self; +} + +impl DecodeError for ReaderError { + fn missing_required(name: &str) -> Self { + Self::MissingRequiredColumn { + name: name.to_string(), + } + } + fn conversion(column: &'static str, detail: String) -> Self { + Self::Conversion { column, detail } + } +} + +impl DecodeError for AuditReaderError { + fn missing_required(name: &str) -> Self { + Self::MissingRequiredColumn { + name: name.to_string(), + } + } + fn conversion(column: &'static str, detail: String) -> Self { + Self::Conversion { column, detail } + } +} + +pub(crate) fn required_string( + batch: &RecordBatch, + name: &'static str, + row_offset: usize, +) -> Result, E> { + let col = required_column(batch, name)?; + let arr = StrCol::try_new(col).ok_or_else(|| { + E::conversion( + name, + format!( + "expected Utf8/Utf8View string array, got {:?}", + col.data_type() + ), + ) + })?; + let mut out = Vec::with_capacity(arr.len()); + for i in 0..arr.len() { + let Some(v) = arr.get(i) else { + return Err(E::conversion( + name, + format!("row {}: null on a REQUIRED column", row_offset + i), + )); + }; + out.push(v.to_string()); + } + Ok(out) +} + +pub(crate) fn required_u64( + batch: &RecordBatch, + name: &'static str, + row_offset: usize, +) -> Result, E> { + let col = required_column(batch, name)?; + let arr = col.as_primitive_opt::().ok_or_else(|| { + E::conversion( + name, + format!("expected UInt64Array, got {:?}", col.data_type()), + ) + })?; + materialize_required_primitive(arr, name, row_offset) +} + +pub(crate) fn required_u32( + batch: &RecordBatch, + name: &'static str, + row_offset: usize, +) -> Result, E> { + let col = required_column(batch, name)?; + let arr = col.as_primitive_opt::().ok_or_else(|| { + E::conversion( + name, + format!("expected UInt32Array, got {:?}", col.data_type()), + ) + })?; + materialize_required_primitive(arr, name, row_offset) +} + +pub(crate) fn required_u8( + batch: &RecordBatch, + name: &'static str, + row_offset: usize, +) -> Result, E> { + let col = required_column(batch, name)?; + let arr = col.as_primitive_opt::().ok_or_else(|| { + E::conversion( + name, + format!("expected UInt8Array, got {:?}", col.data_type()), + ) + })?; + materialize_required_primitive(arr, name, row_offset) +} + +pub(crate) fn required_f32( + batch: &RecordBatch, + name: &'static str, + row_offset: usize, +) -> Result, E> { + let col = required_column(batch, name)?; + let arr = col.as_primitive_opt::().ok_or_else(|| { + E::conversion( + name, + format!("expected Float32Array, got {:?}", col.data_type()), + ) + })?; + materialize_required_primitive(arr, name, row_offset) +} + +/// Materialise a primitive Arrow array into `Vec`, +/// erroring on any NULL slot. Plain `arr.values().to_vec()` would +/// silently turn NULL slots into zero (the underlying primitive +/// buffer's default fill), masking file corruption. Fast-paths +/// the null-free case so the common path is still a single +/// buffer copy. +pub(crate) fn materialize_required_primitive< + T: arrow_array::types::ArrowPrimitiveType, + E: DecodeError, +>( + arr: &arrow_array::PrimitiveArray, + name: &'static str, + row_offset: usize, +) -> Result, E> { + if arr.null_count() == 0 { + return Ok(arr.values().to_vec()); + } + for i in 0..arr.len() { + if arr.is_null(i) { + return Err(E::conversion( + name, + format!("row {}: null on a REQUIRED column", row_offset + i), + )); + } + } + // Validity buffer reported nulls but no row matched — + // unreachable in practice. + Ok(arr.values().to_vec()) +} + +pub(crate) fn required_bool( + batch: &RecordBatch, + name: &'static str, + row_offset: usize, +) -> Result, E> { + let col = required_column(batch, name)?; + let arr = col.as_boolean_opt().ok_or_else(|| { + E::conversion( + name, + format!("expected BooleanArray, got {:?}", col.data_type()), + ) + })?; + let mut out = Vec::with_capacity(arr.len()); + for i in 0..arr.len() { + if arr.is_null(i) { + return Err(E::conversion( + name, + format!("row {}: null on a REQUIRED column", row_offset + i), + )); + } + out.push(arr.value(i)); + } + Ok(out) +} + +pub(crate) fn required_timestamp( + batch: &RecordBatch, + name: &'static str, + row_offset: usize, +) -> Result, E> { + let col = required_column(batch, name)?; + let arr = col + .as_primitive_opt::() + .ok_or_else(|| { + E::conversion( + name, + format!( + "expected TimestampNanosecondArray, got {:?}", + col.data_type() + ), + ) + })?; + let mut out = Vec::with_capacity(arr.len()); + for i in 0..arr.len() { + if arr.is_null(i) { + return Err(E::conversion( + name, + format!("row {}: null on a REQUIRED column", row_offset + i), + )); + } + out.push(arr.value(i)); + } + Ok(out) +} + +pub(crate) fn required_column<'a, E: DecodeError>( + batch: &'a RecordBatch, + name: &'static str, +) -> Result<&'a dyn Array, E> { + let idx = batch + .schema() + .index_of(name) + .map_err(|_| E::missing_required(name))?; + Ok(batch.column(idx).as_ref()) +} + +pub(crate) fn optional_string( + batch: &RecordBatch, + name: &'static str, +) -> Result>>, E> { + let Some(col) = optional_column(batch, name) else { + return Ok(None); + }; + let arr = StrCol::try_new(col).ok_or_else(|| { + E::conversion( + name, + format!( + "expected Utf8/Utf8View string array, got {:?}", + col.data_type() + ), + ) + })?; + let mut out = Vec::with_capacity(arr.len()); + for i in 0..arr.len() { + out.push(arr.get(i).map(ToString::to_string)); + } + Ok(Some(out)) +} + +pub(crate) fn optional_binary( + batch: &RecordBatch, + name: &'static str, +) -> Result>>>, E> { + let Some(col) = optional_column(batch, name) else { + return Ok(None); + }; + let arr = BinCol::try_new(col).ok_or_else(|| { + E::conversion( + name, + format!( + "expected Binary/BinaryView array, got {:?}", + col.data_type() + ), + ) + })?; + let mut out = Vec::with_capacity(arr.len()); + for i in 0..arr.len() { + out.push(arr.get(i).map(<[u8]>::to_vec)); + } + Ok(Some(out)) +} + +pub(crate) fn optional_timestamp( + batch: &RecordBatch, + name: &'static str, +) -> Result>>, E> { + let Some(col) = optional_column(batch, name) else { + return Ok(None); + }; + let arr = col + .as_primitive_opt::() + .ok_or_else(|| { + E::conversion( + name, + format!( + "expected TimestampNanosecondArray, got {:?}", + col.data_type() + ), + ) + })?; + let mut out = Vec::with_capacity(arr.len()); + for i in 0..arr.len() { + out.push(if arr.is_null(i) { + None + } else { + Some(arr.value(i)) + }); + } + Ok(Some(out)) +} + +pub(crate) fn optional_fixed_bytes16( + batch: &RecordBatch, + name: &'static str, +) -> Result>>, E> { + optional_fixed_bytes::<16, E>(batch, name) +} + +pub(crate) fn optional_fixed_bytes8( + batch: &RecordBatch, + name: &'static str, +) -> Result>>, E> { + optional_fixed_bytes::<8, E>(batch, name) +} + +pub(crate) fn optional_fixed_bytes( + batch: &RecordBatch, + name: &'static str, +) -> Result>>, E> { + let Some(col) = optional_column(batch, name) else { + return Ok(None); + }; + let arr = col.as_fixed_size_binary_opt().ok_or_else(|| { + E::conversion( + name, + format!("expected FixedSizeBinaryArray, got {:?}", col.data_type()), + ) + })?; + if usize::try_from(arr.value_length()).ok() != Some(N) { + return Err(E::conversion( + name, + format!( + "expected FixedSizeBinary({N}), got FixedSizeBinary({})", + arr.value_length(), + ), + )); + } + let mut out = Vec::with_capacity(arr.len()); + for i in 0..arr.len() { + out.push(if arr.is_null(i) { + None + } else { + let slice = arr.value(i); + let mut buf = [0u8; N]; + buf.copy_from_slice(slice); + Some(buf) + }); + } + Ok(Some(out)) +} + +pub(crate) fn optional_column<'a>( + batch: &'a RecordBatch, + name: &'static str, +) -> Option<&'a dyn Array> { + let idx = batch.schema().index_of(name).ok()?; + Some(batch.column(idx).as_ref()) +} + +/// Per-row `Option` for a nullable `UInt64` column (the §3.7 +/// `template_id` / `compaction_generation` / `compaction_rows` +/// columns). +pub(crate) fn optional_u64( + batch: &RecordBatch, + name: &'static str, +) -> Result>, E> { + let Some(col) = optional_column(batch, name) else { + return Ok(Vec::new()); + }; + let arr = col.as_primitive_opt::().ok_or_else(|| { + E::conversion( + name, + format!("expected UInt64Array, got {:?}", col.data_type()), + ) + })?; + Ok((0..arr.len()) + .map(|i| (!arr.is_null(i)).then(|| arr.value(i))) + .collect()) +} + +/// Per-row `Option` for a nullable `UInt32` column. +pub(crate) fn optional_u32( + batch: &RecordBatch, + name: &'static str, +) -> Result>, E> { + let Some(col) = optional_column(batch, name) else { + return Ok(Vec::new()); + }; + let arr = col.as_primitive_opt::().ok_or_else(|| { + E::conversion( + name, + format!("expected UInt32Array, got {:?}", col.data_type()), + ) + })?; + Ok((0..arr.len()) + .map(|i| (!arr.is_null(i)).then(|| arr.value(i))) + .collect()) +} + +/// Per-row `Option>` for the nullable +/// `compaction_input_files` `LIST` column. The element field +/// is non-nullable, so a NULL element is a corrupt row. +pub(crate) fn optional_string_list( + batch: &RecordBatch, + name: &'static str, +) -> Result>>, E> { + let Some(col) = optional_column(batch, name) else { + return Ok(Vec::new()); + }; + let list = col.as_list_opt::().ok_or_else(|| { + E::conversion(name, "column is not a LIST as declared".to_string()) + })?; + let mut out = Vec::with_capacity(list.len()); + for row_idx in 0..list.len() { + if list.is_null(row_idx) { + out.push(None); + continue; + } + let elements = list.value(row_idx); + let strs = StrCol::try_new(elements.as_ref()) + .ok_or_else(|| E::conversion(name, "list element is not Utf8/Utf8View".to_string()))?; + let mut row = Vec::with_capacity(strs.len()); + for i in 0..strs.len() { + let Some(v) = strs.get(i) else { + return Err(E::conversion( + name, + format!( + "batch row {row_idx} element {i}: NULL but the element field is non-nullable", + ), + )); + }; + row.push(v.to_string()); + } + out.push(Some(row)); + } + Ok(out) +} + +/// Per-row `Option>` for the nullable `alias_member_ids` +/// `LIST` column. NULL list ⇒ `None` (not an alias row); +/// empty list ⇒ `Some(vec![])` — the §3.7 empty-vs-NULL distinction. +/// The element field is non-nullable, so a NULL element is a corrupt +/// row. +pub(crate) fn optional_u64_list( + batch: &RecordBatch, + name: &'static str, +) -> Result>>, E> { + let Some(col) = optional_column(batch, name) else { + return Ok(Vec::new()); + }; + let list = col.as_list_opt::().ok_or_else(|| { + E::conversion(name, "column is not a LIST as declared".to_string()) + })?; + let mut out = Vec::with_capacity(list.len()); + for row_idx in 0..list.len() { + if list.is_null(row_idx) { + out.push(None); + continue; + } + let elements = list.value(row_idx); + let ids = elements + .as_primitive_opt::() + .ok_or_else(|| E::conversion(name, "list element is not UInt64".to_string()))?; + let mut row = Vec::with_capacity(ids.len()); + for i in 0..ids.len() { + if ids.is_null(i) { + return Err(E::conversion( + name, + format!( + "batch row {row_idx} element {i}: NULL but the element field is non-nullable", + ), + )); + } + row.push(ids.value(i)); + } + out.push(Some(row)); + } + Ok(out) +} + +#[cfg(test)] +mod tests { + use super::*; + use arrow_array::{Int64Array, StringArray, UInt64Array}; + use arrow_schema::{DataType, Field, Schema}; + use std::sync::Arc; + + fn batch(fields: Vec, cols: Vec>) -> RecordBatch { + RecordBatch::try_new(Arc::new(Schema::new(fields)), cols).expect("batch") + } + + /// The one baseline-presence check errors identically through + /// either reader's error type. + #[test] + fn baseline_check_names_the_missing_column_for_both_errors() { + let expected = Schema::new(vec![Field::new("must", DataType::UInt64, false)]); + let file = Schema::new(vec![Field::new("other", DataType::UInt64, false)]); + let r: Result<(), ReaderError> = require_baseline_columns(&file, &expected); + assert!(matches!(r, Err(ReaderError::MissingRequiredColumn { name }) if name == "must")); + let a: Result<(), AuditReaderError> = require_baseline_columns(&file, &expected); + assert!( + matches!(a, Err(AuditReaderError::MissingRequiredColumn { name }) if name == "must") + ); + let ok: Result<(), ReaderError> = require_baseline_columns(&expected, &expected); + assert!(ok.is_ok()); + } + + /// NULL on a REQUIRED column is a per-row Conversion error naming + /// the file-global row index. + #[test] + fn required_accessors_reject_null_with_the_offset_row_index() { + let b = batch( + vec![Field::new("s", DataType::Utf8, true)], + vec![Arc::new(StringArray::from(vec![Some("a"), None]))], + ); + let r: Result, ReaderError> = required_string(&b, "s", 10); + assert!( + matches!(r, Err(ReaderError::Conversion { column: "s", detail }) if detail.contains("row 11")), + ); + } + + /// §3.9 absent-OPTIONAL handling: `optional_string` models absence + /// as `None`; the audit-shaped list/scalar accessors keep their + /// historical empty-vec shape. + #[test] + fn absent_optional_columns_decode_to_their_documented_shapes() { + let b = batch( + vec![Field::new("present", DataType::Int64, false)], + vec![Arc::new(Int64Array::from(vec![1_i64]))], + ); + let s: Result>>, ReaderError> = optional_string(&b, "absent"); + assert!(matches!(s, Ok(None))); + let u: Result>, ReaderError> = optional_u64(&b, "absent"); + assert!(matches!(u, Ok(v) if v.is_empty())); + } + + /// A shape mismatch is a Conversion error naming the column and + /// the found type — identical text through either error. + #[test] + fn type_mismatch_is_a_conversion_error() { + let b = batch( + vec![Field::new("n", DataType::UInt64, false)], + vec![Arc::new(UInt64Array::from(vec![7_u64]))], + ); + let r: Result, ReaderError> = required_string(&b, "n", 0); + assert!(matches!( + r, + Err(ReaderError::Conversion { column: "n", .. }) + )); + } +} diff --git a/crates/ourios-parquet/src/lib.rs b/crates/ourios-parquet/src/lib.rs index 193fcc9c1..fd90f3f6e 100644 --- a/crates/ourios-parquet/src/lib.rs +++ b/crates/ourios-parquet/src/lib.rs @@ -30,6 +30,7 @@ pub mod audit_record_batch; pub mod audit_sink; pub mod audit_writer; pub mod compaction; +mod decode; pub mod manifest; pub mod partition; pub mod promoted; diff --git a/crates/ourios-parquet/src/reader.rs b/crates/ourios-parquet/src/reader.rs index bc016b002..527da9609 100644 --- a/crates/ourios-parquet/src/reader.rs +++ b/crates/ourios-parquet/src/reader.rs @@ -32,10 +32,13 @@ use std::io; use std::path::{Path, PathBuf}; use arrow_array::cast::AsArray; -use arrow_array::types::{ - Float32Type, Int32Type, TimestampNanosecondType, UInt8Type, UInt32Type, UInt64Type, -}; -use arrow_array::{Array, BinaryViewArray, RecordBatch, StringViewArray, StructArray}; + +// The shared accessor family (epic #745 wave 2) — one implementation +// for both readers, generic over the error type. +#[allow(clippy::wildcard_imports)] +use crate::decode::*; +use arrow_array::types::Int32Type; +use arrow_array::{Array, RecordBatch, StructArray}; use ourios_core::audit::ParamType; use ourios_core::record::{BodyKind, MinedRecord, Param}; use ourios_core::tenant::TenantId; @@ -147,7 +150,7 @@ impl Reader { fn from_bytes(bytes: bytes::Bytes, file_path: PathBuf) -> Result { let builder = ParquetRecordBatchReaderBuilder::try_new(bytes).map_err(ReaderError::Parquet)?; - require_baseline_columns(builder.schema())?; + require_baseline_columns(builder.schema(), &crate::data_schema())?; let inner = builder.build().map_err(ReaderError::Parquet)?; Ok(Self { @@ -172,7 +175,7 @@ impl Reader { })?; let builder = ParquetRecordBatchReaderBuilder::try_new(file).map_err(ReaderError::Parquet)?; - require_baseline_columns(builder.schema())?; + require_baseline_columns(builder.schema(), &crate::data_schema())?; let inner = builder.build().map_err(ReaderError::Parquet)?; Ok(Self { inner, @@ -367,69 +370,6 @@ pub enum ShapeValidation { Skip, } -/// A string column in either arrow representation. The parquet reader -/// yields plain `Utf8`; `DataFusion` (the querier's scan path) yields -/// `Utf8View` by default — one decoder serves both (RFC 0021 §3.1). -/// `pub(crate)` so the audit reader's accessors share it rather than -/// re-growing the plain-`Utf8`-only shape this replaced (epic #745 -/// wave 0: the audit path had exactly that divergence). -pub(crate) enum StrCol<'a> { - Plain(&'a arrow_array::StringArray), - View(&'a StringViewArray), -} - -impl<'a> StrCol<'a> { - pub(crate) fn try_new(col: &'a dyn Array) -> Option { - if let Some(a) = col.as_string_opt::() { - return Some(Self::Plain(a)); - } - col.as_string_view_opt().map(Self::View) - } - - pub(crate) fn len(&self) -> usize { - match self { - Self::Plain(a) => a.len(), - Self::View(a) => a.len(), - } - } - - pub(crate) fn get(&self, i: usize) -> Option<&str> { - match self { - Self::Plain(a) => (!a.is_null(i)).then(|| a.value(i)), - Self::View(a) => (!a.is_null(i)).then(|| a.value(i)), - } - } -} - -/// [`StrCol`]'s binary counterpart (`Binary` | `BinaryView`). -enum BinCol<'a> { - Plain(&'a arrow_array::BinaryArray), - View(&'a BinaryViewArray), -} - -impl<'a> BinCol<'a> { - fn try_new(col: &'a dyn Array) -> Option { - if let Some(a) = col.as_binary_opt::() { - return Some(Self::Plain(a)); - } - col.as_binary_view_opt().map(Self::View) - } - - fn len(&self) -> usize { - match self { - Self::Plain(a) => a.len(), - Self::View(a) => a.len(), - } - } - - fn get(&self, i: usize) -> Option<&[u8]> { - match self { - Self::Plain(a) => (!a.is_null(i)).then(|| a.value(i)), - Self::View(a) => (!a.is_null(i)).then(|| a.value(i)), - } - } -} - fn validate_row_vs_partition( record: &MinedRecord, expected: &PartitionKey, @@ -488,24 +428,6 @@ fn store_io_err(op: &'static str, path: &Path, err: StoreError) -> ReaderError { } } -/// RFC 0005 §3.9: every baseline REQUIRED (non-nullable) column must be -/// present in the file's schema, else a hard error. Shared by -/// [`Reader::open_file`] and [`Reader::open_bytes`]. -fn require_baseline_columns(file_schema: &arrow_schema::Schema) -> Result<(), ReaderError> { - for expected_field in crate::data_schema().fields() { - if !expected_field.is_nullable() - && file_schema - .column_with_name(expected_field.name()) - .is_none() - { - return Err(ReaderError::MissingRequiredColumn { - name: expected_field.name().clone(), - }); - } - } - Ok(()) -} - /// Convert one Arrow `RecordBatch` to a `Vec` per /// RFC 0005 §3.2. Handles the §3.9 "missing OPTIONAL column → /// `None`" rule by checking column presence before unpacking. @@ -933,185 +855,6 @@ fn decode_separators_column( // (OPTIONAL columns — `None` when the file's schema doesn't // carry the column, per §3.9's missing-column carve-out). -fn required_string( - batch: &RecordBatch, - name: &'static str, - row_offset: usize, -) -> Result, ReaderError> { - let col = required_column(batch, name)?; - let arr = StrCol::try_new(col).ok_or_else(|| ReaderError::Conversion { - column: name, - detail: format!( - "expected Utf8/Utf8View string array, got {:?}", - col.data_type() - ), - })?; - let mut out = Vec::with_capacity(arr.len()); - for i in 0..arr.len() { - let Some(v) = arr.get(i) else { - return Err(ReaderError::Conversion { - column: name, - detail: format!("row {}: null on a REQUIRED column", row_offset + i), - }); - }; - out.push(v.to_string()); - } - Ok(out) -} - -fn required_u64( - batch: &RecordBatch, - name: &'static str, - row_offset: usize, -) -> Result, ReaderError> { - let col = required_column(batch, name)?; - let arr = col - .as_primitive_opt::() - .ok_or_else(|| ReaderError::Conversion { - column: name, - detail: format!("expected UInt64Array, got {:?}", col.data_type()), - })?; - materialize_required_primitive(arr, name, row_offset) -} - -fn required_u32( - batch: &RecordBatch, - name: &'static str, - row_offset: usize, -) -> Result, ReaderError> { - let col = required_column(batch, name)?; - let arr = col - .as_primitive_opt::() - .ok_or_else(|| ReaderError::Conversion { - column: name, - detail: format!("expected UInt32Array, got {:?}", col.data_type()), - })?; - materialize_required_primitive(arr, name, row_offset) -} - -fn required_u8( - batch: &RecordBatch, - name: &'static str, - row_offset: usize, -) -> Result, ReaderError> { - let col = required_column(batch, name)?; - let arr = col - .as_primitive_opt::() - .ok_or_else(|| ReaderError::Conversion { - column: name, - detail: format!("expected UInt8Array, got {:?}", col.data_type()), - })?; - materialize_required_primitive(arr, name, row_offset) -} - -fn required_f32( - batch: &RecordBatch, - name: &'static str, - row_offset: usize, -) -> Result, ReaderError> { - let col = required_column(batch, name)?; - let arr = col - .as_primitive_opt::() - .ok_or_else(|| ReaderError::Conversion { - column: name, - detail: format!("expected Float32Array, got {:?}", col.data_type()), - })?; - materialize_required_primitive(arr, name, row_offset) -} - -/// Materialise a primitive Arrow array into `Vec`, -/// erroring on any NULL slot. Plain `arr.values().to_vec()` would -/// silently turn NULL slots into zero (the underlying primitive -/// buffer's default fill), masking file corruption. Fast-paths -/// the null-free case so the common path is still a single -/// buffer copy. -fn materialize_required_primitive( - arr: &arrow_array::PrimitiveArray, - name: &'static str, - row_offset: usize, -) -> Result, ReaderError> { - if arr.null_count() == 0 { - return Ok(arr.values().to_vec()); - } - for i in 0..arr.len() { - if arr.is_null(i) { - return Err(ReaderError::Conversion { - column: name, - detail: format!("row {}: null on a REQUIRED column", row_offset + i), - }); - } - } - // Validity buffer reported nulls but no row matched — - // unreachable in practice. - Ok(arr.values().to_vec()) -} - -fn required_bool( - batch: &RecordBatch, - name: &'static str, - row_offset: usize, -) -> Result, ReaderError> { - let col = required_column(batch, name)?; - let arr = col - .as_boolean_opt() - .ok_or_else(|| ReaderError::Conversion { - column: name, - detail: format!("expected BooleanArray, got {:?}", col.data_type()), - })?; - let mut out = Vec::with_capacity(arr.len()); - for i in 0..arr.len() { - if arr.is_null(i) { - return Err(ReaderError::Conversion { - column: name, - detail: format!("row {}: null on a REQUIRED column", row_offset + i), - }); - } - out.push(arr.value(i)); - } - Ok(out) -} - -fn required_timestamp( - batch: &RecordBatch, - name: &'static str, - row_offset: usize, -) -> Result, ReaderError> { - let col = required_column(batch, name)?; - let arr = col - .as_primitive_opt::() - .ok_or_else(|| ReaderError::Conversion { - column: name, - detail: format!( - "expected TimestampNanosecondArray, got {:?}", - col.data_type() - ), - })?; - let mut out = Vec::with_capacity(arr.len()); - for i in 0..arr.len() { - if arr.is_null(i) { - return Err(ReaderError::Conversion { - column: name, - detail: format!("row {}: null on a REQUIRED column", row_offset + i), - }); - } - out.push(arr.value(i)); - } - Ok(out) -} - -fn required_column<'a>( - batch: &'a RecordBatch, - name: &'static str, -) -> Result<&'a dyn Array, ReaderError> { - let idx = batch - .schema() - .index_of(name) - .map_err(|_| ReaderError::MissingRequiredColumn { - name: name.to_string(), - })?; - Ok(batch.column(idx).as_ref()) -} - /// RFC 0018 §3.1 — decode the per-row `scope_attributes` value: an absent /// column, a NULL cell, or `"[]"` all mean "no scope attributes" (empty vec); /// any other value is canonical JSON to decode. @@ -1137,130 +880,6 @@ fn decode_optional_scope_attributes( } } -fn optional_string( - batch: &RecordBatch, - name: &'static str, -) -> Result>>, ReaderError> { - let Some(col) = optional_column(batch, name) else { - return Ok(None); - }; - let arr = StrCol::try_new(col).ok_or_else(|| ReaderError::Conversion { - column: name, - detail: format!( - "expected Utf8/Utf8View string array, got {:?}", - col.data_type() - ), - })?; - let mut out = Vec::with_capacity(arr.len()); - for i in 0..arr.len() { - out.push(arr.get(i).map(ToString::to_string)); - } - Ok(Some(out)) -} - -fn optional_binary( - batch: &RecordBatch, - name: &'static str, -) -> Result>>>, ReaderError> { - let Some(col) = optional_column(batch, name) else { - return Ok(None); - }; - let arr = BinCol::try_new(col).ok_or_else(|| ReaderError::Conversion { - column: name, - detail: format!( - "expected Binary/BinaryView array, got {:?}", - col.data_type() - ), - })?; - let mut out = Vec::with_capacity(arr.len()); - for i in 0..arr.len() { - out.push(arr.get(i).map(<[u8]>::to_vec)); - } - Ok(Some(out)) -} - -fn optional_timestamp( - batch: &RecordBatch, - name: &'static str, -) -> Result>>, ReaderError> { - let Some(col) = optional_column(batch, name) else { - return Ok(None); - }; - let arr = col - .as_primitive_opt::() - .ok_or_else(|| ReaderError::Conversion { - column: name, - detail: format!( - "expected TimestampNanosecondArray, got {:?}", - col.data_type() - ), - })?; - let mut out = Vec::with_capacity(arr.len()); - for i in 0..arr.len() { - out.push(if arr.is_null(i) { - None - } else { - Some(arr.value(i)) - }); - } - Ok(Some(out)) -} - -fn optional_fixed_bytes16( - batch: &RecordBatch, - name: &'static str, -) -> Result>>, ReaderError> { - optional_fixed_bytes::<16>(batch, name) -} - -fn optional_fixed_bytes8( - batch: &RecordBatch, - name: &'static str, -) -> Result>>, ReaderError> { - optional_fixed_bytes::<8>(batch, name) -} - -fn optional_fixed_bytes( - batch: &RecordBatch, - name: &'static str, -) -> Result>>, ReaderError> { - let Some(col) = optional_column(batch, name) else { - return Ok(None); - }; - let arr = col - .as_fixed_size_binary_opt() - .ok_or_else(|| ReaderError::Conversion { - column: name, - detail: format!("expected FixedSizeBinaryArray, got {:?}", col.data_type()), - })?; - if usize::try_from(arr.value_length()).ok() != Some(N) { - return Err(ReaderError::Conversion { - column: name, - detail: format!( - "expected FixedSizeBinary({N}), got FixedSizeBinary({})", - arr.value_length(), - ), - }); - } - let mut out = Vec::with_capacity(arr.len()); - for i in 0..arr.len() { - out.push(if arr.is_null(i) { - None - } else { - let slice = arr.value(i); - let mut buf = [0u8; N]; - buf.copy_from_slice(slice); - Some(buf) - }); - } - Ok(Some(out)) -} - -fn optional_column<'a>(batch: &'a RecordBatch, name: &'static str) -> Option<&'a dyn Array> { - let idx = batch.schema().index_of(name).ok()?; - Some(batch.column(idx).as_ref()) -} - #[cfg(test)] mod tests { //! Reader-side schema-evolution and shape-invariant scenarios