use datafusion::arrow::error::ArrowError;
use datafusion::common::SchemaError;
use datafusion::error::DataFusionError;
use datafusion::sql::sqlparser::parser::ParserError;
pub mod matched {
pub const OPTIMIZER_RULE_FAILED: &str = "Optimizer rule";
pub const TOKENIZER_UNTERMINATED: &str = "Unterminated";
pub const TOKENIZER_EOF: &str = "EOF";
pub const TIMESTAMP_OVERFLOW: &str = "exceeds the representable i64 range";
pub const TIMESTAMP_NS: &str = "Timestamp(ns)";
pub const CSV_FILE_CONTEXT: &str = "Error when processing CSV file ";
}
pub fn shown_path(location: &str) -> String {
#[cfg(all(unix, not(target_arch = "wasm32")))]
if !location.starts_with('/') && !location.contains("://") {
let absolute = format!("/{location}");
if std::path::Path::new(&absolute).exists() {
return absolute;
}
}
location.to_string()
}
const TIMESTAMP_RANGE_HINT: &str = "a timestamp in nanoseconds holds only dates from 1677-09-21 \
to 2262-04-11, and date_trunc and the other timestamp functions convert to one. Filter \
the dates first (WHERE d BETWEEN DATE '1678-01-01' AND DATE '2262-04-11'), or use \
extract(year FROM d) or date_part instead of date_trunc";
pub fn user_message(err: &DataFusionError) -> String {
let message = message_of(err);
if message.contains(matched::TIMESTAMP_OVERFLOW) && message.contains(matched::TIMESTAMP_NS) {
return format!("{message}; {TIMESTAMP_RANGE_HINT}");
}
message
}
fn message_of(err: &DataFusionError) -> String {
match err {
DataFusionError::ArrowError(e, _) => arrow_message(e),
DataFusionError::ParquetError(e) => format!("Parquet: {e}"),
DataFusionError::ObjectStore(e) => e.to_string(),
DataFusionError::IoError(e) => e.to_string(),
DataFusionError::SQL(e, _) => parser_message(e),
DataFusionError::NotImplemented(m)
| DataFusionError::Plan(m)
| DataFusionError::Configuration(m)
| DataFusionError::Execution(m)
| DataFusionError::ResourcesExhausted(m)
| DataFusionError::Substrait(m)
| DataFusionError::Ffi(m) => m.clone(),
DataFusionError::Internal(m) => format!("internal error: {m}"),
DataFusionError::SchemaError(e, _) => schema_message(e),
DataFusionError::ExecutionJoin(e) => e.to_string(),
DataFusionError::External(e) => external_message(e.as_ref()),
DataFusionError::Context(context, inner)
if context.starts_with(matched::OPTIMIZER_RULE_FAILED) =>
{
message_of(inner)
}
DataFusionError::Context(context, inner) => {
match context.strip_prefix(matched::CSV_FILE_CONTEXT) {
Some(location) => format!(
"cannot read {}: {}",
shown_path(location),
message_of(inner)
),
None => format!("{context}: {}", message_of(inner)),
}
}
DataFusionError::Diagnostic(_, inner) => message_of(inner),
DataFusionError::Collection(errs) => errs
.first()
.map(message_of)
.unwrap_or_else(|| "unknown error".to_string()),
DataFusionError::Shared(inner) => message_of(inner),
}
}
fn schema_message(e: &SchemaError) -> String {
let message = e.to_string();
let SchemaError::FieldNotFound {
field,
valid_fields,
} = e
else {
return message;
};
let part_of =
|name: &str| name.split('-').count() > 1 && name.split('-').any(|p| p == field.name);
match valid_fields.iter().find(|c| part_of(&c.name)) {
Some(column) => {
let (first, rest) = message.split_once('\n').unwrap_or((&message, ""));
let mut out = format!(
"{first}\nA name that contains '-' must be quoted, or it is read as a \
subtraction: write \"{}\"",
column.name.replace('"', "\"\"")
);
out.push_str(&format!(", not {}.", column.name));
if !rest.is_empty() {
out.push('\n');
out.push_str(rest);
}
out
}
None => message,
}
}
fn parser_message(e: &ParserError) -> String {
match e {
ParserError::TokenizerError(m) | ParserError::ParserError(m) => {
format!("syntax error: {m}")
}
ParserError::RecursionLimitExceeded => {
"syntax error: the statement is nested too deeply".to_string()
}
}
}
fn arrow_message(e: &ArrowError) -> String {
match e {
ArrowError::ExternalError(source) => external_message(source.as_ref()),
other => other.to_string(),
}
}
fn external_message(e: &(dyn std::error::Error + Send + Sync + 'static)) -> String {
if let Some(df) = e.downcast_ref::<DataFusionError>() {
message_of(df)
} else if let Some(arrow) = e.downcast_ref::<ArrowError>() {
arrow_message(arrow)
} else {
e.to_string()
}
}
#[cfg(test)]
mod tests {
use std::sync::Arc;
use datafusion::common::tree_node::Transformed;
use datafusion::common::DFSchema;
use datafusion::logical_expr::{EmptyRelation, LogicalPlan};
use datafusion::optimizer::{Optimizer, OptimizerConfig, OptimizerContext, OptimizerRule};
use datafusion::prelude::SessionContext;
use datafusion::sql::sqlparser::dialect::GenericDialect;
use datafusion::sql::sqlparser::tokenizer::Tokenizer;
use super::*;
#[derive(Debug)]
struct Failing;
impl OptimizerRule for Failing {
fn name(&self) -> &str {
"failing"
}
fn supports_rewrite(&self) -> bool {
true
}
fn rewrite(
&self,
_plan: LogicalPlan,
_config: &dyn OptimizerConfig,
) -> datafusion::error::Result<Transformed<LogicalPlan>> {
Err(DataFusionError::Plan("the cause".into()))
}
}
#[test]
fn datafusion_still_labels_a_failed_optimizer_rule() {
let optimizer = Optimizer::with_rules(vec![Arc::new(Failing)]);
let config = OptimizerContext::new().with_skip_failing_rules(false);
let plan = LogicalPlan::EmptyRelation(EmptyRelation {
produce_one_row: false,
schema: Arc::new(DFSchema::empty()),
});
let err = optimizer
.optimize(plan, &config, |_, _| {})
.expect_err("the rule fails");
let DataFusionError::Context(label, _) = &err else {
panic!(
"DataFusion no longer wraps a failed optimizer rule's error in a context: {err:?}"
);
};
assert!(
label.starts_with(matched::OPTIMIZER_RULE_FAILED),
"DataFusion's label for a failed optimizer rule changed; update \
matched::OPTIMIZER_RULE_FAILED: {label}"
);
assert_eq!(user_message(&err), "the cause");
}
#[tokio::test]
async fn datafusion_still_returns_syntax_errors_typed() {
let err = SessionContext::new()
.sql("SELEC 1")
.await
.expect_err("a syntax error");
assert!(
matches!(err.find_root(), DataFusionError::SQL(..)),
"DataFusion no longer returns a syntax error as DataFusionError::SQL: {err:?}"
);
let message = user_message(&err);
assert!(
message.starts_with("syntax error: ") && !message.contains("ParserError"),
"{message}"
);
}
fn tokenizer_error(sql: &str) -> String {
Tokenizer::new(&GenericDialect {}, sql)
.tokenize()
.expect_err("the tokenizer fails")
.message
}
#[test]
fn sqlparser_still_says_unterminated_and_eof() {
for sql in ["SELECT 'abc", "SELECT 'it''s"] {
let message = tokenizer_error(sql);
assert!(
message.starts_with(matched::TOKENIZER_UNTERMINATED),
"sqlparser's message for an open string changed; update \
matched::TOKENIZER_UNTERMINATED: {sql}: {message}"
);
}
for sql in ["SELECT \"abc", "SELECT 1 /* a note"] {
let message = tokenizer_error(sql);
assert!(
message.contains(matched::TOKENIZER_EOF),
"sqlparser's message for an open quoted name or comment changed; update \
matched::TOKENIZER_EOF: {sql}: {message}"
);
}
}
#[tokio::test]
async fn datafusion_still_names_a_bad_csv_file_by_its_location() {
let dir = tempfile::tempdir().unwrap();
let path = std::fs::canonicalize(dir.path()).unwrap().join("bad.csv");
std::fs::write(&path, "a,b\n1,2\n1,2,3\n").unwrap();
let ctx = SessionContext::new();
let err = ctx
.register_csv("t", path.to_str().unwrap(), Default::default())
.await
.unwrap_err();
let DataFusionError::Context(context, _) = &err else {
panic!("DataFusion no longer gives a bad CSV file context; update matched::CSV_FILE_CONTEXT: {err:?}");
};
assert!(
context.starts_with(matched::CSV_FILE_CONTEXT),
"DataFusion's context for a bad CSV file changed; update \
matched::CSV_FILE_CONTEXT: {context}"
);
let message = user_message(&err);
#[cfg(unix)]
assert!(
message.starts_with(&format!("cannot read {}: ", path.display())),
"{message}"
);
#[cfg(not(unix))]
assert!(message.starts_with("cannot read "), "{message}");
}
#[test]
fn strips_variant_prefixes() {
let err = DataFusionError::Plan("table 'x' not found".into());
assert_eq!(user_message(&err), "table 'x' not found");
let err = DataFusionError::SQL(
Box::new(ParserError::ParserError("Expected: an expression".into())),
None,
);
assert_eq!(user_message(&err), "syntax error: Expected: an expression");
let err = DataFusionError::External(Box::new(DataFusionError::Execution("boom".into())));
assert_eq!(user_message(&err), "boom");
}
#[tokio::test]
async fn a_hyphenated_name_says_to_quote_it() {
let ctx = SessionContext::new();
let sql = "SELECT date-created FROM (SELECT 1 AS \"date-created\", 2 AS x) t";
let err = ctx.sql(sql).await.unwrap_err();
assert_eq!(
user_message(&err),
"No field named date.\n\
A name that contains '-' must be quoted, or it is read as a subtraction: write \
\"date-created\", not date-created.\n\
Valid fields are t.\"date-created\", t.x."
);
let sql = "SELECT nope FROM (SELECT 1 AS \"date-created\") t";
let err = ctx.sql(sql).await.unwrap_err();
assert!(
!user_message(&err).contains("quoted"),
"{}",
user_message(&err)
);
}
#[tokio::test]
async fn an_out_of_range_date_in_date_trunc_suggests_extract() {
let ctx = SessionContext::new();
let err = ctx
.sql("SELECT date_trunc('year', CAST(-100000000 AS DATE))")
.await
.unwrap()
.collect()
.await
.unwrap_err();
let raw = message_of(&err);
assert!(
raw.contains(matched::TIMESTAMP_OVERFLOW) && raw.contains(matched::TIMESTAMP_NS),
"arrow's message for an out-of-range timestamp cast changed; update \
matched::TIMESTAMP_OVERFLOW and TIMESTAMP_NS: {raw}"
);
let message = user_message(&err);
assert!(
message.ends_with(&format!("; {TIMESTAMP_RANGE_HINT}")),
"{message}"
);
assert!(message.contains("extract(year FROM d)"), "{message}");
}
}