(datafusion) branch gh-readonly-queue/main/pr-26286-cdc2fc041b6a812a24933abc40da6475dbbf5729 created (now 5009f42e4a)

[gi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

github-merge-queue[bot] pushed a change to branch gh-readonly-queue/main/pr-26286-cdc2fc041b6a812a24933abc40da6475dbbf5729
in repository https://gitbox.apache.org/repos/asf/datafusion.git


      at 5009f42e4a Account for TopK common-prefix converter memory (#26286)

This branch includes the following new commits:

     new 5009f42e4a Account for TopK common-prefix converter memory (#26286)

The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails.  The revisions
listed as "add" were already present in the repository and have only
been added to this reference.



---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org

(datafusion) 01/01: Account for TopK common-prefix converter memory (#26286)

[gi...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-26286-cdc2fc041b6a812a24933abc40da6475dbbf5729
in repository https://gitbox.apache.org/repos/asf/datafusion.git

commit 5009f42e4a24d231580222e5628c1e203f07614b
Author: kosiew <ko...@gmail.com>
AuthorDate: Fri Oct 16 02:30:23 2026 +0000

    Account for TopK common-prefix converter memory (#26286)
    
    ## Which issue does this PR close?
    
    - Part of #23393
    
    ## Rationale for this change
    
    `TopK` underreports retained memory when a common sort prefix is present
    because its memory accounting omits the optional
    `common_sort_prefix_converter`. This can cause memory reservations to
    underestimate TopK's retained memory, weakening memory-limit
    enforcement.
    
    ## What changes are included in this PR?
    
    - Include `common_sort_prefix_converter.size()` exactly once in
    `TopK::size()` when the converter is present.
    - Preserve the existing size calculation when the converter is absent.
    - Add a regression test covering both construction paths and checking
    that the memory reservation matches the reported size after batch
    insertion.
    
    ## Are these changes tested?
    
    Added `test_topk_size_common_sort_prefix_converter`, which verifies:
    
    - A common-prefix converter is created and reports a nonzero size.
    - Without a common prefix, `TopK::size()` retains its existing formula.
    - With a common prefix, the reported size increases by the converter's
    size.
    - After `insert_batch()`, the memory reservation matches `TopK::size()`.
    
    The test does not rely on global allocator measurements.
    
    ## Are there any user-facing changes?
    
    No changes to query results or public APIs. The change corrects internal
    memory accounting for TopK with a common sort prefix, potentially
    improving memory-limit enforcement.
    
    ## LLM-generated code disclosure
    
    This PR includes LLM-generated code and comments. All LLM-generated
    content has been manually reviewed.
---
 datafusion/physical-plan/src/topk/mod.rs | 45 ++++++++++++++++++++++++++++++++
 1 file changed, 45 insertions(+)

diff --git a/datafusion/physical-plan/src/topk/mod.rs b/datafusion/physical-plan/src/topk/mod.rs
index aee902f2b7..ef275b301a 100644
--- a/datafusion/physical-plan/src/topk/mod.rs
+++ b/datafusion/physical-plan/src/topk/mod.rs
@@ -882,6 +882,10 @@ impl TopK {
     fn size(&self) -> usize {
         size_of::<Self>()
             + self.row_converter.size()
+            + self
+                .common_sort_prefix_converter
+                .as_ref()
+                .map_or(0, RowConverter::size)
             + self.scratch_rows.size()
             + self.heap.size()
     }
@@ -3306,6 +3310,47 @@ mod tests {
         )
     }
 
+    #[test]
+    fn test_topk_size_common_sort_prefix_converter() -> Result<()> {
+        let schema = make_ab_schema();
+        let sort_expr = PhysicalSortExpr {
+            expr: col("a", schema.as_ref())?,
+            options: SortOptions::default(),
+        };
+        let make_topk = |common_sort_prefix| {
+            TopK::try_new(
+                0,
+                Arc::clone(&schema),
+                common_sort_prefix,
+                LexOrdering::from([sort_expr.clone()]),
+                3,
+                2,
+                Arc::new(RuntimeEnv::default()),
+                &ExecutionPlanMetricsSet::new(),
+                make_topk_filter(),
+            )
+        };
+        let without_prefix = make_topk(vec![])?;
+        let with_prefix = make_topk(vec![sort_expr.clone()])?;
+        let converter_size = with_prefix
+            .common_sort_prefix_converter
+            .as_ref()
+            .expect("common sort prefix converter")
+            .size();
+        assert!(converter_size > 0);
+        assert!(without_prefix.common_sort_prefix_converter.is_none());
+        assert_eq!(
+            without_prefix.size(),
+            size_of::<TopK>()
+                + without_prefix.row_converter.size()
+                + without_prefix.scratch_rows.size()
+                + without_prefix.heap.size()
+        );
+        assert_eq!(with_prefix.size(), without_prefix.size() + converter_size);
+
+        Ok(())
+    }
+
     /// Reads the `output_batches` and `output_rows` metrics recorded by an emit.
     fn output_batches_and_rows(metrics: &ExecutionPlanMetricsSet) -> (usize, usize) {
         let metrics = metrics.clone_inner();


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org