SubhamSinghal commented on code in PR #27136:
URL: https://github.com/apache/datafusion/pull/27136#discussion_r4501526613
##########
datafusion/physical-plan/src/topk/mod.rs:
##########
@@ -8125,4 +7947,115 @@ mod tests {
);
Ok(())
}
+
+ /// One retained group must pin only its own rows, not the batches they
+ /// arrived in.
+ ///
+ /// Regression for the operator pinning every whole input batch any
+ /// retained row came from: batch `j` gives every partition `p` a pair of
+ /// rows at `|j - p|`, so partition `p` keeps improving until batch `p` and
+ /// is then fixed at the pair of zeros. Every batch therefore keeps two
live
+ /// rows forever while its others are evicted, and the store pinned all
+ /// `B` batches — `2B²` rows — to retain `2B`. Gathering alone would still
+ /// pin `~B²` admitted rows, so this needs compaction as well.
+ #[tokio::test]
+ async fn
test_partitioned_topk_dense_rank_store_bounded_when_survivors_spread()
+ -> Result<()> {
+ const B: i32 = 64;
+ let (schema, mut state) = build_partitioned_topk_dense_rank(1)?;
+ for j in 0..B {
+ let pks: Vec<i32> = (0..B).flat_map(|p| [p, p]).collect();
+ let vals: Vec<i32> = (0..B)
+ .flat_map(|p| {
+ let v = (j - p).abs();
+ [v, v]
+ })
+ .collect();
+ state.insert_batch(&pk_val_batch(&schema, pks, vals)?)?;
+ assert!(
+ state.core.store.total_rows
+ <= STORE_COMPACTION_RATIO * state.core.live_slots,
+ "batch {j}: store holds {} rows for {} retained",
+ state.core.store.total_rows,
+ state.core.live_slots
+ );
+ }
+ assert_eq!(state.core.live_slots, 2 * B as usize);
+
+ // Compaction rewrote every slot's coordinates, so the rows emit
+ // resolves are the check that it repointed them: a slot left at a
+ // pre-compaction coordinate would surface some other partition's row.
+ let expected: Vec<(i32, i32)> = (0..B).flat_map(|p| [(p, 0), (p,
0)]).collect();
Review Comment:
Addressed in 5f96ec422cf781200fb32420e09c6aaa6ed4fed8
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
