Merge pull request scylladb#1127 from Lorak-mmk/fix-concurrent-ddl

Fix concurrent DDL queries

(cherry picked from commit 64647f9)

CONTRIBUTING.md

@@ -49,6 +49,18 @@ The above commands will leave a running ScyllaDB cluster in the background.
 To stop it, use `make down`.\
 Starting a cluster without running any test is possible with `make up`.
 
+### Writing tests that need to connect to Scylla
+
+If you test requires connecting to Scylla, there are a few things you should consider.
+
+1. Such tests are considered integration tests and should be placed in `scylla/tests/integration`.
+2. To avoid name conflicts while creating a keyspace use `unique_keyspace_name` function from `utils` module.
+3. This `utils` module (`scylla/tests/integration/utils.rs`) contains other functions that may be helpful for writing tests.
+   For example `create_new_session_builder` or `test_with_3_node_cluster`.
+4. To perform DDL queries (creating / altering / dropping a keyspace / table /type) use `ddl` method from the utils module.
+   To do this, import the `PerformDDL` trait (`use crate::utils::PerformDDL;`). Then you can call `ddl` method on a
+   `Session`.
+
 ### Tracing in tests
 
 By default cargo captures `print!` macro's output from tests and prints them for failed tests.

scylla/src/transport/caching_session.rs

@@ -329,7 +329,9 @@ where
 mod tests {
     use crate::query::Query;
     use crate::statement::PagingState;
-    use crate::test_utils::{create_new_session_builder, scylla_supports_tablets, setup_tracing};
+    use crate::test_utils::{
+        create_new_session_builder, scylla_supports_tablets, setup_tracing, PerformDDL,
+    };
     use crate::transport::partitioner::PartitionerName;
     use crate::transport::session::Session;
     use crate::utils::test_utils::unique_keyspace_name;
@@ -358,18 +360,15 @@ mod tests {
         }
 
         session
-            .query_unpaged(create_ks, &[])
+            .ddl(create_ks)
             .await
             .expect("Could not create keyspace");
 
         session
-            .query_unpaged(
-                format!(
-                    "CREATE TABLE IF NOT EXISTS {}.test_table (a int primary key, b int)",
-                    ks
-                ),
-                &[],
-            )
+            .ddl(format!(
+                "CREATE TABLE IF NOT EXISTS {}.test_table (a int primary key, b int)",
+                ks
+            ))
             .await
             .expect("Could not create table");
 
@@ -566,10 +565,7 @@ mod tests {
         let session: CachingSession = create_caching_session().await;
 
         session
-            .execute_unpaged(
-                "CREATE TABLE IF NOT EXISTS test_batch_table (a int, b int, primary key (a, b))",
-                (),
-            )
+            .ddl("CREATE TABLE IF NOT EXISTS test_batch_table (a int, b int, primary key (a, b))")
             .await
             .unwrap();
 
@@ -689,7 +685,7 @@ mod tests {
         let session: CachingSession = CachingSession::from(new_for_test(true).await, 100);
 
         session
-            .execute_unpaged("CREATE TABLE tbl (a int PRIMARY KEY, b int)", ())
+            .ddl("CREATE TABLE tbl (a int PRIMARY KEY, b int)")
             .await
             .unwrap();
 
@@ -745,10 +741,7 @@ mod tests {
         let session: CachingSession = CachingSession::from(new_for_test(false).await, 100);
 
         session
-            .execute_unpaged(
-                "CREATE TABLE tbl (a int PRIMARY KEY) with cdc = {'enabled': true}",
-                &(),
-            )
+            .ddl("CREATE TABLE tbl (a int PRIMARY KEY) with cdc = {'enabled': true}")
             .await
             .unwrap();
 

scylla/src/transport/connection.rs

@@ -2397,7 +2397,7 @@ mod tests {
     use crate::transport::connection::open_connection;
     use crate::transport::node::ResolvedContactPoint;
     use crate::transport::topology::UntranslatedEndpoint;
-    use crate::utils::test_utils::unique_keyspace_name;
+    use crate::utils::test_utils::{unique_keyspace_name, PerformDDL};
     use crate::SessionBuilder;
     use futures::{StreamExt, TryStreamExt};
     use std::collections::HashMap;
@@ -2452,17 +2452,14 @@ mod tests {
                 .build()
                 .await
                 .unwrap();
-            session.query_unpaged(format!("CREATE KEYSPACE IF NOT EXISTS {} WITH REPLICATION = {{'class' : 'NetworkTopologyStrategy', 'replication_factor' : 1}}", ks.clone()), &[]).await.unwrap();
+            session.ddl(format!("CREATE KEYSPACE IF NOT EXISTS {} WITH REPLICATION = {{'class' : 'NetworkTopologyStrategy', 'replication_factor' : 1}}", ks.clone())).await.unwrap();
             session.use_keyspace(ks.clone(), false).await.unwrap();
             session
-                .query_unpaged("DROP TABLE IF EXISTS connection_query_iter_tab", &[])
+                .ddl("DROP TABLE IF EXISTS connection_query_iter_tab")
                 .await
                 .unwrap();
             session
-                .query_unpaged(
-                    "CREATE TABLE IF NOT EXISTS connection_query_iter_tab (p int primary key)",
-                    &[],
-                )
+                .ddl("CREATE TABLE IF NOT EXISTS connection_query_iter_tab (p int primary key)")
                 .await
                 .unwrap();
         }
@@ -2548,13 +2545,10 @@ mod tests {
                 .build()
                 .await
                 .unwrap();
-            session.query_unpaged(format!("CREATE KEYSPACE IF NOT EXISTS {} WITH REPLICATION = {{'class' : 'NetworkTopologyStrategy', 'replication_factor' : 1}}", ks.clone()), &[]).await.unwrap();
+            session.ddl(format!("CREATE KEYSPACE IF NOT EXISTS {} WITH REPLICATION = {{'class' : 'NetworkTopologyStrategy', 'replication_factor' : 1}}", ks.clone())).await.unwrap();
             session.use_keyspace(ks.clone(), false).await.unwrap();
             session
-                .query_unpaged(
-                    "CREATE TABLE IF NOT EXISTS t (p int primary key, v blob)",
-                    &[],
-                )
+                .ddl("CREATE TABLE IF NOT EXISTS t (p int primary key, v blob)")
                 .await
                 .unwrap();
         }
@@ -2580,7 +2574,7 @@ mod tests {
                 .await
                 .unwrap();
 
-            connection.query_unpaged("TRUNCATE t").await.unwrap();
+            connection.ddl("TRUNCATE t").await.unwrap();
 
             let mut futs = Vec::new();