Support for all custom value operations on plugin custom values (#12088)

# Description

Adds support for the following operations on plugin custom values, in
addition to `to_base_value` which was already present:

- `follow_path_int()`
- `follow_path_string()`
- `partial_cmp()`
- `operation()`
- `Drop` (notification, if opted into with
`CustomValue::notify_plugin_on_drop`)

There are additionally customizable methods within the `Plugin` and
`StreamingPlugin` traits for implementing these functions in a way that
requires access to the plugin state, as a registered handle model such
as might be used in a dataframes plugin would.

`Value::append` was also changed to handle custom values correctly.

# User-Facing Changes

- Signature of `CustomValue::follow_path_string` and
`CustomValue::follow_path_int` changed to give access to the span of the
custom value itself, useful for some errors.
- Plugins using custom values have to be recompiled because the engine
will try to do custom value operations that aren't supported
- Plugins can do more things 🎉 

# Tests + Formatting
Tests were added for all of the new custom values functionality.

- 🟢 `toolkit fmt`
- 🟢 `toolkit clippy`
- 🟢 `toolkit test`
- 🟢 `toolkit test stdlib`

# After Submitting
- [ ] Document protocol reference `CustomValueOp` variants:
  - [ ] `FollowPathInt`
  - [ ] `FollowPathString`
  - [ ] `PartialCmp`
  - [ ] `Operation`
  - [ ] `Dropped`
- [ ] Document `notify_on_drop` optional field in `PluginCustomValue`

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

@@ -34,13 +34,23 @@ impl CustomValue for NuDataFrame {
         self
     }
 
-    fn follow_path_int(&self, count: usize, span: Span) -> Result<Value, ShellError> {
-        self.get_value(count, span)
+    fn follow_path_int(
+        &self,
+        _self_span: Span,
+        count: usize,
+        path_span: Span,
+    ) -> Result<Value, ShellError> {
+        self.get_value(count, path_span)
     }
 
-    fn follow_path_string(&self, column_name: String, span: Span) -> Result<Value, ShellError> {
-        let column = self.column(&column_name, span)?;
-        Ok(column.into_value(span))
+    fn follow_path_string(
+        &self,
+        _self_span: Span,
+        column_name: String,
+        path_span: Span,
+    ) -> Result<Value, ShellError> {
+        let column = self.column(&column_name, path_span)?;
+        Ok(column.into_value(path_span))
     }
 
     fn partial_cmp(&self, other: &Value) -> Option<std::cmp::Ordering> {

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

@@ -372,19 +372,29 @@ impl CustomValue for SQLiteDatabase {
         self
     }
 
-    fn follow_path_int(&self, _count: usize, span: Span) -> Result<Value, ShellError> {
+    fn follow_path_int(
+        &self,
+        _self_span: Span,
+        _index: usize,
+        path_span: Span,
+    ) -> Result<Value, ShellError> {
         // In theory we could support this, but tables don't have an especially well-defined order
-        Err(ShellError::IncompatiblePathAccess { type_name: "SQLite databases do not support integer-indexed access. Try specifying a table name instead".into(), span })
+        Err(ShellError::IncompatiblePathAccess { type_name: "SQLite databases do not support integer-indexed access. Try specifying a table name instead".into(), span: path_span })
     }
 
-    fn follow_path_string(&self, _column_name: String, span: Span) -> Result<Value, ShellError> {
-        let db = open_sqlite_db(&self.path, span)?;
+    fn follow_path_string(
+        &self,
+        _self_span: Span,
+        _column_name: String,
+        path_span: Span,
+    ) -> Result<Value, ShellError> {
+        let db = open_sqlite_db(&self.path, path_span)?;
 
-        read_single_table(db, _column_name, span, self.ctrlc.clone()).map_err(|e| {
+        read_single_table(db, _column_name, path_span, self.ctrlc.clone()).map_err(|e| {
             ShellError::GenericError {
                 error: "Failed to read from SQLite database".into(),
                 msg: e.to_string(),
-                span: Some(span),
+                span: Some(path_span),
                 help: None,
                 inner: vec![],
             }

crates/nu-plugin/Cargo.toml

@@ -16,7 +16,7 @@ nu-protocol = { path = "../nu-protocol", version = "0.91.1" }
 
 bincode = "1.3"
 rmp-serde = "1.1"
-serde = { version = "1.0" }
+serde = "1.0"
 serde_json = { workspace = true }
 log = "0.4"
 miette = { workspace = true }

crates/nu-plugin/src/plugin/interface/engine.rs

@@ -12,8 +12,8 @@ use nu_protocol::{
 
 use crate::{
     protocol::{
-        CallInfo, CustomValueOp, EngineCall, EngineCallId, EngineCallResponse, PluginCall,
-        PluginCallId, PluginCallResponse, PluginCustomValue, PluginInput, PluginOption,
+        CallInfo, CustomValueOp, EngineCall, EngineCallId, EngineCallResponse, Ordering,
+        PluginCall, PluginCallId, PluginCallResponse, PluginCustomValue, PluginInput, PluginOption,
         ProtocolInfo,
     },
     LabeledError, PluginOutput,
@@ -683,6 +683,16 @@ impl EngineInterface {
         self.write(PluginOutput::Option(PluginOption::GcDisabled(disabled)))?;
         self.flush()
     }
+
+    /// Write a call response of [`Ordering`], for `partial_cmp`.
+    pub(crate) fn write_ordering(
+        &self,
+        ordering: Option<impl Into<Ordering>>,
+    ) -> Result<(), ShellError> {
+        let response = PluginCallResponse::Ordering(ordering.map(|o| o.into()));
+        self.write(PluginOutput::CallResponse(self.context()?, response))?;
+        self.flush()
+    }
 }
 
 impl Interface for EngineInterface {

crates/nu-plugin/src/plugin/interface/engine/tests.rs

@@ -496,7 +496,7 @@ fn manager_consume_call_custom_value_op_forwards_to_receiver_with_context() -> R
             op,
         } => {
             assert_eq!(Some(32), engine.context);
-            assert_eq!("TestCustomValue", custom_value.item.name);
+            assert_eq!("TestCustomValue", custom_value.item.name());
             assert!(
                 matches!(op, CustomValueOp::ToBaseValue),
                 "incorrect op: {op:?}"
@@ -600,11 +600,12 @@ fn manager_prepare_pipeline_data_embeds_deserialization_errors_in_streams() -> R
 {
     let manager = TestCase::new().engine();
 
-    let invalid_custom_value = PluginCustomValue {
-        name: "Invalid".into(),
-        data: vec![0; 8], // should fail to decode to anything
-        source: None,
-    };
+    let invalid_custom_value = PluginCustomValue::new(
+        "Invalid".into(),
+        vec![0; 8], // should fail to decode to anything
+        false,
+        None,
+    );
 
     let span = Span::new(20, 30);
     let data = manager.prepare_pipeline_data(
@@ -965,9 +966,9 @@ fn interface_prepare_pipeline_data_serializes_custom_values() -> Result<(), Shel
         .expect("custom value is not a PluginCustomValue, probably not serialized");
 
     let expected = test_plugin_custom_value();
-    assert_eq!(expected.name, custom_value.name);
-    assert_eq!(expected.data, custom_value.data);
-    assert!(custom_value.source.is_none());
+    assert_eq!(expected.name(), custom_value.name());
+    assert_eq!(expected.data(), custom_value.data());
+    assert!(custom_value.source().is_none());
 
     Ok(())
 }
@@ -994,9 +995,9 @@ fn interface_prepare_pipeline_data_serializes_custom_values_in_streams() -> Resu
         .expect("custom value is not a PluginCustomValue, probably not serialized");
 
     let expected = test_plugin_custom_value();
-    assert_eq!(expected.name, custom_value.name);
-    assert_eq!(expected.data, custom_value.data);
-    assert!(custom_value.source.is_none());
+    assert_eq!(expected.name(), custom_value.name());
+    assert_eq!(expected.data(), custom_value.data());
+    assert!(custom_value.source().is_none());
 
     Ok(())
 }

crates/nu-plugin/src/plugin/interface/plugin.rs

@@ -6,15 +6,15 @@ use std::{
 };
 
 use nu_protocol::{
-    IntoInterruptiblePipelineData, ListStream, PipelineData, PluginSignature, ShellError, Spanned,
-    Value,
+    ast::Operator, IntoInterruptiblePipelineData, IntoSpanned, ListStream, PipelineData,
+    PluginSignature, ShellError, Span, Spanned, Value,
 };
 
 use crate::{
     plugin::{context::PluginExecutionContext, gc::PluginGc, PluginSource},
     protocol::{
-        CallInfo, CustomValueOp, EngineCall, EngineCallId, EngineCallResponse, PluginCall,
-        PluginCallId, PluginCallResponse, PluginCustomValue, PluginInput, PluginOption,
+        CallInfo, CustomValueOp, EngineCall, EngineCallId, EngineCallResponse, Ordering,
+        PluginCall, PluginCallId, PluginCallResponse, PluginCustomValue, PluginInput, PluginOption,
         PluginOutput, ProtocolInfo, StreamId, StreamMessage,
     },
     sequence::Sequence,
@@ -454,6 +454,9 @@ impl InterfaceManager for PluginInterfaceManager {
                 let response = match response {
                     PluginCallResponse::Error(err) => PluginCallResponse::Error(err),
                     PluginCallResponse::Signature(sigs) => PluginCallResponse::Signature(sigs),
+                    PluginCallResponse::Ordering(ordering) => {
+                        PluginCallResponse::Ordering(ordering)
+                    }
                     PluginCallResponse::PipelineData(data) => {
                         // If there's an error with initializing this stream, change it to a plugin
                         // error response, but send it anyway
@@ -804,22 +807,93 @@ impl PluginInterface {
         }
     }
 
-    /// Collapse a custom value to its base value.
-    pub(crate) fn custom_value_to_base_value(
+    /// Do a custom value op that expects a value response (i.e. most of them)
+    fn custom_value_op_expecting_value(
         &self,
         value: Spanned<PluginCustomValue>,
+        op: CustomValueOp,
     ) -> Result<Value, ShellError> {
+        let op_name = op.name();
         let span = value.span;
-        let call = PluginCall::CustomValueOp(value, CustomValueOp::ToBaseValue);
+        let call = PluginCall::CustomValueOp(value, op);
         match self.plugin_call(call, &None)? {
             PluginCallResponse::PipelineData(out_data) => Ok(out_data.into_value(span)),
             PluginCallResponse::Error(err) => Err(err.into()),
             _ => Err(ShellError::PluginFailedToDecode {
-                msg: "Received unexpected response to plugin CustomValueOp::ToBaseValue call"
-                    .into(),
+                msg: format!("Received unexpected response to custom value {op_name}() call"),
+            }),
+        }
+    }
+
+    /// Collapse a custom value to its base value.
+    pub(crate) fn custom_value_to_base_value(
+        &self,
+        value: Spanned<PluginCustomValue>,
+    ) -> Result<Value, ShellError> {
+        self.custom_value_op_expecting_value(value, CustomValueOp::ToBaseValue)
+    }
+
+    /// Follow a numbered cell path on a custom value - e.g. `value.0`.
+    pub(crate) fn custom_value_follow_path_int(
+        &self,
+        value: Spanned<PluginCustomValue>,
+        index: Spanned<usize>,
+    ) -> Result<Value, ShellError> {
+        self.custom_value_op_expecting_value(value, CustomValueOp::FollowPathInt(index))
+    }
+
+    /// Follow a named cell path on a custom value - e.g. `value.column`.
+    pub(crate) fn custom_value_follow_path_string(
+        &self,
+        value: Spanned<PluginCustomValue>,
+        column_name: Spanned<String>,
+    ) -> Result<Value, ShellError> {
+        self.custom_value_op_expecting_value(value, CustomValueOp::FollowPathString(column_name))
+    }
+
+    /// Invoke comparison logic for custom values.
+    pub(crate) fn custom_value_partial_cmp(
+        &self,
+        value: PluginCustomValue,
+        mut other_value: Value,
+    ) -> Result<Option<Ordering>, ShellError> {
+        PluginCustomValue::verify_source(&mut other_value, &self.state.source)?;
+        // Note: the protocol is always designed to have a span with the custom value, but this
+        // operation doesn't support one.
+        let call = PluginCall::CustomValueOp(
+            value.into_spanned(Span::unknown()),
+            CustomValueOp::PartialCmp(other_value),
+        );
+        match self.plugin_call(call, &None)? {
+            PluginCallResponse::Ordering(ordering) => Ok(ordering),
+            PluginCallResponse::Error(err) => Err(err.into()),
+            _ => Err(ShellError::PluginFailedToDecode {
+                msg: "Received unexpected response to custom value partial_cmp() call".into(),
             }),
         }
     }
+
+    /// Invoke functionality for an operator on a custom value.
+    pub(crate) fn custom_value_operation(
+        &self,
+        left: Spanned<PluginCustomValue>,
+        operator: Spanned<Operator>,
+        mut right: Value,
+    ) -> Result<Value, ShellError> {
+        PluginCustomValue::verify_source(&mut right, &self.state.source)?;
+        self.custom_value_op_expecting_value(left, CustomValueOp::Operation(operator, right))
+    }
+
+    /// Notify the plugin about a dropped custom value.
+    pub(crate) fn custom_value_dropped(&self, value: PluginCustomValue) -> Result<(), ShellError> {
+        // Note: the protocol is always designed to have a span with the custom value, but this
+        // operation doesn't support one.
+        self.custom_value_op_expecting_value(
+            value.into_spanned(Span::unknown()),
+            CustomValueOp::Dropped,
+        )
+        .map(|_| ())
+    }
 }
 
 /// Check that custom values in call arguments come from the right source

crates/nu-plugin/src/plugin/interface/plugin/tests.rs

@@ -649,7 +649,7 @@ fn manager_prepare_pipeline_data_adds_source_to_values() -> Result<(), ShellErro
         .downcast_ref()
         .expect("custom value is not a PluginCustomValue");
 
-    if let Some(source) = &custom_value.source {
+    if let Some(source) = custom_value.source() {
         assert_eq!("test", source.name());
     } else {
         panic!("source was not set");
@@ -679,7 +679,7 @@ fn manager_prepare_pipeline_data_adds_source_to_list_streams() -> Result<(), She
         .downcast_ref()
         .expect("custom value is not a PluginCustomValue");
 
-    if let Some(source) = &custom_value.source {
+    if let Some(source) = custom_value.source() {
         assert_eq!("test", source.name());
     } else {
         panic!("source was not set");
@@ -1092,12 +1092,13 @@ fn interface_custom_value_to_base_value() -> Result<(), ShellError> {
 fn normal_values(interface: &PluginInterface) -> Vec<Value> {
     vec![
         Value::test_int(5),
-        Value::test_custom_value(Box::new(PluginCustomValue {
-            name: "SomeTest".into(),
-            data: vec![1, 2, 3],
+        Value::test_custom_value(Box::new(PluginCustomValue::new(
+            "SomeTest".into(),
+            vec![1, 2, 3],
+            false,
             // Has the same source, so it should be accepted
-            source: Some(interface.state.source.clone()),
-        })),
+            Some(interface.state.source.clone()),
+        ))),
     ]
 }
 
@@ -1145,17 +1146,19 @@ fn bad_custom_values() -> Vec<Value> {
         // Native custom value (not PluginCustomValue) should be rejected
         Value::test_custom_value(Box::new(expected_test_custom_value())),
         // Has no source, so it should be rejected
-        Value::test_custom_value(Box::new(PluginCustomValue {
-            name: "SomeTest".into(),
-            data: vec![1, 2, 3],
-            source: None,
-        })),
+        Value::test_custom_value(Box::new(PluginCustomValue::new(
+            "SomeTest".into(),
+            vec![1, 2, 3],
+            false,
+            None,
+        ))),
         // Has a different source, so it should be rejected
-        Value::test_custom_value(Box::new(PluginCustomValue {
-            name: "SomeTest".into(),
-            data: vec![1, 2, 3],
-            source: Some(PluginSource::new_fake("pluto").into()),
-        })),
+        Value::test_custom_value(Box::new(PluginCustomValue::new(
+            "SomeTest".into(),
+            vec![1, 2, 3],
+            false,
+            Some(PluginSource::new_fake("pluto").into()),
+        ))),
     ]
 }