This is an automated email from the ASF dual-hosted git repository.
Jefffrey 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 76326e8377 fix(arrow-ipc): return an error instead of panicking on
malformed IPC schema (#10647)
76326e8377 is described below
commit 76326e837767957f001b198527a5bd8d38e0fc56
Author: Aditya Mishra <[email protected]>
AuthorDate: Fri Aug 14 12:43:40 2026 +0530
fix(arrow-ipc): return an error instead of panicking on malformed IPC
schema (#10647)
# Which issue does this PR close?
- Closes #10575.
- Closes #10437.
# Rationale for this change
`arrow_ipc::convert::fb_to_schema` is infallible and panics on schema
messages that the flatbuffer verifier itself accepts. the first thing it
does is `fb.fields().unwrap()`, and `fields` is optional in the
flatbuffer schema, so a message with no fields aborts the process.
`get_data_type` has the same problem deeper down: about 20 more
`panic!`/`unimplemented!` and 30 `unwrap()` reachable from decoder input
(unknown float precision, `Type NONE`, out of range enum tags, and so
on).
every reader path funnels through `fb_to_schema`, so there was no way to
read Arrow IPC without exposing the process to an abort on untrusted
input. it also reaches parquet (an untrusted `.parquet` whose footer
carries the `ARROW:schema` key) and arrow-flight.
`StreamReader::try_new` returns `Result`, so malformed input should be
an `Err`, not a process abort.
# What changes are included in this PR?
- `get_data_type` (already `pub(crate)`) now returns `Result<DataType,
ArrowError>`; every input reachable `unwrap`/`panic!`/`unimplemented!`
is now a returned `ParseError`.
- new `pub fn try_fb_to_schema` and a private fallible field conversion.
the readers in arrow-ipc, the parquet `ARROW:schema` path, and
arrow-flight are switched onto it.
- `fb_to_schema` and the public `From<crate::Field>` impl keep their
signatures for compatibility and delegate to the fallible path.
# Are these changes tested?
yes. a schema message with no fields, one with an unknown float
precision, and one with `Type NONE` now return `Err` instead of
aborting, asserted by new tests in `arrow-ipc`. the flatbuffer verifier
accepts all three, so they exercise the exact gap. existing round trip
tests still pass.
# Are there any user-facing changes?
no breaking changes. `try_fb_to_schema` is additive, and `fb_to_schema`
still exists with the same signature.
---
arrow-flight/src/sql/client.rs | 4 +-
arrow-flight/src/utils.rs | 4 +-
.../flight_client_scenarios/integration_test.rs | 2 +-
arrow-ipc/benches/ipc_reader.rs | 4 +-
arrow-ipc/src/convert.rs | 385 +++++++++++++++------
arrow-ipc/src/reader.rs | 16 +-
arrow-ipc/src/reader/stream.rs | 8 +-
arrow-ipc/src/writer.rs | 6 +-
arrow/examples/zero_copy_ipc.rs | 4 +-
parquet/src/arrow/schema/mod.rs | 6 +-
10 files changed, 305 insertions(+), 134 deletions(-)
diff --git a/arrow-flight/src/sql/client.rs b/arrow-flight/src/sql/client.rs
index 8ef80da50b..6ea1a04fc1 100644
--- a/arrow-flight/src/sql/client.rs
+++ b/arrow-flight/src/sql/client.rs
@@ -19,7 +19,7 @@
use arrow_buffer::Buffer;
use arrow_ipc::MessageHeader;
-use arrow_ipc::convert::fb_to_schema;
+use arrow_ipc::convert::try_fb_to_schema;
use arrow_ipc::reader::read_record_batch;
use arrow_ipc::root_as_message;
use arrow_schema::SchemaRef;
@@ -664,7 +664,7 @@ pub fn arrow_data_from_flight_data(
)
})?;
- let arrow_schema = fb_to_schema(ipc_schema);
+ let arrow_schema = try_fb_to_schema(ipc_schema)?;
Ok(ArrowFlightData::Schema(arrow_schema))
}
MessageHeader::DictionaryBatch => {
diff --git a/arrow-flight/src/utils.rs b/arrow-flight/src/utils.rs
index 07a1801914..0cc55ee75d 100644
--- a/arrow-flight/src/utils.rs
+++ b/arrow-flight/src/utils.rs
@@ -23,7 +23,7 @@ use std::sync::Arc;
use arrow_array::{ArrayRef, RecordBatch};
use arrow_buffer::Buffer;
-use arrow_ipc::convert::fb_to_schema;
+use arrow_ipc::convert::try_fb_to_schema;
use arrow_ipc::writer::IpcWriteContext;
use arrow_ipc::{reader, root_as_message, writer, writer::IpcWriteOptions};
use arrow_schema::{ArrowError, Schema, SchemaRef};
@@ -39,7 +39,7 @@ pub fn flight_data_to_batches(flight_data: &[FlightData]) ->
Result<Vec<RecordBa
let ipc_schema: arrow_ipc::Schema = message
.header_as_schema()
.ok_or_else(|| ArrowError::CastError("Cannot get header as
Schema".to_string()))?;
- let schema = fb_to_schema(ipc_schema);
+ let schema = try_fb_to_schema(ipc_schema)?;
let schema = Arc::new(schema);
let mut batches = vec![];
diff --git
a/arrow-integration-testing/src/flight_client_scenarios/integration_test.rs
b/arrow-integration-testing/src/flight_client_scenarios/integration_test.rs
index 950a64562e..559ede93d1 100644
--- a/arrow-integration-testing/src/flight_client_scenarios/integration_test.rs
+++ b/arrow-integration-testing/src/flight_client_scenarios/integration_test.rs
@@ -286,7 +286,7 @@ async fn receive_schema_flight_data(resp: &mut
Streaming<FlightData>) -> Option<
let ipc_schema: ipc::Schema = message
.header_as_schema()
.expect("Unable to read IPC message as schema");
- let schema = ipc::convert::fb_to_schema(ipc_schema);
+ let schema = ipc::convert::try_fb_to_schema(ipc_schema).unwrap();
Some(schema)
}
diff --git a/arrow-ipc/benches/ipc_reader.rs b/arrow-ipc/benches/ipc_reader.rs
index ef1de88d32..5d7c373ee5 100644
--- a/arrow-ipc/benches/ipc_reader.rs
+++ b/arrow-ipc/benches/ipc_reader.rs
@@ -18,7 +18,7 @@
use arrow_array::builder::{Date32Builder, Decimal128Builder, Int32Builder};
use arrow_array::{RecordBatch, builder::StringBuilder};
use arrow_buffer::Buffer;
-use arrow_ipc::convert::fb_to_schema;
+use arrow_ipc::convert::try_fb_to_schema;
use arrow_ipc::reader::{FileDecoder, FileReader, StreamReader,
read_footer_length};
use arrow_ipc::writer::{FileWriter, IpcWriteOptions, StreamWriter};
use arrow_ipc::{Block, CompressionType, root_as_footer};
@@ -215,7 +215,7 @@ impl IPCBufferDecoder {
let footer_len =
read_footer_length(buffer[trailer_start..].try_into().unwrap()).unwrap();
let footer = root_as_footer(&buffer[trailer_start -
footer_len..trailer_start]).unwrap();
- let schema = fb_to_schema(footer.schema().unwrap());
+ let schema = try_fb_to_schema(footer.schema().unwrap()).unwrap();
let mut decoder = FileDecoder::new(Arc::new(schema), footer.version());
diff --git a/arrow-ipc/src/convert.rs b/arrow-ipc/src/convert.rs
index 6429f76f8a..7e1fc962f4 100644
--- a/arrow-ipc/src/convert.rs
+++ b/arrow-ipc/src/convert.rs
@@ -34,11 +34,11 @@ use DataType::*;
/// Low level Arrow [Schema] to IPC bytes converter
///
-/// See also [`fb_to_schema`] for the reverse operation
+/// See also [`try_fb_to_schema`] for the reverse operation
///
/// # Example
/// ```
-/// # use arrow_ipc::convert::{fb_to_schema, IpcSchemaEncoder};
+/// # use arrow_ipc::convert::{try_fb_to_schema, IpcSchemaEncoder};
/// # use arrow_ipc::root_as_schema;
/// # use arrow_ipc::writer::DictionaryTracker;
/// # use arrow_schema::{DataType, Field, Schema};
@@ -59,7 +59,7 @@ use DataType::*;
///
/// // convert the IPC bytes back to an Arrow schema
/// let ipc_schema = root_as_schema(ipc_bytes).unwrap();
-/// let schema2 = fb_to_schema(ipc_schema);
+/// let schema2 = try_fb_to_schema(ipc_schema).unwrap();
/// assert_eq!(schema, schema2);
/// ```
#[derive(Debug)]
@@ -158,52 +158,73 @@ pub fn schema_to_fb_offset<'a>(
}
/// Convert an IPC Field to Arrow Field
+///
+/// This panics on malformed input; every reader path uses the fallible
+/// conversion instead. kept for backwards compatibility only.
impl From<crate::Field<'_>> for Field {
fn from(field: crate::Field) -> Field {
- let arrow_field = if let Some(dictionary) = field.dictionary() {
- #[allow(deprecated)]
- Field::new_dict(
- field.name().unwrap_or_default(),
- get_data_type(field, true),
- field.nullable(),
- dictionary.id(),
- dictionary.isOrdered(),
- )
- } else {
- Field::new(
- field.name().unwrap_or_default(),
- get_data_type(field, true),
- field.nullable(),
- )
- };
+ try_field_from(field).expect("invalid IPC field")
+ }
+}
- let mut metadata_map = HashMap::default();
- if let Some(list) = field.custom_metadata() {
- for kv in list {
- if let (Some(k), Some(v)) = (kv.key(), kv.value()) {
- metadata_map.insert(k.to_string(), v.to_string());
- }
+/// Convert an IPC Field to Arrow Field
+fn try_field_from(field: crate::Field) -> Result<Field, ArrowError> {
+ let arrow_field = if let Some(dictionary) = field.dictionary() {
+ #[allow(deprecated)]
+ Field::new_dict(
+ field.name().unwrap_or_default(),
+ get_data_type(field, true)?,
+ field.nullable(),
+ dictionary.id(),
+ dictionary.isOrdered(),
+ )
+ } else {
+ Field::new(
+ field.name().unwrap_or_default(),
+ get_data_type(field, true)?,
+ field.nullable(),
+ )
+ };
+
+ let mut metadata_map = HashMap::default();
+ if let Some(list) = field.custom_metadata() {
+ for kv in list {
+ if let (Some(k), Some(v)) = (kv.key(), kv.value()) {
+ metadata_map.insert(k.to_string(), v.to_string());
}
}
-
- arrow_field.with_metadata(metadata_map)
}
+
+ Ok(arrow_field.with_metadata(metadata_map))
}
/// Deserialize an ipc [`crate::Schema`] from flat buffers to an arrow
[Schema].
+#[deprecated(since = "60.0.0", note = "Use `try_fb_to_schema` instead")]
pub fn fb_to_schema(fb: crate::Schema) -> Schema {
+ try_fb_to_schema(fb).expect("invalid IPC schema")
+}
+
+/// Deserialize an ipc [`crate::Schema`] from flat buffers to an arrow
[Schema].
+///
+/// Returns an error on schema messages that the flatbuffer
+/// verifier accepts but that are not valid Arrow.
+pub fn try_fb_to_schema(fb: crate::Schema) -> Result<Schema, ArrowError> {
let mut fields: Vec<Field> = vec![];
- let c_fields = fb.fields().unwrap();
+ let c_fields = fb
+ .fields()
+ .ok_or_else(|| ArrowError::ParseError("IPC schema has no
fields".to_string()))?;
let len = c_fields.len();
for i in 0..len {
let c_field: crate::Field = c_fields.get(i);
match c_field.type_type() {
crate::Type::Decimal if fb.endianness() == crate::Endianness::Big
=> {
- unimplemented!("Big Endian is not supported for Decimal!")
+ return Err(ArrowError::ParseError(
+ "Big Endian is not supported for Decimal!".to_string(),
+ ));
}
_ => (),
};
- fields.push(c_field.into());
+ fields.push(try_field_from(c_field)?);
}
let mut metadata: HashMap<String, String> = HashMap::default();
@@ -220,14 +241,14 @@ pub fn fb_to_schema(fb: crate::Schema) -> Schema {
}
}
}
- Schema::new_with_metadata(fields, metadata)
+ Ok(Schema::new_with_metadata(fields, metadata))
}
/// Try deserialize flat buffer format bytes into a schema
pub fn try_schema_from_flatbuffer_bytes(bytes: &[u8]) -> Result<Schema,
ArrowError> {
if let Ok(ipc) = crate::root_as_message(bytes) {
- if let Some(schema) = ipc.header_as_schema().map(fb_to_schema) {
- Ok(schema)
+ if let Some(schema) = ipc.header_as_schema() {
+ try_fb_to_schema(schema)
} else {
Err(ArrowError::ParseError(
"Unable to get head as schema".to_string(),
@@ -287,15 +308,27 @@ pub fn try_schema_from_ipc_buffer(buffer: &[u8]) ->
Result<Schema, ArrowError> {
let ipc_schema = msg.header_as_schema().ok_or_else(|| {
ArrowError::ParseError("Unable to convert flight info to a
schema".to_string())
})?;
- Ok(fb_to_schema(ipc_schema))
+ try_fb_to_schema(ipc_schema)
}
/// Get the Arrow data type from the flatbuffer Field table
-pub(crate) fn get_data_type(field: crate::Field, may_be_dictionary: bool) ->
DataType {
+pub(crate) fn get_data_type(
+ field: crate::Field,
+ may_be_dictionary: bool,
+) -> Result<DataType, ArrowError> {
+ // helper: the flatbuffer verifier accepts a Field whose `type_type` tag
does
+ // not match the union value present, so every `type_as_*` accessor can
return
+ // None. surface that as an error instead of unwrapping.
+ fn type_err(what: &str) -> ArrowError {
+ ArrowError::ParseError(format!("IPC schema field is missing its {what}
type"))
+ }
+
if let Some(dictionary) = field.dictionary()
&& may_be_dictionary
{
- let int = dictionary.indexType().unwrap();
+ let int = dictionary
+ .indexType()
+ .ok_or_else(|| type_err("dictionary index"))?;
let index_type = match (int.bitWidth(), int.is_signed()) {
(8, true) => DataType::Int8,
(8, false) => DataType::UInt8,
@@ -305,16 +338,25 @@ pub(crate) fn get_data_type(field: crate::Field,
may_be_dictionary: bool) -> Dat
(32, false) => DataType::UInt32,
(64, true) => DataType::Int64,
(64, false) => DataType::UInt64,
- _ => panic!("Unexpected bitwidth and signed"),
+ _ => {
+ return Err(ArrowError::ParseError(format!(
+ "Index type with bit width of {} and signed of {} not
supported",
+ int.bitWidth(),
+ int.is_signed()
+ )));
+ }
};
- return DataType::Dictionary(Box::new(index_type),
Box::new(get_data_type(field, false)));
+ return Ok(DataType::Dictionary(
+ Box::new(index_type),
+ Box::new(get_data_type(field, false)?),
+ ));
}
- match field.type_type() {
+ let data_type = match field.type_type() {
crate::Type::Null => DataType::Null,
crate::Type::Bool => DataType::Boolean,
crate::Type::Int => {
- let int = field.type_as_int().unwrap();
+ let int = field.type_as_int().ok_or_else(|| type_err("int"))?;
match (int.bitWidth(), int.is_signed()) {
(8, true) => DataType::Int8,
(8, false) => DataType::UInt8,
@@ -324,10 +366,12 @@ pub(crate) fn get_data_type(field: crate::Field,
may_be_dictionary: bool) -> Dat
(32, false) => DataType::UInt32,
(64, true) => DataType::Int64,
(64, false) => DataType::UInt64,
- z => panic!(
- "Int type with bit width of {} and signed of {} not
supported",
- z.0, z.1
- ),
+ z => {
+ return Err(ArrowError::ParseError(format!(
+ "Int type with bit width of {} and signed of {} not
supported",
+ z.0, z.1
+ )));
+ }
}
}
crate::Type::Binary => DataType::Binary,
@@ -337,41 +381,57 @@ pub(crate) fn get_data_type(field: crate::Field,
may_be_dictionary: bool) -> Dat
crate::Type::Utf8View => DataType::Utf8View,
crate::Type::LargeUtf8 => DataType::LargeUtf8,
crate::Type::FixedSizeBinary => {
- let fsb = field.type_as_fixed_size_binary().unwrap();
+ let fsb = field
+ .type_as_fixed_size_binary()
+ .ok_or_else(|| type_err("fixed size binary"))?;
DataType::FixedSizeBinary(fsb.byteWidth())
}
crate::Type::FloatingPoint => {
- let float = field.type_as_floating_point().unwrap();
+ let float = field
+ .type_as_floating_point()
+ .ok_or_else(|| type_err("floating point"))?;
match float.precision() {
crate::Precision::HALF => DataType::Float16,
crate::Precision::SINGLE => DataType::Float32,
crate::Precision::DOUBLE => DataType::Float64,
- z => panic!("FloatingPoint type with precision of {z:?} not
supported"),
+ z => {
+ return Err(ArrowError::ParseError(format!(
+ "FloatingPoint type with precision of {z:?} not
supported"
+ )));
+ }
}
}
crate::Type::Date => {
- let date = field.type_as_date().unwrap();
+ let date = field.type_as_date().ok_or_else(|| type_err("date"))?;
match date.unit() {
crate::DateUnit::DAY => DataType::Date32,
crate::DateUnit::MILLISECOND => DataType::Date64,
- z => panic!("Date type with unit of {z:?} not supported"),
+ z => {
+ return Err(ArrowError::ParseError(format!(
+ "Date type with unit of {z:?} not supported"
+ )));
+ }
}
}
crate::Type::Time => {
- let time = field.type_as_time().unwrap();
+ let time = field.type_as_time().ok_or_else(|| type_err("time"))?;
match (time.bitWidth(), time.unit()) {
(32, crate::TimeUnit::SECOND) =>
DataType::Time32(TimeUnit::Second),
(32, crate::TimeUnit::MILLISECOND) =>
DataType::Time32(TimeUnit::Millisecond),
(64, crate::TimeUnit::MICROSECOND) =>
DataType::Time64(TimeUnit::Microsecond),
(64, crate::TimeUnit::NANOSECOND) =>
DataType::Time64(TimeUnit::Nanosecond),
- z => panic!(
- "Time type with bit width of {} and unit of {:?} not
supported",
- z.0, z.1
- ),
+ z => {
+ return Err(ArrowError::ParseError(format!(
+ "Time type with bit width of {} and unit of {:?} not
supported",
+ z.0, z.1
+ )));
+ }
}
}
crate::Type::Timestamp => {
- let timestamp = field.type_as_timestamp().unwrap();
+ let timestamp = field
+ .type_as_timestamp()
+ .ok_or_else(|| type_err("timestamp"))?;
let timezone: Option<_> = timestamp.timezone().map(|tz| tz.into());
match timestamp.unit() {
crate::TimeUnit::SECOND =>
DataType::Timestamp(TimeUnit::Second, timezone),
@@ -382,132 +442,164 @@ pub(crate) fn get_data_type(field: crate::Field,
may_be_dictionary: bool) -> Dat
DataType::Timestamp(TimeUnit::Microsecond, timezone)
}
crate::TimeUnit::NANOSECOND =>
DataType::Timestamp(TimeUnit::Nanosecond, timezone),
- z => panic!("Timestamp type with unit of {z:?} not supported"),
+ z => {
+ return Err(ArrowError::ParseError(format!(
+ "Timestamp type with unit of {z:?} not supported"
+ )));
+ }
}
}
crate::Type::Interval => {
- let interval = field.type_as_interval().unwrap();
+ let interval = field
+ .type_as_interval()
+ .ok_or_else(|| type_err("interval"))?;
match interval.unit() {
crate::IntervalUnit::YEAR_MONTH =>
DataType::Interval(IntervalUnit::YearMonth),
crate::IntervalUnit::DAY_TIME =>
DataType::Interval(IntervalUnit::DayTime),
crate::IntervalUnit::MONTH_DAY_NANO => {
DataType::Interval(IntervalUnit::MonthDayNano)
}
- z => panic!("Interval type with unit of {z:?} unsupported"),
+ z => {
+ return Err(ArrowError::ParseError(format!(
+ "Interval type with unit of {z:?} unsupported"
+ )));
+ }
}
}
crate::Type::Duration => {
- let duration = field.type_as_duration().unwrap();
+ let duration = field
+ .type_as_duration()
+ .ok_or_else(|| type_err("duration"))?;
match duration.unit() {
crate::TimeUnit::SECOND =>
DataType::Duration(TimeUnit::Second),
crate::TimeUnit::MILLISECOND =>
DataType::Duration(TimeUnit::Millisecond),
crate::TimeUnit::MICROSECOND =>
DataType::Duration(TimeUnit::Microsecond),
crate::TimeUnit::NANOSECOND =>
DataType::Duration(TimeUnit::Nanosecond),
- z => panic!("Duration type with unit of {z:?} unsupported"),
+ z => {
+ return Err(ArrowError::ParseError(format!(
+ "Duration type with unit of {z:?} unsupported"
+ )));
+ }
}
}
crate::Type::List => {
- let children = field.children().unwrap();
- if children.len() != 1 {
- panic!("expect a list to have one child")
- }
- DataType::List(Arc::new(children.get(0).into()))
+ let child = single_child(&field, "list")?;
+ DataType::List(Arc::new(try_field_from(child)?))
}
crate::Type::LargeList => {
- let children = field.children().unwrap();
- if children.len() != 1 {
- panic!("expect a large list to have one child")
- }
- DataType::LargeList(Arc::new(children.get(0).into()))
+ let child = single_child(&field, "large list")?;
+ DataType::LargeList(Arc::new(try_field_from(child)?))
}
crate::Type::ListView => {
- let children = field.children().unwrap();
- if children.len() != 1 {
- panic!("expect a listview to have one child")
- }
- DataType::ListView(Arc::new(children.get(0).into()))
+ let child = single_child(&field, "listview")?;
+ DataType::ListView(Arc::new(try_field_from(child)?))
}
crate::Type::LargeListView => {
- let children = field.children().unwrap();
- if children.len() != 1 {
- panic!("expect a large listview to have one child")
- }
- DataType::LargeListView(Arc::new(children.get(0).into()))
+ let child = single_child(&field, "large listview")?;
+ DataType::LargeListView(Arc::new(try_field_from(child)?))
}
crate::Type::FixedSizeList => {
- let children = field.children().unwrap();
- if children.len() != 1 {
- panic!("expect a list to have one child")
- }
- let fsl = field.type_as_fixed_size_list().unwrap();
- DataType::FixedSizeList(Arc::new(children.get(0).into()),
fsl.listSize())
+ let child = single_child(&field, "list")?;
+ let fsl = field
+ .type_as_fixed_size_list()
+ .ok_or_else(|| type_err("fixed size list"))?;
+ DataType::FixedSizeList(Arc::new(try_field_from(child)?),
fsl.listSize())
}
crate::Type::Struct_ => {
let fields = match field.children() {
- Some(children) => children.iter().map(Field::from).collect(),
+ Some(children) => children
+ .iter()
+ .map(try_field_from)
+ .collect::<Result<Vec<_>, _>>()?
+ .into(),
None => Fields::empty(),
};
DataType::Struct(fields)
}
crate::Type::RunEndEncoded => {
- let children = field.children().unwrap();
+ let children = field
+ .children()
+ .ok_or_else(|| ArrowError::ParseError("IPC list has no
children".to_string()))?;
if children.len() != 2 {
- panic!(
+ return Err(ArrowError::ParseError(format!(
"RunEndEncoded type should have exactly two children.
Found {}",
children.len()
- )
+ )));
}
- let run_ends_field = children.get(0).into();
- let values_field = children.get(1).into();
+ let run_ends_field = try_field_from(children.get(0))?;
+ let values_field = try_field_from(children.get(1))?;
DataType::RunEndEncoded(Arc::new(run_ends_field),
Arc::new(values_field))
}
crate::Type::Map => {
- let map = field.type_as_map().unwrap();
- let children = field.children().unwrap();
- if children.len() != 1 {
- panic!("expect a map to have one child")
- }
- DataType::Map(Arc::new(children.get(0).into()), map.keysSorted())
+ let map = field.type_as_map().ok_or_else(|| type_err("map"))?;
+ let child = single_child(&field, "map")?;
+ DataType::Map(Arc::new(try_field_from(child)?), map.keysSorted())
}
crate::Type::Decimal => {
- let fsb = field.type_as_decimal().unwrap();
+ let fsb = field.type_as_decimal().ok_or_else(||
type_err("decimal"))?;
let bit_width = fsb.bitWidth();
- let precision: u8 = fsb.precision().try_into().unwrap();
- let scale: i8 = fsb.scale().try_into().unwrap();
+ let precision: u8 = fsb.precision().try_into().map_err(|_| {
+ ArrowError::ParseError(format!("Invalid decimal precision {}",
fsb.precision()))
+ })?;
+ let scale: i8 = fsb.scale().try_into().map_err(|_| {
+ ArrowError::ParseError(format!("Invalid decimal scale {}",
fsb.scale()))
+ })?;
match bit_width {
32 => DataType::Decimal32(precision, scale),
64 => DataType::Decimal64(precision, scale),
128 => DataType::Decimal128(precision, scale),
256 => DataType::Decimal256(precision, scale),
- _ => panic!("Unexpected decimal bit width {bit_width}"),
+ _ => {
+ return Err(ArrowError::ParseError(format!(
+ "Unexpected decimal bit width {bit_width}"
+ )));
+ }
}
}
crate::Type::Union => {
- let union = field.type_as_union().unwrap();
+ let union = field.type_as_union().ok_or_else(||
type_err("union"))?;
let union_mode = match union.mode() {
crate::UnionMode::Dense => UnionMode::Dense,
crate::UnionMode::Sparse => UnionMode::Sparse,
- mode => panic!("Unexpected union mode: {mode:?}"),
+ mode => {
+ return Err(ArrowError::ParseError(format!(
+ "Unexpected union mode: {mode:?}"
+ )));
+ }
};
let mut fields = vec![];
if let Some(children) = field.children() {
for i in 0..children.len() {
- fields.push(Field::from(children.get(i)));
+ fields.push(try_field_from(children.get(i))?);
}
};
let fields = match union.typeIds() {
None => UnionFields::from_fields(fields),
- Some(ids) => UnionFields::try_new(ids.iter().map(|i| i as i8),
fields)
- .expect("invalid union field"),
+ Some(ids) => UnionFields::try_new(ids.iter().map(|i| i as i8),
fields)?,
};
DataType::Union(fields, union_mode)
}
- t => unimplemented!("Type {:?} not supported", t),
+ t => return Err(ArrowError::ParseError(format!("Type {t:?} not
supported"))),
+ };
+ Ok(data_type)
+}
+
+/// Return the single child of a flatbuffer Field, erroring if it does not have
+/// exactly one.
+fn single_child<'a>(field: &crate::Field<'a>, kind: &str) ->
Result<crate::Field<'a>, ArrowError> {
+ let children = field
+ .children()
+ .ok_or_else(|| ArrowError::ParseError(format!("IPC {kind} has no
children")))?;
+ if children.len() != 1 {
+ return Err(ArrowError::ParseError(format!(
+ "expect a {kind} to have one child"
+ )));
}
+ Ok(children.get(0))
}
pub(crate) struct FBFieldType<'b> {
@@ -1002,6 +1094,76 @@ impl MessageBuffer {
mod tests {
use super::*;
+ /// Build a schema flatbuffer with no `fields` vector, using `build_field`
to
+ /// optionally add a single field. Returns bytes that the flatbuffer
verifier
+ /// accepts, so `try_fb_to_schema` is the layer that must reject them.
+ fn schema_bytes_without_fields() -> Vec<u8> {
+ let mut fbb = FlatBufferBuilder::new();
+ // Schema table with `fields` left unset -> fb.fields() == None.
+ let schema = crate::SchemaBuilder::new(&mut fbb).finish();
+ fbb.finish(schema, None);
+ fbb.finished_data().to_vec()
+ }
+
+ /// Finish `fbb` as a schema holding the single `field`, assert the
verifier
+ /// accepts the bytes but `try_fb_to_schema` rejects them (rather than
panics).
+ fn assert_single_field_schema_is_err<'a>(
+ mut fbb: FlatBufferBuilder<'a>,
+ field: WIPOffset<crate::Field<'a>>,
+ ) {
+ let fields = fbb.create_vector(&[field]);
+ let schema = {
+ let mut b = crate::SchemaBuilder::new(&mut fbb);
+ b.add_fields(fields);
+ b.finish()
+ };
+ fbb.finish(schema, None);
+ let fb = crate::root_as_schema(fbb.finished_data()).expect("verifier
should accept");
+ assert!(try_fb_to_schema(fb).is_err());
+ }
+
+ #[test]
+ fn try_fb_to_schema_missing_fields_is_err() {
+ let bytes = schema_bytes_without_fields();
+ // the verifier accepts it...
+ let fb = crate::root_as_schema(&bytes).expect("verifier should
accept");
+ // ...but conversion must return Err, not panic.
+ assert!(try_fb_to_schema(fb).is_err());
+ }
+
+ #[test]
+ fn try_fb_to_schema_unknown_float_precision_is_err() {
+ let mut fbb = FlatBufferBuilder::new();
+ let name = fbb.create_string("f");
+ let float = {
+ let mut b = crate::FloatingPointBuilder::new(&mut fbb);
+ // a precision the verifier does not range-check.
+ b.add_precision(crate::Precision(99));
+ b.finish()
+ };
+ let field = {
+ let mut b = crate::FieldBuilder::new(&mut fbb);
+ b.add_name(name);
+ b.add_type_type(crate::Type::FloatingPoint);
+ b.add_type_(float.as_union_value());
+ b.finish()
+ };
+ assert_single_field_schema_is_err(fbb, field);
+ }
+
+ #[test]
+ fn try_fb_to_schema_type_none_is_err() {
+ let mut fbb = FlatBufferBuilder::new();
+ let name = fbb.create_string("f");
+ let field = {
+ let mut b = crate::FieldBuilder::new(&mut fbb);
+ b.add_name(name);
+ // type_type left as the union default (Type::NONE).
+ b.finish()
+ };
+ assert_single_field_schema_is_err(fbb, field);
+ }
+
#[test]
fn convert_schema_round_trip() {
let md: HashMap<String, String> = [("Key".to_string(),
"value".to_string())]
@@ -1218,7 +1380,7 @@ mod tests {
// read back fields
let ipc = crate::root_as_schema(fb.finished_data()).unwrap();
- let schema2 = fb_to_schema(ipc);
+ let schema2 = try_fb_to_schema(ipc).unwrap();
assert_eq!(schema, schema2);
}
@@ -1264,7 +1426,10 @@ mod tests {
assert_eq!(schema.endianness(), schema2.endianness());
assert!(schema.features().is_none());
assert!(schema2.features().is_none());
- assert_eq!(fb_to_schema(schema), fb_to_schema(schema2));
+ assert_eq!(
+ try_fb_to_schema(schema).unwrap(),
+ try_fb_to_schema(schema2).unwrap()
+ );
assert_eq!(ipc.version(), ipc2.version());
assert_eq!(ipc.header_type(), ipc2.header_type());
diff --git a/arrow-ipc/src/reader.rs b/arrow-ipc/src/reader.rs
index f3856b1d32..748330aad4 100644
--- a/arrow-ipc/src/reader.rs
+++ b/arrow-ipc/src/reader.rs
@@ -967,7 +967,7 @@ pub fn read_footer_length(buf: [u8; 10]) -> Result<usize,
ArrowError> {
/// # use arrow_array::*;
/// # use arrow_array::types::Int32Type;
/// # use arrow_buffer::Buffer;
-/// # use arrow_ipc::convert::fb_to_schema;
+/// # use arrow_ipc::convert::try_fb_to_schema;
/// # use arrow_ipc::reader::{FileDecoder, read_footer_length};
/// # use arrow_ipc::root_as_footer;
/// # use arrow_ipc::writer::FileWriter;
@@ -995,7 +995,7 @@ pub fn read_footer_length(buf: [u8; 10]) -> Result<usize,
ArrowError> {
/// let footer_len =
read_footer_length(buffer[trailer_start..].try_into().unwrap()).unwrap();
/// let footer = root_as_footer(&buffer[trailer_start -
footer_len..trailer_start]).unwrap();
///
-/// let back = fb_to_schema(footer.schema().unwrap());
+/// let back = try_fb_to_schema(footer.schema().unwrap()).unwrap();
/// assert_eq!(&back, schema.as_ref());
///
/// let mut decoder = FileDecoder::new(schema, footer.version());
@@ -1256,7 +1256,7 @@ impl FileReaderBuilder {
));
}
- let schema = Arc::new(crate::convert::fb_to_schema(ipc_schema));
+ let schema = Arc::new(crate::convert::try_fb_to_schema(ipc_schema)?);
let projected_schema = match &self.projection {
Some(projection) => Arc::new(schema.project(projection)?),
@@ -1604,7 +1604,7 @@ impl<R: Read> StreamReader<R> {
let schema = message.header_as_schema().ok_or_else(|| {
ArrowError::ParseError("Failed to parse schema from message
header".to_string())
})?;
- let schema = crate::convert::fb_to_schema(schema);
+ let schema = crate::convert::try_fb_to_schema(schema)?;
// Create an array of optional dictionary value arrays, one per field.
let dictionaries_by_id = HashMap::new();
@@ -1686,7 +1686,7 @@ impl<R: Read> StreamReader<R> {
let schema = message.header_as_schema().ok_or_else(|| {
ArrowError::ParseError("Failed to parse schema from
message header".to_string())
})?;
- let arrow_schema = crate::convert::fb_to_schema(schema);
+ let arrow_schema = crate::convert::try_fb_to_schema(schema)?;
IpcMessage::Schema(arrow_schema)
}
Message::MessageHeader::RecordBatch => {
@@ -1955,7 +1955,7 @@ impl<R: Read> MessageReader<R> {
mod tests {
use std::io::Cursor;
- use crate::convert::fb_to_schema;
+ use crate::convert::try_fb_to_schema;
use crate::writer::{
DictionaryTracker, IpcDataGenerator, IpcWriteOptions,
unslice_run_array, write_message,
};
@@ -2530,7 +2530,7 @@ mod tests {
let footer = root_as_footer(&buffer[trailer_start -
footer_len..trailer_start])
.map_err(|e| ArrowError::InvalidArgumentError(format!("Invalid
footer: {e}")))?;
- let schema = fb_to_schema(footer.schema().unwrap());
+ let schema = try_fb_to_schema(footer.schema().unwrap()).unwrap();
let mut decoder = unsafe {
FileDecoder::new(Arc::new(schema), footer.version())
@@ -3654,7 +3654,7 @@ mod tests {
let msg = parse_message(&schema_bytes).expect("parse_message");
let ipc_schema = msg.header_as_schema().expect("header_as_schema");
- let new_schema = fb_to_schema(ipc_schema);
+ let new_schema = try_fb_to_schema(ipc_schema).unwrap();
assert_eq!(schema, new_schema);
}
diff --git a/arrow-ipc/src/reader/stream.rs b/arrow-ipc/src/reader/stream.rs
index e18d32cea2..f136b6b69d 100644
--- a/arrow-ipc/src/reader/stream.rs
+++ b/arrow-ipc/src/reader/stream.rs
@@ -229,8 +229,12 @@ impl StreamDecoder {
));
}
- let ipc_schema =
message.header_as_schema().unwrap();
- let schema =
crate::convert::fb_to_schema(ipc_schema);
+ let ipc_schema =
message.header_as_schema().ok_or_else(|| {
+ ArrowError::ParseError(
+ "Unable to read IPC message as
schema".to_string(),
+ )
+ })?;
+ let schema =
crate::convert::try_fb_to_schema(ipc_schema)?;
self.state = DecoderState::default();
self.schema = Some(Arc::new(schema));
}
diff --git a/arrow-ipc/src/writer.rs b/arrow-ipc/src/writer.rs
index 20d3441a3f..a9cc68bc13 100644
--- a/arrow-ipc/src/writer.rs
+++ b/arrow-ipc/src/writer.rs
@@ -2771,7 +2771,7 @@ mod tests {
use arrow_buffer::ScalarBuffer;
use crate::MetadataVersion;
- use crate::convert::fb_to_schema;
+ use crate::convert::try_fb_to_schema;
use crate::reader::*;
use crate::root_as_footer;
@@ -4630,7 +4630,7 @@ mod tests {
let footer =
root_as_footer(&buffer[trailer_start -
footer_len..trailer_start]).unwrap();
- let schema = fb_to_schema(footer.schema().unwrap());
+ let schema = try_fb_to_schema(footer.schema().unwrap()).unwrap();
// Importantly we set `require_alignment`, checking that 16-byte
alignment is sufficient
// for `read_record_batch` later on to read the data in a
zero-copy manner.
@@ -4682,7 +4682,7 @@ mod tests {
let trailer_start = buffer.len() - 10;
let footer_len =
read_footer_length(buffer[trailer_start..].try_into().unwrap()).unwrap();
let footer = root_as_footer(&buffer[trailer_start -
footer_len..trailer_start]).unwrap();
- let schema = fb_to_schema(footer.schema().unwrap());
+ let schema = try_fb_to_schema(footer.schema().unwrap()).unwrap();
// Importantly we set `require_alignment`, otherwise the error later
is suppressed due to copying
// to an aligned buffer in `ArrayDataBuilder.build_aligned`.
diff --git a/arrow/examples/zero_copy_ipc.rs b/arrow/examples/zero_copy_ipc.rs
index 0c80572cf4..c05fce42fd 100644
--- a/arrow/examples/zero_copy_ipc.rs
+++ b/arrow/examples/zero_copy_ipc.rs
@@ -24,7 +24,7 @@ use arrow::array::{RecordBatch, record_batch};
use arrow::error::Result;
use arrow_buffer::Buffer;
use arrow_cast::pretty::pretty_format_batches;
-use arrow_ipc::convert::fb_to_schema;
+use arrow_ipc::convert::try_fb_to_schema;
use arrow_ipc::reader::{FileDecoder, read_footer_length};
use arrow_ipc::writer::FileWriter;
use arrow_ipc::{Block, root_as_footer};
@@ -101,7 +101,7 @@ impl IPCBufferDecoder {
let footer_len =
read_footer_length(buffer[trailer_start..].try_into().unwrap()).unwrap();
let footer = root_as_footer(&buffer[trailer_start -
footer_len..trailer_start]).unwrap();
- let schema = fb_to_schema(footer.schema().unwrap());
+ let schema = try_fb_to_schema(footer.schema().unwrap()).unwrap();
let mut decoder = FileDecoder::new(Arc::new(schema), footer.version());
diff --git a/parquet/src/arrow/schema/mod.rs b/parquet/src/arrow/schema/mod.rs
index 110c799b42..be5577b2c4 100644
--- a/parquet/src/arrow/schema/mod.rs
+++ b/parquet/src/arrow/schema/mod.rs
@@ -269,8 +269,10 @@ fn get_arrow_schema_from_metadata(encoded_meta: &str) ->
Result<Schema> {
match arrow_ipc::root_as_message(slice) {
Ok(message) => message
.header_as_schema()
- .map(arrow_ipc::convert::fb_to_schema)
- .ok_or_else(|| arrow_err!("the message is not Arrow
Schema")),
+ .ok_or_else(|| arrow_err!("the message is not Arrow
Schema"))
+ .and_then(|schema| {
+
arrow_ipc::convert::try_fb_to_schema(schema).map_err(Into::into)
+ }),
Err(err) => {
// The flatbuffers implementation returns an error on
verification error.
Err(arrow_err!(
