rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-05-31 04:10:38 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore: tune return msg (#2506)

Browse Source 
* chore: test return msg

* fix: test_child_error

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>

* chore: fix test

* chore: minor fix grpc return value

* chore: format return msg

* chore: use root error as return value

* chore: fix empty err display

* chore: iter through external error

* chore: remove err msg

* chore: remove unused field

---------

Signed-off-by: Ruihang Xia <waynestxia@gmail.com>
Co-authored-by: Ruihang Xia <waynestxia@gmail.com>
This commit is contained in:
shuiyisong
2023-09-27 18:40:25 +08:00
committed by GitHub
parent 9282e59a3b
commit ee8d472aae
32 changed files with 138 additions and 191 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
src/catalog/src/error.rs
View File

@@ -216,9 +216,8 @@ pub enum Error {
#[snafu(display("Illegal access to catalog: {} and schema: {}", catalog, schema))]
QueryAccessDenied { catalog: String, schema: String },
#[snafu(display("msg: {}", msg))]
#[snafu(display(""))]
Datafusion {
msg: String,
#[snafu(source)]
error: DataFusionError,
location: Location,

4
src/cmd/src/cli/repl.rs
View File

@@ -35,7 +35,7 @@ use query::QueryEngine;
use rustyline::error::ReadlineError;
use rustyline::Editor;
use session::context::QueryContext;
use snafu::{ErrorCompat, ResultExt};
use snafu::ResultExt;
use substrait::{DFLogicalSubstraitConvertor, SubstraitPlan};
use crate::cli::cmd::ReplCommand;
@@ -148,7 +148,7 @@ impl Repl {
.await
.map_err(|e| {
let status_code = e.status_code();
let root_cause = e.iter_chain().last().unwrap();
let root_cause = e.output_msg();
println!("Error: {}({status_code}), {root_cause}", status_code as u32)
})
.is_ok()

37
src/common/error/src/ext.rs
View File

@@ -18,7 +18,7 @@ use std::sync::Arc;
use crate::status_code::StatusCode;
/// Extension to [`Error`](std::error::Error) in std.
pub trait ErrorExt: std::error::Error + StackError {
pub trait ErrorExt: StackError {
/// Map this error to [StatusCode].
fn status_code(&self) -> StatusCode {
StatusCode::Unknown
@@ -34,12 +34,43 @@ pub trait ErrorExt: std::error::Error + StackError {
/// Returns the error as [Any](std::any::Any) so that it can be
/// downcast to a specific implementation.
fn as_any(&self) -> &dyn Any;
fn output_msg(&self) -> String
where
Self: Sized,
{
let error = self.last();
if let Some(external_error) = error.source() {
let external_root = external_error.sources().last().unwrap();
if error.to_string().is_empty() {
format!("{external_root}")
} else {
format!("{error}: {external_root}")
}
} else {
format!("{error}")
}
}
}
pub trait StackError {
pub trait StackError: std::error::Error {
fn debug_fmt(&self, layer: usize, buf: &mut Vec<String>);
fn next(&self) -> Option<&dyn StackError>;
fn last(&self) -> &dyn StackError
where
Self: Sized,
{
let Some(mut result) = self.next() else {
return self;
};
while let Some(err) = result.next() {
result = err;
}
result
}
}
impl<T: ?Sized + StackError> StackError for Arc<T> {
@@ -52,7 +83,7 @@ impl<T: ?Sized + StackError> StackError for Arc<T> {
}
}
impl<T: ?Sized + StackError> StackError for Box<T> {
impl<T: StackError> StackError for Box<T> {
fn debug_fmt(&self, layer: usize, buf: &mut Vec<String>) {
self.as_ref().debug_fmt(layer, buf)
}

1
src/common/error/src/lib.rs
View File

@@ -11,6 +11,7 @@
// 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.
#![feature(error_iter)]
pub mod ext;
pub mod format;

12
src/common/procedure/src/local/runner.rs
View File

@@ -453,14 +453,13 @@ mod tests {
use std::sync::Arc;
use async_trait::async_trait;
use common_error::ext::PlainError;
use common_error::ext::{ErrorExt, PlainError};
use common_error::mock::MockError;
use common_error::status_code::StatusCode;
use common_test_util::temp_dir::create_temp_dir;
use futures_util::future::BoxFuture;
use futures_util::FutureExt;
use object_store::ObjectStore;
use snafu::ErrorCompat;
use super::*;
use crate::local::test_util;
@@ -943,14 +942,7 @@ mod tests {
// Run the runner and execute the procedure.
runner.run().await;
let err = meta
.state()
.error()
.unwrap()
.iter_chain()
.last()
.unwrap()
.to_string();
let err = meta.state().error().unwrap().output_msg();
assert!(err.contains("subprocedure failed"), "{err}");
}
}

2
src/common/recordbatch/src/error.rs
View File

@@ -55,7 +55,7 @@ pub enum Error {
location: Location,
},
#[snafu(display("Failed to poll stream"))]
#[snafu(display(""))]
PollStream {
#[snafu(source)]
error: datafusion::error::DataFusionError,

32
src/query/src/datafusion.rs
View File

@@ -277,9 +277,7 @@ impl QueryEngine for DatafusionQueryEngine {
Ok(DataFrame::DataFusion(
self.state
.read_table(table)
.context(error::DatafusionSnafu {
msg: "Fail to create dataframe for table",
})
.context(error::DatafusionSnafu)
.map_err(BoxedError::new)
.context(QueryExecutionSnafu)?,
))
@@ -295,9 +293,7 @@ impl LogicalOptimizer for DatafusionQueryEngine {
.state
.session_state()
.optimize(df_plan)
.context(error::DatafusionSnafu {
msg: "Fail to optimize logical plan",
})
.context(error::DatafusionSnafu)
.map_err(BoxedError::new)
.context(QueryExecutionSnafu)?;
@@ -321,9 +317,7 @@ impl PhysicalPlanner for DatafusionQueryEngine {
let physical_plan = state
.create_physical_plan(df_plan)
.await
.context(error::DatafusionSnafu {
msg: "Fail to create physical plan",
})
.context(error::DatafusionSnafu)
.map_err(BoxedError::new)
.context(QueryExecutionSnafu)?;
@@ -394,9 +388,7 @@ impl QueryExecutor for DatafusionQueryEngine {
assert_eq!(1, plan.output_partitioning().partition_count());
let df_stream = plan
.execute(0, task_ctx)
.context(error::DatafusionSnafu {
msg: "Failed to execute DataFusion merge exec",
})
.context(error::DatafusionSnafu)
.map_err(BoxedError::new)
.context(QueryExecutionSnafu)?;
let stream = RecordBatchStreamAdapter::try_new(df_stream)
@@ -447,35 +439,27 @@ pub async fn execute_show_with_filter(
let context = SessionContext::new();
context
.register_batch(table_name, record_batch.into_df_record_batch())
.context(error::DatafusionSnafu {
msg: "Fail to register a record batch as a table",
})
.context(error::DatafusionSnafu)
.map_err(BoxedError::new)
.context(QueryExecutionSnafu)?;
let mut dataframe = context
.sql(&format!("SELECT * FROM {table_name}"))
.await
.context(error::DatafusionSnafu {
msg: "Fail to execute a sql",
})
.context(error::DatafusionSnafu)
.map_err(BoxedError::new)
.context(QueryExecutionSnafu)?;
if let Some(filter) = filter {
let filter = convert_filter_to_df_filter(filter)?;
dataframe = dataframe
.filter(filter)
.context(error::DatafusionSnafu {
msg: "Fail to filter",
})
.context(error::DatafusionSnafu)
.map_err(BoxedError::new)
.context(QueryExecutionSnafu)?
}
let df_batches = dataframe
.collect()
.await
.context(error::DatafusionSnafu {
msg: "Fail to collect the record batches",
})
.context(error::DatafusionSnafu)
.map_err(BoxedError::new)
.context(QueryExecutionSnafu)?;
let mut batches = Vec::with_capacity(df_batches.len());

3
src/query/src/datafusion/error.rs
View File

@@ -25,9 +25,8 @@ use snafu::{Location, Snafu};
#[snafu(visibility(pub))]
#[stack_trace_debug]
pub enum InnerError {
#[snafu(display("msg: {}", msg))]
#[snafu(display(""))]
Datafusion {
msg: &'static str,
#[snafu(source)]
error: DataFusionError,
location: Location,

5
src/query/src/error.rs
View File

@@ -121,7 +121,7 @@ pub enum Error {
location: Location,
},
#[snafu(display("DataFusion error"))]
#[snafu(display(""))]
DataFusion {
#[snafu(source)]
error: DataFusionError,
@@ -140,9 +140,8 @@ pub enum Error {
source: sql::error::Error,
},
#[snafu(display("Cannot plan SQL: {}", sql))]
#[snafu(display(""))]
PlanSql {
sql: String,
#[snafu(source)]
error: DataFusionError,
location: Location,

11
src/query/src/planner.rs
View File

@@ -76,14 +76,9 @@ impl DfLogicalPlanner {
let sql_to_rel = SqlToRel::new_with_options(&context_provider, parser_options);
let result = sql_to_rel.statement_to_plan(df_stmt).with_context(|_| {
let sql = if let Statement::Query(query) = stmt {
query.inner.to_string()
} else {
format!("{stmt:?}")
};
PlanSqlSnafu { sql }
})?;
let result = sql_to_rel
.statement_to_plan(df_stmt)
.context(PlanSqlSnafu)?;
let plan = RangePlanRewriter::new(table_provider, context_provider)
.rewrite(result)
.await?;

2
src/script/src/python/error.rs
View File

@@ -76,7 +76,7 @@ pub enum Error {
error: ArrowError,
},
#[snafu(display("DataFusion error"))]
#[snafu(display(""))]
DataFusion {
location: SnafuLocation,
#[snafu(source)]

11
src/servers/src/error.rs
View File

@@ -27,7 +27,7 @@ use common_telemetry::logging;
use datatypes::prelude::ConcreteDataType;
use query::parser::PromQuery;
use serde_json::json;
use snafu::{ErrorCompat, Location, Snafu};
use snafu::{Location, Snafu};
use tonic::Code;
#[derive(Snafu)]
@@ -511,7 +511,6 @@ macro_rules! define_into_tonic_status {
impl From<$Error> for tonic::Status {
fn from(err: $Error) -> Self {
use common_error::{GREPTIME_ERROR_CODE, GREPTIME_ERROR_MSG};
use snafu::ErrorCompat;
use tonic::codegen::http::{HeaderMap, HeaderValue};
use tonic::metadata::MetadataMap;
@@ -521,16 +520,16 @@ macro_rules! define_into_tonic_status {
// (which is a very rare case), just ignore. Client will use Tonic status code and message.
let status_code = err.status_code();
headers.insert(GREPTIME_ERROR_CODE, HeaderValue::from(status_code as u32));
let root_error = err.iter_chain().last().unwrap();
let root_error = err.output_msg();
if let Ok(err_msg) = HeaderValue::from_bytes(root_error.to_string().as_bytes()) {
if let Ok(err_msg) = HeaderValue::from_bytes(root_error.as_bytes()) {
let _ = headers.insert(GREPTIME_ERROR_MSG, err_msg);
}
let metadata = MetadataMap::from_headers(headers);
tonic::Status::with_metadata(
$crate::error::status_to_tonic_code(status_code),
err.to_string(),
root_error,
metadata,
)
}
@@ -548,7 +547,7 @@ impl From<std::io::Error> for Error {
impl IntoResponse for Error {
fn into_response(self) -> Response {
let error_msg = self.iter_chain().last().unwrap().to_string();
let error_msg = self.output_msg();
let status = match self {
Error::InfluxdbLineProtocol { .. }
| Error::InfluxdbLinesWrite { .. }

2
src/servers/src/grpc/prom_query_gateway.rs
View File

@@ -124,7 +124,7 @@ impl PrometheusGatewayService {
Err(err) => {
return PrometheusJsonResponse::error(
err.status_code().to_string(),
err.to_string(),
err.output_msg(),
)
.0
}

7
src/servers/src/http.rs
View File

@@ -52,7 +52,7 @@ use futures::FutureExt;
use schemars::JsonSchema;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use snafu::{ensure, ErrorCompat, ResultExt};
use snafu::{ensure, ResultExt};
use tokio::sync::oneshot::{self, Sender};
use tokio::sync::Mutex;
use tower::timeout::TimeoutLayer;
@@ -314,10 +314,7 @@ impl JsonResponse {
}
},
Err(e) => {
return Self::with_error(
e.iter_chain().last().unwrap().to_string(),
e.status_code(),
);
return Self::with_error(e.output_msg(), e.status_code());
}
}
}

14
src/servers/src/http/opentsdb.rs
View File

@@ -17,12 +17,13 @@ use std::collections::HashMap;
use axum::extract::{Query, RawBody, State};
use axum::http::StatusCode as HttpStatusCode;
use axum::{Extension, Json};
use common_error::ext::ErrorExt;
use hyper::Body;
use serde::{Deserialize, Serialize};
use session::context::QueryContextRef;
use snafu::{ErrorCompat, ResultExt};
use snafu::ResultExt;
use crate::error::{self, Error, Result};
use crate::error::{self, Result};
use crate::opentsdb::codec::DataPoint;
use crate::query_handler::OpentsdbProtocolHandlerRef;
@@ -154,13 +155,13 @@ impl OpentsdbDebuggingResponse {
self.success += 1;
}
fn on_failed(&mut self, datapoint: DataPointRequest, error: Error) {
fn on_failed(&mut self, datapoint: DataPointRequest, error: impl ErrorExt) {
self.failed += 1;
if let Some(details) = self.errors.as_mut() {
let error = OpentsdbDetailError {
datapoint,
error: error.iter_chain().last().unwrap().to_string(),
error: error.output_msg(),
};
details.push(error);
};
@@ -169,7 +170,6 @@ impl OpentsdbDebuggingResponse {
#[cfg(test)]
mod test {
use snafu::ErrorCompat;
use super::*;
@@ -229,13 +229,13 @@ mod test {
let body = Body::from("");
let result = parse_data_points(body).await;
assert!(result.is_err());
let err = result.unwrap_err().iter_chain().last().unwrap().to_string();
let err = result.unwrap_err().output_msg();
assert!(err.contains("EOF while parsing a value at line 1 column 0"));
let body = Body::from("hello world");
let result = parse_data_points(body).await;
assert!(result.is_err());
let err = result.unwrap_err().iter_chain().last().unwrap().to_string();
let err = result.unwrap_err().output_msg();
assert!(err.contains("expected value at line 1 column 1"));
}
}

16
src/servers/src/http/prometheus.rs
View File

@@ -162,7 +162,7 @@ impl PrometheusJsonResponse {
..Default::default()
}))
} else {
Self::error(err.status_code().to_string(), err.to_string())
Self::error(err.status_code().to_string(), err.output_msg())
}
}
}
@@ -323,7 +323,7 @@ pub async fn instant_query(
let (metric_name, result_type) = match retrieve_metric_name_and_result_type(&prom_query.query) {
Ok((metric_name, result_type)) => (metric_name.unwrap_or_default(), result_type),
Err(err) => {
return PrometheusJsonResponse::error(err.status_code().to_string(), err.to_string())
return PrometheusJsonResponse::error(err.status_code().to_string(), err.output_msg())
}
};
PrometheusJsonResponse::from_query_result(result, metric_name, result_type).await
@@ -357,7 +357,7 @@ pub async fn range_query(
let result = handler.do_query(&prom_query, query_ctx).await;
let metric_name = match retrieve_metric_name_and_result_type(&prom_query.query) {
Err(err) => {
return PrometheusJsonResponse::error(err.status_code().to_string(), err.to_string())
return PrometheusJsonResponse::error(err.status_code().to_string(), err.output_msg())
}
Ok((metric_name, _)) => metric_name.unwrap_or_default(),
};
@@ -430,7 +430,7 @@ pub async fn labels_query(
return PrometheusJsonResponse::success(PrometheusResponse::Labels(labels))
}
Err(e) => {
return PrometheusJsonResponse::error(e.status_code().to_string(), e.to_string())
return PrometheusJsonResponse::error(e.status_code().to_string(), e.output_msg())
}
}
}
@@ -466,7 +466,7 @@ pub async fn labels_query(
{
return PrometheusJsonResponse::error(
err.status_code().to_string(),
err.to_string(),
err.output_msg(),
);
}
}
@@ -690,7 +690,7 @@ pub async fn label_values_query(
let mut table_names = match handler.catalog_manager().table_names(catalog, schema).await {
Ok(table_names) => table_names,
Err(e) => {
return PrometheusJsonResponse::error(e.status_code().to_string(), e.to_string());
return PrometheusJsonResponse::error(e.status_code().to_string(), e.output_msg());
}
};
table_names.sort_unstable();
@@ -723,7 +723,7 @@ pub async fn label_values_query(
{
return PrometheusJsonResponse::error(
err.status_code().to_string(),
err.to_string(),
err.output_msg(),
);
}
}
@@ -837,7 +837,7 @@ pub async fn series_query(
let result = handler.do_query(&prom_query, query_ctx.clone()).await;
if let Err(err) = retrieve_series_from_query_result(result, &mut series, &table_name).await
{
return PrometheusJsonResponse::error(err.status_code().to_string(), err.to_string());
return PrometheusJsonResponse::error(err.status_code().to_string(), err.output_msg());
}
}
PrometheusJsonResponse::success(PrometheusResponse::Series(series))

7
src/servers/src/mysql/writer.rs
View File

@@ -14,7 +14,7 @@
use std::ops::Deref;
use common_error::ext::BoxedError;
use common_error::ext::{BoxedError, ErrorExt};
use common_query::Output;
use common_recordbatch::{RecordBatch, SendableRecordBatchStream};
use datatypes::prelude::{ConcreteDataType, Value};
@@ -26,7 +26,6 @@ use opensrv_mysql::{
};
use session::context::QueryContextRef;
use snafu::prelude::*;
use snafu::ErrorCompat;
use tokio::io::AsyncWrite;
use crate::error::{self, Error, OtherSnafu, Result};
@@ -205,14 +204,14 @@ impl<'a, W: AsyncWrite + Unpin> MysqlResultWriter<'a, W> {
Ok(())
}
async fn write_query_error(error: Error, w: QueryResultWriter<'a, W>) -> Result<()> {
async fn write_query_error(error: impl ErrorExt, w: QueryResultWriter<'a, W>) -> Result<()> {
increment_counter!(
METRIC_ERROR_COUNTER,
&[(METRIC_PROTOCOL_LABEL, METRIC_ERROR_COUNTER_LABEL_MYSQL)]
);
let kind = ErrorKind::ER_INTERNAL_ERROR;
let error = error.iter_chain().last().unwrap().to_string();
let error = error.output_msg();
w.error(kind, error.as_bytes()).await?;
Ok(())
}

4
src/servers/src/opentsdb/connection.rs
View File

@@ -116,7 +116,7 @@ mod tests {
use std::io::Write;
use bytes::BufMut;
use snafu::ErrorCompat;
use common_error::ext::ErrorExt;
use tokio_test::io::Builder;
use super::*;
@@ -187,7 +187,7 @@ mod tests {
buffer.writer().write_all(b"Hello Wor\xffld.\r\n").unwrap();
let result = conn.parse_line();
assert!(result.is_err());
let err = result.unwrap_err().iter_chain().last().unwrap().to_string();
let err = result.unwrap_err().output_msg();
assert!(err.contains("invalid utf-8 sequence"));
}

5
src/servers/src/opentsdb/handler.rs
View File

@@ -14,6 +14,7 @@
//! Modified from Tokio's mini-redis example.
use common_error::ext::ErrorExt;
use common_telemetry::timer;
use session::context::QueryContextBuilder;
use tokio::io::{AsyncRead, AsyncWrite};
@@ -95,11 +96,11 @@ impl<S: AsyncWrite + AsyncRead + Unpin> Handler<S> {
let _timer = timer!(crate::metrics::METRIC_TCP_OPENTSDB_LINE_WRITE_ELAPSED);
let result = self.query_handler.exec(&data_point, ctx.clone()).await;
if let Err(e) = result {
self.connection.write_line(e.to_string()).await?;
self.connection.write_line(e.output_msg()).await?;
}
}
Err(e) => {
self.connection.write_line(e.to_string()).await?;
self.connection.write_line(e.output_msg()).await?;
}
}
}

4
src/servers/src/postgres/handler.rs
View File

@@ -15,6 +15,7 @@
use std::sync::Arc;
use async_trait::async_trait;
use common_error::ext::ErrorExt;
use common_query::Output;
use common_recordbatch::error::Result as RecordBatchResult;
use common_recordbatch::RecordBatch;
@@ -31,7 +32,6 @@ use pgwire::api::{ClientInfo, Type};
use pgwire::error::{ErrorInfo, PgWireError, PgWireResult};
use query::query_engine::DescribeResult;
use session::Session;
use snafu::ErrorCompat;
use sql::dialect::PostgreSqlDialect;
use sql::parser::ParserContext;
@@ -91,7 +91,7 @@ fn output_to_query_response<'a>(
Err(e) => Ok(Response::Error(Box::new(ErrorInfo::new(
"ERROR".to_string(),
"XX000".to_string(),
e.iter_chain().last().unwrap().to_string(),
e.output_msg(),
)))),
}
}

2
src/servers/tests/http/opentsdb_test.rs
View File

@@ -153,7 +153,7 @@ async fn test_opentsdb_put() {
assert_eq!(result.status(), 400);
assert_eq!(
result.text().await,
"{\"error\":\"expected value at line 1 column 1\"}"
"{\"error\":\"Invalid OpenTSDB Json request: expected value at line 1 column 1\"}"
);
// internal server error

3
src/sql/src/error.rs
View File

@@ -59,9 +59,8 @@ pub enum Error {
UnsupportedDefaultValue { column_name: String, expr: Expr },
// Syntax error from sql parser.
#[snafu(display("Syntax error, sql: {}", sql))]
#[snafu(display(""))]
Syntax {
sql: String,
#[snafu(source)]
error: ParserError,
},

10
src/sql/src/parser.rs
View File

@@ -37,7 +37,7 @@ impl<'a> ParserContext<'a> {
let parser = Parser::new(dialect)
.try_with_sql(sql)
.context(SyntaxSnafu { sql })?;
.context(SyntaxSnafu)?;
let mut parser_ctx = ParserContext { sql, parser };
let mut expecting_statement_delimiter = false;
@@ -67,12 +67,12 @@ impl<'a> ParserContext<'a> {
pub fn parse_function(sql: &'a str, dialect: &dyn Dialect) -> Result<Expr> {
let mut parser = Parser::new(dialect)
.try_with_sql(sql)
.context(SyntaxSnafu { sql })?;
.context(SyntaxSnafu)?;
let function_name = parser.parse_identifier().context(SyntaxSnafu { sql })?;
let function_name = parser.parse_identifier().context(SyntaxSnafu)?;
parser
.parse_function(ObjectName(vec![function_name]))
.context(SyntaxSnafu { sql })
.context(SyntaxSnafu)
}
/// Parses parser context to a set of statements.
@@ -143,7 +143,7 @@ impl<'a> ParserContext<'a> {
Err(ParserError::ParserError(format!(
"Expected {expected}, found: {found}",
)))
.context(SyntaxSnafu { sql: self.sql })
.context(SyntaxSnafu)
}
pub fn matches_keyword(&mut self, expected: Keyword) -> bool {

10
src/sql/src/parsers/alter_parser.rs
View File

@@ -25,9 +25,7 @@ use crate::statements::statement::Statement;
impl<'a> ParserContext<'a> {
pub(crate) fn parse_alter(&mut self) -> Result<Statement> {
let alter_table = self
.parse_alter_table()
.context(error::SyntaxSnafu { sql: self.sql })?;
let alter_table = self.parse_alter_table().context(error::SyntaxSnafu)?;
Ok(Statement::Alter(alter_table))
}
@@ -98,7 +96,7 @@ impl<'a> ParserContext<'a> {
mod tests {
use std::assert_matches::assert_matches;
use snafu::ErrorCompat;
use common_error::ext::ErrorExt;
use sqlparser::ast::{ColumnOption, DataType};
use super::*;
@@ -215,7 +213,7 @@ mod tests {
fn test_parse_alter_drop_column() {
let sql = "ALTER TABLE my_metric_1 DROP a";
let result = ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}).unwrap_err();
let err = result.iter_chain().last().unwrap().to_string();
let err = result.output_msg();
assert!(err.contains("expect keyword COLUMN after ALTER TABLE DROP"));
let sql = "ALTER TABLE my_metric_1 DROP COLUMN a";
@@ -245,7 +243,7 @@ mod tests {
fn test_parse_alter_rename_table() {
let sql = "ALTER TABLE test_table table_t";
let result = ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}).unwrap_err();
let err = result.iter_chain().last().unwrap().to_string();
let err = result.output_msg();
assert!(err.contains("expect keyword ADD or DROP or RENAME after ALTER TABLE"));
let sql = "ALTER TABLE test_table RENAME table_t";

12
src/sql/src/parsers/copy_parser.rs
View File

@@ -57,7 +57,7 @@ impl<'a> ParserContext<'a> {
self.parser
.expect_keyword(Keyword::TO)
.context(error::SyntaxSnafu { sql: self.sql })?;
.context(error::SyntaxSnafu)?;
let (with, connection, location) = self.parse_copy_to()?;
Ok(CopyDatabaseArgument {
@@ -89,7 +89,7 @@ impl<'a> ParserContext<'a> {
} else {
self.parser
.expect_keyword(Keyword::FROM)
.context(error::SyntaxSnafu { sql: self.sql })?;
.context(error::SyntaxSnafu)?;
Ok(CopyTable::From(self.parse_copy_table_from(table_name)?))
}
}
@@ -107,7 +107,7 @@ impl<'a> ParserContext<'a> {
let options = self
.parser
.parse_options(Keyword::WITH)
.context(error::SyntaxSnafu { sql: self.sql })?;
.context(error::SyntaxSnafu)?;
let with = options
.into_iter()
@@ -119,7 +119,7 @@ impl<'a> ParserContext<'a> {
let connection_options = self
.parser
.parse_options(Keyword::CONNECTION)
.context(error::SyntaxSnafu { sql: self.sql })?;
.context(error::SyntaxSnafu)?;
let connection = connection_options
.into_iter()
@@ -148,7 +148,7 @@ impl<'a> ParserContext<'a> {
let options = self
.parser
.parse_options(Keyword::WITH)
.context(error::SyntaxSnafu { sql: self.sql })?;
.context(error::SyntaxSnafu)?;
let with = options
.into_iter()
@@ -160,7 +160,7 @@ impl<'a> ParserContext<'a> {
let connection_options = self
.parser
.parse_options(Keyword::CONNECTION)
.context(error::SyntaxSnafu { sql: self.sql })?;
.context(error::SyntaxSnafu)?;
let connection = connection_options
.into_iter()

53
src/sql/src/parsers/create_parser.rs
View File

@@ -69,7 +69,7 @@ impl<'a> ParserContext<'a> {
let _ = self.parser.next_token();
self.parser
.expect_keyword(Keyword::TABLE)
.context(error::SyntaxSnafu { sql: self.sql })?;
.context(error::SyntaxSnafu)?;
let if_not_exists =
self.parser
.parse_keywords(&[Keyword::IF, Keyword::NOT, Keyword::EXISTS]);
@@ -86,7 +86,7 @@ impl<'a> ParserContext<'a> {
let options = self
.parser
.parse_options(Keyword::WITH)
.context(error::SyntaxSnafu { sql: self.sql })?
.context(error::SyntaxSnafu)?
.into_iter()
.filter_map(|option| {
if let Some(v) = parse_option_string(option.value) {
@@ -159,7 +159,7 @@ impl<'a> ParserContext<'a> {
let options = self
.parser
.parse_options(Keyword::WITH)
.context(error::SyntaxSnafu { sql: self.sql })?;
.context(error::SyntaxSnafu)?;
for option in options.iter() {
ensure!(
valid_table_option(&option.name.value),
@@ -200,7 +200,7 @@ impl<'a> ParserContext<'a> {
let column_list = self
.parser
.parse_parenthesized_column_list(Mandatory, false)
.context(error::SyntaxSnafu { sql: self.sql })?;
.context(error::SyntaxSnafu)?;
let entries = self.parse_comma_separated(Self::parse_partition_entry)?;
@@ -219,16 +219,13 @@ impl<'a> ParserContext<'a> {
actual: self.peek_token_as_string(),
})?;
let name = self
.parser
.parse_identifier()
.context(error::SyntaxSnafu { sql: self.sql })?;
let name = self.parser.parse_identifier().context(error::SyntaxSnafu)?;
self.parser
.expect_keyword(Keyword::VALUES)
.and_then(|_| self.parser.expect_token(&LESS))
.and_then(|_| self.parser.expect_token(&THAN))
.context(error::SyntaxSnafu { sql: self.sql })?;
.context(error::SyntaxSnafu)?;
let value_list = self.parse_comma_separated(Self::parse_value_list)?;
@@ -242,10 +239,7 @@ impl<'a> ParserContext<'a> {
let _ = self.parser.next_token();
SqlValue::Number(MAXVALUE.to_string(), false)
}
_ => self
.parser
.parse_value()
.context(error::SyntaxSnafu { sql: self.sql })?,
_ => self.parser.parse_value().context(error::SyntaxSnafu)?,
};
Ok(value)
}
@@ -320,9 +314,7 @@ impl<'a> ParserContext<'a> {
columns: &mut Vec<ColumnDef>,
constraints: &mut Vec<TableConstraint>,
) -> Result<()> {
let mut column = self
.parse_column_def()
.context(SyntaxSnafu { sql: self.sql })?;
let mut column = self.parse_column_def().context(SyntaxSnafu)?;
let mut time_index_opt_idx = None;
for (index, opt) in column.options.iter().enumerate() {
@@ -496,11 +488,7 @@ impl<'a> ParserContext<'a> {
fn parse_optional_table_constraint(&mut self) -> Result<Option<TableConstraint>> {
let name = if self.parser.parse_keyword(Keyword::CONSTRAINT) {
Some(
self.parser
.parse_identifier()
.context(error::SyntaxSnafu { sql: self.sql })?,
)
Some(self.parser.parse_identifier().context(error::SyntaxSnafu)?)
} else {
None
};
@@ -519,7 +507,7 @@ impl<'a> ParserContext<'a> {
let columns = self
.parser
.parse_parenthesized_column_list(Mandatory, false)
.context(error::SyntaxSnafu { sql: self.sql })?;
.context(error::SyntaxSnafu)?;
Ok(Some(TableConstraint::Unique {
name,
columns,
@@ -541,7 +529,7 @@ impl<'a> ParserContext<'a> {
let columns = self
.parser
.parse_parenthesized_column_list(Mandatory, false)
.context(error::SyntaxSnafu { sql: self.sql })?;
.context(error::SyntaxSnafu)?;
ensure!(
columns.len() == 1,
@@ -837,7 +825,7 @@ mod tests {
use std::collections::HashMap;
use common_catalog::consts::FILE_ENGINE;
use snafu::ErrorCompat;
use common_error::ext::ErrorExt;
use sqlparser::ast::ColumnOption::NotNull;
use super::*;
@@ -1401,10 +1389,7 @@ ENGINE=mito";
let result = ParserContext::create_with_dialect(sql, &GreptimeDbDialect {});
assert!(result
.unwrap_err()
.iter_chain()
.last()
.unwrap()
.to_string()
.output_msg()
.contains("sql parser error: Expected BY, found: RANGE"));
let sql = r"
@@ -1418,10 +1403,7 @@ ENGINE=mito";
let result = ParserContext::create_with_dialect(sql, &GreptimeDbDialect {});
assert!(result
.unwrap_err()
.iter_chain()
.last()
.unwrap()
.to_string()
.output_msg()
.contains("sql parser error: Expected LESS, found: THAN"));
let sql = r"
@@ -1435,10 +1417,7 @@ ENGINE=mito";
let result = ParserContext::create_with_dialect(sql, &GreptimeDbDialect {});
assert!(result
.unwrap_err()
.iter_chain()
.last()
.unwrap()
.to_string()
.output_msg()
.contains("Expected a concrete value, found: MAXVALU"));
}
@@ -1547,7 +1526,7 @@ ENGINE=mito";
fn test_invalid_column_name() {
let sql = "create table foo(user string, i timestamp time index)";
let result = ParserContext::create_with_dialect(sql, &GreptimeDbDialect {});
let err = result.unwrap_err().iter_chain().last().unwrap().to_string();
let err = result.unwrap_err().output_msg();
assert!(err.contains("Cannot use keyword 'user' as column name"));
// If column name is quoted, it's valid even same with keyword.

5
src/sql/src/parsers/delete_parser.rs
View File

@@ -24,10 +24,7 @@ use crate::statements::statement::Statement;
impl<'a> ParserContext<'a> {
pub(crate) fn parse_delete(&mut self) -> Result<Statement> {
let _ = self.parser.next_token();
let spstatement = self
.parser
.parse_delete()
.context(error::SyntaxSnafu { sql: self.sql })?;
let spstatement = self.parser.parse_delete().context(error::SyntaxSnafu)?;
match spstatement {
SpStatement::Delete { .. } => {

5
src/sql/src/parsers/insert_parser.rs
View File

@@ -24,10 +24,7 @@ use crate::statements::statement::Statement;
impl<'a> ParserContext<'a> {
pub(crate) fn parse_insert(&mut self) -> Result<Statement> {
let _ = self.parser.next_token();
let spstatement = self
.parser
.parse_insert()
.context(error::SyntaxSnafu { sql: self.sql })?;
let spstatement = self.parser.parse_insert().context(error::SyntaxSnafu)?;
match spstatement {
SpStatement::Insert { .. } => {

12
src/sql/src/parsers/query_parser.rs
View File

@@ -22,10 +22,7 @@ use crate::statements::statement::Statement;
impl<'a> ParserContext<'a> {
/// Parses select and it's variants.
pub(crate) fn parse_query(&mut self) -> Result<Statement> {
let spquery = self
.parser
.parse_query()
.context(error::SyntaxSnafu { sql: self.sql })?;
let spquery = self.parser.parse_query().context(error::SyntaxSnafu)?;
Ok(Statement::Query(Box::new(Query::try_from(spquery)?)))
}
@@ -33,7 +30,7 @@ impl<'a> ParserContext<'a> {
#[cfg(test)]
mod tests {
use snafu::ErrorCompat;
use common_error::ext::ErrorExt;
use crate::dialect::GreptimeDbDialect;
use crate::parser::ParserContext;
@@ -55,10 +52,7 @@ mod tests {
assert!(result.is_err());
assert!(result
.unwrap_err()
.iter_chain()
.last()
.unwrap()
.to_string()
.output_msg()
.contains("Expected an expression"));
}
}

26
src/sql/src/parsers/tql_parser.rs
View File

@@ -45,8 +45,7 @@ impl<'a> ParserContext<'a> {
&& w.quote_style.is_none() =>
{
let _ = self.parser.next_token();
self.parse_tql_eval()
.context(error::SyntaxSnafu { sql: self.sql })
self.parse_tql_eval().context(error::SyntaxSnafu)
}
Keyword::EXPLAIN => {
@@ -56,8 +55,7 @@ impl<'a> ParserContext<'a> {
Keyword::ANALYZE => {
let _ = self.parser.next_token();
self.parse_tql_analyze()
.context(error::SyntaxSnafu { sql: self.sql })
self.parse_tql_analyze().context(error::SyntaxSnafu)
}
_ => self.unsupported(self.peek_token_as_string()),
}
@@ -136,8 +134,8 @@ impl<'a> ParserContext<'a> {
let start = Self::parse_string_or_number(parser, Token::Comma).unwrap_or("0".to_string());
let end = Self::parse_string_or_number(parser, Token::Comma).unwrap_or("0".to_string());
let step = Self::parse_string_or_number(parser, Token::RParen).unwrap_or("5m".to_string());
let query = Self::parse_tql_query(parser, self.sql, delimiter)
.context(error::SyntaxSnafu { sql: self.sql })?;
let query =
Self::parse_tql_query(parser, self.sql, delimiter).context(error::SyntaxSnafu)?;
Ok(Statement::Tql(Tql::Explain(TqlExplain {
query,
@@ -166,7 +164,7 @@ impl<'a> ParserContext<'a> {
#[cfg(test)]
mod tests {
use snafu::ErrorCompat;
use common_error::ext::ErrorExt;
use super::*;
use crate::dialect::GreptimeDbDialect;
@@ -286,21 +284,11 @@ mod tests {
// Invalid duration
let sql = "TQL EVAL (1676887657, 1676887659, 1m) http_requests_total{environment=~'staging|testing|development',method!='GET'} @ 1609746000 offset 5m";
let result = ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}).unwrap_err();
assert!(result
.iter_chain()
.last()
.unwrap()
.to_string()
.contains("Expected ), found: m"));
assert!(result.output_msg().contains("Expected ), found: m"));
// missing end
let sql = "TQL EVAL (1676887657, '1m') http_requests_total{environment=~'staging|testing|development',method!='GET'} @ 1609746000 offset 5m";
let result = ParserContext::create_with_dialect(sql, &GreptimeDbDialect {}).unwrap_err();
assert!(result
.iter_chain()
.last()
.unwrap()
.to_string()
.contains("Expected ,, found: )"));
assert!(result.output_msg().contains("Expected ,, found: )"));
}
}

2
src/table/src/error.rs
View File

@@ -30,7 +30,7 @@ pub type Result<T> = std::result::Result<T, Error>;
#[snafu(visibility(pub))]
#[stack_trace_debug]
pub enum Error {
#[snafu(display("Datafusion error"))]
#[snafu(display(""))]
Datafusion {
#[snafu(source)]
error: DataFusionError,

7
tests/runner/src/env.rs
View File

@@ -28,7 +28,6 @@ use client::{
Client, Database as DB, Error as ClientError, DEFAULT_CATALOG_NAME, DEFAULT_SCHEMA_NAME,
};
use common_error::ext::ErrorExt;
use common_error::snafu::ErrorCompat;
use common_query::Output;
use common_recordbatch::RecordBatches;
use serde::Serialize;
@@ -365,10 +364,10 @@ impl Database for GreptimeDB {
Ok(recordbatches) => result = Ok(Output::RecordBatches(recordbatches)),
Err(e) => {
let status_code = e.status_code();
let source_error = e.iter_chain().last().unwrap();
let msg = e.output_msg();
result = ServerSnafu {
code: status_code,
msg: source_error.to_string(),
msg,
}
.fail();
}
@@ -453,7 +452,7 @@ impl Display for ResultDisplayer {
},
Err(e) => {
let status_code = e.status_code();
let root_cause = e.iter_chain().last().unwrap();
let root_cause = e.output_msg();
write!(
f,
"Error: {}({status_code}), {root_cause}",