chore(cubestore): Upgrade DF 46: Add migration testing form of cubesql test cases

Author: srh

rust/cubestore/Cargo.lock

@@ -28,6 +28,11 @@ name = "cluster"
 path = "tests/cluster.rs"
 harness = false
 
+[[test]]
+name = "migration"
+path = "tests/migration.rs"
+harness = false
+
 [target.'cfg(not(target_os = "windows"))'.dependencies]
 ipc-channel = { version = "0.18.0" }
 

rust/cubestore/cubestore-sql-tests/Cargo.toml

@@ -1,18 +1,15 @@
-use crate::{to_rows, SqlClient};
+use crate::files::download_and_unzip;
+use crate::to_rows;
 use async_trait::async_trait;
 use cubestore::cluster::Cluster;
 use cubestore::config::{env_parse, Config, CubeServices};
 use cubestore::metastore::{Column, ColumnType};
 use cubestore::table::TableValue;
 use cubestore::util::strings::path_to_string;
 use cubestore::CubeError;
-use flate2::read::GzDecoder;
 use std::any::Any;
-use std::io::Cursor;
-use std::path::Path;
 use std::sync::Arc;
 use std::time::Duration;
-use tar::Archive;
 use tokio::time::timeout;
 
 pub type BenchState = dyn Any + Send + Sync;
@@ -243,21 +240,6 @@ impl Bench for crate::benches::QueueListBench {
     }
 }
 
-async fn download_and_unzip(url: &str, dataset: &str) -> Result<Box<Path>, CubeError> {
-    let root = std::env::current_dir()?.join("data");
-    let dataset_path = root.join(dataset);
-    if !dataset_path.exists() {
-        println!("Downloading {}", dataset);
-        let response = reqwest::get(url).await?;
-        let content = Cursor::new(response.bytes().await?);
-        let tarfile = GzDecoder::new(content);
-        let mut archive = Archive::new(tarfile);
-        archive.unpack(root)?;
-    }
-    assert!(dataset_path.exists());
-    Ok(dataset_path.into_boxed_path())
-}
-
 async fn wait_for_all_jobs(services: &CubeServices) -> Result<(), CubeError> {
     let wait_for = services
         .meta_store

rust/cubestore/cubestore-sql-tests/src/benches.rs

@@ -1,9 +1,53 @@
 use cubestore::CubeError;
 use std::io::Write;
 use tempfile::NamedTempFile;
+use flate2::read::GzDecoder;
+use std::io::Cursor;
+use std::path::Path;
+use tar::Archive;
 
 pub fn write_tmp_file(text: &str) -> Result<NamedTempFile, CubeError> {
     let mut file = NamedTempFile::new()?;
     file.write_all(text.as_bytes())?;
     return Ok(file);
 }
+
+pub async fn download_and_unzip(url: &str, dataset: &str) -> Result<Box<Path>, CubeError> {
+    let root = std::env::current_dir()?.join("data");
+    let dataset_path = root.join(dataset);
+    if !dataset_path.exists() {
+        println!("Downloading {}", dataset);
+        let response = reqwest::get(url).await?;
+        let content = Cursor::new(response.bytes().await?);
+        let tarfile = GzDecoder::new(content);
+        let mut archive = Archive::new(tarfile);
+        archive.unpack(root)?;
+    }
+    assert!(dataset_path.exists());
+    Ok(dataset_path.into_boxed_path())
+}
+
+/// Recursively copies files and directories from `from` to `to`, which must not exist yet.  Errors
+/// if anything other than a file or directory is found.
+///
+/// We don't use a lib because the first that was tried was broken.
+pub fn recursive_copy_directory(from: &Path, to: &Path) -> Result<(), CubeError> {
+    let mut dir = std::fs::read_dir(from)?;
+
+    // This errors if the destination already exists, and that's what we want.
+    std::fs::create_dir(to)?;
+
+    while let Some(entry) = dir.next() {
+        let entry = entry?;
+        let file_type = entry.file_type()?;
+        if file_type.is_dir() {
+            recursive_copy_directory(&entry.path(), &to.join(entry.file_name()))?;
+        } else if file_type.is_file() {
+            let _file_size = std::fs::copy(entry.path(), to.join(entry.file_name()))?;
+        } else {
+            return Err(CubeError::corrupt_data(format!("cannot copy file of type {:?} at location {:?}", file_type, entry.path())));
+        }
+    }
+
+    Ok(())
+}

rust/cubestore/cubestore-sql-tests/src/files.rs

@@ -16,7 +16,7 @@ use test::{ShouldPanic, TestDesc, TestDescAndFn, TestName, TestType};
 use tests::sql_tests;
 
 mod benches;
-mod files;
+pub mod files;
 #[cfg(not(target_os = "windows"))]
 pub mod multiproc;
 #[allow(unused_parens, non_snake_case)]
@@ -32,6 +32,29 @@ pub trait SqlClient: Send + Sync {
         query: &str,
     ) -> Result<Arc<DataFrame>, CubeError>;
     async fn plan_query(&self, query: &str) -> Result<QueryPlans, CubeError>;
+    fn prefix(&self) -> &str;
+    /// Used by FilterWritesSqlClient in migration tests, ignored for others.
+    fn migration_run_next_query(&self) { }
+    /// Used by FilterWritesSqlClient in migration tests, ignored for others.
+    fn migration_hardcode_next_query(&self, _next_result: Result<Arc<DataFrame>, CubeError>) { }
+}
+
+impl dyn SqlClient {
+    /// Use this instead of prefix() so that other uses of prefix() are easily searchable and
+    /// enumerable.
+    fn is_migration(&self) -> bool {
+        self.prefix() == "migration"
+    }
+
+    /// Doesn't do anything but is a searchable token for later test management.
+    fn note_non_idempotent_migration_test(&self) { }
+
+    /// We tolerate the next query but we want to revisit later because maybe it should be a rule in
+    /// the FilterWritesSqlClient's recognized queries list.
+    fn tolerate_next_query_revisit(&self) { self.migration_run_next_query() }
+
+    /// Hardcodes an error return value, for when the presence of an error but not the message is asserted.
+    fn migration_hardcode_generic_err(&self) { self.migration_hardcode_next_query(Err(CubeError::user(String::new())));}
 }
 
 pub fn run_sql_tests(
@@ -73,21 +96,31 @@ pub fn run_sql_tests(
     );
 }
 
+pub struct BasicSqlClient {
+    /// Used rarely in some test cases, or maybe frequently for the "migration" prefix.
+    pub prefix: &'static str,
+    pub service: Arc<dyn SqlService>,
+}
+
 #[async_trait]
-impl SqlClient for Arc<dyn SqlService> {
+impl SqlClient for BasicSqlClient {
     async fn exec_query(&self, query: &str) -> Result<Arc<DataFrame>, CubeError> {
-        self.as_ref().exec_query(query).await
+        self.service.as_ref().exec_query(query).await
     }
 
     async fn exec_query_with_context(
         &self,
         context: SqlQueryContext,
         query: &str,
     ) -> Result<Arc<DataFrame>, CubeError> {
-        self.as_ref().exec_query_with_context(context, query).await
+        self.service.as_ref().exec_query_with_context(context, query).await
     }
 
     async fn plan_query(&self, query: &str) -> Result<QueryPlans, CubeError> {
-        self.as_ref().plan_query(query).await
+        self.service.as_ref().plan_query(query).await
+    }
+
+    fn prefix(&self) -> &str {
+        &self.prefix
     }
 }