Repository: nushell/nushell
File: crates/nu-command/src/filters/chunk_by.rs (Rust)

Code before:

    match input {
        PipelineData::Empty => Ok(PipelineData::Empty),
        PipelineData::Value(Value::Range { .. }, ..)
        | PipelineData::Value(Value::List { .. }, ..)
        | PipelineData::ListStream(..) => {
            let closure = ClosureEval::new(engine_state, stack, closure);

            let result = chunk_value_stream(input.into_iter(), closure, head);

            Ok(result.into_pipeline_data(head, engine_state.signals().clone()))
        }
        PipelineData::ByteStream(stream, ..) => {
            if let Some(chunks) = stream.chunks() {
                let closure = ClosureEval::new(engine_state, stack, closure);

                let mapped_chunks =
                    chunks.map(move |value| value.unwrap_or_else(|err| Value::error(err, head)));

                let result = chunk_value_stream(mapped_chunks, closure, head);

                Ok(result.into_pipeline_data(head, engine_state.signals().clone()))
            } else {
                Ok(PipelineData::Empty)
            }
        }

        PipelineData::Value(..) => Err(input.unsupported_input_error("list", head)),
    }
    .map(|data| data.set_metadata(metadata))
}

fn chunk_value_stream<I>(
    iterator: I,
    mut closure: ClosureEval,
    head: Span,
) -> impl Iterator<Item = Value> + 'static + Send
where
    I: Iterator<Item = Value> + 'static + Send,
{
    chunk_iter_by(iterator, move |value| {
        match closure.run_with_value(value.clone()) {
            Ok(data) => data.into_value(head).unwrap_or_else(|error| {
                Value::error(chain_error_with_input(error, value.is_error(), head), head)
            }),

            Err(error) => Value::error(chain_error_with_input(error, value.is_error(), head), head),
        }
    })
    .map(move |it| Value::list(it, head))
}

Review comment (suggestion):
I think this should return a type error. `ByteStream`s are of type `string` or `binary`, but `chunk-by` only operates on `list`s. 

Code after:
                }
                Some(value) => {
                    let value_key = (self.closure)(&value);

                    if value_key == head_key {
                        result.push(value);
                    } else {
                        self.last_value = Some((value, value_key));
                        return Some(result);
                    }
                }
            }
        }
    }
}

/// An iterator with the semantics of the chunk_by operation.
fn chunk_iter_by<I, T, F, K>(iterator: I, signals: Signals, closure: F) -> Chunk<I, T, F, K>
where
    I: Iterator<Item = T>,
    F: FnMut(&T) -> K,
    K: PartialEq,
{
    Chunk {
        closure,
        iterator,
        last_value: None,
        done: false,
        signals,
    }
}

pub fn chunk_by(
    engine_state: &EngineState,
    stack: &mut Stack,
    call: &Call,
    input: PipelineData,
) -> Result<PipelineData, ShellError> {
    let head = call.head;
    let closure: Closure = call.req(engine_state, stack, 0)?;

    let metadata = input.metadata();

    match input {
        PipelineData::Empty => Ok(PipelineData::Empty),
        PipelineData::Value(Value::Range { .. }, ..)
        | PipelineData::Value(Value::List { .. }, ..)
        | PipelineData::ListStream(..) => {
            let closure = ClosureEval::new(engine_state, stack, closure);

            let result = chunk_value_stream(