diff --git a/datafusion/functions-aggregate-common/src/aggregate/groups_accumulator/accumulate.rs b/datafusion/functions-aggregate-common/src/aggregate/groups_accumulator/accumulate.rs index fb84d5882fad1..36ba0fba0bd34 100644 --- a/datafusion/functions-aggregate-common/src/aggregate/groups_accumulator/accumulate.rs +++ b/datafusion/functions-aggregate-common/src/aggregate/groups_accumulator/accumulate.rs @@ -31,11 +31,8 @@ use datafusion_expr_common::groups_accumulator::{EmitTo, GroupSelection}; /// handle each input null value specially (e.g. for `SUM` to mark the /// corresponding sum as null) /// -/// If there are filters present, `NullState` tracks if it has seen -/// *any* value for that group (as some values may be filtered -/// out). Without a filter, the accumulator is only passed groups that -/// had at least one value to accumulate so they do not need to track -/// if they have seen values for a particular group. +/// `NullState` tracks if it has seen *any* value for each group when filters or +/// sparse group indices may omit input for a registered group. #[derive(Debug)] pub enum SeenValues { /// All groups seen so far have seen at least one non-null value @@ -86,6 +83,31 @@ impl SeenValues { } } +/// Returns true when all newly registered groups are present in `group_indices`. +/// +/// Group indices are assigned in first-seen order, so an unfiltered batch visits +/// new groups in ascending order. Pre-filtered input can omit a new group, making +/// the indices sparse even though the accumulator no longer receives a filter. +fn new_groups_are_dense( + group_indices: &[usize], + first_new_group: usize, + total_num_groups: usize, +) -> bool { + if first_new_group == total_num_groups { + return true; + } + + let mut next_new_group = first_new_group; + for &group_index in group_indices { + if group_index == next_new_group { + next_new_group += 1; + } else if group_index > next_new_group { + return false; + } + } + next_new_group == total_num_groups +} + /// Track the accumulator null state per row: if any values for that /// group were null and if any values have been seen at all for that group. /// @@ -104,11 +126,8 @@ impl SeenValues { /// handle each input null value specially (e.g. for `SUM` to mark the /// corresponding sum as null) /// -/// If there are filters present, `NullState` tracks if it has seen -/// *any* value for that group (as some values may be filtered -/// out). Without a filter, the accumulator is only passed groups that -/// had at least one value to accumulate so they do not need to track -/// if they have seen values for a particular group. +/// `NullState` tracks if it has seen *any* value for each group when filters or +/// sparse group indices may omit input for a registered group. /// /// [`GroupsAccumulator`]: datafusion_expr_common::groups_accumulator::GroupsAccumulator #[derive(Debug)] @@ -173,10 +192,13 @@ impl NullState { T: ArrowPrimitiveType + Send, F: FnMut(usize, T::Native) + Send, { - // skip null handling if no nulls in input or accumulator - if let SeenValues::All { num_values } = &mut self.seen_values - && opt_filter.is_none() + // Skip per-value null handling when every input value is valid and all + // newly registered groups are represented. Pre-filtered inputs can have + // sparse group indices despite not passing a filter to the accumulator. + if opt_filter.is_none() && values.null_count() == 0 + && let SeenValues::All { num_values } = &mut self.seen_values + && new_groups_are_dense(group_indices, *num_values, total_num_groups) { accumulate(group_indices, values, None, value_fn); *num_values = total_num_groups; @@ -213,10 +235,13 @@ impl NullState { let data = values.values(); assert_eq!(data.len(), group_indices.len()); - // skip null handling if no nulls in input or accumulator - if let SeenValues::All { num_values } = &mut self.seen_values - && opt_filter.is_none() + // Skip per-value null handling when every input value is valid and all + // newly registered groups are represented. Pre-filtered inputs can have + // sparse group indices despite not passing a filter to the accumulator. + if opt_filter.is_none() && values.null_count() == 0 + && let SeenValues::All { num_values } = &mut self.seen_values + && new_groups_are_dense(group_indices, *num_values, total_num_groups) { group_indices .iter() diff --git a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common.rs b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common.rs index 6aa246beedb9d..b9efcd714a958 100644 --- a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common.rs +++ b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common.rs @@ -15,10 +15,14 @@ // specific language governing permissions and limitations // under the License. +use std::borrow::Cow; use std::marker::PhantomData; use std::sync::Arc; -use arrow::array::{ArrayRef, AsArray, new_null_array}; +use arrow::array::{ + Array, ArrayRef, AsArray, BooleanArray, new_empty_array, new_null_array, +}; +use arrow::compute::{filter_record_batch, prep_null_mask_filter}; use arrow::datatypes::SchemaRef; use arrow::record_batch::RecordBatch; use datafusion_common::{Result, internal_err}; @@ -185,7 +189,7 @@ impl AggregateHashTable { .enumerate() .map(|(idx, acc)| { self.aggregate_argument_metrics - .time(idx, || acc.evaluate_acc_args(batch)) + .time(idx, || acc.evaluate_compacted_args(batch)) }) .collect::>>()?; drop(timer); @@ -217,6 +221,8 @@ impl AggregateHashTable { state .group_values .intern(group_values, &mut state.batch_group_indices)?; + // Register groups from the full input. Each filtered aggregate compacts + // this row-aligned vector independently immediately before its update. let group_indices = &state.batch_group_indices; let total_num_groups = state.group_values.len(); @@ -409,13 +415,13 @@ pub(super) type AggregateAccumulator = HashAggregateAccumulator; /// /// Arguments: /// * accumulator to update. -/// * accumulator's evaluated arguments and optional filter. +/// * accumulator's compacted arguments and optional row-aligned selection. /// * one group index per input row, mapping each row to its interned group. /// * total number of groups currently interned in that buffer, including newly /// interned groups. pub(super) type AggregateBatchFn = fn( &mut AggregateAccumulator, - &EvaluatedAccumulatorArgs, + &CompactedAccumulatorArgs, &[usize], usize, ) -> Result<()>; @@ -429,17 +435,21 @@ pub(super) type AggregateBatchFn = fn( pub(super) type MaterializeAccumulatorFn = fn(&mut AggregateAccumulator, EmitTo) -> Result>; -/// Evaluated aggregate arguments and filter for one input batch. -/// -/// For example, `AVG(x + 1) FILTER (WHERE x > 0)` evaluates both `x + 1` -/// and `x > 0`. -/// -/// These arrays can be passed directly to [`GroupsAccumulator`]. -pub(super) struct EvaluatedAccumulatorArgs { - /// Evaluated argument arrays. Some aggregate functions take multiple arguments. +/// Aggregate arguments compacted according to one aggregate's `FILTER`. +pub(super) struct CompactedAccumulatorArgs { + /// Argument arrays containing only selected rows. Some aggregate functions take + /// multiple arguments. pub(super) arguments: Vec, - /// Evaluated filter array, `Some` if the aggregate has a `FILTER` expression. - pub(super) filter: Option, + /// Original row-aligned selection used only to compact the matching group IDs. + pub(super) selection: Option, +} + +/// Evaluated aggregate arguments that preserve one output row per input row. +pub(super) struct RowAlignedAccumulatorArgs { + /// Row-aligned argument arrays. Rejected rows are represented as nulls. + pub(super) arguments: Vec, + /// Original row-aligned filter passed through to state conversion. + pub(super) filter: Option, } /// Evaluated all group by keys and accumulator args. @@ -451,8 +461,8 @@ pub(super) struct EvaluatedAggregateBatch { /// arrays for the current input batch. pub(super) grouping_set_args: Vec>, - /// Evaluated arguments and filters, one entry per aggregate expression. - pub(super) accumulator_args: Vec, + /// Compacted arguments and selections, one entry per aggregate expression. + pub(super) accumulator_args: Vec, } /// Buffer for the aggregate hash table's group keys and accumulator states. @@ -531,6 +541,42 @@ impl MaterializedAggregateOutput { } } +/// Compacts row-aligned group indices using an aggregate filter. +/// +/// Returns `None` when every row is selected so callers can reuse the input +/// slice without allocating. At high selectivity, copying contiguous selected +/// ranges avoids branching once per row. +fn compact_group_indices( + group_indices: &[usize], + filter: &BooleanArray, +) -> Option> { + debug_assert_eq!(group_indices.len(), filter.len()); + + let filter = match filter.null_count() { + 0 => Cow::Borrowed(filter), + _ => Cow::Owned(prep_null_mask_filter(filter)), + }; + let mask = filter.values(); + let selected_rows = mask.count_set_bits(); + + if selected_rows == group_indices.len() { + return None; + } + + let mut compacted = Vec::with_capacity(selected_rows); + // Match scatter's strategy: above 80% selectivity, copy contiguous ranges. + if selected_rows * 5 > group_indices.len() * 4 { + for (start, end) in mask.set_slices() { + compacted.extend_from_slice(&group_indices[start..end]); + } + } else { + compacted.extend(mask.set_indices().map(|index| group_indices[index])); + } + debug_assert_eq!(compacted.len(), selected_rows); + + Some(compacted) +} + impl HashAggregateAccumulator { pub(super) fn new( aggregate_expr: Arc, @@ -561,25 +607,62 @@ impl HashAggregateAccumulator { /// Evaluate aggregate arguments and filter for one input batch. /// /// For example, `AVG(2 / x) FILTER (WHERE x > 0)` evaluates `x > 0` - /// first, then evaluates `2 / x` only for selected rows. - /// Filtered rows will be evaluated to `NULL`, and won't trigger errors - /// such as divide by zero. + /// first, then evaluates `2 / x` against a compact batch containing only + /// selected rows. Filtered rows won't trigger errors such as divide by zero. /// - /// These arrays can be passed directly to [`GroupsAccumulator`] next. - pub(super) fn evaluate_acc_args( + /// Before updating [`GroupsAccumulator`], the retained selection is used to + /// compact the matching group IDs and is not passed through. + pub(super) fn evaluate_compacted_args( &self, batch: &RecordBatch, - ) -> Result { - let filter = self - .filter - .as_ref() - .map(|filter| { - filter - .evaluate(batch) - .and_then(|value| value.into_array(batch.num_rows())) + ) -> Result { + let selection = self.evaluate_filter(batch)?; + let selected_rows = selection.as_ref().map(|selection| selection.true_count()); + let filtered_batch = match (selection.as_ref(), selected_rows) { + (Some(selection), Some(selected_rows)) + if selected_rows > 0 && selected_rows < batch.num_rows() => + { + Some(filter_record_batch(batch, selection)?) + } + _ => None, + }; + let argument_batch = match selected_rows { + None => Some(batch), + Some(0) => None, + Some(selected_rows) if selected_rows == batch.num_rows() => Some(batch), + Some(_) => filtered_batch.as_ref(), + }; + let arguments = self + .arguments + .iter() + .map(|expr| { + if let Some(argument_batch) = argument_batch { + expr.evaluate(argument_batch) + .and_then(|value| value.into_array(argument_batch.num_rows())) + } else { + let data_type = expr.data_type(batch.schema_ref().as_ref())?; + Ok(new_empty_array(&data_type)) + } }) - .transpose()?; - let selection = filter.as_ref().map(|filter| filter.as_boolean()); + .collect::>()?; + + Ok(CompactedAccumulatorArgs { + arguments, + selection, + }) + } + + /// Evaluates selected arguments while preserving the input batch row count. + /// + /// Skip-partial conversion produces one state row per input row, so rejected + /// rows remain as null argument values and the filter is passed to + /// [`GroupsAccumulator::convert_to_state`]. + pub(super) fn evaluate_row_aligned_args( + &self, + batch: &RecordBatch, + ) -> Result { + let filter = self.evaluate_filter(batch)?; + let selection = filter.as_ref(); let arguments = self .arguments .iter() @@ -593,7 +676,19 @@ impl HashAggregateAccumulator { }) .collect::>()?; - Ok(EvaluatedAccumulatorArgs { arguments, filter }) + Ok(RowAlignedAccumulatorArgs { arguments, filter }) + } + + fn evaluate_filter(&self, batch: &RecordBatch) -> Result> { + self.filter + .as_ref() + .map(|filter| { + filter + .evaluate(batch) + .and_then(|value| value.into_array(batch.num_rows())) + .map(|filter| filter.as_boolean().clone()) + }) + .transpose() } pub(super) fn size(&self) -> usize { @@ -602,26 +697,30 @@ impl HashAggregateAccumulator { pub(super) fn update_batch( &mut self, - values: &EvaluatedAccumulatorArgs, + values: &CompactedAccumulatorArgs, group_indices: &[usize], total_num_groups: usize, ) -> Result<()> { - let filter = values.filter.as_ref().map(|filter| filter.as_boolean()); + let filtered_group_indices = values + .selection + .as_ref() + .and_then(|selection| compact_group_indices(group_indices, selection)); + let group_indices = filtered_group_indices.as_deref().unwrap_or(group_indices); self.accumulator.update_batch( &values.arguments, group_indices, - filter, + None, total_num_groups, ) } pub(super) fn merge_batch( &mut self, - values: &EvaluatedAccumulatorArgs, + values: &CompactedAccumulatorArgs, group_indices: &[usize], total_num_groups: usize, ) -> Result<()> { - debug_assert!(values.filter.is_none()); + debug_assert!(values.selection.is_none()); self.accumulator .merge_batch(&values.arguments, group_indices, total_num_groups) } @@ -647,24 +746,25 @@ impl HashAggregateAccumulator { self.accumulator.state(emit_to) } + /// Converts evaluated row-aligned arguments directly to partial state. pub(super) fn convert_to_state( - &mut self, - values: &EvaluatedAccumulatorArgs, + &self, + values: &RowAlignedAccumulatorArgs, ) -> Result> { - let opt_filter = values.filter.as_ref().map(|filter| filter.as_boolean()); self.accumulator - .convert_to_state(&values.arguments, opt_filter) + .convert_to_state(&values.arguments, values.filter.as_ref()) } pub(super) fn null_arguments( &self, input_schema: &SchemaRef, + num_rows: usize, ) -> Result> { self.arguments .iter() .map(|expr| { let data_type = expr.data_type(input_schema)?; - Ok(new_null_array(&data_type, 1)) + Ok(new_null_array(&data_type, num_rows)) }) .collect() } @@ -690,10 +790,35 @@ impl AggregateHashTableState { mod tests { use std::sync::Arc; - use arrow::array::{Array, Int32Array}; + use arrow::array::{Array, BooleanArray, Int32Array, Int64Array}; use arrow::datatypes::{DataType, Field, Schema}; + use datafusion_functions_aggregate::sum::sum_udaf; + use datafusion_physical_expr::aggregate::AggregateExprBuilder; + use datafusion_physical_expr::expressions::Column; use super::*; + use crate::metrics::ExecutionPlanMetricsSet; + + #[test] + fn compact_group_indices_uses_filter_bitmap() { + let group_indices = (0..10).collect::>(); + let all_true = BooleanArray::from(vec![true; 10]); + assert_eq!(compact_group_indices(&group_indices, &all_true), None); + + let high_selectivity = + BooleanArray::from((0..10).map(|index| index != 4).collect::>()); + assert_eq!( + compact_group_indices(&group_indices, &high_selectivity), + Some(vec![0, 1, 2, 3, 5, 6, 7, 8, 9]) + ); + + let with_nulls = + BooleanArray::from(vec![Some(true), None, Some(false), Some(true), None]); + assert_eq!( + compact_group_indices(&group_indices[..5], &with_nulls), + Some(vec![0, 3]) + ); + } #[test] fn materialized_aggregate_output_slices_batches_until_exhausted() -> Result<()> { @@ -717,6 +842,89 @@ mod tests { Ok(()) } + #[test] + fn convert_to_state_preserves_rows_and_metrics() -> Result<()> { + let schema = Arc::new(Schema::new(vec![ + Field::new("value", DataType::Int64, false), + Field::new("include", DataType::Boolean, false), + ])); + let batch = RecordBatch::try_new( + Arc::clone(&schema), + vec![ + Arc::new(Int64Array::from(vec![10, 20, 30, 40])), + Arc::new(BooleanArray::from(vec![true, false, true, false])), + ], + )?; + let accumulator = sum_accumulator(&schema, "include", 1)?; + let metrics = ExecutionPlanMetricsSet::new(); + let group_by_metrics = GroupByMetrics::new(&metrics, 0); + let argument_metrics = AggregateArgumentMetrics::new(&metrics, 0, ["SUM(value)"]); + let accumulator_metrics = AggregateAccumulatorMetrics::new( + &metrics, + 0, + ["SUM(value)"], + &[AccumulatorPhase::ConvertToState], + ); + + let values = { + let _timer = group_by_metrics.aggregate_arguments_time.timer(); + argument_metrics.time(0, || accumulator.evaluate_row_aligned_args(&batch))? + }; + let state = + accumulator_metrics.time(0, AccumulatorPhase::ConvertToState, || { + accumulator.convert_to_state(&values) + })?; + + assert_eq!( + int64_options(&state[0]), + vec![Some(10), None, Some(30), None] + ); + let metrics = metrics.clone_inner(); + for metric_name in [ + "aggregate_arguments_time", + "agg_expr_0_arguments_time", + "agg_expr_0_convert_to_state_time", + ] { + assert!( + metrics + .sum_by_name(metric_name) + .is_some_and(|time| { time.as_usize() > 0 }) + ); + } + + Ok(()) + } + + fn sum_accumulator( + schema: &SchemaRef, + filter_name: &str, + filter_index: usize, + ) -> Result { + let argument: Arc = Arc::new(Column::new("value", 0)); + let aggregate_expr = Arc::new( + AggregateExprBuilder::new(sum_udaf(), vec![Arc::clone(&argument)]) + .schema(Arc::clone(schema)) + .alias("SUM(value)") + .build()?, + ); + let accumulator = create_group_accumulator(&aggregate_expr)?; + Ok(HashAggregateAccumulator::new( + aggregate_expr, + vec![argument], + Some(Arc::new(Column::new(filter_name, filter_index))), + accumulator, + )) + } + + fn int64_options(array: &ArrayRef) -> Vec> { + array + .as_any() + .downcast_ref::() + .unwrap() + .iter() + .collect() + } + fn int32_values(batch: &RecordBatch, column: usize) -> Vec { let array = batch .column(column) diff --git a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common_ordered.rs b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common_ordered.rs index 4ced967a0977b..fb92e380d9879 100644 --- a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common_ordered.rs +++ b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/common_ordered.rs @@ -256,7 +256,7 @@ impl OrderedAggregateTable { .enumerate() .map(|(idx, acc)| { self.aggregate_argument_metrics - .time(idx, || acc.evaluate_acc_args(batch)) + .time(idx, || acc.evaluate_compacted_args(batch)) }) .collect::>>()?; drop(timer); @@ -386,6 +386,8 @@ impl OrderedAggregateTable { self.buffer .group_values .intern(group_values, &mut self.buffer.group_indices)?; + // Group values and ordering always observe the full input. Each filtered + // aggregate compacts these IDs independently immediately before update. let total_num_groups = self.buffer.group_values.len(); if total_num_groups > starting_num_groups { self.buffer.group_ordering.new_groups( diff --git a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/partial_table.rs b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/partial_table.rs index 54997f0537b87..dd7558aaf2bdd 100644 --- a/datafusion/physical-plan/src/aggregates/aggregate_hash_table/partial_table.rs +++ b/datafusion/physical-plan/src/aggregates/aggregate_hash_table/partial_table.rs @@ -19,18 +19,20 @@ use std::collections::HashMap; use std::marker::PhantomData; use std::sync::Arc; -use arrow::array::{ArrayRef, BooleanArray, new_null_array}; +use arrow::array::{ArrayRef, new_null_array}; use arrow::datatypes::SchemaRef; use arrow::record_batch::RecordBatch; use datafusion_common::{Result, assert_eq_or_internal_err}; use crate::aggregates::group_values::{AccumulatorPhase, new_group_values}; use crate::aggregates::order::GroupOrdering; -use crate::aggregates::{AggregateExec, group_id_array, max_duplicate_ordinal}; +use crate::aggregates::{ + AggregateExec, evaluate_group_by, group_id_array, max_duplicate_ordinal, +}; use super::common::{ AggregateHashTable, AggregateHashTableBuffer, AggregateHashTableState, - EvaluatedAccumulatorArgs, HashAggregateAccumulator, PartialMarker, PartialSkipMarker, + CompactedAccumulatorArgs, HashAggregateAccumulator, PartialMarker, PartialSkipMarker, }; /// Implementation specific to partial aggregation, where the table stores @@ -137,9 +139,8 @@ impl AggregateHashTable { /// GROUP BY GROUPING SETS (()); /// ``` /// - /// The synthetic row is filtered out before accumulator update so aggregates - /// see the same state they would see for an empty input, rather than a real - /// null-valued row. + /// Accumulators receive zero argument rows and zero group IDs, together with the + /// full registered group count, so they produce the same state as empty input. fn init_empty_grouping_sets(&mut self) -> Result<()> { let state = self.state.building_mut(); if !state.group_by.has_grouping_set() || !state.group_values.is_empty() { @@ -181,15 +182,14 @@ impl AggregateHashTable { if any_interned { let total_groups = state.group_values.len(); - let false_filter = BooleanArray::from(vec![false]); for (idx, acc) in state.accumulators.iter_mut().enumerate() { - let null_args = acc.null_arguments(&self.input_schema)?; - let values = EvaluatedAccumulatorArgs { + let null_args = acc.null_arguments(&self.input_schema, 0)?; + let values = CompactedAccumulatorArgs { arguments: null_args, - filter: Some(Arc::new(false_filter.clone())), + selection: None, }; accumulator_metrics.time(idx, AccumulatorPhase::Update, || { - acc.update_batch(&values, &[0], total_groups) + acc.update_batch(&values, &[], total_groups) })?; } } @@ -203,31 +203,30 @@ impl AggregateHashTable { &mut self, batch: &RecordBatch, ) -> Result { - let evaluated_batch = self.evaluate_batch(batch)?; + let state = self.state.building(); + let timer = self.group_by_metrics.time_calculating_group_ids.timer(); + let grouping_set_args = evaluate_group_by(&state.group_by, batch)?; + drop(timer); assert_eq_or_internal_err!( - evaluated_batch.grouping_set_args.len(), + grouping_set_args.len(), 1, "group_values expected to have single element" ); - let mut output = evaluated_batch - .grouping_set_args - .into_iter() - .next() - .unwrap_or_default(); + let mut output = grouping_set_args.into_iter().next().unwrap_or_default(); let accumulator_metrics = Arc::clone(&self.aggregate_accumulator_metrics); - let state = self.state.building_mut(); - for (idx, (acc, values)) in state - .accumulators - .iter_mut() - .zip(evaluated_batch.accumulator_args.iter()) - .enumerate() - { + for (idx, acc) in state.accumulators.iter().enumerate() { + let values = { + let _timer = self.group_by_metrics.aggregate_arguments_time.timer(); + self.aggregate_argument_metrics + .time(idx, || acc.evaluate_row_aligned_args(batch)) + }?; + output.extend(accumulator_metrics.time( idx, AccumulatorPhase::ConvertToState, - || acc.convert_to_state(values), + || acc.convert_to_state(&values), )?); } diff --git a/datafusion/sqllogictest/test_files/aggregate_filter_selection.slt b/datafusion/sqllogictest/test_files/aggregate_filter_selection.slt index ca583785843c2..7261899b5b351 100644 --- a/datafusion/sqllogictest/test_files/aggregate_filter_selection.slt +++ b/datafusion/sqllogictest/test_files/aggregate_filter_selection.slt @@ -44,9 +44,81 @@ ORDER BY g; 3 NULL 4 2 +# Each aggregate must compact group indices using its own FILTER. Cover both +# primitive and boolean group accumulators, including groups rejected by both. +query IIB +SELECT g, + SUM(v) FILTER (WHERE v = 2), + BOOL_AND(v > 0) FILTER (WHERE v = 5) +FROM aggregate_filter_selection +GROUP BY g +ORDER BY g; +---- +1 NULL NULL +2 2 NULL +3 NULL NULL +4 NULL true + +statement ok +SET datafusion.execution.target_partitions = 1; + +query I +COPY ( + SELECT * + FROM (VALUES + (1, 10, 0), + (1, 10, 2), + (1, 11, 5), + (2, 20, 0), + (2, 21, 4) + ) AS t(sort_col, group_col, v) + ORDER BY sort_col, group_col +) +TO 'test_files/scratch/aggregate_filter_selection/ordered.parquet' +STORED AS PARQUET; +---- +5 + +statement ok +CREATE EXTERNAL TABLE aggregate_filter_selection_ordered ( + sort_col INT, + group_col INT, + v BIGINT +) +STORED AS PARQUET +WITH ORDER (sort_col) +LOCATION 'test_files/scratch/aggregate_filter_selection'; + +# Ordered single aggregation must also apply FILTER before evaluating arguments. +query III +SELECT sort_col, group_col, SUM(10 / v) FILTER (WHERE v <> 0) +FROM aggregate_filter_selection_ordered +GROUP BY sort_col, group_col +ORDER BY sort_col, group_col; +---- +1 10 5 +1 11 2 +2 20 NULL +2 21 2 + statement ok SET datafusion.execution.target_partitions = 2; +# Repartitioning ordered input selects ordered partial aggregation. +query III +SELECT sort_col, group_col, SUM(10 / v) FILTER (WHERE v <> 0) +FROM aggregate_filter_selection_ordered +GROUP BY sort_col, group_col +ORDER BY sort_col, group_col; +---- +1 10 5 +1 11 2 +2 20 NULL +2 21 2 + +statement ok +DROP TABLE aggregate_filter_selection_ordered; + statement ok SET datafusion.execution.batch_size = 1;