Add metadata to some filters (#11160)

<!--
if this PR closes one or more issues, you can automatically link the PR
with
them by using one of the [*linking
keywords*](https://docs.github.com/en/issues/tracking-your-work-with-issues/linking-a-pull-request-to-an-issue#linking-a-pull-request-to-an-issue-using-a-keyword),
e.g.
- this PR should close #xxxx
- fixes #xxxx

you can also mention related issues, PRs or discussions!
-->

# Description
This PR preserves metadata when running some filters. As discussed on
discord that helps when running for example `ls | reject modified`
because it keeps colouring and links.
<!--
Thank you for improving Nushell. Please, check our [contributing
guide](../CONTRIBUTING.md) and talk to the core team before making major
changes.

Description of your pull request goes here. **Provide examples and/or
screenshots** if your changes affect the user experience.
-->

# 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 std testing; testing run-tests --path
crates/nu-std"` 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/append.rs

@@ -118,8 +118,7 @@ only unwrap the outer list, and leave the variable's contents untouched."#
         Ok(input
             .into_iter()
             .chain(other.into_pipeline_data())
-            .into_pipeline_data(engine_state.ctrlc.clone())
-            .set_metadata(metadata))
+            .into_pipeline_data_with_metadata(metadata, engine_state.ctrlc.clone()))
     }
 }
 

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

@@ -145,8 +145,7 @@ fn getcol(
             Ok(input_cols
                 .into_iter()
                 .map(move |x| Value::string(x, head))
-                .into_pipeline_data(ctrlc)
-                .set_metadata(metadata))
+                .into_pipeline_data_with_metadata(metadata, ctrlc))
         }
         PipelineData::ExternalStream { .. } => Err(ShellError::OnlySupportsThisInputType {
             exp_input_type: "record or table".into(),

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

@@ -78,51 +78,56 @@ fn default(
     call: &Call,
     input: PipelineData,
 ) -> Result<PipelineData, ShellError> {
+    let metadata = input.metadata();
     let value: Value = call.req(engine_state, stack, 0)?;
     let column: Option<Spanned<String>> = call.opt(engine_state, stack, 1)?;
 
     let ctrlc = engine_state.ctrlc.clone();
 
     if let Some(column) = column {
-        input.map(
-            move |item| {
-                let span = item.span();
-                match item {
-                    Value::Record {
-                        val: mut record, ..
-                    } => {
-                        let mut found = false;
+        input
+            .map(
+                move |item| {
+                    let span = item.span();
+                    match item {
+                        Value::Record {
+                            val: mut record, ..
+                        } => {
+                            let mut found = false;
 
-                        for (col, val) in record.iter_mut() {
-                            if *col == column.item {
-                                found = true;
-                                if matches!(val, Value::Nothing { .. }) {
-                                    *val = value.clone();
+                            for (col, val) in record.iter_mut() {
+                                if *col == column.item {
+                                    found = true;
+                                    if matches!(val, Value::Nothing { .. }) {
+                                        *val = value.clone();
+                                    }
                                 }
                             }
-                        }
 
-                        if !found {
-                            record.push(column.item.clone(), value.clone());
-                        }
+                            if !found {
+                                record.push(column.item.clone(), value.clone());
+                            }
 
-                        Value::record(record, span)
+                            Value::record(record, span)
+                        }
+                        _ => item,
                     }
-                    _ => item,
-                }
-            },
-            ctrlc,
-        )
+                },
+                ctrlc,
+            )
+            .map(|x| x.set_metadata(metadata))
     } else if input.is_nothing() {
         Ok(value.into_pipeline_data())
     } else {
-        input.map(
-            move |item| match item {
-                Value::Nothing { .. } => value.clone(),
-                x => x,
-            },
-            ctrlc,
-        )
+        input
+            .map(
+                move |item| match item {
+                    Value::Nothing { .. } => value.clone(),
+                    x => x,
+                },
+                ctrlc,
+            )
+            .map(|x| x.set_metadata(metadata))
     }
 }
 

crates/nu-command/src/filters/drop/column.rs

@@ -94,6 +94,7 @@ fn drop_cols(
     // `[{a: 1}, {b: 2}] | drop column`
     // This will drop the column "a" instead of "b" even though column "b"
     // is displayed farther to the right.
+    let metadata = input.metadata();
     match input {
         PipelineData::ListStream(mut stream, ..) => {
             if let Some(mut first) = stream.next() {
@@ -106,7 +107,7 @@ fn drop_cols(
                             Err(e) => Value::error(e, head),
                         }
                     }))
-                    .into_pipeline_data(engine_state.ctrlc.clone()))
+                    .into_pipeline_data_with_metadata(metadata, engine_state.ctrlc.clone()))
             } else {
                 Ok(PipelineData::Empty)
             }
@@ -121,14 +122,14 @@ fn drop_cols(
                             drop_record_cols(val, head, &drop_cols)?
                         }
                     }
-                    Ok(Value::list(vals, span).into_pipeline_data())
+                    Ok(Value::list(vals, span).into_pipeline_data_with_metadata(metadata))
                 }
                 Value::Record {
                     val: mut record, ..
                 } => {
                     let len = record.len().saturating_sub(columns);
                     record.truncate(len);
-                    Ok(Value::record(record, span).into_pipeline_data())
+                    Ok(Value::record(record, span).into_pipeline_data_with_metadata(metadata))
                 }
                 // Propagate errors
                 Value::Error { error, .. } => Err(*error),

crates/nu-command/src/filters/drop/drop_.rs

@@ -96,8 +96,7 @@ impl Command for Drop {
 
         if rows_to_drop == 0 {
             Ok(v.into_iter()
-                .into_pipeline_data(engine_state.ctrlc.clone())
-                .set_metadata(metadata))
+                .into_pipeline_data_with_metadata(metadata, engine_state.ctrlc.clone()))
         } else {
             let k = if vlen < rows_to_drop {
                 0
@@ -106,9 +105,7 @@ impl Command for Drop {
             };
 
             let iter = v.into_iter().take(k as usize);
-            Ok(iter
-                .into_pipeline_data(engine_state.ctrlc.clone())
-                .set_metadata(metadata))
+            Ok(iter.into_pipeline_data_with_metadata(metadata, engine_state.ctrlc.clone()))
         }
     }
 }

crates/nu-command/src/filters/drop/nth.rs

@@ -104,6 +104,7 @@ impl Command for DropNth {
         call: &Call,
         input: PipelineData,
     ) -> Result<PipelineData, ShellError> {
+        let metadata = input.metadata();
         let number_or_range = extract_int_or_range(engine_state, stack, call)?;
         let mut lower_bound = None;
         let rows = match number_or_range {
@@ -165,14 +166,14 @@ impl Command for DropNth {
                 .into_iter()
                 .take(lower_bound)
                 .collect::<Vec<_>>()
-                .into_pipeline_data(engine_state.ctrlc.clone()))
+                .into_pipeline_data_with_metadata(metadata, engine_state.ctrlc.clone()))
         } else {
             Ok(DropNthIterator {
                 input: input.into_iter(),
                 rows,
                 current: 0,
             }
-            .into_pipeline_data(engine_state.ctrlc.clone()))
+            .into_pipeline_data_with_metadata(metadata, engine_state.ctrlc.clone()))
         }
     }
 }

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

@@ -85,8 +85,7 @@ impl Command for Every {
                     None
                 }
             })
-            .into_pipeline_data(engine_state.ctrlc.clone())
-            .set_metadata(metadata))
+            .into_pipeline_data_with_metadata(metadata, engine_state.ctrlc.clone()))
     }
 }
 

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

@@ -392,7 +392,7 @@ fn find_with_rest_and_highlight(
                 },
                 ctrlc,
             ),
-        PipelineData::ListStream(stream, meta) => Ok(ListStream::from_stream(
+        PipelineData::ListStream(stream, metadata) => Ok(ListStream::from_stream(
             stream
                 .map(move |mut x| {
                     let span = x.span();
@@ -421,8 +421,7 @@ fn find_with_rest_and_highlight(
                 }),
             ctrlc.clone(),
         )
-        .into_pipeline_data(ctrlc)
-        .set_metadata(meta)),
+        .into_pipeline_data_with_metadata(metadata, ctrlc)),
         PipelineData::ExternalStream { stdout: None, .. } => Ok(PipelineData::empty()),
         PipelineData::ExternalStream {
             stdout: Some(stream),
@@ -583,8 +582,7 @@ fn split_string_if_multiline(input: PipelineData, head_span: Span) -> PipelineDa
                         .collect(),
                     span,
                 )
-                .into_pipeline_data()
-                .set_metadata(input.metadata())
+                .into_pipeline_data_with_metadata(input.metadata())
             } else {
                 input
             }

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

@@ -106,9 +106,7 @@ fn first_helper(
 
     // early exit for `first 0`
     if rows_desired == 0 {
-        return Ok(Vec::<Value>::new()
-            .into_pipeline_data(ctrlc)
-            .set_metadata(metadata));
+        return Ok(Vec::<Value>::new().into_pipeline_data_with_metadata(metadata, ctrlc));
     }
 
     match input {
@@ -126,8 +124,7 @@ fn first_helper(
                         Ok(vals
                             .into_iter()
                             .take(rows_desired)
-                            .into_pipeline_data(ctrlc)
-                            .set_metadata(metadata))
+                            .into_pipeline_data_with_metadata(metadata, ctrlc))
                     }
                 }
                 Value::Binary { val, .. } => {
@@ -152,8 +149,7 @@ fn first_helper(
                         Ok(val
                             .into_range_iter(ctrlc.clone())?
                             .take(rows_desired)
-                            .into_pipeline_data(ctrlc)
-                            .set_metadata(metadata))
+                            .into_pipeline_data_with_metadata(metadata, ctrlc))
                     }
                 }
                 // Propagate errors by explicitly matching them before the final case.
@@ -176,8 +172,7 @@ fn first_helper(
             } else {
                 Ok(ls
                     .take(rows_desired)
-                    .into_pipeline_data(ctrlc)
-                    .set_metadata(metadata))
+                    .into_pipeline_data_with_metadata(metadata, ctrlc))
             }
         }
         PipelineData::ExternalStream { span, .. } => Err(ShellError::OnlySupportsThisInputType {

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

@@ -70,9 +70,7 @@ impl Command for Group {
             span: call.head,
         };
 
-        Ok(each_group_iterator
-            .into_pipeline_data(ctrlc)
-            .set_metadata(metadata))
+        Ok(each_group_iterator.into_pipeline_data_with_metadata(metadata, ctrlc))
     }
 }
 

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

@@ -73,7 +73,7 @@ impl Command for Headers {
         let (old_headers, new_headers) = extract_headers(&value, config)?;
         let new_headers = replace_headers(value, &old_headers, &new_headers)?;
 
-        Ok(new_headers.into_pipeline_data().set_metadata(metadata))
+        Ok(new_headers.into_pipeline_data_with_metadata(metadata))
     }
 }
 

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

@@ -109,6 +109,7 @@ fn insert(
     call: &Call,
     input: PipelineData,
 ) -> Result<PipelineData, ShellError> {
+    let metadata = input.metadata();
     let span = call.head;
 
     let cell_path: CellPath = call.req(engine_state, stack, 0)?;
@@ -129,47 +130,49 @@ fn insert(
         let orig_env_vars = stack.env_vars.clone();
         let orig_env_hidden = stack.env_hidden.clone();
 
-        input.map(
-            move |mut input| {
-                // with_env() is used here to ensure that each iteration uses
-                // a different set of environment variables.
-                // Hence, a 'cd' in the first loop won't affect the next loop.
-                stack.with_env(&orig_env_vars, &orig_env_hidden);
+        input
+            .map(
+                move |mut input| {
+                    // with_env() is used here to ensure that each iteration uses
+                    // a different set of environment variables.
+                    // Hence, a 'cd' in the first loop won't affect the next loop.
+                    stack.with_env(&orig_env_vars, &orig_env_hidden);
 
-                // Element argument
-                if let Some(var) = block.signature.get_positional(0) {
-                    if let Some(var_id) = &var.var_id {
-                        stack.add_var(*var_id, input.clone())
-                    }
-                }
-
-                let output = eval_block(
-                    &engine_state,
-                    &mut stack,
-                    &block,
-                    input.clone().into_pipeline_data(),
-                    redirect_stdout,
-                    redirect_stderr,
-                );
-
-                match output {
-                    Ok(pd) => {
-                        let span = pd.span().unwrap_or(span);
-                        if let Err(e) = input.insert_data_at_cell_path(
-                            &cell_path.members,
-                            pd.into_value(span),
-                            span,
-                        ) {
-                            return Value::error(e, span);
+                    // Element argument
+                    if let Some(var) = block.signature.get_positional(0) {
+                        if let Some(var_id) = &var.var_id {
+                            stack.add_var(*var_id, input.clone())
                         }
-
-                        input
                     }
-                    Err(e) => Value::error(e, span),
-                }
-            },
-            ctrlc,
-        )
+
+                    let output = eval_block(
+                        &engine_state,
+                        &mut stack,
+                        &block,
+                        input.clone().into_pipeline_data(),
+                        redirect_stdout,
+                        redirect_stderr,
+                    );
+
+                    match output {
+                        Ok(pd) => {
+                            let span = pd.span().unwrap_or(span);
+                            if let Err(e) = input.insert_data_at_cell_path(
+                                &cell_path.members,
+                                pd.into_value(span),
+                                span,
+                            ) {
+                                return Value::error(e, span);
+                            }
+
+                            input
+                        }
+                        Err(e) => Value::error(e, span),
+                    }
+                },
+                ctrlc,
+            )
+            .map(|x| x.set_metadata(metadata))
     } else {
         if let Some(PathMember::Int { val, .. }) = cell_path.members.first() {
             let mut input = input.into_iter();
@@ -187,22 +190,24 @@ fn insert(
                 .into_iter()
                 .chain(vec![replacement])
                 .chain(input)
-                .into_pipeline_data(ctrlc));
+                .into_pipeline_data_with_metadata(metadata, ctrlc));
         }
-        input.map(
-            move |mut input| {
-                let replacement = replacement.clone();
+        input
+            .map(
+                move |mut input| {
+                    let replacement = replacement.clone();
 
-                if let Err(e) =
-                    input.insert_data_at_cell_path(&cell_path.members, replacement, span)
-                {
-                    return Value::error(e, span);
-                }
+                    if let Err(e) =
+                        input.insert_data_at_cell_path(&cell_path.members, replacement, span)
+                    {
+                        return Value::error(e, span);
+                    }
 
-                input
-            },
-            ctrlc,
-        )
+                    input
+                },
+                ctrlc,
+            )
+            .map(|x| x.set_metadata(metadata))
     }
 }
 

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

@@ -68,6 +68,7 @@ impl Command for Join {
         call: &Call,
         input: PipelineData,
     ) -> Result<nu_protocol::PipelineData, nu_protocol::ShellError> {
+        let metadata = input.metadata();
         let table_2: Value = call.req(engine_state, stack, 0)?;
         let l_on: Value = call.req(engine_state, stack, 1)?;
         let r_on: Value = call
@@ -87,7 +88,7 @@ impl Command for Join {
                 Value::String { val: r_on, .. },
             ) => {
                 let result = join(rows_1, rows_2, l_on, r_on, join_type, span);
-                Ok(PipelineData::Value(result, None))
+                Ok(PipelineData::Value(result, metadata))
             }
             _ => Err(ShellError::UnsupportedInput {
                 msg: "(PipelineData<table>, table, string, string)".into(),

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

@@ -96,9 +96,7 @@ impl Command for Last {
 
         // early exit for `last 0`
         if rows_desired == 0 {
-            return Ok(Vec::<Value>::new()
-                .into_pipeline_data(ctrlc)
-                .set_metadata(metadata));
+            return Ok(Vec::<Value>::new().into_pipeline_data_with_metadata(metadata, ctrlc));
         }
 
         match input {
@@ -118,12 +116,12 @@ impl Command for Last {
 
                 if return_single_element {
                     if let Some(last) = buf.pop_back() {
-                        Ok(last.into_pipeline_data().set_metadata(metadata))
+                        Ok(last.into_pipeline_data_with_metadata(metadata))
                     } else {
                         Ok(PipelineData::empty().set_metadata(metadata))
                     }
                 } else {
-                    Ok(buf.into_pipeline_data(ctrlc).set_metadata(metadata))
+                    Ok(buf.into_pipeline_data_with_metadata(metadata, ctrlc))
                 }
             }
             PipelineData::Value(val, _) => {
@@ -143,8 +141,7 @@ impl Command for Last {
                                 .rev()
                                 .take(rows_desired)
                                 .rev()
-                                .into_pipeline_data(ctrlc)
-                                .set_metadata(metadata))
+                                .into_pipeline_data_with_metadata(metadata, ctrlc))
                         }
                     }
                     Value::Binary { val, .. } => {

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

@@ -123,8 +123,7 @@ only unwrap the outer list, and leave the variable's contents untouched."#
             .into_pipeline_data()
             .into_iter()
             .chain(input)
-            .into_pipeline_data(engine_state.ctrlc.clone())
-            .set_metadata(metadata))
+            .into_pipeline_data_with_metadata(metadata, engine_state.ctrlc.clone()))
     }
 }
 

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

@@ -215,6 +215,7 @@ fn reject(
     cell_paths: Vec<CellPath>,
 ) -> Result<PipelineData, ShellError> {
     let mut unique_rows: HashSet<usize> = HashSet::new();
+    let metadata = input.metadata();
     let val = input.into_value(span);
     let mut val = val;
     let mut new_columns = vec![];
@@ -257,7 +258,7 @@ fn reject(
     for cell_path in new_columns {
         val.remove_data_at_cell_path(&cell_path.members)?;
     }
-    Ok(val.into_pipeline_data())
+    Ok(val.into_pipeline_data_with_metadata(metadata))
 }
 
 #[cfg(test)]

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

@@ -72,9 +72,7 @@ impl Command for Reverse {
         #[allow(clippy::needless_collect)]
         let v: Vec<_> = input.into_iter_strict(call.head)?.collect();
         let iter = v.into_iter().rev();
-        Ok(iter
-            .into_pipeline_data(engine_state.ctrlc.clone())
-            .set_metadata(metadata))
+        Ok(iter.into_pipeline_data_with_metadata(metadata, engine_state.ctrlc.clone()))
     }
 }
 

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

@@ -244,8 +244,7 @@ fn select(
             rows: unique_rows.into_iter().peekable(),
             current: 0,
         }
-        .into_pipeline_data(engine_state.ctrlc.clone())
-        .set_metadata(metadata)
+        .into_pipeline_data_with_metadata(metadata, engine_state.ctrlc.clone())
     } else {
         input
     };
@@ -285,8 +284,7 @@ fn select(
 
                     Ok(output
                         .into_iter()
-                        .into_pipeline_data(engine_state.ctrlc.clone())
-                        .set_metadata(metadata))
+                        .into_pipeline_data_with_metadata(metadata, engine_state.ctrlc.clone()))
                 }
                 _ => {
                     if !columns.is_empty() {
@@ -303,10 +301,9 @@ fn select(
                         }
 
                         Ok(Value::record(record, call_span)
-                            .into_pipeline_data()
-                            .set_metadata(metadata))
+                            .into_pipeline_data_with_metadata(metadata))
                     } else {
-                        Ok(v.into_pipeline_data().set_metadata(metadata))
+                        Ok(v.into_pipeline_data_with_metadata(metadata))
                     }
                 }
             }
@@ -332,9 +329,7 @@ fn select(
                 }
             }
 
-            Ok(values
-                .into_pipeline_data(engine_state.ctrlc.clone())
-                .set_metadata(metadata))
+            Ok(values.into_pipeline_data_with_metadata(metadata, engine_state.ctrlc.clone()))
         }
         _ => Ok(PipelineData::empty()),
     }

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

@@ -38,9 +38,7 @@ impl Command for Shuffle {
         let mut v: Vec<_> = input.into_iter_strict(call.head)?.collect();
         v.shuffle(&mut thread_rng());
         let iter = v.into_iter();
-        Ok(iter
-            .into_pipeline_data(engine_state.ctrlc.clone())
-            .set_metadata(metadata))
+        Ok(iter.into_pipeline_data_with_metadata(metadata, engine_state.ctrlc.clone()))
     }
 
     fn examples(&self) -> Vec<Example> {

crates/nu-command/src/filters/skip/skip_.rs

@@ -130,22 +130,17 @@ impl Command for Skip {
                     }
                 }
 
-                Ok(Value::binary(output, bytes_span)
-                    .into_pipeline_data()
-                    .set_metadata(metadata))
+                Ok(Value::binary(output, bytes_span).into_pipeline_data_with_metadata(metadata))
             }
             PipelineData::Value(Value::Binary { val, .. }, metadata) => {
                 let bytes = val.into_iter().skip(n).collect::<Vec<_>>();
 
-                Ok(Value::binary(bytes, input_span)
-                    .into_pipeline_data()
-                    .set_metadata(metadata))
+                Ok(Value::binary(bytes, input_span).into_pipeline_data_with_metadata(metadata))
             }
             _ => Ok(input
                 .into_iter_strict(call.head)?
                 .skip(n)
-                .into_pipeline_data(ctrlc)
-                .set_metadata(metadata)),
+                .into_pipeline_data_with_metadata(metadata, ctrlc)),
         }
     }
 }

crates/nu-command/src/filters/skip/skip_until.rs

@@ -113,8 +113,7 @@ impl Command for SkipUntil {
                     pipeline_data.into_value(span).is_true()
                 })
             })
-            .into_pipeline_data(ctrlc)
-            .set_metadata(metadata))
+            .into_pipeline_data_with_metadata(metadata, ctrlc))
     }
 }
 

crates/nu-command/src/filters/skip/skip_while.rs

@@ -118,8 +118,7 @@ impl Command for SkipWhile {
                     pipeline_data.into_value(span).is_true()
                 })
             })
-            .into_pipeline_data(ctrlc)
-            .set_metadata(metadata))
+            .into_pipeline_data_with_metadata(metadata, ctrlc))
     }
 }
 

crates/nu-command/src/filters/take/take_.rs

@@ -60,8 +60,7 @@ impl Command for Take {
                     Value::List { vals, .. } => Ok(vals
                         .into_iter()
                         .take(rows_desired)
-                        .into_pipeline_data(ctrlc)
-                        .set_metadata(metadata)),
+                        .into_pipeline_data_with_metadata(metadata, ctrlc)),
                     Value::Binary { val, .. } => {
                         let slice: Vec<u8> = val.into_iter().take(rows_desired).collect();
                         Ok(PipelineData::Value(Value::binary(slice, span), metadata))
@@ -69,8 +68,7 @@ impl Command for Take {
                     Value::Range { val, .. } => Ok(val
                         .into_range_iter(ctrlc.clone())?
                         .take(rows_desired)
-                        .into_pipeline_data(ctrlc)
-                        .set_metadata(metadata)),
+                        .into_pipeline_data_with_metadata(metadata, ctrlc)),
                     // Propagate errors by explicitly matching them before the final case.
                     Value::Error { error, .. } => Err(*error),
                     other => Err(ShellError::OnlySupportsThisInputType {
@@ -83,8 +81,7 @@ impl Command for Take {
             }
             PipelineData::ListStream(ls, metadata) => Ok(ls
                 .take(rows_desired)
-                .into_pipeline_data(ctrlc)
-                .set_metadata(metadata)),
+                .into_pipeline_data_with_metadata(metadata, ctrlc)),
             PipelineData::ExternalStream { span, .. } => {
                 Err(ShellError::OnlySupportsThisInputType {
                     exp_input_type: "list, binary or range".into(),

crates/nu-command/src/filters/take/take_until.rs

@@ -75,6 +75,7 @@ impl Command for TakeUntil {
         call: &Call,
         input: PipelineData,
     ) -> Result<PipelineData, ShellError> {
+        let metadata = input.metadata();
         let span = call.head;
 
         let capture_block: Closure = call.req(engine_state, stack, 0)?;
@@ -109,7 +110,7 @@ impl Command for TakeUntil {
                     pipeline_data.into_value(span).is_true()
                 })
             })
-            .into_pipeline_data(ctrlc))
+            .into_pipeline_data_with_metadata(metadata, ctrlc))
     }
 }
 

crates/nu-command/src/filters/take/take_while.rs

@@ -75,6 +75,7 @@ impl Command for TakeWhile {
         call: &Call,
         input: PipelineData,
     ) -> Result<PipelineData, ShellError> {
+        let metadata = input.metadata();
         let span = call.head;
 
         let capture_block: Closure = call.req(engine_state, stack, 0)?;
@@ -109,7 +110,7 @@ impl Command for TakeWhile {
                     pipeline_data.into_value(span).is_true()
                 })
             })
-            .into_pipeline_data(ctrlc))
+            .into_pipeline_data_with_metadata(metadata, ctrlc))
     }
 }
 

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

@@ -291,7 +291,7 @@ pub fn transpose(
             metadata,
         ))
     } else {
-        Ok(result_data.into_pipeline_data(ctrlc).set_metadata(metadata))
+        Ok(result_data.into_pipeline_data_with_metadata(metadata, ctrlc))
     }
 }
 

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

@@ -311,9 +311,7 @@ pub fn uniq(
         uniq_values.into_iter().map(|v| v.val).collect()
     };
 
-    Ok(Value::list(result, head)
-        .into_pipeline_data()
-        .set_metadata(metadata))
+    Ok(Value::list(result, head).into_pipeline_data_with_metadata(metadata))
 }
 
 fn sort(iter: IntoIter<String, ValueCounter>) -> Vec<ValueCounter> {

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

@@ -128,6 +128,7 @@ fn upsert(
     call: &Call,
     input: PipelineData,
 ) -> Result<PipelineData, ShellError> {
+    let metadata = input.metadata();
     let span = call.head;
 
     let cell_path: CellPath = call.req(engine_state, stack, 0)?;
@@ -148,43 +149,45 @@ fn upsert(
         let orig_env_vars = stack.env_vars.clone();
         let orig_env_hidden = stack.env_hidden.clone();
 
-        input.map(
-            move |mut input| {
-                // with_env() is used here to ensure that each iteration uses
-                // a different set of environment variables.
-                // Hence, a 'cd' in the first loop won't affect the next loop.
-                stack.with_env(&orig_env_vars, &orig_env_hidden);
+        input
+            .map(
+                move |mut input| {
+                    // with_env() is used here to ensure that each iteration uses
+                    // a different set of environment variables.
+                    // Hence, a 'cd' in the first loop won't affect the next loop.
+                    stack.with_env(&orig_env_vars, &orig_env_hidden);
 
-                if let Some(var) = block.signature.get_positional(0) {
-                    if let Some(var_id) = &var.var_id {
-                        stack.add_var(*var_id, input.clone())
-                    }
-                }
-
-                let output = eval_block(
-                    &engine_state,
-                    &mut stack,
-                    &block,
-                    input.clone().into_pipeline_data(),
-                    redirect_stdout,
-                    redirect_stderr,
-                );
-
-                match output {
-                    Ok(pd) => {
-                        if let Err(e) =
-                            input.upsert_data_at_cell_path(&cell_path.members, pd.into_value(span))
-                        {
-                            return Value::error(e, span);
+                    if let Some(var) = block.signature.get_positional(0) {
+                        if let Some(var_id) = &var.var_id {
+                            stack.add_var(*var_id, input.clone())
                         }
-
-                        input
                     }
-                    Err(e) => Value::error(e, span),
-                }
-            },
-            ctrlc,
-        )
+
+                    let output = eval_block(
+                        &engine_state,
+                        &mut stack,
+                        &block,
+                        input.clone().into_pipeline_data(),
+                        redirect_stdout,
+                        redirect_stderr,
+                    );
+
+                    match output {
+                        Ok(pd) => {
+                            if let Err(e) = input
+                                .upsert_data_at_cell_path(&cell_path.members, pd.into_value(span))
+                            {
+                                return Value::error(e, span);
+                            }
+
+                            input
+                        }
+                        Err(e) => Value::error(e, span),
+                    }
+                },
+                ctrlc,
+            )
+            .map(|x| x.set_metadata(metadata))
     } else {
         if let Some(PathMember::Int { val, span, .. }) = cell_path.members.first() {
             let mut input = input.into_iter();
@@ -208,21 +211,24 @@ fn upsert(
                 .into_iter()
                 .chain(vec![replacement])
                 .chain(input)
-                .into_pipeline_data(ctrlc));
+                .into_pipeline_data_with_metadata(metadata, ctrlc));
         }
 
-        input.map(
-            move |mut input| {
-                let replacement = replacement.clone();
+        input
+            .map(
+                move |mut input| {
+                    let replacement = replacement.clone();
 
-                if let Err(e) = input.upsert_data_at_cell_path(&cell_path.members, replacement) {
-                    return Value::error(e, span);
-                }
+                    if let Err(e) = input.upsert_data_at_cell_path(&cell_path.members, replacement)
+                    {
+                        return Value::error(e, span);
+                    }
 
-                input
-            },
-            ctrlc,
-        )
+                    input
+                },
+                ctrlc,
+            )
+            .map(|x| x.set_metadata(metadata))
     }
 }
 

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

@@ -149,8 +149,7 @@ fn values(
                 Value::List { vals, .. } => match get_values(&vals, head, span) {
                     Ok(cols) => Ok(cols
                         .into_iter()
-                        .into_pipeline_data(ctrlc)
-                        .set_metadata(metadata)),
+                        .into_pipeline_data_with_metadata(metadata, ctrlc)),
                     Err(err) => Err(err),
                 },
                 Value::CustomValue { val, .. } => {
@@ -158,15 +157,13 @@ fn values(
                     match get_values(&[input_as_base_value], head, span) {
                         Ok(cols) => Ok(cols
                             .into_iter()
-                            .into_pipeline_data(ctrlc)
-                            .set_metadata(metadata)),
+                            .into_pipeline_data_with_metadata(metadata, ctrlc)),
                         Err(err) => Err(err),
                     }
                 }
                 Value::Record { val, .. } => Ok(val
                     .into_values()
-                    .into_pipeline_data(ctrlc)
-                    .set_metadata(metadata)),
+                    .into_pipeline_data_with_metadata(metadata, ctrlc)),
                 Value::LazyRecord { val, .. } => {
                     let record = match val.collect()? {
                         Value::Record { val, .. } => val,
@@ -178,8 +175,7 @@ fn values(
                     };
                     Ok(record
                         .into_values()
-                        .into_pipeline_data(ctrlc)
-                        .set_metadata(metadata))
+                        .into_pipeline_data_with_metadata(metadata, ctrlc))
                 }
                 // Propagate errors
                 Value::Error { error, .. } => Err(*error),
@@ -196,8 +192,7 @@ fn values(
             match get_values(&vals, head, head) {
                 Ok(cols) => Ok(cols
                     .into_iter()
-                    .into_pipeline_data(ctrlc)
-                    .set_metadata(metadata)),
+                    .into_pipeline_data_with_metadata(metadata, ctrlc)),
                 Err(err) => Err(err),
             }
         }

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

@@ -102,8 +102,7 @@ not supported."#
                     Err(err) => Some(Value::error(err, span)),
                 }
             })
-            .into_pipeline_data(ctrlc)
-            .set_metadata(metadata))
+            .into_pipeline_data_with_metadata(metadata, ctrlc))
     }
 
     fn examples(&self) -> Vec<Example> {

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

@@ -135,9 +135,7 @@ impl Command for Window {
             remainder,
         };
 
-        Ok(each_group_iterator
-            .into_pipeline_data(ctrlc)
-            .set_metadata(metadata))
+        Ok(each_group_iterator.into_pipeline_data_with_metadata(metadata, ctrlc))
     }
 }
 

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

@@ -48,17 +48,14 @@ impl Command for Wrap {
             | PipelineData::ListStream { .. } => Ok(input
                 .into_iter()
                 .map(move |x| Value::record(record! { name.clone() => x }, span))
-                .into_pipeline_data(engine_state.ctrlc.clone())
-                .set_metadata(metadata)),
+                .into_pipeline_data_with_metadata(metadata, engine_state.ctrlc.clone())),
             PipelineData::ExternalStream { .. } => Ok(Value::record(
                 record! { name => input.into_value(call.head) },
                 span,
             )
-            .into_pipeline_data()
-            .set_metadata(metadata)),
+            .into_pipeline_data_with_metadata(metadata)),
             PipelineData::Value(input, ..) => Ok(Value::record(record! { name => input }, span)
-                .into_pipeline_data()
-                .set_metadata(metadata)),
+                .into_pipeline_data_with_metadata(metadata)),
         }
     }
 

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

@@ -100,8 +100,7 @@ impl Command for Zip {
             .into_iter()
             .zip(other.into_pipeline_data())
             .map(move |(x, y)| Value::list(vec![x, y], head))
-            .into_pipeline_data(ctrlc)
-            .set_metadata(metadata))
+            .into_pipeline_data_with_metadata(metadata, ctrlc))
     }
 }