Copy-on-write for record values (#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-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)
                         }
-                        _ => item,
+
+                        if !found {
+                            record.to_mut().push(column.item.clone(), value.clone());
+                        }
+
+                        item
                     }
+                    _ => item,
                 },
                 ctrlc,
             )

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

@@ -106,7 +106,7 @@ fn drop_cols(
                 Ok(PipelineData::Empty)
             }
         }
-        PipelineData::Value(v, ..) => {
+        PipelineData::Value(mut v, ..) => {
             let span = v.span();
             match v {
                 Value::List { mut vals, .. } => {
@@ -119,11 +119,12 @@ fn drop_cols(
                     Ok(Value::list(vals, span).into_pipeline_data_with_metadata(metadata))
                 }
                 Value::Record {
-                    val: mut record, ..
+                    val: ref mut record,
+                    ..
                 } => {
                     let len = record.len().saturating_sub(columns);
-                    record.truncate(len);
-                    Ok(Value::record(*record, span).into_pipeline_data_with_metadata(metadata))
+                    record.to_mut().truncate(len);
+                    Ok(v.into_pipeline_data_with_metadata(metadata))
                 }
                 // Propagate errors
                 Value::Error { error, .. } => Err(*error),
@@ -143,7 +144,7 @@ fn drop_cols(
 fn drop_cols_set(val: &mut Value, head: Span, drop: usize) -> Result<HashSet<String>, ShellError> {
     if let Value::Record { val: record, .. } = val {
         let len = record.len().saturating_sub(drop);
-        Ok(record.drain(len..).map(|(col, _)| col).collect())
+        Ok(record.to_mut().drain(len..).map(|(col, _)| col).collect())
     } else {
         Err(unsupported_value_error(val, head))
     }
@@ -155,7 +156,7 @@ fn drop_record_cols(
     drop_cols: &HashSet<String>,
 ) -> Result<(), ShellError> {
     if let Value::Record { val, .. } = val {
-        val.retain(|col, _| !drop_cols.contains(col));
+        val.to_mut().retain(|col, _| !drop_cols.contains(col));
         Ok(())
     } else {
         Err(unsupported_value_error(val, head))

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

@@ -156,15 +156,15 @@ fn flat_value(columns: &[CellPath], item: Value, all: bool) -> Vec<Value> {
             let mut out = IndexMap::<String, Value>::new();
             let mut inner_table = None;
 
-            for (column_index, (column, value)) in val.into_iter().enumerate() {
+            for (column_index, (column, value)) in val.into_owned().into_iter().enumerate() {
                 let column_requested = columns.iter().find(|c| c.to_string() == column);
                 let need_flatten = { columns.is_empty() || column_requested.is_some() };
                 let span = value.span();
 
                 match value {
-                    Value::Record { val, .. } => {
+                    Value::Record { ref val, .. } => {
                         if need_flatten {
-                            for (col, val) in *val {
+                            for (col, val) in val.clone().into_owned() {
                                 if out.contains_key(&col) {
                                     out.insert(format!("{column}_{col}"), val);
                                 } else {
@@ -172,9 +172,9 @@ fn flat_value(columns: &[CellPath], item: Value, all: bool) -> Vec<Value> {
                                 }
                             }
                         } else if out.contains_key(&column) {
-                            out.insert(format!("{column}_{column}"), Value::record(*val, span));
+                            out.insert(format!("{column}_{column}"), value);
                         } else {
-                            out.insert(column, Value::record(*val, span));
+                            out.insert(column, value);
                         }
                     }
                     Value::List { vals, .. } => {

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

@@ -149,6 +149,7 @@ fn replace_headers(
             if let Value::Record { val: record, .. } = value {
                 Ok(Value::record(
                     record
+                        .into_owned()
                         .into_iter()
                         .filter_map(|(col, val)| {
                             old_headers

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

@@ -86,6 +86,7 @@ impl Command for Items {
             PipelineData::Empty => Ok(PipelineData::Empty),
             PipelineData::Value(v, ..) => match v {
                 Value::Record { val, .. } => Ok(val
+                    .into_owned()
                     .into_iter()
                     .map_while(run_for_each_item)
                     .into_pipeline_data(ctrlc)),
@@ -99,6 +100,7 @@ impl Command for Items {
                         })?,
                     };
                     Ok(record
+                        .into_owned()
                         .into_iter()
                         .map_while(run_for_each_item)
                         .into_pipeline_data(ctrlc))

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

@@ -162,7 +162,7 @@ fn rename(
                                 block_info.clone()
                             {
                                 record
-                                    .into_iter()
+                                    .into_owned().into_iter()
                                     .map(|(col, val)| {
                                         stack.with_env(&env_vars, &env_hidden);
 
@@ -191,7 +191,7 @@ fn rename(
                                         // record columns are unique so we can track the number
                                         // of renamed columns to check if any were missed
                                         let mut renamed = 0;
-                                        let record = record.into_iter().map(|(col, val)| {
+                                        let record = record.into_owned().into_iter().map(|(col, val)| {
                                             let col = if let Some(col) = columns.get(&col) {
                                                 renamed += 1;
                                                 col.clone()
@@ -222,7 +222,7 @@ fn rename(
                                         }
                                     }
                                     None => Ok(record
-                                        .into_iter()
+                                        .into_owned().into_iter()
                                         .enumerate()
                                         .map(|(i, (col, val))| {
                                             (columns.get(i).cloned().unwrap_or(col), val)
@@ -237,7 +237,7 @@ fn rename(
                         }
                     }
                     // Propagate errors by explicitly matching them before the final case.
-                    Value::Error { .. } => item.clone(),
+                    Value::Error { .. } => item,
                     other => Value::error(
                         ShellError::OnlySupportsThisInputType {
                             exp_input_type: "record".into(),

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

@@ -144,7 +144,14 @@ impl Command for Sort {
             // Records have two sorting methods, toggled by presence or absence of -v
             PipelineData::Value(Value::Record { val, .. }, ..) => {
                 let sort_by_value = call.has_flag(engine_state, stack, "values")?;
-                let record = sort_record(*val, span, sort_by_value, reverse, insensitive, natural);
+                let record = sort_record(
+                    val.into_owned(),
+                    span,
+                    sort_by_value,
+                    reverse,
+                    insensitive,
+                    natural,
+                );
                 Ok(record.into_pipeline_data())
             }
             // Other values are sorted here

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

@@ -187,11 +187,11 @@ pub fn data_split(
             let span = v.span();
             match v {
                 Value::Record { val: grouped, .. } => {
-                    for (outer_key, list) in grouped.into_iter() {
+                    for (outer_key, list) in grouped.into_owned() {
                         match data_group(&list, splitter, span) {
                             Ok(grouped_vals) => {
                                 if let Value::Record { val: sub, .. } = grouped_vals {
-                                    for (inner_key, subset) in sub.into_iter() {
+                                    for (inner_key, subset) in sub.into_owned() {
                                         let s: &mut IndexMap<String, Value> =
                                             splits.entry(inner_key).or_default();
 

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

@@ -195,6 +195,7 @@ fn sort_attributes(val: Value) -> Value {
         Value::Record { val, .. } => {
             // TODO: sort inplace
             let sorted = val
+                .into_owned()
                 .into_iter()
                 .sorted_by(|a, b| a.0.cmp(&b.0))
                 .collect_vec();

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

@@ -157,7 +157,9 @@ fn values(
                     }
                 }
                 Value::Record { val, .. } => Ok(val
-                    .into_values()
+                    .values()
+                    .cloned()
+                    .collect::<Vec<_>>()
                     .into_pipeline_data_with_metadata(metadata, ctrlc)),
                 Value::LazyRecord { val, .. } => {
                     let record = match val.collect()? {
@@ -169,6 +171,7 @@ fn values(
                         })?,
                     };
                     Ok(record
+                        .into_owned()
                         .into_values()
                         .into_pipeline_data_with_metadata(metadata, ctrlc))
                 }

crates/nu-command/src/formats/to/text.rs

@@ -125,6 +125,7 @@ fn local_into_string(value: Value, separator: &str, config: &Config) -> String {
             .collect::<Vec<_>>()
             .join(separator),
         Value::Record { val, .. } => val
+            .into_owned()
             .into_iter()
             .map(|(x, y)| format!("{}: {}", x, local_into_string(y, ", ", config)))
             .collect::<Vec<_>>()

crates/nu-command/src/formats/to/xml.rs

@@ -325,8 +325,8 @@ impl Job {
             // alternatives like {tag: a attributes: {} content: []}, {tag: a attribbutes: null
             // content: null}, {tag: a}. See to_xml_entry for more
             let attrs = match attrs {
-                Value::Record { val, .. } => val,
-                Value::Nothing { .. } => Box::new(Record::new()),
+                Value::Record { val, .. } => val.into_owned(),
+                Value::Nothing { .. } => Record::new(),
                 _ => {
                     return Err(ShellError::CantConvert {
                         to_type: "XML".into(),
@@ -350,7 +350,7 @@ impl Job {
                 }
             };
 
-            self.write_tag(entry_span, tag, tag_span, *attrs, content)
+            self.write_tag(entry_span, tag, tag_span, attrs, content)
         }
     }
 

crates/nu-command/src/generators/generate.rs

@@ -136,7 +136,7 @@ used as the next argument to the closure, otherwise generation stops.
                     match value {
                         // {out: ..., next: ...} -> output and continue
                         Value::Record { val, .. } => {
-                            let iter = val.into_iter();
+                            let iter = val.into_owned().into_iter();
                             let mut out = None;
                             let mut next = None;
                             let mut err = None;

crates/nu-command/src/help/help_.rs

@@ -139,19 +139,20 @@ pub fn highlight_search_in_table(
     let search_string = search_string.to_folded_case();
     let mut matches = vec![];
 
-    for record in table {
-        let span = record.span();
-        let (mut record, record_span) = if let Value::Record { val, .. } = record {
-            (val, span)
-        } else {
+    for mut value in table {
+        let Value::Record {
+            val: ref mut record,
+            ..
+        } = value
+        else {
             return Err(ShellError::NushellFailedSpanned {
                 msg: "Expected record".to_string(),
-                label: format!("got {}", record.get_type()),
-                span: record.span(),
+                label: format!("got {}", value.get_type()),
+                span: value.span(),
             });
         };
 
-        let has_match = record.iter_mut().try_fold(
+        let has_match = record.to_mut().iter_mut().try_fold(
             false,
             |acc: bool, (col, val)| -> Result<bool, ShellError> {
                 if !searched_cols.contains(&col.as_str()) {
@@ -180,7 +181,7 @@ pub fn highlight_search_in_table(
         )?;
 
         if has_match {
-            matches.push(Value::record(*record, record_span));
+            matches.push(value);
         }
     }
 

crates/nu-command/src/math/utils.rs

@@ -80,15 +80,15 @@ pub fn calculate(
             ),
             _ => mf(vals, span, name),
         },
-        PipelineData::Value(Value::Record { val: record, .. }, ..) => {
-            let mut record = record;
+        PipelineData::Value(Value::Record { val, .. }, ..) => {
+            let mut record = val.into_owned();
             record
                 .iter_mut()
                 .try_for_each(|(_, val)| -> Result<(), ShellError> {
                     *val = mf(slice::from_ref(val), span, name)?;
                     Ok(())
                 })?;
-            Ok(Value::record(*record, span))
+            Ok(Value::record(record, span))
         }
         PipelineData::Value(Value::Range { val, .. }, ..) => {
             let new_vals: Result<Vec<Value>, ShellError> = val

crates/nu-command/src/network/http/client.rs

@@ -222,7 +222,7 @@ pub fn send_request(
         Value::Record { val, .. } if body_type == BodyType::Form => {
             let mut data: Vec<(String, String)> = Vec::with_capacity(val.len());
 
-            for (col, val) in *val {
+            for (col, val) in val.into_owned() {
                 data.push((col, val.coerce_into_string()?))
             }
 

crates/nu-command/src/network/url/join.rs

@@ -92,6 +92,7 @@ impl Command for SubCommand {
                 match value {
                     Value::Record { val, .. } => {
                         let url_components = val
+                            .into_owned()
                             .into_iter()
                             .try_fold(UrlComponents::new(), |url, (k, v)| {
                                 url.add_component(k, v, span, engine_state)
@@ -179,6 +180,7 @@ impl UrlComponents {
             return match value {
                 Value::Record { val, .. } => {
                     let mut qs = val
+                        .into_owned()
                         .into_iter()
                         .map(|(k, v)| match v.coerce_into_string() {
                             Ok(val) => Ok(format!("{k}={val}")),

crates/nu-command/src/viewers/griddle.rs

@@ -108,7 +108,7 @@ prints out the list properly."#
                 // dbg!("value::record");
                 let mut items = vec![];
 
-                for (i, (c, v)) in val.into_iter().enumerate() {
+                for (i, (c, v)) in val.into_owned().into_iter().enumerate() {
                     items.push((i, c, v.to_expanded_string(", ", config)))
                 }
 

crates/nu-command/src/viewers/table.rs

@@ -392,7 +392,7 @@ fn handle_table_command(
         }
         PipelineData::Value(Value::Record { val, .. }, ..) => {
             input.data = PipelineData::Empty;
-            handle_record(input, cfg, *val)
+            handle_record(input, cfg, val.into_owned())
         }
         PipelineData::Value(Value::LazyRecord { val, .. }, ..) => {
             input.data = val.collect()?.into_pipeline_data();
@@ -557,7 +557,7 @@ fn handle_row_stream(
                 stream.map(move |mut x| match &mut x {
                     Value::Record { val: record, .. } => {
                         // Only the name column gets special colors, for now
-                        if let Some(value) = record.get_mut("name") {
+                        if let Some(value) = record.to_mut().get_mut("name") {
                             let span = value.span();
                             if let Value::String { val, .. } = value {
                                 if let Some(val) = render_path_name(val, &config, &ls_colors, span)
@@ -583,7 +583,7 @@ fn handle_row_stream(
             ListStream::from_stream(
                 stream.map(move |mut x| match &mut x {
                     Value::Record { val: record, .. } => {
-                        for (rec_col, rec_val) in record.iter_mut() {
+                        for (rec_col, rec_val) in record.to_mut().iter_mut() {
                             // Every column in the HTML theme table except 'name' is colored
                             if rec_col != "name" {
                                 continue;