tests: introduce new test for compression

test are trying both algorithms (snappy, lz4), while
counting the size of the frames coming into scylla

for reference we also have one test doing the same without
compression

test main structure was borrowed from java-driver, the usage
of proxy to count the packets is unique to this change.

Author: fruch

scylla/tests/integration/compression.rs

@@ -0,0 +1,114 @@
+use crate::utils::{test_with_3_node_cluster, unique_keyspace_name, PerformDDL};
+use scylla::client::session::Session;
+use scylla::client::session_builder::SessionBuilder;
+use scylla::statement::query::Query;
+use scylla::frame::Compression;
+
+use scylla_proxy::{ProxyError, WorkerError, ShardAwareness, RequestRule, Reaction, RequestReaction, Condition, RequestOpcode};
+
+use tokio::sync::mpsc;
+use std::sync::Arc;
+
+async fn test_compression(compression: Option<Compression>, text_size: usize, expected_frame_total_size_range: std::ops::Range<usize>) {
+    let res = test_with_3_node_cluster(ShardAwareness::QueryNode, |proxy_uris, translation_map, mut running_proxy| async move {
+    
+        let request_rule = |tx| {
+            RequestRule(
+                Condition::or(Condition::RequestOpcode(RequestOpcode::Execute), Condition::RequestOpcode(RequestOpcode::Query)),
+                RequestReaction::noop().with_feedback_when_performed(tx),
+            )
+        };
+
+        let (request_tx, mut request_rx) = mpsc::unbounded_channel();
+        for running_node in running_proxy.running_nodes.iter_mut() {
+            running_node.change_request_rules(Some(vec![request_rule(request_tx.clone())]));
+        }
+
+        let session: Session = SessionBuilder::new()
+            .known_node(proxy_uris[0].as_str())
+            .address_translator(Arc::new(translation_map))
+            .compression(compression)
+            .build()
+            .await
+            .unwrap();
+
+        let ks = unique_keyspace_name();
+        session.ddl(format!("CREATE KEYSPACE IF NOT EXISTS {} WITH REPLICATION = {{'class' : 'NetworkTopologyStrategy', 'replication_factor' : 3}}", ks)).await.unwrap();
+        session.use_keyspace(ks, false).await.unwrap();
+        session
+            .ddl("CREATE TABLE test (k text PRIMARY KEY, t text, i int, f float)")
+            .await
+            .unwrap();
+
+        let q = Query::from("INSERT INTO test (k, t, i, f) VALUES (?, ?, ?, ?)");
+        let large_string = "a".repeat(text_size);
+        session.query_unpaged(q.clone(), ("key", large_string.clone(), 42_i32, 24.03_f32)).await.unwrap();
+
+        let result: Vec<(String, String, i32, f32)> = session
+            .query_unpaged("SELECT k, t, i, f FROM test WHERE k = 'key'", &[])
+            .await
+            .unwrap()
+            .into_rows_result()
+            .unwrap()
+            .rows::<(String, String, i32, f32)>()
+            .unwrap()
+            .collect::<Result<_, _>>()
+            .unwrap();
+        
+        assert_eq!(result, vec![((String::from("key"), String::from(large_string), 42_i32, 24.03_f32))]);
+       
+
+        let mut total_frame_size = 0;
+        while let Ok((request_frame, _shard)) = request_rx.try_recv() {
+            total_frame_size += request_frame.body.len();
+        }
+        println!("Total frame size: {}", total_frame_size);
+        assert!(expected_frame_total_size_range.contains(&total_frame_size));
+
+        running_proxy
+
+    }).await;
+
+    match res {
+        Ok(()) => (),
+        Err(ProxyError::Worker(WorkerError::DriverDisconnected(_))) => (),
+        Err(err) => panic!("{}", err),
+    }
+
+}
+
+#[tokio::test]
+#[cfg(not(scylla_cloud_tests))]
+async fn should_execute_queries_without_compression() {
+    test_compression(None, 1024, 4200..5000).await;
+}
+
+#[tokio::test]
+#[cfg(not(scylla_cloud_tests))]
+async fn should_execute_queries_without_compression_10mb() {
+    test_compression(None, 1024 * 10, 13300..15000).await;
+}
+
+#[tokio::test]
+#[cfg(not(scylla_cloud_tests))]
+async fn should_execute_queries_with_snappy_compression() {
+    test_compression(Some(Compression::Snappy), 1024, 3100..4000).await;
+}
+
+#[tokio::test]
+#[cfg(not(scylla_cloud_tests))]
+async fn should_execute_queries_with_snappy_compression_10mb() {
+    test_compression(Some(Compression::Snappy), 1024 * 10, 3300..4000).await;
+}
+
+#[tokio::test]
+#[cfg(not(scylla_cloud_tests))]
+async fn should_execute_queries_with_lz4_compression(){
+    test_compression(Some(Compression::Lz4), 1024, 3100..4000).await;
+}
+
+#[tokio::test]
+#[cfg(not(scylla_cloud_tests))]
+async fn should_execute_queries_with_lz4_compression_10mb(){
+    test_compression(Some(Compression::Lz4), 1024 * 10, 3300..4000).await;
+}

scylla/tests/integration/main.rs

@@ -18,4 +18,5 @@ mod silent_prepare_batch;
 mod silent_prepare_query;
 mod skip_metadata_optimization;
 mod tablets;
+mod compression;
 pub(crate) mod utils;