allow select to stream more (#14492)

# Description

closes https://github.com/nushell/nushell/issues/14487

This PR tries to allow the `select` to stream better by changing the for
loops that collected the output into a `Vec<Value>` prior to returning
it into a map that returns the data as it is processed.

One curiosity, `select` transforms the input into a `PipelineIterator`.
If I remove this code, it still passes all tests. I'm not sure all this
`PipelineIterator` code is even needed. I left it for someone to tell me
if it's necessary.

# User-Facing Changes
<!-- List of all changes that impact the user experience here. This
helps us keep track of breaking changes. -->

# Tests + Formatting
<!--
Don't forget to add tests that cover your changes.

Make sure you've run and fixed any issues with these commands:

- `cargo fmt --all -- --check` to check standard code formatting (`cargo
fmt --all` applies these changes)
- `cargo clippy --workspace -- -D warnings -D clippy::unwrap_used` to
check that you're using the standard code style
- `cargo test --workspace` to check that all tests pass (on Windows make
sure to [enable developer
mode](https://learn.microsoft.com/en-us/windows/apps/get-started/developer-mode-features-and-debugging))
- `cargo run -- -c "use toolkit.nu; toolkit test stdlib"` to run the
tests for the standard library

> **Note**
> from `nushell` you can also use the `toolkit` as follows
> ```bash
> use toolkit.nu # or use an `env_change` hook to activate it
automatically
> toolkit check pr
> ```
-->

# After Submitting
<!-- If your PR had any user-facing changes, update [the
documentation](https://github.com/nushell/nushell.github.io) after the
PR is merged, if necessary. This will help us keep the docs up to date.
-->

crates/nu-command/src/filters/select.rs

@@ -206,7 +206,6 @@ fn select(
     let columns = new_columns;
 
     let input = if !unique_rows.is_empty() {
-        // let skip = call.has_flag(engine_state, stack, "skip")?;
         let metadata = input.metadata();
         let pipeline_iter: PipelineIterator = input.into_iter();
 
@@ -231,37 +230,31 @@ fn select(
                 Value::List {
                     vals: input_vals, ..
                 } => {
-                    let mut output = vec![];
-                    let mut columns_with_value = Vec::new();
-                    for input_val in input_vals {
-                        if !columns.is_empty() {
-                            let mut record = Record::new();
-                            for path in &columns {
-                                //FIXME: improve implementation to not clone
-                                match input_val.clone().follow_cell_path(&path.members, false) {
-                                    Ok(fetcher) => {
-                                        record.push(path.to_column_name(), fetcher);
-                                        if !columns_with_value.contains(&path) {
-                                            columns_with_value.push(path);
+                    Ok(input_vals
+                        .into_iter()
+                        .map(move |input_val| {
+                            if !columns.is_empty() {
+                                let mut record = Record::new();
+                                for path in &columns {
+                                    //FIXME: improve implementation to not clone
+                                    match input_val.clone().follow_cell_path(&path.members, false) {
+                                        Ok(fetcher) => {
+                                            record.push(path.to_column_name(), fetcher);
                                         }
-                                    }
-                                    Err(e) => {
-                                        return Err(e);
+                                        Err(e) => return Value::error(e, call_span),
                                     }
                                 }
+
+                                Value::record(record, span)
+                            } else {
+                                input_val.clone()
                             }
-
-                            output.push(Value::record(record, span))
-                        } else {
-                            output.push(input_val)
-                        }
-                    }
-
-                    Ok(output.into_iter().into_pipeline_data_with_metadata(
-                        call_span,
-                        engine_state.signals().clone(),
-                        metadata,
-                    ))
+                        })
+                        .into_pipeline_data_with_metadata(
+                            call_span,
+                            engine_state.signals().clone(),
+                            metadata,
+                        ))
                 }
                 _ => {
                     if !columns.is_empty() {
@@ -286,31 +279,29 @@ fn select(
             }
         }
         PipelineData::ListStream(stream, metadata, ..) => {
-            let mut values = vec![];
-
-            for x in stream {
-                if !columns.is_empty() {
-                    let mut record = Record::new();
-                    for path in &columns {
-                        //FIXME: improve implementation to not clone
-                        match x.clone().follow_cell_path(&path.members, false) {
-                            Ok(value) => {
-                                record.push(path.to_column_name(), value);
+            Ok(stream
+                .map(move |x| {
+                    if !columns.is_empty() {
+                        let mut record = Record::new();
+                        for path in &columns {
+                            //FIXME: improve implementation to not clone
+                            match x.clone().follow_cell_path(&path.members, false) {
+                                Ok(value) => {
+                                    record.push(path.to_column_name(), value);
+                                }
+                                Err(e) => return Value::error(e, call_span),
                             }
-                            Err(e) => return Err(e),
                         }
+                        Value::record(record, call_span)
+                    } else {
+                        x
                     }
-                    values.push(Value::record(record, call_span));
-                } else {
-                    values.push(x);
-                }
-            }
-
-            Ok(values.into_pipeline_data_with_metadata(
-                call_span,
-                engine_state.signals().clone(),
-                metadata,
-            ))
+                })
+                .into_pipeline_data_with_metadata(
+                    call_span,
+                    engine_state.signals().clone(),
+                    metadata,
+                ))
         }
         _ => Ok(PipelineData::empty()),
     }

crates/nu-command/tests/commands/select.rs

@@ -63,7 +63,7 @@ fn complex_nested_columns() {
 fn fails_if_given_unknown_column_name() {
     let actual = nu!(pipeline(
         r#"
-            echo [
+            [
                 [first_name, last_name, rusty_at, type];
 
                 [Andrés Robalino '10/11/2013' A]
@@ -71,7 +71,6 @@ fn fails_if_given_unknown_column_name() {
                 [Yehuda Katz '10/11/2013' A]
             ]
             | select rrusty_at first_name
-            | length
         "#
     ));