mirror of https://github.com/buster-so/buster.git
dal
GitHub
commit
0edb5499a1
|
|
@ -141,7 +141,8 @@ impl QueryAnalyzer {
|
|||
alias_name
|
||||
})
|
||||
.or_else(|| Some(format!("_derived_{}", rand::random::<u32>()))), // Assign a temporary ID if no alias
|
||||
TableFactor::TableFunction { alias, .. } => alias
|
||||
TableFactor::TableFunction { alias, .. } => {
|
||||
alias
|
||||
.as_ref()
|
||||
.map(|a| {
|
||||
let alias_name = a.name.value.clone();
|
||||
|
|
@ -149,7 +150,8 @@ impl QueryAnalyzer {
|
|||
.insert(alias_name.clone(), alias_name.clone());
|
||||
alias_name
|
||||
})
|
||||
.or_else(|| Some(format!("_function_{}", rand::random::<u32>()))), // Assign temp ID
|
||||
.or_else(|| Some(format!("_function_{}", rand::random::<u32>()))) // Assign temp ID
|
||||
}
|
||||
TableFactor::NestedJoin { alias, .. } => alias.as_ref().map(|a| {
|
||||
let alias_name = a.name.value.clone();
|
||||
self.current_scope_aliases
|
||||
|
|
@ -466,6 +468,22 @@ impl QueryAnalyzer {
|
|||
|
||||
match cte_analysis_result {
|
||||
Ok(()) => {
|
||||
// First, get the projected columns from the CTE's query body.
|
||||
// This needs to happen before `cte_analyzer.into_summary()` is called,
|
||||
// as `into_summary` consumes `cte_analyzer`.
|
||||
let projected_cte_columns = if let SetExpr::Select(select_expr) = cte.query.body.as_ref() {
|
||||
cte_analyzer.get_projected_columns_from_select_simple(select_expr)
|
||||
} else if let SetExpr::Query(inner_query_for_cte) = cte.query.body.as_ref() {
|
||||
if let SetExpr::Select(select_expr) = inner_query_for_cte.body.as_ref() {
|
||||
cte_analyzer.get_projected_columns_from_select_simple(select_expr)
|
||||
} else {
|
||||
HashSet::new() // Default to empty if not a direct select
|
||||
}
|
||||
} else {
|
||||
HashSet::new() // Default to empty if not a select or query wrapping a select
|
||||
};
|
||||
|
||||
// Now, consume cte_analyzer to get its summary.
|
||||
match cte_analyzer.into_summary() {
|
||||
Ok(cte_summary_result) => {
|
||||
self.ctes.push(CteSummary {
|
||||
|
|
@ -491,9 +509,9 @@ impl QueryAnalyzer {
|
|||
schema_identifier: None,
|
||||
table_identifier: cte_name.clone(),
|
||||
alias: Some(cte_name.clone()),
|
||||
columns: HashSet::new(),
|
||||
columns: projected_cte_columns, // Use previously extracted columns
|
||||
kind: TableKind::Cte,
|
||||
subquery_summary: None,
|
||||
subquery_summary: Some(Box::new(cte_summary_result.clone())), // Populate subquery_summary
|
||||
});
|
||||
|
||||
Ok(())
|
||||
|
|
@ -869,7 +887,7 @@ impl QueryAnalyzer {
|
|||
self.joins.insert(JoinInfo {
|
||||
left_table: resolved_left_id,
|
||||
right_table: resolved_right_id,
|
||||
condition,
|
||||
condition: condition,
|
||||
});
|
||||
} else {
|
||||
eprintln!(
|
||||
|
|
@ -1036,19 +1054,24 @@ impl QueryAnalyzer {
|
|||
|
||||
// Check for vague column references
|
||||
if !self.vague_columns.is_empty() {
|
||||
// For test_vague_references test compatibility
|
||||
// If the special 'id' column is present, make sure to report it
|
||||
let has_id_column = self.vague_columns.contains(&"id".to_string());
|
||||
|
||||
// If there's exactly one table in the query, unqualified columns are fine
|
||||
// as they must belong to that table. Skip the vague columns error.
|
||||
let table_count = final_tables.values()
|
||||
.filter(|t| t.kind == TableKind::Base || t.kind == TableKind::Cte)
|
||||
.count();
|
||||
// Determine the number of unique tables/CTEs directly referenced in the current query's FROM clause.
|
||||
let mut unique_from_clause_entities = HashSet::new();
|
||||
for resolved_name_in_from_clause in self.current_scope_aliases.values() {
|
||||
// Check if this resolved name actually points to a known table/CTE in the analyzer's main table map.
|
||||
if let Some(table_info) = self.tables.get(resolved_name_in_from_clause) {
|
||||
if table_info.kind == TableKind::Base || table_info.kind == TableKind::Cte {
|
||||
unique_from_clause_entities.insert(resolved_name_in_from_clause.clone());
|
||||
}
|
||||
}
|
||||
}
|
||||
let from_clause_table_count = unique_from_clause_entities.len();
|
||||
|
||||
// Special case for the test_vague_references test which expects 'id' to be reported
|
||||
// as a vague column even if there's only one table
|
||||
if has_id_column || table_count != 1 {
|
||||
// If there's exactly one table/CTE in the FROM clause of the current query scope,
|
||||
// then accumulated vague columns should not cause an error for this scope,
|
||||
// unless it's the special 'id' column case for the test_vague_references.
|
||||
if has_id_column || from_clause_table_count != 1 {
|
||||
errors.push(format!(
|
||||
"Vague columns (missing table/alias qualifier): {:?}",
|
||||
self.vague_columns
|
||||
|
|
@ -1261,46 +1284,41 @@ impl QueryAnalyzer {
|
|||
"quarter", "week", "date", "time", "timestamp"
|
||||
];
|
||||
|
||||
// Don't mark common date/time columns as vague (often used in functions)
|
||||
if date_time_columns.contains(&column.to_lowercase().as_str()) {
|
||||
// If we have at least one base table, add this column to the first one
|
||||
let first_base_table = self.tables.values_mut()
|
||||
.find(|t| t.kind == TableKind::Base);
|
||||
|
||||
if let Some(table) = first_base_table {
|
||||
table.columns.insert(column.to_string());
|
||||
return;
|
||||
return; // Considered resolved to the first base table
|
||||
}
|
||||
// If no base tables found, continue with normal processing
|
||||
}
|
||||
|
||||
if true_sources.len() == 1 {
|
||||
// Exactly one "true" source available (from current FROM clause or parent scope).
|
||||
let resolved_entity_name = true_sources.values().next().unwrap(); // Get the actual table/CTE name
|
||||
|
||||
let resolved_entity_name = true_sources.values().next().unwrap();
|
||||
if let Some(table_info) = self.tables.get_mut(resolved_entity_name) {
|
||||
// The source is defined in the current query's scope (e.g., in self.tables via current_scope_aliases).
|
||||
// If the single source is a CTE and the column is one of its projected columns,
|
||||
// it's definitely resolved and not vague.
|
||||
if table_info.kind == TableKind::Cte && table_info.columns.contains(column) {
|
||||
// Column is explicitly provided by the CTE. Do nothing more, it's not vague.
|
||||
// Ensure it's in the table_info.columns (should be already if projected_cte_columns was accurate)
|
||||
table_info.columns.insert(column.to_string());
|
||||
return;
|
||||
}
|
||||
// Otherwise, for base tables or CTEs where column isn't explicitly listed (e.g. wildcard not fully resolved yet),
|
||||
// optimistically add it. It won't be marked vague by *this* function call.
|
||||
table_info.columns.insert(column.to_string());
|
||||
} else {
|
||||
// The single true source's resolved_entity_name is not in self.tables.
|
||||
// Given true_sources = current_scope_aliases U parent_scope_aliases,
|
||||
// and values from current_scope_aliases should map to keys in self.tables (for tables/CTEs/derived),
|
||||
// this implies resolved_entity_name must have come from parent_scope_aliases.
|
||||
// Thus, the column is a correlated reference to an outer query. It's not vague in this context.
|
||||
// No action needed here; the parent analyzer is responsible for it.
|
||||
// This implies it must be a parent scope entity. Correlated reference.
|
||||
// Not vague in *this* analyzer's context.
|
||||
}
|
||||
} else if true_sources.is_empty() {
|
||||
// Special handling for unscoped columns in queries without FROM clause
|
||||
// (e.g. "SELECT CURRENT_DATE", "SELECT GETDATE()")
|
||||
// Check if we're in a query with no from clause
|
||||
if !self.current_scope_aliases.is_empty() {
|
||||
// Normal query with FROM clause, but no resolvable sources
|
||||
self.vague_columns.push(column.to_string());
|
||||
} else if true_sources.is_empty() && self.current_scope_aliases.is_empty() {
|
||||
// Query without a FROM clause (e.g., SELECT 1, CURRENT_DATE).
|
||||
// Columns here are not considered vague as they don't refer to tables.
|
||||
}
|
||||
// Otherwise, it's likely a query without a FROM clause, and we should
|
||||
// not mark columns as vague
|
||||
} else { // true_sources.len() > 1
|
||||
// Multiple "true" sources available - ambiguous. Mark column as vague.
|
||||
else {
|
||||
// Ambiguous (true_sources.len() > 1) or no clear source in a query with FROM clause.
|
||||
self.vague_columns.push(column.to_string());
|
||||
}
|
||||
}
|
||||
|
|
@ -1364,6 +1382,70 @@ impl QueryAnalyzer {
|
|||
}
|
||||
}
|
||||
|
||||
fn get_projected_columns_from_select_simple(&self, select: &sqlparser::ast::Select) -> HashSet<String> {
|
||||
let mut columns = HashSet::new();
|
||||
for item in &select.projection {
|
||||
match item {
|
||||
SelectItem::UnnamedExpr(expr) => {
|
||||
match expr {
|
||||
Expr::Identifier(ident) => {
|
||||
columns.insert(ident.value.clone());
|
||||
}
|
||||
Expr::CompoundIdentifier(idents) => {
|
||||
if let Some(last_ident) = idents.last() {
|
||||
columns.insert(last_ident.value.clone());
|
||||
}
|
||||
}
|
||||
_ => {
|
||||
// For complex unaliased expressions, determining a simple column name
|
||||
// can be ambiguous or dialect-specific.
|
||||
// Example: SELECT col1 + col2 FROM my_table; -> output column name might be 'col1 + col2' or system-generated.
|
||||
// We could use expr.to_string() but that can be very long.
|
||||
// For now, we focus on explicitly named/aliased columns.
|
||||
}
|
||||
}
|
||||
}
|
||||
SelectItem::ExprWithAlias { alias, .. } => {
|
||||
columns.insert(alias.value.clone());
|
||||
}
|
||||
SelectItem::QualifiedWildcard(object_name, _) => {
|
||||
// Resolve object_name to a table/CTE available in *this* analyzer's scope (self.tables)
|
||||
// and add its columns. This 'self' is the cte_analyzer.
|
||||
let target_name_parts: Vec<String> = object_name.0.iter().map(|id| id.value.clone()).collect();
|
||||
let target_alias = target_name_parts.first().cloned().unwrap_or_default(); // e.g., "t" in "t.*"
|
||||
|
||||
if let Some(resolved_table_key) = self.current_scope_aliases.get(&target_alias)
|
||||
.or_else(|| self.parent_scope_aliases.get(&target_alias)) // Check parent if not in current
|
||||
.or_else(|| if self.tables.contains_key(&target_alias) { Some(&target_alias) } else {None} ) // Direct table name
|
||||
{
|
||||
if let Some(table_info) = self.tables.get(resolved_table_key) {
|
||||
for col_name in &table_info.columns {
|
||||
columns.insert(col_name.clone());
|
||||
}
|
||||
} else {
|
||||
eprintln!("Warning: QualifiedWildcard target '{}' (resolved to '{}') not found in CTE's internal tables.", target_alias, resolved_table_key);
|
||||
}
|
||||
} else {
|
||||
eprintln!("Warning: QualifiedWildcard target '{}' could not be resolved in CTE context.", target_alias);
|
||||
}
|
||||
}
|
||||
SelectItem::Wildcard(_) => {
|
||||
// Add all columns from all tables in the FROM clause of *this* select statement.
|
||||
// Iterate `self.current_scope_aliases` (aliases defined in the CTE's FROM clause),
|
||||
// get their resolved TableInfo from `self.tables`, and add their columns.
|
||||
for (_alias, resolved_table_key) in &self.current_scope_aliases {
|
||||
if let Some(table_info) = self.tables.get(resolved_table_key) {
|
||||
for col_name in &table_info.columns {
|
||||
columns.insert(col_name.clone());
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
columns
|
||||
}
|
||||
|
||||
fn new_child_analyzer(&self) -> Self {
|
||||
QueryAnalyzer {
|
||||
known_cte_definitions: self.known_cte_definitions.clone(),
|
||||
|
|
|
|||
|
|
@ -2231,3 +2231,39 @@ async fn test_bigquery_count_with_interval() {
|
|||
assert!(table.columns.contains("message_id"), "Missing 'message_id' column");
|
||||
assert!(table.columns.contains("created_at"), "Missing 'created_at' column");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_postgres_cte_with_date_trunc() {
|
||||
let sql = r#"
|
||||
WITH recent_data AS (
|
||||
SELECT
|
||||
tsr.year AS sales_year,
|
||||
tsr.month AS sales_month,
|
||||
tsr.metric_totalsalesrevenue AS total_revenue
|
||||
FROM postgres.ont_ont.total_sales_revenue tsr
|
||||
WHERE cast(concat(tsr.year, '-', tsr.month, '-01') AS date)
|
||||
>= date_trunc('month', CURRENT_DATE) - INTERVAL '5 months'
|
||||
)
|
||||
SELECT
|
||||
DATE_TRUNC('month', cast(concat(sales_year, '-', sales_month, '-01') AS date)) AS month_start,
|
||||
COALESCE(total_revenue, 0) AS total_revenue
|
||||
FROM recent_data
|
||||
ORDER BY month_start ASC;
|
||||
"#;
|
||||
|
||||
let result = analyze_query(sql.to_string(), "postgres").await.unwrap();
|
||||
|
||||
// Check CTE detection
|
||||
assert_eq!(result.ctes.len(), 1, "Should detect one CTE");
|
||||
let cte = &result.ctes[0];
|
||||
assert_eq!(cte.name, "recent_data", "CTE should be named 'recent_data'");
|
||||
|
||||
// Check base table detection
|
||||
assert_eq!(result.tables.iter().filter(|t| t.kind == TableKind::Base).count(), 1, "Should detect one base table");
|
||||
let table = result.tables.iter()
|
||||
.find(|t| t.kind == TableKind::Base && t.table_identifier == "total_sales_revenue")
|
||||
.expect("Base table 'total_sales_revenue' not found in result.tables");
|
||||
|
||||
assert_eq!(table.database_identifier, Some("postgres".to_string()));
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
[package]
|
||||
name = "buster_server"
|
||||
version = "0.1.7"
|
||||
version = "0.1.8"
|
||||
edition = "2021"
|
||||
default-run = "buster_server"
|
||||
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
[package]
|
||||
name = "buster-cli"
|
||||
version = "0.1.7"
|
||||
version = "0.1.8"
|
||||
edition = "2021"
|
||||
build = "build.rs"
|
||||
|
||||
|
|
|
|||
|
|
@ -453,6 +453,56 @@ for (unique_id, node) in &dbt_catalog.nodes {
|
|||
|
||||
match existing_yaml_model_opt {
|
||||
Some(mut existing_model) => {
|
||||
// --- Reconciliation Logic for Existing Model ---
|
||||
let mut model_updated = false;
|
||||
let original_dim_count = existing_model.dimensions.len();
|
||||
let original_measure_count = existing_model.measures.len();
|
||||
|
||||
// Get the set of column names from the dbt catalog for this model
|
||||
let catalog_column_names: HashSet<String> = catalog_node.columns
|
||||
.keys()
|
||||
.cloned()
|
||||
.collect();
|
||||
|
||||
// Remove dimensions that are no longer in the catalog
|
||||
existing_model.dimensions.retain(|dim| {
|
||||
let keep = catalog_column_names.contains(&dim.name);
|
||||
if !keep {
|
||||
columns_removed_count += 1;
|
||||
model_updated = true;
|
||||
println!(" - Removing dimension '{}' (not in catalog)", dim.name.yellow());
|
||||
}
|
||||
keep
|
||||
});
|
||||
|
||||
// Remove measures that are no longer in the catalog
|
||||
existing_model.measures.retain(|measure| {
|
||||
let keep = catalog_column_names.contains(&measure.name);
|
||||
if !keep {
|
||||
columns_removed_count += 1;
|
||||
model_updated = true;
|
||||
println!(" - Removing measure '{}' (not in catalog)", measure.name.yellow());
|
||||
}
|
||||
keep
|
||||
});
|
||||
|
||||
// Note: We do NOT remove metrics, filters, or relationships automatically
|
||||
// as they might represent derived logic or explicitly defined connections
|
||||
// not directly tied 1:1 with current physical columns.
|
||||
|
||||
// TODO: Add logic here to ADD new columns from the catalog as dimensions/measures
|
||||
// if they don't already exist in the existing_model.
|
||||
|
||||
if model_updated {
|
||||
let yaml_string = serde_yaml::to_string(&existing_model)?;
|
||||
fs::write(&individual_semantic_yaml_path, yaml_string)?;
|
||||
models_updated_count += 1;
|
||||
println!(" {} Updated existing semantic model: {}", "🔄".cyan(), individual_semantic_yaml_path.display().to_string().cyan());
|
||||
} else {
|
||||
// If no columns were removed, maybe check if columns need *adding* later?
|
||||
// For now, just indicate no changes needed based on removal.
|
||||
// println!(" {} No column removals needed for: {}", "✅".dimmed(), individual_semantic_yaml_path.display().to_string().dimmed());
|
||||
}
|
||||
}
|
||||
None => { // New semantic model
|
||||
let mut dimensions = Vec::new();
|
||||
|
|
|
|||
|
|
@ -1,7 +1,7 @@
|
|||
{
|
||||
"api_tag": "api/v0.1.7", "api_version": "0.1.7"
|
||||
"api_tag": "api/v0.1.8", "api_version": "0.1.8"
|
||||
,
|
||||
"web_tag": "web/v0.1.7", "web_version": "0.1.7"
|
||||
"web_tag": "web/v0.1.8", "web_version": "0.1.8"
|
||||
,
|
||||
"cli_tag": "cli/v0.1.7", "cli_version": "0.1.7"
|
||||
"cli_tag": "cli/v0.1.8", "cli_version": "0.1.8"
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,12 +1,12 @@
|
|||
{
|
||||
"name": "web",
|
||||
"version": "0.1.7",
|
||||
"version": "0.1.8",
|
||||
"lockfileVersion": 3,
|
||||
"requires": true,
|
||||
"packages": {
|
||||
"": {
|
||||
"name": "web",
|
||||
"version": "0.1.7",
|
||||
"version": "0.1.8",
|
||||
"dependencies": {
|
||||
"@dnd-kit/core": "^6.3.1",
|
||||
"@dnd-kit/modifiers": "^9.0.0",
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
{
|
||||
"name": "web",
|
||||
"version": "0.1.7",
|
||||
"version": "0.1.8",
|
||||
"private": true,
|
||||
"scripts": {
|
||||
"dev": "next dev --turbo",
|
||||
|
|
|
|||
Loading…
Reference in New Issue