proto-common: decoding a malformed scalar, union type or CSV options panics instead of erroring #27004

Describe the bug

datafusion-proto-common's from_proto conversions panic at four sites when a peer sends input that a well-formed encoder never produces. Nearby code returns an error in the same situations.

1. Decimal scalar value of the wrong length. Decimal32Value, Decimal64Value, Decimal128Value and Decimal256Value pass their value bytes to vec_to_array, which panics when the length is not the width:

   datafusion/datafusion/proto-common/src/from_proto/mod.rs

   Lines 552 to 568 in 636590b

   	let array = vec_to_array(val.value.clone());
   	Self::Decimal32(Some(i32::from_be_bytes(array)), val.p as u8, val.s as i8)
   	}
   	Value::Decimal64Value(val) => {
   	let array = vec_to_array(val.value.clone());
   	Self::Decimal64(Some(i64::from_be_bytes(array)), val.p as u8, val.s as i8)
   	}
   	Value::Decimal128Value(val) => {
   	let array = vec_to_array(val.value.clone());
   	Self::Decimal128(
   	Some(i128::from_be_bytes(array)),
   	val.p as u8,
   	val.s as i8,
   	)
   	}
   	Value::Decimal256Value(val) => {
   	let array = vec_to_array(val.value.clone());

   
   

   datafusion/datafusion/proto-common/src/from_proto/mod.rs

   Lines 1421 to 1425 in 636590b

   	fn vec_to_array<T, const N: usize>(v: Vec<T>) -> [T; N] {
   	v.try_into().unwrap_or_else(|v: Vec<T>| {
   	panic!("Expected a Vec of length {} but it was {}", N, v.len())
   	})
   	}

2. Nested scalar whose batch disagrees with its kind. The nested ScalarValue arm decodes an IPC batch, then calls record_batch.column(0), which panics on a zero-field schema. It then calls as_list / as_fixed_size_list / as_list_view / as_struct / as_map, which panic when the column is not the type the value kind names:

   datafusion/datafusion/proto-common/src/from_proto/mod.rs

   Lines 523 to 544 in 636590b

   	let arr = record_batch.column(0);
   	match value {
   	Value::ListValue(_) => {
   	Self::List(arr.as_list::<i32>().to_owned().into())
   	}
   	Value::LargeListValue(_) => {
   	Self::LargeList(arr.as_list::<i64>().to_owned().into())
   	}
   	Value::FixedSizeListValue(_) => {
   	Self::FixedSizeList(arr.as_fixed_size_list().to_owned().into())
   	}
   	Value::ListViewValue(_) => {
   	Self::ListView(arr.as_list_view::<i32>().to_owned().into())
   	}
   	Value::LargeListViewValue(_) => {
   	Self::LargeListView(arr.as_list_view::<i64>().to_owned().into())
   	}
   	Value::StructValue(_) => {
   	Self::Struct(arr.as_struct().to_owned().into())
   	}
   	Value::MapValue(_) => Self::Map(arr.as_map().to_owned().into()),
   	_ => unreachable!(),

3. Union with more than 128 fields and no type_ids. UnionFields::from_fields panics with "UnionFields cannot contain more than 128 fields". The else branch just below calls UnionFields::try_new and maps its error. UnionFields::try_from_fields is the non-panicking equivalent:

   datafusion/datafusion/proto-common/src/from_proto/mod.rs

   Lines 319 to 325 in 636590b

   	let union_fields = if union.type_ids.is_empty() {
   	UnionFields::from_fields(union_fields)
   	} else {
   	let type_ids = union.type_ids.iter().map(|i| *i as i8);
   	UnionFields::try_new(type_ids, union_fields).map_err(|e| {
   	DataFusionError::from(e).context("Deserializing Union DataType")
   	})?

4. CSV options with an empty delimiter or quote. These are indexed with [0]:

   datafusion/datafusion/proto-common/src/from_proto/mod.rs

   Lines 1129 to 1130 in 636590b

   	delimiter: proto_opts.delimiter[0],
   	quote: proto_opts.quote[0],

The same code is in 55.1.0, at src/from_proto/mod.rs lines 549-565 and 1256, 521-541, 323, and 1009-1010.

To Reproduce

Decode a physical plan with datafusion_proto::bytes::physical_plan_from_bytes (or a ScalarValue / Schema with TryFrom), where the input contains one of:

1. a Decimal32Value literal whose value is 3 bytes long;
2. a ListValue literal whose IPC batch holds a single Int32 column;
3. a schema with a Union field of 129 children and an empty type_ids;
4. CsvOptions with an empty delimiter.

What we observed, and what we did not. We decode plan stages from peers, so we keep a patched copy of 55.1.0 in which each site returns an error. In our CI, inputs 1-3 reach their sites in that copy and get the error placed there. A test pins each error message, and a mutation of that message makes the test fail. We have not run the unpatched crate on these inputs. The panics above are the documented behaviour of the callees (vec_to_array's own panic!, as_list's "panicking if not possible", from_fields's expect). Input 4 has no test on our side.

Expected behavior

Each site returns Err naming the bad field: as_*_opt accessors for site 2, UnionFields::try_from_fields for site 3, and .first().ok_or_else(..) for site 4. The caller can then refuse the plan and keep running.

Additional context

This matters wherever a process decodes plans it did not encode itself, such as a distributed worker that receives plan stages from peers. Under panic = "abort", one malformed message stops the whole process. #24726 fixed the same class of problem for Constraint and TableParquetOptions. #26998 and #26999 are related cases in datafusion-proto and datafusion-physical-plan.

The patch is small and each site has a decode test. I can open a PR if that is welcome.