Repository: pola-rs/polars
File: crates/polars-mem-engine/src/planner/lp.rs (Rust)

Code before:
                    input,
                    keys: phys_keys,
                    aggs: phys_aggs,
                    options,
                    input_schema,
                    output_schema,
                    slice: _slice,
                    apply,
                }));
            }

            // We first check if we can partition the group_by on the latest moment.
            let partitionable = partitionable_gb(&keys, &aggs, &input_schema, expr_arena, &apply);
            if partitionable {
                let from_partitioned_ds = lp_arena.iter(input).any(|(_, lp)| {
                    if let Union { options, .. } = lp {
                        options.from_partitioned_ds
                    } else {
                        false
                    }
                });
                let builder = get_streaming_executor_builder();

                let input = recurse!(input, state)?;
                let executor = Box::new(GroupByStreamingExec::new(
                    input,
                    builder,
                    root,
                    lp_arena,
                    expr_arena,
                    phys_keys,
                    phys_aggs,
                    maintain_order,
                    output_schema,
                    _slice,
                    from_partitioned_ds,
                ));

                Ok(executor)
            } else {
                let input = recurse!(input, state)?;
                Ok(Box::new(executors::GroupByExec::new(
                    input,
                    phys_keys,
                    phys_aggs,
                    apply,
                    maintain_order,
                    input_schema,
                    output_schema,
                    options.slice,
                )))

Review comment (suggestion):
Could we perform this clone inside of `GroupByStreamingExec::new`, next to where the mutation takes place? That would make it clearer why we're performing the clone.


Code after:
                    input,
                    keys: phys_keys,
                    aggs: phys_aggs,
                    options,
                    input_schema,
                    output_schema,
                    slice: _slice,
                    apply,
                }));
            }

            // We first check if we can partition the group_by on the latest moment.
            let partitionable = partitionable_gb(&keys, &aggs, &input_schema, expr_arena, &apply);
            if partitionable {
                let from_partitioned_ds = lp_arena.iter(input).any(|(_, lp)| {
                    if let Union { options, .. } = lp {
                        options.from_partitioned_ds
                    } else {
                        false
                    }
                });
                let builder = get_streaming_executor_builder();

                let input = recurse!(input, state)?;

                let gb_root = if state.has_cache_parent {
                    lp_arena.add(lp_arena.get(root).clone())
                } else {
                    root
                };

                let executor = Box::new(GroupByStreamingExec::new(
                    input,
                    builder,
                    gb_root,
                    lp_arena,
                    expr_arena,
                    phys_keys,
                    phys_aggs,
                    maintain_order,
                    output_schema,
                    _slice,
                    from_partitioned_ds,
                ));

                Ok(executor)
            } else {
                let input = recurse!(input, state)?;
                Ok(Box::new(executors::GroupByExec::new(
                    input,
                    phys_keys,