datafusion_comet_spark_expr/
unbound.rs

1// Licensed to the Apache Software Foundation (ASF) under one
2// or more contributor license agreements.  See the NOTICE file
3// distributed with this work for additional information
4// regarding copyright ownership.  The ASF licenses this file
5// to you under the Apache License, Version 2.0 (the
6// "License"); you may not use this file except in compliance
7// with the License.  You may obtain a copy of the License at
8//
9//   http://www.apache.org/licenses/LICENSE-2.0
10//
11// Unless required by applicable law or agreed to in writing,
12// software distributed under the License is distributed on an
13// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14// KIND, either express or implied.  See the License for the
15// specific language governing permissions and limitations
16// under the License.
17
18use arrow::array::RecordBatch;
19use arrow::datatypes::{DataType, Schema};
20use datafusion::common::{internal_err, Result};
21use datafusion::physical_expr::PhysicalExpr;
22use datafusion::physical_plan::ColumnarValue;
23use std::fmt::Formatter;
24use std::{hash::Hash, sync::Arc};
25
26/// This is similar to `UnKnownColumn` in DataFusion, but it has data type.
27/// This is only used when the column is not bound to a schema, for example, the
28/// inputs to aggregation functions in final aggregation. In the case, we cannot
29/// bind the aggregation functions to the input schema which is grouping columns
30/// and aggregate buffer attributes in Spark (DataFusion has different design).
31/// But when creating certain aggregation functions, we need to know its input
32/// data types. As `UnKnownColumn` doesn't have data type, we implement this
33/// `UnboundColumn` to carry the data type.
34#[derive(Debug, Hash, PartialEq, Eq, Clone)]
35pub struct UnboundColumn {
36    name: String,
37    datatype: DataType,
38}
39
40impl UnboundColumn {
41    /// Create a new unbound column expression
42    pub fn new(name: &str, datatype: DataType) -> Self {
43        Self {
44            name: name.to_owned(),
45            datatype,
46        }
47    }
48
49    /// Get the column name
50    pub fn name(&self) -> &str {
51        &self.name
52    }
53}
54
55impl std::fmt::Display for UnboundColumn {
56    fn fmt(&self, f: &mut std::fmt::Formatter) -> std::fmt::Result {
57        write!(f, "{}, datatype: {}", self.name, self.datatype)
58    }
59}
60
61impl PhysicalExpr for UnboundColumn {
62    /// Return a reference to Any that can be used for downcasting
63    fn as_any(&self) -> &dyn std::any::Any {
64        self
65    }
66
67    fn fmt_sql(&self, _: &mut Formatter<'_>) -> std::fmt::Result {
68        unimplemented!()
69    }
70
71    /// Get the data type of this expression, given the schema of the input
72    fn data_type(&self, _input_schema: &Schema) -> Result<DataType> {
73        Ok(self.datatype.clone())
74    }
75
76    /// Decide whether this expression is nullable, given the schema of the input
77    fn nullable(&self, _input_schema: &Schema) -> Result<bool> {
78        Ok(true)
79    }
80
81    /// Evaluate the expression
82    fn evaluate(&self, _batch: &RecordBatch) -> Result<ColumnarValue> {
83        internal_err!("UnboundColumn::evaluate() should not be called")
84    }
85
86    fn children(&self) -> Vec<&Arc<dyn PhysicalExpr>> {
87        vec![]
88    }
89
90    fn with_new_children(
91        self: Arc<Self>,
92        _children: Vec<Arc<dyn PhysicalExpr>>,
93    ) -> Result<Arc<dyn PhysicalExpr>> {
94        Ok(self)
95    }
96}
datafusion_comet_spark_expr/unbound.rs

datafusion_comet_spark_expr/
unbound.rs
