emit-all path

Author: Rich-T-kid

datafusion/physical-expr-common/src/binary_map.rs

@@ -523,6 +523,18 @@ where
         // Copy only the surviving bytes into the fresh builder.
         self.buffer.append_slice(&frozen[end..]);
 
+        // Rebase non-inline offsets: buffer now starts at what was `end`.
+        // Only surviving entries (offset >= end) need rebasing; emitted entries
+        // (offset < end) are removed by drain_emitted and must not be touched.
+        if end > 0 {
+            for entry in self.map.iter_mut() {
+                if entry.len.as_usize() > SHORT_VALUE_LEN && entry.offset_or_inline >= end
+                {
+                    entry.offset_or_inline -= end;
+                }
+            }
+        }
+
         let nulls = self.null.and_then(|(_, null_idx)| {
             if null_idx >= cursor && null_idx < cursor + n {
                 Some(single_null_buffer(n, null_idx - cursor))

datafusion/physical-plan/src/aggregates/group_values/single_group_by/bytes.rs

@@ -85,13 +85,19 @@ impl<O: OffsetSizeTrait> GroupValues for GroupValuesBytes<O> {
     }
 
     fn emit(&mut self, emit_to: EmitTo) -> Result<Vec<ArrayRef>> {
-        let n = match emit_to {
-            EmitTo::All => self.num_groups,
-            EmitTo::First(n) => n.min(self.num_groups),
+        let group_values = match emit_to {
+            EmitTo::All => {
+                self.num_groups = 0;
+                self.map.take().into_state()
+            }
+            EmitTo::First(n) => {
+                let n = n.min(self.num_groups);
+                let group_values = self.map.emit(n);
+                self.map.drain_emitted(n);
+                self.num_groups -= n;
+                group_values
+            }
         };
-        let group_values = self.map.emit(n);
-        self.map.drain_emitted(n);
-        self.num_groups -= n;
         Ok(vec![group_values])
     }