name: String,

models: Vec<String>,

tests: Option<Vec<String>>,

Check,

Run,

Test {

#[structopt(long)]

fail_fast: bool,

},

Docs,

match table_factor {

TableFactor::Table { name, .. } => vec.push(format!("{}", name)),

TableFactor::NestedJoin(nested_join) => {

get_refs_table_factor(&nested_join.relation, vec);

for join in nested_join.joins.iter() {

get_refs_table_factor(&join.relation, vec);

}

}

TableFactor::Derived {

subquery: query, ..

} => {

get_refs(query, vec);

}

}

match body {

SetExpr::Query(q) => get_refs(q, vec),

SetExpr::Select(select) => {

select

.from

.iter()

.for_each(|table| get_refs_table_factor(&table.relation, vec));

select.projection.iter().for_each(|x| match x {

SelectItem::ExprWithAlias { expr, .. } => get_refs_expr(expr, vec),

SelectItem::UnnamedExpr(expr) => get_refs_expr(expr, vec),

_ => {}

});

}

_ => {}

}

match expr {

Expr::Between { low, high, .. } => {

get_refs_expr(low, vec);

get_refs_expr(high, vec);

}

Expr::BinaryOp { left, right, .. } => {

get_refs_expr(left, vec);

get_refs_expr(right, vec);

}

Expr::Cast { expr, .. } => {

get_refs_expr(expr, vec);

}

Expr::Collate { expr, .. } => {

get_refs_expr(expr, vec);

}

Expr::Exists(query) => get_refs(query, vec),

Expr::Extract { expr, .. } => get_refs_expr(expr, vec),

Expr::Function(Function { args, .. }) => {

for arg in args {

get_refs_expr(arg, vec);

}

}

Expr::InSubquery { expr, subquery, .. } => {

get_refs_expr(expr, vec);

get_refs(subquery, vec);

}

Expr::IsNotNull(expr) => {

get_refs_expr(expr, vec);

}

Expr::IsNull(expr) => {

get_refs_expr(expr, vec);

}

Expr::ListAgg(ListAgg { expr, .. }) => {

get_refs_expr(expr, vec);

}

Expr::Nested(expr) => {

get_refs_expr(expr, vec);

}

Expr::Subquery(query) => get_refs(query, vec),

Expr::UnaryOp { expr, .. } => get_refs_expr(expr, vec),

_ => {}

}

let mut res = HashMap::new();

for path in models.iter() {

let sql = fs::read_to_string(path).map_err(|_x| format!("Error while reading {}", path))?;

let statements = Parser::parse_sql(&PowerSqlDialect {}, &sql)

.map_err(|err| format!("Parse Error in {}: {}", path, err))?;

for statement in statements {

let name = match &statement {

Statement::CreateView { name, .. } => format!("{}", name),

Statement::CreateTable {

name,

query: Some(_),

..

} => format!("{}", name),

_ => unimplemented!("Only (materialized) view and create table as supported "),

};

res.insert(name, statement);

}

}

Ok(res)

let mut res = vec![];

for path in models.iter() {

let sql = fs::read_to_string(path).map_err(|_x| format!("Error while reading {}", path))?;

let statements = Parser::parse_sql(&PowerSqlDialect {}, &sql)

.map_err(|err| format!("Parse Error in {}: {}", path, err))?;

for statement in statements {

let query = match statement {

Statement::Assert {

message: Some(Expr::Value(Value::SingleQuotedString(msg))),

condition,

} => Test {

condition,

message: msg,

},

_ => unimplemented!("Only assert statements are supported in test files"),

};

res.push(query)

}

}

Ok(res)

match statement {

Statement::CreateView { query, .. } => query,

Statement::CreateTable {

query: Some(query), ..

} => query,

Statement::Query(query) => query,

_ => unreachable!("Expected view, table, of query in fn get_query"),

}

match statement {

Statement::CreateView { query, .. } => get_refs(query, vec),

Statement::CreateTable {

query: Some(query), ..

} => get_refs(query, vec),

_ => unreachable!("Expected view or table in fn get_refs_statement"),

}

asts.iter()

.map(|(src, stmt)| {

let mut x = vec![];

get_refs_statement(&stmt, &mut x);

(

src.clone(),

x.iter()

.filter(|elem| asts.contains_key(*elem))

.cloned()

.collect(),

)

})

.collect()

let mut visited_all = HashSet::new();

for (model, model_deps) in deps.iter() {

let mut visited = HashSet::new();

let mut stack = model_deps.clone();

visited.insert(model);

while let Some(x) = stack.pop() {

if visited_all.contains(&x) {

continue;

}

visited_all.insert(x.clone());

let d = deps.get(&x).ok_or(format!("Model {} not found", &x))?;

for i in d.iter() {

if visited.contains(i) {

return Err(format!("Loop detected while checking model {}", model));

}

visited.insert(i);

}

stack.extend(d.clone());

}

}

Ok(())

let mut graph = HashMap::new();

let mut nodes = Vec::new();

for (model, model_deps) in deps.iter() {

// for each model collect number of parents

graph.entry(model.as_str()).or_insert(ModelNode {

live_parents: 0,

next_nodes: vec![],

});

for m in model_deps {

nodes.push((model.as_str(), m.as_str()));

}

}

for (to, from) in nodes {

let x = graph.get_mut(from).unwrap();

x.next_nodes.push(to.to_string());

let mut y = graph.get_mut(to).unwrap();

y.live_parents += 1;

}

Ok(graph)

let mut tests_models = vec![];

if let Some(tests) = tests {

for dir in tests {

for entry in WalkDir::new(dir.to_string()) {

let entry = entry.unwrap();

if let Some(ext) = entry.path().extension() {

{

if ext == "sql" {

tests_models.push(entry.path().to_str().unwrap().to_string());

}

}

}

}

}

}

tests_models

let opt = Opt::from_args();

// Load project

let contents = fs::read_to_string("powersql.toml").expect("No powersql.toml file found");

let config = toml::from_str(&contents);

let config: PowerSqlConfig = match config {

Err(x) => return Err(x.to_string()),

Ok(conf) => conf,

};

let mut models = vec![];

for dir in config.project.models {

for entry in WalkDir::new(dir.to_string()) {

let entry = entry.unwrap();

if let Some(ext) = entry.path().extension() {

{

if ext == "sql" {

models.push(entry.path().to_str().unwrap().to_string());

}

}

}

}

}

let mut asts = load_asts(&models)?;

let dependencies: HashMap<String, Vec<String>> = get_dependencies(&asts);

detect_cycles(&dependencies)?;

match opt.command {

Command::Check => {

let mut graph = build_graph(&dependencies)?;

let mut nodes: Vec<_> = graph

.iter()

.filter(|(_m, node)| node.live_parents == 0)

.map(|(x, _)| (*x).to_string())

.collect();

let mut ty_env = im::HashMap::new();

while let Some(m) = nodes.pop() {

println!("Checking {}", m);

let node = graph.get(m.as_str()).unwrap().clone();

let ty = types::get_model_type(get_query(asts.get(&m).unwrap()), ty_env.clone())?;

println!("{} {:?}", m, ty);

ty_env = ty_env.update(m.to_string(), ty);

println!("ty_env {:?}", ty_env);

for n in node.next_nodes.iter() {

let mut node = graph.get_mut(n.as_str()).unwrap();

node.live_parents -= 1;

if node.live_parents == 0 {

nodes.push(n.to_string());

}

}

}

let test_models = find_test_files(config.project.tests);

let tests = load_tests(&test_models)?;

for Test { condition, .. } in tests {

let ty = types::expr_type(&condition, &HashMap::new(), ty_env.clone(), true)?;

match ty {

types::BaseType::Any | types::BaseType::Boolean => {}

_ => return Err(format!("Expected boolean in test, got {:?}", ty)),

}

}

}

Command::Run => {

let mut graph = build_graph(&dependencies)?;

let mut nodes: Vec<_> = graph

.iter()

.filter(|(_m, node)| node.live_parents == 0)

.map(|(x, _)| (*x).to_string())

.collect();

let mut executor = get_executor()

.await

.map_err(|x| format!("Connection error: {}", x))?;

while let Some(m) = nodes.pop() {

println!("Executing {}", m);

executor.execute(&m, asts.get_mut(&m).unwrap()).await?;

println!("Ready {}", m);

println!("Graph {:?}", graph);

let node = graph.get(m.as_str()).unwrap().clone();

for n in node.next_nodes.iter() {

let mut node = graph.get_mut(n.as_str()).unwrap();

node.live_parents -= 1;

if node.live_parents == 0 {

nodes.push(n.to_string());

}

}

}

}

Command::Docs => {

let arrows: Vec<String> = dependencies

.iter()

.flat_map(|(x, y)| y.iter().map(move |z| format!("{z} -> {x}", x = x, z = z)))

.collect();

let graph = build_graph(&dependencies)?;

std::fs::create_dir("docs");

let mut f =

File::create("docs/docs.md").map_err(|_x| "Could not create docs/docs.md")?;

for (x, deps) in graph {

f.write_fmt(format_args!("#{}\n", x));

}

}

Command::Test { fail_fast } => {

let mut exit_code = 0;

let test_models = find_test_files(config.project.tests);

let tests = load_tests(&test_models)?;

let mut executor = get_executor().await?;

for Test { condition, message } in tests.iter() {

print!("{}", message);

let query = format!("SELECT ({}) AS condition", condition);

let succeeded = executor.query_bool(&query).await?;

if succeeded {

println!("...OK")

} else {

if fail_fast {

std::process::exit(1);

}

exit_code = 1;

println!("...ERROR")

}

}

std::process::exit(exit_code);

}

}

Ok(())

let sql = "create materialized view x as select a from t";

let ast = Parser::parse_sql(&PowerSqlDialect {}, sql).unwrap()[0].clone();

let x = get_dependencies(&hashmap! {"x".to_string() => ast});

assert_eq!(x, hashmap! {"x".to_string() => vec![]})

let sql =

"create materialized view x as select a from t join x on 1=1; create view t as select 1";

let ast = Parser::parse_sql(&PowerSqlDialect {}, sql).unwrap();

let x = get_dependencies(

&hashmap! {"x".to_string() => ast[0].clone(), "t".to_string() => ast[1].clone()},

);

assert_eq!(

x,

hashmap! {"x".to_string() => vec!["t".to_string()], "t".to_string() => vec![]}

);

assert!(matches!(

detect_cycles(&hashmap! {

"a".to_string() => vec!["b".to_string()],

"b".to_string() => vec!["a".to_string()]

}),

Err(_)

));

assert!(matches!(

detect_cycles(&hashmap! {

"a".to_string() => vec!["b".to_string()],

"b".to_string() => vec!["c".to_string()]

}),

Err(_)

));

assert!(matches!(

detect_cycles(&hashmap! {

"a".to_string() => vec!["b".to_string()],

"b".to_string() => vec![]

}),

Ok(_)

));

let deps = hashmap! {

"a".to_string() => vec!["b".to_string()],

"b".to_string() => vec!["c".to_string()],

"c".to_string() => vec![],

};

let plan = build_graph(&deps).unwrap();

assert_eq!(

plan,

hashmap! {

"a" => ModelNode {

live_parents: 1,

next_nodes: vec![]

},

"b" => ModelNode {

live_parents: 1,

next_nodes: vec!["a".to_string()]

},

"c" => ModelNode {

live_parents: 0,

next_nodes: vec!["b".to_string()]

}

}

);