Structural Sum finalize; thread ExecutionCtx through aggregate finalize

mhk197 · mhk197 · commit a9bf18cfe24d · 2026-07-09T17:02:08.000-07:00
Signed-off-by: Matt Katz &lt;mhkatz97@gmail.com&gt;
diff --git a/vortex-array/benches/aggregate_grouped.rs b/vortex-array/benches/aggregate_grouped.rs
@@ -163,12 +163,12 @@ where
         list_element_dtype(list_view),
     )
     .unwrap();
-    acc.accumulate_list(list_view, &mut SESSION.create_execution_ctx())
-        .unwrap();
+    let mut ctx = SESSION.create_execution_ctx();
+    acc.accumulate_list(list_view, &mut ctx).unwrap();
     let result = acc
-        .finish()
+        .finish(&mut ctx)
         .unwrap()
-        .execute::<Canonical>(&mut SESSION.create_execution_ctx())
+        .execute::<Canonical>(&mut ctx)
         .unwrap()
         .into_array();
     divan::black_box(result)
diff --git a/vortex-array/src/aggregate_fn/accumulator_grouped.rs b/vortex-array/src/aggregate_fn/accumulator_grouped.rs
@@ -238,7 +238,7 @@ pub trait DynGroupedAccumulator: 'static + Send {
 
     /// Finish the accumulation and return the final aggregate results for all groups.
     /// Resets the accumulator state for the next round of accumulation.
-    fn finish(&mut self) -> VortexResult<ArrayRef>;
+    fn finish(&mut self, ctx: &mut ExecutionCtx) -> VortexResult<ArrayRef>;
 }
 
 impl<V: AggregateFnVTable> DynGroupedAccumulator for GroupedAccumulator<V> {
@@ -276,9 +276,15 @@ impl<V: AggregateFnVTable> DynGroupedAccumulator for GroupedAccumulator<V> {
         Ok(ChunkedArray::try_new(states, self.partial_dtype.clone())?.into_array())
     }
 
-    fn finish(&mut self) -> VortexResult<ArrayRef> {
-        let states = self.flush()?;
-        let results = self.vtable.finalize(states)?;
+    fn finish(&mut self, ctx: &mut ExecutionCtx) -> VortexResult<ArrayRef> {
+        // The single-batch case (one accumulate_list call) skips the chunked wrapper so that
+        // finalize implementations can operate on the batch's states directly.
+        let states = if self.partials.len() == 1 {
+            self.partials.pop().vortex_expect("checked length")
+        } else {
+            self.flush()?
+        };
+        let results = self.vtable.finalize(states, ctx)?;
 
         vortex_ensure!(
             results.dtype() == &self.return_dtype,
diff --git a/vortex-array/src/aggregate_fn/combined.rs b/vortex-array/src/aggregate_fn/combined.rs
@@ -245,11 +245,11 @@ impl<T: BinaryCombined> AggregateFnVTable for Combined<T> {
         unreachable!("Combined::try_accumulate handles all batches")
     }
 
-    fn finalize(&self, states: ArrayRef) -> VortexResult<ArrayRef> {
+    fn finalize(&self, states: ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult<ArrayRef> {
         let l_field = states.get_item(FieldName::from(self.0.left_name()))?;
         let r_field = states.get_item(FieldName::from(self.0.right_name()))?;
-        let l_finalized = self.0.left().finalize(l_field)?;
-        let r_finalized = self.0.right().finalize(r_field)?;
+        let l_finalized = self.0.left().finalize(l_field, ctx)?;
+        let r_finalized = self.0.right().finalize(r_field, ctx)?;
         BinaryCombined::finalize(&self.0, l_finalized, r_finalized)
     }
 
diff --git a/vortex-array/src/aggregate_fn/fns/all_nan/mod.rs b/vortex-array/src/aggregate_fn/fns/all_nan/mod.rs
@@ -121,7 +121,7 @@ impl AggregateFnVTable for AllNan {
         Ok(())
     }
 
-    fn finalize(&self, partials: ArrayRef) -> VortexResult<ArrayRef> {
+    fn finalize(&self, partials: ArrayRef, _ctx: &mut ExecutionCtx) -> VortexResult<ArrayRef> {
         Ok(partials)
     }
 
diff --git a/vortex-array/src/aggregate_fn/fns/all_non_distinct/mod.rs b/vortex-array/src/aggregate_fn/fns/all_non_distinct/mod.rs
@@ -233,7 +233,7 @@ impl AggregateFnVTable for AllNonDistinct {
         }
     }
 
-    fn finalize(&self, _partials: ArrayRef) -> VortexResult<ArrayRef> {
+    fn finalize(&self, _partials: ArrayRef, _ctx: &mut ExecutionCtx) -> VortexResult<ArrayRef> {
         vortex_bail!("AllNonDistinct does not support array finalization");
     }
 
diff --git a/vortex-array/src/aggregate_fn/fns/all_non_nan/mod.rs b/vortex-array/src/aggregate_fn/fns/all_non_nan/mod.rs
@@ -111,7 +111,7 @@ impl AggregateFnVTable for AllNonNan {
         Ok(())
     }
 
-    fn finalize(&self, partials: ArrayRef) -> VortexResult<ArrayRef> {
+    fn finalize(&self, partials: ArrayRef, _ctx: &mut ExecutionCtx) -> VortexResult<ArrayRef> {
         Ok(partials)
     }
 
diff --git a/vortex-array/src/aggregate_fn/fns/all_non_null/mod.rs b/vortex-array/src/aggregate_fn/fns/all_non_null/mod.rs
@@ -101,7 +101,7 @@ impl AggregateFnVTable for AllNonNull {
         Ok(())
     }
 
-    fn finalize(&self, partials: ArrayRef) -> VortexResult<ArrayRef> {
+    fn finalize(&self, partials: ArrayRef, _ctx: &mut ExecutionCtx) -> VortexResult<ArrayRef> {
         Ok(partials)
     }
 
diff --git a/vortex-array/src/aggregate_fn/fns/all_null/mod.rs b/vortex-array/src/aggregate_fn/fns/all_null/mod.rs
@@ -104,7 +104,7 @@ impl AggregateFnVTable for AllNull {
         Ok(())
     }
 
-    fn finalize(&self, partials: ArrayRef) -> VortexResult<ArrayRef> {
+    fn finalize(&self, partials: ArrayRef, _ctx: &mut ExecutionCtx) -> VortexResult<ArrayRef> {
         Ok(partials)
     }
 
diff --git a/vortex-array/src/aggregate_fn/fns/bounded_max/mod.rs b/vortex-array/src/aggregate_fn/fns/bounded_max/mod.rs
@@ -279,7 +279,7 @@ impl AggregateFnVTable for BoundedMax {
         Ok(())
     }
 
-    fn finalize(&self, partials: ArrayRef) -> VortexResult<ArrayRef> {
+    fn finalize(&self, partials: ArrayRef, _ctx: &mut ExecutionCtx) -> VortexResult<ArrayRef> {
         partials.get_item(BOUNDED_MAX_BOUND)
     }
 
diff --git a/vortex-array/src/aggregate_fn/fns/bounded_min/mod.rs b/vortex-array/src/aggregate_fn/fns/bounded_min/mod.rs
@@ -197,7 +197,7 @@ impl AggregateFnVTable for BoundedMin {
         Ok(())
     }
 
-    fn finalize(&self, partials: ArrayRef) -> VortexResult<ArrayRef> {
+    fn finalize(&self, partials: ArrayRef, _ctx: &mut ExecutionCtx) -> VortexResult<ArrayRef> {
         Ok(partials)
     }
 
diff --git a/vortex-array/src/aggregate_fn/fns/count/grouped.rs b/vortex-array/src/aggregate_fn/fns/count/grouped.rs
@@ -122,7 +122,7 @@ mod tests {
             elem_dtype.clone(),
         )?;
         acc.accumulate_list(groups, ctx)?;
-        acc.finish()
+        acc.finish(ctx)
     }
 
     /// Reference valid-counts (non-nullable `U64`), one per group.
@@ -235,7 +235,7 @@ mod tests {
         let mut acc =
             GroupedAccumulator::try_new(Count, NumericalAggregateOpts::include_nans(), elem_dtype)?;
         acc.accumulate_list(&groups, &mut ctx)?;
-        let actual = acc.finish()?;
+        let actual = acc.finish(&mut ctx)?;
         let expected = PrimitiveArray::new(buffer![2u64, 1], Validity::NonNullable).into_array();
         assert_arrays_eq!(&actual, &expected, &mut ctx);
         Ok(())
diff --git a/vortex-array/src/aggregate_fn/fns/count/mod.rs b/vortex-array/src/aggregate_fn/fns/count/mod.rs
@@ -115,7 +115,7 @@ impl AggregateFnVTable for Count {
         unreachable!("Count::try_accumulate handles all arrays")
     }
 
-    fn finalize(&self, partials: ArrayRef) -> VortexResult<ArrayRef> {
+    fn finalize(&self, partials: ArrayRef, _ctx: &mut ExecutionCtx) -> VortexResult<ArrayRef> {
         Ok(partials)
     }
 
diff --git a/vortex-array/src/aggregate_fn/fns/first/mod.rs b/vortex-array/src/aggregate_fn/fns/first/mod.rs
@@ -117,7 +117,7 @@ impl AggregateFnVTable for First {
         unreachable!("First::try_accumulate handles all arrays")
     }
 
-    fn finalize(&self, partials: ArrayRef) -> VortexResult<ArrayRef> {
+    fn finalize(&self, partials: ArrayRef, _ctx: &mut ExecutionCtx) -> VortexResult<ArrayRef> {
         Ok(partials)
     }
 
diff --git a/vortex-array/src/aggregate_fn/fns/is_constant/mod.rs b/vortex-array/src/aggregate_fn/fns/is_constant/mod.rs
@@ -420,7 +420,7 @@ impl AggregateFnVTable for IsConstant {
         }
     }
 
-    fn finalize(&self, partials: ArrayRef) -> VortexResult<ArrayRef> {
+    fn finalize(&self, partials: ArrayRef, _ctx: &mut ExecutionCtx) -> VortexResult<ArrayRef> {
         partials.get_item(NAMES.get(0).vortex_expect("out of bounds").clone())
     }
 
diff --git a/vortex-array/src/aggregate_fn/fns/is_sorted/mod.rs b/vortex-array/src/aggregate_fn/fns/is_sorted/mod.rs
@@ -514,7 +514,7 @@ impl AggregateFnVTable for IsSorted {
         }
     }
 
-    fn finalize(&self, partials: ArrayRef) -> VortexResult<ArrayRef> {
+    fn finalize(&self, partials: ArrayRef, _ctx: &mut ExecutionCtx) -> VortexResult<ArrayRef> {
         partials.get_item(NAMES.get(0).vortex_expect("out of bounds").clone())
     }
 
diff --git a/vortex-array/src/aggregate_fn/fns/last/mod.rs b/vortex-array/src/aggregate_fn/fns/last/mod.rs
@@ -115,7 +115,7 @@ impl AggregateFnVTable for Last {
         unreachable!("Last::try_accumulate handles all arrays")
     }
 
-    fn finalize(&self, partials: ArrayRef) -> VortexResult<ArrayRef> {
+    fn finalize(&self, partials: ArrayRef, _ctx: &mut ExecutionCtx) -> VortexResult<ArrayRef> {
         Ok(partials)
     }
 
diff --git a/vortex-array/src/aggregate_fn/fns/max/mod.rs b/vortex-array/src/aggregate_fn/fns/max/mod.rs
@@ -205,7 +205,7 @@ impl AggregateFnVTable for Max {
         Ok(())
     }
 
-    fn finalize(&self, partials: ArrayRef) -> VortexResult<ArrayRef> {
+    fn finalize(&self, partials: ArrayRef, _ctx: &mut ExecutionCtx) -> VortexResult<ArrayRef> {
         Ok(partials)
     }
 
diff --git a/vortex-array/src/aggregate_fn/fns/min/mod.rs b/vortex-array/src/aggregate_fn/fns/min/mod.rs
@@ -205,7 +205,7 @@ impl AggregateFnVTable for Min {
         Ok(())
     }
 
-    fn finalize(&self, partials: ArrayRef) -> VortexResult<ArrayRef> {
+    fn finalize(&self, partials: ArrayRef, _ctx: &mut ExecutionCtx) -> VortexResult<ArrayRef> {
         Ok(partials)
     }
 
diff --git a/vortex-array/src/aggregate_fn/fns/min_max/mod.rs b/vortex-array/src/aggregate_fn/fns/min_max/mod.rs
@@ -427,7 +427,7 @@ impl AggregateFnVTable for MinMax {
         }
     }
 
-    fn finalize(&self, partials: ArrayRef) -> VortexResult<ArrayRef> {
+    fn finalize(&self, partials: ArrayRef, _ctx: &mut ExecutionCtx) -> VortexResult<ArrayRef> {
         Ok(partials)
     }
 
diff --git a/vortex-array/src/aggregate_fn/fns/nan_count/mod.rs b/vortex-array/src/aggregate_fn/fns/nan_count/mod.rs
@@ -172,7 +172,7 @@ impl AggregateFnVTable for NanCount {
         }
     }
 
-    fn finalize(&self, partials: ArrayRef) -> VortexResult<ArrayRef> {
+    fn finalize(&self, partials: ArrayRef, _ctx: &mut ExecutionCtx) -> VortexResult<ArrayRef> {
         Ok(partials)
     }
 
diff --git a/vortex-array/src/aggregate_fn/fns/null_count/mod.rs b/vortex-array/src/aggregate_fn/fns/null_count/mod.rs
@@ -143,7 +143,7 @@ impl AggregateFnVTable for NullCount {
         Ok(())
     }
 
-    fn finalize(&self, partials: ArrayRef) -> VortexResult<ArrayRef> {
+    fn finalize(&self, partials: ArrayRef, _ctx: &mut ExecutionCtx) -> VortexResult<ArrayRef> {
         Ok(partials)
     }
 
diff --git a/vortex-array/src/aggregate_fn/fns/sum/grouped.rs b/vortex-array/src/aggregate_fn/fns/sum/grouped.rs
@@ -214,8 +214,9 @@ mod tests {
             NumericalAggregateOpts::default(),
             elem_dtype.clone(),
         )?;
-        acc.accumulate_list(groups, &mut array_session().create_execution_ctx())?;
-        acc.finish()
+        let mut ctx = array_session().create_execution_ctx();
+        acc.accumulate_list(groups, &mut ctx)?;
+        acc.finish(&mut ctx)
     }
 
     /// Reference sums computed exactly like the generic slow path: per-group scalar [`sum`]
@@ -401,8 +402,9 @@ mod tests {
 
         let mut acc =
             GroupedAccumulator::try_new(Sum, NumericalAggregateOpts::include_nans(), elem_dtype)?;
-        acc.accumulate_list(&groups, &mut array_session().create_execution_ctx())?;
-        let actual = acc.finish()?;
+        let mut ctx2 = array_session().create_execution_ctx();
+        acc.accumulate_list(&groups, &mut ctx2)?;
+        let actual = acc.finish(&mut ctx2)?;
 
         let mut ctx = array_session().create_execution_ctx();
         // Group 0 contains a NaN -> NaN sum; group 1 sums normally.
diff --git a/vortex-array/src/aggregate_fn/fns/sum/mod.rs b/vortex-array/src/aggregate_fn/fns/sum/mod.rs
@@ -12,6 +12,7 @@ use vortex_error::VortexResult;
 use vortex_error::vortex_bail;
 use vortex_error::vortex_err;
 use vortex_error::vortex_panic;
+use vortex_mask::Mask;
 use vortex_session::VortexSession;
 use vortex_session::registry::CachedId;
 
@@ -29,8 +30,13 @@ use crate::aggregate_fn::AggregateFnId;
 use crate::aggregate_fn::AggregateFnVTable;
 use crate::aggregate_fn::DynAccumulator;
 use crate::aggregate_fn::NumericalAggregateOpts;
-use crate::arrays::ConstantArray;
-use crate::arrays::scalar_fn::ScalarFnFactoryExt;
+use crate::arrays::Bool;
+use crate::arrays::BoolArray;
+use crate::arrays::Struct;
+use crate::arrays::StructArray;
+use crate::arrays::bool::BoolArrayExt;
+use crate::arrays::masked::mask_validity_canonical;
+use crate::arrays::struct_::StructArrayExt;
 use crate::dtype::DType;
 use crate::dtype::DecimalDType;
 use crate::dtype::FieldName;
@@ -45,10 +51,7 @@ use crate::expr::stats::StatsProvider;
 use crate::expr::stats::StatsProviderExt;
 use crate::scalar::DecimalValue;
 use crate::scalar::Scalar;
-use crate::scalar_fn::EmptyOptions;
-use crate::scalar_fn::fns::fill_null::FillNull;
-use crate::scalar_fn::fns::get_item::GetItem;
-use crate::scalar_fn::fns::mask::Mask;
+use crate::validity::Validity;
 
 /// The monoid sum of an array: zero when there are no valid values, null on overflow.
 ///
@@ -415,26 +418,35 @@ impl AggregateFnVTable for Sum {
         Ok(())
     }
 
-    fn finalize(&self, partials: ArrayRef) -> VortexResult<ArrayRef> {
+    fn finalize(&self, partials: ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult<ArrayRef> {
         // Entries that saw no valid values finalize to null (SQL `SUM`), while a null `sum`
-        // field (overflow) and null partial rows (e.g. null groups) stay null via the mask's
-        // validity intersection.
-        //
-        // The expressions are built unoptimized: `optimize` costs multiples of the whole
-        // aggregation at small group counts, and the caller's execution evaluates the lazy
-        // expression as-is.
+        // field (overflow) and null partial rows (e.g. null groups) stay null via the
+        // validity intersection. Implemented structurally rather than as a `get_item`/`mask`
+        // expression: expression construction and dispatch cost multiples of the whole
+        // aggregation at small group counts.
         let len = partials.len();
-        let sum = GetItem.try_new_array(len, FieldName::from("sum"), [partials.clone()])?;
-        let seen = GetItem.try_new_array(len, FieldName::from("seen"), [partials])?;
-        let seen = FillNull.try_new_array(
-            len,
-            EmptyOptions,
-            [
-                seen,
-                ConstantArray::new(Scalar::bool(false, Nullability::NonNullable), len).into_array(),
-            ],
-        )?;
-        Mask.try_new_array(len, EmptyOptions, [sum, seen])
+        let states = match partials.as_opt::<Struct>() {
+            Some(states) => states.into_owned(),
+            None => partials.execute::<StructArray>(ctx)?,
+        };
+        let struct_mask = states.as_ref().validity()?.execute_mask(len, ctx)?;
+        let seen = states.unmasked_field_by_name("seen")?.clone();
+        let seen = match seen.as_opt::<Bool>() {
+            Some(seen) => seen.into_owned(),
+            None => seen.execute::<BoolArray>(ctx)?,
+        };
+        let valid = &struct_mask & &Mask::from_buffer(seen.to_bit_buffer());
+
+        let sum = states.unmasked_field_by_name("sum")?.clone();
+        if valid.all_true() {
+            // Every partial row is a seen, valid group: the sums are already the result.
+            return Ok(sum);
+        }
+        let sum = sum.execute::<Canonical>(ctx)?;
+        Ok(
+            mask_validity_canonical(sum, Validity::from_mask(valid, Nullability::Nullable), ctx)?
+                .into_array(),
+        )
     }
 
     fn finalize_scalar(&self, partial: &Self::Partial) -> VortexResult<Scalar> {
@@ -844,8 +856,9 @@ mod tests {
             NumericalAggregateOpts::default(),
             elem_dtype.clone(),
         )?;
-        acc.accumulate_list(groups, &mut array_session().create_execution_ctx())?;
-        acc.finish()
+        let mut ctx = array_session().create_execution_ctx();
+        acc.accumulate_list(groups, &mut ctx)?;
+        acc.finish(&mut ctx)
     }
 
     #[test]
@@ -939,15 +952,15 @@ mod tests {
             PrimitiveArray::new(buffer![1i32, 2, 3, 4], Validity::NonNullable).into_array();
         let groups1 = FixedSizeListArray::try_new(elements1, 2, Validity::NonNullable, 2)?;
         acc.accumulate_list(&groups1.into_array(), &mut ctx)?;
-        let result1 = acc.finish()?;
+        let result1 = acc.finish(&mut ctx)?;
 
         let expected1 = PrimitiveArray::from_option_iter([Some(3i64), Some(7i64)]).into_array();
         assert_arrays_eq!(&result1, &expected1, &mut ctx);
 
         let elements2 = PrimitiveArray::new(buffer![10i32, 20], Validity::NonNullable).into_array();
         let groups2 = FixedSizeListArray::try_new(elements2, 2, Validity::NonNullable, 1)?;
         acc.accumulate_list(&groups2.into_array(), &mut ctx)?;
-        let result2 = acc.finish()?;
+        let result2 = acc.finish(&mut ctx)?;
 
         let expected2 = PrimitiveArray::from_option_iter([Some(30i64)]).into_array();
         assert_arrays_eq!(&result2, &expected2, &mut ctx);
diff --git a/vortex-array/src/aggregate_fn/fns/uncompressed_size_in_bytes/mod.rs b/vortex-array/src/aggregate_fn/fns/uncompressed_size_in_bytes/mod.rs
@@ -177,7 +177,7 @@ impl AggregateFnVTable for UncompressedSizeInBytes {
         Ok(())
     }
 
-    fn finalize(&self, partials: ArrayRef) -> VortexResult<ArrayRef> {
+    fn finalize(&self, partials: ArrayRef, _ctx: &mut ExecutionCtx) -> VortexResult<ArrayRef> {
         Ok(partials)
     }
 
diff --git a/vortex-array/src/aggregate_fn/foreign.rs b/vortex-array/src/aggregate_fn/foreign.rs
@@ -108,7 +108,7 @@ impl AggregateFnVTable for ForeignAggregateFnVTable {
         vortex_bail!("Cannot execute unknown aggregate function '{}'", self.id)
     }
 
-    fn finalize(&self, _states: ArrayRef) -> VortexResult<ArrayRef> {
+    fn finalize(&self, _states: ArrayRef, _ctx: &mut ExecutionCtx) -> VortexResult<ArrayRef> {
         vortex_bail!("Cannot execute unknown aggregate function '{}'", self.id)
     }
 
diff --git a/vortex-array/src/aggregate_fn/proto.rs b/vortex-array/src/aggregate_fn/proto.rs
diff --git a/vortex-array/src/aggregate_fn/vtable.rs b/vortex-array/src/aggregate_fn/vtable.rs

Original file line number	Diff line number	Diff line change
`@@ -245,11 +245,11 @@ impl<T: BinaryCombined> AggregateFnVTable for Combined<T> {`
`245`	`245`	`unreachable!("Combined::try_accumulate handles all batches")`
`246`	`246`	`}`
`247`	`247`
`248`		`- fn finalize(&self, states: ArrayRef) -> VortexResult<ArrayRef> {`
	`248`	`+ fn finalize(&self, states: ArrayRef, ctx: &mut ExecutionCtx) -> VortexResult<ArrayRef> {`
`249`	`249`	`let l_field = states.get_item(FieldName::from(self.0.left_name()))?;`
`250`	`250`	`let r_field = states.get_item(FieldName::from(self.0.right_name()))?;`
`251`		`- let l_finalized = self.0.left().finalize(l_field)?;`
`252`		`- let r_finalized = self.0.right().finalize(r_field)?;`
	`251`	`+ let l_finalized = self.0.left().finalize(l_field, ctx)?;`
	`252`	`+ let r_finalized = self.0.right().finalize(r_field, ctx)?;`
`253`	`253`	`BinaryCombined::finalize(&self.0, l_finalized, r_finalized)`
`254`	`254`	`}`
`255`	`255`
Original file line number	Diff line number	Diff line change
`@@ -121,7 +121,7 @@ impl AggregateFnVTable for AllNan {`
`121`	`121`	`Ok(())`
`122`	`122`	`}`
`123`	`123`
`124`		`- fn finalize(&self, partials: ArrayRef) -> VortexResult<ArrayRef> {`
	`124`	`+ fn finalize(&self, partials: ArrayRef, _ctx: &mut ExecutionCtx) -> VortexResult<ArrayRef> {`
`125`	`125`	`Ok(partials)`
`126`	`126`	`}`
`127`	`127`
Original file line number	Diff line number	Diff line change
`@@ -233,7 +233,7 @@ impl AggregateFnVTable for AllNonDistinct {`
`233`	`233`	`}`
`234`	`234`	`}`
`235`	`235`
`236`		`- fn finalize(&self, _partials: ArrayRef) -> VortexResult<ArrayRef> {`
	`236`	`+ fn finalize(&self, _partials: ArrayRef, _ctx: &mut ExecutionCtx) -> VortexResult<ArrayRef> {`
`237`	`237`	`vortex_bail!("AllNonDistinct does not support array finalization");`
`238`	`238`	`}`
`239`	`239`
Original file line number	Diff line number	Diff line change
`@@ -111,7 +111,7 @@ impl AggregateFnVTable for AllNonNan {`
`111`	`111`	`Ok(())`
`112`	`112`	`}`
`113`	`113`
`114`		`- fn finalize(&self, partials: ArrayRef) -> VortexResult<ArrayRef> {`
	`114`	`+ fn finalize(&self, partials: ArrayRef, _ctx: &mut ExecutionCtx) -> VortexResult<ArrayRef> {`
`115`	`115`	`Ok(partials)`
`116`	`116`	`}`
`117`	`117`
Original file line number	Diff line number	Diff line change
`@@ -101,7 +101,7 @@ impl AggregateFnVTable for AllNonNull {`
`101`	`101`	`Ok(())`
`102`	`102`	`}`
`103`	`103`
`104`		`- fn finalize(&self, partials: ArrayRef) -> VortexResult<ArrayRef> {`
	`104`	`+ fn finalize(&self, partials: ArrayRef, _ctx: &mut ExecutionCtx) -> VortexResult<ArrayRef> {`
`105`	`105`	`Ok(partials)`
`106`	`106`	`}`
`107`	`107`
Original file line number	Diff line number	Diff line change
`@@ -104,7 +104,7 @@ impl AggregateFnVTable for AllNull {`
`104`	`104`	`Ok(())`
`105`	`105`	`}`
`106`	`106`
`107`		`- fn finalize(&self, partials: ArrayRef) -> VortexResult<ArrayRef> {`
	`107`	`+ fn finalize(&self, partials: ArrayRef, _ctx: &mut ExecutionCtx) -> VortexResult<ArrayRef> {`
`108`	`108`	`Ok(partials)`
`109`	`109`	`}`
`110`	`110`
Original file line number	Diff line number	Diff line change
`@@ -279,7 +279,7 @@ impl AggregateFnVTable for BoundedMax {`
`279`	`279`	`Ok(())`
`280`	`280`	`}`
`281`	`281`
`282`		`- fn finalize(&self, partials: ArrayRef) -> VortexResult<ArrayRef> {`
	`282`	`+ fn finalize(&self, partials: ArrayRef, _ctx: &mut ExecutionCtx) -> VortexResult<ArrayRef> {`
`283`	`283`	`partials.get_item(BOUNDED_MAX_BOUND)`
`284`	`284`	`}`
`285`	`285`
Original file line number	Diff line number	Diff line change
`@@ -197,7 +197,7 @@ impl AggregateFnVTable for BoundedMin {`
`197`	`197`	`Ok(())`
`198`	`198`	`}`
`199`	`199`
`200`		`- fn finalize(&self, partials: ArrayRef) -> VortexResult<ArrayRef> {`
	`200`	`+ fn finalize(&self, partials: ArrayRef, _ctx: &mut ExecutionCtx) -> VortexResult<ArrayRef> {`
`201`	`201`	`Ok(partials)`
`202`	`202`	`}`
`203`	`203`