Add per-field UpdateOp support for advanced database operations

proto/sf/substreams/sink/database/v1/database.proto

@@ -41,4 +41,16 @@
   string name = 1;
   string new_value = 2;
   string old_value = 3;
+
+  // UpdateOp specifies how the field value should be applied during UPSERT.
+  // Only used when TableChange.operation is OPERATION_UPSERT.
+  // Defaults to UPDATE_OP_SET if not specified.
+  enum UpdateOp {
+    UPDATE_OP_SET = 0;          // SET: column = new_value (default)
+    UPDATE_OP_ADD = 1;          // ADD: column = COALESCE(column, 0) + new_value
+    UPDATE_OP_MAX = 2;          // MAX: column = GREATEST(COALESCE(column, new_value), new_value)
+    UPDATE_OP_MIN = 3;          // MIN: column = LEAST(COALESCE(column, new_value), new_value)
+    UPDATE_OP_SET_IF_NULL = 4;  // SET_IF_NULL: column = COALESCE(column, new_value)
+  }
+  UpdateOp update_op = 4;
 }

src/change.rs

@@ -1,3 +1,4 @@
+use crate::pb::database::field::UpdateOp;
 use crate::pb::database::Field;
 use std::str;
 use substreams::pb::substreams::store_delta::Operation;
@@ -76,6 +77,7 @@ impl<T: AsString> ToField for (T, T) {
             name: name.as_ref().to_string(),
             old_value: self.0.as_string(),
             new_value: self.1.as_string(),
+            update_op: UpdateOp::Set as i32,
         }
     }
 }
@@ -88,6 +90,7 @@ impl<T: AsString> ToField for (Option<T>, T) {
                 name: name.as_ref().to_string(),
                 old_value: "".to_string(),
                 new_value: new.as_string(),
+                update_op: UpdateOp::Set as i32,
             },
         }
     }
@@ -101,6 +104,7 @@ impl<T: AsString> ToField for (T, Option<T>) {
                 name: name.as_ref().to_string(),
                 old_value: old.as_string(),
                 new_value: "".to_string(),
+                update_op: UpdateOp::Set as i32,
             },
         }
     }

src/pb/sf.substreams.sink.database.v1.rs

@@ -87,5 +87,53 @@ pub struct Field {
     pub new_value: ::prost::alloc::string::String,
     #[prost(string, tag="3")]
     pub old_value: ::prost::alloc::string::String,
+    #[prost(enumeration="field::UpdateOp", tag="4")]
+    pub update_op: i32,
+}
+/// Nested message and enum types in `Field`.
+pub mod field {
+    /// UpdateOp specifies how the field value should be applied during UPSERT.
+    /// Only used when TableChange.operation is OPERATION_UPSERT.
+    /// Defaults to UPDATE_OP_SET if not specified.
+    #[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord, ::prost::Enumeration)]
+    #[repr(i32)]
+    pub enum UpdateOp {
+        /// SET: column = new_value (default)
+        Set = 0,
+        /// ADD: column = COALESCE(column, 0) + new_value
+        Add = 1,
+        /// MAX: column = GREATEST(COALESCE(column, new_value), new_value)
+        Max = 2,
+        /// MIN: column = LEAST(COALESCE(column, new_value), new_value)
+        Min = 3,
+        /// SET_IF_NULL: column = COALESCE(column, new_value)
+        SetIfNull = 4,
+    }
+    impl UpdateOp {
+        /// String value of the enum field names used in the ProtoBuf definition.
+        ///
+        /// The values are not transformed in any way and thus are considered stable
+        /// (if the ProtoBuf definition does not change) and safe for programmatic use.
+        pub fn as_str_name(&self) -> &'static str {
+            match self {
+                UpdateOp::Set => "UPDATE_OP_SET",
+                UpdateOp::Add => "UPDATE_OP_ADD",
+                UpdateOp::Max => "UPDATE_OP_MAX",
+                UpdateOp::Min => "UPDATE_OP_MIN",
+                UpdateOp::SetIfNull => "UPDATE_OP_SET_IF_NULL",
+            }
+        }
+        /// Creates an enum from field names used in the ProtoBuf definition.
+        pub fn from_str_name(value: &str) -> ::core::option::Option<Self> {
+            match value {
+                "UPDATE_OP_SET" => Some(Self::Set),
+                "UPDATE_OP_ADD" => Some(Self::Add),
+                "UPDATE_OP_MAX" => Some(Self::Max),
+                "UPDATE_OP_MIN" => Some(Self::Min),
+                "UPDATE_OP_SET_IF_NULL" => Some(Self::SetIfNull),
+                _ => None,
+            }
+        }
+    }
 }
 // @@protoc_insertion_point(module)