diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index 1a093a2d2a5..21a6b8d678f 100644 --- a/.github/workflows/rust.yml +++ b/.github/workflows/rust.yml @@ -157,6 +157,48 @@ jobs: env: RUSTFLAGS: "-D warnings" + test-windows-filesystem: + if: ${{ github.repository == 'GreptimeTeam/greptimedb' && (github.event_name != 'pull_request' || github.event.pull_request.draft == false) }} + name: Test Windows filesystem + runs-on: windows-2022 + timeout-minutes: 90 + env: + CARGO_INCREMENTAL: 0 + CARGO_PROFILE_DEV_DEBUG: 0 + CARGO_PROFILE_TEST_DEBUG: 0 + RUSTFLAGS: "-D warnings -C linker=lld-link" + RUST_BACKTRACE: 1 + RUST_MIN_STACK: 8388608 # 8MB + steps: + - run: git config --global core.autocrlf false + - uses: actions/checkout@v4 + with: + persist-credentials: false + - uses: ./.github/actions/setup-cyborg + - uses: arduino/setup-protoc@v3 + with: + repo-token: ${{ secrets.GITHUB_TOKEN }} + - uses: KyleMayes/install-llvm-action@v1 + with: + version: "14.0" + - uses: actions-rust-lang/setup-rust-toolchain@v1 + with: + cache: false + - name: Rust Cache + uses: Swatinem/rust-cache@v2 + with: + shared-key: "test-windows-filesystem" + cache-all-crates: "true" + save-if: ${{ github.ref == 'refs/heads/main' }} + - name: Install Cargo Nextest + uses: taiki-e/install-action@nextest + - name: Run filesystem regression tests + # Keep this focused and omit dashboard, which requires WSL on Windows. + run: >- + cargo nextest run --locked --lib + -p object-store -p common-datasource -p cli + -E 'package(=object-store) | package(=common-datasource) | (package(=cli) & test(metadata::snapshot::tests::))' + conflict-check: if: ${{ github.repository == 'GreptimeTeam/greptimedb' && (github.event_name != 'pull_request' || github.event.pull_request.draft == false) }} name: Check for conflict diff --git a/src/cli/src/metadata/snapshot.rs b/src/cli/src/metadata/snapshot.rs index 59b2599ad9c..9c010c2235b 100644 --- a/src/cli/src/metadata/snapshot.rs +++ b/src/cli/src/metadata/snapshot.rs @@ -12,11 +12,13 @@ // See the License for the specific language governing permissions and // limitations under the License. +use std::path::{Component, Path, PathBuf}; + use async_trait::async_trait; use clap::{Parser, Subcommand}; use common_error::ext::BoxedError; use common_meta::snapshot::MetadataSnapshotManager; -use object_store::{ObjectStore, services}; +use object_store::ObjectStore; use crate::Tool; use crate::common::{ObjectStoreConfig, StoreConfig, new_fs_object_store}; @@ -270,25 +272,44 @@ fn build_object_store_and_resolve_file_path( fs_root: &str, file_path: &str, ) -> Result<(ObjectStore, String), BoxedError> { - let object_store = object_store.build().map_err(BoxedError::new)?; - let object_store = match object_store { - Some(object_store) => object_store, - None => new_fs_object_store(fs_root)?, - }; + if let Some(object_store) = object_store.build().map_err(BoxedError::new)? { + return Ok((object_store, file_path.to_string())); + } - let file_path = if object_store.info().scheme() == services::FS_SCHEME { - resolve_relative_path_with_current_dir(file_path).map_err(BoxedError::new)? + let absolute_path = + resolve_relative_path_with_current_dir(file_path).map_err(BoxedError::new)?; + let (root, key) = split_fs_snapshot_path(fs_root, &absolute_path); + Ok((new_fs_object_store(&root)?, key)) +} + +/// Separates a local absolute path into an operator root and a relative key. +fn split_fs_snapshot_path(fs_root: &str, absolute_path: &str) -> (String, String) { + let path = Path::new(absolute_path); + // OpenDAL keys cannot contain a Windows drive/UNC prefix or a root directory. + // Keep the filesystem anchor in the operator root, not in the object key. + let anchor: PathBuf = path + .components() + .take_while(|c| matches!(c, Component::Prefix(_) | Component::RootDir)) + .collect(); + let key = path + .components() + .filter(|c| !matches!(c, Component::Prefix(_) | Component::RootDir)) + .map(|c| c.as_os_str().to_string_lossy()) + .collect::>() + .join("/"); + // Preserve the existing custom-root semantics: cwd-resolved keys are rooted + // under --dir. The default root follows the input path's drive on Windows. + let root = if fs_root == "/" { + anchor.to_string_lossy() } else { - file_path.to_string() + fs_root.into() }; - - Ok((object_store, file_path)) + (root.into_owned(), key) } #[cfg(test)] mod tests { use std::env; - use std::path::Path; use std::sync::Arc; use std::time::Duration; @@ -311,44 +332,159 @@ mod tests { #[tokio::test] async fn test_cmd_resolve_file_path() { - common_telemetry::init_default_ut_logging(); - let cmd = RestoreCommand::parse_from([ - "", - "--file_name", - "metadata_snapshot.metadata.fb", - "--backend", - "memory-store", - "--store-addrs", - "memory://", - ]); - let tool = cmd.build().await.unwrap(); let current_dir = env::current_dir().unwrap(); - let file_path = current_dir.join("metadata_snapshot.metadata.fb"); - assert_eq!(tool.file_path, file_path.to_string_lossy().to_string()); + for file_path in [ + "metadata_snapshot.metadata.fb".to_string(), + current_dir + .join("metadata_snapshot.metadata.fb") + .display() + .to_string(), + ] { + let cmd = RestoreCommand::parse_from([ + "", + "--file_name", + &file_path, + "--backend", + "memory-store", + "--store-addrs", + "memory://", + ]); + let tool = cmd.build().await.unwrap(); + assert!(!Path::new(&tool.file_path).is_absolute()); + assert!( + Path::new(&tool.file_path) + .components() + .all(|c| matches!(c, Component::Normal(_))) + ); + let expected = current_dir + .components() + .filter_map(|c| match c { + Component::Normal(part) => Some(part.to_str().unwrap()), + _ => None, + }) + .chain(["metadata_snapshot.metadata.fb"]) + .collect::>() + .join("/"); + assert_eq!(tool.file_path, expected); + } + } + #[cfg(windows)] + #[test] + fn test_split_windows_snapshot_path() { + for (path, root, key) in [ + ( + r"C:\backup\snapshot.metadata.fb", + r"C:\", + "backup/snapshot.metadata.fb", + ), + ( + r"D:\备份 dir\snapshot.metadata.fb", + r"D:\", + "备份 dir/snapshot.metadata.fb", + ), + ( + r"\\server\share\backup\snapshot.metadata.fb", + r"\\server\share\", + "backup/snapshot.metadata.fb", + ), + ( + r"\\?\C:\backup\snapshot.metadata.fb", + r"\\?\C:\", + "backup/snapshot.metadata.fb", + ), + ] { + assert_eq!( + split_fs_snapshot_path("/", path), + (root.to_string(), key.to_string()) + ); + assert_eq!( + split_fs_snapshot_path(r"D:\root", path), + (r"D:\root".to_string(), key.to_string()) + ); + } + } + + #[tokio::test] + async fn test_fs_snapshot_paths() { + let temp_dir = tempfile::tempdir().unwrap(); + for name in ["snapshot.metadata.fb", "nested dir/快照.metadata.fb"] { + let path = temp_dir.path().join(name); + let (store, key) = build_object_store_and_resolve_file_path( + ObjectStoreConfig::default(), + "/", + path.to_str().unwrap(), + ) + .unwrap(); + assert!(!path.exists()); + store.write(&key, "snapshot").await.unwrap(); + assert_eq!(std::fs::read(&path).unwrap(), b"snapshot"); + assert_eq!(store.read(&key).await.unwrap().to_bytes(), &b"snapshot"[..]); + } + } + + #[tokio::test] + async fn test_fs_snapshot_custom_root() { + let temp_dir = tempfile::tempdir().unwrap(); + let (store, key) = build_object_store_and_resolve_file_path( + ObjectStoreConfig::default(), + temp_dir.path().to_str().unwrap(), + "snapshot.metadata.fb", + ) + .unwrap(); + let current_dir = env::current_dir().unwrap(); + let expected = current_dir + .components() + .filter_map(|c| match c { + Component::Normal(part) => Some(part.to_str().unwrap()), + _ => None, + }) + .chain(["snapshot.metadata.fb"]) + .collect::>() + .join("/"); + assert_eq!(key, expected); + store.write(&key, "snapshot").await.unwrap(); + assert_eq!( + std::fs::read(temp_dir.path().join(expected)).unwrap(), + b"snapshot" + ); + } + + #[tokio::test] + async fn test_fs_snapshot_rejects_parent_traversal() { + let temp_dir = tempfile::tempdir().unwrap(); + let path = temp_dir.path().join("../snapshot.metadata.fb"); + let (store, key) = build_object_store_and_resolve_file_path( + ObjectStoreConfig::default(), + "/", + path.to_str().unwrap(), + ) + .unwrap(); + assert!(key.split('/').any(|part| part == "..")); + let err = store.write(&key, "snapshot").await.unwrap_err(); + assert_eq!(err.kind(), object_store::ErrorKind::NotFound); + assert!(err.to_string().contains("path escapes the configured root")); + } + + #[test] + fn test_remote_snapshot_key_unchanged() { let cmd = RestoreCommand::parse_from([ "", - "--file_name", - "metadata_snapshot.metadata.fb", "--backend", "memory-store", "--store-addrs", "memory://", + "--s3", + "--s3-bucket", + "snapshots", + "--s3-region", + "us-east-1", ]); - let tool = cmd.build().await.unwrap(); - assert_eq!(tool.file_path, file_path.to_string_lossy().to_string()); - - let cmd = RestoreCommand::parse_from([ - "", - "--file_name", - "metadata_snapshot.metadata.fb", - "--backend", - "memory-store", - "--store-addrs", - "memory://", - ]); - let tool = cmd.build().await.unwrap(); - assert_eq!(tool.file_path, file_path.to_string_lossy().to_string()); + let key = "nested/snapshot.metadata.fb"; + let (store, resolved) = + build_object_store_and_resolve_file_path(cmd.object_store, "/ignored", key).unwrap(); + assert_eq!(store.info().scheme(), object_store::services::S3_SCHEME); + assert_eq!(resolved, key); } async fn setup_backup_file(object_store: ObjectStore, file_path: &str) { diff --git a/src/common/datasource/src/file_format/csv.rs b/src/common/datasource/src/file_format/csv.rs index d016d4eabe9..d4cfb45088c 100644 --- a/src/common/datasource/src/file_format/csv.rs +++ b/src/common/datasource/src/file_format/csv.rs @@ -647,16 +647,15 @@ mod tests { let compressed_file_name = format!("test_compressed_csv.{}", compression_type.file_extension()); let compressed_file_path = temp_dir.path().join(&compressed_file_name); - let compressed_file_path_str = compressed_file_path.to_str().unwrap(); // Create a simple file store for testing - let store = test_store("/"); + let store = test_store(temp_dir.path().to_str().unwrap()); // Export CSV with compression let rows = stream_to_csv( Box::pin(DfRecordBatchStreamAdapter::new(recordbatches.as_stream())), store, - compressed_file_path_str, + &compressed_file_name, 1024, 1, &format, @@ -708,13 +707,13 @@ mod tests { } // Verify the compressed file can be decompressed and content matches original data - let store = test_store("/"); + let store = test_store(temp_dir.path().to_str().unwrap()); let schema = Arc::new( CsvFormat { compression_type, ..Default::default() } - .infer_schema(&store, compressed_file_path_str) + .infer_schema(&store, &compressed_file_name) .await .unwrap(), ); @@ -722,7 +721,7 @@ mod tests { let stream = file_to_stream( &store, - compressed_file_path_str, + &compressed_file_name, csv_source.clone(), None, compression_type, @@ -761,13 +760,13 @@ mod tests { ) .unwrap(); - let store = test_store("/"); + let store = test_store(temp_dir.path().to_str().unwrap()); let schema = Arc::new(arrow_schema::Schema::new(vec![ Field::new("id", DataType::UInt32, false), Field::new("name", DataType::Utf8, false), Field::new("value", DataType::Float64, false), ])); - let path = csv_file_path.to_str().unwrap(); + let path = "input.csv"; let stream = tolerant_csv_stream(&store, path, schema, vec![0, 1, 2], &CsvFormat::default()) @@ -793,13 +792,13 @@ mod tests { let csv_file_path = temp_dir.path().join("input.csv"); std::fs::write(&csv_file_path, "id,name,value\n1,Alice,10.5\n2,Bob\n").unwrap(); - let store = test_store("/"); + let store = test_store(temp_dir.path().to_str().unwrap()); let schema = Arc::new(arrow_schema::Schema::new(vec![ Field::new("id", DataType::UInt32, false), Field::new("name", DataType::Utf8, false), Field::new("value", DataType::Float64, false), ])); - let path = csv_file_path.to_str().unwrap(); + let path = "input.csv"; let stream = tolerant_csv_stream(&store, path, schema, vec![0, 1, 2], &CsvFormat::default()) diff --git a/src/common/datasource/src/file_format/json.rs b/src/common/datasource/src/file_format/json.rs index 3c24a4b7e27..23ce461fcb2 100644 --- a/src/common/datasource/src/file_format/json.rs +++ b/src/common/datasource/src/file_format/json.rs @@ -313,16 +313,15 @@ mod tests { let compressed_file_name = format!("test_compressed_json.{}", compression_type.file_extension()); let compressed_file_path = temp_dir.path().join(&compressed_file_name); - let compressed_file_path_str = compressed_file_path.to_str().unwrap(); // Create a simple file store for testing - let store = test_store("/"); + let store = test_store(temp_dir.path().to_str().unwrap()); // Export JSON with compression let rows = stream_to_json( Box::pin(DfRecordBatchStreamAdapter::new(recordbatches.as_stream())), store, - compressed_file_path_str, + &compressed_file_name, 1024, 1, &format, @@ -374,13 +373,13 @@ mod tests { } // Verify the compressed file can be decompressed and content matches original data - let store = test_store("/"); + let store = test_store(temp_dir.path().to_str().unwrap()); let schema = Arc::new( JsonFormat { compression_type, ..Default::default() } - .infer_schema(&store, compressed_file_path_str) + .infer_schema(&store, &compressed_file_name) .await .unwrap(), ); @@ -388,7 +387,7 @@ mod tests { let stream = file_to_stream( &store, - compressed_file_path_str, + &compressed_file_name, json_source.clone(), None, compression_type, diff --git a/src/common/datasource/src/file_format/tests.rs b/src/common/datasource/src/file_format/tests.rs index 27b86e1b114..14be2b1777f 100644 --- a/src/common/datasource/src/file_format/tests.rs +++ b/src/common/datasource/src/file_format/tests.rs @@ -66,13 +66,12 @@ impl Test<'_> { #[tokio::test] async fn test_json_opener() { - let store = test_store("/"); + let root = find_workspace_path("/src/common/datasource/tests"); + let store = test_store(root.to_str().unwrap()); let schema = basic_schema_with_time_format(); let file_source = Arc::new(JsonSource::new(schema)).with_batch_size(test_util::TEST_BATCH_SIZE); - let path = &find_workspace_path("/src/common/datasource/tests/json/basic.json") - .display() - .to_string(); + let path = "json/basic.json"; let tests = [ Test { @@ -108,11 +107,10 @@ async fn test_json_opener() { #[tokio::test] async fn test_csv_opener() { - let store = test_store("/"); + let root = find_workspace_path("/src/common/datasource/tests"); + let store = test_store(root.to_str().unwrap()); let schema = basic_schema_with_time_format(); - let path = &find_workspace_path("/src/common/datasource/tests/csv/basic.csv") - .display() - .to_string(); + let path = "csv/basic.csv"; let file_source = CsvSource::new(schema).with_batch_size(test_util::TEST_BATCH_SIZE); @@ -150,13 +148,12 @@ async fn test_csv_opener() { #[tokio::test(flavor = "multi_thread")] async fn test_parquet_exec() { - let store = test_store("/"); + let root = find_workspace_path("/src/common/datasource/tests"); + let store = test_store(root.to_str().unwrap()); let schema = test_basic_schema(); - let path = &find_workspace_path("/src/common/datasource/tests/parquet/basic.parquet") - .display() - .to_string(); + let path = "parquet/basic.parquet"; let parquet_source = ParquetSource::new(schema) .with_parquet_file_reader_factory(Arc::new(DefaultParquetFileReaderFactory::new(store))); @@ -191,11 +188,10 @@ async fn test_parquet_exec() { #[tokio::test] async fn test_orc_opener() { - let path = &find_workspace_path("/src/common/datasource/tests/orc/test.orc") - .display() - .to_string(); + let path = "orc/test.orc"; - let store = test_store("/"); + let root = find_workspace_path("/src/common/datasource/tests"); + let store = test_store(root.to_str().unwrap()); let schema = Arc::new(OrcFormat.infer_schema(&store, path).await.unwrap()); let file_source = Arc::new(OrcSource::new(schema.into())); diff --git a/src/common/datasource/src/test_util.rs b/src/common/datasource/src/test_util.rs index 4cefda88890..f95d24930b2 100644 --- a/src/common/datasource/src/test_util.rs +++ b/src/common/datasource/src/test_util.rs @@ -56,8 +56,8 @@ pub fn test_store(root: &str) -> ObjectStore { pub fn test_tmp_store(root: &str) -> (ObjectStore, TempDir) { let dir = create_temp_dir(root); - let builder = Fs::default(); - (ObjectStore::new(builder.root("/")).unwrap(), dir) + let store = test_store(dir.path().to_str().unwrap()); + (store, dir) } pub fn test_basic_schema() -> SchemaRef { @@ -94,9 +94,11 @@ pub(crate) fn scan_config( .build() } -pub async fn setup_stream_to_json_test(origin_path: &str, threshold: impl Fn(usize) -> usize) { - let store = test_store("/"); - +pub async fn setup_stream_to_json_test( + store: &ObjectStore, + origin_path: &str, + threshold: impl Fn(usize) -> usize, +) { let schema = basic_schema_with_time_format(); let json_opener = JsonOpener::new( @@ -117,9 +119,9 @@ pub async fn setup_stream_to_json_test(origin_path: &str, threshold: impl Fn(usi .build() .unwrap(); - let (tmp_store, dir) = test_tmp_store("test_stream_to_json"); + let (tmp_store, _dir) = test_tmp_store("test_stream_to_json"); - let output_path = format!("{}/{}", dir.path().display(), "output"); + let output_path = "output"; let json_format = JsonFormat::default(); @@ -127,7 +129,7 @@ pub async fn setup_stream_to_json_test(origin_path: &str, threshold: impl Fn(usi stream_to_json( Box::pin(stream), tmp_store.clone(), - &output_path, + output_path, threshold(size), 8, &json_format, @@ -136,18 +138,17 @@ pub async fn setup_stream_to_json_test(origin_path: &str, threshold: impl Fn(usi .is_ok() ); - let written = tmp_store.read(&output_path).await.unwrap(); + let written = tmp_store.read(output_path).await.unwrap(); let origin = store.read(origin_path).await.unwrap(); assert_eq_lines(written.to_vec(), origin.to_vec()); } pub async fn setup_stream_to_csv_test( + store: &ObjectStore, origin_path: &str, format_path: &str, threshold: impl Fn(usize) -> usize, ) { - let store = test_store("/"); - let schema = basic_schema_with_time_format(); let csv_source = CsvSource::new(schema).with_batch_size(TEST_BATCH_SIZE); @@ -168,9 +169,9 @@ pub async fn setup_stream_to_csv_test( .build() .unwrap(); - let (tmp_store, dir) = test_tmp_store("test_stream_to_csv"); + let (tmp_store, _dir) = test_tmp_store("test_stream_to_csv"); - let output_path = format!("{}/{}", dir.path().display(), "output"); + let output_path = "output"; let csv_format = CsvFormat { timestamp_format: Some("%m-%d-%Y".to_string()), @@ -183,7 +184,7 @@ pub async fn setup_stream_to_csv_test( stream_to_csv( Box::pin(stream), tmp_store.clone(), - &output_path, + output_path, threshold(size), 8, &csv_format, @@ -192,7 +193,7 @@ pub async fn setup_stream_to_csv_test( .is_ok() ); - let written = tmp_store.read(&output_path).await.unwrap(); + let written = tmp_store.read(output_path).await.unwrap(); let format_expect = store.read(format_path).await.unwrap(); assert_eq_lines(written.to_vec(), format_expect.to_vec()); } diff --git a/src/common/datasource/src/tests.rs b/src/common/datasource/src/tests.rs index 8be9add7c88..b983ef86627 100644 --- a/src/common/datasource/src/tests.rs +++ b/src/common/datasource/src/tests.rs @@ -18,34 +18,31 @@ use crate::test_util; #[tokio::test] async fn test_stream_to_json() { - let origin_path = &find_workspace_path("/src/common/datasource/tests/json/basic.json") - .display() - .to_string(); + let root = find_workspace_path("/src/common/datasource/tests/json"); + let store = test_util::test_store(root.to_str().unwrap()); + let origin_path = "basic.json"; // A small threshold // Triggers the flush each writes - test_util::setup_stream_to_json_test(origin_path, |size| size / 2).await; + test_util::setup_stream_to_json_test(&store, origin_path, |size| size / 2).await; // A large threshold // Only triggers the flush at last - test_util::setup_stream_to_json_test(origin_path, |size| size * 2).await; + test_util::setup_stream_to_json_test(&store, origin_path, |size| size * 2).await; } #[tokio::test] async fn test_stream_to_csv() { - let origin_path = &find_workspace_path("/src/common/datasource/tests/csv/basic.csv") - .display() - .to_string(); - - let format_path = &find_workspace_path("/src/common/datasource/tests/csv/basic_format.csv") - .display() - .to_string(); + let root = find_workspace_path("/src/common/datasource/tests/csv"); + let store = test_util::test_store(root.to_str().unwrap()); + let origin_path = "basic.csv"; + let format_path = "basic_format.csv"; // A small threshold // Triggers the flush each writes - test_util::setup_stream_to_csv_test(origin_path, format_path, |size| size / 2).await; + test_util::setup_stream_to_csv_test(&store, origin_path, format_path, |size| size / 2).await; // A large threshold // Only triggers the flush at last - test_util::setup_stream_to_csv_test(origin_path, format_path, |size| size * 2).await; + test_util::setup_stream_to_csv_test(&store, origin_path, format_path, |size| size * 2).await; }