datafusion_comet_spark_expr/string_funcs/
string_space.rs#![allow(deprecated)]
use crate::kernels::strings::string_space;
use arrow::record_batch::RecordBatch;
use arrow_schema::{DataType, Schema};
use datafusion::logical_expr::ColumnarValue;
use datafusion_common::DataFusionError;
use datafusion_physical_expr::PhysicalExpr;
use std::{
any::Any,
fmt::{Display, Formatter},
hash::Hash,
sync::Arc,
};
#[derive(Debug, Eq)]
pub struct StringSpaceExpr {
pub child: Arc<dyn PhysicalExpr>,
}
impl Hash for StringSpaceExpr {
fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
self.child.hash(state);
}
}
impl PartialEq for StringSpaceExpr {
fn eq(&self, other: &Self) -> bool {
self.child.eq(&other.child)
}
}
impl StringSpaceExpr {
pub fn new(child: Arc<dyn PhysicalExpr>) -> Self {
Self { child }
}
}
impl Display for StringSpaceExpr {
fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
write!(f, "StringSpace [child: {}] ", self.child)
}
}
impl PhysicalExpr for StringSpaceExpr {
fn as_any(&self) -> &dyn Any {
self
}
fn data_type(&self, input_schema: &Schema) -> datafusion_common::Result<DataType> {
match self.child.data_type(input_schema)? {
DataType::Dictionary(key_type, _) => {
Ok(DataType::Dictionary(key_type, Box::new(DataType::Utf8)))
}
_ => Ok(DataType::Utf8),
}
}
fn nullable(&self, _: &Schema) -> datafusion_common::Result<bool> {
Ok(true)
}
fn evaluate(&self, batch: &RecordBatch) -> datafusion_common::Result<ColumnarValue> {
let arg = self.child.evaluate(batch)?;
match arg {
ColumnarValue::Array(array) => {
let result = string_space(&array)?;
Ok(ColumnarValue::Array(result))
}
_ => Err(DataFusionError::Execution(
"StringSpace(scalar) should be fold in Spark JVM side.".to_string(),
)),
}
}
fn children(&self) -> Vec<&Arc<dyn PhysicalExpr>> {
vec![&self.child]
}
fn with_new_children(
self: Arc<Self>,
children: Vec<Arc<dyn PhysicalExpr>>,
) -> datafusion_common::Result<Arc<dyn PhysicalExpr>> {
Ok(Arc::new(StringSpaceExpr::new(Arc::clone(&children[0]))))
}
}