From 612404c47f0b61fc111d141e83e714307ccf6a30 Mon Sep 17 00:00:00 2001 From: junxiang Mu <1948535941@qq.com> Date: Fri, 25 Jul 2025 15:07:44 +0800 Subject: [PATCH] Test: add e2e_test for s3select Signed-off-by: junxiang Mu <1948535941@qq.com> --- Cargo.lock | 117 ++++++++- crates/e2e_test/Cargo.toml | 2 + crates/e2e_test/src/reliant/mod.rs | 1 + crates/e2e_test/src/reliant/sql.rs | 402 +++++++++++++++++++++++++++++ 4 files changed, 518 insertions(+), 4 deletions(-) create mode 100644 crates/e2e_test/src/reliant/sql.rs diff --git a/Cargo.lock b/Cargo.lock index 3098046bb..3a23fd84a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -672,6 +672,36 @@ version = "1.5.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8" +[[package]] +name = "aws-config" +version = "1.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c0baa720ebadea158c5bda642ac444a2af0cdf7bb66b46d1e4533de5d1f449d0" +dependencies = [ + "aws-credential-types", + "aws-runtime", + "aws-sdk-sso", + "aws-sdk-ssooidc", + "aws-sdk-sts", + "aws-smithy-async", + "aws-smithy-http", + "aws-smithy-json", + "aws-smithy-runtime", + "aws-smithy-runtime-api", + "aws-smithy-types", + "aws-types", + "bytes", + "fastrand", + "hex", + "http 1.3.1", + "ring", + "time", + "tokio", + "tracing", + "url", + "zeroize", +] + [[package]] name = "aws-credential-types" version = "1.2.4" @@ -734,9 +764,9 @@ dependencies = [ [[package]] name = "aws-sdk-s3" -version = "1.98.0" +version = "1.99.0" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "029e89cae7e628553643aecb3a3f054a0a0912ff0fd1f5d6a0b4fda421dce64b" +checksum = "b2d64d68c93000d5792b2a25fbeaafb90985fa80a1c8adfe93f24fb271296f5f" dependencies = [ "aws-credential-types", "aws-runtime", @@ -766,6 +796,73 @@ dependencies = [ "url", ] +[[package]] +name = "aws-sdk-sso" +version = "1.77.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "18f2f37fea82468fe3f5a059542c05392ef680c4f7f00e0db02df8b6e5c7d0c6" +dependencies = [ + "aws-credential-types", + "aws-runtime", + "aws-smithy-async", + "aws-smithy-http", + "aws-smithy-json", + "aws-smithy-runtime", + "aws-smithy-runtime-api", + "aws-smithy-types", + "aws-types", + "bytes", + "fastrand", + "http 0.2.12", + "regex-lite", + "tracing", +] + +[[package]] +name = "aws-sdk-ssooidc" +version = "1.78.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ecb4f6eada20e0193450cd48b12ed05e1e66baac86f39160191651b932f2b7d9" +dependencies = [ + "aws-credential-types", + "aws-runtime", + "aws-smithy-async", + "aws-smithy-http", + "aws-smithy-json", + "aws-smithy-runtime", + "aws-smithy-runtime-api", + "aws-smithy-types", + "aws-types", + "bytes", + "fastrand", + "http 0.2.12", + "regex-lite", + "tracing", +] + +[[package]] +name = "aws-sdk-sts" +version = "1.79.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "317377afba3498fca4948c5d32b399ef9a5ad35561a1e8a6f2ac7273dabf802d" +dependencies = [ + "aws-credential-types", + "aws-runtime", + "aws-smithy-async", + "aws-smithy-http", + "aws-smithy-json", + "aws-smithy-query", + "aws-smithy-runtime", + "aws-smithy-runtime-api", + "aws-smithy-types", + "aws-smithy-xml", + "aws-types", + "fastrand", + "http 0.2.12", + "regex-lite", + "tracing", +] + [[package]] name = "aws-sigv4" version = "1.3.3" @@ -904,6 +1001,16 @@ dependencies = [ "aws-smithy-runtime-api", ] +[[package]] +name = "aws-smithy-query" +version = "0.60.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f2fbd61ceb3fe8a1cb7352e42689cec5335833cd9f94103a61e98f9bb61c64bb" +dependencies = [ + "aws-smithy-types", + "urlencoding", +] + [[package]] name = "aws-smithy-runtime" version = "1.8.5" @@ -982,9 +1089,9 @@ dependencies = [ [[package]] name = "aws-types" -version = "1.3.7" +version = "1.3.8" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8a322fec39e4df22777ed3ad8ea868ac2f94cd15e1a55f6ee8d8d6305057689a" +checksum = "b069d19bf01e46298eaedd7c6f283fe565a59263e53eebec945f3e6398f42390" dependencies = [ "aws-credential-types", "aws-smithy-async", @@ -3466,6 +3573,8 @@ checksum = "92773504d58c093f6de2459af4af33faa518c13451eb8f2b5698ed3d36e7c813" name = "e2e_test" version = "0.0.5" dependencies = [ + "aws-config", + "aws-sdk-s3", "bytes", "flatbuffers 25.2.10", "futures", diff --git a/crates/e2e_test/Cargo.toml b/crates/e2e_test/Cargo.toml index 4edaffcd7..da5b1d308 100644 --- a/crates/e2e_test/Cargo.toml +++ b/crates/e2e_test/Cargo.toml @@ -39,3 +39,5 @@ rustfs-madmin.workspace = true rustfs-filemeta.workspace = true bytes.workspace = true serial_test = "3.2.0" +aws-sdk-s3 = "1.99.0" +aws-config = "1.8.3" diff --git a/crates/e2e_test/src/reliant/mod.rs b/crates/e2e_test/src/reliant/mod.rs index e3d11ebc2..00bf9d98a 100644 --- a/crates/e2e_test/src/reliant/mod.rs +++ b/crates/e2e_test/src/reliant/mod.rs @@ -14,3 +14,4 @@ mod lock; mod node_interact_test; +mod sql; diff --git a/crates/e2e_test/src/reliant/sql.rs b/crates/e2e_test/src/reliant/sql.rs new file mode 100644 index 000000000..4ca02d682 --- /dev/null +++ b/crates/e2e_test/src/reliant/sql.rs @@ -0,0 +1,402 @@ +#![cfg(test)] +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use aws_config::meta::region::RegionProviderChain; +use aws_sdk_s3::Client; +use aws_sdk_s3::config::{Credentials, Region}; +use aws_sdk_s3::types::{ + CsvInput, CsvOutput, ExpressionType, FileHeaderInfo, InputSerialization, JsonInput, JsonOutput, JsonType, OutputSerialization, +}; +use bytes::Bytes; +use serial_test::serial; +use std::error::Error; + +const ENDPOINT: &str = "http://localhost:9000"; +const ACCESS_KEY: &str = "rustfsadmin"; +const SECRET_KEY: &str = "rustfsadmin"; +const BUCKET: &str = "test-sql-bucket"; +const CSV_OBJECT: &str = "test-data.csv"; +const JSON_OBJECT: &str = "test-data.json"; + +async fn create_aws_s3_client() -> Result> { + let region_provider = RegionProviderChain::default_provider().or_else(Region::new("us-east-1")); + let shared_config = aws_config::defaults(aws_config::BehaviorVersion::latest()) + .region(region_provider) + .credentials_provider(Credentials::new(ACCESS_KEY, SECRET_KEY, None, None, "static")) + .endpoint_url(ENDPOINT) + .load() + .await; + + let client = Client::from_conf( + aws_sdk_s3::Config::from(&shared_config) + .to_builder() + .force_path_style(true) // Important for S3-compatible services + .build(), + ); + + Ok(client) +} + +async fn setup_test_bucket(client: &Client) -> Result<(), Box> { + match client.create_bucket().bucket(BUCKET).send().await { + Ok(_) => {} + Err(e) => { + let error_str = e.to_string(); + if !error_str.contains("BucketAlreadyOwnedByYou") && !error_str.contains("BucketAlreadyExists") { + return Err(e.into()); + } + } + } + Ok(()) +} + +async fn upload_test_csv(client: &Client) -> Result<(), Box> { + let csv_data = "name,age,city\nAlice,30,New York\nBob,25,Los Angeles\nCharlie,35,Chicago\nDiana,28,Boston"; + + client + .put_object() + .bucket(BUCKET) + .key(CSV_OBJECT) + .body(Bytes::from(csv_data.as_bytes()).into()) + .send() + .await?; + + Ok(()) +} + +async fn upload_test_json(client: &Client) -> Result<(), Box> { + let json_data = r#"{"name":"Alice","age":30,"city":"New York"} +{"name":"Bob","age":25,"city":"Los Angeles"} +{"name":"Charlie","age":35,"city":"Chicago"} +{"name":"Diana","age":28,"city":"Boston"}"#; + + client + .put_object() + .bucket(BUCKET) + .key(JSON_OBJECT) + .body(Bytes::from(json_data.as_bytes()).into()) + .send() + .await?; + Ok(()) +} + +async fn process_select_response( + mut event_stream: aws_sdk_s3::operation::select_object_content::SelectObjectContentOutput, +) -> Result> { + let mut total_data = Vec::new(); + + while let Ok(Some(event)) = event_stream.payload.recv().await { + match event { + aws_sdk_s3::types::SelectObjectContentEventStream::Records(records_event) => { + if let Some(payload) = records_event.payload { + let data = payload.into_inner(); + total_data.extend_from_slice(&data); + } + } + aws_sdk_s3::types::SelectObjectContentEventStream::End(_) => { + break; + } + _ => { + // Handle other event types (Stats, Progress, Cont, etc.) + } + } + } + + Ok(String::from_utf8(total_data)?) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[serial] +#[ignore = "requires running RustFS server at localhost:9000"] +async fn test_select_object_content_csv_basic() -> Result<(), Box> { + let client = create_aws_s3_client().await?; + setup_test_bucket(&client).await?; + upload_test_csv(&client).await?; + + // Construct SelectObjectContent request - basic query + let sql = "SELECT * FROM S3Object WHERE age > 28"; + + let csv_input = CsvInput::builder().file_header_info(FileHeaderInfo::Use).build(); + + let input_serialization = InputSerialization::builder().csv(csv_input).build(); + + let csv_output = CsvOutput::builder().build(); + let output_serialization = OutputSerialization::builder().csv(csv_output).build(); + + let response = client + .select_object_content() + .bucket(BUCKET) + .key(CSV_OBJECT) + .expression(sql) + .expression_type(ExpressionType::Sql) + .input_serialization(input_serialization) + .output_serialization(output_serialization) + .send() + .await?; + + let result_str = process_select_response(response).await?; + + println!("CSV Select result: {result_str}"); + + // Verify results contain records with age > 28 + assert!(result_str.contains("Alice,30,New York")); + assert!(result_str.contains("Charlie,35,Chicago")); + assert!(!result_str.contains("Bob,25,Los Angeles")); + assert!(!result_str.contains("Diana,28,Boston")); + + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[serial] +#[ignore = "requires running RustFS server at localhost:9000"] +async fn test_select_object_content_csv_aggregation() -> Result<(), Box> { + let client = create_aws_s3_client().await?; + setup_test_bucket(&client).await?; + upload_test_csv(&client).await?; + + // Construct aggregation query - use simpler approach + let sql = "SELECT name, age FROM S3Object WHERE age >= 25"; + + let csv_input = CsvInput::builder().file_header_info(FileHeaderInfo::Use).build(); + + let input_serialization = InputSerialization::builder().csv(csv_input).build(); + + let csv_output = CsvOutput::builder().build(); + let output_serialization = OutputSerialization::builder().csv(csv_output).build(); + + let response = client + .select_object_content() + .bucket(BUCKET) + .key(CSV_OBJECT) + .expression(sql) + .expression_type(ExpressionType::Sql) + .input_serialization(input_serialization) + .output_serialization(output_serialization) + .send() + .await?; + + let result_str = process_select_response(response).await?; + + println!("CSV Aggregation result: {result_str}"); + + // Verify query results - should include records with age >= 25 + assert!(result_str.contains("Alice")); + assert!(result_str.contains("Bob")); + assert!(result_str.contains("Charlie")); + assert!(result_str.contains("Diana")); + assert!(result_str.contains("30")); + assert!(result_str.contains("25")); + assert!(result_str.contains("35")); + assert!(result_str.contains("28")); + + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[serial] +#[ignore = "requires running RustFS server at localhost:9000"] +async fn test_select_object_content_json_basic() -> Result<(), Box> { + let client = create_aws_s3_client().await?; + setup_test_bucket(&client).await?; + upload_test_json(&client).await?; + + // Construct JSON query + let sql = "SELECT s.name, s.age FROM S3Object s WHERE s.age > 28"; + + let json_input = JsonInput::builder().set_type(Some(JsonType::Document)).build(); + + let input_serialization = InputSerialization::builder().json(json_input).build(); + + let json_output = JsonOutput::builder().build(); + let output_serialization = OutputSerialization::builder().json(json_output).build(); + + let response = client + .select_object_content() + .bucket(BUCKET) + .key(JSON_OBJECT) + .expression(sql) + .expression_type(ExpressionType::Sql) + .input_serialization(input_serialization) + .output_serialization(output_serialization) + .send() + .await?; + + let result_str = process_select_response(response).await?; + + println!("JSON Select result: {result_str}"); + + // Verify JSON query results + assert!(result_str.contains("Alice")); + assert!(result_str.contains("Charlie")); + assert!(result_str.contains("30")); + assert!(result_str.contains("35")); + + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[serial] +#[ignore = "requires running RustFS server at localhost:9000"] +async fn test_select_object_content_csv_limit() -> Result<(), Box> { + let client = create_aws_s3_client().await?; + setup_test_bucket(&client).await?; + upload_test_csv(&client).await?; + + // Test LIMIT clause + let sql = "SELECT * FROM S3Object LIMIT 2"; + + let csv_input = CsvInput::builder().file_header_info(FileHeaderInfo::Use).build(); + + let input_serialization = InputSerialization::builder().csv(csv_input).build(); + + let csv_output = CsvOutput::builder().build(); + let output_serialization = OutputSerialization::builder().csv(csv_output).build(); + + let response = client + .select_object_content() + .bucket(BUCKET) + .key(CSV_OBJECT) + .expression(sql) + .expression_type(ExpressionType::Sql) + .input_serialization(input_serialization) + .output_serialization(output_serialization) + .send() + .await?; + + let result_str = process_select_response(response).await?; + + println!("CSV Limit result: {result_str}"); + + // Verify only first 2 records are returned + let lines: Vec<&str> = result_str.lines().filter(|line| !line.trim().is_empty()).collect(); + assert_eq!(lines.len(), 2, "Should return exactly 2 records"); + + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[serial] +#[ignore = "requires running RustFS server at localhost:9000"] +async fn test_select_object_content_csv_order_by() -> Result<(), Box> { + let client = create_aws_s3_client().await?; + setup_test_bucket(&client).await?; + upload_test_csv(&client).await?; + + // Test ORDER BY clause + let sql = "SELECT name, age FROM S3Object ORDER BY age DESC LIMIT 2"; + + let csv_input = CsvInput::builder().file_header_info(FileHeaderInfo::Use).build(); + + let input_serialization = InputSerialization::builder().csv(csv_input).build(); + + let csv_output = CsvOutput::builder().build(); + let output_serialization = OutputSerialization::builder().csv(csv_output).build(); + + let response = client + .select_object_content() + .bucket(BUCKET) + .key(CSV_OBJECT) + .expression(sql) + .expression_type(ExpressionType::Sql) + .input_serialization(input_serialization) + .output_serialization(output_serialization) + .send() + .await?; + + let result_str = process_select_response(response).await?; + + println!("CSV Order By result: {result_str}"); + + // Verify ordered by age descending + let lines: Vec<&str> = result_str.lines().filter(|line| !line.trim().is_empty()).collect(); + assert!(lines.len() >= 2, "Should return at least 2 records"); + + // Check if contains highest age records + assert!(result_str.contains("Charlie,35")); + assert!(result_str.contains("Alice,30")); + + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[serial] +#[ignore = "requires running RustFS server at localhost:9000"] +async fn test_select_object_content_error_handling() -> Result<(), Box> { + let client = create_aws_s3_client().await?; + setup_test_bucket(&client).await?; + upload_test_csv(&client).await?; + + // Test invalid SQL query + let sql = "SELECT * FROM S3Object WHERE invalid_column > 10"; + + let csv_input = CsvInput::builder().file_header_info(FileHeaderInfo::Use).build(); + + let input_serialization = InputSerialization::builder().csv(csv_input).build(); + + let csv_output = CsvOutput::builder().build(); + let output_serialization = OutputSerialization::builder().csv(csv_output).build(); + + // This query should fail because invalid_column doesn't exist + let result = client + .select_object_content() + .bucket(BUCKET) + .key(CSV_OBJECT) + .expression(sql) + .expression_type(ExpressionType::Sql) + .input_serialization(input_serialization) + .output_serialization(output_serialization) + .send() + .await; + + // Verify query fails (expected behavior) + assert!(result.is_err(), "Query with invalid column should fail"); + + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[serial] +#[ignore = "requires running RustFS server at localhost:9000"] +async fn test_select_object_content_nonexistent_object() -> Result<(), Box> { + let client = create_aws_s3_client().await?; + setup_test_bucket(&client).await?; + + // Test query on nonexistent object + let sql = "SELECT * FROM S3Object"; + + let csv_input = CsvInput::builder().file_header_info(FileHeaderInfo::Use).build(); + + let input_serialization = InputSerialization::builder().csv(csv_input).build(); + + let csv_output = CsvOutput::builder().build(); + let output_serialization = OutputSerialization::builder().csv(csv_output).build(); + + let result = client + .select_object_content() + .bucket(BUCKET) + .key("nonexistent.csv") + .expression(sql) + .expression_type(ExpressionType::Sql) + .input_serialization(input_serialization) + .output_serialization(output_serialization) + .send() + .await; + + // Verify query fails (expected behavior) + assert!(result.is_err(), "Query on nonexistent object should fail"); + + Ok(()) +}