Skip to content

Commit 1b40ffa

Browse files
compheadcomphead
authored
Review use of panic in datafusion-proto crate (#3365)
* removed panics from proto crate * fixes * fixed comments * fix build * fixed comments * Fixing proto json * formatting * fix build * addresed comments
1 parent 4258751 commit 1b40ffa

5 files changed

Lines changed: 134 additions & 129 deletions

File tree

‎datafusion/proto/build.rs‎

Lines changed: 10 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,9 @@ fn main() -> Result<(), String> {
3232
fn build() -> Result<(), String> {
3333
use std::io::Write;
3434

35-
let out = std::path::PathBuf::from(std::env::var("OUT_DIR").unwrap());
35+
let out = std::path::PathBuf::from(
36+
std::env::var("OUT_DIR").expect("Cannot find OUT_DIR environment vairable"),
37+
);
3638
let descriptor_path = out.join("proto_descriptor.bin");
3739

3840
prost_build::Config::new()
@@ -42,10 +44,15 @@ fn build() -> Result<(), String> {
4244
.compile_protos(&["proto/datafusion.proto"], &["proto"])
4345
.map_err(|e| format!("protobuf compilation failed: {}", e))?;
4446

45-
let descriptor_set = std::fs::read(descriptor_path).unwrap();
47+
let descriptor_set = std::fs::read(&descriptor_path)
48+
.expect(&*format!("Cannot read {:?}", &descriptor_path));
49+
4650
pbjson_build::Builder::new()
4751
.register_descriptors(&descriptor_set)
48-
.unwrap()
52+
.expect(&*format!(
53+
"Cannot register descriptors {:?}",
54+
&descriptor_set
55+
))
4956
.build(&[".datafusion"])
5057
.map_err(|e| format!("pbjson compilation failed: {}", e))?;
5158

‎datafusion/proto/src/bytes/mod.rs‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -137,7 +137,8 @@ pub fn logical_plan_to_bytes_with_extension_codec(
137137
/// Deserialize a LogicalPlan from json
138138
#[cfg(feature = "json")]
139139
pub fn logical_plan_from_json(json: &str, ctx: &SessionContext) -> Result<LogicalPlan> {
140-
let back: protobuf::LogicalPlanNode = serde_json::from_str(json).unwrap();
140+
let back: protobuf::LogicalPlanNode = serde_json::from_str(json)
141+
.map_err(|e| DataFusionError::Plan(format!("Error serializing plan: {}", e)))?;
141142
let extension_codec = DefaultExtensionCodec {};
142143
back.try_into_logical_plan(ctx, &extension_codec)
143144
}

‎datafusion/proto/src/from_proto.rs‎

Lines changed: 8 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -378,11 +378,13 @@ impl From<&protobuf::StringifiedPlan> for StringifiedPlan {
378378
plan_type: match stringified_plan
379379
.plan_type
380380
.as_ref()
381-
.unwrap()
382-
.plan_type_enum
383-
.as_ref()
384-
.unwrap()
385-
{
381+
.and_then(|pt| pt.plan_type_enum.as_ref())
382+
.unwrap_or_else(|| {
383+
panic!(
384+
"Cannot create protobuf::StringifiedPlan from {:?}",
385+
stringified_plan
386+
)
387+
}) {
386388
InitialLogicalPlan(_) => PlanType::InitialLogicalPlan,
387389
OptimizedLogicalPlan(OptimizedLogicalPlanType { optimizer_name }) => {
388390
PlanType::OptimizedLogicalPlan {
@@ -1341,6 +1343,7 @@ impl From<protobuf::IntervalUnit> for IntervalUnit {
13411343
}
13421344
}
13431345

1346+
// panic here because no better way to convert from Vec to Array
13441347
fn vec_to_array<T, const N: usize>(v: Vec<T>) -> [T; N] {
13451348
v.try_into().unwrap_or_else(|v: Vec<T>| {
13461349
panic!("Expected a Vec of length {} but it was {}", N, v.len())

‎datafusion/proto/src/logical_plan.rs‎

Lines changed: 7 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -381,12 +381,12 @@ impl AsLogicalPlan for LogicalPlanNode {
381381
FileFormatType::Avro(..) => Arc::new(AvroFormat::default()),
382382
};
383383

384-
// let table_path = ListingTableUrl::parse(&scan.paths)?;
385384
let table_paths = &scan
386385
.paths
387386
.iter()
388-
.map(|p| ListingTableUrl::parse(p).unwrap())
389-
.collect::<Vec<ListingTableUrl>>();
387+
.map(ListingTableUrl::parse)
388+
.collect::<Result<Vec<_>, _>>()?;
389+
390390
let options = ListingOptions {
391391
file_extension: scan.file_extension.clone(),
392392
format: file_format,
@@ -635,7 +635,10 @@ impl AsLogicalPlan for LogicalPlanNode {
635635
)));
636636
}
637637

638-
let mut builder = LogicalPlanBuilder::from(input_plans.pop().unwrap());
638+
let first = input_plans.pop().ok_or_else(|| DataFusionError::Internal(String::from(
639+
"Protobuf deserialization error, Union was require at least two input.",
640+
)))?;
641+
let mut builder = LogicalPlanBuilder::from(first);
639642
for plan in input_plans {
640643
builder = builder.union(plan)?;
641644
}

‎datafusion/proto/src/to_proto.rs‎

Lines changed: 107 additions & 116 deletions
Original file line numberDiff line numberDiff line change
@@ -224,10 +224,12 @@ impl From<&DataType> for protobuf::arrow_type::ArrowTypeEnum {
224224
fractional: *fractional as u64,
225225
}),
226226
DataType::Decimal256(_, _) => {
227-
unimplemented!("The Decimal256 data type is not yet supported")
227+
unimplemented!("Proto serialization error: The Decimal256 data type is not yet supported")
228228
}
229229
DataType::Map(_, _) => {
230-
unimplemented!("The Map data type is not yet supported")
230+
unimplemented!(
231+
"Proto serialization error: The Map data type is not yet supported"
232+
)
231233
}
232234
}
233235
}
@@ -612,7 +614,7 @@ impl TryFrom<&Expr> for protobuf::LogicalExprNode {
612614
expr_type: Some(ExprType::AggregateExpr(aggregate_expr)),
613615
}
614616
}
615-
Expr::ScalarVariable(_, _) => unimplemented!(),
617+
Expr::ScalarVariable(_, _) => return Err(Error::General("Proto serialization error: Scalar Variable not supported".to_string())),
616618
Expr::ScalarFunction { ref fun, ref args } => {
617619
let fun: protobuf::ScalarFunction = fun.try_into()?;
618620
let args: Vec<Self> = args
@@ -820,7 +822,7 @@ impl TryFrom<&Expr> for protobuf::LogicalExprNode {
820822
Expr::ScalarSubquery(_) | Expr::InSubquery { .. } | Expr::Exists { .. } => {
821823
// we would need to add logical plan operators to datafusion.proto to support this
822824
// see discussion in https://github.com/apache/arrow-datafusion/issues/2565
823-
unimplemented!("subquery expressions are not supported yet")
825+
return Err(Error::General("Proto serialization error: Expr::ScalarSubquery(_) | Expr::InSubquery { .. } | Expr::Exists { .. } not supported".to_string()))
824826
}
825827
Expr::GetIndexedField { key, expr } => Self {
826828
expr_type: Some(ExprType::GetIndexedField(Box::new(
@@ -865,7 +867,8 @@ impl TryFrom<&Expr> for protobuf::LogicalExprNode {
865867
})),
866868
},
867869

868-
Expr::QualifiedWildcard { .. } | Expr::TryCast { .. } => unimplemented!(),
870+
Expr::QualifiedWildcard { .. } | Expr::TryCast { .. } =>
871+
return Err(Error::General("Proto serialization error: Expr::QualifiedWildcard { .. } | Expr::TryCast { .. } not supported".to_string())),
869872
};
870873

871874
Ok(expr_node)
@@ -945,127 +948,115 @@ impl TryFrom<&ScalarValue> for protobuf::ScalarValue {
945948
Value::LargeUtf8Value(s.to_owned())
946949
})
947950
}
948-
scalar::ScalarValue::List(value, boxed_field) => {
949-
println!("Current field of list: {:?}", boxed_field);
950-
match value {
951-
Some(values) => {
952-
if values.is_empty() {
953-
protobuf::ScalarValue {
954-
value: Some(protobuf::scalar_value::Value::ListValue(
955-
protobuf::ScalarListValue {
956-
field: Some(boxed_field.as_ref().into()),
957-
values: Vec::new(),
958-
},
959-
)),
951+
scalar::ScalarValue::List(value, boxed_field) => match value {
952+
Some(values) => {
953+
if values.is_empty() {
954+
protobuf::ScalarValue {
955+
value: Some(protobuf::scalar_value::Value::ListValue(
956+
protobuf::ScalarListValue {
957+
field: Some(boxed_field.as_ref().into()),
958+
values: Vec::new(),
959+
},
960+
)),
961+
}
962+
} else {
963+
let scalar_type = match boxed_field.data_type() {
964+
DataType::List(field) => field.as_ref().data_type(),
965+
unsupported => {
966+
return Err(Error::General(format!("Proto serialization error: {:?} not supported to convert to DataType::List", unsupported)));
960967
}
961-
} else {
962-
let scalar_type = match boxed_field.data_type() {
963-
DataType::List(field) => field.as_ref().data_type(),
964-
unsupported => {
965-
todo!("Proper error handling {}", unsupported)
966-
}
967-
};
968-
println!("Current scalar type for list: {:?}", scalar_type);
968+
};
969969

970-
let type_checked_values: Vec<protobuf::ScalarValue> = values
971-
.iter()
972-
.map(|scalar| match (scalar, scalar_type) {
973-
(
974-
scalar::ScalarValue::List(_, list_type),
975-
DataType::List(field),
976-
) => {
977-
if let DataType::List(list_field) =
978-
list_type.data_type()
970+
let type_checked_values: Vec<protobuf::ScalarValue> = values
971+
.iter()
972+
.map(|scalar| match (scalar, scalar_type) {
973+
(
974+
scalar::ScalarValue::List(_, list_type),
975+
DataType::List(field),
976+
) => {
977+
if let DataType::List(list_field) =
978+
list_type.data_type()
979+
{
980+
let scalar_datatype = field.data_type();
981+
let list_datatype = list_field.data_type();
982+
if std::mem::discriminant(list_datatype)
983+
!= std::mem::discriminant(scalar_datatype)
979984
{
980-
let scalar_datatype = field.data_type();
981-
let list_datatype = list_field.data_type();
982-
if std::mem::discriminant(list_datatype)
983-
!= std::mem::discriminant(scalar_datatype)
984-
{
985-
return Err(
986-
Error::inconsistent_list_typing(
987-
list_datatype,
988-
scalar_datatype,
989-
),
990-
);
991-
}
992-
scalar.try_into()
993-
} else {
994-
Err(Error::inconsistent_list_designated(
995-
scalar,
996-
boxed_field.data_type(),
997-
))
985+
return Err(Error::inconsistent_list_typing(
986+
list_datatype,
987+
scalar_datatype,
988+
));
998989
}
999-
}
1000-
(
1001-
scalar::ScalarValue::Boolean(_),
1002-
DataType::Boolean,
1003-
) => scalar.try_into(),
1004-
(
1005-
scalar::ScalarValue::Float32(_),
1006-
DataType::Float32,
1007-
) => scalar.try_into(),
1008-
(
1009-
scalar::ScalarValue::Float64(_),
1010-
DataType::Float64,
1011-
) => scalar.try_into(),
1012-
(scalar::ScalarValue::Int8(_), DataType::Int8) => {
1013990
scalar.try_into()
991+
} else {
992+
Err(Error::inconsistent_list_designated(
993+
scalar,
994+
boxed_field.data_type(),
995+
))
1014996
}
1015-
(scalar::ScalarValue::Int16(_), DataType::Int16) => {
1016-
scalar.try_into()
1017-
}
1018-
(scalar::ScalarValue::Int32(_), DataType::Int32) => {
1019-
scalar.try_into()
1020-
}
1021-
(scalar::ScalarValue::Int64(_), DataType::Int64) => {
1022-
scalar.try_into()
1023-
}
1024-
(scalar::ScalarValue::UInt8(_), DataType::UInt8) => {
1025-
scalar.try_into()
1026-
}
1027-
(
1028-
scalar::ScalarValue::UInt16(_),
1029-
DataType::UInt16,
1030-
) => scalar.try_into(),
1031-
(
1032-
scalar::ScalarValue::UInt32(_),
1033-
DataType::UInt32,
1034-
) => scalar.try_into(),
1035-
(
1036-
scalar::ScalarValue::UInt64(_),
1037-
DataType::UInt64,
1038-
) => scalar.try_into(),
1039-
(scalar::ScalarValue::Utf8(_), DataType::Utf8) => {
1040-
scalar.try_into()
1041-
}
1042-
(
1043-
scalar::ScalarValue::LargeUtf8(_),
1044-
DataType::LargeUtf8,
1045-
) => scalar.try_into(),
1046-
_ => Err(Error::inconsistent_list_designated(
1047-
scalar,
1048-
boxed_field.data_type(),
1049-
)),
1050-
})
1051-
.collect::<Result<Vec<_>, _>>()?;
1052-
protobuf::ScalarValue {
1053-
value: Some(protobuf::scalar_value::Value::ListValue(
1054-
protobuf::ScalarListValue {
1055-
field: Some(boxed_field.as_ref().into()),
1056-
values: type_checked_values,
1057-
},
997+
}
998+
(scalar::ScalarValue::Boolean(_), DataType::Boolean) => {
999+
scalar.try_into()
1000+
}
1001+
(scalar::ScalarValue::Float32(_), DataType::Float32) => {
1002+
scalar.try_into()
1003+
}
1004+
(scalar::ScalarValue::Float64(_), DataType::Float64) => {
1005+
scalar.try_into()
1006+
}
1007+
(scalar::ScalarValue::Int8(_), DataType::Int8) => {
1008+
scalar.try_into()
1009+
}
1010+
(scalar::ScalarValue::Int16(_), DataType::Int16) => {
1011+
scalar.try_into()
1012+
}
1013+
(scalar::ScalarValue::Int32(_), DataType::Int32) => {
1014+
scalar.try_into()
1015+
}
1016+
(scalar::ScalarValue::Int64(_), DataType::Int64) => {
1017+
scalar.try_into()
1018+
}
1019+
(scalar::ScalarValue::UInt8(_), DataType::UInt8) => {
1020+
scalar.try_into()
1021+
}
1022+
(scalar::ScalarValue::UInt16(_), DataType::UInt16) => {
1023+
scalar.try_into()
1024+
}
1025+
(scalar::ScalarValue::UInt32(_), DataType::UInt32) => {
1026+
scalar.try_into()
1027+
}
1028+
(scalar::ScalarValue::UInt64(_), DataType::UInt64) => {
1029+
scalar.try_into()
1030+
}
1031+
(scalar::ScalarValue::Utf8(_), DataType::Utf8) => {
1032+
scalar.try_into()
1033+
}
1034+
(
1035+
scalar::ScalarValue::LargeUtf8(_),
1036+
DataType::LargeUtf8,
1037+
) => scalar.try_into(),
1038+
_ => Err(Error::inconsistent_list_designated(
1039+
scalar,
1040+
boxed_field.data_type(),
10581041
)),
1059-
}
1042+
})
1043+
.collect::<Result<Vec<_>, _>>()?;
1044+
protobuf::ScalarValue {
1045+
value: Some(protobuf::scalar_value::Value::ListValue(
1046+
protobuf::ScalarListValue {
1047+
field: Some(boxed_field.as_ref().into()),
1048+
values: type_checked_values,
1049+
},
1050+
)),
10601051
}
10611052
}
1062-
None => protobuf::ScalarValue {
1063-
value: Some(protobuf::scalar_value::Value::NullListValue(
1064-
boxed_field.as_ref().try_into()?,
1065-
)),
1066-
},
10671053
}
1068-
}
1054+
None => protobuf::ScalarValue {
1055+
value: Some(protobuf::scalar_value::Value::NullListValue(
1056+
boxed_field.as_ref().try_into()?,
1057+
)),
1058+
},
1059+
},
10691060
datafusion::scalar::ScalarValue::Date32(val) => {
10701061
create_proto_scalar(val, PrimitiveScalarType::Date32, |s| {
10711062
Value::Date32Value(*s)

0 commit comments

Comments
 (0)