Repository: scylladb/scylla-rust-driver (rust)

Issue:
Expose a type that represents Unset (Value with -2 len)
Currently, we only support Option<T> -> None -> -1 (Null) but inserting null is the same as inserting a tombstone as far as I am aware and it's better to do unset in the event the value is not set (undefined).

node has support for automatically handling unset as undefined.

We could expose a type that is identical to Option but serializes to 0 length to support unset, or add a configuration option to make Option<T> use unset instead of null.

Please let me know how you'd like to proceed with this task.

Pull request description:
Expose a type that represents unset value
Fixes #159 by adding `struct Unset` and `enum MaybeUnset<V: Value>`.

## Pre-review checklist

<!--
    Make sure you took care of the issues on the list.
    Put 'x' into those boxes which apply.
    You can also create the PR now and click on all relevant checkboxes.
    See CONTRIBUTING.md for more details.
-->

- [x] I have split my patch into logically separate commits.
- [x] All commit messages clearly explain what they change and why.
- [x] I added relevant tests for new features and bug fixes.
- [x] All commits compile, pass static checks and pass test.
- [x] PR description sums up the changes and reasons why they should be introduced.
- [x] I added appropriate `Fixes:` annotations to PR description.


Patch:
diff --git a/scylla/src/frame/value.rs b/scylla/src/frame/value.rs
index a40e54ad..69eec3ba 100644
--- a/scylla/src/frame/value.rs
+++ b/scylla/src/frame/value.rs
@@ -14,6 +14,15 @@ pub trait Value {
 #[error("Value too big to be sent in a request - max 2GiB allowed")]
 pub struct ValueTooBig;
 
+/// Represents an unset value
+pub struct Unset;
+
+/// Enum providing a way to represent a value that might be unset
+pub enum MaybeUnset<V: Value> {
+    Unset,
+    Set(V),
+}
+
 /// Keeps a buffer with serialized Values
 /// Allows adding new Values and iterating over serialized ones
 #[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord)]
@@ -218,6 +227,23 @@ impl<T: Value> Value for Option<T> {
     }
 }
 
+impl Value for Unset {
+    fn serialize(&self, buf: &mut Vec<u8>) -> Result<(), ValueTooBig> {
+        // Unset serializes itself to empty value with length = -2
+        buf.put_i32(-2);
+        Ok(())
+    }
+}
+
+impl<V: Value> Value for MaybeUnset<V> {
+    fn serialize(&self, buf: &mut Vec<u8>) -> Result<(), ValueTooBig> {
+        match self {
+            MaybeUnset::Set(v) => v.serialize(buf),
+            MaybeUnset::Unset => Unset.serialize(buf),
+        }
+    }
+}
+
 // Every &impl Value should also implement Value
 impl<T: Value> Value for &T {
     fn serialize(&self, buf: &mut Vec<u8>) -> Result<(), ValueTooBig> {
@@ -294,9 +320,9 @@ impl_value_for_tuple!(T0, T1, T2, T3, T4, T5; 0, 1, 2, 3, 4, 5);
 impl_value_for_tuple!(T0, T1, T2, T3, T4, T5, T6; 0, 1, 2, 3, 4, 5, 6);
 impl_value_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7; 0, 1, 2, 3, 4, 5, 6, 7);
 impl_value_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8; 0, 1, 2, 3, 4, 5, 6, 7, 8);
-impl_value_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8, T9; 
+impl_value_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8, T9;
                            0, 1, 2, 3, 4, 5, 6, 7, 8, 9);
-impl_value_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10; 
+impl_value_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10;
                            0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
 impl_value_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11;
                            0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11);
@@ -388,9 +414,9 @@ impl_value_list_for_tuple!(T0, T1, T2, T3, T4, T5; 0, 1, 2, 3, 4, 5);
 impl_value_list_for_tuple!(T0, T1, T2, T3, T4, T5, T6; 0, 1, 2, 3, 4, 5, 6);
 impl_value_list_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7; 0, 1, 2, 3, 4, 5, 6, 7);
 impl_value_list_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8; 0, 1, 2, 3, 4, 5, 6, 7, 8);
-impl_value_list_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8, T9; 
+impl_value_list_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8, T9;
                            0, 1, 2, 3, 4, 5, 6, 7, 8, 9);
-impl_value_list_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10; 
+impl_value_list_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10;
                            0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
 impl_value_list_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11;
                            0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11);
@@ -520,17 +546,17 @@ impl_batch_values_for_tuple!(T0, T1, T2, T3, T4, T5; 0, 1, 2, 3, 4, 5; 6);
 impl_batch_values_for_tuple!(T0, T1, T2, T3, T4, T5, T6; 0, 1, 2, 3, 4, 5, 6; 7);
 impl_batch_values_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7; 0, 1, 2, 3, 4, 5, 6, 7; 8);
 impl_batch_values_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8; 0, 1, 2, 3, 4, 5, 6, 7, 8; 9);
-impl_batch_values_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8, T9; 
+impl_batch_values_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8, T9;
                              0, 1, 2, 3, 4, 5, 6, 7, 8, 9; 10);
-impl_batch_values_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10; 
+impl_batch_values_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10;
                              0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10; 11);
-impl_batch_values_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11; 
+impl_batch_values_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11;
                              0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11; 12);
-impl_batch_values_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12; 
+impl_batch_values_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12;
                              0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12; 13);
-impl_batch_values_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13; 
+impl_batch_values_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13;
                              0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13; 14);
-impl_batch_values_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14; 
+impl_batch_values_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14;
                              0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14; 15);
 impl_batch_values_for_tuple!(T0, T1, T2, T3, T4, T5, T6, T7, T8, T9, T10, T11, T12, T13, T14, T15;
                              0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15; 16);
diff --git a/scylla/src/transport/connection.rs b/scylla/src/transport/connection.rs
index 9d4e8e81..fa443c40 100644
--- a/scylla/src/transport/connection.rs
+++ b/scylla/src/transport/connection.rs
@@ -32,7 +32,6 @@ use crate::transport::Compression;
 pub struct Connection {
     submit_channel: mpsc::Sender<Task>,
     _worker_handle: RemoteHandle<()>,
-    connect_address: SocketAddr,
     source_port: u16,
     shard_info: Option<ShardInfo>,
     config: ConnectionConfig,
@@ -99,7 +98,6 @@ impl Connection {
             submit_channel: sender,
             _worker_handle,
             source_port,
-            connect_address: addr,
             shard_info: None,
             config,
             is_shard_aware: false,
@@ -411,10 +409,6 @@ impl Connection {
     fn set_is_shard_aware(&mut self, is_shard_aware: bool) {
         self.is_shard_aware = is_shard_aware;
     }
-
-    pub fn get_connect_address(&self) -> SocketAddr {
-        self.connect_address
-    }
 }
 
 pub async fn open_connection(
diff --git a/scylla/src/transport/topology.rs b/scylla/src/transport/topology.rs
index 23e49ba2..5cea3f76 100644
--- a/scylla/src/transport/topology.rs
+++ b/scylla/src/transport/topology.rs
@@ -174,13 +174,12 @@ async fn query_peers(conn: &Connection, connect_port: u16) -> Result<Vec<Peer>,
 
     let typed_peers_rows =
         peers_rows.into_typed::<(IpAddr, Option<String>, Option<String>, Option<Vec<String>>)>();
+    let typed_local_rows =
+        local_rows.into_typed::<(IpAddr, Option<String>, Option<String>, Option<Vec<String>>)>();
 
-    // For the local node we should use connection's address instead of rpc_address unless SNI is enabled (TODO)
-    // Replace address in local_rows with connection's address
-    let local_address: IpAddr = conn.get_connect_address().ip();
-    let typed_local_rows = local_rows
-        .into_typed::<(IpAddr, Option<String>, Option<String>, Option<Vec<String>>)>()
-        .map(|res| res.map(|(_addr, dc, rack, tokens)| (local_address, dc, rack, tokens)));
+    // Here we are considering the local node to be just a regular peer, but datastax claims that unless
+    // you have SNI enabled, you shouldn't trust rpc_address (So they just use the initial connection's address)
+    // we could expose this from conn & use it directly instead of system.local.rpc_address
 
     for row in typed_peers_rows.chain(typed_local_rows) {
         let (ip_address, datacenter, rack, tokens) = row.map_err(|_| {


Test patch:
diff --git a/scylla/src/frame/value_tests.rs b/scylla/src/frame/value_tests.rs
index 0d923164..76d3671e 100644
--- a/scylla/src/frame/value_tests.rs
+++ b/scylla/src/frame/value_tests.rs
@@ -1,6 +1,6 @@
 use super::value::{
-    BatchValues, SerializeValuesError, SerializedResult, SerializedValues, Value, ValueList,
-    ValueTooBig,
+    BatchValues, MaybeUnset, SerializeValuesError, SerializedResult, SerializedValues, Unset,
+    Value, ValueList, ValueTooBig,
 };
 use bytes::BufMut;
 use std::borrow::Cow;
@@ -33,6 +33,17 @@ fn option_value() {
     assert_eq!(serialized(null_i32), &(-1_i32).to_be_bytes()[..]);
 }
 
+#[test]
+fn unset_value() {
+    assert_eq!(serialized(Unset), &(-2_i32).to_be_bytes()[..]);
+
+    let unset_i32: MaybeUnset<i32> = MaybeUnset::Unset;
+    assert_eq!(serialized(unset_i32), &(-2_i32).to_be_bytes()[..]);
+
+    let set_i32: MaybeUnset<i32> = MaybeUnset::Set(32);
+    assert_eq!(serialized(set_i32), vec![0, 0, 0, 4, 0, 0, 0, 32]);
+}
+
 #[test]
 fn ref_value() {
     assert_eq!(serialized(&1_i32), serialized(1_i32));