Skip to content

feat: fan out multi-column-alias lineage to every alias #60

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
takaebato merged 1 commit into from
Jul 5, 2026
Merged

feat: fan out multi-column-alias lineage to every alias #60

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
39 changes: 26 additions & 13 deletions sql-insight/src/resolver/binder.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -56,10 +56,10 @@ use sqlparser::ast::{
use sqlparser::tokenizer::Span;

use super::logical_plan::{
Aggregate, AlterTable, Assignment, Binding, BoundColumn, Columns, CreateTableAs, CreateView,
Cte, CteRef, Delete, Drop, Expr, Filter, Insert, Join, LogicalPlan, Merge, MergeClause,
NamedExpr, Projection, Scan, SchemaSource, SetOp, Sort, SubqueryAlias, TableFunction, Update,
Values, With,
output_slots, slot_count, Aggregate, AlterTable, Assignment, Binding, BoundColumn, Columns,
CreateTableAs, CreateView, Cte, CteRef, Delete, Drop, Expr, Filter, Insert, Join, LogicalPlan,
Merge, MergeClause, NamedExpr, OutputNames, Projection, Scan, SchemaSource, SetOp, Sort,
SubqueryAlias, TableFunction, Update, Values, With,
};
use super::origins::output_operands;
use crate::casing::{CaseRule, IdentifierStyle};
Expand Down Expand Up @@ -330,10 +330,8 @@ impl<'a> Binder<'a> {
let Some(operand) = operands.first() else {
return;
};
let anonymous = operand
.outputs
.iter()
.filter(|ne| ne.name.is_none())
let anonymous = output_slots(operand.outputs)
.filter(|(name, _)| name.is_none())
.count();
if anonymous == 0 {
return;
Expand Down Expand Up @@ -370,9 +368,9 @@ impl<'a> Binder<'a> {
return;
}
if let Some(operand) = output_operands(input).first() {
let outputs = operand.outputs;
if !outputs.is_empty() && outputs.len() != explicit.len() {
self.record_created_columns_arity_mismatch(target, explicit.len(), outputs.len());
let outputs = slot_count(operand.outputs);
if outputs != 0 && outputs != explicit.len() {
self.record_created_columns_arity_mismatch(target, explicit.len(), outputs);
}
}
}
Expand Down Expand Up @@ -679,8 +677,23 @@ fn rename_outputs(op: &mut LogicalPlan, names: &[Ident]) {
}
match op {
LogicalPlan::Projection(p) => {
for (ne, n) in p.exprs.iter_mut().zip(names) {
ne.name = Some(n.clone());
let mut names = names.iter();
for ne in p.exprs.iter_mut() {
match &mut ne.names {
OutputNames::Single(slot) => {
if let Some(n) = names.next() {
*slot = Some(n.clone());
}
}
// A fan occupies one position per name — each renames.
OutputNames::Fan(fan) => {
for slot in fan.iter_mut() {
if let Some(n) = names.next() {
*slot = n.clone();
}
}
}
}
}
}
LogicalPlan::Sort(s) => rename_outputs(&mut s.input, names),
Expand Down
63 changes: 24 additions & 39 deletions sql-insight/src/resolver/binder/expr.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,38 +8,24 @@ impl<'a> Binder<'a> {
pub(super) fn bind_select_item(&mut self, item: &SelectItem, scope: &Scope) -> Vec<NamedExpr> {
match item {
SelectItem::UnnamedExpr(expr) => vec![NamedExpr {
name: inferred_name(expr),
names: OutputNames::Single(inferred_name(expr)),
expr: self.bind_expr(expr, scope),
}],
SelectItem::ExprWithAlias { expr, alias } => vec![NamedExpr {
name: Some(alias.clone()),
names: OutputNames::Single(Some(alias.clone())),
expr: self.bind_expr(expr, scope),
}],
// Spark `expr AS (a, b, …)`: one expression projected under several
// output names (e.g. `explode(arr) AS (key, value)`). Because
// `reads` is occurrence-based, the expression's columns must be
// counted exactly once — so only the first alias carries the bound
// expression (with its reads / lineage); the remaining aliases are
// extra output columns that trace to nothing, like a
// `(VALUES …) AS v(x)` derived column. Splitting one expression's
// lineage across N outputs isn't representable without re-reading
// it, so those tail outputs are best-effort.
SelectItem::ExprWithAliases { expr, aliases } => {
let bound = self.bind_expr(expr, scope);
let mut names = aliases.iter();
let head = NamedExpr {
name: names.next().cloned(),
expr: bound,
};
std::iter::once(head)
.chain(names.map(|a| NamedExpr {
name: Some(a.clone()),
// An empty `Call` reads nothing and originates from
// nothing (a `Derived` output).
expr: Expr::Call { args: Vec::new() },
}))
.collect()
}
// output names (e.g. `explode(arr) AS (key, value)`) — one *fan*
// item occupying one output position per alias. The expression
// exists once, so `reads` count it once (occurrence-based); the
// slot view expands the names, so lineage fans out — every output
// column traces to the expression's sources (`arr → key` *and*
// `arr → value`).
SelectItem::ExprWithAliases { expr, aliases } => vec![NamedExpr {
names: OutputNames::Fan(aliases.clone()),
expr: self.bind_expr(expr, scope),
}],
// A wildcard isn't expanded (the rigor cost is too high for a
// SQL-text-only library); record it so consumers know this
// projection's column lineage is incomplete. A `REPLACE (expr AS
Expand Down Expand Up @@ -67,7 +53,7 @@ impl<'a> Binder<'a> {
// explicit outputs follow.
let mut out = match kind {
SelectItemQualifiedWildcardKind::Expr(expr) => vec![NamedExpr {
name: None,
names: OutputNames::Single(None),
expr: Expr::Call {
args: vec![self.bind_expr(expr, scope)],
},
Expand All @@ -94,7 +80,7 @@ impl<'a> Binder<'a> {
.iter()
.flat_map(|replace| &replace.items)
.map(|element| NamedExpr {
name: Some(element.column_name.clone()),
names: OutputNames::Single(Some(element.column_name.clone())),
expr: self.bind_expr(&element.expr, scope),
})
.collect()
Expand Down Expand Up @@ -767,28 +753,27 @@ impl<'a> Binder<'a> {
pub(super) fn output_cols(&self, exprs: &[NamedExpr]) -> Vec<OutputCol> {
exprs
.iter()
.map(|ne| {
.flat_map(|ne| {
// An identity output re-reads a real base column (so a later
// clause-alias / pipe reference to it reads that column). Only a
// `Base` column qualifies: a `Derived` passthrough (a pipe-
// carried alias, a derived-table column) traces back through the
// projection chain, not to a base table — marking it identity
// would let a later stage fall through to the base relation and
// fabricate a phantom read.
let identity = match &ne.expr {
Expr::Column(c) => {
// fabricate a phantom read. A fan's names are always introduced
// aliases (never the column itself), so never identity.
let identity = match (&ne.names, &ne.expr) {
(OutputNames::Single(Some(name)), Expr::Column(c)) => {
matches!(c.binding, Binding::Base { .. })
&& ne
.name
.as_ref()
.is_some_and(|n| self.eq(self.style.casing.column, n, &c.name))
&& self.eq(self.style.casing.column, name, &c.name)
}
_ => false,
};
OutputCol {
name: ne.name.clone(),
// One `OutputCol` per output position (a fan expands).
(0..ne.names.count()).map(move |j| OutputCol {
name: ne.names.get(j).flatten().cloned(),
identity,
}
})
})
.collect()
}
Expand Down
14 changes: 8 additions & 6 deletions sql-insight/src/resolver/binder/query.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -184,7 +184,9 @@ impl<'a> Binder<'a> {
.iter()
.chain(group_by_expr)
.map(|e| NamedExpr {
name: e.expr.alias.clone().or_else(|| inferred_name(&e.expr.expr)),
names: OutputNames::Single(
e.expr.alias.clone().or_else(|| inferred_name(&e.expr.expr)),
),
expr: self.bind_expr(&e.expr.expr, scope),
})
.collect();
Expand Down Expand Up @@ -311,7 +313,7 @@ impl<'a> Binder<'a> {
expr: Expr::Column(Box::new(
self.resolve(std::slice::from_ref(&name), &pass_scope),
)),
name: Some(name),
names: OutputNames::Single(Some(name)),
})
.collect()
}
Expand All @@ -329,13 +331,13 @@ impl<'a> Binder<'a> {
for a in assignments {
for column in assignment_target_columns(&a.target) {
let ne = NamedExpr {
name: Some(column.clone()),
names: OutputNames::Single(Some(column.clone())),
expr: self.bind_expr(&a.value, scope),
};
// Pipe outputs are always single-named passthroughs.
match exprs.iter_mut().find(|e| {
e.name
.as_ref()
.is_some_and(|n| self.eq(self.style.casing.column, n, &column))
matches!(&e.names, OutputNames::Single(Some(n))
if self.eq(self.style.casing.column, n, &column))
}) {
Some(slot) => *slot = ne,
None => exprs.push(ne),
Expand Down
4 changes: 2 additions & 2 deletions sql-insight/src/resolver/binder/statement.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -207,7 +207,7 @@ impl<'a> Binder<'a> {
LogicalPlan::Values(v) => v.rows.first().map(Vec::len),
other => output_operands(other)
.first()
.map(|operand| operand.outputs.len())
.map(|operand| slot_count(operand.outputs))
.filter(|n| *n > 0),
}
};
Expand Down Expand Up @@ -311,7 +311,7 @@ impl<'a> Binder<'a> {
for assignment in &insert.assignments {
for column in assignment_target_columns(&assignment.target) {
exprs.push(NamedExpr {
name: Some(column.clone()),
names: OutputNames::Single(Some(column.clone())),
expr: self.bind_expr(&assignment.value, &scope),
});
columns.push(column);
Expand Down
33 changes: 21 additions & 12 deletions sql-insight/src/resolver/lineage.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,7 +12,8 @@
use sqlparser::ast::Ident;

use super::logical_plan::{
dml_roots, is_dml_root, peel_with, Expr, LogicalPlan, MergeClause, NamedExpr, Update,
dml_roots, is_dml_root, output_slots, peel_with, Expr, LogicalPlan, MergeClause, NamedExpr,
Update,
};
use super::origins::{
conflict_value_origins, enter_withs, origins_of_expr, output_operands, TraceContext,
Expand Down Expand Up @@ -197,18 +198,21 @@ fn query_output_lineage<'a>(
// column). Position already restarts per branch, aligning them; a plain
// query has a single operand, so this is a no-op there.
let result_names: Vec<Option<Ident>> = match operands.first() {
Some(o) => o.outputs.iter().map(|ne| ne.name.clone()).collect(),
Some(o) => output_slots(o.outputs).map(|(n, _)| n.cloned()).collect(),
None => Vec::new(),
};
for operand in &operands {
let outputs = operand.outputs;
operand.trace(context, |input, cx| {
for (position, ne) in outputs.iter().enumerate() {
// Position-by-position over the slot view: a fan yields its
// shared expression once per alias, so every alias of
// `explode(arr) AS (k, v)` gets its own `arr` edge.
for (position, (_, expr)) in output_slots(outputs).enumerate() {
let target = ColumnTarget::QueryOutput {
name: result_names.get(position).cloned().flatten(),
position,
};
emit_edges(origins_of_expr(&ne.expr, input, cx), target, out);
emit_edges(origins_of_expr(expr, input, cx), target, out);
}
});
}
Expand Down Expand Up @@ -263,9 +267,12 @@ fn relation_lineage<'a>(
for operand in output_operands(input) {
let outputs = operand.outputs;
operand.trace(context, |src_input, cx| {
for (target_column, ne) in columns.iter().zip(outputs) {
// Pair positionally with the slot view: a fan feeds one target
// column per alias (`INSERT … SELECT explode(arr) AS (k, v)`
// feeds both).
for (target_column, (_, expr)) in columns.iter().zip(output_slots(outputs)) {
let tgt = ColumnTarget::Relation(target_column.clone());
emit_edges(origins_of_expr(&ne.expr, src_input, cx), tgt, out);
emit_edges(origins_of_expr(expr, src_input, cx), tgt, out);
}
});
}
Expand Down Expand Up @@ -300,7 +307,7 @@ fn created_relation_lineage<'a>(
let operands = output_operands(input);
let result_names: Vec<Option<Ident>> = if explicit.is_empty() {
match operands.first() {
Some(o) => o.outputs.iter().map(|ne| ne.name.clone()).collect(),
Some(o) => output_slots(o.outputs).map(|(n, _)| n.cloned()).collect(),
None => Vec::new(),
}
} else {
Expand All @@ -309,7 +316,9 @@ fn created_relation_lineage<'a>(
for operand in &operands {
let outputs = operand.outputs;
operand.trace(context, |src_input, cx| {
for (position, ne) in outputs.iter().enumerate() {
// Position-by-position over the slot view, so a CTAS / CREATE
// VIEW over `explode(arr) AS (k, v)` feeds both created columns.
for (position, (_, expr)) in output_slots(outputs).enumerate() {
let Some(name) = result_names.get(position).cloned().flatten() else {
continue;
};
Expand All @@ -321,7 +330,7 @@ fn created_relation_lineage<'a>(
},
resolution: ResolutionKind::Inferred,
});
emit_edges(origins_of_expr(&ne.expr, src_input, cx), tgt, out);
emit_edges(origins_of_expr(expr, src_input, cx), tgt, out);
}
});
}
Expand All @@ -336,12 +345,12 @@ fn returning_lineage<'a>(
context: &mut TraceContext<'a>,
out: &mut Vec<ColumnLineageEdge>,
) {
for (position, ne) in returning.iter().enumerate() {
for (position, (name, expr)) in output_slots(returning).enumerate() {
let target = ColumnTarget::QueryOutput {
name: ne.name.clone(),
name: name.cloned(),
position,
};
emit_edges(origins_of_expr(&ne.expr, input, context), target, out);
emit_edges(origins_of_expr(expr, input, context), target, out);
}
}

Expand Down
Loading