kosiew commented on code in PR #27136:
URL: https://github.com/apache/datafusion/pull/27136#discussion_r4491395286
##########
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:
Optional: could this survivor-spread regression include a unique,
non-ordering payload column and assert the exact retained payloads after
compaction? The two `(p, 0)` survivors per partition are indistinguishable, so
duplicating one while losing the other could still satisfy the current
assertions. This would verify row identity without requiring SQL ordering among
ties.
--
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]
