rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-10-12 15:02:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: use relative object keys for Windows filesystem access

Browse Source 
Signed-off-by: WenyXu <wenymedia@gmail.com>
This commit is contained in:
WenyXu
2026-09-22 12:46:31 +00:00
parent 2388257c35
commit 9ed4c008ff
7 changed files with 272 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/rust.yml
+42
View File
@@ -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
src/cli/src/metadata/snapshot.rs
+177 -41
View File
@@ -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::<Vec<_>>()
.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::<Vec<_>>()
.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::<Vec<_>>()
.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) {
src/common/datasource/src/file_format/csv.rs
+9 -10
View File
@@ -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())
src/common/datasource/src/file_format/json.rs
+5 -6
View File
@@ -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,
src/common/datasource/src/file_format/tests.rs
+12 -16
View File
@@ -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()));
src/common/datasource/src/test_util.rs
+16 -15
View File
@@ -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());
}
src/common/datasource/src/tests.rs
+11 -14
View File
@@ -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;
}