Bug: UDAF unexpectedly returns non-empty result for empty table

[niebayes]

Describe the bug

Applying a UDAF on an empty table would return a non-empty result, a single null value.

To Reproduce

I have written a simple test for reproducing the bug. For any UDAF, such as last_value and first_value, calling it on an empty table would give a result containing a non empty result.

#[tokio::test]
    async fn test_last_value() {
        use std::sync::Arc;

        use arrow_array::RecordBatch;
        use arrow_schema::{DataType, Field, Schema};
        use datafusion::assert_batches_eq;
        use datafusion::physical_plan::collect;
        use datafusion::prelude::SessionContext;

        let schema = Arc::new(Schema::new(vec![
            Field::new("id", DataType::Int32, false),
            Field::new("value", DataType::Int32, false),
        ]));
        let batch = RecordBatch::new_empty(schema);

        let ctx = SessionContext::new();
        ctx.register_batch("t", batch).unwrap();

        let plan = ctx
            .sql("select last_value(value order by id) from t")
            .await
            .unwrap()
            .logical_plan()
            .clone();
        let exec_plan = ctx.state().create_physical_plan(&plan).await.unwrap();
        let batches = collect(exec_plan, ctx.task_ctx()).await.unwrap();

        assert_eq!(batches.iter().map(|b| b.num_rows()).sum::<usize>(), 1);
        assert_batches_eq!(
            &[
                "+----------------------------------------------------+",
                "| last_value(t.value) ORDER BY [t.id ASC NULLS LAST] |",
                "+----------------------------------------------------+",
                "|                                                    |",
                "+----------------------------------------------------+",
            ],
            &batches
        );
    }

Expected behavior

Calling a UDAF on an empty table should given an empty result.

Additional context

No response

[samueleresca]

This issue is also affecting other aggregates (e.g. sum()). I think the root cause is the parsing between ScalarValue to ArrayRef happening at the finalisation of the aggregation.

The evaluate function of TrivialFirstValueAccumulator, FirstValueAccumulator, etc., produces a ScalarValue. The ScalarValue is parsed to ArrayRef in finalize_aggregation:

datafusion/datafusion/physical-plan/src/aggregates/mod.rs

Lines 1341 to 1344 in d19bf52

	accumulators
	.iter_mut()
	.map(|accumulator| accumulator.evaluate().and_then(|v| v.to_array()))
	.collect()

The .to_array() function is always setting the size of the Array to 1:

datafusion/datafusion/common/src/scalar/mod.rs

Lines 2042 to 2044 in d19bf52

	pub fn to_array(&self) -> Result<ArrayRef> {
	self.to_array_of_size(1)
	}

[Jefffrey]

I don't think this is a bug, rather it's what is expected for aggregate functions. We are consistent with other systems such as:

DuckDB

D with t as (select 1 as id, 2 as value where 1=0) select last(value order by id) from t;
┌─────────────────────────────┐
│ "last"("value" ORDER BY id) │
│            int32            │
├─────────────────────────────┤
│            NULL             │
└─────────────────────────────┘

Postgres

postgres=# with t as (select 1 as id, 2 as value where 1=0) select sum(value) from t where 1=0;
 sum
-----

(1 row)

• Using sum as they don't have last/first (only for window functions)

Spark

>>> spark.sql("with t as (select 1 as id, 2 as value where 1=0) select last(value) from t").show()
+-----------+
|last(value)|
+-----------+
|       NULL|
+-----------+

I'll close this as not planned, feel free to leave a comment if you think this should stay open 👍