|
15 | 15 | // specific language governing permissions and limitations |
16 | 16 | // under the License. |
17 | 17 |
|
18 | | -use crate::logical_plan::producer::SubstraitProducer; |
| 18 | +use crate::logical_plan::producer::{SubstraitProducer, to_substrait_type_from_field}; |
19 | 19 | use datafusion::common::DFSchemaRef; |
20 | | -use datafusion::logical_expr::expr; |
21 | 20 | use datafusion::logical_expr::expr::AggregateFunctionParams; |
| 21 | +use datafusion::logical_expr::{Expr, ExprSchemable, expr}; |
22 | 22 | use substrait::proto::aggregate_function::AggregationInvocation; |
23 | 23 | use substrait::proto::aggregate_rel::Measure; |
24 | 24 | use substrait::proto::function_argument::ArgType; |
@@ -54,13 +54,15 @@ pub fn from_aggregate_function( |
54 | 54 | }); |
55 | 55 | } |
56 | 56 | let function_anchor = producer.register_function(func.name().to_string()); |
| 57 | + let (_, output_field) = Expr::AggregateFunction(agg_fn.clone()).to_field(schema)?; |
| 58 | + let output_type = to_substrait_type_from_field(producer, &output_field)?; |
57 | 59 | #[expect(deprecated)] |
58 | 60 | Ok(Measure { |
59 | 61 | measure: Some(AggregateFunction { |
60 | 62 | function_reference: function_anchor, |
61 | 63 | arguments, |
62 | 64 | sorts, |
63 | | - output_type: None, |
| 65 | + output_type: Some(output_type), |
64 | 66 | invocation: match distinct { |
65 | 67 | true => AggregationInvocation::Distinct as i32, |
66 | 68 | false => AggregationInvocation::All as i32, |
@@ -93,3 +95,51 @@ fn to_substrait_sort_field( |
93 | 95 | sort_kind: Some(SortKind::Direction(sort_kind.into())), |
94 | 96 | }) |
95 | 97 | } |
| 98 | + |
| 99 | +#[cfg(test)] |
| 100 | +mod tests { |
| 101 | + use crate::logical_plan::producer::{ |
| 102 | + DefaultSubstraitProducer, SubstraitProducer, to_substrait_type, |
| 103 | + }; |
| 104 | + use datafusion::arrow::datatypes::{DataType, Field, Schema}; |
| 105 | + use datafusion::common::{DFSchema, DFSchemaRef}; |
| 106 | + use datafusion::execution::SessionStateBuilder; |
| 107 | + use datafusion::functions_aggregate::expr_fn::{avg, count, min, sum}; |
| 108 | + use datafusion::logical_expr::Expr; |
| 109 | + use datafusion::prelude::col; |
| 110 | + |
| 111 | + #[test] |
| 112 | + fn aggregate_function_output_type() -> datafusion::common::Result<()> { |
| 113 | + let state = SessionStateBuilder::default().build(); |
| 114 | + let schema = |
| 115 | + DFSchemaRef::new(DFSchema::try_from(Schema::new(vec![Field::new( |
| 116 | + "i", |
| 117 | + DataType::Int64, |
| 118 | + false, |
| 119 | + )]))?); |
| 120 | + let mut producer = DefaultSubstraitProducer::new(&state); |
| 121 | + |
| 122 | + // (aggregate, expected output type, expected nullability) |
| 123 | + let cases = [ |
| 124 | + (count(col("i")), DataType::Int64, false), |
| 125 | + (sum(col("i")), DataType::Int64, true), |
| 126 | + (avg(col("i")), DataType::Float64, true), |
| 127 | + (min(col("i")), DataType::Int64, true), |
| 128 | + ]; |
| 129 | + |
| 130 | + for (expr, expected_type, expected_nullable) in cases { |
| 131 | + let Expr::AggregateFunction(agg_fn) = &expr else { |
| 132 | + panic!("AggregateFunction expected, got {expr}") |
| 133 | + }; |
| 134 | + let measure = producer.handle_aggregate_function(agg_fn, &schema)?; |
| 135 | + let expected = |
| 136 | + to_substrait_type(&mut producer, &expected_type, expected_nullable)?; |
| 137 | + let output_type = measure |
| 138 | + .measure |
| 139 | + .expect("Measure should contain an AggregateFunction") |
| 140 | + .output_type; |
| 141 | + assert_eq!(output_type, Some(expected), "output_type for {expr}"); |
| 142 | + } |
| 143 | + Ok(()) |
| 144 | + } |
| 145 | +} |
0 commit comments