Repository: AlexPikalov/cdrs (language: rust)

Issue:
Updating a set of udt results with cluster error
If a table contains a set of frozen UDT and we execute `update table set column = column + ? where ...` and pass a UDT using query_values, the cluster responds with `Server error: CString { string: "java.lang.IllegalArgumentException" }`. Executing the same query using text, results in a success.

The stack trace in Cassandra is:

```
java.lang.IllegalArgumentException: null
cassandra_1         |   at java.nio.Buffer.limit(Buffer.java:275) ~[na:1.8.0_222]
cassandra_1         |   at org.apache.cassandra.utils.ByteBufferUtil.readBytes(ByteBufferUtil.java:651) ~[apache-cassandra-3.11.4.jar:3.11.4]
cassandra_1         |   at org.apache.cassandra.serializers.CollectionSerializer.readValue(CollectionSerializer.java:105) ~[apache-cassandra-3.11.4.jar:3.11.4]
cassandra_1         |   at org.apache.cassandra.serializers.SetSerializer.deserializeForNativeProtocol(SetSerializer.java:101) ~[apache-cassandra-3.11.4.jar:3.11.4]
cassandra_1         |   at org.apache.cassandra.cql3.Sets$Value.fromSerialized(Sets.java:158) ~[apache-cassandra-3.11.4.jar:3.11.4]
cassandra_1         |   at org.apache.cassandra.cql3.Sets$Marker.bind(Sets.java:251) ~[apache-cassandra-3.11.4.jar:3.11.4]
cassandra_1         |   at org.apache.cassandra.cql3.Sets$Adder.execute(Sets.java:285) ~[apache-cassandra-3.11.4.jar:3.11.4]
cassandra_1         |   at org.apache.cassandra.cql3.statements.UpdateStatement.addUpdateForKey(UpdateStatement.java:94) ~[apache-cassandra-3.11.4.jar:3.11.4]
cassandra_1         |   at org.apache.cassandra.cql3.statements.ModificationStatement.addUpdates(ModificationStatement.java:694) ~[apache-cassandra-3.11.4.jar:3.11.4]
cassandra_1         |   at org.apache.cassandra.cql3.statements.ModificationStatement.getMutations(ModificationStatement.java:635) ~[apache-cassandra-3.11.4.jar:3.11.4]
cassandra_1         |   at org.apache.cassandra.cql3.statements.ModificationStatement.executeWithoutCondition(ModificationStatement.java:437) ~[apache-cassandra-3.11.4.jar:3.11.4]
cassandra_1         |   at org.apache.cassandra.cql3.statements.ModificationStatement.execute(ModificationStatement.java:425) ~[apache-cassandra-3.11.4.jar:3.11.4]
cassandra_1         |   at org.apache.cassandra.cql3.QueryProcessor.processStatement(QueryProcessor.java:225) ~[apache-cassandra-3.11.4.jar:3.11.4]
cassandra_1         |   at org.apache.cassandra.cql3.QueryProcessor.processPrepared(QueryProcessor.java:532) ~[apache-cassandra-3.11.4.jar:3.11.4]
cassandra_1         |   at org.apache.cassandra.cql3.QueryProcessor.processPrepared(QueryProcessor.java:509) ~[apache-cassandra-3.11.4.jar:3.11.4]
cassandra_1         |   at org.apache.cassandra.transport.messages.ExecuteMessage.execute(ExecuteMessage.java:146) ~[apache-cassandra-3.11.4.jar:3.11.4]
cassandra_1         |   at org.apache.cassandra.transport.Message$Dispatcher.channelRead0(Message.java:566) [apache-cassandra-3.11.4.jar:3.11.4]
cassandra_1         |   at org.apache.cassandra.transport.Message$Dispatcher.channelRead0(Message.java:410) [apache-cassandra-3.11.4.jar:3.11.4]
cassandra_1         |   at io.netty.channel.SimpleChannelInboundHandler.channelRead(SimpleChannelInboundHandler.java:105) [netty-all-4.0.44.Final.jar:4.0.44.Final]
cassandra_1         |   at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:357) [netty-all-4.0.44.Final.jar:4.0.44.Final]
cassandra_1         |   at io.netty.channel.AbstractChannelHandlerContext.access$600(AbstractChannelHandlerContext.java:35) [netty-all-4.0.44.Final.jar:4.0.44.Final]
cassandra_1         |   at io.netty.channel.AbstractChannelHandlerContext$7.run(AbstractChannelHandlerContext.java:348) [netty-all-4.0.44.Final.jar:4.0.44.Final]
cassandra_1         |   at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:511) [na:1.8.0_222]
cassandra_1         |   at org.apache.cassandra.concurrent.AbstractLocalAwareExecutorService$FutureTask.run(AbstractLocalAwareExecutorService.java:162) [apache-cassandra-3.11.4.jar:3.11.4]
cassandra_1         |   at org.apache.cassandra.concurrent.SEPWorker.run(SEPWorker.java:114) [apache-cassandra-3.11.4.jar:3.11.4]
cassandra_1         |   at java.lang.Thread.run(Thread.java:748) [na:1.8.0_222]

```



Discussion / hints:
Hello @krojew 

Would you be able to provide a full code snippet including table schemas that may help to reproduce the issue?

Thanks
DB preparation:

```
CREATE TYPE udt1 (
    id uuid,
    col1 text,
    col2 tinyint,
    col3 tinyint
);

CREATE TABLE table1 (
    id uuid PRIMARY KEY,
    udts set<frozen<udt1>>
);

INSERT INTO table1 (id, udts) VALUES (5bd8877a-e2b2-4d6f-aafd-c3f72a6964cf, {{
  id: 08f49fa5-934b-4aff-8a87-f3a3287296ba,
  col1: 'text',
  col2: 1,
  col3: 2
}});
```

Rust:

```
use cdrs::authenticators::StaticPasswordAuthenticator;
use cdrs::cluster::session::{new as new_session};
use cdrs::cluster::{ClusterTcpConfig, NodeTcpConfigBuilder};
use cdrs::consistency::Consistency;
use cdrs::frame::traits::IntoBytes;
use cdrs::load_balancing::RoundRobin;
use cdrs::query::*;
use cdrs::query_values;
use cdrs::types::prelude::*;
use cdrs_helpers_derive::*;
use uuid::Uuid;

#[derive(IntoCDRSValue)]
struct Model {
  id: Uuid,
  col1: String,
  col2: i8,
  col3: i8,
}

fn main() {
    log4rs::init_file("log4rs.yml", Default::default()).expect("Error initializing log4rs");

    let node = NodeTcpConfigBuilder::new(
        "localhost:9042",
        StaticPasswordAuthenticator::new("cassandra", "cassandra")
    ).build();
    let cluster_config = ClusterTcpConfig(vec![node]);
    let session = new_session(&cluster_config, RoundRobin::new()).expect("session should be created");
    let query = session.prepare("UPDATE bug_284.table1 SET udts = udts + ? WHERE id = ?").unwrap();

    let row = Model {
      id: Uuid::parse_str("68f49fa5-934b-4aff-8a87-f3a32872a6ba").unwrap(),
      col1: "abc".into(),
      col2: 1,
      col3: 2,
    };

    let params = QueryParamsBuilder::new().consistency(Consistency::Quorum).values(query_values!(row, Uuid::parse_str("5bd8877a-e2b2-4d6f-aafd-c3f72a6964cf").unwrap()));
    session.exec_with_params(&query, params.finalize()).unwrap();
}

```

Fix patch:
diff --git a/.travis.yml b/.travis.yml
index 5770bc2..fbcd3b2 100644
--- a/.travis.yml
+++ b/.travis.yml
@@ -1,6 +1,6 @@
 dist: trusty
 language: rust
-cache: cargo
+# cache: cargo
 rust:
   - stable
   - beta
@@ -19,9 +19,9 @@ install:
 script:
   # - cargo fmt -- --write-mode=diff
   - cargo build --verbose
-  - cargo test --verbose
-  - cargo test --verbose --features e2e-tests
-  - cargo test --features ssl
+  - cargo test --verbose --no-fail-fast
+  - cargo test --verbose --no-fail-fast --features e2e-tests
+  - cargo test --no-fail-fast --features ssl
 
 deploy:
   # deploy master to production


Test patch:
diff --git a/tests/derive_traits.rs b/tests/derive_traits.rs
index bbd062b..d8dfa8f 100644
--- a/tests/derive_traits.rs
+++ b/tests/derive_traits.rs
@@ -16,6 +16,8 @@ mod common;
 #[cfg(feature = "e2e-tests")]
 use common::*;
 
+#[cfg(feature = "e2e-tests")]
+use cdrs::consistency::Consistency;
 #[cfg(feature = "e2e-tests")]
 use cdrs::error::Result as CDRSResult;
 #[cfg(feature = "e2e-tests")]
@@ -27,12 +29,16 @@ use cdrs::query::QueryExecutor;
 #[cfg(feature = "e2e-tests")]
 use cdrs::query::QueryValues;
 #[cfg(feature = "e2e-tests")]
+use cdrs::query::*;
+#[cfg(feature = "e2e-tests")]
 use cdrs::types::blob::Blob;
 #[cfg(feature = "e2e-tests")]
 use cdrs::types::from_cdrs::FromCDRSByName;
 #[cfg(feature = "e2e-tests")]
 use cdrs::types::map::Map;
 #[cfg(feature = "e2e-tests")]
+use cdrs::types::prelude::*;
+#[cfg(feature = "e2e-tests")]
 use cdrs::types::rows::Row;
 #[cfg(feature = "e2e-tests")]
 use cdrs::types::udt::UDT;
@@ -41,6 +47,8 @@ use cdrs::types::value::{Bytes, Value};
 #[cfg(feature = "e2e-tests")]
 use cdrs::types::{AsRust, AsRustType, IntoRustByName};
 #[cfg(feature = "e2e-tests")]
+use cdrs_helpers_derive::*;
+#[cfg(feature = "e2e-tests")]
 use std::str::FromStr;
 #[cfg(feature = "e2e-tests")]
 use time::Timespec;
@@ -254,3 +262,99 @@ fn alter_udt_add() {
     );
   }
 }
+
+#[test]
+#[cfg(feature = "e2e-tests")]
+fn update_list_with_udt() {
+  let drop_table_cql = "DROP TABLE IF EXISTS cdrs_test.update_list_with_udt";
+  let drop_type_cql = "DROP TYPE IF EXISTS cdrs_test.update_list_with_udt";
+  let create_type_cql = "CREATE TYPE cdrs_test.update_list_with_udt (id uuid,
+    text text)";
+  let create_table_cql =
+    "CREATE TABLE IF NOT EXISTS cdrs_test.update_list_with_udt \
+     (id uuid PRIMARY KEY, udts_set set<frozen<cdrs_test.update_list_with_udt>>)";
+  let session = setup_multiple(&[
+    drop_table_cql,
+    drop_type_cql,
+    create_type_cql,
+    create_table_cql,
+  ])
+  .expect("setup");
+
+  #[derive(Clone, Debug, IntoCDRSValue, TryFromRow, PartialEq)]
+  struct RowStruct {
+    id: Uuid,
+    udts_set: Vec<MyUdt>,
+  }
+
+  impl RowStruct {
+    fn into_query_values(self) -> QueryValues {
+      query_values!("id" => self.id, "udts_set" => self.udts_set)
+    }
+  }
+
+  #[derive(Clone, Debug, IntoCDRSValue, TryFromUDT, PartialEq)]
+  struct MyUdt {
+    pub id: Uuid,
+    pub text: String,
+  }
+
+  let row_struct = RowStruct {
+    id: Uuid::parse_str("5bd8877a-e2b2-4d6f-aafd-c3f72a6964cf").expect("row id"),
+    udts_set: vec![MyUdt {
+      id: Uuid::parse_str("08f49fa5-934b-4aff-8a87-f3a3287296ba").expect("udt id"),
+      text: "text".into(),
+    }],
+  };
+
+  let cql = "INSERT INTO cdrs_test.update_list_with_udt \
+             (id, udts_set) VALUES (?, ?)";
+  session
+    .query_with_values(cql, row_struct.clone().into_query_values())
+    .expect("insert");
+
+  let query = session
+    .prepare("UPDATE cdrs_test.update_list_with_udt SET udts_set = udts_set + ? WHERE id = ?")
+    .expect("prepare query");
+  let params = QueryParamsBuilder::new()
+    .consistency(Consistency::Quorum)
+    .values(query_values!(
+      vec![MyUdt {
+        id: Uuid::parse_str("68f49fa5-934b-4aff-8a87-f3a32872a6ba").expect("udt id"),
+        text: "abc".into(),
+      }],
+      Uuid::parse_str("5bd8877a-e2b2-4d6f-aafd-c3f72a6964cf").unwrap()
+    ));
+  session
+    .exec_with_params(&query, params.finalize())
+    .expect("update set");
+
+  let expected_row_struct = RowStruct {
+    id: Uuid::parse_str("5bd8877a-e2b2-4d6f-aafd-c3f72a6964cf").expect("row id"),
+    udts_set: vec![
+      MyUdt {
+        id: Uuid::parse_str("08f49fa5-934b-4aff-8a87-f3a3287296ba").expect("udt id"),
+        text: "text".into(),
+      },
+      MyUdt {
+        id: Uuid::parse_str("68f49fa5-934b-4aff-8a87-f3a32872a6ba").expect("udt id"),
+        text: "abc".into(),
+      },
+    ],
+  };
+
+  let cql = "SELECT * FROM cdrs_test.update_list_with_udt";
+  let rows = session
+    .query(cql)
+    .expect("query")
+    .get_body()
+    .expect("get body")
+    .into_rows()
+    .expect("into rows");
+
+  assert_eq!(rows.len(), 1);
+  for row in rows {
+    let altered_row: RowStruct = RowStruct::try_from_row(row).expect("into RowStruct");
+    assert_eq!(altered_row, expected_row_struct);
+  }
+}