collect: don't require a closure (#12788)

# Description

This changes the `collect` command so that it doesn't require a closure.
Still allowed, optionally.

Before:

```nushell
open foo.json | insert foo bar | collect { save -f foo.json }
```

After:

```nushell
open foo.json | insert foo bar | collect | save -f foo.json
```

The closure argument isn't really necessary, as collect values are also
supported as `PipelineData`.

# User-Facing Changes
- `collect` command changed

# Tests + Formatting
Example changed to reflect.

# After Submitting
- [ ] release notes
- [ ] we may want to deprecate the closure arg?

crates/nu-cmd-lang/src/core_commands/collect.rs

@@ -1,5 +1,5 @@
 use nu_engine::{command_prelude::*, get_eval_block, redirect_env};
-use nu_protocol::engine::Closure;
+use nu_protocol::{engine::Closure, DataSource, PipelineMetadata};
 
 #[derive(Clone)]
 pub struct Collect;
@@ -12,7 +12,7 @@ impl Command for Collect {
     fn signature(&self) -> Signature {
         Signature::build("collect")
             .input_output_types(vec![(Type::Any, Type::Any)])
-            .required(
+            .optional(
                 "closure",
                 SyntaxShape::Closure(Some(vec![SyntaxShape::Any])),
                 "The closure to run once the stream is collected.",
@@ -26,7 +26,14 @@ impl Command for Collect {
     }
 
     fn usage(&self) -> &str {
-        "Collect a stream into a value and then run a closure with the collected value as input."
+        "Collect a stream into a value."
+    }
+
+    fn extra_usage(&self) -> &str {
+        r#"If provided, run a closure with the collected value as input.
+
+The entire stream will be collected into one value in memory, so if the stream
+is particularly large, this can cause high memory usage."#
     }
 
     fn run(
@@ -36,46 +43,59 @@ impl Command for Collect {
         call: &Call,
         input: PipelineData,
     ) -> Result<PipelineData, ShellError> {
-        let closure: Closure = call.req(engine_state, stack, 0)?;
+        let closure: Option<Closure> = call.opt(engine_state, stack, 0)?;
 
-        let block = engine_state.get_block(closure.block_id);
-        let mut stack_captures =
-            stack.captures_to_stack_preserve_out_dest(closure.captures.clone());
+        let metadata = match input.metadata() {
+            // Remove the `FilePath` metadata, because after `collect` it's no longer necessary to
+            // check where some input came from.
+            Some(PipelineMetadata {
+                data_source: DataSource::FilePath(_),
+            }) => None,
+            other => other,
+        };
 
-        let metadata = input.metadata();
         let input = input.into_value(call.head)?;
+        let result;
 
-        let mut saved_positional = None;
-        if let Some(var) = block.signature.get_positional(0) {
-            if let Some(var_id) = &var.var_id {
-                stack_captures.add_var(*var_id, input.clone());
-                saved_positional = Some(*var_id);
+        if let Some(closure) = closure {
+            let block = engine_state.get_block(closure.block_id);
+            let mut stack_captures =
+                stack.captures_to_stack_preserve_out_dest(closure.captures.clone());
+
+            let mut saved_positional = None;
+            if let Some(var) = block.signature.get_positional(0) {
+                if let Some(var_id) = &var.var_id {
+                    stack_captures.add_var(*var_id, input.clone());
+                    saved_positional = Some(*var_id);
+                }
             }
+
+            let eval_block = get_eval_block(engine_state);
+
+            result = eval_block(
+                engine_state,
+                &mut stack_captures,
+                block,
+                input.into_pipeline_data_with_metadata(metadata),
+            );
+
+            if call.has_flag(engine_state, stack, "keep-env")? {
+                redirect_env(engine_state, stack, &stack_captures);
+                // for when we support `data | let x = $in;`
+                // remove the variables added earlier
+                for (var_id, _) in closure.captures {
+                    stack_captures.remove_var(var_id);
+                }
+                if let Some(u) = saved_positional {
+                    stack_captures.remove_var(u);
+                }
+                // add any new variables to the stack
+                stack.vars.extend(stack_captures.vars);
+            }
+        } else {
+            result = Ok(input.into_pipeline_data_with_metadata(metadata));
         }
 
-        let eval_block = get_eval_block(engine_state);
-
-        let result = eval_block(
-            engine_state,
-            &mut stack_captures,
-            block,
-            input.into_pipeline_data(),
-        )
-        .map(|x| x.set_metadata(metadata));
-
-        if call.has_flag(engine_state, stack, "keep-env")? {
-            redirect_env(engine_state, stack, &stack_captures);
-            // for when we support `data | let x = $in;`
-            // remove the variables added earlier
-            for (var_id, _) in closure.captures {
-                stack_captures.remove_var(var_id);
-            }
-            if let Some(u) = saved_positional {
-                stack_captures.remove_var(u);
-            }
-            // add any new variables to the stack
-            stack.vars.extend(stack_captures.vars);
-        }
         result
     }
 
@@ -88,7 +108,7 @@ impl Command for Collect {
             },
             Example {
                 description: "Read and write to the same file",
-                example: "open file.txt | collect { save -f file.txt }",
+                example: "open file.txt | collect | save -f file.txt",
                 result: None,
             },
         ]