use super::plan::agg as proto_agg;
use super::schema::data_type::Kind as DataTypeKind;
use super::schema::metadata_value::Value as MetadataValueKind;
use super::schema::primitive_type::Kind as PrimitiveTypeKind;
use super::schema::SimplePrimitiveType as Simple;
use super::{
expressions as proto_expr, operation as proto_op, plan as proto_plan, schema as proto_schema,
};
use crate::expressions::{
ArrayData, BinaryExpression, BinaryExpressionOp, BinaryPredicate, BinaryPredicateOp,
ColumnName, DecimalData, Expression, ExpressionFieldPatch, ExpressionStructPatch,
JunctionPredicate, JunctionPredicateOp, MapData, MapToStructExpression, OpaqueExpression,
OpaquePredicate, ParseJsonExpression, Predicate, Scalar, StructData, UnaryExpression,
UnaryExpressionOp, UnaryPredicate, UnaryPredicateOp, VariadicExpression, VariadicExpressionOp,
};
use crate::plans::ir::nodes::{
Agg, Aggregate, FileType, Filter, Load, LoadColumnFileMeta, Operator, Project, ScanFile,
ScanJson, ScanParquet, SemiJoin, Values,
};
use crate::plans::ir::plan::{Plan, PlanNode};
use crate::plans::{IoOperation, Operation};
use crate::schema::{
ArrayType, DataType, DecimalType, MapType, MetadataValue, PrimitiveType, StructField,
StructType,
};
use crate::{DeltaResult, Error, FileMeta, FileSlice};
fn convert_vec<'a, T, U>(items: &'a [T]) -> Vec<U>
where
U: From<&'a T>,
{
items.iter().map(U::from).collect()
}
fn convert_expr_vec<E>(items: &[E]) -> Vec<proto_expr::Expression>
where
E: AsRef<Expression>,
{
items.iter().map(|e| e.as_ref().into()).collect()
}
impl From<&Operation> for proto_op::Operation {
fn from(op: &Operation) -> Self {
let op = match op {
Operation::IoOperation(io) => proto_op::operation::Op::Io(io.into()),
Operation::QueryPlan(plan) => proto_op::operation::Op::QueryPlan(plan.into()),
};
proto_op::Operation { op: Some(op) }
}
}
impl From<&IoOperation> for proto_op::IoOperation {
fn from(io: &IoOperation) -> Self {
use proto_op::io_operation::Op;
let op = match io {
IoOperation::FileListing { url } => Op::FileListing(proto_op::FileListing {
url: url.to_string(),
}),
IoOperation::ReadBytes { files } => Op::ReadBytes(proto_op::ReadBytes {
files: convert_vec(files),
}),
IoOperation::WriteBytes {
url,
data,
overwrite,
} => Op::WriteBytes(proto_op::WriteBytes {
url: url.to_string(),
data: data.to_vec(),
overwrite: *overwrite,
}),
IoOperation::HeadFile { url } => Op::HeadFile(proto_op::HeadFile {
url: url.to_string(),
}),
IoOperation::AtomicCopy {
source,
destination,
} => Op::AtomicCopy(proto_op::AtomicCopy {
source: source.to_string(),
destination: destination.to_string(),
}),
IoOperation::ParquetFooter { file } => Op::ParquetFooter(proto_op::ParquetFooter {
file: Some(file.into()),
}),
};
proto_op::IoOperation { op: Some(op) }
}
}
impl From<&FileSlice> for proto_op::FileSlice {
fn from(slice: &FileSlice) -> Self {
let (url, range) = slice;
proto_op::FileSlice {
url: url.to_string(),
range_start: range.as_ref().map(|r| r.start),
range_end: range.as_ref().map(|r| r.end),
}
}
}
impl From<&FileMeta> for proto_plan::FileMeta {
fn from(meta: &FileMeta) -> Self {
proto_plan::FileMeta {
location: meta.location.to_string(),
size: meta.size,
last_modified: meta.last_modified,
}
}
}
impl From<&Plan> for proto_plan::Plan {
fn from(plan: &Plan) -> Self {
proto_plan::Plan {
nodes: convert_vec(&plan.nodes),
}
}
}
impl From<&PlanNode> for proto_plan::PlanNode {
fn from(node: &PlanNode) -> Self {
proto_plan::PlanNode {
op: Some((&node.op).into()),
inputs: node.inputs.iter().map(|&i| i as u32).collect(),
}
}
}
impl From<&Operator> for proto_plan::Operator {
fn from(op: &Operator) -> Self {
use proto_plan::operator::Op;
let op = match op {
Operator::ScanParquet(n) => Op::ScanParquet(n.into()),
Operator::ScanJson(n) => Op::ScanJson(n.into()),
Operator::Values(n) => Op::Values(n.into()),
Operator::Project(n) => Op::Project(n.into()),
Operator::Filter(n) => Op::Filter(n.into()),
Operator::Load(n) => Op::Load(n.into()),
Operator::Aggregate(n) => Op::Aggregate(n.into()),
Operator::SemiJoin(n) => Op::SemiJoin(n.into()),
Operator::UnionAll(_) => Op::UnionAll(proto_plan::UnionAllNode {}),
};
proto_plan::Operator { op: Some(op) }
}
}
impl From<&ScanFile> for proto_plan::ScanFile {
fn from(file: &ScanFile) -> Self {
proto_plan::ScanFile {
meta: Some((&file.meta).into()),
file_constants: convert_vec(&file.file_constants),
}
}
}
impl From<&ScanParquet> for proto_plan::ScanParquetNode {
fn from(node: &ScanParquet) -> Self {
proto_plan::ScanParquetNode {
files: convert_vec(&node.files),
file_constant_columns: node.file_constant_columns.clone(),
schema: Some(node.schema.as_ref().into()),
}
}
}
impl From<&ScanJson> for proto_plan::ScanJsonNode {
fn from(node: &ScanJson) -> Self {
proto_plan::ScanJsonNode {
files: convert_vec(&node.files),
file_constant_columns: node.file_constant_columns.clone(),
schema: Some(node.schema.as_ref().into()),
}
}
}
impl From<&Values> for proto_plan::ValuesNode {
fn from(node: &Values) -> Self {
let rows = node
.rows
.iter()
.map(|row| proto_plan::ValuesRow {
values: convert_vec(row),
})
.collect();
proto_plan::ValuesNode {
schema: Some(node.schema.as_ref().into()),
rows,
}
}
}
impl From<&Project> for proto_plan::ProjectNode {
fn from(node: &Project) -> Self {
proto_plan::ProjectNode {
expr: Some(node.expr.as_ref().into()),
schema: Some(node.schema.as_ref().into()),
}
}
}
impl From<&Filter> for proto_plan::FilterNode {
fn from(node: &Filter) -> Self {
proto_plan::FilterNode {
predicate: Some(node.predicate.as_ref().into()),
}
}
}
impl From<&Load> for proto_plan::LoadNode {
fn from(node: &Load) -> Self {
proto_plan::LoadNode {
schema: Some(node.schema.as_ref().into()),
file_type: proto_plan::FileType::from(node.file_type) as i32,
base_url: node.base_url.as_ref().map(ToString::to_string),
file_constant_columns: node.file_constant_columns.clone(),
file_meta: Some((&node.file_meta).into()),
dv_column: Some((&node.dv_column).into()),
}
}
}
impl From<&LoadColumnFileMeta> for proto_plan::LoadColumnFileMeta {
fn from(meta: &LoadColumnFileMeta) -> Self {
proto_plan::LoadColumnFileMeta {
path_column: Some((&meta.path_column).into()),
file_size_column: Some((&meta.file_size_column).into()),
num_records_column: Some((&meta.num_records_column).into()),
}
}
}
impl From<&Aggregate> for proto_plan::AggregateNode {
fn from(node: &Aggregate) -> Self {
proto_plan::AggregateNode {
group_by: convert_vec(&node.group_by),
aggs: convert_vec(&node.aggs),
schema: Some(node.schema.as_ref().into()),
}
}
}
impl From<&Agg> for proto_plan::Agg {
fn from(agg: &Agg) -> Self {
let func = match agg {
Agg::Min { value } => proto_agg::Func::Min(proto_plan::MinAgg {
value: Some(value.into()),
}),
Agg::Max { value } => proto_agg::Func::Max(proto_plan::MaxAgg {
value: Some(value.into()),
}),
Agg::MinNonNullBy { value, key } => {
proto_agg::Func::MinNonNullBy(proto_plan::MinNonNullByAgg {
value: Some(value.into()),
key: Some(key.into()),
})
}
Agg::MaxNonNullBy { value, key } => {
proto_agg::Func::MaxNonNullBy(proto_plan::MaxNonNullByAgg {
value: Some(value.into()),
key: Some(key.into()),
})
}
};
proto_plan::Agg { func: Some(func) }
}
}
impl From<&SemiJoin> for proto_plan::SemiJoinNode {
fn from(node: &SemiJoin) -> Self {
proto_plan::SemiJoinNode {
inverted: node.inverted,
probe_keys: convert_vec(&node.probe_keys),
build_keys: convert_vec(&node.build_keys),
}
}
}
impl From<FileType> for proto_plan::FileType {
fn from(file_type: FileType) -> Self {
match file_type {
FileType::Parquet => proto_plan::FileType::Parquet,
FileType::Json => proto_plan::FileType::Json,
}
}
}
impl From<&Expression> for proto_expr::Expression {
fn from(expr: &Expression) -> Self {
use proto_expr::expression::Kind;
let kind = match expr {
Expression::Literal(scalar) => Kind::Literal(scalar.into()),
Expression::Column(column) => Kind::Column(column.into()),
Expression::Predicate(pred) => Kind::Predicate(Box::new(pred.as_ref().into())),
Expression::Struct(exprs, nullability) => {
let nullability_predicate =
nullability.as_ref().map(|n| Box::new(n.as_ref().into()));
Kind::StructExpr(Box::new(proto_expr::StructExpression {
exprs: convert_expr_vec(exprs),
nullability_predicate,
}))
}
Expression::StructPatch(patch) => Kind::Transform(patch.into()),
Expression::Unary(unary) => Kind::Unary(Box::new(unary.into())),
Expression::Binary(binary) => Kind::Binary(Box::new(binary.into())),
Expression::Variadic(variadic) => Kind::Variadic(variadic.into()),
Expression::Opaque(opaque) => Kind::Opaque(opaque.into()),
Expression::Unknown(name) => Kind::Unknown(name.clone()),
Expression::ParseJson(parse_json) => Kind::ParseJson(Box::new(parse_json.into())),
Expression::MapToStruct(map_to_struct) => {
Kind::MapToStruct(Box::new(map_to_struct.into()))
}
};
proto_expr::Expression { kind: Some(kind) }
}
}
impl From<&Predicate> for proto_expr::Predicate {
fn from(pred: &Predicate) -> Self {
use proto_expr::predicate::Kind;
let kind = match pred {
Predicate::BooleanExpression(expr) => Kind::BooleanExpression(Box::new(expr.into())),
Predicate::Not(inner) => Kind::Not(Box::new(inner.as_ref().into())),
Predicate::Unary(unary) => Kind::Unary(Box::new(unary.into())),
Predicate::Binary(binary) => Kind::Binary(Box::new(binary.into())),
Predicate::Junction(junction) => Kind::Junction(junction.into()),
Predicate::Opaque(opaque) => Kind::Opaque(opaque.into()),
Predicate::Unknown(name) => Kind::Unknown(name.clone()),
};
proto_expr::Predicate { kind: Some(kind) }
}
}
impl From<&ColumnName> for proto_expr::ColumnName {
fn from(column: &ColumnName) -> Self {
proto_expr::ColumnName {
path: column.path().to_vec(),
}
}
}
impl From<&UnaryExpression> for proto_expr::UnaryExpression {
fn from(unary: &UnaryExpression) -> Self {
proto_expr::UnaryExpression {
op: proto_expr::UnaryExpressionOp::from(unary.op) as i32,
expr: Some(Box::new(unary.expr.as_ref().into())),
}
}
}
impl From<&BinaryExpression> for proto_expr::BinaryExpression {
fn from(binary: &BinaryExpression) -> Self {
proto_expr::BinaryExpression {
op: proto_expr::BinaryExpressionOp::from(binary.op) as i32,
left: Some(Box::new(binary.left.as_ref().into())),
right: Some(Box::new(binary.right.as_ref().into())),
}
}
}
impl From<&VariadicExpression> for proto_expr::VariadicExpression {
fn from(variadic: &VariadicExpression) -> Self {
proto_expr::VariadicExpression {
op: proto_expr::VariadicExpressionOp::from(variadic.op) as i32,
exprs: convert_vec(&variadic.exprs),
}
}
}
impl From<&OpaqueExpression> for proto_expr::OpaqueExpression {
fn from(opaque: &OpaqueExpression) -> Self {
proto_expr::OpaqueExpression {
name: opaque.op.name().to_string(),
exprs: convert_vec(&opaque.exprs),
}
}
}
impl From<&ParseJsonExpression> for proto_expr::ParseJsonExpression {
fn from(parse_json: &ParseJsonExpression) -> Self {
proto_expr::ParseJsonExpression {
json_expr: Some(Box::new(parse_json.json_expr.as_ref().into())),
output_schema: Some(parse_json.output_schema.as_ref().into()),
}
}
}
impl From<&MapToStructExpression> for proto_expr::MapToStructExpression {
fn from(map_to_struct: &MapToStructExpression) -> Self {
proto_expr::MapToStructExpression {
map_expr: Some(Box::new(map_to_struct.map_expr.as_ref().into())),
}
}
}
impl From<&UnaryPredicate> for proto_expr::UnaryPredicate {
fn from(unary: &UnaryPredicate) -> Self {
proto_expr::UnaryPredicate {
op: proto_expr::UnaryPredicateOp::from(unary.op) as i32,
expr: Some(Box::new(unary.expr.as_ref().into())),
}
}
}
impl From<&BinaryPredicate> for proto_expr::BinaryPredicate {
fn from(binary: &BinaryPredicate) -> Self {
proto_expr::BinaryPredicate {
op: proto_expr::BinaryPredicateOp::from(binary.op) as i32,
left: Some(Box::new(binary.left.as_ref().into())),
right: Some(Box::new(binary.right.as_ref().into())),
}
}
}
impl From<&JunctionPredicate> for proto_expr::JunctionPredicate {
fn from(junction: &JunctionPredicate) -> Self {
proto_expr::JunctionPredicate {
op: proto_expr::JunctionPredicateOp::from(junction.op) as i32,
preds: convert_vec(&junction.preds),
}
}
}
impl From<&OpaquePredicate> for proto_expr::OpaquePredicate {
fn from(opaque: &OpaquePredicate) -> Self {
proto_expr::OpaquePredicate {
name: opaque.op.name().to_string(),
exprs: convert_vec(&opaque.exprs),
}
}
}
impl From<&ExpressionStructPatch> for proto_expr::Transform {
fn from(patch: &ExpressionStructPatch) -> Self {
let field_transforms = patch
.field_patches
.iter()
.map(|(name, field_patch)| (name.clone(), field_patch.into()))
.collect();
proto_expr::Transform {
input_path: patch.input_path.as_ref().map(Into::into),
field_transforms,
prepended_fields: convert_expr_vec(&patch.prepended_fields),
appended_fields: convert_expr_vec(&patch.appended_fields),
}
}
}
impl From<&ExpressionFieldPatch> for proto_expr::FieldTransform {
fn from(field_patch: &ExpressionFieldPatch) -> Self {
proto_expr::FieldTransform {
exprs: convert_expr_vec(&field_patch.insertions),
is_replace: !field_patch.keep_input,
optional: field_patch.optional,
}
}
}
impl From<UnaryExpressionOp> for proto_expr::UnaryExpressionOp {
fn from(op: UnaryExpressionOp) -> Self {
match op {
UnaryExpressionOp::ToJson => proto_expr::UnaryExpressionOp::ToJson,
}
}
}
impl From<BinaryExpressionOp> for proto_expr::BinaryExpressionOp {
fn from(op: BinaryExpressionOp) -> Self {
match op {
BinaryExpressionOp::Plus => proto_expr::BinaryExpressionOp::Plus,
BinaryExpressionOp::Minus => proto_expr::BinaryExpressionOp::Minus,
BinaryExpressionOp::Multiply => proto_expr::BinaryExpressionOp::Multiply,
BinaryExpressionOp::Divide => proto_expr::BinaryExpressionOp::Divide,
}
}
}
impl From<VariadicExpressionOp> for proto_expr::VariadicExpressionOp {
fn from(op: VariadicExpressionOp) -> Self {
match op {
VariadicExpressionOp::Coalesce => proto_expr::VariadicExpressionOp::Coalesce,
VariadicExpressionOp::Array => proto_expr::VariadicExpressionOp::Array,
}
}
}
impl From<UnaryPredicateOp> for proto_expr::UnaryPredicateOp {
fn from(op: UnaryPredicateOp) -> Self {
match op {
UnaryPredicateOp::IsNull => proto_expr::UnaryPredicateOp::IsNull,
}
}
}
impl From<BinaryPredicateOp> for proto_expr::BinaryPredicateOp {
fn from(op: BinaryPredicateOp) -> Self {
match op {
BinaryPredicateOp::LessThan => proto_expr::BinaryPredicateOp::LessThan,
BinaryPredicateOp::GreaterThan => proto_expr::BinaryPredicateOp::GreaterThan,
BinaryPredicateOp::Equal => proto_expr::BinaryPredicateOp::Equal,
BinaryPredicateOp::Distinct => proto_expr::BinaryPredicateOp::Distinct,
BinaryPredicateOp::In => proto_expr::BinaryPredicateOp::In,
}
}
}
impl From<JunctionPredicateOp> for proto_expr::JunctionPredicateOp {
fn from(op: JunctionPredicateOp) -> Self {
match op {
JunctionPredicateOp::And => proto_expr::JunctionPredicateOp::And,
JunctionPredicateOp::Or => proto_expr::JunctionPredicateOp::Or,
}
}
}
impl From<&Scalar> for proto_expr::Scalar {
fn from(scalar: &Scalar) -> Self {
use proto_expr::scalar::Value;
let value = match scalar {
Scalar::Integer(v) => Value::Integer(*v),
Scalar::Long(v) => Value::Long(*v),
Scalar::Short(v) => Value::Short(*v as i32),
Scalar::Byte(v) => Value::Byte(*v as i32),
Scalar::Float(v) => Value::Float(*v),
Scalar::Double(v) => Value::Double(*v),
Scalar::String(v) => Value::String(v.clone()),
Scalar::Boolean(v) => Value::Boolean(*v),
Scalar::Timestamp(v) => Value::Timestamp(*v),
Scalar::TimestampNtz(v) => Value::TimestampNtz(*v),
#[cfg(feature = "nanosecond-timestamps")]
Scalar::TimestampNanos(v) => Value::TimestampNanos(*v),
#[cfg(feature = "nanosecond-timestamps")]
Scalar::TimestampNanosNtz(v) => Value::TimestampNanosNtz(*v),
Scalar::Date(v) => Value::Date(*v),
Scalar::Binary(v) => Value::Binary(v.clone()),
Scalar::Decimal(decimal) => Value::Decimal(decimal.into()),
Scalar::Null(data_type) => Value::Null(data_type.into()),
Scalar::Struct(struct_data) => Value::Struct(struct_data.into()),
Scalar::Array(array_data) => Value::Array(array_data.into()),
Scalar::Map(map_data) => Value::Map(map_data.into()),
};
proto_expr::Scalar { value: Some(value) }
}
}
impl From<&DecimalData> for proto_expr::DecimalData {
fn from(decimal: &DecimalData) -> Self {
proto_expr::DecimalData {
bits: decimal.bits().to_be_bytes().to_vec(),
decimal_type: Some((*decimal.ty()).into()),
}
}
}
impl From<&StructData> for proto_expr::StructData {
fn from(struct_data: &StructData) -> Self {
proto_expr::StructData {
fields: convert_vec(struct_data.fields()),
values: convert_vec(struct_data.values()),
}
}
}
impl From<&ArrayData> for proto_expr::ArrayData {
fn from(array_data: &ArrayData) -> Self {
proto_expr::ArrayData {
array_type: Some(array_data.array_type().into()),
elements: convert_vec(array_data.array_elements()),
}
}
}
impl From<&MapData> for proto_expr::MapData {
fn from(map_data: &MapData) -> Self {
let pairs = map_data
.pairs()
.iter()
.map(|(key, value)| proto_expr::MapEntry {
key: Some(key.into()),
value: Some(value.into()),
})
.collect();
proto_expr::MapData {
map_type: Some(map_data.map_type().into()),
pairs,
}
}
}
impl From<&DataType> for proto_schema::DataType {
fn from(data_type: &DataType) -> Self {
let kind = match data_type {
DataType::Primitive(primitive) => DataTypeKind::Primitive(primitive.into()),
DataType::Array(array) => DataTypeKind::Array(Box::new(array.as_ref().into())),
DataType::Struct(struct_type) => DataTypeKind::Struct(struct_type.as_ref().into()),
DataType::Map(map) => DataTypeKind::Map(Box::new(map.as_ref().into())),
DataType::Variant(_) => DataTypeKind::Variant(proto_schema::VariantType {}),
};
proto_schema::DataType { kind: Some(kind) }
}
}
impl From<&PrimitiveType> for proto_schema::PrimitiveType {
fn from(primitive: &PrimitiveType) -> Self {
let kind = match primitive {
PrimitiveType::String => Kind::Simple(Simple::String as i32),
PrimitiveType::Long => Kind::Simple(Simple::Long as i32),
PrimitiveType::Integer => Kind::Simple(Simple::Integer as i32),
PrimitiveType::Short => Kind::Simple(Simple::Short as i32),
PrimitiveType::Byte => Kind::Simple(Simple::Byte as i32),
PrimitiveType::Float => Kind::Simple(Simple::Float as i32),
PrimitiveType::Double => Kind::Simple(Simple::Double as i32),
PrimitiveType::Boolean => Kind::Simple(Simple::Boolean as i32),
PrimitiveType::Binary => Kind::Simple(Simple::Binary as i32),
PrimitiveType::Date => Kind::Simple(Simple::Date as i32),
PrimitiveType::Timestamp => Kind::Simple(Simple::Timestamp as i32),
PrimitiveType::TimestampNtz => Kind::Simple(Simple::TimestampNtz as i32),
#[cfg(feature = "nanosecond-timestamps")]
PrimitiveType::TimestampNanos => Kind::Simple(Simple::TimestampNanos as i32),
#[cfg(feature = "nanosecond-timestamps")]
PrimitiveType::TimestampNanosNtz => Kind::Simple(Simple::TimestampNanosNtz as i32),
PrimitiveType::Decimal(decimal) => Kind::Decimal((*decimal).into()),
PrimitiveType::Void => Kind::Simple(Simple::Void as i32),
PrimitiveType::IntervalYearMonth => Kind::Simple(Simple::IntervalYearMonth as i32),
PrimitiveType::IntervalDayTime => Kind::Simple(Simple::IntervalDayTime as i32),
};
proto_schema::PrimitiveType { kind: Some(kind) }
}
}
impl From<DecimalType> for proto_schema::DecimalType {
fn from(decimal: DecimalType) -> Self {
proto_schema::DecimalType {
precision: u32::from(decimal.precision()),
scale: u32::from(decimal.scale()),
}
}
}
impl From<&ArrayType> for proto_schema::ArrayType {
fn from(array: &ArrayType) -> Self {
proto_schema::ArrayType {
element_type: Some(Box::new(array.element_type().into())),
contains_null: array.contains_null(),
}
}
}
impl From<&MapType> for proto_schema::MapType {
fn from(map: &MapType) -> Self {
proto_schema::MapType {
key_type: Some(Box::new(map.key_type().into())),
value_type: Some(Box::new(map.value_type().into())),
value_contains_null: map.value_contains_null(),
}
}
}
impl From<&StructType> for proto_schema::StructType {
fn from(struct_type: &StructType) -> Self {
proto_schema::StructType {
fields: struct_type.fields().map(Into::into).collect(),
}
}
}
impl From<&StructField> for proto_schema::StructField {
fn from(field: &StructField) -> Self {
let metadata = field
.metadata
.iter()
.map(|(key, value)| (key.clone(), value.into()))
.collect();
proto_schema::StructField {
name: field.name.clone(),
data_type: Some((&field.data_type).into()),
nullable: field.nullable,
metadata,
}
}
}
impl From<&MetadataValue> for proto_schema::MetadataValue {
fn from(metadata: &MetadataValue) -> Self {
let value = match metadata {
MetadataValue::Number(n) => MetadataValueKind::Number(*n),
MetadataValue::String(s) => MetadataValueKind::String(s.clone()),
MetadataValue::Boolean(b) => MetadataValueKind::Boolean(*b),
MetadataValue::Other(json) => MetadataValueKind::OtherJson(json.to_string()),
};
proto_schema::MetadataValue { value: Some(value) }
}
}
impl TryFrom<proto_schema::StructType> for StructType {
type Error = Error;
fn try_from(proto: proto_schema::StructType) -> DeltaResult<Self> {
let fields = proto
.fields
.into_iter()
.map(StructField::try_from)
.collect::<DeltaResult<Vec<_>>>()?;
StructType::try_new(fields)
}
}
impl TryFrom<proto_schema::StructField> for StructField {
type Error = Error;
fn try_from(proto: proto_schema::StructField) -> DeltaResult<Self> {
let data_type = proto
.data_type
.ok_or_else(|| Error::schema("StructField proto missing data_type"))?;
let metadata = proto
.metadata
.into_iter()
.map(|(key, value)| Ok::<_, Error>((key, MetadataValue::try_from(value)?)))
.collect::<DeltaResult<std::collections::HashMap<_, _>>>()?;
Ok(StructField {
name: proto.name,
data_type: DataType::try_from(data_type)?,
nullable: proto.nullable,
metadata,
})
}
}
impl TryFrom<proto_schema::DataType> for DataType {
type Error = Error;
fn try_from(proto: proto_schema::DataType) -> DeltaResult<Self> {
let kind = proto
.kind
.ok_or_else(|| Error::schema("DataType proto missing kind"))?;
let data_type = match kind {
DataTypeKind::Primitive(primitive) => DataType::Primitive(primitive.try_into()?),
DataTypeKind::Array(array) => DataType::Array(Box::new((*array).try_into()?)),
DataTypeKind::Struct(struct_type) => {
DataType::Struct(Box::new(struct_type.try_into()?))
}
DataTypeKind::Map(map) => DataType::Map(Box::new((*map).try_into()?)),
DataTypeKind::Variant(_) => DataType::unshredded_variant(),
};
Ok(data_type)
}
}
impl TryFrom<proto_schema::PrimitiveType> for PrimitiveType {
type Error = Error;
fn try_from(proto: proto_schema::PrimitiveType) -> DeltaResult<Self> {
let kind = proto
.kind
.ok_or_else(|| Error::schema("PrimitiveType proto missing kind"))?;
let primitive = match kind {
PrimitiveTypeKind::Simple(simple) => {
let simple = Simple::try_from(simple).map_err(|_| {
Error::schema(format!("unknown SimplePrimitiveType value: {simple}"))
})?;
match simple {
Simple::String => PrimitiveType::String,
Simple::Long => PrimitiveType::Long,
Simple::Integer => PrimitiveType::Integer,
Simple::Short => PrimitiveType::Short,
Simple::Byte => PrimitiveType::Byte,
Simple::Float => PrimitiveType::Float,
Simple::Double => PrimitiveType::Double,
Simple::Boolean => PrimitiveType::Boolean,
Simple::Binary => PrimitiveType::Binary,
Simple::Date => PrimitiveType::Date,
Simple::Timestamp => PrimitiveType::Timestamp,
Simple::TimestampNtz => PrimitiveType::TimestampNtz,
Simple::Void => PrimitiveType::Void,
Simple::IntervalYearMonth => PrimitiveType::IntervalYearMonth,
Simple::IntervalDayTime => PrimitiveType::IntervalDayTime,
Simple::Unspecified => {
return Err(Error::schema("SimplePrimitiveType is unspecified"))
}
}
}
PrimitiveTypeKind::Decimal(decimal) => PrimitiveType::Decimal(decimal.try_into()?),
};
Ok(primitive)
}
}
impl TryFrom<proto_schema::DecimalType> for DecimalType {
type Error = Error;
fn try_from(proto: proto_schema::DecimalType) -> DeltaResult<Self> {
let precision = u8::try_from(proto.precision).map_err(|_| {
Error::invalid_decimal(format!("precision out of range: {}", proto.precision))
})?;
let scale = u8::try_from(proto.scale)
.map_err(|_| Error::invalid_decimal(format!("scale out of range: {}", proto.scale)))?;
DecimalType::try_new(precision, scale)
}
}
impl TryFrom<proto_schema::ArrayType> for ArrayType {
type Error = Error;
fn try_from(proto: proto_schema::ArrayType) -> DeltaResult<Self> {
let element_type = proto
.element_type
.ok_or_else(|| Error::schema("ArrayType proto missing element_type"))?;
Ok(ArrayType::new(
DataType::try_from(*element_type)?,
proto.contains_null,
))
}
}
impl TryFrom<proto_schema::MapType> for MapType {
type Error = Error;
fn try_from(proto: proto_schema::MapType) -> DeltaResult<Self> {
let key_type = proto
.key_type
.ok_or_else(|| Error::schema("MapType proto missing key_type"))?;
let value_type = proto
.value_type
.ok_or_else(|| Error::schema("MapType proto missing value_type"))?;
Ok(MapType::new(
DataType::try_from(*key_type)?,
DataType::try_from(*value_type)?,
proto.value_contains_null,
))
}
}
impl TryFrom<proto_schema::MetadataValue> for MetadataValue {
type Error = Error;
fn try_from(proto: proto_schema::MetadataValue) -> DeltaResult<Self> {
let value = proto
.value
.ok_or_else(|| Error::schema("MetadataValue proto missing value"))?;
let metadata = match value {
MetadataValueKind::Number(n) => MetadataValue::Number(n),
MetadataValueKind::String(s) => MetadataValue::String(s),
MetadataValueKind::Boolean(b) => MetadataValue::Boolean(b),
MetadataValueKind::OtherJson(json) => {
MetadataValue::Other(serde_json::from_str(&json)?)
}
};
Ok(metadata)
}
}
#[cfg(test)]
mod tests {
use std::sync::Arc;
use bytes::Bytes;
use rstest::rstest;
use url::Url;
use crate::expressions::{
lit, ArrayData, BinaryExpressionOp, BinaryPredicateOp, ColumnName, DecimalData, Expression,
ExpressionStructPatchBuilder, JunctionPredicateOp, MapData, OpaqueExpressionOp,
OpaquePredicateOp, Predicate, Scalar, ScalarExpressionEvaluator, StructData,
UnaryExpressionOp, UnaryPredicateOp, VariadicExpressionOp,
};
use crate::kernel_predicates::{
DirectDataSkippingPredicateEvaluator, DirectPredicateEvaluator,
IndirectDataSkippingPredicateEvaluator,
};
use crate::plans::ir::nodes::{
Agg, Aggregate, FileType, Filter, Load, LoadColumnFileMeta, Operator, Project, ScanFile,
ScanJson, ScanParquet, SemiJoin, UnionAll, Values,
};
use crate::plans::ir::plan::{Plan, PlanNode};
use crate::plans::proto::{
expressions as proto_expr, operation as proto_op, plan as proto_plan,
schema as proto_schema,
};
use crate::plans::{IoOperation, Operation};
use crate::schema::{
ArrayType, DataType, DecimalType, MapType, MetadataValue, PrimitiveType, SchemaRef,
StructField, StructType,
};
use crate::{DeltaResult, FileMeta, FileSlice};
#[derive(Debug, PartialEq)]
struct TestOpaqueExprOp;
impl OpaqueExpressionOp for TestOpaqueExprOp {
fn name(&self) -> &str {
"test_opaque_expr"
}
fn eval_expr_scalar(
&self,
_eval_expr: &ScalarExpressionEvaluator<'_>,
_exprs: &[Expression],
) -> DeltaResult<Scalar> {
Ok(Scalar::Integer(0))
}
}
#[derive(Debug, PartialEq)]
struct TestOpaquePredOp;
impl OpaquePredicateOp for TestOpaquePredOp {
fn name(&self) -> &str {
"test_opaque_pred"
}
fn eval_pred_scalar(
&self,
_eval_expr: &ScalarExpressionEvaluator<'_>,
_eval_pred: &DirectPredicateEvaluator<'_>,
_exprs: &[Expression],
_inverted: bool,
) -> DeltaResult<Option<bool>> {
Ok(Some(true))
}
fn eval_as_data_skipping_predicate(
&self,
_evaluator: &DirectDataSkippingPredicateEvaluator<'_>,
_exprs: &[Expression],
_inverted: bool,
) -> Option<bool> {
None
}
fn as_data_skipping_predicate(
&self,
_evaluator: &IndirectDataSkippingPredicateEvaluator<'_>,
_exprs: &[Expression],
_inverted: bool,
) -> Option<Predicate> {
None
}
}
fn sample_file_meta() -> FileMeta {
FileMeta {
location: Url::parse("memory:///table/part-0.parquet").unwrap(),
last_modified: 123,
size: 456,
}
}
fn sample_schema() -> SchemaRef {
Arc::new(StructType::try_new(vec![StructField::nullable("id", DataType::INTEGER)]).unwrap())
}
fn decode(op: &Operation) -> proto_op::Operation {
let bytes = op.to_proto_bytes();
prost::Message::decode(bytes.as_slice()).expect("decode succeeds")
}
fn io_op(op: &Operation) -> proto_op::io_operation::Op {
let Some(proto_op::operation::Op::Io(io)) = decode(op).op else {
panic!("expected an IoOperation");
};
io.op.expect("io op present")
}
fn expr_kind_of(expr: Expression) -> proto_expr::expression::Kind {
proto_expr::Expression::from(&expr)
.kind
.expect("expression kind present")
}
fn pred_kind_of(pred: Predicate) -> proto_expr::predicate::Kind {
proto_expr::Predicate::from(&pred)
.kind
.expect("predicate kind present")
}
fn scalar_value_of(scalar: Scalar) -> proto_expr::scalar::Value {
proto_expr::Scalar::from(&scalar)
.value
.expect("scalar value present")
}
#[rstest]
#[case(
Operation::IoOperation(IoOperation::head_file(Url::parse("memory:///h").unwrap())),
"io"
)]
#[case(Operation::QueryPlan(Plan { nodes: vec![] }), "query_plan")]
fn from_operation(#[case] op: Operation, #[case] expected: &str) {
use proto_op::operation::Op;
let kind = match decode(&op).op.unwrap() {
Op::Io(_) => "io",
Op::QueryPlan(_) => "query_plan",
};
assert_eq!(kind, expected);
}
#[test]
fn from_io_operation_file_listing() {
let url = "memory:///table/";
let op = Operation::IoOperation(IoOperation::file_listing(Url::parse(url).unwrap()));
let proto_op::io_operation::Op::FileListing(file_listing) = io_op(&op) else {
panic!("expected FileListing");
};
assert_eq!(file_listing.url, url);
}
#[test]
fn from_io_operation_read_bytes() {
let files = vec![
(Url::parse("memory:///a").unwrap(), Some(0u64..10u64)),
(Url::parse("memory:///b").unwrap(), None),
];
let op = Operation::IoOperation(IoOperation::read_bytes(files));
let proto_op::io_operation::Op::ReadBytes(read_bytes) = io_op(&op) else {
panic!("expected ReadBytes");
};
assert_eq!(read_bytes.files.len(), 2);
assert_eq!(read_bytes.files[0].url, "memory:///a");
}
#[test]
fn from_io_operation_write_bytes() {
let op = Operation::IoOperation(IoOperation::write_bytes(
Url::parse("memory:///out").unwrap(),
Bytes::from_static(b"hello"),
true,
));
let proto_op::io_operation::Op::WriteBytes(write_bytes) = io_op(&op) else {
panic!("expected WriteBytes");
};
assert_eq!(write_bytes.url, "memory:///out");
assert_eq!(write_bytes.data, b"hello");
assert!(write_bytes.overwrite);
}
#[test]
fn from_io_operation_head_file() {
let op = Operation::IoOperation(IoOperation::head_file(
Url::parse("memory:///head").unwrap(),
));
let proto_op::io_operation::Op::HeadFile(head_file) = io_op(&op) else {
panic!("expected HeadFile");
};
assert_eq!(head_file.url, "memory:///head");
}
#[test]
fn from_io_operation_atomic_copy() {
let op = Operation::IoOperation(IoOperation::atomic_copy(
Url::parse("memory:///src").unwrap(),
Url::parse("memory:///dst").unwrap(),
));
let proto_op::io_operation::Op::AtomicCopy(atomic_copy) = io_op(&op) else {
panic!("expected AtomicCopy");
};
assert_eq!(atomic_copy.source, "memory:///src");
assert_eq!(atomic_copy.destination, "memory:///dst");
}
#[test]
fn from_io_operation_parquet_footer() {
let op = Operation::IoOperation(IoOperation::parquet_footer(sample_file_meta()));
let proto_op::io_operation::Op::ParquetFooter(parquet_footer) = io_op(&op) else {
panic!("expected ParquetFooter");
};
let file = parquet_footer.file.expect("file meta present");
assert_eq!(file.location, "memory:///table/part-0.parquet");
assert_eq!(file.size, 456);
assert_eq!(file.last_modified, 123);
}
#[rstest]
#[case(Some(0u64..10u64), Some(0), Some(10))]
#[case(None, None, None)]
fn from_file_slice(
#[case] range: Option<std::ops::Range<u64>>,
#[case] expected_start: Option<u64>,
#[case] expected_end: Option<u64>,
) {
let slice: FileSlice = (Url::parse("memory:///a").unwrap(), range);
let proto = proto_op::FileSlice::from(&slice);
assert_eq!(proto.url, "memory:///a");
assert_eq!(proto.range_start, expected_start);
assert_eq!(proto.range_end, expected_end);
}
#[test]
fn from_file_meta() {
let proto = proto_plan::FileMeta::from(&sample_file_meta());
assert_eq!(proto.location, "memory:///table/part-0.parquet");
assert_eq!(proto.size, 456);
assert_eq!(proto.last_modified, 123);
}
#[test]
fn from_plan() {
let schema = Arc::new(
StructType::try_new(vec![
StructField::nullable("id", DataType::INTEGER),
StructField::not_null("name", DataType::STRING),
])
.unwrap(),
);
let plan = Plan {
nodes: vec![
PlanNode {
op: Operator::ScanParquet(ScanParquet {
files: vec![ScanFile::new(sample_file_meta())],
file_constant_columns: vec![],
schema: schema.clone(),
}),
inputs: vec![],
},
PlanNode {
op: Operator::Filter(Filter {
predicate: Arc::new(Predicate::gt(
Expression::Column(ColumnName::new(["id"])),
lit(5i32),
)),
}),
inputs: vec![0],
},
],
};
let Some(proto_op::operation::Op::QueryPlan(plan)) = decode(&Operation::QueryPlan(plan)).op
else {
panic!("expected a QueryPlan");
};
assert_eq!(plan.nodes.len(), 2);
let source = &plan.nodes[0];
assert!(source.inputs.is_empty());
let Some(proto_plan::operator::Op::ScanParquet(scan)) = &source.op.as_ref().unwrap().op
else {
panic!("expected ScanParquet");
};
assert_eq!(scan.files.len(), 1);
assert_eq!(scan.schema.as_ref().unwrap().fields.len(), 2);
let filter_node = &plan.nodes[1];
assert_eq!(filter_node.inputs.len(), 1);
assert_eq!(filter_node.inputs[0], 0);
let Some(proto_plan::operator::Op::Filter(filter)) = &filter_node.op.as_ref().unwrap().op
else {
panic!("expected Filter");
};
let predicate = filter.predicate.as_ref().unwrap();
let Some(proto_expr::predicate::Kind::Binary(binary)) = &predicate.kind else {
panic!("expected a binary predicate");
};
assert_eq!(binary.op, proto_expr::BinaryPredicateOp::GreaterThan as i32);
assert!(matches!(
binary.left.as_ref().unwrap().kind,
Some(proto_expr::expression::Kind::Column(_))
));
}
#[rstest]
#[case(
Operator::ScanParquet(ScanParquet {
files: vec![],
file_constant_columns: vec![],
schema: sample_schema(),
}),
"scan_parquet"
)]
#[case(
Operator::ScanJson(ScanJson {
files: vec![],
file_constant_columns: vec![],
schema: sample_schema(),
}),
"scan_json"
)]
#[case(Operator::Values(Values { schema: sample_schema(), rows: vec![] }), "values")]
#[case(
Operator::Project(Project {
expr: Arc::new(Expression::struct_from([lit(1)])),
schema: sample_schema(),
}),
"project"
)]
#[case(Operator::Filter(Filter { predicate: Arc::new(Predicate::literal(true)) }), "filter")]
#[case(
Operator::Load(Load {
schema: sample_schema(),
file_type: FileType::Parquet,
base_url: None,
file_constant_columns: vec![],
file_meta: sample_load_column_file_meta(),
dv_column: ColumnName::new(["dv"]),
}),
"load"
)]
#[case(
Operator::Aggregate(Aggregate {
group_by: vec![],
aggs: vec![],
schema: sample_schema(),
}),
"aggregate"
)]
#[case(
Operator::SemiJoin(SemiJoin { inverted: false, probe_keys: vec![], build_keys: vec![] }),
"semi_join"
)]
#[case(Operator::UnionAll(UnionAll), "union_all")]
fn from_operator(#[case] op: Operator, #[case] expected: &str) {
use proto_plan::operator::Op;
let kind = match proto_plan::Operator::from(&op).op.unwrap() {
Op::ScanParquet(_) => "scan_parquet",
Op::ScanJson(_) => "scan_json",
Op::Values(_) => "values",
Op::Project(_) => "project",
Op::Filter(_) => "filter",
Op::Load(_) => "load",
Op::Aggregate(_) => "aggregate",
Op::SemiJoin(_) => "semi_join",
Op::UnionAll(_) => "union_all",
};
assert_eq!(kind, expected);
}
#[test]
fn from_scan_file() {
let scan_file = ScanFile {
meta: sample_file_meta(),
file_constants: vec![Scalar::Integer(1), Scalar::String("x".into())],
};
let proto = proto_plan::ScanFile::from(&scan_file);
assert!(proto.meta.is_some());
assert_eq!(proto.file_constants.len(), 2);
}
#[test]
fn from_scan_parquet() {
let node = ScanParquet {
files: vec![ScanFile::new(sample_file_meta())],
file_constant_columns: vec!["c".to_string()],
schema: sample_schema(),
};
let proto = proto_plan::ScanParquetNode::from(&node);
assert_eq!(proto.files.len(), 1);
assert_eq!(proto.file_constant_columns.len(), 1);
assert!(proto.schema.is_some());
}
#[test]
fn from_scan_json() {
let node = ScanJson {
files: vec![ScanFile::new(sample_file_meta())],
file_constant_columns: vec!["c".to_string()],
schema: sample_schema(),
};
let proto = proto_plan::ScanJsonNode::from(&node);
assert_eq!(proto.files.len(), 1);
assert_eq!(proto.file_constant_columns.len(), 1);
assert!(proto.schema.is_some());
}
#[test]
fn from_values() {
let node = Values {
schema: sample_schema(),
rows: vec![vec![Scalar::Integer(1)], vec![Scalar::Integer(2)]],
};
let proto = proto_plan::ValuesNode::from(&node);
assert!(proto.schema.is_some());
assert_eq!(proto.rows.len(), 2);
assert_eq!(proto.rows[0].values.len(), 1);
}
#[test]
fn from_project() {
let node = Project {
expr: Arc::new(Expression::struct_from([lit(1)])),
schema: sample_schema(),
};
let proto = proto_plan::ProjectNode::from(&node);
assert!(proto.expr.is_some());
assert!(proto.schema.is_some());
}
#[test]
fn from_filter() {
let node = Filter {
predicate: Arc::new(Predicate::literal(true)),
};
let proto = proto_plan::FilterNode::from(&node);
assert!(proto.predicate.is_some());
}
fn sample_load_column_file_meta() -> LoadColumnFileMeta {
LoadColumnFileMeta {
path_column: ColumnName::new(["path"]),
file_size_column: ColumnName::new(["size"]),
num_records_column: ColumnName::new(["num_records"]),
}
}
#[rstest]
#[case(
FileType::Json,
Some(Url::parse("memory:///base/").unwrap()),
Some("memory:///base/")
)]
#[case(FileType::Parquet, None, None)]
fn from_load(
#[case] file_type: FileType,
#[case] base_url: Option<Url>,
#[case] expected_base_url: Option<&str>,
) {
let node = Load {
schema: sample_schema(),
file_type,
base_url,
file_constant_columns: vec!["c".to_string()],
file_meta: sample_load_column_file_meta(),
dv_column: ColumnName::new(["dv"]),
};
let proto = proto_plan::LoadNode::from(&node);
assert!(proto.schema.is_some());
assert_eq!(
proto.file_type,
proto_plan::FileType::from(file_type) as i32
);
assert_eq!(proto.base_url.as_deref(), expected_base_url);
assert_eq!(proto.file_constant_columns.len(), 1);
assert!(proto.dv_column.is_some());
let file_meta = proto.file_meta.expect("file_meta present");
assert!(file_meta.path_column.is_some());
assert!(file_meta.file_size_column.is_some());
assert!(file_meta.num_records_column.is_some());
}
#[test]
fn from_aggregate() {
let node = Aggregate {
group_by: vec![ColumnName::new(["g"])],
aggs: vec![Agg::max(ColumnName::new(["a"]))],
schema: sample_schema(),
};
let proto = proto_plan::AggregateNode::from(&node);
assert_eq!(proto.group_by.len(), 1);
assert_eq!(proto.aggs.len(), 1);
assert!(proto.aggs[0].func.is_some());
assert!(proto.schema.is_some());
}
#[rstest]
#[case(Agg::min(ColumnName::new(["a"])), "min")]
#[case(Agg::max(ColumnName::new(["a"])), "max")]
#[case(Agg::min_non_null_by(ColumnName::new(["a"]), ColumnName::new(["k"])), "min_non_null_by")]
#[case(Agg::max_non_null_by(ColumnName::new(["a"]), ColumnName::new(["k"])), "max_non_null_by")]
fn from_agg(#[case] agg: Agg, #[case] expected: &str) {
use proto_plan::agg::Func;
let proto = proto_plan::Agg::from(&agg);
let kind = match proto.func.unwrap() {
Func::Min(_) => "min",
Func::Max(_) => "max",
Func::MinNonNullBy(_) => "min_non_null_by",
Func::MaxNonNullBy(_) => "max_non_null_by",
};
assert_eq!(kind, expected);
}
#[rstest]
#[case(true)]
#[case(false)]
fn from_semi_join(#[case] inverted: bool) {
let node = SemiJoin {
inverted,
probe_keys: vec![ColumnName::new(["p"])],
build_keys: vec![ColumnName::new(["b"])],
};
let proto = proto_plan::SemiJoinNode::from(&node);
assert_eq!(proto.inverted, inverted);
assert_eq!(proto.probe_keys.len(), 1);
assert_eq!(proto.build_keys.len(), 1);
}
#[rstest]
#[case(FileType::Parquet, proto_plan::FileType::Parquet)]
#[case(FileType::Json, proto_plan::FileType::Json)]
fn from_file_type(#[case] value: FileType, #[case] expected: proto_plan::FileType) {
assert_eq!(proto_plan::FileType::from(value) as i32, expected as i32);
}
#[rstest]
#[case(lit(1), "literal")]
#[case(Expression::Column(ColumnName::new(["a"])), "column")]
#[case(Expression::Predicate(Box::new(Predicate::literal(true))), "predicate")]
#[case(Expression::struct_from([lit(1)]), "struct_expr")]
#[case(
Expression::StructPatch(
ExpressionStructPatchBuilder::new().append(lit(1)).build().unwrap(),
),
"transform"
)]
#[case(Expression::unary(UnaryExpressionOp::ToJson, lit(1)), "unary")]
#[case(Expression::binary(BinaryExpressionOp::Plus, lit(1), lit(2)), "binary")]
#[case(Expression::coalesce([lit(1), lit(2)]), "variadic")]
#[case(Expression::opaque(TestOpaqueExprOp, [lit(1)]), "opaque")]
#[case(Expression::parse_json(lit("{}"), sample_schema()), "parse_json")]
#[case(Expression::map_to_struct(Expression::Column(ColumnName::new(["m"]))), "map_to_struct")]
#[case(Expression::unknown("x"), "unknown")]
fn from_expression(#[case] expr: Expression, #[case] expected: &str) {
use proto_expr::expression::Kind;
let kind = match proto_expr::Expression::from(&expr).kind.unwrap() {
Kind::Literal(_) => "literal",
Kind::Column(_) => "column",
Kind::Predicate(_) => "predicate",
Kind::StructExpr(_) => "struct_expr",
Kind::Transform(_) => "transform",
Kind::Unary(_) => "unary",
Kind::Binary(_) => "binary",
Kind::Variadic(_) => "variadic",
Kind::IfExpr(_) => "if_expr",
Kind::Opaque(_) => "opaque",
Kind::ParseJson(_) => "parse_json",
Kind::MapToStruct(_) => "map_to_struct",
Kind::Unknown(_) => "unknown",
};
assert_eq!(kind, expected);
}
#[rstest]
#[case(Predicate::literal(true), "boolean_expression")]
#[case(Predicate::not(Predicate::literal(true)), "not")]
#[case(Predicate::is_null(lit(1)), "unary")]
#[case(Predicate::gt(lit(1), lit(2)), "binary")]
#[case(
Predicate::and(Predicate::literal(true), Predicate::literal(false)),
"junction"
)]
#[case(Predicate::opaque(TestOpaquePredOp, [lit(1)]), "opaque")]
#[case(Predicate::Unknown("x".to_string()), "unknown")]
fn from_predicate(#[case] pred: Predicate, #[case] expected: &str) {
use proto_expr::predicate::Kind;
let kind = match proto_expr::Predicate::from(&pred).kind.unwrap() {
Kind::BooleanExpression(_) => "boolean_expression",
Kind::Not(_) => "not",
Kind::Unary(_) => "unary",
Kind::Binary(_) => "binary",
Kind::Junction(_) => "junction",
Kind::Opaque(_) => "opaque",
Kind::Unknown(_) => "unknown",
};
assert_eq!(kind, expected);
}
#[test]
fn from_column_name() {
let proto = proto_expr::ColumnName::from(&ColumnName::new(["a", "b", "c"]));
assert_eq!(proto.path, vec!["a", "b", "c"]);
}
#[test]
fn from_struct_expression() {
let proto_expr::expression::Kind::StructExpr(plain) =
expr_kind_of(Expression::struct_from([lit(1), lit(2)]))
else {
panic!("expected a struct expression");
};
assert_eq!(plain.exprs.len(), 2);
assert!(plain.nullability_predicate.is_none());
let proto_expr::expression::Kind::StructExpr(guarded) = expr_kind_of(
Expression::struct_with_nullability_from([lit(1)], lit(true)),
) else {
panic!("expected a struct expression");
};
assert_eq!(guarded.exprs.len(), 1);
assert!(guarded.nullability_predicate.is_some());
}
#[test]
fn from_unary_expression() {
let proto_expr::expression::Kind::Unary(unary) =
expr_kind_of(Expression::unary(UnaryExpressionOp::ToJson, lit(1)))
else {
panic!("expected a unary expression");
};
assert_eq!(unary.op, proto_expr::UnaryExpressionOp::ToJson as i32);
assert!(unary.expr.is_some());
}
#[test]
fn from_binary_expression() {
let proto_expr::expression::Kind::Binary(binary) =
expr_kind_of(Expression::binary(BinaryExpressionOp::Plus, lit(1), lit(2)))
else {
panic!("expected a binary expression");
};
assert_eq!(binary.op, proto_expr::BinaryExpressionOp::Plus as i32);
assert!(binary.left.is_some());
assert!(binary.right.is_some());
}
#[test]
fn from_variadic_expression() {
let proto_expr::expression::Kind::Variadic(variadic) =
expr_kind_of(Expression::coalesce([lit(1), lit(2), lit(3)]))
else {
panic!("expected a variadic expression");
};
assert_eq!(
variadic.op,
proto_expr::VariadicExpressionOp::Coalesce as i32
);
assert_eq!(variadic.exprs.len(), 3);
}
#[test]
fn from_opaque_expression() {
let proto_expr::expression::Kind::Opaque(opaque) =
expr_kind_of(Expression::opaque(TestOpaqueExprOp, [lit(1), lit(2)]))
else {
panic!("expected an opaque expression");
};
assert_eq!(opaque.name, "test_opaque_expr");
assert_eq!(opaque.exprs.len(), 2);
}
#[test]
fn from_parse_json_expression() {
let proto_expr::expression::Kind::ParseJson(parse_json) =
expr_kind_of(Expression::parse_json(lit("{}"), sample_schema()))
else {
panic!("expected a parse_json expression");
};
assert!(parse_json.json_expr.is_some());
assert!(parse_json.output_schema.is_some());
}
#[test]
fn from_map_to_struct_expression() {
let proto_expr::expression::Kind::MapToStruct(map_to_struct) = expr_kind_of(
Expression::map_to_struct(Expression::Column(ColumnName::new(["m"]))),
) else {
panic!("expected a map_to_struct expression");
};
assert!(map_to_struct.map_expr.is_some());
}
#[test]
fn from_unary_predicate() {
let proto_expr::predicate::Kind::Unary(unary) = pred_kind_of(Predicate::is_null(lit(1)))
else {
panic!("expected a unary predicate");
};
assert_eq!(unary.op, proto_expr::UnaryPredicateOp::IsNull as i32);
assert!(unary.expr.is_some());
}
#[test]
fn from_binary_predicate() {
let proto_expr::predicate::Kind::Binary(binary) =
pred_kind_of(Predicate::gt(lit(1), lit(2)))
else {
panic!("expected a binary predicate");
};
assert_eq!(binary.op, proto_expr::BinaryPredicateOp::GreaterThan as i32);
assert!(binary.left.is_some());
assert!(binary.right.is_some());
}
#[test]
fn from_junction_predicate() {
let proto_expr::predicate::Kind::Junction(junction) = pred_kind_of(Predicate::and(
Predicate::literal(true),
Predicate::literal(false),
)) else {
panic!("expected a junction predicate");
};
assert_eq!(junction.op, proto_expr::JunctionPredicateOp::And as i32);
assert_eq!(junction.preds.len(), 2);
}
#[test]
fn from_opaque_predicate() {
let proto_expr::predicate::Kind::Opaque(opaque) =
pred_kind_of(Predicate::opaque(TestOpaquePredOp, [lit(1), lit(2)]))
else {
panic!("expected an opaque predicate");
};
assert_eq!(opaque.name, "test_opaque_pred");
assert_eq!(opaque.exprs.len(), 2);
}
#[test]
fn from_struct_patch() {
let patch = ExpressionStructPatchBuilder::new()
.replace("a", lit(1))
.insert_after("b", lit(2))
.drop_if_exists("c")
.prepend(lit(0))
.append(lit(3))
.build()
.unwrap();
let transform = proto_expr::Transform::from(&patch);
assert!(transform.field_transforms["a"].is_replace);
assert!(!transform.field_transforms["a"].optional);
assert!(!transform.field_transforms["b"].is_replace);
assert!(transform.field_transforms["c"].is_replace);
assert!(transform.field_transforms["c"].optional);
assert!(transform.input_path.is_none());
assert_eq!(transform.prepended_fields.len(), 1);
assert_eq!(transform.appended_fields.len(), 1);
}
#[rstest]
#[case(UnaryExpressionOp::ToJson, proto_expr::UnaryExpressionOp::ToJson)]
fn from_unary_expression_op(
#[case] op: UnaryExpressionOp,
#[case] expected: proto_expr::UnaryExpressionOp,
) {
assert_eq!(
proto_expr::UnaryExpressionOp::from(op) as i32,
expected as i32
);
}
#[rstest]
#[case(BinaryExpressionOp::Plus, proto_expr::BinaryExpressionOp::Plus)]
#[case(BinaryExpressionOp::Minus, proto_expr::BinaryExpressionOp::Minus)]
#[case(BinaryExpressionOp::Multiply, proto_expr::BinaryExpressionOp::Multiply)]
#[case(BinaryExpressionOp::Divide, proto_expr::BinaryExpressionOp::Divide)]
fn from_binary_expression_op(
#[case] op: BinaryExpressionOp,
#[case] expected: proto_expr::BinaryExpressionOp,
) {
assert_eq!(
proto_expr::BinaryExpressionOp::from(op) as i32,
expected as i32
);
}
#[rstest]
#[case(
VariadicExpressionOp::Coalesce,
proto_expr::VariadicExpressionOp::Coalesce
)]
#[case(VariadicExpressionOp::Array, proto_expr::VariadicExpressionOp::Array)]
fn from_variadic_expression_op(
#[case] op: VariadicExpressionOp,
#[case] expected: proto_expr::VariadicExpressionOp,
) {
assert_eq!(
proto_expr::VariadicExpressionOp::from(op) as i32,
expected as i32
);
}
#[rstest]
#[case(UnaryPredicateOp::IsNull, proto_expr::UnaryPredicateOp::IsNull)]
fn from_unary_predicate_op(
#[case] op: UnaryPredicateOp,
#[case] expected: proto_expr::UnaryPredicateOp,
) {
assert_eq!(
proto_expr::UnaryPredicateOp::from(op) as i32,
expected as i32
);
}
#[rstest]
#[case(BinaryPredicateOp::LessThan, proto_expr::BinaryPredicateOp::LessThan)]
#[case(
BinaryPredicateOp::GreaterThan,
proto_expr::BinaryPredicateOp::GreaterThan
)]
#[case(BinaryPredicateOp::Equal, proto_expr::BinaryPredicateOp::Equal)]
#[case(BinaryPredicateOp::Distinct, proto_expr::BinaryPredicateOp::Distinct)]
#[case(BinaryPredicateOp::In, proto_expr::BinaryPredicateOp::In)]
fn from_binary_predicate_op(
#[case] op: BinaryPredicateOp,
#[case] expected: proto_expr::BinaryPredicateOp,
) {
assert_eq!(
proto_expr::BinaryPredicateOp::from(op) as i32,
expected as i32
);
}
#[rstest]
#[case(JunctionPredicateOp::And, proto_expr::JunctionPredicateOp::And)]
#[case(JunctionPredicateOp::Or, proto_expr::JunctionPredicateOp::Or)]
fn from_junction_predicate_op(
#[case] op: JunctionPredicateOp,
#[case] expected: proto_expr::JunctionPredicateOp,
) {
assert_eq!(
proto_expr::JunctionPredicateOp::from(op) as i32,
expected as i32
);
}
#[rstest]
#[case(Scalar::Integer(7), "integer")]
#[case(Scalar::Long(7), "long")]
#[case(Scalar::Short(7), "short")]
#[case(Scalar::Byte(7), "byte")]
#[case(Scalar::Float(1.5), "float")]
#[case(Scalar::Double(1.5), "double")]
#[case(Scalar::String("hi".into()), "string")]
#[case(Scalar::Boolean(true), "boolean")]
#[case(Scalar::Timestamp(9), "timestamp")]
#[case(Scalar::TimestampNtz(9), "timestamp_ntz")]
#[case(Scalar::Date(9), "date")]
#[case(Scalar::Binary(vec![1, 2, 3]), "binary")]
#[case(
Scalar::Decimal(
DecimalData::try_new(1234i128, DecimalType::try_new(10, 2).unwrap()).unwrap(),
),
"decimal"
)]
#[case(Scalar::Null(DataType::LONG), "null")]
#[case(
Scalar::Struct(
StructData::try_new(
vec![StructField::nullable("a", DataType::INTEGER)],
vec![Scalar::Integer(1)],
)
.unwrap(),
),
"struct"
)]
#[case(
Scalar::Array(
ArrayData::try_new(ArrayType::new(DataType::INTEGER, false), [1, 2, 3]).unwrap(),
),
"array"
)]
#[case(
Scalar::Map(
MapData::try_new(MapType::new(DataType::STRING, DataType::INTEGER, false), [("k", 1)])
.unwrap(),
),
"map"
)]
fn from_scalar(#[case] value: Scalar, #[case] expected: &str) {
use proto_expr::scalar::Value;
let kind = match proto_expr::Scalar::from(&value).value.unwrap() {
Value::Integer(_) => "integer",
Value::Long(_) => "long",
Value::Short(_) => "short",
Value::Byte(_) => "byte",
Value::Float(_) => "float",
Value::Double(_) => "double",
Value::String(_) => "string",
Value::Boolean(_) => "boolean",
Value::Timestamp(_) => "timestamp",
Value::TimestampNtz(_) => "timestamp_ntz",
Value::TimestampNanos(_) => "timestamp_nanos",
Value::TimestampNanosNtz(_) => "timestamp_nanos_ntz",
Value::Date(_) => "date",
Value::Binary(_) => "binary",
Value::Decimal(_) => "decimal",
Value::Null(_) => "null",
Value::Struct(_) => "struct",
Value::Array(_) => "array",
Value::Map(_) => "map",
};
assert_eq!(kind, expected);
}
#[test]
fn from_decimal_data() {
let decimal = DecimalData::try_new(1234i128, DecimalType::try_new(10, 2).unwrap()).unwrap();
let proto_expr::scalar::Value::Decimal(decimal) = scalar_value_of(Scalar::Decimal(decimal))
else {
panic!("expected a decimal scalar");
};
assert_eq!(decimal.bits, 1234i128.to_be_bytes().to_vec());
let decimal_type = decimal.decimal_type.expect("decimal type present");
assert_eq!((decimal_type.precision, decimal_type.scale), (10, 2));
}
#[test]
fn from_struct_data() {
let struct_data = StructData::try_new(
vec![StructField::nullable("a", DataType::INTEGER)],
vec![Scalar::Integer(1)],
)
.unwrap();
let proto_expr::scalar::Value::Struct(struct_data) =
scalar_value_of(Scalar::Struct(struct_data))
else {
panic!("expected a struct scalar");
};
assert_eq!(struct_data.fields.len(), 1);
assert_eq!(struct_data.values.len(), 1);
assert_eq!(struct_data.fields[0].name, "a");
}
#[test]
fn from_array_data() {
let array_data =
ArrayData::try_new(ArrayType::new(DataType::INTEGER, false), [1, 2, 3]).unwrap();
let proto_expr::scalar::Value::Array(array_data) =
scalar_value_of(Scalar::Array(array_data))
else {
panic!("expected an array scalar");
};
assert!(array_data.array_type.is_some());
assert_eq!(array_data.elements.len(), 3);
}
#[test]
fn from_map_data() {
let map_data = MapData::try_new(
MapType::new(DataType::STRING, DataType::INTEGER, false),
[("k", 1)],
)
.unwrap();
let proto_expr::scalar::Value::Map(map_data) = scalar_value_of(Scalar::Map(map_data))
else {
panic!("expected a map scalar");
};
assert!(map_data.map_type.is_some());
assert_eq!(map_data.pairs.len(), 1);
assert!(map_data.pairs[0].key.is_some());
assert!(map_data.pairs[0].value.is_some());
}
#[rstest]
#[case(DataType::INTEGER, "primitive")]
#[case(ArrayType::new(DataType::INTEGER, true).into(), "array")]
#[case(
StructType::try_new(vec![StructField::nullable("a", DataType::INTEGER)])
.unwrap()
.into(),
"struct"
)]
#[case(MapType::new(DataType::STRING, DataType::INTEGER, true).into(), "map")]
#[case(DataType::unshredded_variant(), "variant")]
fn from_data_type(#[case] value: DataType, #[case] expected: &str) {
use proto_schema::data_type::Kind;
let kind = match proto_schema::DataType::from(&value).kind.unwrap() {
Kind::Primitive(_) => "primitive",
Kind::Array(_) => "array",
Kind::Struct(_) => "struct",
Kind::Map(_) => "map",
Kind::Variant(_) => "variant",
};
assert_eq!(kind, expected);
}
#[rstest]
#[case(PrimitiveType::String, proto_schema::SimplePrimitiveType::String)]
#[case(PrimitiveType::Long, proto_schema::SimplePrimitiveType::Long)]
#[case(PrimitiveType::Integer, proto_schema::SimplePrimitiveType::Integer)]
#[case(PrimitiveType::Short, proto_schema::SimplePrimitiveType::Short)]
#[case(PrimitiveType::Byte, proto_schema::SimplePrimitiveType::Byte)]
#[case(PrimitiveType::Float, proto_schema::SimplePrimitiveType::Float)]
#[case(PrimitiveType::Double, proto_schema::SimplePrimitiveType::Double)]
#[case(PrimitiveType::Boolean, proto_schema::SimplePrimitiveType::Boolean)]
#[case(PrimitiveType::Binary, proto_schema::SimplePrimitiveType::Binary)]
#[case(PrimitiveType::Date, proto_schema::SimplePrimitiveType::Date)]
#[case(PrimitiveType::Timestamp, proto_schema::SimplePrimitiveType::Timestamp)]
#[case(
PrimitiveType::TimestampNtz,
proto_schema::SimplePrimitiveType::TimestampNtz
)]
#[case(PrimitiveType::Void, proto_schema::SimplePrimitiveType::Void)]
#[case(
PrimitiveType::IntervalYearMonth,
proto_schema::SimplePrimitiveType::IntervalYearMonth
)]
#[case(
PrimitiveType::IntervalDayTime,
proto_schema::SimplePrimitiveType::IntervalDayTime
)]
fn from_primitive_type(
#[case] primitive: PrimitiveType,
#[case] expected: proto_schema::SimplePrimitiveType,
) {
let Some(proto_schema::primitive_type::Kind::Simple(simple)) =
proto_schema::PrimitiveType::from(&primitive).kind
else {
panic!("expected a simple primitive type");
};
assert_eq!(simple, expected as i32);
}
#[test]
fn from_decimal_type() {
let primitive = PrimitiveType::decimal(10, 2).unwrap();
let Some(proto_schema::primitive_type::Kind::Decimal(decimal)) =
proto_schema::PrimitiveType::from(&primitive).kind
else {
panic!("expected a decimal primitive type");
};
assert_eq!((decimal.precision, decimal.scale), (10, 2));
}
#[rstest]
#[case(true)]
#[case(false)]
fn from_array_type(#[case] contains_null: bool) {
let proto =
proto_schema::ArrayType::from(&ArrayType::new(DataType::INTEGER, contains_null));
assert!(proto.element_type.is_some());
assert_eq!(proto.contains_null, contains_null);
}
#[rstest]
#[case(true)]
#[case(false)]
fn from_map_type(#[case] value_contains_null: bool) {
let proto = proto_schema::MapType::from(&MapType::new(
DataType::STRING,
DataType::INTEGER,
value_contains_null,
));
assert!(proto.key_type.is_some());
assert!(proto.value_type.is_some());
assert_eq!(proto.value_contains_null, value_contains_null);
}
#[test]
fn from_struct_type() {
let struct_type = StructType::try_new(vec![
StructField::nullable("a", DataType::INTEGER),
StructField::not_null("b", DataType::STRING),
])
.unwrap();
let proto = proto_schema::StructType::from(&struct_type);
assert_eq!(proto.fields.len(), 2);
assert!(proto.fields[0].nullable);
assert!(!proto.fields[1].nullable);
}
#[test]
fn from_struct_field() {
let field = StructField::nullable("a", DataType::INTEGER)
.with_metadata([("k", MetadataValue::Number(7))]);
let proto = proto_schema::StructField::from(&field);
assert_eq!(proto.name, "a");
assert!(proto.nullable);
assert!(proto.data_type.is_some());
assert_eq!(
proto.metadata["k"].value,
Some(proto_schema::metadata_value::Value::Number(7))
);
}
#[rstest]
#[case(
MetadataValue::Number(5),
proto_schema::metadata_value::Value::Number(5)
)]
#[case(
MetadataValue::String("s".to_string()),
proto_schema::metadata_value::Value::String("s".to_string())
)]
#[case(
MetadataValue::Boolean(true),
proto_schema::metadata_value::Value::Boolean(true)
)]
#[case(
MetadataValue::Other(serde_json::json!([1, 2])),
proto_schema::metadata_value::Value::OtherJson("[1,2]".to_string())
)]
fn from_metadata_value(
#[case] metadata: MetadataValue,
#[case] expected: proto_schema::metadata_value::Value,
) {
assert_eq!(
proto_schema::MetadataValue::from(&metadata).value.unwrap(),
expected
);
}
fn assert_data_type_round_trips(data_type: DataType) {
let decoded = DataType::try_from(proto_schema::DataType::from(&data_type));
assert_eq!(decoded.expect("decode succeeds"), data_type);
}
fn assert_schema_round_trips(schema: StructType) {
let decoded = StructType::try_from(proto_schema::StructType::from(&schema));
assert_eq!(decoded.expect("decode succeeds"), schema);
}
#[rstest]
fn round_trip_primitive(
#[values(
PrimitiveType::String,
PrimitiveType::Long,
PrimitiveType::Integer,
PrimitiveType::Short,
PrimitiveType::Byte,
PrimitiveType::Float,
PrimitiveType::Double,
PrimitiveType::Boolean,
PrimitiveType::Binary,
PrimitiveType::Date,
PrimitiveType::Timestamp,
PrimitiveType::TimestampNtz,
PrimitiveType::Void,
PrimitiveType::IntervalYearMonth,
PrimitiveType::IntervalDayTime
)]
primitive: PrimitiveType,
) {
assert_data_type_round_trips(DataType::Primitive(primitive));
}
#[rstest]
#[case(1, 0)]
#[case(10, 2)]
#[case(38, 0)]
#[case(38, 38)]
fn round_trip_decimal(#[case] precision: u8, #[case] scale: u8) {
assert_data_type_round_trips(DataType::Primitive(
PrimitiveType::decimal(precision, scale).unwrap(),
));
}
#[rstest]
#[case(DataType::from(ArrayType::new(DataType::INTEGER, true)))]
#[case(DataType::from(ArrayType::new(DataType::STRING, false)))]
#[case(DataType::from(MapType::new(DataType::STRING, DataType::LONG, true)))]
#[case(DataType::from(MapType::new(DataType::INTEGER, DataType::BOOLEAN, false)))]
#[case(DataType::from(ArrayType::new(
MapType::new(DataType::STRING, DataType::LONG, true),
true
)))]
#[case(DataType::from(StructType::try_new(vec![
StructField::nullable("a", DataType::INTEGER),
StructField::not_null("b", DataType::STRING),
]).unwrap()))]
fn round_trip_composite(#[case] data_type: DataType) {
assert_data_type_round_trips(data_type);
}
#[rstest]
#[case(MetadataValue::Number(7))]
#[case(MetadataValue::String("hello".to_string()))]
#[case(MetadataValue::Boolean(true))]
#[case(MetadataValue::Other(serde_json::json!({"nested": [1, 2, 3]})))]
fn round_trip_field_metadata(#[case] value: MetadataValue) {
let field = StructField::nullable("f", DataType::INTEGER).with_metadata([("k", value)]);
let decoded = StructField::try_from(proto_schema::StructField::from(&field));
assert_eq!(decoded.expect("decode succeeds"), field);
}
#[test]
fn round_trip_full_schema() {
let schema = StructType::try_new(vec![
StructField::nullable("id", DataType::LONG)
.with_metadata([("k", MetadataValue::Number(7))]),
StructField::not_null("name", DataType::STRING),
StructField::nullable("scores", ArrayType::new(DataType::INTEGER, true)),
StructField::nullable(
"attrs",
MapType::new(DataType::STRING, DataType::LONG, true),
),
StructField::nullable(
"price",
DataType::Primitive(PrimitiveType::decimal(10, 2).unwrap()),
),
StructField::nullable(
"nested",
StructType::try_new(vec![StructField::not_null("inner", DataType::BOOLEAN)])
.unwrap(),
),
])
.unwrap();
assert_schema_round_trips(schema);
}
#[test]
fn variant_decodes_to_unshredded() {
let shredded = DataType::Variant(Box::new(
StructType::try_new(vec![
StructField::not_null("metadata", DataType::BINARY),
StructField::not_null("value", DataType::BINARY),
StructField::nullable("typed_value", DataType::INTEGER),
])
.unwrap(),
));
let decoded = DataType::try_from(proto_schema::DataType::from(&shredded));
assert_eq!(
decoded.expect("decode succeeds"),
DataType::unshredded_variant()
);
}
fn primitive_data_type(
kind: Option<proto_schema::primitive_type::Kind>,
) -> proto_schema::DataType {
proto_schema::DataType {
kind: Some(proto_schema::data_type::Kind::Primitive(
proto_schema::PrimitiveType { kind },
)),
}
}
fn simple_primitive_data_type(value: i32) -> proto_schema::DataType {
primitive_data_type(Some(proto_schema::primitive_type::Kind::Simple(value)))
}
fn decimal_data_type(precision: u32, scale: u32) -> proto_schema::DataType {
primitive_data_type(Some(proto_schema::primitive_type::Kind::Decimal(
proto_schema::DecimalType { precision, scale },
)))
}
#[rstest]
#[case::missing_data_type_kind(proto_schema::DataType { kind: None })]
#[case::missing_primitive_kind(primitive_data_type(None))]
#[case::unspecified_primitive(simple_primitive_data_type(
proto_schema::SimplePrimitiveType::Unspecified as i32
))]
#[case::unknown_primitive(simple_primitive_data_type(9999))]
#[case::decimal_zero_precision(decimal_data_type(0, 0))]
#[case::decimal_precision_too_large(decimal_data_type(39, 0))]
#[case::decimal_scale_exceeds_precision(decimal_data_type(5, 6))]
#[case::decimal_precision_out_of_u8_range(decimal_data_type(256, 0))]
#[case::array_missing_element_type(proto_schema::DataType {
kind: Some(proto_schema::data_type::Kind::Array(Box::new(proto_schema::ArrayType {
element_type: None,
contains_null: true,
}))),
})]
#[case::map_missing_key_type(proto_schema::DataType {
kind: Some(proto_schema::data_type::Kind::Map(Box::new(proto_schema::MapType {
key_type: None,
value_type: Some(Box::new(proto_schema::DataType::from(&DataType::STRING))),
value_contains_null: true,
}))),
})]
#[case::map_missing_value_type(proto_schema::DataType {
kind: Some(proto_schema::data_type::Kind::Map(Box::new(proto_schema::MapType {
key_type: Some(Box::new(proto_schema::DataType::from(&DataType::STRING))),
value_type: None,
value_contains_null: true,
}))),
})]
fn data_type_decode_rejects_invalid(#[case] proto: proto_schema::DataType) {
assert!(DataType::try_from(proto).is_err());
}
#[test]
fn struct_field_decode_rejects_missing_data_type() {
let proto = proto_schema::StructField {
name: "f".to_string(),
data_type: None,
nullable: true,
metadata: Default::default(),
};
assert!(StructField::try_from(proto).is_err());
}
#[rstest]
#[case::missing_value(proto_schema::MetadataValue { value: None })]
#[case::invalid_other_json(proto_schema::MetadataValue {
value: Some(proto_schema::metadata_value::Value::OtherJson("not json".to_string())),
})]
fn metadata_value_decode_rejects_invalid(#[case] proto: proto_schema::MetadataValue) {
assert!(MetadataValue::try_from(proto).is_err());
}
}