This is an automated email from the ASF dual-hosted git repository.
alamb pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-rs.git
The following commit(s) were added to refs/heads/main by this push:
new 07a363632b [Json] Replace `ArrayData` with typed Array construction in
json-reader (#9497)
07a363632b is described below
commit 07a363632bb57a1cab74b31ce4c67c8ec9814ebd
Author: Liam Bao <[email protected]>
AuthorDate: Thu Apr 2 17:43:28 2026 -0400
[Json] Replace `ArrayData` with typed Array construction in json-reader
(#9497)
# Which issue does this PR close?
<!--
We generally require a GitHub issue to be filed for all bug fixes and
enhancements and this helps us generate change logs for our releases.
You can link an issue to this PR using the GitHub syntax.
-->
- Part of #9298.
# Rationale for this change
<!--
Why are you proposing this change? If this is already explained clearly
in the issue then this section is not needed.
Explaining clearly why changes are proposed helps reviewers understand
your changes and offer better suggestions for fixes.
-->
While implementing `ListViewArrayDecoder` in arrow-json, I noticed we
could potentially retire `ArrayDataBuilder` inside `ListArrayDecoder`.
Therefore, I'd like to use a small PR here to make sure there's no
regression
# What changes are included in this PR?
<!--
There is no need to duplicate the description in the issue here but it
is sometimes worth providing a summary of the individual changes in this
PR.
-->
Replace `ArrayDataBuilder` with `GenericListArray` in `ListArrayDecoder`
# Are these changes tested?
<!--
We typically require tests for all PRs in order to:
1. Prevent the code from being accidentally broken by subsequent changes
2. Serve as another way to document the expected behavior of the code
If tests are not included in your PR, please explain why (for example,
are they covered by existing tests)?
-->
Covered by existing tests
# Are there any user-facing changes?
<!--
If there are user-facing changes then we may require documentation to be
updated before approving the PR.
If there are any breaking changes to public APIs, please call them out.
-->
No
---
arrow-json/src/reader/binary_array.rs | 21 +++---
arrow-json/src/reader/boolean_array.rs | 9 +--
arrow-json/src/reader/decimal_array.rs | 15 ++--
arrow-json/src/reader/list_array.rs | 60 +++++++++-------
arrow-json/src/reader/map_array.rs | 106 ++++++++++++++++-------------
arrow-json/src/reader/mod.rs | 30 ++++----
arrow-json/src/reader/null_array.rs | 11 +--
arrow-json/src/reader/primitive_array.rs | 10 +--
arrow-json/src/reader/run_end_array.rs | 15 ++--
arrow-json/src/reader/string_array.rs | 16 ++---
arrow-json/src/reader/string_view_array.rs | 12 ++--
arrow-json/src/reader/struct_array.rs | 30 ++++----
arrow-json/src/reader/timestamp_array.rs | 10 +--
13 files changed, 187 insertions(+), 158 deletions(-)
diff --git a/arrow-json/src/reader/binary_array.rs
b/arrow-json/src/reader/binary_array.rs
index 712eb6bb4d..b1b736e838 100644
--- a/arrow-json/src/reader/binary_array.rs
+++ b/arrow-json/src/reader/binary_array.rs
@@ -15,12 +15,13 @@
// specific language governing permissions and limitations
// under the License.
-use arrow_array::builder::{BinaryViewBuilder, FixedSizeBinaryBuilder,
GenericBinaryBuilder};
-use arrow_array::{Array, GenericStringArray, OffsetSizeTrait};
-use arrow_data::ArrayData;
-use arrow_schema::ArrowError;
use std::io::Write;
use std::marker::PhantomData;
+use std::sync::Arc;
+
+use arrow_array::builder::{BinaryViewBuilder, FixedSizeBinaryBuilder,
GenericBinaryBuilder};
+use arrow_array::{ArrayRef, GenericStringArray, OffsetSizeTrait};
+use arrow_schema::ArrowError;
use crate::reader::ArrayDecoder;
use crate::reader::tape::{Tape, TapeElement};
@@ -87,7 +88,7 @@ pub struct BinaryArrayDecoder<O: OffsetSizeTrait> {
}
impl<O: OffsetSizeTrait> ArrayDecoder for BinaryArrayDecoder<O> {
- fn decode(&mut self, tape: &Tape<'_>, pos: &[u32]) -> Result<ArrayData,
ArrowError> {
+ fn decode(&mut self, tape: &Tape<'_>, pos: &[u32]) -> Result<ArrayRef,
ArrowError> {
let data_capacity = estimate_data_capacity(tape, pos)?;
if O::from_usize(data_capacity).is_none() {
@@ -113,7 +114,7 @@ impl<O: OffsetSizeTrait> ArrayDecoder for
BinaryArrayDecoder<O> {
}
}
- Ok(builder.finish().into_data())
+ Ok(Arc::new(builder.finish()))
}
}
@@ -129,7 +130,7 @@ impl FixedSizeBinaryArrayDecoder {
}
impl ArrayDecoder for FixedSizeBinaryArrayDecoder {
- fn decode(&mut self, tape: &Tape<'_>, pos: &[u32]) -> Result<ArrayData,
ArrowError> {
+ fn decode(&mut self, tape: &Tape<'_>, pos: &[u32]) -> Result<ArrayRef,
ArrowError> {
let mut builder = FixedSizeBinaryBuilder::with_capacity(pos.len(),
self.len);
// Preallocate for the decoded byte width (FixedSizeBinary len), not
the hex string length.
let mut scratch = Vec::with_capacity(self.len as usize);
@@ -148,7 +149,7 @@ impl ArrayDecoder for FixedSizeBinaryArrayDecoder {
}
}
- Ok(builder.finish().into_data())
+ Ok(Arc::new(builder.finish()))
}
}
@@ -156,7 +157,7 @@ impl ArrayDecoder for FixedSizeBinaryArrayDecoder {
pub struct BinaryViewDecoder {}
impl ArrayDecoder for BinaryViewDecoder {
- fn decode(&mut self, tape: &Tape<'_>, pos: &[u32]) -> Result<ArrayData,
ArrowError> {
+ fn decode(&mut self, tape: &Tape<'_>, pos: &[u32]) -> Result<ArrayRef,
ArrowError> {
let data_capacity = estimate_data_capacity(tape, pos)?;
let mut builder = BinaryViewBuilder::with_capacity(data_capacity);
let mut scratch = Vec::new();
@@ -175,7 +176,7 @@ impl ArrayDecoder for BinaryViewDecoder {
}
}
- Ok(builder.finish().into_data())
+ Ok(Arc::new(builder.finish()))
}
}
diff --git a/arrow-json/src/reader/boolean_array.rs
b/arrow-json/src/reader/boolean_array.rs
index cb2587edcb..17c0586dfa 100644
--- a/arrow-json/src/reader/boolean_array.rs
+++ b/arrow-json/src/reader/boolean_array.rs
@@ -15,9 +15,10 @@
// specific language governing permissions and limitations
// under the License.
-use arrow_array::Array;
+use std::sync::Arc;
+
+use arrow_array::ArrayRef;
use arrow_array::builder::BooleanBuilder;
-use arrow_data::ArrayData;
use arrow_schema::ArrowError;
use crate::reader::ArrayDecoder;
@@ -27,7 +28,7 @@ use crate::reader::tape::{Tape, TapeElement};
pub struct BooleanArrayDecoder {}
impl ArrayDecoder for BooleanArrayDecoder {
- fn decode(&mut self, tape: &Tape<'_>, pos: &[u32]) -> Result<ArrayData,
ArrowError> {
+ fn decode(&mut self, tape: &Tape<'_>, pos: &[u32]) -> Result<ArrayRef,
ArrowError> {
let mut builder = BooleanBuilder::with_capacity(pos.len());
for p in pos {
match tape.get(*p) {
@@ -38,6 +39,6 @@ impl ArrayDecoder for BooleanArrayDecoder {
}
}
- Ok(builder.finish().into_data())
+ Ok(Arc::new(builder.finish()))
}
}
diff --git a/arrow-json/src/reader/decimal_array.rs
b/arrow-json/src/reader/decimal_array.rs
index 07a5e182a3..c9936e04a4 100644
--- a/arrow-json/src/reader/decimal_array.rs
+++ b/arrow-json/src/reader/decimal_array.rs
@@ -16,12 +16,12 @@
// under the License.
use std::marker::PhantomData;
+use std::sync::Arc;
-use arrow_array::Array;
+use arrow_array::ArrayRef;
use arrow_array::builder::PrimitiveBuilder;
use arrow_array::types::DecimalType;
use arrow_cast::parse::parse_decimal;
-use arrow_data::ArrayData;
use arrow_schema::ArrowError;
use crate::reader::ArrayDecoder;
@@ -48,7 +48,7 @@ impl<D> ArrayDecoder for DecimalArrayDecoder<D>
where
D: DecimalType,
{
- fn decode(&mut self, tape: &Tape<'_>, pos: &[u32]) -> Result<ArrayData,
ArrowError> {
+ fn decode(&mut self, tape: &Tape<'_>, pos: &[u32]) -> Result<ArrayRef,
ArrowError> {
let mut builder = PrimitiveBuilder::<D>::with_capacity(pos.len());
for p in pos {
@@ -94,9 +94,10 @@ where
}
}
- Ok(builder
- .finish()
- .with_precision_and_scale(self.precision, self.scale)?
- .into_data())
+ Ok(Arc::new(
+ builder
+ .finish()
+ .with_precision_and_scale(self.precision, self.scale)?,
+ ))
}
}
diff --git a/arrow-json/src/reader/list_array.rs
b/arrow-json/src/reader/list_array.rs
index ea23403c4b..b11124576d 100644
--- a/arrow-json/src/reader/list_array.rs
+++ b/arrow-json/src/reader/list_array.rs
@@ -15,20 +15,24 @@
// specific language governing permissions and limitations
// under the License.
+use std::marker::PhantomData;
+use std::sync::Arc;
+
+use arrow_array::builder::BooleanBufferBuilder;
+use arrow_array::{ArrayRef, GenericListArray, OffsetSizeTrait, make_array};
+use arrow_buffer::buffer::NullBuffer;
+use arrow_buffer::{Buffer, OffsetBuffer, ScalarBuffer};
+use arrow_data::ArrayDataBuilder;
+use arrow_schema::{ArrowError, DataType, FieldRef};
+
use crate::reader::tape::{Tape, TapeElement};
use crate::reader::{ArrayDecoder, DecoderContext};
-use arrow_array::OffsetSizeTrait;
-use arrow_array::builder::BooleanBufferBuilder;
-use arrow_buffer::{Buffer, buffer::NullBuffer};
-use arrow_data::{ArrayData, ArrayDataBuilder};
-use arrow_schema::{ArrowError, DataType};
-use std::marker::PhantomData;
pub type ListArrayDecoder<O> = ListLikeArrayDecoder<O, false>;
pub type ListViewArrayDecoder<O> = ListLikeArrayDecoder<O, true>;
pub struct ListLikeArrayDecoder<O, const IS_VIEW: bool> {
- data_type: DataType,
+ field: FieldRef,
decoder: Box<dyn ArrayDecoder>,
phantom: PhantomData<O>,
is_nullable: bool,
@@ -50,7 +54,7 @@ impl<O: OffsetSizeTrait, const IS_VIEW: bool>
ListLikeArrayDecoder<O, IS_VIEW> {
let decoder = ctx.make_decoder(field.data_type(),
field.is_nullable())?;
Ok(Self {
- data_type: data_type.clone(),
+ field: field.clone(),
decoder,
phantom: Default::default(),
is_nullable,
@@ -59,7 +63,7 @@ impl<O: OffsetSizeTrait, const IS_VIEW: bool>
ListLikeArrayDecoder<O, IS_VIEW> {
}
impl<O: OffsetSizeTrait, const IS_VIEW: bool> ArrayDecoder for
ListLikeArrayDecoder<O, IS_VIEW> {
- fn decode(&mut self, tape: &Tape<'_>, pos: &[u32]) -> Result<ArrayData,
ArrowError> {
+ fn decode(&mut self, tape: &Tape<'_>, pos: &[u32]) -> Result<ArrayRef,
ArrowError> {
let mut child_pos = Vec::with_capacity(pos.len());
let mut offsets = Vec::with_capacity(pos.len() + 1);
offsets.push(O::from_usize(0).unwrap());
@@ -91,34 +95,42 @@ impl<O: OffsetSizeTrait, const IS_VIEW: bool> ArrayDecoder
for ListLikeArrayDeco
}
let offset = O::from_usize(child_pos.len()).ok_or_else(|| {
- ArrowError::JsonError(format!("offset overflow decoding {}",
self.data_type))
+ ArrowError::JsonError(format!("offset overflow decoding
{}ListArray", O::PREFIX))
})?;
offsets.push(offset);
}
- let child_data = self.decoder.decode(tape, &child_pos)?;
+ let values = self.decoder.decode(tape, &child_pos)?;
let nulls = nulls.as_mut().map(|x| NullBuffer::new(x.finish()));
- let mut data = ArrayDataBuilder::new(self.data_type.clone())
- .len(pos.len())
- .nulls(nulls)
- .child_data(vec![child_data]);
-
if IS_VIEW {
let mut sizes = Vec::with_capacity(offsets.len() - 1);
for i in 1..offsets.len() {
sizes.push(offsets[i] - offsets[i - 1]);
}
offsets.pop();
- data = data
- .add_buffer(Buffer::from_vec(offsets))
- .add_buffer(Buffer::from_vec(sizes));
+ let data_type = if O::IS_LARGE {
+ DataType::LargeListView(self.field.clone())
+ } else {
+ DataType::ListView(self.field.clone())
+ };
+ // SAFETY: offsets and sizes are constructed correctly from the
tape
+ let array_data = unsafe {
+ ArrayDataBuilder::new(data_type)
+ .len(pos.len())
+ .nulls(nulls)
+ .child_data(vec![values.to_data()])
+ .add_buffer(Buffer::from_vec(offsets))
+ .add_buffer(Buffer::from_vec(sizes))
+ .build_unchecked()
+ };
+ Ok(make_array(array_data))
} else {
- data = data.add_buffer(Buffer::from_vec(offsets));
- }
+ // SAFETY: offsets are built monotonically starting from 0
+ let offsets = unsafe {
OffsetBuffer::<O>::new_unchecked(ScalarBuffer::from(offsets)) };
- // Safety
- // Validated lengths above
- Ok(unsafe { data.build_unchecked() })
+ let array = GenericListArray::<O>::try_new(self.field.clone(),
offsets, values, nulls)?;
+ Ok(Arc::new(array))
+ }
}
}
diff --git a/arrow-json/src/reader/map_array.rs
b/arrow-json/src/reader/map_array.rs
index ff22b588c5..4ec855a666 100644
--- a/arrow-json/src/reader/map_array.rs
+++ b/arrow-json/src/reader/map_array.rs
@@ -15,16 +15,21 @@
// specific language governing permissions and limitations
// under the License.
-use crate::reader::tape::{Tape, TapeElement};
-use crate::reader::{ArrayDecoder, DecoderContext};
+use std::sync::Arc;
+
use arrow_array::builder::{BooleanBufferBuilder, BufferBuilder};
-use arrow_buffer::ArrowNativeType;
+use arrow_array::{ArrayRef, MapArray, StructArray};
use arrow_buffer::buffer::NullBuffer;
-use arrow_data::{ArrayData, ArrayDataBuilder};
-use arrow_schema::{ArrowError, DataType};
+use arrow_buffer::{ArrowNativeType, OffsetBuffer, ScalarBuffer};
+use arrow_schema::{ArrowError, DataType, FieldRef, Fields};
+
+use crate::reader::tape::{Tape, TapeElement};
+use crate::reader::{ArrayDecoder, DecoderContext};
pub struct MapArrayDecoder {
- data_type: DataType,
+ entries_field: FieldRef,
+ key_value_fields: Fields,
+ ordered: bool,
keys: Box<dyn ArrayDecoder>,
values: Box<dyn ArrayDecoder>,
is_nullable: bool,
@@ -36,28 +41,38 @@ impl MapArrayDecoder {
data_type: &DataType,
is_nullable: bool,
) -> Result<Self, ArrowError> {
- let fields = match data_type {
+ let (entries_field, ordered) = match data_type {
DataType::Map(_, true) => {
return Err(ArrowError::NotYetImplemented(
"Decoding MapArray with sorted fields".to_string(),
));
}
- DataType::Map(f, _) => match f.data_type() {
- DataType::Struct(fields) if fields.len() == 2 => fields,
- d => {
- return Err(ArrowError::InvalidArgumentError(format!(
- "MapArray must contain struct with two fields, got {d}"
- )));
- }
- },
+ DataType::Map(f, ordered) => (f.clone(), *ordered),
_ => unreachable!(),
};
- let keys = ctx.make_decoder(fields[0].data_type(),
fields[0].is_nullable())?;
- let values = ctx.make_decoder(fields[1].data_type(),
fields[1].is_nullable())?;
+ let key_value_fields = match entries_field.data_type() {
+ DataType::Struct(fields) if fields.len() == 2 => fields.clone(),
+ d => {
+ return Err(ArrowError::InvalidArgumentError(format!(
+ "MapArray must contain struct with two fields, got {d}"
+ )));
+ }
+ };
+
+ let keys = ctx.make_decoder(
+ key_value_fields[0].data_type(),
+ key_value_fields[0].is_nullable(),
+ )?;
+ let values = ctx.make_decoder(
+ key_value_fields[1].data_type(),
+ key_value_fields[1].is_nullable(),
+ )?;
Ok(Self {
- data_type: data_type.clone(),
+ entries_field,
+ key_value_fields,
+ ordered,
keys,
values,
is_nullable,
@@ -66,15 +81,7 @@ impl MapArrayDecoder {
}
impl ArrayDecoder for MapArrayDecoder {
- fn decode(&mut self, tape: &Tape<'_>, pos: &[u32]) -> Result<ArrayData,
ArrowError> {
- let s = match &self.data_type {
- DataType::Map(f, _) => match f.data_type() {
- s @ DataType::Struct(_) => s,
- _ => unreachable!(),
- },
- _ => unreachable!(),
- };
-
+ fn decode(&mut self, tape: &Tape<'_>, pos: &[u32]) -> Result<ArrayRef,
ArrowError> {
let mut offsets = BufferBuilder::<i32>::new(pos.len() + 1);
offsets.append(0);
@@ -110,34 +117,37 @@ impl ArrayDecoder for MapArrayDecoder {
}
let offset = i32::from_usize(key_pos.len()).ok_or_else(|| {
- ArrowError::JsonError(format!("offset overflow decoding {}",
self.data_type))
+ ArrowError::JsonError("offset overflow decoding
MapArray".to_string())
})?;
offsets.append(offset)
}
assert_eq!(key_pos.len(), value_pos.len());
- let key_data = self.keys.decode(tape, &key_pos)?;
- let value_data = self.values.decode(tape, &value_pos)?;
-
- let struct_data = ArrayDataBuilder::new(s.clone())
- .len(key_pos.len())
- .child_data(vec![key_data, value_data]);
-
- // Safety:
- // Valid by construction
- let struct_data = unsafe { struct_data.build_unchecked() };
+ let key_array = self.keys.decode(tape, &key_pos)?;
+ let value_array = self.values.decode(tape, &value_pos)?;
+
+ // SAFETY: fields/arrays match the schema, lengths are equal, no nulls
+ let entries = unsafe {
+ StructArray::new_unchecked_with_length(
+ self.key_value_fields.clone(),
+ vec![key_array, value_array],
+ None,
+ key_pos.len(),
+ )
+ };
let nulls = nulls.as_mut().map(|x| NullBuffer::new(x.finish()));
-
- let builder = ArrayDataBuilder::new(self.data_type.clone())
- .len(pos.len())
- .buffers(vec![offsets.finish()])
- .nulls(nulls)
- .child_data(vec![struct_data]);
-
- // Safety:
- // Valid by construction
- Ok(unsafe { builder.build_unchecked() })
+ // SAFETY: offsets are built monotonically starting from 0
+ let offsets = unsafe {
OffsetBuffer::new_unchecked(ScalarBuffer::from(offsets.finish())) };
+
+ let array = MapArray::try_new(
+ self.entries_field.clone(),
+ offsets,
+ entries,
+ nulls,
+ self.ordered,
+ )?;
+ Ok(Arc::new(array))
}
}
diff --git a/arrow-json/src/reader/mod.rs b/arrow-json/src/reader/mod.rs
index 7039d3500e..497ec4c3f3 100644
--- a/arrow-json/src/reader/mod.rs
+++ b/arrow-json/src/reader/mod.rs
@@ -133,25 +133,22 @@
//! ```
//!
-use crate::StructMode;
-use crate::reader::binary_array::{
- BinaryArrayDecoder, BinaryViewDecoder, FixedSizeBinaryArrayDecoder,
-};
use std::borrow::Cow;
use std::io::BufRead;
use std::sync::Arc;
-use chrono::Utc;
-use serde_core::Serialize;
-
+use arrow_array::cast::AsArray;
use arrow_array::timezone::Tz;
use arrow_array::types::*;
-use arrow_array::{RecordBatch, RecordBatchReader, StructArray,
downcast_integer, make_array};
-use arrow_data::ArrayData;
+use arrow_array::{ArrayRef, RecordBatch, RecordBatchReader, downcast_integer};
use arrow_schema::{ArrowError, DataType, FieldRef, Schema, SchemaRef,
TimeUnit};
-pub use schema::*;
-pub use value_iter::ValueIter;
+use chrono::Utc;
+use serde_core::Serialize;
+use crate::StructMode;
+use crate::reader::binary_array::{
+ BinaryArrayDecoder, BinaryViewDecoder, FixedSizeBinaryArrayDecoder,
+};
use crate::reader::boolean_array::BooleanArrayDecoder;
use crate::reader::decimal_array::DecimalArrayDecoder;
use crate::reader::list_array::{ListArrayDecoder, ListViewArrayDecoder};
@@ -165,6 +162,9 @@ use crate::reader::struct_array::StructArrayDecoder;
use crate::reader::tape::{Tape, TapeDecoder};
use crate::reader::timestamp_array::TimestampArrayDecoder;
+pub use schema::*;
+pub use value_iter::ValueIter;
+
mod binary_array;
mod boolean_array;
mod decimal_array;
@@ -669,9 +669,9 @@ impl Decoder {
self.tape_decoder.clear();
let batch = match self.is_field {
- true => RecordBatch::try_new(self.schema.clone(),
vec![make_array(decoded)])?,
+ true => RecordBatch::try_new(self.schema.clone(), vec![decoded])?,
false => {
-
RecordBatch::from(StructArray::from(decoded)).with_schema(self.schema.clone())?
+
RecordBatch::from(decoded.as_struct().clone()).with_schema(self.schema.clone())?
}
};
@@ -681,7 +681,7 @@ impl Decoder {
trait ArrayDecoder: Send {
/// Decode elements from `tape` starting at the indexes contained in `pos`
- fn decode(&mut self, tape: &Tape<'_>, pos: &[u32]) -> Result<ArrayData,
ArrowError>;
+ fn decode(&mut self, tape: &Tape<'_>, pos: &[u32]) -> Result<ArrayRef,
ArrowError>;
}
/// Context for decoder creation, containing configuration.
@@ -819,7 +819,7 @@ mod tests {
use arrow_array::cast::AsArray;
use arrow_array::{
Array, BooleanArray, Float64Array, GenericListViewArray, ListArray,
OffsetSizeTrait,
- StringArray, StringViewArray,
+ StringArray, StringViewArray, StructArray, make_array,
};
use arrow_buffer::{ArrowNativeType, Buffer};
use arrow_cast::display::{ArrayFormatter, FormatOptions};
diff --git a/arrow-json/src/reader/null_array.rs
b/arrow-json/src/reader/null_array.rs
index aa16678c0a..9c6ac3d288 100644
--- a/arrow-json/src/reader/null_array.rs
+++ b/arrow-json/src/reader/null_array.rs
@@ -15,21 +15,24 @@
// specific language governing permissions and limitations
// under the License.
+use std::sync::Arc;
+
+use arrow_array::{ArrayRef, NullArray};
+use arrow_schema::ArrowError;
+
use crate::reader::ArrayDecoder;
use crate::reader::tape::{Tape, TapeElement};
-use arrow_data::{ArrayData, ArrayDataBuilder};
-use arrow_schema::{ArrowError, DataType};
#[derive(Default)]
pub struct NullArrayDecoder {}
impl ArrayDecoder for NullArrayDecoder {
- fn decode(&mut self, tape: &Tape<'_>, pos: &[u32]) -> Result<ArrayData,
ArrowError> {
+ fn decode(&mut self, tape: &Tape<'_>, pos: &[u32]) -> Result<ArrayRef,
ArrowError> {
for p in pos {
if !matches!(tape.get(*p), TapeElement::Null) {
return Err(tape.error(*p, "null"));
}
}
- ArrayDataBuilder::new(DataType::Null).len(pos.len()).build()
+ Ok(Arc::new(NullArray::new(pos.len())))
}
}
diff --git a/arrow-json/src/reader/primitive_array.rs
b/arrow-json/src/reader/primitive_array.rs
index b2bffe45e4..559b82ea83 100644
--- a/arrow-json/src/reader/primitive_array.rs
+++ b/arrow-json/src/reader/primitive_array.rs
@@ -15,15 +15,15 @@
// specific language governing permissions and limitations
// under the License.
-use num_traits::NumCast;
use std::marker::PhantomData;
+use std::sync::Arc;
use arrow_array::builder::PrimitiveBuilder;
-use arrow_array::{Array, ArrowPrimitiveType};
+use arrow_array::{ArrayRef, ArrowPrimitiveType};
use arrow_cast::parse::Parser;
-use arrow_data::ArrayData;
use arrow_schema::{ArrowError, DataType};
use half::f16;
+use num_traits::NumCast;
use crate::reader::ArrayDecoder;
use crate::reader::tape::{Tape, TapeElement};
@@ -93,7 +93,7 @@ where
P: ArrowPrimitiveType + Parser,
P::Native: ParseJsonNumber + NumCast,
{
- fn decode(&mut self, tape: &Tape<'_>, pos: &[u32]) -> Result<ArrayData,
ArrowError> {
+ fn decode(&mut self, tape: &Tape<'_>, pos: &[u32]) -> Result<ArrayRef,
ArrowError> {
let mut builder =
PrimitiveBuilder::<P>::with_capacity(pos.len()).with_data_type(self.data_type.clone());
let d = &self.data_type;
@@ -154,6 +154,6 @@ where
}
}
- Ok(builder.finish().into_data())
+ Ok(Arc::new(builder.finish()))
}
}
diff --git a/arrow-json/src/reader/run_end_array.rs
b/arrow-json/src/reader/run_end_array.rs
index 8b658fa6dd..1a007ccbb6 100644
--- a/arrow-json/src/reader/run_end_array.rs
+++ b/arrow-json/src/reader/run_end_array.rs
@@ -17,15 +17,16 @@
use std::marker::PhantomData;
-use crate::reader::tape::Tape;
-use crate::reader::{ArrayDecoder, DecoderContext};
use arrow_array::types::RunEndIndexType;
-use arrow_array::{Array, PrimitiveArray, new_empty_array};
+use arrow_array::{Array, ArrayRef, PrimitiveArray, make_array,
new_empty_array};
use arrow_buffer::{ArrowNativeType, ScalarBuffer};
use arrow_data::transform::MutableArrayData;
use arrow_data::{ArrayData, ArrayDataBuilder};
use arrow_schema::{ArrowError, DataType};
+use crate::reader::tape::Tape;
+use crate::reader::{ArrayDecoder, DecoderContext};
+
pub struct RunEndEncodedArrayDecoder<R> {
data_type: DataType,
decoder: Box<dyn ArrayDecoder>,
@@ -56,13 +57,13 @@ impl<R: RunEndIndexType> RunEndEncodedArrayDecoder<R> {
}
impl<R: RunEndIndexType + Send> ArrayDecoder for RunEndEncodedArrayDecoder<R> {
- fn decode(&mut self, tape: &Tape<'_>, pos: &[u32]) -> Result<ArrayData,
ArrowError> {
+ fn decode(&mut self, tape: &Tape<'_>, pos: &[u32]) -> Result<ArrayRef,
ArrowError> {
let len = pos.len();
if len == 0 {
- return Ok(new_empty_array(&self.data_type).to_data());
+ return Ok(new_empty_array(&self.data_type));
}
- let flat_data = self.decoder.decode(tape, pos)?;
+ let flat_data = self.decoder.decode(tape, pos)?.to_data();
let mut run_ends: Vec<R::Native> = Vec::new();
let mut mutable = MutableArrayData::new(vec![&flat_data], false, len);
@@ -102,7 +103,7 @@ impl<R: RunEndIndexType + Send> ArrayDecoder for
RunEndEncodedArrayDecoder<R> {
// Safety:
// run_ends are strictly increasing with the last value equal to len,
// and values has the same length as run_ends
- Ok(unsafe { data.build_unchecked() })
+ Ok(make_array(unsafe { data.build_unchecked() }))
}
}
diff --git a/arrow-json/src/reader/string_array.rs
b/arrow-json/src/reader/string_array.rs
index 7ab5d343c9..6cdfa06013 100644
--- a/arrow-json/src/reader/string_array.rs
+++ b/arrow-json/src/reader/string_array.rs
@@ -15,18 +15,18 @@
// specific language governing permissions and limitations
// under the License.
+use std::marker::PhantomData;
+use std::sync::Arc;
+
use arrow_array::builder::GenericStringBuilder;
-use arrow_array::{Array, GenericStringArray, OffsetSizeTrait};
-use arrow_data::ArrayData;
+use arrow_array::{ArrayRef, GenericStringArray, OffsetSizeTrait};
use arrow_schema::ArrowError;
-use std::marker::PhantomData;
+use itoa;
+use ryu;
use crate::reader::ArrayDecoder;
use crate::reader::tape::{Tape, TapeElement};
-use itoa;
-use ryu;
-
const TRUE: &str = "true";
const FALSE: &str = "false";
@@ -45,7 +45,7 @@ impl<O: OffsetSizeTrait> StringArrayDecoder<O> {
}
impl<O: OffsetSizeTrait> ArrayDecoder for StringArrayDecoder<O> {
- fn decode(&mut self, tape: &Tape<'_>, pos: &[u32]) -> Result<ArrayData,
ArrowError> {
+ fn decode(&mut self, tape: &Tape<'_>, pos: &[u32]) -> Result<ArrayRef,
ArrowError> {
let coerce_primitive = self.coerce_primitive;
let mut data_capacity = 0;
@@ -130,6 +130,6 @@ impl<O: OffsetSizeTrait> ArrayDecoder for
StringArrayDecoder<O> {
}
}
- Ok(builder.finish().into_data())
+ Ok(Arc::new(builder.finish()))
}
}
diff --git a/arrow-json/src/reader/string_view_array.rs
b/arrow-json/src/reader/string_view_array.rs
index dbc27e9c50..5364317dfd 100644
--- a/arrow-json/src/reader/string_view_array.rs
+++ b/arrow-json/src/reader/string_view_array.rs
@@ -15,12 +15,13 @@
// specific language governing permissions and limitations
// under the License.
-use arrow_array::Array;
+use std::fmt::Write;
+use std::sync::Arc;
+
+use arrow_array::ArrayRef;
use arrow_array::builder::GenericByteViewBuilder;
use arrow_array::types::StringViewType;
-use arrow_data::ArrayData;
use arrow_schema::ArrowError;
-use std::fmt::Write;
use crate::reader::ArrayDecoder;
use crate::reader::tape::{Tape, TapeElement};
@@ -39,7 +40,7 @@ impl StringViewArrayDecoder {
}
impl ArrayDecoder for StringViewArrayDecoder {
- fn decode(&mut self, tape: &Tape<'_>, pos: &[u32]) -> Result<ArrayData,
ArrowError> {
+ fn decode(&mut self, tape: &Tape<'_>, pos: &[u32]) -> Result<ArrayRef,
ArrowError> {
let coerce = self.coerce_primitive;
let mut data_capacity = 0;
for &p in pos {
@@ -159,7 +160,6 @@ impl ArrayDecoder for StringViewArrayDecoder {
}
}
- let array = builder.finish();
- Ok(array.into_data())
+ Ok(Arc::new(builder.finish()))
}
}
diff --git a/arrow-json/src/reader/struct_array.rs
b/arrow-json/src/reader/struct_array.rs
index 9191afb8e6..00dc55a5fd 100644
--- a/arrow-json/src/reader/struct_array.rs
+++ b/arrow-json/src/reader/struct_array.rs
@@ -15,13 +15,16 @@
// specific language governing permissions and limitations
// under the License.
-use crate::reader::tape::{Tape, TapeElement};
-use crate::reader::{ArrayDecoder, DecoderContext, StructMode};
+use std::collections::HashMap;
+use std::sync::Arc;
+
use arrow_array::builder::BooleanBufferBuilder;
+use arrow_array::{Array, ArrayRef, StructArray};
use arrow_buffer::buffer::NullBuffer;
-use arrow_data::{ArrayData, ArrayDataBuilder};
use arrow_schema::{ArrowError, DataType, Fields};
-use std::collections::HashMap;
+
+use crate::reader::tape::{Tape, TapeElement};
+use crate::reader::{ArrayDecoder, DecoderContext, StructMode};
/// Reusable buffer for tape positions, indexed by (field_idx, row_idx).
/// A value of 0 indicates the field is absent for that row.
@@ -116,7 +119,7 @@ impl StructArrayDecoder {
}
impl ArrayDecoder for StructArrayDecoder {
- fn decode(&mut self, tape: &Tape<'_>, pos: &[u32]) -> Result<ArrayData,
ArrowError> {
+ fn decode(&mut self, tape: &Tape<'_>, pos: &[u32]) -> Result<ArrayRef,
ArrowError> {
let fields = struct_fields(&self.data_type);
let row_count = pos.len();
let field_count = fields.len();
@@ -216,7 +219,7 @@ impl ArrayDecoder for StructArrayDecoder {
}
}
- let child_data = self
+ let child_arrays = self
.decoders
.iter_mut()
.enumerate()
@@ -234,7 +237,7 @@ impl ArrayDecoder for StructArrayDecoder {
let nulls = nulls.as_mut().map(|x| NullBuffer::new(x.finish()));
- for (c, f) in child_data.iter().zip(fields) {
+ for (c, f) in child_arrays.iter().zip(fields) {
// Sanity check
assert_eq!(c.len(), pos.len());
if let Some(a) = c.nulls() {
@@ -249,14 +252,11 @@ impl ArrayDecoder for StructArrayDecoder {
}
}
- let data = ArrayDataBuilder::new(self.data_type.clone())
- .len(pos.len())
- .nulls(nulls)
- .child_data(child_data);
-
- // Safety
- // Validated lengths above
- Ok(unsafe { data.build_unchecked() })
+ // SAFETY: fields, child array lengths, and nullability are validated
above
+ let array = unsafe {
+ StructArray::new_unchecked_with_length(fields.clone(),
child_arrays, nulls, row_count)
+ };
+ Ok(Arc::new(array))
}
}
diff --git a/arrow-json/src/reader/timestamp_array.rs
b/arrow-json/src/reader/timestamp_array.rs
index ff24d5391f..3fe4dc07af 100644
--- a/arrow-json/src/reader/timestamp_array.rs
+++ b/arrow-json/src/reader/timestamp_array.rs
@@ -15,15 +15,15 @@
// specific language governing permissions and limitations
// under the License.
-use chrono::TimeZone;
use std::marker::PhantomData;
+use std::sync::Arc;
-use arrow_array::Array;
+use arrow_array::ArrayRef;
use arrow_array::builder::PrimitiveBuilder;
use arrow_array::types::ArrowTimestampType;
use arrow_cast::parse::string_to_datetime;
-use arrow_data::ArrayData;
use arrow_schema::{ArrowError, DataType, TimeUnit};
+use chrono::TimeZone;
use crate::reader::ArrayDecoder;
use crate::reader::tape::{Tape, TapeElement};
@@ -51,7 +51,7 @@ where
P: ArrowTimestampType,
Tz: TimeZone + Send,
{
- fn decode(&mut self, tape: &Tape<'_>, pos: &[u32]) -> Result<ArrayData,
ArrowError> {
+ fn decode(&mut self, tape: &Tape<'_>, pos: &[u32]) -> Result<ArrayRef,
ArrowError> {
let mut builder =
PrimitiveBuilder::<P>::with_capacity(pos.len()).with_data_type(self.data_type.clone());
@@ -105,6 +105,6 @@ where
}
}
- Ok(builder.finish().into_data())
+ Ok(Arc::new(builder.finish()))
}
}
