From cbdfd7ba957e034b724038126cfea23dea4dcf83 Mon Sep 17 00:00:00 2001 From: Jens Holdgaard Pedersen Date: Thu, 27 Aug 2026 13:10:27 +0200 Subject: [PATCH 1/3] refactor(parquet): one accessor family decodes for both readers MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Epic #745 wave 2 slice A part 1: the line-for-line twin required_*/ optional_* accessors, the view-tolerant StrCol/BinCol wrappers, and the §3.8/§3.9 baseline-column check unify in decode.rs, generic over a two-constructor DecodeError trait implemented by ReaderError and AuditReaderError — the divergence class that produced the audit Utf8View bug (#746) structurally ends: a fix lands once or not at all, and both readers' diagnostics are built from one text. Reader- specific decode (scope attributes, the record/event assemblers) stays in each reader. One deliberate seam kept: the audit reader's historical absent-column-as-empty-vec shape for the hash column is an explicit unwrap_or_default at its one call site. -694 lines of twins replaced by one ~520-line shared module. Behavior identical: parquet 172 passed (same as baseline), querier 260 (RFC0005.2/.3/.4 compat trio, RFC 0009 suite incl. rfc0009_6 cross-amendment merge, rfc0042_8 re-projection all green). Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01JZXtbyWoQY19ZGtNecDfgv Signed-off-by: Jens Holdgaard Pedersen --- crates/ourios-parquet/src/audit_reader.rs | 314 +----------- crates/ourios-parquet/src/decode.rs | 569 ++++++++++++++++++++++ crates/ourios-parquet/src/lib.rs | 1 + crates/ourios-parquet/src/reader.rs | 399 +-------------- 4 files changed, 589 insertions(+), 694 deletions(-) create mode 100644 crates/ourios-parquet/src/decode.rs 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..76f418542 --- /dev/null +++ b/crates/ourios-parquet/src/decode.rs @@ -0,0 +1,569 @@ +//! 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: 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`]. +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().clone())); + } + } + 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) +} 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 From f8b04f8e0e6f8f06f4e206453b669ea546e8fd0e Mon Sep 17 00:00:00 2001 From: Jens Holdgaard Pedersen Date: Thu, 27 Aug 2026 13:25:49 +0200 Subject: [PATCH 2/3] docs(parquet): baseline-check doc names both readers; drop a clone Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01JZXtbyWoQY19ZGtNecDfgv Signed-off-by: Jens Holdgaard Pedersen --- crates/ourios-parquet/src/decode.rs | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/crates/ourios-parquet/src/decode.rs b/crates/ourios-parquet/src/decode.rs index 76f418542..8fb1da5d7 100644 --- a/crates/ourios-parquet/src/decode.rs +++ b/crates/ourios-parquet/src/decode.rs @@ -78,9 +78,10 @@ impl<'a> BinCol<'a> { } } -/// 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`]. +/// RFC 0005 §3.9 / §3.7: 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, @@ -91,7 +92,7 @@ pub(crate) fn require_baseline_columns( .column_with_name(expected_field.name()) .is_none() { - return Err(E::missing_required(&expected_field.name().clone())); + return Err(E::missing_required(expected_field.name())); } } Ok(()) From 8c3c98ed74c97ec0c64da89733432b70052f2396 Mon Sep 17 00:00:00 2001 From: Jens Holdgaard Pedersen Date: Thu, 27 Aug 2026 13:36:51 +0200 Subject: [PATCH 3/3] test(parquet): colocated cases for the shared decode invariants MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review round: the baseline check, null-on-required rejection with file-global row indices, absent-optional shapes (None vs the audit- historical empty vec) and type-mismatch conversion errors get direct cases beside the shared module, each pinned through both error types where the seam matters. The section citation reads §3.9 (reader contract) with §3.8 as policy. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01JZXtbyWoQY19ZGtNecDfgv Signed-off-by: Jens Holdgaard Pedersen --- crates/ourios-parquet/src/decode.rs | 75 ++++++++++++++++++++++++++++- 1 file changed, 74 insertions(+), 1 deletion(-) diff --git a/crates/ourios-parquet/src/decode.rs b/crates/ourios-parquet/src/decode.rs index 8fb1da5d7..bf2b00e2d 100644 --- a/crates/ourios-parquet/src/decode.rs +++ b/crates/ourios-parquet/src/decode.rs @@ -78,7 +78,8 @@ impl<'a> BinCol<'a> { } } -/// RFC 0005 §3.9 / §3.7: every REQUIRED (non-nullable) column of +/// 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. @@ -568,3 +569,75 @@ pub(crate) fn optional_u64_list( } 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", .. }) + )); + } +}