diff --git a/Cargo.lock b/Cargo.lock index 36895b91..71f283e7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1840,6 +1840,7 @@ dependencies = [ "serde", "serde_bytes", "serde_json", + "simdutf8", "uuid", ] diff --git a/serde_arrow/Cargo.toml b/serde_arrow/Cargo.toml index cdb0cb5f..154ce5bb 100644 --- a/serde_arrow/Cargo.toml +++ b/serde_arrow/Cargo.toml @@ -37,6 +37,7 @@ bytemuck = { version = "1", default-features = false } chrono = { version = "0.4", features = ["std"], default-features = false } half = { version = "2", features = ["bytemuck"], default-features = false } serde = { version = "1.0", features = ["derive", "std"], default-features = false } +simdutf8 = { version = "0.1", default-features = false } # arrow-version:insert: arrow-array-{version} = {{ package = "arrow-array", version = "{version}", optional = true, default-features = false }} arrow-array-60 = { package = "arrow-array", version = "60", optional = true, default-features = false } diff --git a/serde_arrow/src/internal/deserialization/array_deserializer.rs b/serde_arrow/src/internal/deserialization/array_deserializer.rs index e4ac7727..65b52eb1 100644 --- a/serde_arrow/src/internal/deserialization/array_deserializer.rs +++ b/serde_arrow/src/internal/deserialization/array_deserializer.rs @@ -6,7 +6,7 @@ use serde::{ }; use crate::internal::{ - error::{fail, Context, Error, Result}, + error::{fail, try_, Context, ContextSupport, Error, Result}, schema::Strategy, }; @@ -276,6 +276,22 @@ impl<'de> RandomAccessDeserializer<'de> for ArrayDeserializer<'de> { dispatch!(self, Self(this) => this.is_some(idx)) } + fn deserialize_option>(&self, visitor: V, idx: usize) -> Result { + match self { + Self::Utf8(this) => this.deserialize_option(visitor, idx), + Self::LargeUtf8(this) => this.deserialize_option(visitor, idx), + Self::Utf8View(this) => this.deserialize_option(visitor, idx), + _ => try_(|| { + if self.is_some(idx)? { + visitor.visit_some(self.at(idx)) + } else { + visitor.visit_none() + } + }) + .ctx(self), + } + } + fn deserialize_any_some>(&self, visitor: V, idx: usize) -> Result { dispatch!(self, Self(this) => this.deserialize_any_some(visitor, idx)) } diff --git a/serde_arrow/src/internal/deserialization/dictionary_deserializer.rs b/serde_arrow/src/internal/deserialization/dictionary_deserializer.rs index 3c6158b1..4787b289 100644 --- a/serde_arrow/src/internal/deserialization/dictionary_deserializer.rs +++ b/serde_arrow/src/internal/deserialization/dictionary_deserializer.rs @@ -8,7 +8,7 @@ use crate::internal::{ use super::{ enums_as_string_impl::EnumAccess, integer_deserializer::Integer, - random_access_deserializer::RandomAccessDeserializer, + random_access_deserializer::RandomAccessDeserializer, string_deserializer::get_utf8, }; pub struct DictionaryDeserializer<'a, K: Integer, V: Offset> { @@ -28,7 +28,7 @@ impl<'a, K: Integer, V: Offset> DictionaryDeserializer<'a, K, V> { pub fn get_str(&self, idx: usize) -> Result<&str> { let key: usize = self.keys.get_required(idx)?.into_i64()?.try_into()?; - let value: &str = self.values.get_required(key)?; + let value: &str = get_utf8(&self.values, key)?; Ok(value) } } diff --git a/serde_arrow/src/internal/deserialization/list_deserializer.rs b/serde_arrow/src/internal/deserialization/list_deserializer.rs index 1df8665f..c0e2ce89 100644 --- a/serde_arrow/src/internal/deserialization/list_deserializer.rs +++ b/serde_arrow/src/internal/deserialization/list_deserializer.rs @@ -131,4 +131,8 @@ impl<'de> SeqAccess<'de> for ListItemDeserializer<'_, 'de> { self.start += 1; Ok(Some(item)) } + + fn size_hint(&self) -> Option { + Some(self.end.saturating_sub(self.start).min(1024)) + } } diff --git a/serde_arrow/src/internal/deserialization/string_deserializer.rs b/serde_arrow/src/internal/deserialization/string_deserializer.rs index 6f27fbf3..32e78196 100644 --- a/serde_arrow/src/internal/deserialization/string_deserializer.rs +++ b/serde_arrow/src/internal/deserialization/string_deserializer.rs @@ -36,6 +36,13 @@ impl StringDeserializer { } } +pub(super) fn get_utf8<'a, V>(view: &V, idx: usize) -> Result<&'a str> +where + V: ViewAccess<'a, str> + 'a, +{ + ViewAccess::::get_required(view, idx) +} + impl Context for StringDeserializer { fn annotate(&self, annotations: &mut std::collections::BTreeMap) { set_default(annotations, "field", &self.path); @@ -51,6 +58,24 @@ where self.view.is_some(idx) } + fn deserialize_option>( + &self, + visitor: V, + idx: usize, + ) -> Result { + try_(|| match ViewAccess::::get(&self.view, idx)? { + Some(value) => visitor.visit_some( + ValidatedStringDeserializer { + original: self, + value, + } + .at(idx), + ), + None => visitor.visit_none(), + }) + .ctx(self) + } + fn deserialize_any_some>( &self, visitor: V, @@ -64,7 +89,7 @@ where visitor: V, idx: usize, ) -> Result { - try_(|| visitor.visit_borrowed_str(self.view.get_required(idx)?)).ctx(self) + try_(|| visitor.visit_borrowed_str(get_utf8(&self.view, idx)?)).ctx(self) } fn deserialize_string>( @@ -72,7 +97,7 @@ where visitor: V, idx: usize, ) -> Result { - try_(|| visitor.visit_string(self.view.get_required(idx)?.to_owned())).ctx(self) + try_(|| visitor.visit_string(get_utf8(&self.view, idx)?.to_owned())).ctx(self) } fn deserialize_bytes>( @@ -80,7 +105,7 @@ where visitor: V, idx: usize, ) -> Result { - try_(|| visitor.visit_bytes(self.view.get_required(idx)?.as_bytes())).ctx(self) + try_(|| visitor.visit_bytes(get_utf8(&self.view, idx)?.as_bytes())).ctx(self) } fn deserialize_byte_buf>( @@ -88,7 +113,7 @@ where visitor: V, idx: usize, ) -> Result { - try_(|| visitor.visit_byte_buf(self.view.get_required(idx)?.to_owned().into_bytes())) + try_(|| visitor.visit_byte_buf(get_utf8(&self.view, idx)?.to_owned().into_bytes())) .ctx(self) } @@ -100,9 +125,137 @@ where idx: usize, ) -> Result { try_(|| { - let variant = self.view.get_required(idx)?; + let variant = get_utf8(&self.view, idx)?; visitor.visit_enum(EnumAccess(variant)) }) .ctx(self) } } + +struct ValidatedStringDeserializer<'this, 'a, VV> { + original: &'this StringDeserializer, + value: &'a str, +} + +impl Context for ValidatedStringDeserializer<'_, '_, VV> { + fn annotate(&self, annotations: &mut std::collections::BTreeMap) { + self.original.annotate(annotations); + } +} + +impl<'a, VV> RandomAccessDeserializer<'a> for ValidatedStringDeserializer<'_, 'a, VV> +where + VV: StringDeserializerDataType, +{ + fn is_some(&self, _idx: usize) -> Result { + Ok(true) + } + + fn deserialize_any_some>( + &self, + visitor: V, + idx: usize, + ) -> Result { + self.deserialize_str(visitor, idx) + } + + fn deserialize_str>( + &self, + visitor: V, + _idx: usize, + ) -> Result { + try_(|| visitor.visit_borrowed_str(self.value)).ctx(self) + } + + fn deserialize_string>( + &self, + visitor: V, + _idx: usize, + ) -> Result { + try_(|| visitor.visit_string(self.value.to_owned())).ctx(self) + } + + fn deserialize_bytes>( + &self, + visitor: V, + _idx: usize, + ) -> Result { + try_(|| visitor.visit_bytes(self.value.as_bytes())).ctx(self) + } + + fn deserialize_byte_buf>( + &self, + visitor: V, + _idx: usize, + ) -> Result { + try_(|| visitor.visit_byte_buf(self.value.to_owned().into_bytes())).ctx(self) + } + + fn deserialize_enum>( + &self, + _name: &'static str, + _variants: &'static [&'static str], + visitor: V, + _idx: usize, + ) -> Result { + try_(|| visitor.visit_enum(EnumAccess(self.value))).ctx(self) + } +} + +#[cfg(test)] +mod tests { + use marrow::{ + datatypes::{DataType, Field}, + view::{BitsWithOffset, BytesView, View}, + }; + use serde::Deserialize; + + #[derive(Debug, Deserialize, PartialEq)] + struct Row<'a> { + #[serde(borrow)] + value: Option<&'a str>, + } + + #[test] + fn nullable_marrow_string_borrows_valid_values_and_reports_invalid_utf8() { + let fields = [Field { + name: "value".into(), + data_type: DataType::LargeUtf8, + nullable: true, + metadata: Default::default(), + }]; + let offsets = [0_i64, 5, 5]; + let valid = [0b0000_0001]; + let views = [View::LargeUtf8(BytesView { + offsets: &offsets, + data: "café".as_bytes(), + validity: Some(BitsWithOffset { + offset: 0, + data: &valid, + }), + })]; + let decoded: Vec> = crate::from_marrow(&fields, &views).unwrap(); + assert_eq!( + decoded, + vec![ + Row { + value: Some("café") + }, + Row { value: None } + ] + ); + + let invalid_offsets = [0_i64, 2]; + let invalid = [b'a', 0xff]; + let views = [View::LargeUtf8(BytesView { + offsets: &invalid_offsets, + data: &invalid, + validity: Some(BitsWithOffset { + offset: 0, + data: &valid, + }), + })]; + let error = crate::from_marrow::>>(&fields, &views).unwrap_err(); + assert!(error.to_string().contains("invalid UTF-8 data")); + } +} diff --git a/serde_arrow/src/internal/deserialization/struct_deserializer.rs b/serde_arrow/src/internal/deserialization/struct_deserializer.rs index 743e65e8..3d703066 100644 --- a/serde_arrow/src/internal/deserialization/struct_deserializer.rs +++ b/serde_arrow/src/internal/deserialization/struct_deserializer.rs @@ -1,3 +1,5 @@ +use std::sync::atomic::{AtomicPtr, Ordering}; + use marrow::view::{BitsWithOffset, StructView}; use serde::de::{value::StrDeserializer, DeserializeSeed, MapAccess, SeqAccess, Visitor}; @@ -17,6 +19,7 @@ pub struct StructDeserializer<'a> { pub fields: Vec<(String, ArrayDeserializer<'a>)>, pub validity: Option>, pub len: usize, + ordered_fields: AtomicPtr<&'static str>, } impl<'a> StructDeserializer<'a> { @@ -48,6 +51,32 @@ impl<'a> StructDeserializer<'a> { fields, validity, len, + ordered_fields: AtomicPtr::new(std::ptr::null_mut()), + } + } + + fn fields_match(&self, fields: &'static [&'static str]) -> bool { + if self.fields.len() != fields.len() { + return false; + } + + // Derived Deserialize implementations use a static field slice. Cache the + // matching slice's address so subsequent rows skip string comparisons. + let field_ptr = fields.as_ptr() as *mut &'static str; + if self.ordered_fields.load(Ordering::Relaxed) == field_ptr { + return true; + } + + if self + .fields + .iter() + .zip(fields) + .all(|((actual, _), expected)| actual == expected) + { + self.ordered_fields.store(field_ptr, Ordering::Relaxed); + true + } else { + false } } } @@ -93,13 +122,19 @@ impl<'de> RandomAccessDeserializer<'de> for StructDeserializer<'de> { fn deserialize_struct>( &self, _: &'static str, - _: &'static [&'static str], + fields: &'static [&'static str], visitor: V, idx: usize, ) -> Result { - visitor - .visit_map(StructItemDeserializer::new(self, idx)) - .ctx(self) + if self.fields_match(fields) { + visitor + .visit_seq(StructSeqDeserializer::new(self, idx)) + .ctx(self) + } else { + visitor + .visit_map(StructItemDeserializer::new(self, idx)) + .ctx(self) + } } fn deserialize_tuple>( @@ -109,7 +144,7 @@ impl<'de> RandomAccessDeserializer<'de> for StructDeserializer<'de> { idx: usize, ) -> Result { visitor - .visit_seq(StructItemDeserializer::new(self, idx)) + .visit_seq(StructSeqDeserializer::new(self, idx)) .ctx(self) } @@ -121,7 +156,7 @@ impl<'de> RandomAccessDeserializer<'de> for StructDeserializer<'de> { idx: usize, ) -> Result { visitor - .visit_seq(StructItemDeserializer::new(self, idx)) + .visit_seq(StructSeqDeserializer::new(self, idx)) .ctx(self) } } @@ -169,17 +204,94 @@ impl<'de> MapAccess<'de> for StructItemDeserializer<'_, 'de> { } } -impl<'de> SeqAccess<'de> for StructItemDeserializer<'_, 'de> { +struct StructSeqDeserializer<'a, 'de> { + fields: &'a [(String, ArrayDeserializer<'de>)], + item: usize, +} + +impl<'a, 'de> StructSeqDeserializer<'a, 'de> { + fn new(deserializer: &'a StructDeserializer<'de>, item: usize) -> Self { + Self { + fields: &deserializer.fields, + item, + } + } +} + +impl<'de> SeqAccess<'de> for StructSeqDeserializer<'_, 'de> { type Error = Error; fn next_element_seed>(&mut self, seed: T) -> Result> { - let Some((_, field_deserializer)) = self.deserializer.fields.get(self.field) else { + let Some(((_, field_deserializer), rest)) = self.fields.split_first() else { return Ok(None); }; let res = seed.deserialize(field_deserializer.at(self.item))?; - self.field += 1; + self.fields = rest; Ok(Some(res)) } } + +#[cfg(test)] +mod tests { + use marrow::{ + datatypes::{DataType, Field}, + view::{PrimitiveView, View}, + }; + use serde::Deserialize; + + use crate::Deserializer; + + #[derive(Debug, Deserialize, PartialEq)] + struct Ordered { + a: i32, + b: i32, + } + + #[derive(Debug, Deserialize, PartialEq)] + struct Reordered { + b: i32, + a: i32, + } + + #[test] + fn cached_field_match_preserves_reordered_deserialization() { + let fields = [ + Field { + name: "a".into(), + data_type: DataType::Int32, + ..Field::default() + }, + Field { + name: "b".into(), + data_type: DataType::Int32, + ..Field::default() + }, + ]; + let views = [ + View::Int32(PrimitiveView { + values: &[11, 33], + validity: None, + }), + View::Int32(PrimitiveView { + values: &[22, 44], + validity: None, + }), + ]; + let deserializer = Deserializer::from_marrow(&fields, &views).unwrap(); + + assert_eq!( + Ordered::deserialize(deserializer.get(0).unwrap()).unwrap(), + Ordered { a: 11, b: 22 } + ); + assert_eq!( + Reordered::deserialize(deserializer.get(0).unwrap()).unwrap(), + Reordered { a: 11, b: 22 } + ); + assert_eq!( + Ordered::deserialize(deserializer.get(1).unwrap()).unwrap(), + Ordered { a: 33, b: 44 } + ); + } +} diff --git a/serde_arrow/src/internal/deserializer.rs b/serde_arrow/src/internal/deserializer.rs index 569d7cb6..b24449de 100644 --- a/serde_arrow/src/internal/deserializer.rs +++ b/serde_arrow/src/internal/deserializer.rs @@ -530,6 +530,17 @@ impl<'de> SeqAccess<'de> for Private> { self.0.next += 1; Ok(Some(item)) } + + fn size_hint(&self) -> Option { + // Keep preallocation bounded when views come from untrusted Marrow offsets. + Some( + self.0 + .deserializer + .len + .saturating_sub(self.0.next) + .min(1024), + ) + } } #[allow(unused, reason = "trait assertion")] diff --git a/serde_arrow/src/internal/utils/array_view_ext.rs b/serde_arrow/src/internal/utils/array_view_ext.rs index 810d5b19..92bfd4b6 100644 --- a/serde_arrow/src/internal/utils/array_view_ext.rs +++ b/serde_arrow/src/internal/utils/array_view_ext.rs @@ -231,9 +231,38 @@ where { fn get(&self, idx: usize) -> Result> { match ViewAccess::<[u8]>::get(self, idx) { - Ok(Some(data)) => Ok(Some(std::str::from_utf8(data)?)), + // Preserve the detailed std error for malformed Marrow views. + Ok(Some(data)) => match simdutf8::basic::from_utf8(data) { + Ok(value) => Ok(Some(value)), + Err(_) => Ok(Some(std::str::from_utf8(data)?)), + }, Ok(None) => Ok(None), Err(err) => Err(err), } } } + +#[cfg(test)] +mod tests { + use marrow::view::BytesView; + + use super::ViewAccess; + + #[test] + fn utf8_access_validates_marrow_data() { + let valid = BytesView { + offsets: &[0_i32, 5], + data: "café".as_bytes(), + validity: None, + }; + assert_eq!(ViewAccess::::get(&valid, 0).unwrap(), Some("café")); + + let invalid = BytesView { + offsets: &[0_i32, 2], + data: &[b'a', 0xff], + validity: None, + }; + let error = ViewAccess::::get(&invalid, 0).unwrap_err(); + assert!(error.to_string().contains("invalid UTF-8 data")); + } +} diff --git a/serde_arrow_bench/benches/groups/complex.rs b/serde_arrow_bench/benches/groups/complex.rs index ea79e615..60a5c5bc 100644 --- a/serde_arrow_bench/benches/groups/complex.rs +++ b/serde_arrow_bench/benches/groups/complex.rs @@ -11,7 +11,7 @@ use arrow_schema::{DataType, Field, Schema}; use rand::{ distributions::{Standard, Uniform}, prelude::Distribution, - Rng, + Rng, SeedableRng, }; use serde::{Deserialize, Serialize}; @@ -95,8 +95,9 @@ pub fn benchmark_serialize(c: &mut criterion::Criterion) { } pub fn benchmark_deserialize(c: &mut criterion::Criterion) { + let mut rng = rand::rngs::StdRng::seed_from_u64(42); let items = (0..1_000) - .map(|_| Item::random(&mut rand::thread_rng())) + .map(|_| Item::random(&mut rng)) .collect::>(); let mut group = super::new_group(c, "complex_1000_deserialize"); diff --git a/serde_arrow_bench/benches/groups/mod.rs b/serde_arrow_bench/benches/groups/mod.rs index 91fe2e95..cb1d8e82 100644 --- a/serde_arrow_bench/benches/groups/mod.rs +++ b/serde_arrow_bench/benches/groups/mod.rs @@ -3,6 +3,7 @@ use criterion::measurement::Measurement; pub mod binary; pub mod complex; pub mod json_to_arrow; +pub mod nullable_strings; pub mod primitives; pub mod wide_schema; diff --git a/serde_arrow_bench/benches/groups/nullable_strings.rs b/serde_arrow_bench/benches/groups/nullable_strings.rs new file mode 100644 index 00000000..fd868222 --- /dev/null +++ b/serde_arrow_bench/benches/groups/nullable_strings.rs @@ -0,0 +1,64 @@ +use arrow_array::{Array, ArrayRef, LargeStringArray}; +use criterion::Criterion; +use serde::{Deserialize, Serialize}; + +#[derive(Debug, PartialEq, Deserialize, Serialize)] +struct Item { + value: Option, +} + +pub fn benchmark_deserialize(c: &mut Criterion) { + let items = (0..1_000) + .map(|idx| Item { + value: (idx % 10 != 0).then(|| format!("{idx:04}{}", "é".repeat(64))), + }) + .collect::>(); + let mut group = super::new_group(c, "nullable_strings_1000_deserialize"); + + let arrow_fields = crate::impls::serde_arrow_arrow::trace(&items); + let arrow_arrays = serde_arrow::to_arrow(&arrow_fields, &items).unwrap(); + assert_eq!(deserialize_manual(&arrow_arrays), items); + group.bench_function("arrow_manual", |b| { + b.iter(|| criterion::black_box(deserialize_manual(&arrow_arrays))) + }); + let decoded: Vec = serde_arrow::from_arrow(&arrow_fields, &arrow_arrays).unwrap(); + assert_eq!(decoded, items); + group.bench_function("serde_arrow_arrow", |b| { + b.iter(|| { + let decoded: Vec = serde_arrow::from_arrow(&arrow_fields, &arrow_arrays).unwrap(); + criterion::black_box(decoded) + }) + }); + + let marrow_fields = crate::impls::serde_arrow_marrow::trace(&items); + let marrow_arrays = serde_arrow::to_marrow(&marrow_fields, &items).unwrap(); + let marrow_views = marrow_arrays + .iter() + .map(|array| array.as_view()) + .collect::>(); + let decoded: Vec = serde_arrow::from_marrow(&marrow_fields, &marrow_views).unwrap(); + assert_eq!(decoded, items); + group.bench_function("serde_arrow_marrow", |b| { + b.iter(|| { + let decoded: Vec = + serde_arrow::from_marrow(&marrow_fields, &marrow_views).unwrap(); + criterion::black_box(decoded) + }) + }); + + group.finish(); +} + +fn deserialize_manual(arrays: &[ArrayRef]) -> Vec { + let values = arrays[0] + .as_any() + .downcast_ref::() + .unwrap(); + (0..values.len()) + .map(|idx| Item { + value: (!values.is_null(idx)).then(|| values.value(idx).to_owned()), + }) + .collect() +} + +criterion::criterion_group!(benchmark, benchmark_deserialize); diff --git a/serde_arrow_bench/benches/groups/primitives.rs b/serde_arrow_bench/benches/groups/primitives.rs index 13c55b0f..4e28f620 100644 --- a/serde_arrow_bench/benches/groups/primitives.rs +++ b/serde_arrow_bench/benches/groups/primitives.rs @@ -11,7 +11,7 @@ use arrow_schema::Schema; use rand::{ distributions::{Standard, Uniform}, prelude::Distribution, - Rng, + Rng, SeedableRng, }; use serde::{Deserialize, Serialize}; @@ -94,8 +94,9 @@ pub fn benchmark_serialize(c: &mut criterion::Criterion) { } pub fn benchmark_deserialize(c: &mut criterion::Criterion) { + let mut rng = rand::rngs::StdRng::seed_from_u64(42); let items = (0..1_000) - .map(|_| Item::random(&mut rand::thread_rng())) + .map(|_| Item::random(&mut rng)) .collect::>(); let mut group = super::new_group(c, "primitives_1000_deserialize"); diff --git a/serde_arrow_bench/benches/serde_arrow_bench.rs b/serde_arrow_bench/benches/serde_arrow_bench.rs index 92d2f440..7523be51 100644 --- a/serde_arrow_bench/benches/serde_arrow_bench.rs +++ b/serde_arrow_bench/benches/serde_arrow_bench.rs @@ -4,6 +4,7 @@ pub mod impls; criterion::criterion_main!( groups::binary::benchmark, groups::complex::benchmark, + groups::nullable_strings::benchmark, groups::primitives::benchmark, groups::json_to_arrow::benchmark, groups::wide_schema::benchmark,