fix: min, max with append only

dozer-sql/expression/src/aggregate.rs

@@ -5,8 +5,10 @@ pub enum AggregateFunctionType {
     Avg,
     Count,
     Max,
+    MaxAppendOnly,
     MaxValue,
     Min,
+    MinAppendOnly,
     MinValue,
     Sum,
 }
@@ -17,8 +19,10 @@ impl AggregateFunctionType {
             "avg" => Some(AggregateFunctionType::Avg),
             "count" => Some(AggregateFunctionType::Count),
             "max" => Some(AggregateFunctionType::Max),
+            "max_append_only" => Some(AggregateFunctionType::MaxAppendOnly),
             "max_value" => Some(AggregateFunctionType::MaxValue),
             "min" => Some(AggregateFunctionType::Min),
+            "min_append_only" => Some(AggregateFunctionType::MinAppendOnly),
             "min_value" => Some(AggregateFunctionType::MinValue),
             "sum" => Some(AggregateFunctionType::Sum),
             _ => None,
@@ -32,8 +36,10 @@ impl Display for AggregateFunctionType {
             AggregateFunctionType::Avg => f.write_str("AVG"),
             AggregateFunctionType::Count => f.write_str("COUNT"),
             AggregateFunctionType::Max => f.write_str("MAX"),
+            AggregateFunctionType::MaxAppendOnly => f.write_str("MAX_APPEND_ONLY"),
             AggregateFunctionType::MaxValue => f.write_str("MAX_VALUE"),
             AggregateFunctionType::Min => f.write_str("MIN"),
+            AggregateFunctionType::MinAppendOnly => f.write_str("MIN_APPEND_ONLY"),
             AggregateFunctionType::MinValue => f.write_str("MIN_VALUE"),
             AggregateFunctionType::Sum => f.write_str("SUM"),
         }

dozer-sql/expression/src/execution.rs

@@ -721,8 +721,10 @@ fn get_aggregate_function_type(
         AggregateFunctionType::Avg => validate_avg(args, schema),
         AggregateFunctionType::Count => validate_count(args, schema),
         AggregateFunctionType::Max => validate_max(args, schema),
+        AggregateFunctionType::MaxAppendOnly => validate_max_append_only(args, schema),
         AggregateFunctionType::MaxValue => validate_max_value(args, schema),
         AggregateFunctionType::Min => validate_min(args, schema),
+        AggregateFunctionType::MinAppendOnly => validate_min_append_only(args, schema),
         AggregateFunctionType::MinValue => validate_min_value(args, schema),
         AggregateFunctionType::Sum => validate_sum(args, schema),
     }
@@ -863,6 +865,97 @@ fn validate_min(args: &[Expression], schema: &Schema) -> Result<ExpressionType,
             });
         }
     };
+
+    Ok(ExpressionType::new(
+        ret_type,
+        true,
+        SourceDefinition::Dynamic,
+        false,
+    ))
+}
+
+fn validate_max_append_only(args: &[Expression], schema: &Schema) -> Result<ExpressionType, Error> {
+    let arg = validate_one_argument(args, schema, AggregateFunctionType::MaxAppendOnly)?;
+
+    let ret_type = match arg.return_type {
+        FieldType::UInt => FieldType::UInt,
+        FieldType::U128 => FieldType::U128,
+        FieldType::Int => FieldType::Int,
+        FieldType::I128 => FieldType::I128,
+        FieldType::Float => FieldType::Float,
+        FieldType::Decimal => FieldType::Decimal,
+        FieldType::Timestamp => FieldType::Timestamp,
+        FieldType::Date => FieldType::Date,
+        FieldType::Duration => FieldType::Duration,
+        FieldType::Boolean
+        | FieldType::String
+        | FieldType::Text
+        | FieldType::Binary
+        | FieldType::Json
+        | FieldType::Point => {
+            return Err(Error::InvalidFunctionArgumentType {
+                function_name: AggregateFunctionType::MaxAppendOnly.to_string(),
+                argument_index: 0,
+                actual: arg.return_type,
+                expected: vec![
+                    FieldType::Decimal,
+                    FieldType::UInt,
+                    FieldType::U128,
+                    FieldType::Int,
+                    FieldType::I128,
+                    FieldType::Float,
+                    FieldType::Timestamp,
+                    FieldType::Date,
+                    FieldType::Duration,
+                ],
+            });
+        }
+    };
+    Ok(ExpressionType::new(
+        ret_type,
+        true,
+        SourceDefinition::Dynamic,
+        false,
+    ))
+}
+
+fn validate_min_append_only(args: &[Expression], schema: &Schema) -> Result<ExpressionType, Error> {
+    let arg = validate_one_argument(args, schema, AggregateFunctionType::MinAppendOnly)?;
+
+    let ret_type = match arg.return_type {
+        FieldType::UInt => FieldType::UInt,
+        FieldType::U128 => FieldType::U128,
+        FieldType::Int => FieldType::Int,
+        FieldType::I128 => FieldType::I128,
+        FieldType::Float => FieldType::Float,
+        FieldType::Decimal => FieldType::Decimal,
+        FieldType::Timestamp => FieldType::Timestamp,
+        FieldType::Date => FieldType::Date,
+        FieldType::Duration => FieldType::Duration,
+        FieldType::Boolean
+        | FieldType::String
+        | FieldType::Text
+        | FieldType::Binary
+        | FieldType::Json
+        | FieldType::Point => {
+            return Err(Error::InvalidFunctionArgumentType {
+                function_name: AggregateFunctionType::MinAppendOnly.to_string(),
+                argument_index: 0,
+                actual: arg.return_type,
+                expected: vec![
+                    FieldType::Decimal,
+                    FieldType::UInt,
+                    FieldType::U128,
+                    FieldType::Int,
+                    FieldType::I128,
+                    FieldType::Float,
+                    FieldType::Timestamp,
+                    FieldType::Date,
+                    FieldType::Duration,
+                ],
+            });
+        }
+    };
     Ok(ExpressionType::new(
         ret_type,
         true,

dozer-sql/src/aggregation/aggregator.rs

@@ -14,7 +14,9 @@ use std::collections::BTreeMap;
 use dozer_sql_expression::aggregate::AggregateFunctionType;
 use dozer_sql_expression::execution::Expression;
 
+use crate::aggregation::max_append_only::MaxAppendOnlyAggregator;
 use crate::aggregation::max_value::MaxValueAggregator;
+use crate::aggregation::min_append_only::MinAppendOnlyAggregator;
 use crate::aggregation::min_value::MinValueAggregator;
 use crate::errors::PipelineError::{InvalidFunctionArgument, InvalidValue};
 use dozer_sql_expression::aggregate::AggregateFunctionType::MaxValue;
@@ -35,8 +37,10 @@ pub trait Aggregator: Send + Sync + Serialize + DeserializeOwned {
 pub enum AggregatorEnum {
     AvgAggregator,
     MinAggregator,
+    MinAppendOnlyAggregator,
     MinValueAggregator,
     MaxAggregator,
+    MaxAppendOnlyAggregator,
     MaxValueAggregator,
     SumAggregator,
     CountAggregator,
@@ -47,8 +51,10 @@ pub enum AggregatorType {
     Avg,
     Count,
     Max,
+    MaxAppendOnly,
     MaxValue,
     Min,
+    MinAppendOnly,
     MinValue,
     Sum,
 }
@@ -59,8 +65,10 @@ impl Display for AggregatorType {
             AggregatorType::Avg => f.write_str("avg"),
             AggregatorType::Count => f.write_str("count"),
             AggregatorType::Max => f.write_str("max"),
+            AggregatorType::MaxAppendOnly => f.write_str("max_append_only"),
             AggregatorType::MaxValue => f.write_str("max_value"),
             AggregatorType::Min => f.write_str("min"),
+            AggregatorType::MinAppendOnly => f.write_str("min_append_only"),
             AggregatorType::MinValue => f.write_str("min_value"),
             AggregatorType::Sum => f.write_str("sum"),
         }
@@ -72,8 +80,10 @@ pub fn get_aggregator_from_aggregator_type(typ: AggregatorType) -> AggregatorEnu
         AggregatorType::Avg => AvgAggregator::new().into(),
         AggregatorType::Count => CountAggregator::new().into(),
         AggregatorType::Max => MaxAggregator::new().into(),
+        AggregatorType::MaxAppendOnly => MaxAppendOnlyAggregator::new().into(),
         AggregatorType::MaxValue => MaxValueAggregator::new().into(),
         AggregatorType::Min => MinAggregator::new().into(),
+        AggregatorType::MinAppendOnly => MinAppendOnlyAggregator::new().into(),
         AggregatorType::MinValue => MinValueAggregator::new().into(),
         AggregatorType::Sum => SumAggregator::new().into(),
     }
@@ -108,6 +118,20 @@ pub fn get_aggregator_type_from_aggregation_expression(
                 .clone()],
             AggregatorType::Min,
         )),
+        Expression::AggregateFunction {
+            fun: AggregateFunctionType::MinAppendOnly,
+            args,
+        } => Ok((
+            vec![args
+                .get(0)
+                .ok_or_else(|| {
+                    PipelineError::NotEnoughArguments(
+                        AggregateFunctionType::MinAppendOnly.to_string(),
+                    )
+                })?
+                .clone()],
+            AggregatorType::MinAppendOnly,
+        )),
         Expression::AggregateFunction {
             fun: AggregateFunctionType::Max,
             args,
@@ -120,6 +144,20 @@ pub fn get_aggregator_type_from_aggregation_expression(
                 .clone()],
             AggregatorType::Max,
         )),
+        Expression::AggregateFunction {
+            fun: AggregateFunctionType::MaxAppendOnly,
+            args,
+        } => Ok((
+            vec![args
+                .get(0)
+                .ok_or_else(|| {
+                    PipelineError::NotEnoughArguments(
+                        AggregateFunctionType::MaxAppendOnly.to_string(),
+                    )
+                })?
+                .clone()],
+            AggregatorType::MaxAppendOnly,
+        )),
         Expression::AggregateFunction {
             fun: AggregateFunctionType::MaxValue,
             args,