From 2ae9ad86766e09a0e16dcc5c30abe2b2eb811d77 Mon Sep 17 00:00:00 2001
From: Devyn Cairns <devyn.cairns@gmail.com>
Date: Sat, 13 Apr 2024 18:42:03 -0700
Subject: [PATCH] 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
- :green_circle: `toolkit fmt`
- :green_circle: `toolkit clippy`
- :green_circle: `toolkit test`
- :green_circle: `toolkit test stdlib`

# After Submitting
- [ ] use for `EngineState`
- [ ] use for `Value::List`
---
 Cargo.lock                                    |   2 +
 .../src/completions/variable_completions.rs   |  67 +++++------
 .../src/dataframe/values/nu_dataframe/mod.rs  |   8 +-
 .../src/extra/filters/roll/mod.rs             |   2 +-
 .../nu-cmd-extra/src/extra/filters/rotate.rs  |   2 +-
 .../src/extra/filters/update_cells.rs         |   3 +-
 .../nu-cmd-lang/src/core_commands/describe.rs |  15 ++-
 crates/nu-command/src/charting/histogram.rs   |   6 +-
 .../nu-command/src/conversions/into/record.rs |   2 +-
 .../nu-command/src/conversions/into/value.rs  |   3 +-
 .../src/database/commands/into_sqlite.rs      |   2 +-
 .../nu-command/src/database/values/sqlite.rs  |   2 +-
 crates/nu-command/src/debug/inspect_table.rs  |   2 +-
 crates/nu-command/src/env/load_env.rs         |   2 +-
 crates/nu-command/src/filters/columns.rs      |   1 +
 crates/nu-command/src/filters/default.rs      |  38 +++---
 crates/nu-command/src/filters/drop/column.rs  |  13 +-
 crates/nu-command/src/filters/flatten.rs      |  10 +-
 crates/nu-command/src/filters/headers.rs      |   1 +
 crates/nu-command/src/filters/items.rs        |   2 +
 crates/nu-command/src/filters/rename.rs       |   8 +-
 crates/nu-command/src/filters/sort.rs         |   9 +-
 crates/nu-command/src/filters/split_by.rs     |   4 +-
 crates/nu-command/src/filters/uniq.rs         |   1 +
 crates/nu-command/src/filters/values.rs       |   5 +-
 crates/nu-command/src/formats/to/text.rs      |   1 +
 crates/nu-command/src/formats/to/xml.rs       |   6 +-
 crates/nu-command/src/generators/generate.rs  |   2 +-
 crates/nu-command/src/help/help_.rs           |  19 +--
 crates/nu-command/src/math/utils.rs           |   6 +-
 crates/nu-command/src/network/http/client.rs  |   2 +-
 crates/nu-command/src/network/url/join.rs     |   2 +
 crates/nu-command/src/viewers/griddle.rs      |   2 +-
 crates/nu-command/src/viewers/table.rs        |   6 +-
 crates/nu-explore/src/nu_common/table.rs      |   8 +-
 crates/nu-explore/src/nu_common/value.rs      |   2 +-
 crates/nu-explore/src/pager/mod.rs            |   2 +-
 .../nu-plugin-test-support/src/plugin_test.rs |   4 +-
 crates/nu-plugin/Cargo.toml                   |   1 +
 .../src/protocol/plugin_custom_value.rs       |  51 ++------
 crates/nu-protocol/src/config/mod.rs          |  22 ++--
 crates/nu-protocol/src/config/plugin_gc.rs    |   6 +-
 crates/nu-protocol/src/eval_base.rs           |   2 +-
 crates/nu-protocol/src/value/from_value.rs    |   2 +-
 crates/nu-protocol/src/value/mod.rs           |  59 ++++-----
 crates/nu-protocol/src/value/record.rs        |   2 +-
 crates/nu-table/src/types/collapse.rs         |   4 +-
 crates/nu-table/src/unstructured_table.rs     |   5 +-
 crates/nu-utils/Cargo.toml                    |   1 +
 crates/nu-utils/src/lib.rs                    |   2 +
 crates/nu-utils/src/shared_cow.rs             | 113 ++++++++++++++++++
 .../src/dataframe/values/nu_dataframe/mod.rs  |   8 +-
 52 files changed, 328 insertions(+), 222 deletions(-)
 create mode 100644 crates/nu-utils/src/shared_cow.rs

diff --git a/Cargo.lock b/Cargo.lock
index ab45c57268..18d9194675 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -3081,6 +3081,7 @@ dependencies = [
  "miette",
  "nu-engine",
  "nu-protocol",
+ "nu-utils",
  "rmp-serde",
  "semver",
  "serde",
@@ -3210,6 +3211,7 @@ dependencies = [
  "lscolors",
  "nix",
  "num-format",
+ "serde",
  "strip-ansi-escapes",
  "sys-locale",
  "unicase",
diff --git a/crates/nu-cli/src/completions/variable_completions.rs b/crates/nu-cli/src/completions/variable_completions.rs
index 826e1e412a..c8cadc0d0b 100644
--- a/crates/nu-cli/src/completions/variable_completions.rs
+++ b/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 {
diff --git a/crates/nu-cmd-dataframe/src/dataframe/values/nu_dataframe/mod.rs b/crates/nu-cmd-dataframe/src/dataframe/values/nu_dataframe/mod.rs
index bc0a26b6a1..b59175a404 100644
--- a/crates/nu-cmd-dataframe/src/dataframe/values/nu_dataframe/mod.rs
+++ b/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)?
diff --git a/crates/nu-cmd-extra/src/extra/filters/roll/mod.rs b/crates/nu-cmd-extra/src/extra/filters/roll/mod.rs
index 446200e0b8..77c668167a 100644
--- a/crates/nu-cmd-extra/src/extra/filters/roll/mod.rs
+++ b/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),
diff --git a/crates/nu-cmd-extra/src/extra/filters/rotate.rs b/crates/nu-cmd-extra/src/extra/filters/rotate.rs
index f3988bcadb..ba393f45ef 100644
--- a/crates/nu-cmd-extra/src/extra/filters/rotate.rs
+++ b/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);
                 }
diff --git a/crates/nu-cmd-extra/src/extra/filters/update_cells.rs b/crates/nu-cmd-extra/src/extra/filters/update_cells.rs
index cc1fc877c3..12c3520abe 100644
--- a/crates/nu-cmd-extra/src/extra/filters/update_cells.rs
+++ b/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),
                                 _ => (
diff --git a/crates/nu-cmd-lang/src/core_commands/describe.rs b/crates/nu-cmd-lang/src/core_commands/describe.rs
index 46e6e9a91e..70e381b7d6 100644
--- a/crates/nu-cmd-lang/src/core_commands/describe.rs
+++ b/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));
diff --git a/crates/nu-command/src/charting/histogram.rs b/crates/nu-command/src/charting/histogram.rs
index bd94af2ed3..acf10e2c01 100755
--- a/crates/nu-command/src/charting/histogram.rs
+++ b/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);
                                 }
                             }
diff --git a/crates/nu-command/src/conversions/into/record.rs b/crates/nu-command/src/conversions/into/record.rs
index 3df3aaf1b1..a4aebe7a1a 100644
--- a/crates/nu-command/src/conversions/into/record.rs
+++ b/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 {
diff --git a/crates/nu-command/src/conversions/into/value.rs b/crates/nu-command/src/conversions/into/value.rs
index 9c3cda77a2..e19ecf3c0c 100644
--- a/crates/nu-command/src/conversions/into/value.rs
+++ b/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),
                                 _ => (
diff --git a/crates/nu-command/src/database/commands/into_sqlite.rs b/crates/nu-command/src/database/commands/into_sqlite.rs
index 730bf92b78..876d763696 100644
--- a/crates/nu-command/src/database/commands/into_sqlite.rs
+++ b/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))
diff --git a/crates/nu-command/src/database/values/sqlite.rs b/crates/nu-command/src/database/values/sqlite.rs
index 6225d12897..9778f44993 100644
--- a/crates/nu-command/src/database/values/sqlite.rs
+++ b/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([':', '@', '$']) {
diff --git a/crates/nu-command/src/debug/inspect_table.rs b/crates/nu-command/src/debug/inspect_table.rs
index d2a112c70e..c7d5a87d9e 100644
--- a/crates/nu-command/src/debug/inspect_table.rs
+++ b/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
diff --git a/crates/nu-command/src/env/load_env.rs b/crates/nu-command/src/env/load_env.rs
index 0eba17528f..9e622a5803 100644
--- a/crates/nu-command/src/env/load_env.rs
+++ b/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 {
diff --git a/crates/nu-command/src/filters/columns.rs b/crates/nu-command/src/filters/columns.rs
index 6d0535bad1..394370a644 100644
--- a/crates/nu-command/src/filters/columns.rs
+++ b/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)
diff --git a/crates/nu-command/src/filters/default.rs b/crates/nu-command/src/filters/default.rs
index 30619d0f26..3eaa8d342e 100644
--- a/crates/nu-command/src/filters/default.rs
+++ b/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,
             )
diff --git a/crates/nu-command/src/filters/drop/column.rs b/crates/nu-command/src/filters/drop/column.rs
index 1862ac51c3..a19190cbd6 100644
--- a/crates/nu-command/src/filters/drop/column.rs
+++ b/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))
diff --git a/crates/nu-command/src/filters/flatten.rs b/crates/nu-command/src/filters/flatten.rs
index f7f5a1325b..073e84368d 100644
--- a/crates/nu-command/src/filters/flatten.rs
+++ b/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, .. } => {
diff --git a/crates/nu-command/src/filters/headers.rs b/crates/nu-command/src/filters/headers.rs
index 0e34977014..f84f49b4be 100644
--- a/crates/nu-command/src/filters/headers.rs
+++ b/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
diff --git a/crates/nu-command/src/filters/items.rs b/crates/nu-command/src/filters/items.rs
index 60c6b67acf..dc6f37ebad 100644
--- a/crates/nu-command/src/filters/items.rs
+++ b/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))
diff --git a/crates/nu-command/src/filters/rename.rs b/crates/nu-command/src/filters/rename.rs
index 8d132bb99b..8f5780dcf7 100644
--- a/crates/nu-command/src/filters/rename.rs
+++ b/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(),
diff --git a/crates/nu-command/src/filters/sort.rs b/crates/nu-command/src/filters/sort.rs
index cd0818a8c9..451f4bdcef 100644
--- a/crates/nu-command/src/filters/sort.rs
+++ b/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
diff --git a/crates/nu-command/src/filters/split_by.rs b/crates/nu-command/src/filters/split_by.rs
index 7dea03a6e6..349c7ec49b 100644
--- a/crates/nu-command/src/filters/split_by.rs
+++ b/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();
 
diff --git a/crates/nu-command/src/filters/uniq.rs b/crates/nu-command/src/filters/uniq.rs
index 13e3e900bc..e8ac53df52 100644
--- a/crates/nu-command/src/filters/uniq.rs
+++ b/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();
diff --git a/crates/nu-command/src/filters/values.rs b/crates/nu-command/src/filters/values.rs
index 640ef1bb44..691e260163 100644
--- a/crates/nu-command/src/filters/values.rs
+++ b/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))
                 }
diff --git a/crates/nu-command/src/formats/to/text.rs b/crates/nu-command/src/formats/to/text.rs
index 07f91bb258..89efe708f6 100644
--- a/crates/nu-command/src/formats/to/text.rs
+++ b/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<_>>()
diff --git a/crates/nu-command/src/formats/to/xml.rs b/crates/nu-command/src/formats/to/xml.rs
index c5b1238714..f93bc94d28 100644
--- a/crates/nu-command/src/formats/to/xml.rs
+++ b/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)
         }
     }
 
diff --git a/crates/nu-command/src/generators/generate.rs b/crates/nu-command/src/generators/generate.rs
index e8a36e6ab2..fa0450043d 100644
--- a/crates/nu-command/src/generators/generate.rs
+++ b/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;
diff --git a/crates/nu-command/src/help/help_.rs b/crates/nu-command/src/help/help_.rs
index 47f268c5c2..730afae267 100644
--- a/crates/nu-command/src/help/help_.rs
+++ b/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);
         }
     }
 
diff --git a/crates/nu-command/src/math/utils.rs b/crates/nu-command/src/math/utils.rs
index 82ecf2f50b..9abcce06e9 100644
--- a/crates/nu-command/src/math/utils.rs
+++ b/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
diff --git a/crates/nu-command/src/network/http/client.rs b/crates/nu-command/src/network/http/client.rs
index 61ada986c7..dc5adbe0cd 100644
--- a/crates/nu-command/src/network/http/client.rs
+++ b/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()?))
             }
 
diff --git a/crates/nu-command/src/network/url/join.rs b/crates/nu-command/src/network/url/join.rs
index 4d73bcb00f..eaa157ff2e 100644
--- a/crates/nu-command/src/network/url/join.rs
+++ b/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}")),
diff --git a/crates/nu-command/src/viewers/griddle.rs b/crates/nu-command/src/viewers/griddle.rs
index be4f4be49f..afeb6ee0c4 100644
--- a/crates/nu-command/src/viewers/griddle.rs
+++ b/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)))
                 }
 
diff --git a/crates/nu-command/src/viewers/table.rs b/crates/nu-command/src/viewers/table.rs
index e831b151f1..b032637275 100644
--- a/crates/nu-command/src/viewers/table.rs
+++ b/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;
diff --git a/crates/nu-explore/src/nu_common/table.rs b/crates/nu-explore/src/nu_common/table.rs
index b1d0a4c974..cb580e404b 100644
--- a/crates/nu-explore/src/nu_common/table.rs
+++ b/crates/nu-explore/src/nu_common/table.rs
@@ -16,7 +16,7 @@ pub fn try_build_table(
     let span = value.span();
     match value {
         Value::List { vals, .. } => try_build_list(vals, ctrlc, config, span, style_computer),
-        Value::Record { val, .. } => try_build_map(*val, span, style_computer, ctrlc, config),
+        Value::Record { val, .. } => try_build_map(&val, span, style_computer, ctrlc, config),
         val if matches!(val, Value::String { .. }) => {
             nu_value_to_string_clean(&val, config, style_computer).0
         }
@@ -25,7 +25,7 @@ pub fn try_build_table(
 }
 
 fn try_build_map(
-    record: Record,
+    record: &Record,
     span: Span,
     style_computer: &StyleComputer,
     ctrlc: Option<Arc<AtomicBool>>,
@@ -42,11 +42,11 @@ fn try_build_map(
         0,
         false,
     );
-    let result = ExpandedTable::new(None, false, String::new()).build_map(&record, opts);
+    let result = ExpandedTable::new(None, false, String::new()).build_map(record, opts);
     match result {
         Ok(Some(result)) => result,
         Ok(None) | Err(_) => {
-            nu_value_to_string(&Value::record(record, span), config, style_computer).0
+            nu_value_to_string(&Value::record(record.clone(), span), config, style_computer).0
         }
     }
 }
diff --git a/crates/nu-explore/src/nu_common/value.rs b/crates/nu-explore/src/nu_common/value.rs
index 5f25a1131e..5c63c9e780 100644
--- a/crates/nu-explore/src/nu_common/value.rs
+++ b/crates/nu-explore/src/nu_common/value.rs
@@ -87,7 +87,7 @@ pub fn collect_input(value: Value) -> (Vec<String>, Vec<Vec<Value>>) {
     let span = value.span();
     match value {
         Value::Record { val: record, .. } => {
-            let (key, val) = record.into_iter().unzip();
+            let (key, val) = record.into_owned().into_iter().unzip();
             (key, vec![val])
         }
         Value::List { vals, .. } => {
diff --git a/crates/nu-explore/src/pager/mod.rs b/crates/nu-explore/src/pager/mod.rs
index 5e78dba14a..67aac1f312 100644
--- a/crates/nu-explore/src/pager/mod.rs
+++ b/crates/nu-explore/src/pager/mod.rs
@@ -1038,7 +1038,7 @@ fn set_config(hm: &mut HashMap<String, Value>, path: &[&str], value: Value) -> b
             if path.len() == 2 {
                 let key = path[1];
 
-                record.insert(key, value);
+                record.to_mut().insert(key, value);
             } else {
                 let mut hm2: HashMap<String, Value> = HashMap::new();
                 for (k, v) in record.iter() {
diff --git a/crates/nu-plugin-test-support/src/plugin_test.rs b/crates/nu-plugin-test-support/src/plugin_test.rs
index 3f9326a33b..cc4650b4ae 100644
--- a/crates/nu-plugin-test-support/src/plugin_test.rs
+++ b/crates/nu-plugin-test-support/src/plugin_test.rs
@@ -319,8 +319,8 @@ impl PluginTest {
                 // reorder cols and vals to make more logically compare.
                 // more general, if two record have same col and values,
                 // the order of cols shouldn't affect the equal property.
-                let mut a_rec = a_rec.clone();
-                let mut b_rec = b_rec.clone();
+                let mut a_rec = a_rec.clone().into_owned();
+                let mut b_rec = b_rec.clone().into_owned();
                 a_rec.sort_cols();
                 b_rec.sort_cols();
 
diff --git a/crates/nu-plugin/Cargo.toml b/crates/nu-plugin/Cargo.toml
index a085937f01..0e2f755f31 100644
--- a/crates/nu-plugin/Cargo.toml
+++ b/crates/nu-plugin/Cargo.toml
@@ -13,6 +13,7 @@ bench = false
 [dependencies]
 nu-engine = { path = "../nu-engine", version = "0.92.3" }
 nu-protocol = { path = "../nu-protocol", version = "0.92.3" }
+nu-utils = { path = "../nu-utils", version = "0.92.3" }
 
 bincode = "1.3"
 rmp-serde = "1.1"
diff --git a/crates/nu-plugin/src/protocol/plugin_custom_value.rs b/crates/nu-plugin/src/protocol/plugin_custom_value.rs
index 1b3316675a..c7da73bffe 100644
--- a/crates/nu-plugin/src/protocol/plugin_custom_value.rs
+++ b/crates/nu-plugin/src/protocol/plugin_custom_value.rs
@@ -1,10 +1,13 @@
-use std::{cmp::Ordering, sync::Arc};
+use std::cmp::Ordering;
+use std::sync::Arc;
 
 use crate::{
     plugin::{PluginInterface, PluginSource},
     util::with_custom_values_in,
 };
 use nu_protocol::{ast::Operator, CustomValue, IntoSpanned, ShellError, Span, Spanned, Value};
+use nu_utils::SharedCow;
+
 use serde::{Deserialize, Serialize};
 
 #[cfg(test)]
@@ -28,7 +31,7 @@ mod tests;
 #[doc(hidden)]
 pub struct PluginCustomValue {
     #[serde(flatten)]
-    shared: SerdeArc<SharedContent>,
+    shared: SharedCow<SharedContent>,
 
     /// Which plugin the custom value came from. This is not defined on the plugin side. The engine
     /// side is responsible for maintaining it, and it is not sent over the serialization boundary.
@@ -152,11 +155,11 @@ impl PluginCustomValue {
         source: Option<Arc<PluginSource>>,
     ) -> PluginCustomValue {
         PluginCustomValue {
-            shared: SerdeArc(Arc::new(SharedContent {
+            shared: SharedCow::new(SharedContent {
                 name,
                 data,
                 notify_on_drop,
-            })),
+            }),
             source,
         }
     }
@@ -379,7 +382,8 @@ impl Drop for PluginCustomValue {
     fn drop(&mut self) {
         // If the custom value specifies notify_on_drop and this is the last copy, we need to let
         // the plugin know about it if we can.
-        if self.source.is_some() && self.notify_on_drop() && Arc::strong_count(&self.shared) == 1 {
+        if self.source.is_some() && self.notify_on_drop() && SharedCow::ref_count(&self.shared) == 1
+        {
             self.get_plugin(None, "drop")
                 // While notifying drop, we don't need a copy of the source
                 .and_then(|plugin| {
@@ -396,40 +400,3 @@ impl Drop for PluginCustomValue {
         }
     }
 }
-
-/// A serializable `Arc`, to avoid having to have the serde `rc` feature enabled.
-#[derive(Clone, Debug)]
-#[repr(transparent)]
-struct SerdeArc<T>(Arc<T>);
-
-impl<T> Serialize for SerdeArc<T>
-where
-    T: Serialize,
-{
-    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
-    where
-        S: serde::Serializer,
-    {
-        self.0.serialize(serializer)
-    }
-}
-
-impl<'de, T> Deserialize<'de> for SerdeArc<T>
-where
-    T: Deserialize<'de>,
-{
-    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
-    where
-        D: serde::Deserializer<'de>,
-    {
-        T::deserialize(deserializer).map(Arc::new).map(SerdeArc)
-    }
-}
-
-impl<T> std::ops::Deref for SerdeArc<T> {
-    type Target = Arc<T>;
-
-    fn deref(&self) -> &Self::Target {
-        &self.0
-    }
-}
diff --git a/crates/nu-protocol/src/config/mod.rs b/crates/nu-protocol/src/config/mod.rs
index 59c24fac0b..f7c03a521b 100644
--- a/crates/nu-protocol/src/config/mod.rs
+++ b/crates/nu-protocol/src/config/mod.rs
@@ -201,13 +201,13 @@ impl Value {
         // the `2`.
 
         if let Value::Record { val, .. } = self {
-            val.retain_mut(|key, value| {
+            val.to_mut().retain_mut(|key, value| {
                 let span = value.span();
                 match key {
                     // Grouped options
                     "ls" => {
                         if let Value::Record { val, .. } = value {
-                            val.retain_mut(|key2, value| {
+                            val.to_mut().retain_mut(|key2, value| {
                                 let span = value.span();
                                 match key2 {
                                     "use_ls_colors" => {
@@ -236,7 +236,7 @@ impl Value {
                     }
                     "rm" => {
                         if let Value::Record { val, .. } = value {
-                            val.retain_mut(|key2, value| {
+                            val.to_mut().retain_mut(|key2, value| {
                                 let span = value.span();
                                 match key2 {
                                     "always_trash" => {
@@ -263,7 +263,7 @@ impl Value {
                     "history" => {
                         let history = &mut config.history;
                         if let Value::Record { val, .. } = value {
-                            val.retain_mut(|key2, value| {
+                            val.to_mut().retain_mut(|key2, value| {
                                 let span = value.span();
                                 match key2 {
                                     "isolation" => {
@@ -305,7 +305,7 @@ impl Value {
                     }
                     "completions" => {
                         if let Value::Record { val, .. } = value {
-                            val.retain_mut(|key2, value| {
+                            val.to_mut().retain_mut(|key2, value| {
                                 let span = value.span();
                                 match key2 {
                                     "quick" => {
@@ -326,7 +326,7 @@ impl Value {
                                     }
                                     "external" => {
                                         if let Value::Record { val, .. } = value {
-                                            val.retain_mut(|key3, value|
+                                            val.to_mut().retain_mut(|key3, value|
                                                 {
                                                     let span = value.span();
                                                     match key3 {
@@ -393,7 +393,7 @@ impl Value {
                     }
                     "cursor_shape" => {
                         if let Value::Record { val, .. } = value {
-                            val.retain_mut(|key2, value| {
+                            val.to_mut().retain_mut(|key2, value| {
                                 let span = value.span();
                                 let config_point = match key2 {
                                     "vi_insert" => &mut config.cursor_shape_vi_insert,
@@ -426,7 +426,7 @@ impl Value {
                     }
                     "table" => {
                         if let Value::Record { val, .. } = value {
-                            val.retain_mut(|key2, value| {
+                            val.to_mut().retain_mut(|key2, value| {
                                 let span = value.span();
                                 match key2 {
                                     "mode" => {
@@ -451,7 +451,7 @@ impl Value {
                                         }
                                         Value::Record { val, .. } => {
                                             let mut invalid = false;
-                                            val.retain(|key3, value| {
+                                            val.to_mut().retain(|key3, value| {
                                                 match key3 {
                                                     "left" => {
                                                         match value.as_int() {
@@ -546,7 +546,7 @@ impl Value {
                     }
                     "filesize" => {
                         if let Value::Record { val, .. } = value {
-                            val.retain_mut(|key2, value| {
+                            val.to_mut().retain_mut(|key2, value| {
                                 let span = value.span();
                                 match key2 {
                                 "metric" => {
@@ -719,7 +719,7 @@ impl Value {
                     },
                     "datetime_format" => {
                         if let Value::Record { val, .. } = value {
-                            val.retain_mut(|key2, value|
+                            val.to_mut().retain_mut(|key2, value|
                                 {
                                 let span = value.span();
                                 match key2 {
diff --git a/crates/nu-protocol/src/config/plugin_gc.rs b/crates/nu-protocol/src/config/plugin_gc.rs
index 4d859299d2..70beeb6ae3 100644
--- a/crates/nu-protocol/src/config/plugin_gc.rs
+++ b/crates/nu-protocol/src/config/plugin_gc.rs
@@ -39,7 +39,7 @@ impl PluginGcConfigs {
                 self.plugins = HashMap::new();
             }
 
-            val.retain_mut(|key, value| {
+            val.to_mut().retain_mut(|key, value| {
                 let span = value.span();
                 match key {
                     "default" => {
@@ -88,7 +88,7 @@ fn process_plugins(
         // Remove any plugin configs that aren't in the value
         plugins.retain(|key, _| val.contains(key));
 
-        val.retain_mut(|key, value| {
+        val.to_mut().retain_mut(|key, value| {
             if matches!(value, Value::Record { .. }) {
                 plugins.entry(key.to_owned()).or_default().process(
                     &join_path(path, &[key]),
@@ -150,7 +150,7 @@ impl PluginGcConfig {
                 self.stop_after = PluginGcConfig::default().stop_after;
             }
 
-            val.retain_mut(|key, value| {
+            val.to_mut().retain_mut(|key, value| {
                 let span = value.span();
                 match key {
                     "enabled" => process_bool_config(value, errors, &mut self.enabled),
diff --git a/crates/nu-protocol/src/eval_base.rs b/crates/nu-protocol/src/eval_base.rs
index 5b8dc51312..7c240c0b1f 100644
--- a/crates/nu-protocol/src/eval_base.rs
+++ b/crates/nu-protocol/src/eval_base.rs
@@ -75,7 +75,7 @@ pub trait Eval {
                         RecordItem::Spread(_, inner) => {
                             match Self::eval::<D>(state, mut_state, inner)? {
                                 Value::Record { val: inner_val, .. } => {
-                                    for (col_name, val) in *inner_val {
+                                    for (col_name, val) in inner_val.into_owned() {
                                         if let Some(orig_span) = col_names.get(&col_name) {
                                             return Err(ShellError::ColumnDefinedTwice {
                                                 col_name,
diff --git a/crates/nu-protocol/src/value/from_value.rs b/crates/nu-protocol/src/value/from_value.rs
index 4e72814f86..17d325feea 100644
--- a/crates/nu-protocol/src/value/from_value.rs
+++ b/crates/nu-protocol/src/value/from_value.rs
@@ -538,7 +538,7 @@ impl FromValue for Vec<Value> {
 impl FromValue for Record {
     fn from_value(v: Value) -> Result<Self, ShellError> {
         match v {
-            Value::Record { val, .. } => Ok(*val),
+            Value::Record { val, .. } => Ok(val.into_owned()),
             v => Err(ShellError::CantConvert {
                 to_type: "Record".into(),
                 from_type: v.get_type().to_string(),
diff --git a/crates/nu-protocol/src/value/mod.rs b/crates/nu-protocol/src/value/mod.rs
index d747b88400..bd6a6018d0 100644
--- a/crates/nu-protocol/src/value/mod.rs
+++ b/crates/nu-protocol/src/value/mod.rs
@@ -29,7 +29,7 @@ use fancy_regex::Regex;
 use nu_utils::{
     contains_emoji,
     locale::{get_system_locale_string, LOCALE_OVERRIDE_ENV_VAR},
-    IgnoreCaseExt,
+    IgnoreCaseExt, SharedCow,
 };
 use serde::{Deserialize, Serialize};
 use std::{
@@ -110,7 +110,7 @@ pub enum Value {
         internal_span: Span,
     },
     Record {
-        val: Box<Record>,
+        val: SharedCow<Record>,
         // note: spans are being refactored out of Value
         // please use .span() instead of matching this span value
         #[serde(rename = "span")]
@@ -538,7 +538,7 @@ impl Value {
     /// Unwraps the inner [`Record`] value or returns an error if this `Value` is not a record
     pub fn into_record(self) -> Result<Record, ShellError> {
         if let Value::Record { val, .. } = self {
-            Ok(*val)
+            Ok(val.into_owned())
         } else {
             self.cant_convert_to("record")
         }
@@ -1159,7 +1159,7 @@ impl Value {
                     match current {
                         Value::Record { mut val, .. } => {
                             // Make reverse iterate to avoid duplicate column leads to first value, actually last value is expected.
-                            if let Some(found) = val.iter_mut().rev().find(|x| {
+                            if let Some(found) = val.to_mut().iter_mut().rev().find(|x| {
                                 if insensitive {
                                     x.0.eq_ignore_case(column_name)
                                 } else {
@@ -1220,13 +1220,15 @@ impl Value {
                                     let val_span = val.span();
                                     match val {
                                         Value::Record { mut val, .. } => {
-                                            if let Some(found) = val.iter_mut().rev().find(|x| {
-                                                if insensitive {
-                                                    x.0.eq_ignore_case(column_name)
-                                                } else {
-                                                    x.0 == column_name
-                                                }
-                                            }) {
+                                            if let Some(found) =
+                                                val.to_mut().iter_mut().rev().find(|x| {
+                                                    if insensitive {
+                                                        x.0.eq_ignore_case(column_name)
+                                                    } else {
+                                                        x.0 == column_name
+                                                    }
+                                                })
+                                            {
                                                 Ok(std::mem::take(found.1))
                                             } else if *optional {
                                                 Ok(Value::nothing(*origin_span))
@@ -1332,7 +1334,7 @@ impl Value {
                         for val in vals.iter_mut() {
                             match val {
                                 Value::Record { val: record, .. } => {
-                                    if let Some(val) = record.get_mut(col_name) {
+                                    if let Some(val) = record.to_mut().get_mut(col_name) {
                                         val.upsert_data_at_cell_path(path, new_val.clone())?;
                                     } else {
                                         let new_col = if path.is_empty() {
@@ -1344,7 +1346,7 @@ impl Value {
                                                 .upsert_data_at_cell_path(path, new_val.clone())?;
                                             new_col
                                         };
-                                        record.push(col_name, new_col);
+                                        record.to_mut().push(col_name, new_col);
                                     }
                                 }
                                 Value::Error { error, .. } => return Err(*error.clone()),
@@ -1359,7 +1361,7 @@ impl Value {
                         }
                     }
                     Value::Record { val: record, .. } => {
-                        if let Some(val) = record.get_mut(col_name) {
+                        if let Some(val) = record.to_mut().get_mut(col_name) {
                             val.upsert_data_at_cell_path(path, new_val)?;
                         } else {
                             let new_col = if path.is_empty() {
@@ -1369,7 +1371,7 @@ impl Value {
                                 new_col.upsert_data_at_cell_path(path, new_val)?;
                                 new_col
                             };
-                            record.push(col_name, new_col);
+                            record.to_mut().push(col_name, new_col);
                         }
                     }
                     Value::LazyRecord { val, .. } => {
@@ -1456,7 +1458,7 @@ impl Value {
                             let v_span = val.span();
                             match val {
                                 Value::Record { val: record, .. } => {
-                                    if let Some(val) = record.get_mut(col_name) {
+                                    if let Some(val) = record.to_mut().get_mut(col_name) {
                                         val.update_data_at_cell_path(path, new_val.clone())?;
                                     } else {
                                         return Err(ShellError::CantFindColumn {
@@ -1478,7 +1480,7 @@ impl Value {
                         }
                     }
                     Value::Record { val: record, .. } => {
-                        if let Some(val) = record.get_mut(col_name) {
+                        if let Some(val) = record.to_mut().get_mut(col_name) {
                             val.update_data_at_cell_path(path, new_val)?;
                         } else {
                             return Err(ShellError::CantFindColumn {
@@ -1548,7 +1550,7 @@ impl Value {
                                 let v_span = val.span();
                                 match val {
                                     Value::Record { val: record, .. } => {
-                                        if record.remove(col_name).is_none() && !optional {
+                                        if record.to_mut().remove(col_name).is_none() && !optional {
                                             return Err(ShellError::CantFindColumn {
                                                 col_name: col_name.clone(),
                                                 span: *span,
@@ -1568,7 +1570,7 @@ impl Value {
                             Ok(())
                         }
                         Value::Record { val: record, .. } => {
-                            if record.remove(col_name).is_none() && !optional {
+                            if record.to_mut().remove(col_name).is_none() && !optional {
                                 return Err(ShellError::CantFindColumn {
                                     col_name: col_name.clone(),
                                     span: *span,
@@ -1628,7 +1630,7 @@ impl Value {
                                 let v_span = val.span();
                                 match val {
                                     Value::Record { val: record, .. } => {
-                                        if let Some(val) = record.get_mut(col_name) {
+                                        if let Some(val) = record.to_mut().get_mut(col_name) {
                                             val.remove_data_at_cell_path(path)?;
                                         } else if !optional {
                                             return Err(ShellError::CantFindColumn {
@@ -1650,7 +1652,7 @@ impl Value {
                             Ok(())
                         }
                         Value::Record { val: record, .. } => {
-                            if let Some(val) = record.get_mut(col_name) {
+                            if let Some(val) = record.to_mut().get_mut(col_name) {
                                 val.remove_data_at_cell_path(path)?;
                             } else if !optional {
                                 return Err(ShellError::CantFindColumn {
@@ -1720,7 +1722,7 @@ impl Value {
                             let v_span = val.span();
                             match val {
                                 Value::Record { val: record, .. } => {
-                                    if let Some(val) = record.get_mut(col_name) {
+                                    if let Some(val) = record.to_mut().get_mut(col_name) {
                                         if path.is_empty() {
                                             return Err(ShellError::ColumnAlreadyExists {
                                                 col_name: col_name.clone(),
@@ -1747,7 +1749,7 @@ impl Value {
                                             )?;
                                             new_col
                                         };
-                                        record.push(col_name, new_col);
+                                        record.to_mut().push(col_name, new_col);
                                     }
                                 }
                                 Value::Error { error, .. } => return Err(*error.clone()),
@@ -1763,7 +1765,7 @@ impl Value {
                         }
                     }
                     Value::Record { val: record, .. } => {
-                        if let Some(val) = record.get_mut(col_name) {
+                        if let Some(val) = record.to_mut().get_mut(col_name) {
                             if path.is_empty() {
                                 return Err(ShellError::ColumnAlreadyExists {
                                     col_name: col_name.clone(),
@@ -1781,7 +1783,7 @@ impl Value {
                                 new_col.insert_data_at_cell_path(path, new_val, head_span)?;
                                 new_col
                             };
-                            record.push(col_name, new_col);
+                            record.to_mut().push(col_name, new_col);
                         }
                     }
                     Value::LazyRecord { val, .. } => {
@@ -1854,6 +1856,7 @@ impl Value {
         // Check for contained values
         match self {
             Value::Record { ref mut val, .. } => val
+                .to_mut()
                 .iter_mut()
                 .try_for_each(|(_, rec_value)| rec_value.recurse_mut(f)),
             Value::List { ref mut vals, .. } => vals
@@ -1989,7 +1992,7 @@ impl Value {
 
     pub fn record(val: Record, span: Span) -> Value {
         Value::Record {
-            val: Box::new(val),
+            val: SharedCow::new(val),
             internal_span: span,
         }
     }
@@ -2432,8 +2435,8 @@ impl PartialOrd for Value {
                     // reorder cols and vals to make more logically compare.
                     // more general, if two record have same col and values,
                     // the order of cols shouldn't affect the equal property.
-                    let mut lhs = lhs.clone();
-                    let mut rhs = rhs.clone();
+                    let mut lhs = lhs.clone().into_owned();
+                    let mut rhs = rhs.clone().into_owned();
                     lhs.sort_cols();
                     rhs.sort_cols();
 
diff --git a/crates/nu-protocol/src/value/record.rs b/crates/nu-protocol/src/value/record.rs
index 50996ac2b0..6d4f3b3513 100644
--- a/crates/nu-protocol/src/value/record.rs
+++ b/crates/nu-protocol/src/value/record.rs
@@ -151,7 +151,7 @@ impl Record {
     ///
     /// fn remove_foo_recursively(val: &mut Value) {
     ///     if let Value::Record {val, ..} = val {
-    ///         val.retain_mut(keep_non_foo);
+    ///         val.to_mut().retain_mut(keep_non_foo);
     ///     }
     /// }
     ///
diff --git a/crates/nu-table/src/types/collapse.rs b/crates/nu-table/src/types/collapse.rs
index d4d43e86a9..e25b7f1e79 100644
--- a/crates/nu-table/src/types/collapse.rs
+++ b/crates/nu-table/src/types/collapse.rs
@@ -4,6 +4,7 @@ use crate::{
 };
 use nu_color_config::StyleComputer;
 use nu_protocol::{Config, Record, TableMode, Value};
+use nu_utils::SharedCow;
 
 pub struct CollapsedTable;
 
@@ -48,8 +49,9 @@ fn colorize_value(value: &mut Value, config: &Config, style_computer: &StyleComp
             // Take ownership of the record and reassign to &mut
             // We do this to have owned keys through `.into_iter`
             let record = std::mem::take(val);
-            *val = Box::new(
+            *val = SharedCow::new(
                 record
+                    .into_owned()
                     .into_iter()
                     .map(|(mut header, mut val)| {
                         colorize_value(&mut val, config, style_computer);
diff --git a/crates/nu-table/src/unstructured_table.rs b/crates/nu-table/src/unstructured_table.rs
index 58ea117791..5ab0bec3f6 100644
--- a/crates/nu-table/src/unstructured_table.rs
+++ b/crates/nu-table/src/unstructured_table.rs
@@ -89,7 +89,7 @@ fn build_table(
 
 fn convert_nu_value_to_table_value(value: Value, config: &Config) -> TableValue {
     match value {
-        Value::Record { val, .. } => build_vertical_map(*val, config),
+        Value::Record { val, .. } => build_vertical_map(val.into_owned(), config),
         Value::List { vals, .. } => {
             let rebuild_array_as_map = is_valid_record(&vals) && count_columns_in_record(&vals) > 0;
             if rebuild_array_as_map {
@@ -195,7 +195,8 @@ fn build_map_from_record(vals: Vec<Value>, config: &Config) -> TableValue {
     for val in vals {
         match val {
             Value::Record { val, .. } => {
-                for (i, (_key, val)) in val.into_iter().take(count_columns).enumerate() {
+                for (i, (_key, val)) in val.into_owned().into_iter().take(count_columns).enumerate()
+                {
                     let cell = convert_nu_value_to_table_value(val, config);
                     list[i].push(cell);
                 }
diff --git a/crates/nu-utils/Cargo.toml b/crates/nu-utils/Cargo.toml
index fd13b8119a..e659fd6d47 100644
--- a/crates/nu-utils/Cargo.toml
+++ b/crates/nu-utils/Cargo.toml
@@ -21,6 +21,7 @@ lscolors = { workspace = true, default-features = false, features = ["nu-ansi-te
 log = { workspace = true }
 num-format = { workspace = true }
 strip-ansi-escapes = { workspace = true }
+serde = { workspace = true }
 sys-locale = "0.3"
 unicase = "2.7.0"
 
diff --git a/crates/nu-utils/src/lib.rs b/crates/nu-utils/src/lib.rs
index 00dcf36454..f1a915290b 100644
--- a/crates/nu-utils/src/lib.rs
+++ b/crates/nu-utils/src/lib.rs
@@ -4,6 +4,7 @@ mod deansi;
 pub mod emoji;
 pub mod filesystem;
 pub mod locale;
+mod shared_cow;
 pub mod utils;
 
 pub use locale::get_system_locale;
@@ -17,6 +18,7 @@ pub use deansi::{
     strip_ansi_likely, strip_ansi_string_likely, strip_ansi_string_unlikely, strip_ansi_unlikely,
 };
 pub use emoji::contains_emoji;
+pub use shared_cow::SharedCow;
 
 #[cfg(unix)]
 pub use filesystem::users;
diff --git a/crates/nu-utils/src/shared_cow.rs b/crates/nu-utils/src/shared_cow.rs
new file mode 100644
index 0000000000..535b1648cd
--- /dev/null
+++ b/crates/nu-utils/src/shared_cow.rs
@@ -0,0 +1,113 @@
+use serde::{Deserialize, Serialize};
+use std::{fmt, ops, sync::Arc};
+
+/// A container that transparently shares a value when possible, but clones on mutate.
+///
+/// Unlike `Arc`, this is only intended to help save memory usage and reduce the amount of effort
+/// required to clone unmodified values with easy to use copy-on-write.
+///
+/// This should more or less reflect the API of [`std::borrow::Cow`] as much as is sensible.
+#[derive(Clone, PartialEq, Eq, PartialOrd, Ord, Hash, Default)]
+#[repr(transparent)]
+pub struct SharedCow<T: Clone>(Arc<T>);
+
+impl<T: Clone> SharedCow<T> {
+    /// Create a new `Shared` value.
+    pub fn new(value: T) -> SharedCow<T> {
+        SharedCow(Arc::new(value))
+    }
+
+    /// Take an exclusive clone of the shared value, or move and take ownership if it wasn't shared.
+    pub fn into_owned(self: SharedCow<T>) -> T {
+        // Optimized: if the Arc is not shared, just unwraps the Arc
+        match Arc::try_unwrap(self.0) {
+            Ok(value) => value,
+            Err(arc) => (*arc).clone(),
+        }
+    }
+
+    /// Get a mutable reference to the value inside the [`SharedCow`]. This will result in a clone
+    /// being created only if the value was shared with multiple references.
+    pub fn to_mut(&mut self) -> &mut T {
+        Arc::make_mut(&mut self.0)
+    }
+
+    /// Convert the `Shared` value into an `Arc`
+    pub fn into_arc(value: SharedCow<T>) -> Arc<T> {
+        value.0
+    }
+
+    /// Return the number of references to the shared value.
+    pub fn ref_count(value: &SharedCow<T>) -> usize {
+        Arc::strong_count(&value.0)
+    }
+}
+
+impl<T> From<T> for SharedCow<T>
+where
+    T: Clone,
+{
+    fn from(value: T) -> Self {
+        SharedCow::new(value)
+    }
+}
+
+impl<T> From<Arc<T>> for SharedCow<T>
+where
+    T: Clone,
+{
+    fn from(value: Arc<T>) -> Self {
+        SharedCow(value)
+    }
+}
+
+impl<T> fmt::Debug for SharedCow<T>
+where
+    T: fmt::Debug + Clone,
+{
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        // Appears transparent
+        (*self.0).fmt(f)
+    }
+}
+
+impl<T> fmt::Display for SharedCow<T>
+where
+    T: fmt::Display + Clone,
+{
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        (*self.0).fmt(f)
+    }
+}
+
+impl<T: Clone> Serialize for SharedCow<T>
+where
+    T: Serialize,
+{
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        self.0.serialize(serializer)
+    }
+}
+
+impl<'de, T: Clone> Deserialize<'de> for SharedCow<T>
+where
+    T: Deserialize<'de>,
+{
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        T::deserialize(deserializer).map(Arc::new).map(SharedCow)
+    }
+}
+
+impl<T: Clone> ops::Deref for SharedCow<T> {
+    type Target = T;
+
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
diff --git a/crates/nu_plugin_polars/src/dataframe/values/nu_dataframe/mod.rs b/crates/nu_plugin_polars/src/dataframe/values/nu_dataframe/mod.rs
index 209acf594c..8bc3ae56dd 100644
--- a/crates/nu_plugin_polars/src/dataframe/values/nu_dataframe/mod.rs
+++ b/crates/nu_plugin_polars/src/dataframe/values/nu_dataframe/mod.rs
@@ -174,9 +174,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)?