diff --git a/.config/s3s-footprint-baseline.txt b/.config/s3s-footprint-baseline.txt index a579ae56c..e5d110cad 100644 --- a/.config/s3s-footprint-baseline.txt +++ b/.config/s3s-footprint-baseline.txt @@ -12,6 +12,6 @@ # s3_error_lines lines holding s3_error!, same scope, minus those the script # header's rule clears (rustfs_s3_types's macro); fails closed # ecstore_files Rust files referencing s3s paths under crates/ecstore/src -files=127 +files=118 s3_error_lines=303 ecstore_files=0 diff --git a/.config/test-count-baseline.txt b/.config/test-count-baseline.txt index 8d21e5466..5cfd69abf 100644 --- a/.config/test-count-baseline.txt +++ b/.config/test-count-baseline.txt @@ -10,7 +10,7 @@ # intentionally removed. Regenerate with: # scripts/check_test_count_ratchet.sh --update e2e_test 908 18 -rustfs 5710 63 +rustfs 5712 63 rustfs-audit 70 0 rustfs-checksums 26 0 rustfs-common 25 0 @@ -48,7 +48,7 @@ rustfs-rio-v2 34 6 rustfs-s3-client 69 0 rustfs-s3-ops 8 0 rustfs-s3-types 99 0 -rustfs-s3select-api 206 0 +rustfs-s3select-api 207 0 rustfs-s3select-query 189 2 rustfs-scanner 1018 20 rustfs-scanner-metrics 81 0 diff --git a/Cargo.lock b/Cargo.lock index a167851e1..65685c968 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -10014,12 +10014,14 @@ source = "git+https://github.com/rustfs/gateway.git?rev=21283e874c2381bab611e934 dependencies = [ "bytes", "crc-fast", + "futures-core", "hex", "http 1.5.0", "md-5 0.11.0", "percent-encoding", "rustfs-gateway-stream", "rustfs-gateway-xml", + "s3s", "sha1 0.11.0", "sha2 0.11.1", "time", @@ -10833,10 +10835,10 @@ dependencies = [ "parking_lot", "rustfs-common", "rustfs-ecstore", + "rustfs-gateway-types", "rustfs-s3-types", "rustfs-storage-api", "rustfs-test-utils", - "s3s", "serde_json", "serial_test", "thiserror 2.0.21", @@ -10862,7 +10864,6 @@ dependencies = [ "parking_lot", "rustfs-s3select-api", "rustfs-test-utils", - "s3s", "serde", "serde_json", "tokio", diff --git a/crates/s3select-api/Cargo.toml b/crates/s3select-api/Cargo.toml index 164da0f43..f76de2caf 100644 --- a/crates/s3select-api/Cargo.toml +++ b/crates/s3select-api/Cargo.toml @@ -69,6 +69,7 @@ crc-fast.workspace = true rustfs-common.workspace = true datafusion = { workspace = true, default-features = false, features = ["parquet", "recursive_protection", "sql"] } rustfs-ecstore.workspace = true +rustfs-gateway-types.workspace = true rustfs-s3-types = { workspace = true } rustfs-storage-api.workspace = true futures = { workspace = true } @@ -76,7 +77,6 @@ futures-core = { workspace = true } flate2.workspace = true http.workspace = true memchr.workspace = true -s3s = { workspace = true, features = ["minio"] } serde_json = { workspace = true, features = ["raw_value"] } thiserror = { workspace = true } parking_lot.workspace = true diff --git a/crates/s3select-api/src/lib.rs b/crates/s3select-api/src/lib.rs index cebec28a6..75d5d969e 100644 --- a/crates/s3select-api/src/lib.rs +++ b/crates/s3select-api/src/lib.rs @@ -28,6 +28,7 @@ mod input_stream; mod metrics; pub mod object_store; pub mod query; +pub mod request; pub mod server; mod storage_api; pub use csv_input::csv_input_requires_normalization; diff --git a/crates/s3select-api/src/object_store.rs b/crates/s3select-api/src/object_store.rs index c089712a3..b0f120462 100644 --- a/crates/s3select-api/src/object_store.rs +++ b/crates/s3select-api/src/object_store.rs @@ -12,6 +12,11 @@ // See the License for the specific language governing permissions and // limitations under the License. +use crate::request::{CompressionType, InputSerialization, ScanRange, SelectObjectContentInput}; +use crate::request::{ + X_AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_ALGORITHM, X_AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY, + X_AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY_MD5, +}; use crate::{ PrepareSelectObjectSnapshotError, QueryError, SELECT_DEFAULT_READ_BUFFER_SIZE, SelectError, SelectGetObjectReader, SelectInputMetrics, SelectObjectOptions, SelectObjectSnapshot, SelectObjectSnapshotReadError, SelectStorageError, @@ -52,11 +57,6 @@ use http::{HeaderMap, HeaderValue, header::HeaderName}; use memchr::{memchr, memmem}; use rustfs_common::DEFAULT_DELIMITER; use rustfs_s3_types::{S3Error, S3ErrorCode, S3Result}; -use s3s::dto::{CompressionType, InputSerialization, ScanRange, SelectObjectContentInput}; -use s3s::header::{ - X_AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_ALGORITHM, X_AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY, - X_AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY_MD5, -}; use std::collections::VecDeque; use std::ops::Range; use std::sync::Arc; @@ -79,9 +79,10 @@ fn select_default_read_buffer_size_u64() -> u64 { fn compression_format(input: &InputSerialization) -> Result, SelectError> { match input.compression_type.as_ref().map(|value| value.as_str()) { - None | Some(CompressionType::NONE) => Ok(None), - Some(CompressionType::GZIP) => Ok(Some(CompressionFormat::Gzip)), - Some(CompressionType::BZIP2) => Ok(Some(CompressionFormat::Bzip2)), + None => Ok(None), + Some(value) if value == CompressionType::NONE.as_str() => Ok(None), + Some(value) if value == CompressionType::GZIP.as_str() => Ok(Some(CompressionFormat::Gzip)), + Some(value) if value == CompressionType::BZIP2.as_str() => Ok(Some(CompressionFormat::Bzip2)), Some(_) => Err(SelectError::InvalidCompressionFormat), } } @@ -507,7 +508,7 @@ pub(crate) fn is_json_document_input(input: &SelectObjectContentInput) -> bool { .input_serialization .json .as_ref() - .and_then(|json| json.type_.as_ref()) + .and_then(|json| json.r#type.as_ref()) .is_some_and(|json_type| json_type.as_str() == "DOCUMENT") } @@ -1399,6 +1400,14 @@ mod test { use crate::input_stream::{CompressionFormat, MAX_SELECT_RECORD_BYTES, compressed_input_reader, encode_compressed_fixture}; use crate::query::ast::{JsonPathSegment, JsonSource}; use crate::query::session::{QueryExecutionGuard, QueryExecutionOwner, QueryExecutionTracker}; + use crate::request::{ + CompressionType, CsvInput, CsvOutput, ExpressionType, FileHeaderInfo, InputSerialization, JsonInput, JsonOutput, + JsonType, OutputSerialization, ScanRange, SelectObjectContentInput, SelectObjectContentRequest, + }; + use crate::request::{ + X_AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_ALGORITHM, X_AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY, + X_AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY_MD5, + }; use crate::storage_api::SelectPutObjReader; use crate::storage_api::object_store::ObjectIO as _; use crate::{QueryError, SelectError, SelectInputMetrics, SelectStorageError}; @@ -1416,14 +1425,6 @@ mod test { use parking_lot::Mutex; use rustfs_s3_types::S3ErrorCode; use rustfs_test_utils::PutObjectCommitBarrier; - use s3s::dto::{ - CSVInput, CSVOutput, CompressionType, ExpressionType, FileHeaderInfo, InputSerialization, JSONInput, JSONOutput, - JSONType, OutputSerialization, ScanRange, SelectObjectContentInput, SelectObjectContentRequest, - }; - use s3s::header::{ - X_AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_ALGORITHM, X_AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY, - X_AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY_MD5, - }; use std::sync::{ Arc, atomic::{AtomicUsize, Ordering}, @@ -1515,13 +1516,13 @@ mod test { sse_customer_key_md5: None, request: SelectObjectContentRequest { expression: "SELECT * FROM s3object".to_string(), - expression_type: ExpressionType::from_static(ExpressionType::SQL), + expression_type: ExpressionType::SQL, input_serialization: InputSerialization { - csv: Some(CSVInput::default()), + csv: Some(CsvInput::default()), ..Default::default() }, output_serialization: OutputSerialization { - csv: Some(CSVOutput::default()), + csv: Some(CsvOutput::default()), ..Default::default() }, request_progress: None, @@ -1540,15 +1541,15 @@ mod test { sse_customer_key_md5: None, request: SelectObjectContentRequest { expression: expression.to_string(), - expression_type: ExpressionType::from_static(ExpressionType::SQL), + expression_type: ExpressionType::SQL, input_serialization: InputSerialization { - json: Some(JSONInput { - type_: Some(JSONType::from_static(JSONType::DOCUMENT)), + json: Some(JsonInput { + r#type: Some(JsonType::DOCUMENT), }), ..Default::default() }, output_serialization: OutputSerialization { - csv: Some(CSVOutput::default()), + csv: Some(CsvOutput::default()), ..Default::default() }, request_progress: None, @@ -1600,16 +1601,14 @@ mod test { assert_eq!(pinned.json_source, expected); } - fn json_input(bucket: &str, object: &str, json_type: &'static str) -> Arc { + fn json_input(bucket: &str, object: &str, json_type: JsonType) -> Arc { let mut input = (*csv_input(bucket, object)).clone(); input.request.input_serialization = InputSerialization { - json: Some(JSONInput { - type_: Some(JSONType::from_static(json_type)), - }), + json: Some(JsonInput { r#type: Some(json_type) }), ..Default::default() }; input.request.output_serialization = OutputSerialization { - json: Some(JSONOutput::default()), + json: Some(JsonOutput::default()), ..Default::default() }; Arc::new(input) @@ -1961,9 +1960,9 @@ mod test { sse_customer_key_md5: None, request: SelectObjectContentRequest { expression: "SELECT * FROM s3object".to_string(), - expression_type: ExpressionType::from_static(ExpressionType::SQL), + expression_type: ExpressionType::SQL, input_serialization: InputSerialization { - csv: Some(CSVInput { + csv: Some(CsvInput { record_delimiter: Some(record_delimiter.to_string()), file_header_info, ..Default::default() @@ -1971,7 +1970,7 @@ mod test { ..Default::default() }, output_serialization: OutputSerialization { - csv: Some(CSVOutput::default()), + csv: Some(CsvOutput::default()), ..Default::default() }, request_progress: None, @@ -2025,13 +2024,13 @@ mod test { sse_customer_key_md5: Some("secret-customer-key-md5".to_string()), request: SelectObjectContentRequest { expression: "SELECT * FROM s3object".to_string(), - expression_type: ExpressionType::from_static(ExpressionType::SQL), + expression_type: ExpressionType::SQL, input_serialization: InputSerialization { - csv: Some(CSVInput::default()), + csv: Some(CsvInput::default()), ..Default::default() }, output_serialization: OutputSerialization { - csv: Some(CSVOutput::default()), + csv: Some(CsvOutput::default()), ..Default::default() }, request_progress: None, @@ -2078,15 +2077,8 @@ mod test { let selected_start = i64::try_from(b"old_header,value\nskip_old,0\n".len()).expect("fixture offset should fit in i64"); let snapshot = prepare_test_snapshot(BUCKET, OBJECT).await; - let store = scan_range_csv_store( - BUCKET, - OBJECT, - snapshot, - "\n", - Some(FileHeaderInfo::from_static(FileHeaderInfo::USE)), - selected_start, - selected_start, - ); + let store = + scan_range_csv_store(BUCKET, OBJECT, snapshot, "\n", Some(FileHeaderInfo::USE), selected_start, selected_start); let commit_barrier = PutObjectCommitBarrier::before_namespace(BUCKET, OBJECT); let writer = tokio::spawn(async move { env.put_object_bytes(BUCKET, OBJECT, NEW_DATA.to_vec()).await; @@ -2555,9 +2547,9 @@ mod test { sse_customer_key_md5: None, request: SelectObjectContentRequest { expression: "SELECT * FROM s3object".to_string(), - expression_type: ExpressionType::from_static(ExpressionType::SQL), + expression_type: ExpressionType::SQL, input_serialization: InputSerialization { - csv: Some(CSVInput { + csv: Some(CsvInput { record_delimiter: Some("aa".to_string()), file_header_info, ..Default::default() @@ -2565,7 +2557,7 @@ mod test { ..Default::default() }, output_serialization: OutputSerialization { - csv: Some(CSVOutput::default()), + csv: Some(CsvOutput::default()), ..Default::default() }, request_progress: None, @@ -2644,15 +2636,10 @@ mod test { .put_object(bucket, object, &mut reader, &Default::default()) .await .expect("put ScanRange header snapshot fixture"); - let result = make_store( - 8, - 8, - Some(FileHeaderInfo::from_static(FileHeaderInfo::USE)), - prepare_test_snapshot(bucket, object).await, - ) - .get_opts(&Path::from(object), GetOptions::default()) - .await - .expect("read ScanRange with a separate header read"); + let result = make_store(8, 8, Some(FileHeaderInfo::USE), prepare_test_snapshot(bucket, object).await) + .get_opts(&Path::from(object), GetOptions::default()) + .await + .expect("read ScanRange with a separate header read"); let GetResultPayload::Stream(stream) = result.payload else { panic!("expected streaming ScanRange header payload"); }; @@ -2743,16 +2730,16 @@ mod test { sse_customer_key_md5: None, request: SelectObjectContentRequest { expression: "SELECT * FROM s3object".to_string(), - expression_type: ExpressionType::from_static(ExpressionType::SQL), + expression_type: ExpressionType::SQL, input_serialization: InputSerialization { - csv: Some(CSVInput { + csv: Some(CsvInput { record_delimiter: Some("aa".to_string()), ..Default::default() }), ..Default::default() }, output_serialization: OutputSerialization { - csv: Some(CSVOutput::default()), + csv: Some(CsvOutput::default()), ..Default::default() }, request_progress: None, @@ -2839,9 +2826,9 @@ mod test { sse_customer_key_md5: None, request: SelectObjectContentRequest { expression: "SELECT * FROM s3object".to_string(), - expression_type: ExpressionType::from_static(ExpressionType::SQL), + expression_type: ExpressionType::SQL, input_serialization: InputSerialization { - csv: Some(CSVInput { + csv: Some(CsvInput { field_delimiter: Some("\r\n".to_string()), record_delimiter: Some("^Y".to_string()), ..Default::default() @@ -2849,7 +2836,7 @@ mod test { ..Default::default() }, output_serialization: OutputSerialization { - csv: Some(CSVOutput::default()), + csv: Some(CsvOutput::default()), ..Default::default() }, request_progress: None, @@ -3056,13 +3043,11 @@ mod test { .expect("write multiline CSV"); let mut input = (*csv_input(BUCKET, object)).clone(); let csv = input.request.input_serialization.csv.as_mut().expect("CSV input"); - csv.file_header_info = Some(FileHeaderInfo::from_static(FileHeaderInfo::USE)); + csv.file_header_info = Some(FileHeaderInfo::USE); csv.allow_quoted_record_delimiter = Some(true); - input.request.input_serialization.compression_type = compression.map(|format| { - CompressionType::from_static(match format { - CompressionFormat::Gzip => CompressionType::GZIP, - CompressionFormat::Bzip2 => CompressionType::BZIP2, - }) + input.request.input_serialization.compression_type = compression.map(|format| match format { + CompressionFormat::Gzip => CompressionType::GZIP, + CompressionFormat::Bzip2 => CompressionType::BZIP2, }); let metrics = Arc::new(SelectInputMetrics::default()); let store = EcObjectStore::build_with_snapshot( @@ -3125,13 +3110,11 @@ mod test { .expect("write boundary fixture"); let mut input = (*csv_input(BUCKET, &object)).clone(); let csv = input.request.input_serialization.csv.as_mut().expect("CSV input"); - csv.file_header_info = Some(FileHeaderInfo::from_static(FileHeaderInfo::NONE)); + csv.file_header_info = Some(FileHeaderInfo::NONE); csv.allow_quoted_record_delimiter = Some(true); - input.request.input_serialization.compression_type = compression.map(|format| { - CompressionType::from_static(match format { - CompressionFormat::Gzip => CompressionType::GZIP, - CompressionFormat::Bzip2 => CompressionType::BZIP2, - }) + input.request.input_serialization.compression_type = compression.map(|format| match format { + CompressionFormat::Gzip => CompressionType::GZIP, + CompressionFormat::Bzip2 => CompressionType::BZIP2, }); let metrics = Arc::new(SelectInputMetrics::default()); let store = EcObjectStore::build_with_snapshot( @@ -3196,15 +3179,13 @@ mod test { .expect("write Unicode CSV fixture"); let mut input = (*csv_input(BUCKET, object)).clone(); let csv = input.request.input_serialization.csv.as_mut().expect("CSV input"); - csv.file_header_info = Some(FileHeaderInfo::from_static(FileHeaderInfo::USE)); + csv.file_header_info = Some(FileHeaderInfo::USE); csv.quote_character = Some("ع".to_owned()); csv.quote_escape_character = Some("\\".to_owned()); csv.record_delimiter = Some("\n".to_owned()); - input.request.input_serialization.compression_type = compression.map(|format| { - CompressionType::from_static(match format { - CompressionFormat::Gzip => CompressionType::GZIP, - CompressionFormat::Bzip2 => CompressionType::BZIP2, - }) + input.request.input_serialization.compression_type = compression.map(|format| match format { + CompressionFormat::Gzip => CompressionType::GZIP, + CompressionFormat::Bzip2 => CompressionType::BZIP2, }); let start = HEADER.len() + SKIP.len(); if let Some(range_offset) = range_offset { @@ -3266,7 +3247,7 @@ mod test { .expect("put compressed CSV fixture"); let snapshot = prepare_test_snapshot(BUCKET, OBJECT).await; let mut input = (*csv_input(BUCKET, OBJECT)).clone(); - input.request.input_serialization.compression_type = Some(CompressionType::from_static(CompressionType::GZIP)); + input.request.input_serialization.compression_type = Some(CompressionType::GZIP); input.request.scan_range = Some(ScanRange { start: Some(0), end: None, @@ -3299,7 +3280,7 @@ mod test { assert_eq!(store.reader_open_count.load(Ordering::SeqCst), 0); let mut scan_input = (*csv_input(BUCKET, OBJECT)).clone(); - scan_input.request.input_serialization.compression_type = Some(CompressionType::from_static(CompressionType::GZIP)); + scan_input.request.input_serialization.compression_type = Some(CompressionType::GZIP); scan_input.request.scan_range = Some(ScanRange { start: Some(1), end: None, @@ -3364,7 +3345,7 @@ mod test { .expect("put compressed throughput fixture"); let snapshot = prepare_test_snapshot(BUCKET, OBJECT).await; let mut input = (*csv_input(BUCKET, OBJECT)).clone(); - input.request.input_serialization.compression_type = Some(CompressionType::from_static(CompressionType::GZIP)); + input.request.input_serialization.compression_type = Some(CompressionType::GZIP); let metrics = Arc::new(SelectInputMetrics::default()); let store = EcObjectStore::build_with_snapshot( Arc::new(input), @@ -3452,8 +3433,8 @@ mod test { let env = crate::storage_api::select_test_ecstore_env().await; env.make_bucket(BUCKET, false).await; for (object, data, json_type) in [ - (LINES_OBJECT, LINES_DATA, JSONType::LINES), - (DOCUMENT_OBJECT, DOCUMENT_DATA, JSONType::DOCUMENT), + (LINES_OBJECT, LINES_DATA, JsonType::LINES), + (DOCUMENT_OBJECT, DOCUMENT_DATA, JsonType::DOCUMENT), ] { let mut reader = SelectPutObjReader::from_vec(data.to_vec()); env.ecstore @@ -3462,13 +3443,13 @@ mod test { .expect("put JSON input metrics fixture"); let snapshot = prepare_test_snapshot(BUCKET, object).await; let input_metrics = Arc::new(SelectInputMetrics::default()); - let memory_limit = if json_type == JSONType::DOCUMENT { + let memory_limit = if json_type == JsonType::DOCUMENT { JSON_DOCUMENT_TEST_MEMORY_BYTES } else { 1024 * 1024 }; let store = EcObjectStore::build_with_snapshot( - json_input(BUCKET, object, json_type), + json_input(BUCKET, object, json_type.clone()), Arc::new(GreedyMemoryPool::new(memory_limit)), None, Arc::clone(&input_metrics), @@ -3516,7 +3497,7 @@ mod test { .csv .as_mut() .expect("CSV input") - .file_header_info = Some(FileHeaderInfo::from_static(FileHeaderInfo::USE)); + .file_header_info = Some(FileHeaderInfo::USE); input.request.scan_range = Some(ScanRange { start: Some(i64::try_from(RECORD_START).expect("fixture offset should fit in i64")), end: Some(i64::try_from(RECORD_START).expect("fixture offset should fit in i64")), diff --git a/crates/s3select-api/src/query/mod.rs b/crates/s3select-api/src/query/mod.rs index c09640752..0c861175b 100644 --- a/crates/s3select-api/src/query/mod.rs +++ b/crates/s3select-api/src/query/mod.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use s3s::dto::SelectObjectContentInput; +use crate::request::SelectObjectContentInput; use std::sync::Arc; use crate::{SelectInputMetrics, SelectObjectSnapshot}; @@ -92,7 +92,9 @@ impl Query { #[cfg(test)] fn test_query() -> Query { - use s3s::dto::{CSVInput, CSVOutput, ExpressionType, InputSerialization, OutputSerialization, SelectObjectContentRequest}; + use crate::request::{ + CsvInput, CsvOutput, ExpressionType, InputSerialization, OutputSerialization, SelectObjectContentRequest, + }; let input = SelectObjectContentInput { bucket: "bucket".to_string(), @@ -103,13 +105,13 @@ fn test_query() -> Query { sse_customer_key_md5: None, request: SelectObjectContentRequest { expression: "SELECT * FROM S3Object".to_string(), - expression_type: ExpressionType::from_static(ExpressionType::SQL), + expression_type: ExpressionType::SQL, input_serialization: InputSerialization { - csv: Some(CSVInput::default()), + csv: Some(CsvInput::default()), ..Default::default() }, output_serialization: OutputSerialization { - csv: Some(CSVOutput::default()), + csv: Some(CsvOutput::default()), ..Default::default() }, request_progress: None, diff --git a/crates/s3select-api/src/query/session.rs b/crates/s3select-api/src/query/session.rs index 1553d5731..e1a4aabe4 100644 --- a/crates/s3select-api/src/query/session.rs +++ b/crates/s3select-api/src/query/session.rs @@ -13,6 +13,7 @@ // limitations under the License. use crate::query::{Context, Query, ast::JsonSource}; +use crate::request::CompressionType; use crate::{ QueryError, QueryResult, SelectInputMetrics, SelectObjectSnapshot, object_store::{EcObjectStore, is_json_document_input, legacy_json_source_from_input}, @@ -30,7 +31,6 @@ use datafusion::{ prelude::SessionContext, }; use parking_lot::Mutex; -use s3s::dto::CompressionType; use std::sync::{ Arc, Weak, atomic::{AtomicU8, Ordering}, @@ -453,7 +453,7 @@ impl SessionCtxFactory { .input_serialization .compression_type .as_ref() - .is_some_and(|compression| compression.as_str() != CompressionType::NONE); + .is_some_and(|compression| *compression != CompressionType::NONE); let metered_input_requires_single_file_scan = input_metrics.is_some() && context.input.request.input_serialization.parquet.is_none(); let normalized_csv_requires_single_file_scan = @@ -610,6 +610,10 @@ fn test_parquet_batch( #[cfg(test)] mod tests { use super::*; + use crate::request::{ + CsvInput, CsvOutput, ExpressionType, InputSerialization, JsonInput, JsonType, OutputSerialization, ParquetInput, + ScanRange, SelectObjectContentInput, SelectObjectContentRequest, + }; use crate::storage_api::SelectPutObjReader; use crate::storage_api::object_store::ObjectIO as _; use datafusion::{ @@ -620,10 +624,6 @@ mod tests { execution::memory_pool::MemoryLimit, }; use http::HeaderMap; - use s3s::dto::{ - CSVInput, CSVOutput, ExpressionType, InputSerialization, JSONInput, JSONType, OutputSerialization, ParquetInput, - ScanRange, SelectObjectContentInput, SelectObjectContentRequest, - }; use std::io::Write as _; fn test_context() -> Context { @@ -637,13 +637,13 @@ mod tests { sse_customer_key_md5: None, request: SelectObjectContentRequest { expression: "SELECT * FROM S3Object".to_string(), - expression_type: ExpressionType::from_static("SQL"), + expression_type: ExpressionType::from("SQL"), input_serialization: InputSerialization { - csv: Some(CSVInput::default()), + csv: Some(CsvInput::default()), ..Default::default() }, output_serialization: OutputSerialization { - csv: Some(CSVOutput::default()), + csv: Some(CsvOutput::default()), ..Default::default() }, request_progress: None, @@ -687,8 +687,8 @@ mod tests { input.key = OBJECT.to_string(); input.request.expression = expression.to_string(); input.request.input_serialization.csv = None; - input.request.input_serialization.json = Some(JSONInput { - type_: Some(JSONType::from_static(JSONType::DOCUMENT)), + input.request.input_serialization.json = Some(JsonInput { + r#type: Some(JsonType::DOCUMENT), }); } env.make_bucket(bucket, false).await; @@ -790,7 +790,7 @@ mod tests { let mut json_context = test_context(); let json_request = &mut Arc::make_mut(&mut json_context.input).request; json_request.input_serialization.csv = None; - json_request.input_serialization.json = Some(JSONInput::default()); + json_request.input_serialization.json = Some(JsonInput::default()); for context in [&csv_context, &json_context] { let session = factory @@ -838,7 +838,7 @@ mod tests { end: Some(0), }); request.input_serialization.csv = None; - request.input_serialization.json = Some(JSONInput::default()); + request.input_serialization.json = Some(JsonInput::default()); let session = SessionCtxFactory::new(true) .with_target_partitions(2) @@ -854,7 +854,7 @@ mod tests { let mut context = test_context(); let request = &mut Arc::make_mut(&mut context.input).request; request.input_serialization.csv = None; - request.input_serialization.json = Some(JSONInput::default()); + request.input_serialization.json = Some(JsonInput::default()); let session = SessionCtxFactory::new(true) .with_target_partitions(2) @@ -868,8 +868,7 @@ mod tests { #[tokio::test] async fn compressed_input_disables_file_repartitioning_without_metrics() { let mut context = test_context(); - Arc::make_mut(&mut context.input).request.input_serialization.compression_type = - Some(CompressionType::from_static(CompressionType::GZIP)); + Arc::make_mut(&mut context.input).request.input_serialization.compression_type = Some(CompressionType::GZIP); let session = SessionCtxFactory::new(true) .with_target_partitions(2) @@ -885,8 +884,8 @@ mod tests { let mut context = test_context(); let request = &mut Arc::make_mut(&mut context.input).request; request.input_serialization.csv = None; - request.input_serialization.json = Some(JSONInput { - type_: Some(JSONType::from_static(JSONType::DOCUMENT)), + request.input_serialization.json = Some(JsonInput { + r#type: Some(JsonType::DOCUMENT), }); let session = SessionCtxFactory::new(true) diff --git a/crates/s3select-api/src/request.rs b/crates/s3select-api/src/request.rs new file mode 100644 index 000000000..bc730ef7d --- /dev/null +++ b/crates/s3select-api/src/request.rs @@ -0,0 +1,95 @@ +// 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. + +//! The SelectObjectContent request the query engine runs. +//! +//! Responsible for: the request shape the engine reads, built from the gateway +//! DTO shapes (rustfs/backlog#2746). Not responsible for: decoding the request +//! from the wire or authorizing it; the S3 edge converts its own request type +//! into this one. +//! +//! The SSE-C members keep the text the client sent, MD5 included, because the +//! engine forwards all three to the object read, which checks them. The gateway +//! operation input drops the MD5 after its own check, so it is not used here. + +use std::fmt; + +pub use rustfs_gateway_types::dto::{ + CompressionType, CsvInput, CsvOutput, ExpressionType, FileHeaderInfo, InputSerialization, JsonInput, JsonOutput, + OutputSerialization, ParquetInput, QuoteFields, RequestProgress, ScanRange, Type as JsonType, +}; + +/// The SSE-C headers the engine forwards to the object read. +pub const X_AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_ALGORITHM: http::HeaderName = + http::HeaderName::from_static("x-amz-server-side-encryption-customer-algorithm"); +pub const X_AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY: http::HeaderName = + http::HeaderName::from_static("x-amz-server-side-encryption-customer-key"); +pub const X_AMZ_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY_MD5: http::HeaderName = + http::HeaderName::from_static("x-amz-server-side-encryption-customer-key-md5"); + +/// One SelectObjectContent request: the object to read and the query to run. +#[derive(Clone, Default)] +pub struct SelectObjectContentInput { + pub bucket: String, + pub expected_bucket_owner: Option, + pub key: String, + pub sse_customer_algorithm: Option, + pub sse_customer_key: Option, + pub sse_customer_key_md5: Option, + pub request: SelectObjectContentRequest, +} + +/// The query part of a SelectObjectContent request. +#[derive(Clone, Debug, Default)] +pub struct SelectObjectContentRequest { + pub expression: String, + pub expression_type: ExpressionType, + pub input_serialization: InputSerialization, + pub output_serialization: OutputSerialization, + pub request_progress: Option, + pub scan_range: Option, +} + +impl fmt::Debug for SelectObjectContentInput { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let redacted = |value: &Option| value.as_ref().map(|_| ""); + f.debug_struct("SelectObjectContentInput") + .field("bucket", &self.bucket) + .field("expected_bucket_owner", &self.expected_bucket_owner) + .field("key", &self.key) + .field("sse_customer_algorithm", &self.sse_customer_algorithm) + .field("sse_customer_key", &redacted(&self.sse_customer_key)) + .field("sse_customer_key_md5", &redacted(&self.sse_customer_key_md5)) + .field("request", &self.request) + .finish() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn debug_redacts_the_customer_key_and_its_digest() { + let input = SelectObjectContentInput { + sse_customer_algorithm: Some("AES256".to_owned()), + sse_customer_key: Some("secret-customer-key".to_owned()), + sse_customer_key_md5: Some("secret-customer-key-md5".to_owned()), + ..Default::default() + }; + let debug = format!("{input:?}"); + assert!(!debug.contains("secret-customer-key"), "{debug}"); + assert!(debug.contains("AES256"), "{debug}"); + } +} diff --git a/crates/s3select-query/Cargo.toml b/crates/s3select-query/Cargo.toml index 3fbf7778c..883ee5842 100644 --- a/crates/s3select-query/Cargo.toml +++ b/crates/s3select-query/Cargo.toml @@ -49,7 +49,6 @@ datafusion = { workspace = true, default-features = false, features = ["parquet" derive_builder = { workspace = true } futures = { workspace = true } parking_lot = { workspace = true } -s3s = { workspace = true, features = ["minio"] } serde.workspace = true serde_json.workspace = true tokio = { workspace = true, features = ["fs", "rt-multi-thread", "sync", "time"] } diff --git a/crates/s3select-query/src/dispatcher/manager.rs b/crates/s3select-query/src/dispatcher/manager.rs index 06e21393c..c22f42801 100644 --- a/crates/s3select-query/src/dispatcher/manager.rs +++ b/crates/s3select-query/src/dispatcher/manager.rs @@ -37,6 +37,7 @@ use datafusion::{ }; use futures::Stream; use parking_lot::Mutex; +use rustfs_s3select_api::request::{FileHeaderInfo, JsonType, SelectObjectContentInput}; use rustfs_s3select_api::{ QueryError, QueryResult, SelectError, query::{ @@ -53,7 +54,6 @@ use rustfs_s3select_api::{ }, }, }; -use s3s::dto::{FileHeaderInfo, JSONType, SelectObjectContentInput}; use std::sync::LazyLock; use tokio::{ sync::Semaphore, @@ -69,9 +69,9 @@ use crate::{ sql::planner::prepare_s3_select_statement, }; -static IGNORE: LazyLock = LazyLock::new(|| FileHeaderInfo::from_static(FileHeaderInfo::IGNORE)); -static NONE: LazyLock = LazyLock::new(|| FileHeaderInfo::from_static(FileHeaderInfo::NONE)); -static USE: LazyLock = LazyLock::new(|| FileHeaderInfo::from_static(FileHeaderInfo::USE)); +static IGNORE: LazyLock = LazyLock::new(|| FileHeaderInfo::IGNORE); +static NONE: LazyLock = LazyLock::new(|| FileHeaderInfo::NONE); +static USE: LazyLock = LazyLock::new(|| FileHeaderInfo::USE); const EXACT_OBJECT_FILE_EXTENSION: &str = ""; #[derive(Clone)] @@ -555,8 +555,8 @@ fn is_json_document_input(input: &SelectObjectContentInput) -> bool { .input_serialization .json .as_ref() - .and_then(|json| json.type_.as_ref()) - .is_some_and(|json_type| json_type.as_str() == JSONType::DOCUMENT) + .and_then(|json| json.r#type.as_ref()) + .is_some_and(|json_type| *json_type == JsonType::DOCUMENT) } fn validate_json_source_path_input(input: &SelectObjectContentInput, source_path: &[JsonPathSegment]) -> QueryResult<()> { @@ -864,6 +864,10 @@ mod tests { physical_plan::{RecordBatchStream, stream::RecordBatchStreamAdapter}, }; use futures::{StreamExt, TryStreamExt, stream}; + use rustfs_s3select_api::request::{ + CsvInput, CsvOutput, ExpressionType, FileHeaderInfo, InputSerialization, JsonInput, JsonOutput, JsonType, + OutputSerialization, ParquetInput, SelectObjectContentInput, SelectObjectContentRequest, + }; use rustfs_s3select_api::{ QueryError, QueryResult, SelectError, query::{ @@ -881,10 +885,6 @@ mod tests { }, }; use rustfs_test_utils::{PutObjectCommitBarrier, TestECStoreEnv}; - use s3s::dto::{ - CSVInput, CSVOutput, ExpressionType, FileHeaderInfo, InputSerialization, JSONInput, JSONOutput, JSONType, - OutputSerialization, ParquetInput, SelectObjectContentInput, SelectObjectContentRequest, - }; use std::{ pin::Pin, sync::{ @@ -1078,16 +1078,16 @@ mod tests { sse_customer_key_md5: None, request: SelectObjectContentRequest { expression: "SELECT * FROM S3Object".to_string(), - expression_type: ExpressionType::from_static(ExpressionType::SQL), + expression_type: ExpressionType::SQL, input_serialization: InputSerialization { - csv: Some(CSVInput { - file_header_info: Some(FileHeaderInfo::from_static(FileHeaderInfo::USE)), + csv: Some(CsvInput { + file_header_info: Some(FileHeaderInfo::USE), ..Default::default() }), ..Default::default() }, output_serialization: OutputSerialization { - csv: Some(CSVOutput::default()), + csv: Some(CsvOutput::default()), ..Default::default() }, request_progress: None, @@ -1243,15 +1243,15 @@ mod tests { sse_customer_key_md5: None, request: SelectObjectContentRequest { expression: "SELECT old_marker FROM S3Object".to_string(), - expression_type: ExpressionType::from_static(ExpressionType::SQL), + expression_type: ExpressionType::SQL, input_serialization: InputSerialization { - json: Some(JSONInput { - type_: Some(JSONType::from_static(JSONType::LINES)), + json: Some(JsonInput { + r#type: Some(JsonType::LINES), }), ..Default::default() }, output_serialization: OutputSerialization { - json: Some(JSONOutput::default()), + json: Some(JsonOutput::default()), ..Default::default() }, request_progress: None, @@ -1284,7 +1284,7 @@ mod tests { } let mut document_input = (*lines_input).clone(); - document_input.request.input_serialization.json.as_mut().unwrap().type_ = Some(JSONType::from_static(JSONType::DOCUMENT)); + document_input.request.input_serialization.json.as_mut().unwrap().r#type = Some(JsonType::DOCUMENT); assert!(validate_json_source_path_input(&document_input, &nested_path).is_ok()); } @@ -1332,13 +1332,13 @@ mod tests { sse_customer_key_md5: None, request: SelectObjectContentRequest { expression: "SELECT old_marker FROM S3Object".to_string(), - expression_type: ExpressionType::from_static(ExpressionType::SQL), + expression_type: ExpressionType::SQL, input_serialization: InputSerialization { parquet: Some(ParquetInput {}), ..Default::default() }, output_serialization: OutputSerialization { - json: Some(JSONOutput::default()), + json: Some(JsonOutput::default()), ..Default::default() }, request_progress: None, @@ -1493,7 +1493,7 @@ mod tests { .csv .as_mut() .expect("test input should use CSV") - .file_header_info = Some(FileHeaderInfo::from_static("INVALID")); + .file_header_info = Some(FileHeaderInfo::from("INVALID")); let input = Arc::new(input); let optimizer = Arc::new(CascadeOptimizerBuilder::default().build()); let scheduler = Arc::new(LocalScheduler {}); @@ -1546,7 +1546,7 @@ mod tests { input.bucket = format!("select-unicode-quotes-{index}"); input.key = "records".to_owned(); let csv = input.request.input_serialization.csv.as_mut().expect("CSV input"); - csv.file_header_info = Some(FileHeaderInfo::from_static(FileHeaderInfo::NONE)); + csv.file_header_info = Some(FileHeaderInfo::NONE); csv.quote_character = Some(quote.to_owned()); csv.quote_escape_character = Some(escape.to_owned()); csv.field_delimiter = Some(field.to_owned()); @@ -1609,11 +1609,11 @@ mod tests { for (index, (quote, escape, field, record, data, expected)) in cases.into_iter().enumerate() { for header in [FileHeaderInfo::NONE, FileHeaderInfo::USE, FileHeaderInfo::IGNORE] { let mut input = test_input(); - input.bucket = format!("select-quoted-records-{index}-{}", header.to_ascii_lowercase()); + input.bucket = format!("select-quoted-records-{index}-{}", header.as_str().to_ascii_lowercase()); input.key = "records".to_owned(); let csv = input.request.input_serialization.csv.as_mut().expect("CSV input"); csv.allow_quoted_record_delimiter = Some(true); - csv.file_header_info = Some(FileHeaderInfo::from_static(header)); + csv.file_header_info = Some(header.clone()); if index != 0 { csv.quote_character = Some(quote.to_owned()); csv.quote_escape_character = Some(escape.to_owned()); @@ -1674,8 +1674,8 @@ mod tests { input.key = key.to_owned(); let data = if json { input.request.input_serialization.csv = None; - input.request.input_serialization.json = Some(s3s::dto::JSONInput { - type_: Some(JSONType::from_static(JSONType::LINES)), + input.request.input_serialization.json = Some(rustfs_s3select_api::request::JsonInput { + r#type: Some(JsonType::LINES), }); b"{\"value\":\"selected\"}\n".as_slice() } else { @@ -1738,7 +1738,7 @@ mod tests { .csv .as_mut() .expect("test input should use CSV serialization"); - csv.file_header_info = Some(FileHeaderInfo::from_static(FileHeaderInfo::NONE)); + csv.file_header_info = Some(FileHeaderInfo::NONE); csv.record_delimiter = Some("^".to_string()); let input = Arc::new(input); let admission = Arc::new(Semaphore::new(1)); @@ -1889,13 +1889,13 @@ mod tests { input.key = "test.json".to_string(); input.request.expression = "SELECT * FROM S3Object.employees".to_string(); input.request.input_serialization = InputSerialization { - json: Some(JSONInput { - type_: Some(JSONType::from_static(JSONType::DOCUMENT)), + json: Some(JsonInput { + r#type: Some(JsonType::DOCUMENT), }), ..Default::default() }; input.request.output_serialization = OutputSerialization { - json: Some(JSONOutput::default()), + json: Some(JsonOutput::default()), ..Default::default() }; let input = Arc::new(input); @@ -1953,13 +1953,13 @@ mod tests { input.key = "input.json".to_string(); input.request.expression = expression.to_string(); input.request.input_serialization = InputSerialization { - json: Some(JSONInput { - type_: Some(JSONType::from_static(JSONType::DOCUMENT)), + json: Some(JsonInput { + r#type: Some(JsonType::DOCUMENT), }), ..Default::default() }; input.request.output_serialization = OutputSerialization { - json: Some(JSONOutput::default()), + json: Some(JsonOutput::default()), ..Default::default() }; let input = Arc::new(input); @@ -2025,13 +2025,13 @@ mod tests { input.key = "input.json".to_string(); input.request.expression = "SELECT _1 FROM S3Object[*]".to_string(); input.request.input_serialization = InputSerialization { - json: Some(JSONInput { - type_: Some(JSONType::from_static(JSONType::DOCUMENT)), + json: Some(JsonInput { + r#type: Some(JsonType::DOCUMENT), }), ..Default::default() }; input.request.output_serialization = OutputSerialization { - json: Some(JSONOutput::default()), + json: Some(JsonOutput::default()), ..Default::default() }; let input = Arc::new(input); @@ -2506,7 +2506,7 @@ mod tests { input_mut.bucket = "s3select-invalid-sql-precedence".to_string(); input_mut.key = "malformed.json".to_string(); input_mut.request.expression = "SELECT * FROM".to_string(); - input_mut.request.input_serialization.json.as_mut().unwrap().type_ = Some(JSONType::from_static(JSONType::DOCUMENT)); + input_mut.request.input_serialization.json.as_mut().unwrap().r#type = Some(JsonType::DOCUMENT); let env = snapshot_test_env().await; env.make_bucket(&input.bucket, false).await; diff --git a/crates/s3select-query/src/dispatcher/parquet_table.rs b/crates/s3select-query/src/dispatcher/parquet_table.rs index 97949371c..e6b568edd 100644 --- a/crates/s3select-query/src/dispatcher/parquet_table.rs +++ b/crates/s3select-query/src/dispatcher/parquet_table.rs @@ -37,11 +37,11 @@ use datafusion::{ physical_plan::ExecutionPlan, }; use futures::{FutureExt, TryFutureExt, future::BoxFuture}; +use rustfs_s3select_api::request::SelectObjectContentInput; use rustfs_s3select_api::{ QueryResult, object_store::{SelectScanRange, scan_range_from_bounds}, }; -use s3s::dto::SelectObjectContentInput; #[derive(Clone)] pub struct ParquetSelectTable { @@ -232,8 +232,8 @@ mod tests { prelude::SessionContext, }; use rustfs_s3select_api::SelectError; - use s3s::dto::{ - CSVOutput, ExpressionType, InputSerialization, OutputSerialization, ParquetInput, ScanRange, SelectObjectContentRequest, + use rustfs_s3select_api::request::{ + CsvOutput, ExpressionType, InputSerialization, OutputSerialization, ParquetInput, ScanRange, SelectObjectContentRequest, }; use std::{ fs::File, @@ -345,13 +345,13 @@ mod tests { sse_customer_key_md5: None, request: SelectObjectContentRequest { expression: "SELECT * FROM S3Object".to_string(), - expression_type: ExpressionType::from_static(ExpressionType::SQL), + expression_type: ExpressionType::SQL, input_serialization: InputSerialization { parquet: Some(ParquetInput::default()), ..Default::default() }, output_serialization: OutputSerialization { - csv: Some(CSVOutput::default()), + csv: Some(CsvOutput::default()), ..Default::default() }, request_progress: None, diff --git a/crates/s3select-query/src/instance.rs b/crates/s3select-query/src/instance.rs index b01dc1afb..e67d125cc 100644 --- a/crates/s3select-query/src/instance.rs +++ b/crates/s3select-query/src/instance.rs @@ -19,6 +19,7 @@ use std::{ use async_trait::async_trait; use derive_builder::Builder; +use rustfs_s3select_api::request::SelectObjectContentInput; use rustfs_s3select_api::{ QueryResult, query::{ @@ -30,7 +31,6 @@ use rustfs_s3select_api::{ }, server::dbms::{DatabaseManagerSystem, QueryHandle}, }; -use s3s::dto::SelectObjectContentInput; use tokio::sync::Semaphore; use crate::{ @@ -250,6 +250,10 @@ mod tests { use async_trait::async_trait; use datafusion::{arrow::util::pretty, assert_batches_eq}; use parking_lot::Mutex; + use rustfs_s3select_api::request::{ + CsvInput, CsvOutput, ExpressionType, FileHeaderInfo, InputSerialization, OutputSerialization, SelectObjectContentInput, + SelectObjectContentRequest, + }; use rustfs_s3select_api::{ QueryResult, SelectInputMetrics, query::{ @@ -260,10 +264,6 @@ mod tests { }, server::dbms::DatabaseManagerSystem, }; - use s3s::dto::{ - CSVInput, CSVOutput, ExpressionType, FieldDelimiter, FileHeaderInfo, InputSerialization, OutputSerialization, - RecordDelimiter, SelectObjectContentInput, SelectObjectContentRequest, - }; use crate::get_global_db; @@ -312,13 +312,13 @@ mod tests { sse_customer_key_md5: None, request: SelectObjectContentRequest { expression: expression.to_string(), - expression_type: ExpressionType::from_static(ExpressionType::SQL), + expression_type: ExpressionType::SQL, input_serialization: InputSerialization { - csv: Some(CSVInput::default()), + csv: Some(CsvInput::default()), ..Default::default() }, output_serialization: OutputSerialization { - csv: Some(CSVOutput::default()), + csv: Some(CsvOutput::default()), ..Default::default() }, request_progress: None, @@ -391,16 +391,16 @@ mod tests { sse_customer_key_md5: None, request: SelectObjectContentRequest { expression: sql.to_string(), - expression_type: ExpressionType::from_static("SQL"), + expression_type: ExpressionType::from("SQL"), input_serialization: InputSerialization { - csv: Some(CSVInput { - file_header_info: Some(FileHeaderInfo::from_static(FileHeaderInfo::USE)), + csv: Some(CsvInput { + file_header_info: Some(FileHeaderInfo::USE), ..Default::default() }), ..Default::default() }, output_serialization: OutputSerialization { - csv: Some(CSVOutput::default()), + csv: Some(CsvOutput::default()), ..Default::default() }, request_progress: None, @@ -453,18 +453,18 @@ mod tests { sse_customer_key_md5: None, request: SelectObjectContentRequest { expression: sql.to_string(), - expression_type: ExpressionType::from_static("SQL"), + expression_type: ExpressionType::from("SQL"), input_serialization: InputSerialization { - csv: Some(CSVInput { - file_header_info: Some(FileHeaderInfo::from_static(FileHeaderInfo::IGNORE)), - field_delimiter: Some(FieldDelimiter::from("╦")), - record_delimiter: Some(RecordDelimiter::from("\n")), + csv: Some(CsvInput { + file_header_info: Some(FileHeaderInfo::IGNORE), + field_delimiter: Some("╦".to_owned()), + record_delimiter: Some("\n".to_owned()), ..Default::default() }), ..Default::default() }, output_serialization: OutputSerialization { - csv: Some(CSVOutput::default()), + csv: Some(CsvOutput::default()), ..Default::default() }, request_progress: None, diff --git a/crates/s3select-query/src/lib.rs b/crates/s3select-query/src/lib.rs index fade8f9bb..76d649860 100644 --- a/crates/s3select-query/src/lib.rs +++ b/crates/s3select-query/src/lib.rs @@ -25,8 +25,8 @@ pub mod sql; #[cfg(test)] mod test; +use rustfs_s3select_api::request::SelectObjectContentInput; use rustfs_s3select_api::{QueryResult, server::dbms::DatabaseManagerSystem}; -use s3s::dto::SelectObjectContentInput; use std::sync::{Arc, LazyLock}; use crate::{ @@ -89,11 +89,11 @@ pub async fn create_fresh_db() -> QueryResult SelectObjectContentInput { @@ -34,15 +34,15 @@ mod error_handling_tests { sse_customer_key_md5: None, request: SelectObjectContentRequest { expression: sql.to_string(), - expression_type: ExpressionType::from_static("SQL"), + expression_type: ExpressionType::from("SQL"), input_serialization: InputSerialization { - csv: Some(CSVInput { - file_header_info: Some(FileHeaderInfo::from_static(FileHeaderInfo::USE)), + csv: Some(CsvInput { + file_header_info: Some(FileHeaderInfo::USE), ..Default::default() }), ..Default::default() }, - output_serialization: s3s::dto::OutputSerialization::default(), + output_serialization: rustfs_s3select_api::request::OutputSerialization::default(), request_progress: None, scan_range: None, }, diff --git a/crates/s3select-query/src/test/integration_test.rs b/crates/s3select-query/src/test/integration_test.rs index 461dbedb3..a3597df74 100644 --- a/crates/s3select-query/src/test/integration_test.rs +++ b/crates/s3select-query/src/test/integration_test.rs @@ -19,14 +19,14 @@ mod integration_tests { array::{Array, Int64Array, StringArray}, record_batch::RecordBatch, }; + use rustfs_s3select_api::request::{ + CsvInput, CsvOutput, ExpressionType, FileHeaderInfo, InputSerialization, JsonInput, JsonOutput, JsonType, + OutputSerialization, ParquetInput, ScanRange, SelectObjectContentInput, SelectObjectContentRequest, + }; use rustfs_s3select_api::{ QueryError, query::{Context, Query}, }; - use s3s::dto::{ - CSVInput, CSVOutput, ExpressionType, FileHeaderInfo, InputSerialization, JSONInput, JSONOutput, JSONType, - OutputSerialization, ParquetInput, ScanRange, SelectObjectContentInput, SelectObjectContentRequest, - }; use std::sync::Arc; fn assert_ages_descending(output: &[RecordBatch]) { @@ -85,16 +85,16 @@ mod integration_tests { sse_customer_key_md5: None, request: SelectObjectContentRequest { expression: sql.to_string(), - expression_type: ExpressionType::from_static("SQL"), + expression_type: ExpressionType::from("SQL"), input_serialization: InputSerialization { - csv: Some(CSVInput { - file_header_info: Some(FileHeaderInfo::from_static(FileHeaderInfo::USE)), + csv: Some(CsvInput { + file_header_info: Some(FileHeaderInfo::USE), ..Default::default() }), ..Default::default() }, output_serialization: OutputSerialization { - csv: Some(CSVOutput::default()), + csv: Some(CsvOutput::default()), ..Default::default() }, request_progress: None, @@ -104,7 +104,7 @@ mod integration_tests { } /// Build a `SelectObjectContentInput` targeting a JSON DOCUMENT file. - /// Uses `JSONType::DOCUMENT`, which keeps the document-style validation path. + /// Uses `JsonType::DOCUMENT`, which keeps the document-style validation path. fn create_test_json_input(sql: &str) -> SelectObjectContentInput { SelectObjectContentInput { bucket: "test-bucket".to_string(), @@ -115,15 +115,15 @@ mod integration_tests { sse_customer_key_md5: None, request: SelectObjectContentRequest { expression: sql.to_string(), - expression_type: ExpressionType::from_static("SQL"), + expression_type: ExpressionType::from("SQL"), input_serialization: InputSerialization { - json: Some(JSONInput { - type_: Some(JSONType::from_static(JSONType::DOCUMENT)), + json: Some(JsonInput { + r#type: Some(JsonType::DOCUMENT), }), ..Default::default() }, output_serialization: OutputSerialization { - json: Some(JSONOutput::default()), + json: Some(JsonOutput::default()), ..Default::default() }, request_progress: None, @@ -142,15 +142,15 @@ mod integration_tests { sse_customer_key_md5: None, request: SelectObjectContentRequest { expression: sql.to_string(), - expression_type: ExpressionType::from_static("SQL"), + expression_type: ExpressionType::from("SQL"), input_serialization: InputSerialization { - json: Some(JSONInput { - type_: Some(JSONType::from_static(JSONType::LINES)), + json: Some(JsonInput { + r#type: Some(JsonType::LINES), }), ..Default::default() }, output_serialization: OutputSerialization { - json: Some(JSONOutput::default()), + json: Some(JsonOutput::default()), ..Default::default() }, request_progress: None, @@ -169,13 +169,13 @@ mod integration_tests { sse_customer_key_md5: None, request: SelectObjectContentRequest { expression: sql.to_string(), - expression_type: ExpressionType::from_static("SQL"), + expression_type: ExpressionType::from("SQL"), input_serialization: InputSerialization { parquet: Some(ParquetInput {}), ..Default::default() }, output_serialization: OutputSerialization { - json: Some(JSONOutput::default()), + json: Some(JsonOutput::default()), ..Default::default() }, request_progress: None, @@ -259,7 +259,7 @@ mod integration_tests { .csv .as_mut() .expect("CSV input should be configured") - .file_header_info = Some(FileHeaderInfo::from_static(header)); + .file_header_info = Some(header.clone()); let db = get_global_db(input.clone(), true).await.expect("create CSV test database"); let query = Query::new(Context { input: Arc::new(input) }, sql.to_string()); let batches = db diff --git a/rustfs/Cargo.toml b/rustfs/Cargo.toml index 78eb4a4d2..47752f44c 100644 --- a/rustfs/Cargo.toml +++ b/rustfs/Cargo.toml @@ -250,7 +250,7 @@ rustfs-notify = { workspace = true } rustfs-obs = { workspace = true } rustfs-policy = { workspace = true } rustfs-protocols = { workspace = true } -rustfs-gateway-types = { workspace = true } +rustfs-gateway-types = { workspace = true, features = ["compat-s3s-rustfs"] } rustfs-protos = { workspace = true } rustfs-rio = { workspace = true } rustfs-s3-client = { workspace = true } diff --git a/rustfs/src/app/context.rs b/rustfs/src/app/context.rs index 935eee6a8..890c69a01 100644 --- a/rustfs/src/app/context.rs +++ b/rustfs/src/app/context.rs @@ -48,9 +48,9 @@ use rustfs_iam::{ use rustfs_io_metrics::{PerformanceMetrics, internode_metrics::InternodeMetrics}; use rustfs_kms::{KmsServiceManager, ObjectEncryptionService}; use rustfs_lock::LockClient; +use rustfs_s3select_api::request::SelectObjectContentInput; use rustfs_s3select_api::{QueryResult, server::dbms::DatabaseManagerSystem}; use rustfs_tls_runtime::{GlobalPublishedOutboundTlsState, TlsGeneration}; -use s3s::dto::SelectObjectContentInput; use std::{collections::HashMap, sync::Arc, time::SystemTime}; use tokio::sync::RwLock; @@ -974,11 +974,11 @@ mod tests { sse_customer_algorithm: None, sse_customer_key: None, sse_customer_key_md5: None, - request: s3s::dto::SelectObjectContentRequest { + request: rustfs_s3select_api::request::SelectObjectContentRequest { expression: "SELECT * FROM S3Object".to_string(), - expression_type: s3s::dto::ExpressionType::from_static("SQL"), - input_serialization: s3s::dto::InputSerialization::default(), - output_serialization: s3s::dto::OutputSerialization::default(), + expression_type: rustfs_s3select_api::request::ExpressionType::SQL, + input_serialization: rustfs_s3select_api::request::InputSerialization::default(), + output_serialization: rustfs_s3select_api::request::OutputSerialization::default(), request_progress: None, scan_range: None, }, diff --git a/rustfs/src/app/context/handles.rs b/rustfs/src/app/context/handles.rs index 2968af533..6c72df0c2 100644 --- a/rustfs/src/app/context/handles.rs +++ b/rustfs/src/app/context/handles.rs @@ -37,10 +37,10 @@ use rustfs_io_metrics::{PerformanceMetrics, internode_metrics::InternodeMetrics} use rustfs_kms::KmsServiceManager; use rustfs_lock::LockClient; use rustfs_notify::{EventArgs, NotificationError}; +use rustfs_s3select_api::request::SelectObjectContentInput; use rustfs_s3select_api::{QueryResult, server::dbms::DatabaseManagerSystem}; use rustfs_targets::{EventName, arn::TargetID}; use rustfs_tls_runtime::{GlobalPublishedOutboundTlsState, TlsGeneration}; -use s3s::dto::SelectObjectContentInput; use std::{ collections::HashMap, sync::{Arc, OnceLock, RwLock as StdRwLock}, diff --git a/rustfs/src/app/context/interfaces.rs b/rustfs/src/app/context/interfaces.rs index 43052f5ad..d773ee956 100644 --- a/rustfs/src/app/context/interfaces.rs +++ b/rustfs/src/app/context/interfaces.rs @@ -27,10 +27,10 @@ use rustfs_io_metrics::{PerformanceMetrics, internode_metrics::InternodeMetrics} use rustfs_kms::KmsServiceManager; use rustfs_lock::LockClient; use rustfs_notify::{EventArgs, NotificationError}; +use rustfs_s3select_api::request::SelectObjectContentInput; use rustfs_s3select_api::{QueryResult, server::dbms::DatabaseManagerSystem}; use rustfs_targets::{EventName, arn::TargetID}; use rustfs_tls_runtime::{GlobalPublishedOutboundTlsState, TlsGeneration}; -use s3s::dto::SelectObjectContentInput; use std::{collections::HashMap, sync::Arc, time::SystemTime}; use tokio::sync::RwLock; diff --git a/rustfs/src/app/context/runtime_sources.rs b/rustfs/src/app/context/runtime_sources.rs index 4831cd509..15f420e1d 100644 --- a/rustfs/src/app/context/runtime_sources.rs +++ b/rustfs/src/app/context/runtime_sources.rs @@ -33,12 +33,12 @@ use rustfs_io_metrics::{ use rustfs_kms::KmsServiceManager; use rustfs_lock::LockClient; use rustfs_notify::{EventArgs, NotificationError, notifier_global}; +use rustfs_s3select_api::request::SelectObjectContentInput; use rustfs_s3select_api::{QueryResult, server::dbms::DatabaseManagerSystem}; use rustfs_targets::{EventName, arn::TargetID}; use rustfs_tls_runtime::{ GlobalPublishedOutboundTlsState, TlsGeneration, load_global_outbound_tls_generation, load_global_outbound_tls_state, }; -use s3s::dto::SelectObjectContentInput; use std::{collections::HashMap, sync::Arc, time::SystemTime}; use tokio::sync::RwLock; diff --git a/rustfs/src/app/mod.rs b/rustfs/src/app/mod.rs index c00b015fd..0adc53b58 100644 --- a/rustfs/src/app/mod.rs +++ b/rustfs/src/app/mod.rs @@ -27,6 +27,7 @@ pub(crate) mod object_traffic_health; pub mod object_usecase; pub(crate) mod runtime_sources; mod select_object; +mod select_request; pub(crate) mod storage_api; pub(crate) mod table_list_isolation; pub(crate) mod trailer_adapter; diff --git a/rustfs/src/app/runtime_sources.rs b/rustfs/src/app/runtime_sources.rs index 389d97517..be440b176 100644 --- a/rustfs/src/app/runtime_sources.rs +++ b/rustfs/src/app/runtime_sources.rs @@ -22,8 +22,8 @@ pub(crate) use crate::runtime_sources::{ current_notification_system_for_context, current_object_data_cache_handle_for_context, current_object_store_handle_for_context, }; +use rustfs_s3select_api::request::SelectObjectContentInput; use rustfs_s3select_api::{QueryResult, server::dbms::DatabaseManagerSystem}; -use s3s::dto::SelectObjectContentInput; use std::sync::Arc; #[cfg(test)] use tokio::sync::RwLock; diff --git a/rustfs/src/app/select_object.rs b/rustfs/src/app/select_object.rs index 9a51272f5..1d40963c9 100644 --- a/rustfs/src/app/select_object.rs +++ b/rustfs/src/app/select_object.rs @@ -8,6 +8,7 @@ use super::storage_api::select_object::{ validate_ssec_for_read, }; use crate::app::runtime_sources::current_s3select_db; +use crate::app::select_request::select_input_from_s3s; use crate::error::ApiError; use bytes::{Bytes, BytesMut}; use datafusion::arrow::{ @@ -124,8 +125,11 @@ pub async fn execute_select_object_content( log_select_request_summary(&input, &validation); let query_timeout = s3_select_query_timeout(); let query_deadline = Instant::now() + query_timeout; + let engine_input = Arc::new(select_input_from_s3s(input.clone()).map_err(|err| { + S3Error::with_message(S3ErrorCode::InternalError, format!("convert the SelectObjectContent request: {err}")) + })?); let input = Arc::new(input); - let db = timeout_at(query_deadline, current_s3select_db((*input).clone(), false)) + let db = timeout_at(query_deadline, current_s3select_db((*engine_input).clone(), false)) .await .map_err(|_| select_query_timeout_error(query_timeout.as_secs()))? .map_err(map_query_error_to_s3)?; @@ -143,7 +147,7 @@ pub async fn execute_select_object_content( } let snapshot = Arc::new(snapshot); let query = - Query::new_with_snapshot(Context { input: input.clone() }, input.request.expression.clone(), Arc::clone(&snapshot)); + Query::new_with_snapshot(Context { input: engine_input }, input.request.expression.clone(), Arc::clone(&snapshot)); let query_handle = timeout_at(query_deadline, db.execute_admitted(&query, admission)) .await .map_err(|_| select_query_timeout_error(query_timeout.as_secs()))? @@ -607,7 +611,12 @@ fn validate_select_request(headers: &http::HeaderMap, input: &mut SelectObjectCo normalize_input_serialization(&mut input.request.input_serialization)?; let compressed_input = is_compressed_input(&input.request.input_serialization); if compressed_input - && input.request.scan_range.as_ref().is_some_and(is_noop_scan_range) + && input.request.scan_range.as_ref().is_some_and(|range| { + is_noop_scan_range(&rustfs_s3select_api::request::ScanRange { + start: range.start, + end: range.end, + }) + }) && !input .request .input_serialization diff --git a/rustfs/src/app/select_request.rs b/rustfs/src/app/select_request.rs new file mode 100644 index 000000000..26990f064 --- /dev/null +++ b/rustfs/src/app/select_request.rs @@ -0,0 +1,176 @@ +// 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. + +//! SelectObjectContent requests across the legacy S3 edge. +//! +//! Responsible for: converting the s3s request the edge decoded into the +//! request the query engine runs (rustfs/backlog#2746), through the gateway's +//! s3s seam for the serialization shapes. Not responsible for: validating the +//! request; the select use case does that before the engine is built. +//! DELETE BY T4.1 (rustfs/backlog#2783) with the rest of the legacy edge. + +use rustfs_gateway_types::compat::ConversionError; +use rustfs_gateway_types::compat::s3s_5761ddfe::generated::shapes::input_serialization::input_serialization_from_s3s; +use rustfs_gateway_types::compat::s3s_5761ddfe::generated::shapes::output_serialization::output_serialization_from_s3s; +use rustfs_s3select_api::request as engine; + +/// Converts the decoded request, keeping the SSE-C members as the client sent them. +pub(crate) fn select_input_from_s3s( + input: s3s::dto::SelectObjectContentInput, +) -> Result { + let s3s::dto::SelectObjectContentInput { + bucket, + expected_bucket_owner, + key, + sse_customer_algorithm, + sse_customer_key, + sse_customer_key_md5, + request, + } = input; + let s3s::dto::SelectObjectContentRequest { + expression, + expression_type, + input_serialization, + output_serialization, + request_progress, + scan_range, + } = request; + Ok(engine::SelectObjectContentInput { + bucket, + expected_bucket_owner, + key, + sse_customer_algorithm, + sse_customer_key, + sse_customer_key_md5, + request: engine::SelectObjectContentRequest { + expression, + expression_type: engine::ExpressionType::custom(expression_type.as_str().to_owned()), + input_serialization: input_serialization_from_s3s(input_serialization)?, + output_serialization: output_serialization_from_s3s(output_serialization)?, + request_progress: request_progress.map(|progress| engine::RequestProgress { + enabled: progress.enabled, + }), + scan_range: scan_range.map(|range| engine::ScanRange { + start: range.start, + end: range.end, + }), + }, + }) +} + +#[cfg(test)] +mod tests { + use super::select_input_from_s3s; + use s3s::dto::{ + CSVInput, CSVOutput, CompressionType, ExpressionType, FileHeaderInfo, InputSerialization, JSONInput, JSONType, + OutputSerialization, RequestProgress, ScanRange, SelectObjectContentInput, SelectObjectContentRequest, + }; + + fn input(input_serialization: InputSerialization) -> SelectObjectContentInput { + SelectObjectContentInput { + bucket: "bucket".to_owned(), + expected_bucket_owner: Some("owner".to_owned()), + key: "key".to_owned(), + sse_customer_algorithm: Some("AES256".to_owned()), + sse_customer_key: Some("key-text".to_owned()), + sse_customer_key_md5: Some("client-md5".to_owned()), + request: SelectObjectContentRequest { + expression: "SELECT * FROM S3Object".to_owned(), + expression_type: ExpressionType::from_static(ExpressionType::SQL), + input_serialization, + output_serialization: OutputSerialization { + csv: Some(CSVOutput { + field_delimiter: Some(";".to_owned()), + quote_fields: None, + quote_character: Some("'".to_owned()), + quote_escape_character: None, + record_delimiter: Some("\r\n".to_owned()), + }), + json: None, + }, + request_progress: Some(RequestProgress { enabled: Some(true) }), + scan_range: Some(ScanRange { + start: Some(3), + end: Some(9), + }), + }, + } + } + + #[test] + fn keeps_every_member_the_engine_reads() { + let converted = select_input_from_s3s(input(InputSerialization { + csv: Some(CSVInput { + allow_quoted_record_delimiter: Some(true), + comments: Some("#".to_owned()), + field_delimiter: Some("|".to_owned()), + file_header_info: Some(FileHeaderInfo::from_static(FileHeaderInfo::USE)), + quote_character: Some("\"".to_owned()), + quote_escape_character: Some("\\".to_owned()), + record_delimiter: Some("\n".to_owned()), + }), + compression_type: Some(CompressionType::from_static(CompressionType::GZIP)), + json: None, + parquet: None, + })) + .expect("convert"); + + assert_eq!((converted.bucket.as_str(), converted.key.as_str()), ("bucket", "key")); + assert_eq!(converted.expected_bucket_owner.as_deref(), Some("owner")); + assert_eq!(converted.sse_customer_algorithm.as_deref(), Some("AES256")); + assert_eq!(converted.sse_customer_key.as_deref(), Some("key-text")); + assert_eq!(converted.sse_customer_key_md5.as_deref(), Some("client-md5")); + let request = &converted.request; + assert_eq!(request.expression, "SELECT * FROM S3Object"); + assert_eq!(request.expression_type.as_str(), "SQL"); + let csv = request.input_serialization.csv.as_ref().expect("csv input"); + assert_eq!(csv.allow_quoted_record_delimiter, Some(true)); + assert_eq!(csv.comments.as_deref(), Some("#")); + assert_eq!(csv.field_delimiter.as_deref(), Some("|")); + assert_eq!(csv.file_header_info.as_ref().map(|value| value.as_str()), Some("USE")); + assert_eq!(csv.quote_character.as_deref(), Some("\"")); + assert_eq!(csv.quote_escape_character.as_deref(), Some("\\")); + assert_eq!(csv.record_delimiter.as_deref(), Some("\n")); + assert_eq!( + request + .input_serialization + .compression_type + .as_ref() + .map(|value| value.as_str()), + Some("GZIP") + ); + let out = request.output_serialization.csv.as_ref().expect("csv output"); + assert_eq!(out.field_delimiter.as_deref(), Some(";")); + assert_eq!(out.quote_character.as_deref(), Some("'")); + assert_eq!(out.record_delimiter.as_deref(), Some("\r\n")); + assert_eq!(request.request_progress.as_ref().and_then(|progress| progress.enabled), Some(true)); + let range = request.scan_range.as_ref().expect("scan range"); + assert_eq!((range.start, range.end), (Some(3), Some(9))); + } + + #[test] + fn keeps_the_json_type_and_an_unknown_expression_type() { + let mut source = input(InputSerialization { + json: Some(JSONInput { + type_: Some(JSONType::from_static(JSONType::LINES)), + }), + ..Default::default() + }); + source.request.expression_type = ExpressionType::from("SQL2".to_owned()); + let converted = select_input_from_s3s(source).expect("convert"); + let json = converted.request.input_serialization.json.as_ref().expect("json input"); + assert_eq!(json.r#type.as_ref().map(|value| value.as_str()), Some("LINES")); + assert_eq!(converted.request.expression_type.as_str(), "SQL2"); + } +}