Function merge

datafusion/functions-aggregate/src/stddev.rs:400–454

(
        batch1: &RecordBatch,
        batch2: &RecordBatch,
        agg1: Arc<AggregateUDF>,
        agg2: Arc<AggregateUDF>,
        schema: &Schema,
    )

    }

    fn merge(
        batch1: &RecordBatch,
        batch2: &RecordBatch,
        agg1: Arc<AggregateUDF>,
        agg2: Arc<AggregateUDF>,
        schema: &Schema,
    ) -> Result<ScalarValue> {
        let expr = col("a", schema)?;
        let expr_field = expr.return_field(schema)?;

        let args1 = AccumulatorArgs {
            return_field: Field::new("f", DataType::Float64, true).into(),
            schema,
            expr_fields: &[Arc::clone(&expr_field)],
            ignore_nulls: false,
            order_bys: &[],
            name: "a",
            is_distinct: false,
            is_reversed: false,
            exprs: &[Arc::clone(&expr)],
        };

        let args2 = AccumulatorArgs {
            return_field: Field::new("f", DataType::Float64, true).into(),
            schema,
            expr_fields: &[expr_field],
            ignore_nulls: false,
            order_bys: &[],
            name: "a",
            is_distinct: false,
            is_reversed: false,
            exprs: &[expr],
        };

        let mut accum1 = agg1.accumulator(args1)?;
        let mut accum2 = agg2.accumulator(args2)?;

        let value1 = vec![
            col("a", schema)?
                .evaluate(batch1)
                .and_then(|v| v.into_array(batch1.num_rows()))?,
        ];
        let value2 = vec![
            col("a", schema)?
                .evaluate(batch2)
                .and_then(|v| v.into_array(batch2.num_rows()))?,
        ];

        accum1.update_batch(&value1)?;
        accum2.update_batch(&value2)?;
        let state2 = get_accum_scalar_values_as_arrays(accum2.as_mut())?;
        accum1.merge_batch(&state2)?;
        let result = accum1.evaluate()?;
        Ok(result)
    }
}