Add retract_batch method for median accumulator #7664

Activity

1. mustafasrepo

   added

   enhancementNew feature or requestNew feature or request

   on Sep 27, 2023
2. NGA-TRAN commented on Apr 21, 2025

   

   NGA-TRAN

   on Apr 21, 2025

   Contributor

   More actions

   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.
3. Jefffrey commented on Dec 10, 2025

   

   Jefffrey

   on Dec 10, 2025

   Contributor

   More actions

   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
4. Jefffrey

   added

   good first issueGood for newcomersGood for newcomers

   on Dec 10, 2025
5. github-merge-queue

   added a commit that references this issue on Dec 16, 2025

   feat: Support sliding window queries for MedianAccumulator by impleme…

   ...

   This issue will close once commit 933657e is merged into the 'main' branch.

   933657e
6. 2010YOUY01

   closed this as completedin #19278on Dec 16, 2025