You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "rluvaton (via GitHub)" <gi...@apache.org> on 2025/01/02 20:12:21 UTC
[PR] add support for lists in min [datafusion]
rluvaton opened a new pull request, #13991:
URL: https://github.com/apache/datafusion/pull/13991
## Which issue does this PR close?
Closes #13987.
## Rationale for this change
You are now able to run min and max on lists
## What changes are included in this PR?
Added `GenericMinMaxAccumulator` and `GenericSlidingMinMaxAccumulator` which calculate the min and max based on the `arrow-row` crate
## Are these changes tested?
Yes manually but I need to add automated tests
## Are there any user-facing changes?
Yes, users are now allowed to sort lists and every type that is supported by the `arrow-row`
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(datafusion-functions-aggregate): add support for lists and other nested types in `min` and `max` [datafusion]
Posted by "rluvaton (via GitHub)" <gi...@apache.org>.
rluvaton commented on PR #13991:
URL: https://github.com/apache/datafusion/pull/13991#issuecomment-2828299361
Sorry, I could not find time today, you can take this but please add me as co-author for my changes:
[Creating a commit with multiple authors](https://docs.github.com/en/pull-requests/committing-changes-to-your-project/creating-and-editing-commits/creating-a-commit-with-multiple-authors)
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(datafusion-functions-aggregate): add support for lists and other nested types in `min` and `max` [datafusion]
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on code in PR #13991:
URL: https://github.com/apache/datafusion/pull/13991#discussion_r1909655204
##########
datafusion/functions-aggregate/src/min_max/min_max_generic.rs:
##########
@@ -0,0 +1,491 @@
+// 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 crate::min_max::{MovingMax, MovingMin};
+use arrow::array::ArrayRef;
+use arrow::datatypes::DataType;
+use arrow::row::{OwnedRow, Row, RowConverter, SortField};
+use arrow_schema::SortOptions;
+use datafusion_common::Result;
+use datafusion_common::ScalarValue;
+use datafusion_expr::Accumulator;
+use std::fmt::Debug;
+use std::marker::PhantomData;
+use std::mem::size_of_val;
+use std::ops::Deref;
+
+pub(crate) type GenericMinAccumulator = GenericMinMaxAccumulator<MinAccumulatorHelper>;
+pub(crate) type GenericMaxAccumulator = GenericMinMaxAccumulator<MaxAccumulatorHelper>;
+
+pub(crate) type GenericSlidingMinAccumulator =
+ GenericSlidingMinMaxAccumulator<GenericMovingMin>;
+pub(crate) type GenericSlidingMaxAccumulator =
+ GenericSlidingMinMaxAccumulator<GenericMovingMax>;
+
+fn convert_row_to_scalar(
+ row_converter: &RowConverter,
+ owned_row: &OwnedRow,
+) -> Result<ScalarValue> {
+ // Convert the row back to array so we can return it
+ let converted = row_converter.convert_rows(vec![owned_row.row()])?;
+
+ // Get the first value from the array (as we only have one row)
+ ScalarValue::try_from_array(converted[0].deref(), 0)
+}
+
+/// Helper trait for min/max accumulators to avoid code duplication
+pub(crate) trait GenericMinMaxAccumulatorHelper: Debug + Send + Sync {
+ /// Return true if the new value should replace the current value
+ /// for minimum the new value should be less than the current value
+ /// for maximum the new value should be greater than the current value
+ fn should_replace<'a>(current: &Row<'a>, possibly_new: &Row<'a>) -> bool;
+
+ /// Get the minimum/maximum value from an iterator
+ fn get_value_from_iter<Item: Ord + Sized, I: Iterator<Item = Item>>(
+ iter: I,
+ ) -> Option<Item>;
+}
+
+#[derive(Debug)]
+pub(crate) struct MinAccumulatorHelper;
+
+impl GenericMinMaxAccumulatorHelper for MinAccumulatorHelper {
+ /// Should replace the current value if the new value is less than the current value
+ #[inline]
+ fn should_replace<'a>(current: &Row<'a>, possibly_new: &Row<'a>) -> bool {
+ current > possibly_new
+ }
+
+ #[inline]
+ fn get_value_from_iter<Item: Ord + Sized, I: Iterator<Item = Item>>(
+ iter: I,
+ ) -> Option<Item> {
+ iter.min()
+ }
+}
+
+#[derive(Debug)]
+pub(crate) struct MaxAccumulatorHelper;
+
+impl GenericMinMaxAccumulatorHelper for MaxAccumulatorHelper {
+ /// Should replace the current value if the new value is greater than the current value
+ #[inline]
+ fn should_replace<'a>(current: &Row<'a>, possibly_new: &Row<'a>) -> bool {
+ current < possibly_new
+ }
+
+ #[inline]
+ fn get_value_from_iter<Item: Ord + Sized, I: Iterator<Item = Item>>(
+ iter: I,
+ ) -> Option<Item> {
+ iter.max()
+ }
+}
+
+/// Accumulator for min/max of lists
+/// this accumulator is using arrow-row as the internal representation and a way for comparing
+#[derive(Debug)]
+pub(crate) struct GenericMinMaxAccumulator<Helper>
+where
+ Helper: GenericMinMaxAccumulatorHelper,
+{
+ /// Convert the columns to row so we can compare them
+ row_converter: RowConverter,
+
+ /// The current minimum/maximum value
+ current_value: Option<OwnedRow>,
+
+ /// Null row to use for fast filtering
+ null_row: OwnedRow,
+
+ phantom_data: PhantomData<Helper>,
+}
+
+impl<Helper> GenericMinMaxAccumulator<Helper>
+where
+ Helper: GenericMinMaxAccumulatorHelper,
+{
+ pub fn try_new(datatype: &DataType) -> Result<Self> {
+ let converter = RowConverter::new(vec![
+ (SortField::new_with_options(
+ datatype.clone(),
+ SortOptions {
+ descending: false,
+ nulls_first: true,
+ },
+ )),
+ ])?;
+
+ // Create a null row to use for filtering out nulls from the input
+ let null_row = {
+ let null_array = ScalarValue::try_from(datatype)?.to_array_of_size(1)?;
+
+ let rows = converter.convert_columns(&[null_array])?;
+
+ rows.row(0).owned()
+ };
+
+ Ok(Self {
+ row_converter: converter,
+ null_row,
+ current_value: None,
+ phantom_data: PhantomData,
+ })
+ }
+}
+
+impl<Helper> Accumulator for GenericMinMaxAccumulator<Helper>
+where
+ Helper: GenericMinMaxAccumulatorHelper,
+{
+ fn state(&mut self) -> Result<Vec<ScalarValue>> {
+ Ok(vec![self.evaluate()?])
+ }
+
+ fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
+ let rows = self.row_converter.convert_columns(values)?;
+
+ let wanted_value_in_batch = Helper::get_value_from_iter(
+ rows.iter()
+ // Filter out nulls
+ .filter(|row| row != &self.null_row.row()),
+ );
+
+ match (&self.current_value, wanted_value_in_batch) {
+ // Update the minimum/maximum based on the should replace logic
+ (Some(current_val), Some(current_val_in_batch)) => {
+ if Helper::should_replace(¤t_val.row(), ¤t_val_in_batch) {
+ self.current_value = Some(current_val_in_batch.owned());
+ }
+ }
+ // First minimum/maximum for the accumulator
+ (None, Some(new_value)) => {
+ self.current_value = Some(new_value.owned());
+ }
+ // If no minimum/maximum in batch (all nulls), do nothing
+ (_, None) => {}
+ }
+
+ Ok(())
+ }
+
+ fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
+ self.update_batch(states)
+ }
+
+ fn evaluate(&mut self) -> Result<ScalarValue> {
+ // Get the current value or null if no value has been seen
+ let current_value = self.current_value.as_ref().unwrap_or(&self.null_row);
+
+ convert_row_to_scalar(&self.row_converter, current_value)
+ }
+
+ fn size(&self) -> usize {
+ size_of_val(self) - size_of_val(&self.row_converter) + self.row_converter.size()
+ }
+}
+
+/// Helper trait for sliding min/max accumulators to avoid code duplication
+pub(crate) trait GenericSlidingMinMaxAccumulatorHelper:
+ Debug + Default + Send + Sync
+{
+ fn push(&mut self, row: OwnedRow);
+
+ fn pop(&mut self);
+
+ fn value(&self) -> Option<&OwnedRow>;
+}
+
+#[derive(Debug, Default)]
+pub(crate) struct GenericMovingMin(MovingMin<OwnedRow>);
+
+impl GenericSlidingMinMaxAccumulatorHelper for GenericMovingMin {
+ fn push(&mut self, row: OwnedRow) {
+ self.0.push(row);
+ }
+
+ fn pop(&mut self) {
+ self.0.pop();
+ }
+
+ fn value(&self) -> Option<&OwnedRow> {
+ self.0.min()
+ }
+}
+
+#[derive(Debug, Default)]
+pub(crate) struct GenericMovingMax(MovingMax<OwnedRow>);
+
+impl GenericSlidingMinMaxAccumulatorHelper for GenericMovingMax {
+ fn push(&mut self, row: OwnedRow) {
+ self.0.push(row);
+ }
+
+ fn pop(&mut self) {
+ self.0.pop();
+ }
+
+ fn value(&self) -> Option<&OwnedRow> {
+ self.0.max()
+ }
+}
+
+#[derive(Debug)]
+pub(crate) struct GenericSlidingMinMaxAccumulator<
+ MovingWindowHelper: GenericSlidingMinMaxAccumulatorHelper,
+> {
+ /// Convert the columns to row so we can compare them
+ row_converter: RowConverter,
+
+ /// The current minimum value
+ current_value: Option<OwnedRow>,
+
+ moving_helper: MovingWindowHelper,
+
+ /// Null row to use for fast filtering
+ null_row: OwnedRow,
+}
+
+impl<MovingWindowHelper: GenericSlidingMinMaxAccumulatorHelper>
+ GenericSlidingMinMaxAccumulator<MovingWindowHelper>
+{
+ pub fn try_new(datatype: &DataType) -> Result<Self> {
+ let converter = RowConverter::new(vec![
+ (SortField::new_with_options(
+ datatype.clone(),
+ SortOptions {
+ descending: false,
+ nulls_first: true,
+ },
+ )),
+ ])?;
+
+ // Create a null row to use for filtering out nulls from the input
+ let null_row = {
+ let null_array = ScalarValue::try_from(datatype)?.to_array_of_size(1)?;
+
+ let rows = converter.convert_columns(&[null_array])?;
+
+ rows.row(0).owned()
+ };
+
+ Ok(Self {
+ row_converter: converter,
+ null_row,
+ current_value: None,
+ moving_helper: MovingWindowHelper::default(),
+ })
+ }
+}
+
+impl<MovingWindowHelper: GenericSlidingMinMaxAccumulatorHelper> Accumulator
+ for GenericSlidingMinMaxAccumulator<MovingWindowHelper>
+{
+ fn state(&mut self) -> Result<Vec<ScalarValue>> {
+ // Get the current value or null if no value has been seen
+ let min = self.current_value.as_ref().unwrap_or(&self.null_row);
+
+ Ok(vec![convert_row_to_scalar(&self.row_converter, min)?])
+ }
+
+ fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
+ // TODO - should assert getting only one column?
+ let rows = self.row_converter.convert_columns(values)?;
+
+ rows.iter()
+ // Filter out nulls
+ .filter(|row| row != &self.null_row.row())
+ .for_each(|row| self.moving_helper.push(row.owned()));
+
+ self.current_value = self.moving_helper.value().cloned();
+ Ok(())
+ }
+
+ fn retract_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
+ // TODO - should assert getting only one column?
+ let rows = self.row_converter.convert_columns(values)?;
+
+ rows.iter()
+ // Filter out nulls
+ .filter(|row| row != &self.null_row.row())
+ .for_each(|_| {
+ self.moving_helper.pop();
+ });
+
+ self.current_value = self.moving_helper.value().cloned();
+
+ Ok(())
+ }
+
+ fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
+ self.update_batch(states)
+ }
+
+ fn evaluate(&mut self) -> Result<ScalarValue> {
+ // Get the current value or null if no value has been seen
+ let current_value = self.current_value.as_ref().unwrap_or(&self.null_row);
+
+ convert_row_to_scalar(&self.row_converter, current_value)
+ }
+
+ fn supports_retract_batch(&self) -> bool {
+ true
+ }
+
+ fn size(&self) -> usize {
+ size_of_val(self) - size_of_val(&self.row_converter) + self.row_converter.size()
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use crate::min_max::{GenericMaxAccumulator, GenericMinAccumulator};
+ use arrow::array::builder::{BooleanBuilder, GenericListBuilder};
+ use arrow::array::types::Int32Type;
+ use arrow::array::{ArrayRef, ListArray};
+ use arrow_schema::{DataType, Field};
+ use datafusion_common::ScalarValue;
+ use datafusion_expr::Accumulator;
+ use std::ops::Deref;
+ use std::sync::Arc;
+
+ trait CreateArrayAndScalar {
+ fn get_data_type() -> DataType;
+
+ fn create_array(&self) -> ArrayRef;
+ fn create_scalar(&self) -> ScalarValue;
+ }
+
+ impl CreateArrayAndScalar for Vec<Option<Vec<Option<i32>>>> {
+ fn get_data_type() -> DataType {
+ DataType::List(Arc::new(Field::new_list_field(DataType::Int32, true)))
+ }
+
+ fn create_array(&self) -> ArrayRef {
+ Arc::new(ListArray::from_iter_primitive::<Int32Type, _, _>(
+ self.iter().cloned(),
+ ))
+ }
+
+ fn create_scalar(&self) -> ScalarValue {
+ let array = self.create_array();
+ ScalarValue::try_from_array(array.deref(), 0).unwrap()
+ }
+ }
+
+ impl CreateArrayAndScalar for Vec<Option<Vec<Option<bool>>>> {
+ fn get_data_type() -> DataType {
+ DataType::List(Arc::new(Field::new_list_field(DataType::Boolean, true)))
+ }
+
+ fn create_array(&self) -> ArrayRef {
+ let mut builder = GenericListBuilder::<i32, _>::new(BooleanBuilder::new());
+
+ builder.extend(self.iter().cloned());
+
+ Arc::new(builder.finish())
+ }
+
+ fn create_scalar(&self) -> ScalarValue {
+ let array = self.create_array();
+ ScalarValue::try_from_array(array.deref(), 0).unwrap()
+ }
+ }
+
+ fn run_test<T: CreateArrayAndScalar>(
+ acc: &mut dyn Accumulator,
+ values: &[T],
+ expected: &T,
+ ) -> (ScalarValue, ScalarValue) {
+ for batch in values.iter() {
+ let batch = batch.create_array();
+ acc.update_batch(&[batch]).unwrap();
+ }
+ let result = acc.evaluate().unwrap();
+ let expected_scalar = expected.create_scalar();
+
+ (result, expected_scalar)
+ }
+
+ fn test_min_max<T: CreateArrayAndScalar + Clone>(
+ values: &[T],
+ expected_min: &T,
+ expected_max: &T,
+ ) {
+ {
+ let mut acc = GenericMinAccumulator::try_new(&T::get_data_type()).unwrap();
+ let (result, expected_scalar) = run_test(&mut acc, values, expected_min);
+ assert_eq!(
+ result, expected_scalar,
+ "min failed, expected: {expected_scalar}, got: {result}"
+ );
+ }
+ {
+ let mut acc = GenericMaxAccumulator::try_new(&T::get_data_type()).unwrap();
+ let (result, expected_scalar) = run_test(&mut acc, values, expected_max);
+ assert_eq!(
+ result, expected_scalar,
+ "max failed, expected: {expected_scalar}, got: {result}"
+ );
+ }
+ }
+
+ #[test]
+ fn basic_i32_list() {
Review Comment:
```
statement count 0
create table t1 (a int[]) as values ([1,2,3]), ([1,2,1]);
query ?
select min(a) from t1;
----
[1, 2, 1]
statement count 0
drop table t1;
```
You can create test like this, it is easier to maintain than this rust test
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(datafusion-functions-aggregate): add support for lists and other nested types in `min` and `max` [datafusion]
Posted by "jayzhan211 (via GitHub)" <gi...@apache.org>.
jayzhan211 commented on code in PR #13991:
URL: https://github.com/apache/datafusion/pull/13991#discussion_r1909649089
##########
datafusion/functions-aggregate/src/min_max/min_max_generic.rs:
##########
@@ -0,0 +1,491 @@
+// 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 crate::min_max::{MovingMax, MovingMin};
+use arrow::array::ArrayRef;
+use arrow::datatypes::DataType;
+use arrow::row::{OwnedRow, Row, RowConverter, SortField};
+use arrow_schema::SortOptions;
+use datafusion_common::Result;
+use datafusion_common::ScalarValue;
+use datafusion_expr::Accumulator;
+use std::fmt::Debug;
+use std::marker::PhantomData;
+use std::mem::size_of_val;
+use std::ops::Deref;
+
+pub(crate) type GenericMinAccumulator = GenericMinMaxAccumulator<MinAccumulatorHelper>;
+pub(crate) type GenericMaxAccumulator = GenericMinMaxAccumulator<MaxAccumulatorHelper>;
+
+pub(crate) type GenericSlidingMinAccumulator =
+ GenericSlidingMinMaxAccumulator<GenericMovingMin>;
+pub(crate) type GenericSlidingMaxAccumulator =
+ GenericSlidingMinMaxAccumulator<GenericMovingMax>;
+
+fn convert_row_to_scalar(
+ row_converter: &RowConverter,
+ owned_row: &OwnedRow,
+) -> Result<ScalarValue> {
+ // Convert the row back to array so we can return it
+ let converted = row_converter.convert_rows(vec![owned_row.row()])?;
+
+ // Get the first value from the array (as we only have one row)
+ ScalarValue::try_from_array(converted[0].deref(), 0)
+}
+
+/// Helper trait for min/max accumulators to avoid code duplication
+pub(crate) trait GenericMinMaxAccumulatorHelper: Debug + Send + Sync {
+ /// Return true if the new value should replace the current value
+ /// for minimum the new value should be less than the current value
+ /// for maximum the new value should be greater than the current value
+ fn should_replace<'a>(current: &Row<'a>, possibly_new: &Row<'a>) -> bool;
+
+ /// Get the minimum/maximum value from an iterator
+ fn get_value_from_iter<Item: Ord + Sized, I: Iterator<Item = Item>>(
+ iter: I,
+ ) -> Option<Item>;
+}
+
+#[derive(Debug)]
+pub(crate) struct MinAccumulatorHelper;
+
+impl GenericMinMaxAccumulatorHelper for MinAccumulatorHelper {
+ /// Should replace the current value if the new value is less than the current value
+ #[inline]
+ fn should_replace<'a>(current: &Row<'a>, possibly_new: &Row<'a>) -> bool {
+ current > possibly_new
+ }
+
+ #[inline]
+ fn get_value_from_iter<Item: Ord + Sized, I: Iterator<Item = Item>>(
+ iter: I,
+ ) -> Option<Item> {
+ iter.min()
+ }
+}
+
+#[derive(Debug)]
+pub(crate) struct MaxAccumulatorHelper;
+
+impl GenericMinMaxAccumulatorHelper for MaxAccumulatorHelper {
+ /// Should replace the current value if the new value is greater than the current value
+ #[inline]
+ fn should_replace<'a>(current: &Row<'a>, possibly_new: &Row<'a>) -> bool {
+ current < possibly_new
+ }
+
+ #[inline]
+ fn get_value_from_iter<Item: Ord + Sized, I: Iterator<Item = Item>>(
+ iter: I,
+ ) -> Option<Item> {
+ iter.max()
+ }
+}
+
+/// Accumulator for min/max of lists
+/// this accumulator is using arrow-row as the internal representation and a way for comparing
+#[derive(Debug)]
+pub(crate) struct GenericMinMaxAccumulator<Helper>
+where
+ Helper: GenericMinMaxAccumulatorHelper,
+{
+ /// Convert the columns to row so we can compare them
+ row_converter: RowConverter,
+
+ /// The current minimum/maximum value
+ current_value: Option<OwnedRow>,
+
+ /// Null row to use for fast filtering
+ null_row: OwnedRow,
+
+ phantom_data: PhantomData<Helper>,
+}
+
+impl<Helper> GenericMinMaxAccumulator<Helper>
+where
+ Helper: GenericMinMaxAccumulatorHelper,
+{
+ pub fn try_new(datatype: &DataType) -> Result<Self> {
+ let converter = RowConverter::new(vec![
+ (SortField::new_with_options(
+ datatype.clone(),
+ SortOptions {
+ descending: false,
+ nulls_first: true,
+ },
+ )),
+ ])?;
+
+ // Create a null row to use for filtering out nulls from the input
+ let null_row = {
+ let null_array = ScalarValue::try_from(datatype)?.to_array_of_size(1)?;
Review Comment:
```suggestion
let null_array = new_null_array(datatype, 1);
```
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(datafusion-functions-aggregate): add support for lists and other nested types in `min` and `max` [datafusion]
Posted by "gabotechs (via GitHub)" <gi...@apache.org>.
gabotechs commented on PR #13991:
URL: https://github.com/apache/datafusion/pull/13991#issuecomment-2829991915
Work resumed in https://github.com/apache/datafusion/pull/15857
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(datafusion-functions-aggregate): add support for lists and other nested types in `min` and `max` [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on code in PR #13991:
URL: https://github.com/apache/datafusion/pull/13991#discussion_r1903111345
##########
datafusion/functions-aggregate/src/min_max.rs:
##########
@@ -230,7 +235,13 @@ impl AggregateUDFImpl for Max {
}
fn accumulator(&self, acc_args: AccumulatorArgs) -> Result<Box<dyn Accumulator>> {
- Ok(Box::new(MaxAccumulator::try_new(acc_args.return_type)?))
+ if acc_args.return_type.is_nested() {
+ Ok(Box::new(GenericMaxAccumulator::try_new(
+ acc_args.return_type,
+ )?))
Review Comment:
If you wanted to make this less verbose you could probably do something like
```suggestion
GenericMaxAccumulator::try_new(
acc_args.return_type,
).map(Box::new)
```
##########
datafusion/functions-aggregate/src/min_max.rs:
##########
@@ -230,7 +235,13 @@ impl AggregateUDFImpl for Max {
}
fn accumulator(&self, acc_args: AccumulatorArgs) -> Result<Box<dyn Accumulator>> {
- Ok(Box::new(MaxAccumulator::try_new(acc_args.return_type)?))
+ if acc_args.return_type.is_nested() {
+ Ok(Box::new(GenericMaxAccumulator::try_new(
+ acc_args.return_type,
+ )?))
+ } else {
+ Ok(Box::new(MaxAccumulator::try_new(acc_args.return_type)?))
Review Comment:
The pattern used in other accumulators is slightly different I think -- It would have a function on `MaxAccumulator` that explicitly declared the types it supported and then fallback to the `GenericMaxAccumulator` if not.
For example, something like
```rust
if MaxAccumulator::supports_type(&acc_args.return_type) {
Ok(Box::new(MaxAccumulator::try_new(acc_args.return_type)?))
} else {
Ok(Box::new(GenericMaxAccumulator::try_new(
acc_args.return_type,
)?))
}
```
This is a minor detail and not required, I just wanted to point it out
##########
datafusion/sqllogictest/test_files/array_query.slt:
##########
@@ -108,11 +108,15 @@ SELECT * FROM data WHERE column2 is not distinct from null;
# Aggregates
###########
-query error Internal error: Min/Max accumulator not implemented for type List
+query ?
SELECT min(column1) FROM data;
Review Comment:
Can we also add tests for empty input on min/max as well to test boundary conditions? Something like
```sql
SELECT MIN(column1) FROM data WHERE false
```
Which should return a null value?
##########
datafusion/functions-aggregate/src/min_max/min_max_generic.rs:
##########
@@ -0,0 +1,491 @@
+// 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 crate::min_max::{MovingMax, MovingMin};
+use arrow::array::ArrayRef;
+use arrow::datatypes::DataType;
+use arrow::row::{OwnedRow, Row, RowConverter, SortField};
+use arrow_schema::SortOptions;
+use datafusion_common::Result;
+use datafusion_common::ScalarValue;
+use datafusion_expr::Accumulator;
+use std::fmt::Debug;
+use std::marker::PhantomData;
+use std::mem::size_of_val;
+use std::ops::Deref;
+
+pub(crate) type GenericMinAccumulator = GenericMinMaxAccumulator<MinAccumulatorHelper>;
+pub(crate) type GenericMaxAccumulator = GenericMinMaxAccumulator<MaxAccumulatorHelper>;
+
+pub(crate) type GenericSlidingMinAccumulator =
+ GenericSlidingMinMaxAccumulator<GenericMovingMin>;
+pub(crate) type GenericSlidingMaxAccumulator =
+ GenericSlidingMinMaxAccumulator<GenericMovingMax>;
+
+fn convert_row_to_scalar(
+ row_converter: &RowConverter,
+ owned_row: &OwnedRow,
+) -> Result<ScalarValue> {
+ // Convert the row back to array so we can return it
+ let converted = row_converter.convert_rows(vec![owned_row.row()])?;
+
+ // Get the first value from the array (as we only have one row)
+ ScalarValue::try_from_array(converted[0].deref(), 0)
+}
+
+/// Helper trait for min/max accumulators to avoid code duplication
+pub(crate) trait GenericMinMaxAccumulatorHelper: Debug + Send + Sync {
+ /// Return true if the new value should replace the current value
+ /// for minimum the new value should be less than the current value
+ /// for maximum the new value should be greater than the current value
+ fn should_replace<'a>(current: &Row<'a>, possibly_new: &Row<'a>) -> bool;
+
+ /// Get the minimum/maximum value from an iterator
+ fn get_value_from_iter<Item: Ord + Sized, I: Iterator<Item = Item>>(
+ iter: I,
+ ) -> Option<Item>;
+}
+
+#[derive(Debug)]
+pub(crate) struct MinAccumulatorHelper;
+
+impl GenericMinMaxAccumulatorHelper for MinAccumulatorHelper {
+ /// Should replace the current value if the new value is less than the current value
+ #[inline]
+ fn should_replace<'a>(current: &Row<'a>, possibly_new: &Row<'a>) -> bool {
+ current > possibly_new
+ }
+
+ #[inline]
+ fn get_value_from_iter<Item: Ord + Sized, I: Iterator<Item = Item>>(
+ iter: I,
+ ) -> Option<Item> {
+ iter.min()
+ }
+}
+
+#[derive(Debug)]
+pub(crate) struct MaxAccumulatorHelper;
+
+impl GenericMinMaxAccumulatorHelper for MaxAccumulatorHelper {
+ /// Should replace the current value if the new value is greater than the current value
+ #[inline]
+ fn should_replace<'a>(current: &Row<'a>, possibly_new: &Row<'a>) -> bool {
+ current < possibly_new
+ }
+
+ #[inline]
+ fn get_value_from_iter<Item: Ord + Sized, I: Iterator<Item = Item>>(
+ iter: I,
+ ) -> Option<Item> {
+ iter.max()
+ }
+}
+
+/// Accumulator for min/max of lists
+/// this accumulator is using arrow-row as the internal representation and a way for comparing
+#[derive(Debug)]
+pub(crate) struct GenericMinMaxAccumulator<Helper>
+where
+ Helper: GenericMinMaxAccumulatorHelper,
+{
+ /// Convert the columns to row so we can compare them
+ row_converter: RowConverter,
+
+ /// The current minimum/maximum value
+ current_value: Option<OwnedRow>,
+
+ /// Null row to use for fast filtering
+ null_row: OwnedRow,
+
+ phantom_data: PhantomData<Helper>,
+}
+
+impl<Helper> GenericMinMaxAccumulator<Helper>
+where
+ Helper: GenericMinMaxAccumulatorHelper,
+{
+ pub fn try_new(datatype: &DataType) -> Result<Self> {
+ let converter = RowConverter::new(vec![
+ (SortField::new_with_options(
+ datatype.clone(),
+ SortOptions {
+ descending: false,
+ nulls_first: true,
+ },
+ )),
+ ])?;
+
+ // Create a null row to use for filtering out nulls from the input
+ let null_row = {
+ let null_array = ScalarValue::try_from(datatype)?.to_array_of_size(1)?;
+
+ let rows = converter.convert_columns(&[null_array])?;
+
+ rows.row(0).owned()
+ };
+
+ Ok(Self {
+ row_converter: converter,
+ null_row,
+ current_value: None,
+ phantom_data: PhantomData,
+ })
+ }
+}
+
+impl<Helper> Accumulator for GenericMinMaxAccumulator<Helper>
+where
+ Helper: GenericMinMaxAccumulatorHelper,
+{
+ fn state(&mut self) -> Result<Vec<ScalarValue>> {
+ Ok(vec![self.evaluate()?])
+ }
+
+ fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
+ let rows = self.row_converter.convert_columns(values)?;
+
+ let wanted_value_in_batch = Helper::get_value_from_iter(
+ rows.iter()
+ // Filter out nulls
+ .filter(|row| row != &self.null_row.row()),
+ );
+
+ match (&self.current_value, wanted_value_in_batch) {
+ // Update the minimum/maximum based on the should replace logic
+ (Some(current_val), Some(current_val_in_batch)) => {
+ if Helper::should_replace(¤t_val.row(), ¤t_val_in_batch) {
+ self.current_value = Some(current_val_in_batch.owned());
+ }
+ }
+ // First minimum/maximum for the accumulator
+ (None, Some(new_value)) => {
+ self.current_value = Some(new_value.owned());
+ }
+ // If no minimum/maximum in batch (all nulls), do nothing
+ (_, None) => {}
+ }
+
+ Ok(())
+ }
+
+ fn merge_batch(&mut self, states: &[ArrayRef]) -> Result<()> {
+ self.update_batch(states)
+ }
+
+ fn evaluate(&mut self) -> Result<ScalarValue> {
+ // Get the current value or null if no value has been seen
+ let current_value = self.current_value.as_ref().unwrap_or(&self.null_row);
+
+ convert_row_to_scalar(&self.row_converter, current_value)
+ }
+
+ fn size(&self) -> usize {
+ size_of_val(self) - size_of_val(&self.row_converter) + self.row_converter.size()
+ }
+}
+
+/// Helper trait for sliding min/max accumulators to avoid code duplication
+pub(crate) trait GenericSlidingMinMaxAccumulatorHelper:
+ Debug + Default + Send + Sync
+{
+ fn push(&mut self, row: OwnedRow);
+
+ fn pop(&mut self);
+
+ fn value(&self) -> Option<&OwnedRow>;
+}
+
+#[derive(Debug, Default)]
+pub(crate) struct GenericMovingMin(MovingMin<OwnedRow>);
+
+impl GenericSlidingMinMaxAccumulatorHelper for GenericMovingMin {
+ fn push(&mut self, row: OwnedRow) {
+ self.0.push(row);
+ }
+
+ fn pop(&mut self) {
+ self.0.pop();
+ }
+
+ fn value(&self) -> Option<&OwnedRow> {
+ self.0.min()
+ }
+}
+
+#[derive(Debug, Default)]
+pub(crate) struct GenericMovingMax(MovingMax<OwnedRow>);
+
+impl GenericSlidingMinMaxAccumulatorHelper for GenericMovingMax {
+ fn push(&mut self, row: OwnedRow) {
+ self.0.push(row);
+ }
+
+ fn pop(&mut self) {
+ self.0.pop();
+ }
+
+ fn value(&self) -> Option<&OwnedRow> {
+ self.0.max()
+ }
+}
+
+#[derive(Debug)]
+pub(crate) struct GenericSlidingMinMaxAccumulator<
+ MovingWindowHelper: GenericSlidingMinMaxAccumulatorHelper,
+> {
+ /// Convert the columns to row so we can compare them
+ row_converter: RowConverter,
+
+ /// The current minimum value
+ current_value: Option<OwnedRow>,
+
+ moving_helper: MovingWindowHelper,
+
+ /// Null row to use for fast filtering
+ null_row: OwnedRow,
+}
+
+impl<MovingWindowHelper: GenericSlidingMinMaxAccumulatorHelper>
+ GenericSlidingMinMaxAccumulator<MovingWindowHelper>
+{
+ pub fn try_new(datatype: &DataType) -> Result<Self> {
+ let converter = RowConverter::new(vec![
+ (SortField::new_with_options(
+ datatype.clone(),
+ SortOptions {
+ descending: false,
+ nulls_first: true,
+ },
+ )),
+ ])?;
+
+ // Create a null row to use for filtering out nulls from the input
+ let null_row = {
+ let null_array = ScalarValue::try_from(datatype)?.to_array_of_size(1)?;
+
+ let rows = converter.convert_columns(&[null_array])?;
+
+ rows.row(0).owned()
+ };
+
+ Ok(Self {
+ row_converter: converter,
+ null_row,
+ current_value: None,
+ moving_helper: MovingWindowHelper::default(),
+ })
+ }
+}
+
+impl<MovingWindowHelper: GenericSlidingMinMaxAccumulatorHelper> Accumulator
+ for GenericSlidingMinMaxAccumulator<MovingWindowHelper>
+{
+ fn state(&mut self) -> Result<Vec<ScalarValue>> {
+ // Get the current value or null if no value has been seen
+ let min = self.current_value.as_ref().unwrap_or(&self.null_row);
+
+ Ok(vec![convert_row_to_scalar(&self.row_converter, min)?])
+ }
+
+ fn update_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
+ // TODO - should assert getting only one column?
Review Comment:
I think it would be nice to either add the assert or remove the TODO as part of this PR
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(datafusion-functions-aggregate): add support for lists and other nested types in `min` and `max` [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #13991:
URL: https://github.com/apache/datafusion/pull/13991#issuecomment-2794832041
> Hey @alamb @jayzhan211, I see that this PR is closed, is it possible to reopen this and complete this feature? Are there any concerns apart from the tests? We can also take up this code and finish missing requirements. Issue: #15477 cc @gabotechs @LiaCastaneda
Thanks @ologlogn ! From my perspective this test simply needed some more slt tests
> Specifically I think this PR just needs some more tests and it will be good to go
What I would suggest is creating a new PR using the code in this branch as a starting point and then add more tests, get it ready,
fyi @rluvaton
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(datafusion-functions-aggregate): add support for lists and other nested types in `min` and `max` [datafusion]
Posted by "rluvaton (via GitHub)" <gi...@apache.org>.
rluvaton commented on PR #13991:
URL: https://github.com/apache/datafusion/pull/13991#issuecomment-2794835258
I did not meant to close it, sorry
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(datafusion-functions-aggregate): add support for lists and other nested types in `min` and `max` [datafusion]
Posted by "gabotechs (via GitHub)" <gi...@apache.org>.
gabotechs commented on PR #13991:
URL: https://github.com/apache/datafusion/pull/13991#issuecomment-2826530936
@rluvaton do you have an estimate of when this might be shipped? We’re currently blocked by this, so we’d be glad to handle it ourselves if that would help ease your workload.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(datafusion-functions-aggregate): add support for lists and other nested types in `min` and `max` [datafusion]
Posted by "alamb (via GitHub)" <gi...@apache.org>.
alamb commented on PR #13991:
URL: https://github.com/apache/datafusion/pull/13991#issuecomment-2597029091
Marking as draft as I think this PR is no longer waiting on feedback. Please mark it as ready for review when it is ready for another look
Specifically I think this PR just needs some more tests and it will be good to go
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(datafusion-functions-aggregate): add support for lists and other nested types in `min` and `max` [datafusion]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #13991:
URL: https://github.com/apache/datafusion/pull/13991#issuecomment-2731392221
Thank you for your contribution. Unfortunately, this pull request is stale because it has been open 60 days with no activity. Please remove the stale label or comment or this will be closed in 7 days.
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(datafusion-functions-aggregate): add support for lists and other nested types in `min` and `max` [datafusion]
Posted by "ologlogn (via GitHub)" <gi...@apache.org>.
ologlogn commented on PR #13991:
URL: https://github.com/apache/datafusion/pull/13991#issuecomment-2796943616
Thanks @rluvaton! It will be really helpful :)
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(datafusion-functions-aggregate): add support for lists and other nested types in `min` and `max` [datafusion]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #13991: feat(datafusion-functions-aggregate): add support for lists and other nested types in `min` and `max`
URL: https://github.com/apache/datafusion/pull/13991
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(datafusion-functions-aggregate): add support for lists and other nested types in `min` and `max` [datafusion]
Posted by "rluvaton (via GitHub)" <gi...@apache.org>.
rluvaton commented on PR #13991:
URL: https://github.com/apache/datafusion/pull/13991#issuecomment-2826533597
I'm really sorry, had crazy week with the baby, will work on it today
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(datafusion-functions-aggregate): add support for lists and other nested types in `min` and `max` [datafusion]
Posted by "ologlogn (via GitHub)" <gi...@apache.org>.
ologlogn commented on PR #13991:
URL: https://github.com/apache/datafusion/pull/13991#issuecomment-2789591893
Hey @alamb @jayzhan211, I see that this PR is closed, is it possible to reopen this and complete this feature?
Are there any concerns apart from the tests? We can also take up this code and finish missing requirements.
Issue: https://github.com/apache/datafusion/issues/15477
cc @gabotechs @LiaCastaneda
--
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: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org