gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-10-27 14:27:10 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(s3select): run queries on gateway DTO request shapes (#8794)

Browse Source 
* refactor(s3select): run queries on an engine request built from gateway DTO shapes

Part of T1.5 (rustfs/backlog#2746), batch 2.

* fix(s3select): keep the select request conversion in the app layer

The layer guard forbids app -> storage; the conversion moves from storage/s3_compat/select.rs to app/select_request.rs unchanged.
This commit is contained in:
Chris
2026-10-22 08:16:48 +09:00
committed by GitHub
parent 1bd91c8dad
commit be0986115c
25 changed files with 488 additions and 224 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.config/s3s-footprint-baseline.txt
+1 -1
View File
@@ -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
.config/test-count-baseline.txt
+2 -2
View File
@@ -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
Cargo.lock
Generated
+3 -2
View File
@@ -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",
crates/s3select-api/Cargo.toml
+1 -1
View File
@@ -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
crates/s3select-api/src/lib.rs
+1
View File
@@ -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;
crates/s3select-api/src/object_store.rs
+69 -88
View File
@@ -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<Option<CompressionFormat>, 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<SelectObjectContentInput> {
fn json_input(bucket: &str, object: &str, json_type: JsonType) -> Arc<SelectObjectContentInput> {
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")),
crates/s3select-api/src/query/mod.rs
+7 -5
View File
@@ -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,
crates/s3select-api/src/query/session.rs
+17 -18
View File
@@ -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)
crates/s3select-api/src/request.rs
+95
View File
@@ -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<String>,
pub key: String,
pub sse_customer_algorithm: Option<String>,
pub sse_customer_key: Option<String>,
pub sse_customer_key_md5: Option<String>,
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<RequestProgress>,
pub scan_range: Option<ScanRange>,
}
impl fmt::Debug for SelectObjectContentInput {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
let redacted = |value: &Option<String>| value.as_ref().map(|_| "<redacted>");
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}");
}
}
crates/s3select-query/Cargo.toml
-1
View File
@@ -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"] }
crates/s3select-query/src/dispatcher/manager.rs
+38 -38
View File
@@ -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<FileHeaderInfo> = LazyLock::new(|| FileHeaderInfo::from_static(FileHeaderInfo::IGNORE));
static NONE: LazyLock<FileHeaderInfo> = LazyLock::new(|| FileHeaderInfo::from_static(FileHeaderInfo::NONE));
static USE: LazyLock<FileHeaderInfo> = LazyLock::new(|| FileHeaderInfo::from_static(FileHeaderInfo::USE));
static IGNORE: LazyLock<FileHeaderInfo> = LazyLock::new(|| FileHeaderInfo::IGNORE);
static NONE: LazyLock<FileHeaderInfo> = LazyLock::new(|| FileHeaderInfo::NONE);
static USE: LazyLock<FileHeaderInfo> = 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;
crates/s3select-query/src/dispatcher/parquet_table.rs
+5 -5
View File
@@ -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,
crates/s3select-query/src/instance.rs
+18 -18
View File
@@ -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,
crates/s3select-query/src/lib.rs
+5 -5
View File
@@ -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<Arc<dyn DatabaseManagerSystem + Se
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::from("SQL"),
input_serialization: rustfs_s3select_api::request::InputSerialization::default(),
output_serialization: rustfs_s3select_api::request::OutputSerialization::default(),
request_progress: None,
scan_range: None,
},
crates/s3select-query/src/test/error_handling_test.rs
+7 -7
View File
@@ -15,13 +15,13 @@
#[cfg(test)]
mod error_handling_tests {
use crate::get_global_db;
use rustfs_s3select_api::request::{
CsvInput, ExpressionType, FileHeaderInfo, InputSerialization, SelectObjectContentInput, SelectObjectContentRequest,
};
use rustfs_s3select_api::{
QueryError, SelectError,
query::{Context, Query},
};
use s3s::dto::{
CSVInput, ExpressionType, FileHeaderInfo, InputSerialization, SelectObjectContentInput, SelectObjectContentRequest,
};
use std::sync::Arc;
fn create_test_input_with_sql(sql: &str) -> 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,
},
crates/s3select-query/src/test/integration_test.rs
+20 -20
View File
@@ -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
rustfs/Cargo.toml
+1 -1
View File
@@ -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 }
rustfs/src/app/context.rs
+5 -5
View File
@@ -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,
},
rustfs/src/app/context/handles.rs
+1 -1
View File
@@ -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},
rustfs/src/app/context/interfaces.rs
+1 -1
View File
@@ -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;
rustfs/src/app/context/runtime_sources.rs
+1 -1
View File
@@ -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;
rustfs/src/app/mod.rs
+1
View File
@@ -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;
rustfs/src/app/runtime_sources.rs
+1 -1
View File
@@ -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;
rustfs/src/app/select_object.rs
+12 -3
View File
@@ -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
rustfs/src/app/select_request.rs
+176
View File
@@ -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<engine::SelectObjectContentInput, ConversionError> {
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");
}
}