feat(physical-plan): generic Rows-backed GroupColumn keeps mixed schemas on the column-wise path by zhuqi-lucas · Pull Request #23523 · apache/datafusion

zhuqi-lucas · 2026-07-13T02:41:16Z

Which issue does this PR close?

Part of EPIC: complete GroupValuesColumn type coverage (nested types + remaining primitives) #22715 (nested type coverage in GroupValuesColumn EPIC)
Alternative to feat(physical-plan): add GroupColumn support for FixedSizeList<primitive> in multi-column GROUP BY #23128 (per-type approach) — implements the direction @alamb proposed in feat(physical-plan): add GroupColumn support for FixedSizeList<primitive> in multi-column GROUP BY #23128 (comment)
Step toward the terminal goal of retiring GroupValuesRows entirely (Generic Rows-backed GroupColumn so nested GROUP BY keys stay on the column-wise fast path #23404)

Rationale for this change

Today GroupValuesColumn is all-or-nothing: a single nested column in the GROUP BY key (`Struct`, `List`, `FixedSizeList`, …) makes `supported_schema` return `false` and drops the entire aggregation onto the row-wise `GroupValuesRows` fallback — even when every other column would have qualified for the column-wise fast path. For a `GROUP BY int_col, struct_col` shape, the `int_col` pays the row-encoded storage cost for no reason.

What changes are included in this PR?

Add `RowsGroupColumn`: a generic `GroupColumn` backed by a single-field `RowConverter`, wired in as the nested-type dispatch arm of `group_column_supported_type` / `make_group_column`. Native columns keep their type-specialized builders; the nested column pays row-encoding only for its one column.

Gated to `data_type.is_nested()` so intentionally excluded scalar types (Float16, Decimal256) stay on `GroupValuesRows` and the `group_column_supported_type` ⇔ `make_group_column` invariant holds.

Impact

Memory, measured with 4000 groups of `8 × Int64 + 1 × FixedSizeList<Int64, 4>` in `mixed_schema_column_path_uses_less_memory_than_rows_fallback`:

	Bytes	vs baseline
`GroupValuesRows` (today's fallback)	1096 KB	100%
`GroupValuesColumn` + `RowsGroupColumn` fallback	594 KB	54.2%

Speed: not benchmarked as a headline result — the wins come from native columns keeping their type-specialized `equal_to`/`append_val` fast paths instead of falling back to byte-encoded row comparisons.

Are these changes tested?

Yes:

Unit tests inside `row_backed`: FSL / Struct roundtrip, `take_n`, `supports_type` matches `RowConverter::supports_fields`.
`mixed_schema_column_path_uses_less_memory_than_rows_fallback` (mod.rs): the 54.2% memory claim + identical group assignment vs `GroupValuesRows`.
`nested_float_edge_cases_match_rows_fallback`: nested `-0.0` / `NaN` produce the same groupings as `GroupValuesRows` (the correctness invariant to watch, since hashing runs on the raw column and equality runs on the row bytes).
`multi_batch_and_emit_first_matches_rows_fallback`: multi-batch streaming intern + `EmitTo::First` + `take_n`.

All 39 tests in `aggregates::group_values` pass.

Are there any user-facing changes?

No — internal aggregation representation only. Same query results, lower memory footprint on mixed-schema GROUP BY keys.

Follow-ups (out of scope)

Add coverage for any type `RowConverter` cannot encode (currently arrow-rs 59.x handles Map fine; `supports_type` delegates to `RowConverter::supports_fields` so it auto-tracks upstream).
Retire `GroupValuesRows` entirely once coverage is complete (Generic Rows-backed GroupColumn so nested GROUP BY keys stay on the column-wise fast path #23404).

…d schemas on the column-wise path Add `RowsGroupColumn`, a generic `GroupColumn` backed by a single-field `RowConverter` from arrow's row format. Route nested types (`Struct`, `List`, `LargeList`, `FixedSizeList`, and recursive combinations arrow's row converter can encode) to it via `group_column_supported_type` / `make_group_column`. Before this, a single nested column in a GROUP BY key made `supported_schema` return `false` and the whole aggregation fell back to the row-wise `GroupValuesRows`, even when every other column would have qualified for the column-wise fast path. With the generic fallback, only the nested column pays the row-encoding cost; the native columns keep their compact column-wise storage. Impact (measured by `mixed_schema_column_path_uses_less_memory_than_rows_fallback` with 4000 groups of `8 × Int64 + 1 × FixedSizeList<Int64, 4>`): - old (all-rows fallback): 1096 KB - new (column-wise + row-backed nested column): **594 KB (54.2%)** Correctness: - Hashing (`create_hashes` on the raw input columns) already supports nested types, and the caller-side `-0.0/NaN` normalization performed for `GroupValuesRows` keeps hashing aligned with the row-format equality this column implements. Covered by `nested_float_edge_cases_match_rows_fallback`. - Multi-batch streaming intern + `EmitTo::First` / `take_n` covered by `multi_batch_and_emit_first_matches_rows_fallback`. - FSL / Struct roundtrip + `supports_type` invariant covered by the unit tests inside `row_backed`. Follow-ups tracked separately: - Cover types `RowConverter` cannot encode (currently a moving target; if the arrow-rs release DataFusion pins encodes them all — including Map — the fallback is complete). - With full coverage, retire `GroupValuesRows` entirely (motivation: apache#23404). Refs: apache#23404, apache#22715.

Copilot

Pull request overview

This PR improves GroupValuesColumn mixed-schema GROUP BY handling by adding a generic row-format-backed GroupColumn for nested types, so only unsupported (nested) columns pay row-encoding overhead instead of forcing the entire key onto the GroupValuesRows fallback.

Changes:

Introduces RowsGroupColumn, a generic GroupColumn backed by Arrow’s row format for nested (row-encodable) types.
Updates the group_column_supported_type / make_group_column dispatch so nested types use RowsGroupColumn while intentionally excluded scalar types remain on GroupValuesRows.
Adds unit and integration-style tests validating correctness (including float edge cases) and reduced memory usage for mixed schemas.

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 1 comment.

File	Description
datafusion/physical-plan/src/aggregates/group_values/row.rs	Documents and exposes `dictionary_encode_if_necessary` for reuse by the new row-backed column path.
datafusion/physical-plan/src/aggregates/group_values/multi_group_by/row_backed.rs	Adds `RowsGroupColumn` implementation and unit tests for nested types via Arrow row encoding.
datafusion/physical-plan/src/aggregates/group_values/multi_group_by/mod.rs	Wires nested-type fallback into supported-type checks and factory dispatch; adds mixed-schema memory/correctness tests.

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

+    fn rows_to_array<'a>(
+        &self,
+        rows: impl IntoIterator<Item = arrow::row::Row<'a>>,
+    ) -> ArrayRef {
+        let mut arrays = self
+            .row_converter
+            .convert_rows(rows)
+            .expect("row conversion during emit");
+        debug_assert_eq!(arrays.len(), 1, "single-field row converter");
+        let array = arrays.swap_remove(0);
+        dictionary_encode_if_necessary(&array, &self.output_type)
+            .expect("dictionary re-encode during emit")
+    }


Rich-T-kid · 2026-07-13T04:00:58Z

nice! excited to take a look at this

Rich-T-kid · 2026-07-13T18:08:54Z

Speed: not benchmarked as a headline result — the wins come from native columns keeping their type-specialized equal_to/append_val fast paths instead of falling back to byte-encoded row comparisons.

I think speed is an important piece of this PR — could we at least run the existing benchmarks? IIRC the TPC-H/ClickBench/etc benchmarks don't actually include certain data types, such as dictionaries and FixedSizeList, so they may not be the best indicator of performance. But they should at least give us a ballpark sense of whether this causes a regression.

Rich-T-kid

I took a first pass through, mostly looks good. I left some comments. would be happy to take another look.

Rich-T-kid · 2026-07-13T18:14:51Z

+/// Re-apply dictionary / run-end encoding to `array` so it matches `expected`.
+///
+/// Arrow's [`RowConverter`] decodes dictionary and run-end-encoded values to
+/// their plain value type on the way out, so any group-value array produced


Arrow's [RowConverter] decodes dictionary and run-end-encoded values to
their plain value type on the way out

nit: Id say RowConverter decodes dictionarys/REE on their way in since it occurs onRowConverter::append().

Rich-T-kid · 2026-07-13T18:17:50Z

+/// being returned. Shared with the generic row-backed `GroupColumn`.
+///
+/// [`RowConverter`]: arrow::row::RowConverter
+pub(crate) fn dictionary_encode_if_necessary(


the comment above mentions casting arrays to REE but the functions is called Dictionary_encode_if_necessary.

Suggested change

pub(crate) fn dictionary_encode_if_necessary(

pub(crate) fn encode_array_if_necessary(

or something of the sort

Rich-T-kid · 2026-07-13T18:19:26Z

+/// their plain value type on the way out, so any group-value array produced
+/// from the row format must be re-encoded to the schema's expected type before
+/// being returned. Shared with the generic row-backed `GroupColumn`.


I think the cast here shows that Dictionaries and REE arrays should have their own specialized implementation of GroupColumn. For now I think its fine but unlike other implementations like fixedSizeList that can go from

arrow-array -> RowFormat -> arrow-array

dictionary/REE need an extra cast step

arrow-array -> rowFormat -> arrow-array -> (cast) -> arrow-array

Rich-T-kid · 2026-07-13T18:24:09Z

+    // encode them. Gate the fallback to nested types so intentionally-excluded
+    // scalar types (e.g. Float16, Decimal256) stay on `GroupValuesRows` and the
+    // `group_column_supported_type` ⇔ `make_group_column` invariant holds.
+    if data_type.is_nested() {
+        return RowsGroupColumn::supports_type(data_type);
+    }


why are Float16 & Decimal256 arrays not supported for this optimization?

Rich-T-kid · 2026-07-13T18:28:47Z

+        assert!(
+            column_size < rows_size,
+            "expected column-wise path ({column_size}) to use less memory than \
+             the all-rows fallback ({rows_size})"
+        );


this is a nice way to guarantee groupColumns will always consume less memory than the GroupValueRows implementation!

Rich-T-kid · 2026-07-13T18:47:39Z

+    /// Relabel a group-index vector so labels are assigned in order of first
+    /// appearance. Two vectors are equivalent groupings iff their canonical
+    /// forms are equal — this ignores the (opaque, non-semantic) difference in
+    /// group-index numbering between the vectorized column path and the
+    /// sequential rows fallback.
+    fn canonical_grouping(groups: &[usize]) -> Vec<usize> {
+        let mut map = HashMap::new();
+        let mut next = 0usize;
+        groups
+            .iter()
+            .map(|&g| {
+                *map.entry(g).or_insert_with(|| {
+                    let v = next;
+                    next += 1;
+                    v
+                })
+            })
+            .collect()
+    }


This took me a bit of time to understand why its needed. from my understanding its because GroupValues::intern() doesn't specify the order in which new IDs are given. docs

I think the comment is fine as it is but maybe adding something like

GroupValues implementations only guarantee that equal rows receive
/// equal group ids and new rows receive a fresh id — the order in which
/// new ids are handed out is not part of the contract, and can differ
/// between correct implementations

can help readers know why this function has to exist. This is mostly a nit

Rich-T-kid · 2026-07-13T19:18:59Z

+//! and only pays the row-encoding cost on `struct_col`, instead of dragging the
+//! entire key onto `GroupValuesRows`.


nit:

Suggested change

//! and only pays the row-encoding cost on `struct_col`, instead of dragging the

//! entire key onto `GroupValuesRows`.

//! and only pays the row-encoding cost on `struct_col`, instead of dragging both

//! columns onto `GroupValuesRows`.

Rich-T-kid · 2026-07-13T19:46:07Z

+        let mut arrays = self
+            .row_converter
+            .convert_rows(rows)
+            .expect("row conversion during emit");


You may be interested in these issues/ PR in arrow-rs

apache/arrow-rs#10275
apache/arrow-rs#10319

Rich-T-kid · 2026-07-13T19:48:12Z

+    }
+
+    fn take_n(&mut self, n: usize) -> ArrayRef {
+        debug_assert!(n <= self.group_values.num_rows());


does DataFusion guarantee that it wont call GroupValues::emit(emit::to(n)) where n is possibly greater than the number of groups?

Copilot AI review requested due to automatic review settings July 13, 2026 02:41

github-actions Bot added the physical-plan Changes to the physical-plan crate label Jul 13, 2026

zhuqi-lucas requested review from adriangb and alamb July 13, 2026 02:41

Copilot started reviewing on behalf of zhuqi-lucas July 13, 2026 02:41 View session

zhuqi-lucas requested a review from kosiew July 13, 2026 02:44

Copilot AI reviewed Jul 13, 2026

View reviewed changes

Rich-T-kid suggested changes Jul 13, 2026

View reviewed changes

	pub(crate) fn dictionary_encode_if_necessary(
	pub(crate) fn encode_array_if_necessary(

		//! and only pays the row-encoding cost on `struct_col`, instead of dragging the
		//! entire key onto `GroupValuesRows`.

Uh oh!

Conversation

zhuqi-lucas commented Jul 13, 2026

Which issue does this PR close?

Rationale for this change

What changes are included in this PR?

Impact

Are these changes tested?

Are there any user-facing changes?

Follow-ups (out of scope)

Uh oh!

Copilot AI left a comment

Choose a reason for hiding this comment

Pull request overview

Reviewed changes

Uh oh!

Rich-T-kid commented Jul 13, 2026

Uh oh!

Rich-T-kid commented Jul 13, 2026

Uh oh!

Rich-T-kid left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants