allow for $in to affect environment (#6649)

* allow for `$in` to affect environment

* fix for vars, overlays, env_hidden

* fmt

* carry over variables properly

* add test

* modify name, remove redundant

* fmt

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

@@ -1,4 +1,4 @@
-use nu_engine::{eval_block, CallExt};
+use nu_engine::{eval_block, redirect_env, CallExt};
 use nu_protocol::ast::Call;
 use nu_protocol::engine::{CaptureBlock, Command, EngineState, Stack};
 use nu_protocol::{
@@ -20,6 +20,11 @@ impl Command for Collect {
                 SyntaxShape::Block(Some(vec![SyntaxShape::Any])),
                 "the block to run once the stream is collected",
             )
+            .switch(
+                "keep-env",
+                "let the block affect environment variables",
+                None,
+            )
             .category(Category::Filters)
     }
 
@@ -37,26 +42,43 @@ impl Command for Collect {
         let capture_block: CaptureBlock = call.req(engine_state, stack, 0)?;
 
         let block = engine_state.get_block(capture_block.block_id).clone();
-        let mut stack = stack.captures_to_stack(&capture_block.captures);
+        let mut stack_captures = stack.captures_to_stack(&capture_block.captures);
 
         let metadata = input.metadata();
         let input: Value = input.into_value(call.head);
 
+        let mut saved_positional = None;
         if let Some(var) = block.signature.get_positional(0) {
             if let Some(var_id) = &var.var_id {
-                stack.add_var(*var_id, input.clone());
+                stack_captures.add_var(*var_id, input.clone());
+                saved_positional = Some(*var_id);
             }
         }
 
-        eval_block(
+        let result = eval_block(
             engine_state,
-            &mut stack,
+            &mut stack_captures,
             &block,
             input.into_pipeline_data(),
             call.redirect_stdout,
             call.redirect_stderr,
         )
-        .map(|x| x.set_metadata(metadata))
+        .map(|x| x.set_metadata(metadata));
+
+        if call.has_flag("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 capture_block.captures.keys() {
+                stack_captures.vars.remove(var_id);
+            }
+            if let Some(u) = saved_positional {
+                stack_captures.vars.remove(&u);
+            }
+            // add any new variables to the stack
+            stack.vars.extend(stack_captures.vars);
+        }
+        result
     }
 
     fn examples(&self) -> Vec<Example> {