// Copyright 2024 RustFS Team // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. use datafusion::{common::DataFusionError, sql::sqlparser::parser::ParserError}; use std::fmt::Display; use thiserror::Error; pub mod object_store; pub mod query; pub mod server; mod storage_api; #[cfg(test)] mod test; pub type QueryResult = Result; pub(crate) use storage_api::crate_boundary::{ SELECT_DEFAULT_READ_BUFFER_SIZE, SelectGetObjectReader, SelectObjectInfo, SelectObjectOptions, SelectStorageError, SelectStore, resolve_select_object_store_handle, select_is_err_bucket_not_found, select_is_err_object_not_found, select_is_err_version_not_found, }; #[derive(Debug, Error)] pub enum QueryError { #[error("DataFusion error: {source}")] Datafusion { source: Box }, #[error("This feature is not implemented: {err}")] NotImplemented { err: String }, #[error("Multi-statement not allow, found num:{num}, sql:{sql}")] MultiStatement { num: usize, sql: String }, #[error("Failed to build QueryDispatcher. err: {err}")] BuildQueryDispatcher { err: String }, #[error("The query has been canceled")] Cancel, #[error("{source}")] Parser { #[from] source: ParserError, }, #[error("Udf not exists, name:{name}.")] FunctionNotExists { name: String }, #[error("Udf already exists, name:{name}.")] FunctionExists { name: String }, #[error("Store Error, e:{e}.")] StoreError { e: String }, } #[derive(Debug, Error)] #[non_exhaustive] pub enum S3SelectPolicyError { #[error("Unsupported S3 Select SQL structure: {message}")] UnsupportedSqlStructure { message: String }, #[error("S3 Select query concurrency limit reached")] QueryConcurrencyLimit, #[error("S3 Select query exceeded the {seconds}-second execution limit")] QueryTimeout { seconds: u64 }, } impl S3SelectPolicyError { fn from_error<'a>(mut err: &'a (dyn std::error::Error + 'static)) -> Option<&'a Self> { for _ in 0..16 { if let Some(policy_error) = err.downcast_ref::() { return Some(policy_error); } err = err.source()?; } None } } impl QueryError { pub fn s3_select_policy_error(&self) -> Option<&S3SelectPolicyError> { match self { Self::Datafusion { source } => S3SelectPolicyError::from_error(source.as_ref()), _ => None, } } } impl From for QueryError { fn from(value: S3SelectPolicyError) -> Self { Self::Datafusion { source: Box::new(DataFusionError::External(Box::new(value))), } } } impl From for QueryError { fn from(value: DataFusionError) -> Self { match value { DataFusionError::External(e) => match e.downcast::() { Ok(query_error) => *query_error, Err(e) => Self::Datafusion { source: Box::new(DataFusionError::External(e)), }, }, v => Self::Datafusion { source: Box::new(v) }, } } } #[derive(Debug, Clone, PartialEq, Eq)] pub struct ResolvedTable { // path table: String, } impl ResolvedTable { pub fn table(&self) -> &str { &self.table } } impl Display for ResolvedTable { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { let Self { table } = self; write!(f, "{table}") } } #[cfg(test)] mod tests { use super::*; use datafusion::common::DataFusionError; use datafusion::sql::sqlparser::parser::ParserError; #[test] fn test_query_error_display() { let err = QueryError::NotImplemented { err: "feature X".to_string(), }; assert_eq!(err.to_string(), "This feature is not implemented: feature X"); let err = QueryError::MultiStatement { num: 2, sql: "SELECT 1; SELECT 2;".to_string(), }; assert_eq!(err.to_string(), "Multi-statement not allow, found num:2, sql:SELECT 1; SELECT 2;"); let err = S3SelectPolicyError::UnsupportedSqlStructure { message: "JOIN is not supported".to_string(), }; assert_eq!(err.to_string(), "Unsupported S3 Select SQL structure: JOIN is not supported"); let err = QueryError::Cancel; assert_eq!(err.to_string(), "The query has been canceled"); assert_eq!( S3SelectPolicyError::QueryConcurrencyLimit.to_string(), "S3 Select query concurrency limit reached" ); assert_eq!( S3SelectPolicyError::QueryTimeout { seconds: 300 }.to_string(), "S3 Select query exceeded the 300-second execution limit" ); let err = QueryError::FunctionNotExists { name: "my_func".to_string(), }; assert_eq!(err.to_string(), "Udf not exists, name:my_func."); let err = QueryError::StoreError { e: "connection failed".to_string(), }; assert_eq!(err.to_string(), "Store Error, e:connection failed."); } #[test] fn test_query_error_from_datafusion_error() { let df_error = DataFusionError::Plan("invalid plan".to_string()); let query_error: QueryError = df_error.into(); match query_error { QueryError::Datafusion { source, .. } => { assert!(source.to_string().contains("invalid plan")); } _ => panic!("Expected Datafusion error"), } } #[test] fn query_error_variants_remain_source_compatible() { fn exhaustive_match(err: QueryError) { match err { QueryError::Datafusion { .. } | QueryError::NotImplemented { .. } | QueryError::MultiStatement { .. } | QueryError::BuildQueryDispatcher { .. } | QueryError::Cancel | QueryError::Parser { .. } | QueryError::FunctionNotExists { .. } | QueryError::FunctionExists { .. } | QueryError::StoreError { .. } => {} } } exhaustive_match(QueryError::Cancel); } #[test] fn policy_error_is_recoverable_from_query_error() { let err: QueryError = S3SelectPolicyError::QueryTimeout { seconds: 300 }.into(); assert!(matches!( err.s3_select_policy_error(), Some(S3SelectPolicyError::QueryTimeout { seconds: 300 }) )); } #[test] fn test_query_error_from_parser_error() { let parser_error = ParserError::ParserError("syntax error".to_string()); let query_error = QueryError::Parser { source: parser_error }; assert!(query_error.to_string().contains("syntax error")); } #[test] fn test_resolved_table() { let table = ResolvedTable { table: "my_table".to_string(), }; assert_eq!(table.table(), "my_table"); assert_eq!(table.to_string(), "my_table"); } #[test] fn test_resolved_table_clone_and_eq() { let table1 = ResolvedTable { table: "table1".to_string(), }; let table2 = table1.clone(); let table3 = ResolvedTable { table: "table2".to_string(), }; assert_eq!(table1, table2); assert_ne!(table1, table3); } }