refactor Value::follow_cell_path to reduce clones and return Cow (#15640)

# Description
While working on something else, I noticed that
`Value::follow_cell_path` receives `self`.

While it would be ideal for the signature to be `(&'a self, cell_path)
-> &'a Value`, that's not possible because:
1. Selecting a row from a list and field from a record can be done with
a reference but selecting a column from a table requires creating a new
list.
2. `Value::Custom` returns new `Value`s when indexed.

So the signature becomes `(&'a self, cell_path) -> Cow<'a, Value>`.

Another complication that arises is, once a new `Value` is created, and
it is further indexed, the `current` variable
1. can't be `&'a Value`, as the lifetime requirement means it can't
refer to local variables
2. _shouldn't_ be `Cow<'a, Value>`, as once it becomes an owned value,
it can't be borrowed ever again, as `current` is derived from its
previous value in further iterations. So once it's owned, it can't be
indexed by reference, leading to more clones

We need `current` to have _two_ possible lifetimes
1. `'out`: references derived from `&self`
2. `'local`: references derived from an owned value stored in a local
variable

```rust
enum MultiLife<'out, 'local, T>
where
    'out: 'local,
    T: ?Sized,
{
    Out(&'out T),
    Local(&'local T),
}
```
With `current: MultiLife<'out, '_, Value>`, we can traverse values with
minimal clones, and we can transform it to `Cow<'out, Value>` easily
(`MultiLife::Out -> Cow::Borrowed, MultiLife::Local -> Cow::Owned`) to
return it

# User-Facing Changes

# Tests + Formatting

# After Submitting

---------

Co-authored-by: Bahex <17417311+Bahex@users.noreply.github.com>

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

@@ -15,17 +15,8 @@ pub fn empty(
     if !columns.is_empty() {
         for val in input {
             for column in &columns {
-                let val = val.clone();
-                match val.follow_cell_path(&column.members, false) {
-                    Ok(Value::Nothing { .. }) => {}
-                    Ok(_) => {
-                        if negate {
-                            return Ok(Value::bool(true, head).into_pipeline_data());
-                        } else {
-                            return Ok(Value::bool(false, head).into_pipeline_data());
-                        }
-                    }
-                    Err(err) => return Err(err),
+                if !val.follow_cell_path(&column.members, false)?.is_nothing() {
+                    return Ok(Value::bool(negate, head).into_pipeline_data());
                 }
             }
         }

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

@@ -1,3 +1,5 @@
+use std::borrow::Cow;
+
 use nu_engine::command_prelude::*;
 use nu_protocol::{ast::PathMember, Signals};
 
@@ -188,9 +190,11 @@ fn action(
         let input = input.into_value(span)?;
 
         for path in paths {
-            let val = input.clone().follow_cell_path(&path.members, !sensitive);
-
-            output.push(val?);
+            output.push(
+                input
+                    .follow_cell_path(&path.members, !sensitive)?
+                    .into_owned(),
+            );
         }
 
         Ok(output.into_iter().into_pipeline_data(span, signals))
@@ -223,10 +227,10 @@ pub fn follow_cell_path_into_stream(
                 .map(move |value| {
                     let span = value.span();
 
-                    match value.follow_cell_path(&cell_path, insensitive) {
-                        Ok(v) => v,
-                        Err(error) => Value::error(error, span),
-                    }
+                    value
+                        .follow_cell_path(&cell_path, insensitive)
+                        .map(Cow::into_owned)
+                        .unwrap_or_else(|error| Value::error(error, span))
                 })
                 .into_pipeline_data(head, signals);
 

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

@@ -322,11 +322,9 @@ fn group_cell_path(
     let mut groups = IndexMap::<_, Vec<_>>::new();
 
     for value in values.into_iter() {
-        let key = value
-            .clone()
-            .follow_cell_path(&column_name.members, false)?;
+        let key = value.follow_cell_path(&column_name.members, false)?;
 
-        if matches!(key, Value::Nothing { .. }) {
+        if key.is_nothing() {
             continue; // likely the result of a failed optional access, ignore this value
         }
 

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

@@ -1,3 +1,5 @@
+use std::borrow::Cow;
+
 use nu_engine::{command_prelude::*, ClosureEval, ClosureEvalOnce};
 use nu_protocol::ast::PathMember;
 
@@ -299,8 +301,8 @@ fn insert_value_by_closure(
 ) -> Result<(), ShellError> {
     let value_at_path = if first_path_member_int {
         value
-            .clone()
             .follow_cell_path(cell_path, false)
+            .map(Cow::into_owned)
             .unwrap_or(Value::nothing(span))
     } else {
         value.clone()
@@ -319,8 +321,8 @@ fn insert_single_value_by_closure(
 ) -> Result<(), ShellError> {
     let value_at_path = if first_path_member_int {
         value
-            .clone()
             .follow_cell_path(cell_path, false)
+            .map(Cow::into_owned)
             .unwrap_or(Value::nothing(span))
     } else {
         value.clone()

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

@@ -229,45 +229,37 @@ fn select(
             match v {
                 Value::List {
                     vals: input_vals, ..
-                } => {
-                    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 Value::error(e, call_span),
+                } => Ok(input_vals
+                    .into_iter()
+                    .map(move |input_val| {
+                        if !columns.is_empty() {
+                            let mut record = Record::new();
+                            for path in &columns {
+                                match input_val.follow_cell_path(&path.members, false) {
+                                    Ok(fetcher) => {
+                                        record.push(path.to_column_name(), fetcher.into_owned());
                                     }
+                                    Err(e) => return Value::error(e, call_span),
                                 }
-
-                                Value::record(record, span)
-                            } else {
-                                input_val.clone()
                             }
-                        })
-                        .into_pipeline_data_with_metadata(
-                            call_span,
-                            engine_state.signals().clone(),
-                            metadata,
-                        ))
-                }
+
+                            Value::record(record, span)
+                        } else {
+                            input_val.clone()
+                        }
+                    })
+                    .into_pipeline_data_with_metadata(
+                        call_span,
+                        engine_state.signals().clone(),
+                        metadata,
+                    )),
                 _ => {
                     if !columns.is_empty() {
                         let mut record = Record::new();
 
                         for cell_path in columns {
-                            // FIXME: remove clone
-                            match v.clone().follow_cell_path(&cell_path.members, false) {
-                                Ok(result) => {
-                                    record.push(cell_path.to_column_name(), result);
-                                }
-                                Err(e) => return Err(e),
-                            }
+                            let result = v.follow_cell_path(&cell_path.members, false)?;
+                            record.push(cell_path.to_column_name(), result.into_owned());
                         }
 
                         Ok(Value::record(record, call_span)
@@ -278,31 +270,24 @@ fn select(
                 }
             }
         }
-        PipelineData::ListStream(stream, metadata, ..) => {
-            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),
+        PipelineData::ListStream(stream, metadata, ..) => Ok(stream
+            .map(move |x| {
+                if !columns.is_empty() {
+                    let mut record = Record::new();
+                    for path in &columns {
+                        match x.follow_cell_path(&path.members, false) {
+                            Ok(value) => {
+                                record.push(path.to_column_name(), value.into_owned());
                             }
+                            Err(e) => return Value::error(e, call_span),
                         }
-                        Value::record(record, call_span)
-                    } else {
-                        x
                     }
-                })
-                .into_pipeline_data_with_metadata(
-                    call_span,
-                    engine_state.signals().clone(),
-                    metadata,
-                ))
-        }
+                    Value::record(record, call_span)
+                } else {
+                    x
+                }
+            })
+            .into_pipeline_data_with_metadata(call_span, engine_state.signals().clone(), metadata)),
         _ => Ok(PipelineData::empty()),
     }
 }

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

@@ -243,17 +243,17 @@ fn update_value_by_closure(
     cell_path: &[PathMember],
     first_path_member_int: bool,
 ) -> Result<(), ShellError> {
-    let value_at_path = value.clone().follow_cell_path(cell_path, false)?;
+    let value_at_path = value.follow_cell_path(cell_path, false)?;
 
     let arg = if first_path_member_int {
-        &value_at_path
+        value_at_path.as_ref()
     } else {
         &*value
     };
 
     let new_value = closure
         .add_arg(arg.clone())
-        .run_with_input(value_at_path.into_pipeline_data())?
+        .run_with_input(value_at_path.into_owned().into_pipeline_data())?
         .into_value(span)?;
 
     value.update_data_at_cell_path(cell_path, new_value)
@@ -266,17 +266,17 @@ fn update_single_value_by_closure(
     cell_path: &[PathMember],
     first_path_member_int: bool,
 ) -> Result<(), ShellError> {
-    let value_at_path = value.clone().follow_cell_path(cell_path, false)?;
+    let value_at_path = value.follow_cell_path(cell_path, false)?;
 
     let arg = if first_path_member_int {
-        &value_at_path
+        value_at_path.as_ref()
     } else {
         &*value
     };
 
     let new_value = closure
         .add_arg(arg.clone())
-        .run_with_input(value_at_path.into_pipeline_data())?
+        .run_with_input(value_at_path.into_owned().into_pipeline_data())?
         .into_value(span)?;
 
     value.update_data_at_cell_path(cell_path, new_value)

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

@@ -1,3 +1,5 @@
+use std::borrow::Cow;
+
 use nu_engine::{command_prelude::*, ClosureEval, ClosureEvalOnce};
 use nu_protocol::ast::PathMember;
 
@@ -319,15 +321,19 @@ fn upsert_value_by_closure(
     cell_path: &[PathMember],
     first_path_member_int: bool,
 ) -> Result<(), ShellError> {
-    let value_at_path = value.clone().follow_cell_path(cell_path, false);
+    let value_at_path = value.follow_cell_path(cell_path, false);
 
     let arg = if first_path_member_int {
-        value_at_path.clone().unwrap_or(Value::nothing(span))
+        value_at_path
+            .as_deref()
+            .cloned()
+            .unwrap_or(Value::nothing(span))
     } else {
         value.clone()
     };
 
     let input = value_at_path
+        .map(Cow::into_owned)
         .map(IntoPipelineData::into_pipeline_data)
         .unwrap_or(PipelineData::Empty);
 
@@ -346,15 +352,19 @@ fn upsert_single_value_by_closure(
     cell_path: &[PathMember],
     first_path_member_int: bool,
 ) -> Result<(), ShellError> {
-    let value_at_path = value.clone().follow_cell_path(cell_path, false);
+    let value_at_path = value.follow_cell_path(cell_path, false);
 
     let arg = if first_path_member_int {
-        value_at_path.clone().unwrap_or(Value::nothing(span))
+        value_at_path
+            .as_deref()
+            .cloned()
+            .unwrap_or(Value::nothing(span))
     } else {
         value.clone()
     };
 
     let input = value_at_path
+        .map(Cow::into_owned)
         .map(IntoPipelineData::into_pipeline_data)
         .unwrap_or(PipelineData::Empty);
 

crates/nu-command/src/platform/input/list.rs

@@ -89,8 +89,7 @@ impl Command for InputList {
                 .into_iter()
                 .map(move |val| {
                     let display_value = if let Some(ref cellpath) = display_path {
-                        val.clone()
-                            .follow_cell_path(&cellpath.members, false)?
+                        val.follow_cell_path(&cellpath.members, false)?
                             .to_expanded_string(", ", &config)
                     } else {
                         val.to_expanded_string(", ", &config)

crates/nu-command/src/sort_utils.rs

@@ -239,8 +239,8 @@ pub fn compare_cell_path(
     insensitive: bool,
     natural: bool,
 ) -> Result<Ordering, ShellError> {
-    let left = left.clone().follow_cell_path(&cell_path.members, false)?;
-    let right = right.clone().follow_cell_path(&cell_path.members, false)?;
+    let left = left.follow_cell_path(&cell_path.members, false)?;
+    let right = right.follow_cell_path(&cell_path.members, false)?;
     compare_values(&left, &right, insensitive, natural)
 }