Add retract_batch method for median accumulator

[mustafasrepo]

Is your feature request related to a problem or challenge?

Median accumulator implementation can be realized with a sliding algorithm. However, current implementation lacks this implementation.

Describe the solution you'd like

No response

Describe alternatives you've considered

No response

Additional context

No response

[NGA-TRAN]

DataDog is interested in this feature. Our focus is on small sliding windows of 3, 5, 7, and 9, which I believe are quite reasonable to support as they only require a small buffer to compute and return the median. Below are the steps to reproduce:

-- create a table with 3 columns: timestamp, tags, value
CREATE TABLE test_table (
    timestamp INT,
    tags VARCHAR(255),
    value DOUBLE
);

-- insert some data into the table
INSERT INTO test_table (timestamp, tags, value) VALUES
(1, 'tag1', 10),
(2, 'tag1', 20),
(3, 'tag1', 30),
(4, 'tag1', 40),
(5, 'tag1', 50),
(1, 'tag2', 60),
(2, 'tag2', 70),
(3, 'tag2', 80),
(4, 'tag2', 90),
(5, 'tag2', 100);

-- Not supported yet
SELECT
    timestamp,
    tags,
    median(value) OVER (PARTITION BY tags ORDER BY timestamp ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING) AS value_median_3
FROM test_table
ORDER BY tags, timestamp;
-- This feature is not implemented: Aggregate can not be used as a sliding accumulator because `retract_batch` is not implemented: median(test_table.value) PARTITION BY [test_table.tags] ORDER BY [test_table.timestamp ASC NULLS LAST] ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING

-- Other streamable aggregations such as max are supported
SELECT
    timestamp,
    tags,
    value,
    max(value) OVER (PARTITION BY tags ORDER BY timestamp ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING) AS value_max_3
FROM test_table
ORDER BY tags, timestamp;   
+-----------+------+-------+-------------+
| timestamp | tags | value | value_max_3 |
+-----------+------+-------+-------------+
| 1         | tag1 | 10.0  | 20.0        |
| 2         | tag1 | 20.0  | 30.0        |
| 3         | tag1 | 30.0  | 40.0        |
| 4         | tag1 | 40.0  | 50.0        |
| 5         | tag1 | 50.0  | 50.0        |
| 1         | tag2 | 60.0  | 70.0        |
| 2         | tag2 | 70.0  | 80.0        |
| 3         | tag2 | 80.0  | 90.0        |
| 4         | tag2 | 90.0  | 100.0       |
| 5         | tag2 | 100.0 | 100.0       |
+-----------+------+-------+-------------+
10 row(s) fetched.

[Jefffrey]

Some pointers.

Where to do implementation:

https://github.com/apache/datafusion/blob/2a08013af3ccf703bee202c959b40bb0d35bdea1/datafusion/functions-aggregate/src/median.rs

See the methods to implement from trait:

datafusion/datafusion/expr-common/src/accumulator.rs

Lines 298 to 313 in 2a08013

	fn retract_batch(&mut self, _values: &[ArrayRef]) -> Result<()> {
	// TODO add retract for all accumulators
	internal_err!(
	"Retract should be implemented for aggregate functions when used with custom window frame queries"
	)
	}
	/// Does the accumulator support incrementally updating its value
	/// by *removing* values.
	///
	/// If this function returns true, [`Self::retract_batch`] will be
	/// called for sliding window functions such as queries with an
	/// `OVER (ROWS BETWEEN 1 PRECEDING AND 2 FOLLOWING)`
	fn supports_retract_batch(&self) -> bool {
	false
	}

Example retract_batch from other method:

datafusion/datafusion/functions-aggregate/src/average.rs

Lines 554 to 565 in 79a2f5e

	fn retract_batch(&mut self, values: &[ArrayRef]) -> Result<()> {
	let values = values[0].as_primitive::<Float64Type>();
	self.count -= (values.len() - values.null_count()) as u64;
	if let Some(x) = sum(values) {
	self.sum = Some(self.sum.unwrap() - x);
	}
	Ok(())
	}
	fn supports_retract_batch(&self) -> bool {
	true
	}

Add tests here:

https://github.com/apache/datafusion/blob/79a2f5e110f2a3ddbe6d943f951bd6f518549b03/datafusion/sqllogictest/test_files/aggregate.slt

• See suggested tests in above comment