1use datafusion::dataframe::DataFrame;
22use datafusion::error::{DataFusionError, Result};
23use datafusion::prelude::SessionContext;
24use datafusion::sql::parser::Statement as DFStatement;
25use datafusion::sql::sqlparser::ast::Statement as SqlStatement;
26
27pub mod database;
28mod graph_ddl;
29mod parser;
30mod paths;
31pub mod property_graph;
32mod settings;
33
34pub use database::open_session;
35pub use parser::{
36 parse_sql, Attach, CreatePropertyGraph, DescribeGraph, DropPropertyGraph, UrsaParser,
37 UrsaStatement, DEFAULT_RECURSION_LIMIT,
38};
39pub use ursa_engine::database::OpenOptions;
40
41pub fn parse_sql_for(ctx: &SessionContext, sql: &str) -> Result<Vec<UrsaStatement>> {
43 let options = ctx.state().config().options().sql_parser.clone();
44 let dialect = datafusion::sql::sqlparser::dialect::dialect_from_str(options.dialect)
45 .ok_or_else(|| {
46 DataFusionError::Configuration(format!("unknown SQL dialect '{}'", options.dialect))
47 })?;
48 UrsaParser::new(sql, dialect.as_ref(), options.recursion_limit.get())?.parse_statements()
49}
50
51pub fn parse_sql_with_text_for(
54 ctx: &SessionContext,
55 sql: &str,
56) -> Result<Vec<(UrsaStatement, String)>> {
57 let options = ctx.state().config().options().sql_parser.clone();
58 let dialect = datafusion::sql::sqlparser::dialect::dialect_from_str(options.dialect)
59 .ok_or_else(|| {
60 DataFusionError::Configuration(format!("unknown SQL dialect '{}'", options.dialect))
61 })?;
62 UrsaParser::new(sql, dialect.as_ref(), options.recursion_limit.get())?
63 .parse_statements_with_text()
64}
65
66pub async fn execute(ctx: &SessionContext, statement: UrsaStatement) -> Result<DataFrame> {
74 execute_inner(ctx, statement, None).await
75}
76
77pub async fn execute_with_text(
81 ctx: &SessionContext,
82 statement: UrsaStatement,
83 text: &str,
84) -> Result<DataFrame> {
85 execute_inner(ctx, statement, Some(text)).await
86}
87
88async fn execute_inner(
89 ctx: &SessionContext,
90 statement: UrsaStatement,
91 text: Option<&str>,
92) -> Result<DataFrame> {
93 match ursa_engine::database::database(&ctx.state()) {
94 Some(db) => database::execute(ctx, &db, statement, text).await,
95 None => match statement {
96 UrsaStatement::Attach(attach) => database::attach(ctx, attach).await,
97 statement => run(ctx, statement).await,
98 },
99 }
100}
101
102pub(crate) async fn run(ctx: &SessionContext, statement: UrsaStatement) -> Result<DataFrame> {
104 match statement {
105 UrsaStatement::DataFusion(statement) => {
106 let state = ctx.state();
107 match statement.as_ref() {
110 DFStatement::CopyTo(copy) => {
111 ursa_engine::ensure_object_store_for(state.runtime_env(), ©.target).await?
112 }
113 DFStatement::CreateExternalTable(table) => {
114 for location in &table.locations {
115 ursa_engine::ensure_object_store_for(state.runtime_env(), location).await?
116 }
117 }
118 _ => {}
119 }
120 if let DFStatement::Statement(inner) = statement.as_ref() {
122 if let SqlStatement::ShowVariable { variable } = inner.as_ref() {
123 if let Some(frame) = settings::show(ctx, variable).await {
124 return frame;
125 }
126 }
127 }
128 let plan = state.statement_to_plan(*statement).await?;
129 ursa_engine::graph::check_ddl(&state, &plan).await?;
130 ctx.execute_logical_plan(plan).await
131 }
132 UrsaStatement::CreatePropertyGraph(statement) => graph_ddl::create(ctx, *statement).await,
133 UrsaStatement::DropPropertyGraph(statement) => graph_ddl::drop(ctx, statement),
134 UrsaStatement::ShowGraphs => graph_ddl::show(ctx),
135 UrsaStatement::DescribeGraph(statement) => graph_ddl::describe(ctx, statement).await,
136 UrsaStatement::Attach(_) => Err(DataFusionError::Plan(
138 "ATTACH cannot run while replaying a database's catalog".into(),
139 )),
140 }
141}
142
143#[cfg(test)]
144mod tests {
145 use super::*;
146
147 #[tokio::test]
148 async fn runs_datafusion_statements() {
149 let ctx = ursa_engine::new_session();
150 let mut statements = parse_sql_for(
151 &ctx,
152 "CREATE VIEW v AS SELECT 1 AS x UNION ALL SELECT 2; SELECT sum(x) AS s FROM v",
153 )
154 .unwrap();
155 let select = statements.pop().unwrap();
156 execute(&ctx, statements.pop().unwrap()).await.unwrap();
157 let batches = execute(&ctx, select)
158 .await
159 .unwrap()
160 .collect()
161 .await
162 .unwrap();
163 let formatted = datafusion::arrow::util::pretty::pretty_format_batches(&batches)
164 .unwrap()
165 .to_string();
166 assert!(formatted.contains("| 3 |"), "{formatted}");
167 }
168}