sunchao commented on code in PR #7191: URL: https://github.com/apache/datafusion-comet/pull/7191#discussion_r4426175822
########## native/spark-expr/src/agg_funcs/min_max_struct.rs: ########## @@ -0,0 +1,589 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use arrow::array::{new_null_array, Array, ArrayRef, BooleanArray}; +use arrow::compute::{not, nullif, prep_null_mask_filter, SortOptions}; +use arrow::datatypes::{DataType, Field, FieldRef}; +use arrow::row::{OwnedRow, RowConverter, Rows, SortField}; +use datafusion::common::{internal_err, Result, ScalarValue}; +use datafusion::logical_expr::function::{AccumulatorArgs, StateFieldsArgs}; +use datafusion::logical_expr::{ + Accumulator, AggregateUDFImpl, EmitTo, GroupsAccumulator, Signature, Volatility, +}; +use datafusion::physical_expr::expressions::format_state_name; +use std::fmt::Debug; +use std::mem::{size_of, size_of_val}; +use std::sync::Arc; + +/// Spark's `max` or `min` over a struct. +/// +/// Spark orders structs field by field, with a null field below every value. Each input batch is +/// encoded once with Arrow's row format, ascending with nulls first, whose bytes sort in exactly +/// that order. Each group keeps the bytes of its current winner, and memory is accounted by the +/// size of those buffers. Float and double fields are not supported: the row format orders `-0.0` +/// below `0.0`, while Spark treats them as equal. +#[derive(Debug, PartialEq, Eq, Hash)] +pub struct SparkStructMinMax { + signature: Signature, + is_max: bool, +} + +impl SparkStructMinMax { + pub fn new(is_max: bool) -> Self { + Self { + signature: Signature::any(1, Volatility::Immutable), + is_max, + } + } +} + +impl AggregateUDFImpl for SparkStructMinMax { + fn name(&self) -> &str { + if self.is_max { + "max" + } else { + "min" + } + } + + fn signature(&self) -> &Signature { + &self.signature + } + + fn return_type(&self, arg_types: &[DataType]) -> Result<DataType> { + Ok(arg_types[0].clone()) + } + + fn accumulator(&self, args: AccumulatorArgs) -> Result<Box<dyn Accumulator>> { + Ok(Box::new(StructMinMaxAccumulator::try_new( + args.return_field.data_type(), + self.is_max, + )?)) + } + + fn state_fields(&self, args: StateFieldsArgs) -> Result<Vec<FieldRef>> { + Ok(vec![Arc::new(Field::new( + format_state_name(args.name, self.name()), + args.return_field.data_type().clone(), + true, + ))]) + } + + fn groups_accumulator_supported(&self, _args: AccumulatorArgs) -> bool { + true + } + + fn create_groups_accumulator( + &self, + args: AccumulatorArgs, + ) -> Result<Box<dyn GroupsAccumulator>> { + Ok(Box::new(StructMinMaxGroupsAccumulator::try_new( + args.return_field.data_type(), + self.is_max, + )?)) + } +} + +/// Encodes struct values to Spark-ordered row bytes and decodes them back. +#[derive(Debug)] +struct StructRows { + converter: RowConverter, + /// The encoding of a null struct, emitted for a group that saw no value. + null_row: OwnedRow, + is_max: bool, +} + +impl StructRows { + fn try_new(data_type: &DataType, is_max: bool) -> Result<Self> { + if !matches!(data_type, DataType::Struct(_)) { + return internal_err!("Spark struct min/max expects a struct, got {data_type}"); + } + let options = SortOptions { + descending: false, + nulls_first: true, + }; + let converter = RowConverter::new(vec![SortField::new_with_options( + data_type.clone(), + options, + )])?; + let null_row = converter + .convert_columns(&[new_null_array(data_type, 1)])? + .row(0) + .owned(); + Ok(Self { + converter, + null_row, + is_max, + }) + } + + fn encode(&self, values: &ArrayRef) -> Result<Rows> { + Ok(self.converter.convert_columns(&[Arc::clone(values)])?) Review Comment: Yes, I’m happy to keep the redundant-sort work in #7323. I reran the original 200,000-row Parquet file with 1,000 groups and 4 KiB payloads on this revision. The only setting I changed was `spark.comet.expression.Max.enabled`. Both grouped plans kept the same Comet scan, two Comet sorts, and columnar shuffle. The partial and final aggregates switched between Spark and Comet as expected. After two warmups, the medians over five alternating samples were 1,015 ms with Spark aggregation and 948 ms with native aggregation globally, and 2,403 ms and 2,059 ms respectively for the grouped query. Every result matched. The timing ranges overlap, but the earlier wide-payload regression did not reproduce. Encoding just the batch winners addresses my concern here. I used Spark 4.1.3 on JDK 17 with the JVM code from `f62550fc` and the native library built by CI from test merge `1f64e58f`. Its struct accumulator matches the PR head. [CI build](https://github.com/apache/datafusion-comet/actions/runs/40945414246/job/123807007753) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
