fix(functions): manually drop state in function eval_aggr function

common/functions/src/aggregates/adaptors/aggregate_null_unary_adaptor.rs

@@ -234,7 +234,7 @@ impl<const NULLABLE_RESULT: bool> AggregateFunction for AggregateNullUnaryAdapto
     }
 
     unsafe fn drop_state(&self, place: StateAddr) {
-        self.nested.drop_state(place)
+        self.nested.drop_state(self.nested_place(place))
     }
 
     fn convert_const_to_full(&self) -> bool {

common/functions/src/aggregates/adaptors/aggregate_null_variadic_adaptor.rs

@@ -247,7 +247,7 @@ impl<const NULLABLE_RESULT: bool, const STKIP_NULL: bool> AggregateFunction
     }
 
     unsafe fn drop_state(&self, place: StateAddr) {
-        self.nested.drop_state(place)
+        self.nested.drop_state(self.nested_place(place))
     }
 
     fn convert_const_to_full(&self) -> bool {

common/functions/src/aggregates/aggregator_common.rs

@@ -89,16 +89,30 @@ pub fn eval_aggr(
     let cols: Vec<ColumnRef> = columns.iter().map(|c| c.column().clone()).collect();
 
     let func = factory.get(name, params, arguments)?;
+    let data_type = func.return_type()?;
 
     let arena = Bump::new();
     let place = arena.alloc_layout(func.state_layout());
     let addr = place.into();
     func.init_state(addr);
-    func.accumulate(addr, &cols, None, rows)?;
 
-    let data_type = func.return_type()?;
-    let mut builder = data_type.create_mutable(1024);
-    func.merge_result(addr, builder.as_mut())?;
+    let f = func.clone();
+    // we need a temporary function to catch the errors
+    let apply = || -> Result<ColumnRef> {
+        func.accumulate(addr, &cols, None, rows)?;
+        let mut builder = data_type.create_mutable(1024);
+        func.merge_result(addr, builder.as_mut())?;
+
+        Ok(builder.to_column())
+    };
 
-    Ok(builder.to_column())
+    let result = apply();
+
+    if f.need_manual_drop_state() {
+        unsafe {
+            f.drop_state(addr);
+        }
+    }
+    drop(arena);
+    result
 }