Copy-on-write for record values (nushell#12305)

# Description
This adds a `SharedCow` type as a transparent copy-on-write pointer that
clones to unique on mutate.

As an initial test, the `Record` within `Value::Record` is shared.

There are some pretty big wins for performance. I'll post benchmark
results in a comment. The biggest winner is nested access, as that would
have cloned the records for each cell path follow before and it doesn't
have to anymore.

The reusability of the `SharedCow` type is nice and I think it could be
used to clean up the previous work I did with `Arc` in `EngineState`.
It's meant to be a mostly transparent clone-on-write that just clones on
`.to_mut()` or `.into_owned()` if there are actually multiple
references, but avoids cloning if the reference is unique.

# User-Facing Changes
- `Value::Record` field is a different type (plugin authors)

# Tests + Formatting
- 🟢 `toolkit fmt`
- 🟢 `toolkit clippy`
- 🟢 `toolkit test`
- 🟢 `toolkit test stdlib`

# After Submitting
- [ ] use for `EngineState`
- [ ] use for `Value::List`

crates/nu-cli/src/completions/variable_completions.rs

@@ -68,9 +68,7 @@ impl Completer for VariableCompletion {
                         self.var_context.1.clone().into_iter().skip(1).collect();
 
                     if let Some(val) = env_vars.get(&target_var_str) {
-                        for suggestion in
-                            nested_suggestions(val.clone(), nested_levels, current_span)
-                        {
+                        for suggestion in nested_suggestions(val, &nested_levels, current_span) {
                             if options.match_algorithm.matches_u8_insensitive(
                                 options.case_sensitive,
                                 suggestion.suggestion.value.as_bytes(),
@@ -117,8 +115,7 @@ impl Completer for VariableCompletion {
                     nu_protocol::NU_VARIABLE_ID,
                     nu_protocol::Span::new(current_span.start, current_span.end),
                 ) {
-                    for suggestion in
-                        nested_suggestions(nuval, self.var_context.1.clone(), current_span)
+                    for suggestion in nested_suggestions(&nuval, &self.var_context.1, current_span)
                     {
                         if options.match_algorithm.matches_u8_insensitive(
                             options.case_sensitive,
@@ -140,8 +137,7 @@ impl Completer for VariableCompletion {
 
                 // If the value exists and it's of type Record
                 if let Ok(value) = var {
-                    for suggestion in
-                        nested_suggestions(value, self.var_context.1.clone(), current_span)
+                    for suggestion in nested_suggestions(&value, &self.var_context.1, current_span)
                     {
                         if options.match_algorithm.matches_u8_insensitive(
                             options.case_sensitive,
@@ -244,21 +240,21 @@ impl Completer for VariableCompletion {
 // Find recursively the values for sublevels
 // if no sublevels are set it returns the current value
 fn nested_suggestions(
-    val: Value,
-    sublevels: Vec<Vec<u8>>,
+    val: &Value,
+    sublevels: &[Vec<u8>],
     current_span: reedline::Span,
 ) -> Vec<SemanticSuggestion> {
     let mut output: Vec<SemanticSuggestion> = vec![];
-    let value = recursive_value(val, sublevels);
+    let value = recursive_value(val, sublevels).unwrap_or_else(Value::nothing);
 
     let kind = SuggestionKind::Type(value.get_type());
     match value {
         Value::Record { val, .. } => {
             // Add all the columns as completion
-            for (col, _) in val.into_iter() {
+            for col in val.columns() {
                 output.push(SemanticSuggestion {
                     suggestion: Suggestion {
-                        value: col,
+                        value: col.clone(),
                         description: None,
                         style: None,
                         extra: None,
@@ -311,56 +307,47 @@ fn nested_suggestions(
 }
 
 // Extracts the recursive value (e.g: $var.a.b.c)
-fn recursive_value(val: Value, sublevels: Vec<Vec<u8>>) -> Value {
+fn recursive_value(val: &Value, sublevels: &[Vec<u8>]) -> Result<Value, Span> {
     // Go to next sublevel
-    if let Some(next_sublevel) = sublevels.clone().into_iter().next() {
+    if let Some((sublevel, next_sublevels)) = sublevels.split_first() {
         let span = val.span();
         match val {
             Value::Record { val, .. } => {
-                for item in *val {
-                    // Check if index matches with sublevel
-                    if item.0.as_bytes().to_vec() == next_sublevel {
-                        // If matches try to fetch recursively the next
-                        return recursive_value(item.1, sublevels.into_iter().skip(1).collect());
-                    }
+                if let Some((_, value)) = val.iter().find(|(key, _)| key.as_bytes() == sublevel) {
+                    // If matches try to fetch recursively the next
+                    recursive_value(value, next_sublevels)
+                } else {
+                    // Current sublevel value not found
+                    Err(span)
                 }
-
-                // Current sublevel value not found
-                return Value::nothing(span);
             }
             Value::LazyRecord { val, .. } => {
                 for col in val.column_names() {
-                    if col.as_bytes().to_vec() == next_sublevel {
-                        return recursive_value(
-                            val.get_column_value(col).unwrap_or_default(),
-                            sublevels.into_iter().skip(1).collect(),
-                        );
+                    if col.as_bytes() == *sublevel {
+                        let val = val.get_column_value(col).map_err(|_| span)?;
+                        return recursive_value(&val, next_sublevels);
                     }
                 }
 
                 // Current sublevel value not found
-                return Value::nothing(span);
+                Err(span)
             }
             Value::List { vals, .. } => {
                 for col in get_columns(vals.as_slice()) {
-                    if col.as_bytes().to_vec() == next_sublevel {
-                        return recursive_value(
-                            Value::list(vals, span)
-                                .get_data_by_key(&col)
-                                .unwrap_or_default(),
-                            sublevels.into_iter().skip(1).collect(),
-                        );
+                    if col.as_bytes() == *sublevel {
+                        let val = val.get_data_by_key(&col).ok_or(span)?;
+                        return recursive_value(&val, next_sublevels);
                     }
                 }
 
                 // Current sublevel value not found
-                return Value::nothing(span);
+                Err(span)
             }
-            _ => return val,
+            _ => Ok(val.clone()),
         }
+    } else {
+        Ok(val.clone())
     }
-
-    val
 }
 
 impl MatchAlgorithm {

crates/nu-cmd-dataframe/src/dataframe/values/nu_dataframe/mod.rs

@@ -176,9 +176,11 @@ impl NuDataFrame {
 
                     conversion::insert_record(&mut column_values, record, &maybe_schema)?
                 }
-                Value::Record { val: record, .. } => {
-                    conversion::insert_record(&mut column_values, *record, &maybe_schema)?
-                }
+                Value::Record { val: record, .. } => conversion::insert_record(
+                    &mut column_values,
+                    record.into_owned(),
+                    &maybe_schema,
+                )?,
                 _ => {
                     let key = "0".to_string();
                     conversion::insert_value(value, key, &mut column_values, &maybe_schema)?

crates/nu-cmd-extra/src/extra/filters/roll/mod.rs

@@ -58,7 +58,7 @@ fn horizontal_rotate_value(
         Value::Record { val: record, .. } => {
             let rotations = by.map(|n| n % record.len()).unwrap_or(1);
 
-            let (mut cols, mut vals): (Vec<_>, Vec<_>) = record.into_iter().unzip();
+            let (mut cols, mut vals): (Vec<_>, Vec<_>) = record.into_owned().into_iter().unzip();
             if !cells_only {
                 match direction {
                     HorizontalDirection::Right => cols.rotate_right(rotations),

crates/nu-cmd-extra/src/extra/filters/rotate.rs

@@ -171,7 +171,7 @@ pub fn rotate(
             let span = val.span();
             match val {
                 Value::Record { val: record, .. } => {
-                    let (cols, vals): (Vec<_>, Vec<_>) = record.into_iter().unzip();
+                    let (cols, vals): (Vec<_>, Vec<_>) = record.into_owned().into_iter().unzip();
                     old_column_names = cols;
                     new_values.extend_from_slice(&vals);
                 }

crates/nu-cmd-extra/src/extra/filters/update_cells.rs

@@ -154,7 +154,8 @@ impl Iterator for UpdateCellIterator {
                 let span = val.span();
                 match val {
                     Value::Record { val, .. } => Some(Value::record(
-                        val.into_iter()
+                        val.into_owned()
+                            .into_iter()
                             .map(|(col, val)| match &self.columns {
                                 Some(cols) if !cols.contains(&col) => (col, val),
                                 _ => (

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

@@ -267,7 +267,7 @@ fn compact_primitive_description(mut value: Value) -> Value {
         if val.len() != 1 {
             return value;
         }
-        if let Some(type_name) = val.get_mut("type") {
+        if let Some(type_name) = val.to_mut().get_mut("type") {
             return std::mem::take(type_name);
         }
     }
@@ -303,7 +303,8 @@ fn describe_value(
             ),
             head,
         ),
-        Value::Record { mut val, .. } => {
+        Value::Record { val, .. } => {
+            let mut val = val.into_owned();
             for (_k, v) in val.iter_mut() {
                 *v = compact_primitive_description(describe_value(
                     std::mem::take(v),
@@ -317,7 +318,7 @@ fn describe_value(
                 record!(
                     "type" => Value::string("record", head),
                     "lazy" => Value::bool(false, head),
-                    "columns" => Value::record(*val, head),
+                    "columns" => Value::record(val, head),
                 ),
                 head,
             )
@@ -395,10 +396,11 @@ fn describe_value(
 
             if options.collect_lazyrecords {
                 let collected = val.collect()?;
-                if let Value::Record { mut val, .. } =
+                if let Value::Record { val, .. } =
                     describe_value(collected, head, engine_state, options)?
                 {
-                    record.push("length", Value::int(val.len() as i64, head));
+                    let mut val = Record::clone(&val);
+
                     for (_k, v) in val.iter_mut() {
                         *v = compact_primitive_description(describe_value(
                             std::mem::take(v),
@@ -408,7 +410,8 @@ fn describe_value(
                         )?);
                     }
 
-                    record.push("columns", Value::record(*val, head));
+                    record.push("length", Value::int(val.len() as i64, head));
+                    record.push("columns", Value::record(val, head));
                 } else {
                     let cols = val.column_names();
                     record.push("length", Value::int(cols.len() as i64, head));

crates/nu-command/src/charting/histogram.rs

@@ -177,9 +177,9 @@ fn run_histogram(
                 match v {
                     // parse record, and fill valid value to actual input.
                     Value::Record { val, .. } => {
-                        for (c, v) in *val {
-                            if &c == col_name {
-                                if let Ok(v) = HashableValue::from_value(v, head_span) {
+                        for (c, v) in val.iter() {
+                            if c == col_name {
+                                if let Ok(v) = HashableValue::from_value(v.clone(), head_span) {
                                     inputs.push(v);
                                 }
                             }

crates/nu-command/src/conversions/into/record.rs

@@ -131,7 +131,7 @@ fn into_record(
                 .collect(),
             span,
         ),
-        Value::Record { val, .. } => Value::record(*val, span),
+        Value::Record { .. } => input,
         Value::Error { .. } => input,
         other => Value::error(
             ShellError::TypeMismatch {

crates/nu-command/src/conversions/into/value.rs

@@ -108,7 +108,8 @@ impl Iterator for UpdateCellIterator {
                 let span = val.span();
                 match val {
                     Value::Record { val, .. } => Some(Value::record(
-                        val.into_iter()
+                        val.into_owned()
+                            .into_iter()
                             .map(|(col, val)| match &self.columns {
                                 Some(cols) if !cols.contains(&col) => (col, val),
                                 _ => (

crates/nu-command/src/database/commands/into_sqlite.rs

@@ -316,7 +316,7 @@ fn insert_value(
     match stream_value {
         // map each column value into its SQL representation
         Value::Record { val, .. } => {
-            let sql_vals = values_to_sql(val.into_values())?;
+            let sql_vals = values_to_sql(val.values().cloned())?;
 
             insert_statement
                 .execute(rusqlite::params_from_iter(sql_vals))

crates/nu-command/src/database/values/sqlite.rs

@@ -500,7 +500,7 @@ pub fn nu_value_to_params(value: Value) -> Result<NuSqlParams, ShellError> {
         Value::Record { val, .. } => {
             let mut params = Vec::with_capacity(val.len());
 
-            for (mut column, value) in val.into_iter() {
+            for (mut column, value) in val.into_owned().into_iter() {
                 let sql_type_erased = value_to_sql(value)?;
 
                 if !column.starts_with([':', '@', '$']) {

crates/nu-command/src/debug/inspect_table.rs

@@ -199,7 +199,7 @@ mod util {
         let span = value.span();
         match value {
             Value::Record { val: record, .. } => {
-                let (cols, vals): (Vec<_>, Vec<_>) = record.into_iter().unzip();
+                let (cols, vals): (Vec<_>, Vec<_>) = record.into_owned().into_iter().unzip();
                 (
                     cols,
                     vec![vals

crates/nu-command/src/env/load_env.rs

@@ -53,7 +53,7 @@ impl Command for LoadEnv {
             }
             None => match input {
                 PipelineData::Value(Value::Record { val, .. }, ..) => {
-                    for (env_var, rhs) in *val {
+                    for (env_var, rhs) in val.into_owned() {
                         let env_var_ = env_var.as_str();
                         if ["FILE_PWD", "CURRENT_FILE"].contains(&env_var_) {
                             return Err(ShellError::AutomaticEnvVarSetManually {

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

@@ -118,6 +118,7 @@ fn getcol(
                     })
                 }
                 Value::Record { val, .. } => Ok(val
+                    .into_owned()
                     .into_iter()
                     .map(move |(x, _)| Value::string(x, head))
                     .into_pipeline_data(ctrlc)

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

@@ -85,31 +85,29 @@ fn default(
     if let Some(column) = column {
         input
             .map(
-                move |item| {
-                    let span = item.span();
-                    match item {
-                        Value::Record {
-                            val: mut record, ..
-                        } => {
-                            let mut found = false;
+                move |mut item| match item {
+                    Value::Record {
+                        val: ref 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.to_mut().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());
-                            }
-
-                            Value::record(*record, span)
+                        if !found {
+                            record.to_mut().push(column.item.clone(), value.clone());
                         }
-                        _ => item,
+
+                        item
                     }
+                    _ => item,
                 },
                 ctrlc,
             )