diff --git a/docs/plans/2026-08-29-issue-142-column-lineage-cursor-subquery.md b/docs/plans/2026-08-29-issue-142-column-lineage-cursor-subquery.md new file mode 100644 index 0000000..f096022 --- /dev/null +++ b/docs/plans/2026-08-29-issue-142-column-lineage-cursor-subquery.md @@ -0,0 +1,881 @@ +# Issue #142 列级血缘穿透：游标 %ROWTYPE 记录变量与目标列表标量子查询 + +> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task. + +**Goal:** 修复列级血缘（`codeweb lineage t.c --direction upstream`）在两类写入形态下无法穿透到真实源列的问题：(1) `%ROWTYPE` 记录变量写入的三种残留盲区，(2) INSERT..SELECT 目标列表中的标量子查询。 + +**Architecture:** 改动全部位于解析层 `src/parser/extractor.rs` 的 `ColumnAccessExtractor`，不触及 store 结构（无需 bump `STORE_VERSION`）： +- `column_source()`：`%ROWTYPE` 记录字段解析增加「表锚定」与「`SELECT *` 游标 catch-all」两个回退分支； +- `push_column_mapping()`：增加 `Expr::Subquery` 值源分流，子查询首表达式在**子查询自身 FROM 作用域**（临时 alias map + `scope_sole_table` save/restore）下解析； +- `visit_insert()`：`InsertSource::RecordVariable`（整记录 `INSERT INTO t (a,b) VALUES r`）按游标源列位置展开。 + +血缘 walker（`graph/lineage.rs`）与 CLI（`main.rs`）无需改动——修复消除的是 `table: None` 源与空 `sources`，walker 现有递归逻辑（仅对 `ColumnSource::Column{table: Some(t)}` 递归）即可穿透。 + +**Tech Stack:** Rust，ogsql-parser（git 依赖），现有测试 harness（`tests/regress_column_lineage.rs` 端到端 + `src/parser/extractor.rs` 单测模块）。 + +**现状基线（已实测，本分支 feat/issue-142）：** + +| 场景 | 当前输出 | 目标 | +|---|---|---| +| `r cur%ROWTYPE` + 显式游标 + `VALUES (r.id, r.amt)` | `t_dst.id ← t_src.id` ✅（#148 已修） | 保持 + 特征测试锁定 | +| `r t_src%ROWTYPE`（表锚定）+ `VALUES (r.id, r.amt)` | `?.id` ❌ | `t_src.id` | +| `r cur%ROWTYPE` + `SELECT *` 游标 | `?.id` ❌ | `t_src.id`（列名取字段名） | +| 整记录 `INSERT INTO t (a,b) VALUES r`（游标锚定） | 无列映射 ❌ | 按游标源列位置映射 | +| INSERT..SELECT 目标列表标量子查询 | "No column lineage" ❌ | `t_ref.code` | +| 对照组：位置映射 `INSERT INTO t_out (id,code) SELECT s.id, s.amt FROM t_src s` | `t_src.id` ✅ | 保持 | + +--- + +## 关键代码位置（当前实现，改动点） + +`src/parser/extractor.rs`： + +```rust +// L3085 — column_source()：%ROWTYPE 记录字段解析（#147 L2） +fn column_source(&self, names: &[ogsql_parser::Ident]) -> ColumnSource { + let (alias_prefix, column) = split_alias_column(names); + if let Some(record) = &alias_prefix { + if let Some(cursor) = self.record_cursors.get(&record.to_lowercase()) { + if let Some(cols) = self.cursor_sources.get(cursor) { + if let Some(col) = cols.iter() + .find(|c| c.output_name.eq_ignore_ascii_case(&column)) + { + if !col.source_col.is_empty() { + return ColumnSource::Column { table: col.source_table.clone(), column: col.source_col.clone() }; + } + } + } + } + } + let table = match alias_prefix.as_ref() { + Some(a) => self.resolve_alias(a).map(|ta| ta.table.clone()), + None => self.scope_sole_table.clone(), + }; + ColumnSource::Column { table, column } +} +``` + +```rust +// L2883 — visit_insert() 的 INSERT..VALUES/RecordVariable 分支 +ogsql_parser::ast::InsertSource::DefaultValues +| ogsql_parser::ast::InsertSource::Set(_) +| ogsql_parser::ast::InsertSource::RecordVariable(_) => {} // L2930-2932 +``` + +```rust +// L3283 — push_column_mapping()：值源分发的唯一咽喉点（INSERT..SELECT 目标、 +// INSERT..VALUES、UPDATE SET、MERGE 全部经此） +fn push_column_mapping(&mut self, target_table: Option, target_column: String, + position: Option, value: &Expr) { + let (sources, kind, expression) = self.classify_value_expr(value); + self.column_mappings.push(ColumnMapping { target_table, target_column, position, sources, kind, expression }); +} +``` + +```rust +// L3178 — collect_value_sources()：L3241 显式丢弃子查询 +Expr::Exists(_) | Expr::Subquery(_) => {} // ← 标量子查询目标列表零源根因 +``` + +**AST 事实（ogsql-parser，已核实）**：目标列表裸标量子查询 `(SELECT ...)` 解析为 `Expr::Subquery(Box)`（ast/mod.rs:1258）；`Expr::ScalarSublink`（1259-1264）是 `expr OP ANY/ALL/SOME (subquery)` 谓词形态，非值源，不在本计划范围。 + +**测试基础设施（现有）**： +- 单测 helper：`column_mappings_of(sql)`（L4924）、`find_mapping`（L4947）、`sources_for`（L4953）、`col(table, column)`（L4960）。 +- `ColumnAccessExtractor::new_with_context(&ProcedureVarContext)`（L2078）可注入游标/记录上下文——单测接缝。 +- 端到端 harness：`tests/regress_column_lineage.rs` 的 `project_with_sql` + `lineage(root, target, dir, "tree")`。 + +--- + +## Task 1: 标量子查询目标解析（Case 2，主缺陷） + +**Files:** +- Modify: `src/parser/extractor.rs`（`push_column_mapping` L3283 分流 + 新增 `push_subquery_column_mapping`） +- Test: `src/parser/extractor.rs` mod tests（新增单测）+ `tests/regress_column_lineage.rs`（新增端到端） + +**Step 1: 写失败测试（单测，Red）** + +在 `src/parser/extractor.rs` 测试模块（`insert_values_maps_literals_and_columns` L5095 附近）新增： + +```rust +/// #142: a scalar subquery as an INSERT..SELECT target contributes the inner +/// select's FIRST expression as the source, resolved in the subquery's own FROM +/// scope. Correlated refs (`s.id` in WHERE) must NOT leak as sources. +#[test] +fn scalar_subquery_target_resolves_its_first_column() { + let maps = column_mappings_of( + "INSERT INTO t_out (id, code) \ + SELECT s.id, (SELECT r.code FROM t_ref r WHERE r.id = s.id) FROM t_src s", + ); + let m = find_mapping(&maps, "code"); + assert_eq!(m.kind, MappingKind::Direct); + assert_eq!(m.sources, vec![col(Some("t_ref"), "code")]); +} + +/// #142: the choke point is push_column_mapping, so INSERT..VALUES subqueries +/// resolve too. +#[test] +fn scalar_subquery_in_insert_values_resolves() { + let maps = column_mappings_of( + "INSERT INTO t_out (code) VALUES ((SELECT r.code FROM t_ref r WHERE r.id = 1))", + ); + assert_eq!( + find_mapping(&maps, "code").sources, + vec![col(Some("t_ref"), "code")] + ); +} +``` + +**Step 2: 运行确认失败** + +Run: `cargo test --features full scalar_subquery_target_resolves_its_first_column` +Expected: FAIL — `find_mapping(&maps, "code")` 命中映射但 `sources == []`（`collect_value_sources` L3241 丢弃 `Expr::Subquery`）。 + +**Step 3: 写端到端失败测试（Red）** + +`tests/regress_column_lineage.rs` 新增（放在 `cursor_fetch_insert_values_resolves_to_source_columns` 附近）： + +```rust +/// #142: a scalar subquery in the INSERT..SELECT target list must resolve to the +/// subquery's source column, not report "No column lineage". +#[test] +fn scalar_subquery_in_insert_select_target_resolves() { + let dir = TempDir::new().unwrap(); + let root = project_with_sql( + &dir, + r#" +CREATE TABLE t_src(id NUMBER, amt NUMBER); +CREATE TABLE t_ref(id NUMBER, code VARCHAR2(10)); +CREATE TABLE t_out(id NUMBER, code VARCHAR2(10)); +CREATE PROCEDURE p_copy_subquery AS BEGIN + INSERT INTO t_out (id, code) + SELECT s.id, (SELECT r.code FROM t_ref r WHERE r.id = s.id) FROM t_src s; +END; +"#, + ); + let out = lineage(&root, "t_out.code", "upstream", "tree"); + assert!( + !out.contains("No column lineage"), + "scalar subquery target must resolve:\n{out}" + ); + assert!( + out.contains("t_ref.code"), + "subquery source column missing:\n{out}" + ); +} +``` + +**Step 4: 运行确认失败** + +Run: `cargo test --features full --test regress_column_lineage scalar_subquery_in_insert_select_target_resolves` +Expected: FAIL（"No column lineage" 出现在输出中）。 + +**Step 5: 最小实现（Green）** + +改 `push_column_mapping`（L3283）为值源咽喉点分流，并新增 `push_subquery_column_mapping`（置于 `push_column_mapping` 之后、`visit_merge_statement` L3302 之前）： + +```rust + fn push_column_mapping( + &mut self, + target_table: Option, + target_column: String, + position: Option, + value: &Expr, + ) { + // #142: a scalar subquery as a value (`INSERT .. SELECT (SELECT ...)`, + // `VALUES ((SELECT ...))`, `SET x = (SELECT ...)`, MERGE values) contributes + // the inner select's FIRST output expression as the source, resolved in the + // subquery's own FROM scope. + if let Expr::Subquery(select) = peel_parenthesized(value) { + self.push_subquery_column_mapping(target_table, target_column, position, select); + return; + } + let (sources, kind, expression) = self.classify_value_expr(value); + self.column_mappings.push(ColumnMapping { + target_table, + target_column, + position, + sources, + kind, + expression, + }); + } + + /// Column mapping for `target = (SELECT first_expr FROM ...)`: resolve the + /// subquery's first select-list expression against the subquery's own FROM + /// aliases, then restore the enclosing statement's scope. Correlated + /// references (`s.id` in the subquery's WHERE) are not value sources and are + /// intentionally not collected — only the first select-list expression feeds + /// the written column. + fn push_subquery_column_mapping( + &mut self, + target_table: Option, + target_column: String, + position: Option, + select: &SelectStatement, + ) { + let saved_alias_map = self.alias_map.clone(); + self.collect_aliases_from_table_refs(&select.from); + let new_scope = self.scope_sole_table_of(&select.from); + let saved_scope = std::mem::replace(&mut self.scope_sole_table, new_scope); + + let mut sources = Vec::new(); + let mut kind = MappingKind::Derived; + let mut expression: Option = None; + if let Some(SelectTarget::Expr(first, _)) = select.targets.first() { + let first = peel_parenthesized(first); + self.collect_value_sources(first, &mut sources); + // An entirely-literal first target (`(SELECT 'x' FROM dual)`) is a + // constant; collect_value_sources skips literals by design, so record + // it here as a Literal source rather than leaving the mapping empty. + if sources.is_empty() { + if let Expr::Literal(lit) = first { + sources.push(ColumnSource::Literal { + value: format_literal_short(lit), + }); + } + } + if matches!(sources.as_slice(), [ColumnSource::Column { .. }]) { + kind = MappingKind::Direct; + } + expression = Some(format_expr_short(first)); + } + + self.scope_sole_table = saved_scope; + self.alias_map = saved_alias_map; + + self.column_mappings.push(ColumnMapping { + target_table, + target_column, + position, + sources, + kind, + // A plain copy needs no expression text (mirrors `insert_select_maps_columns_by_position`). + expression: if matches!(kind, MappingKind::Direct) { + None + } else { + expression + }, + }); + } +``` + +**Step 6: 运行确认通过** + +Run: `cargo test --features full scalar_subquery` +Expected: PASS（两个单测）。 + +Run: `cargo test --features full --test regress_column_lineage scalar_subquery_in_insert_select_target_resolves` +Expected: PASS（`t_ref.code` 出现在输出，无 "No column lineage"）。 + +**Step 7: 回归现有单测** + +Run: `cargo test --features full insert_values_maps_literals_and_columns` +Run: `cargo test --features full insert_select_maps_columns_by_position` +Expected: PASS（子查询分流不影响普通值）。 + +**Step 8: Commit** + +```bash +git add src/parser/extractor.rs tests/regress_column_lineage.rs +git commit -m "feat(lineage): 标量子查询作为 INSERT 值源时解析其首表达式 (fix #142 部分)" +``` + +--- + +## Task 2: 表锚定 %ROWTYPE 记录字段（Case 1a） + +**Files:** +- Modify: `src/parser/extractor.rs`（`column_source` L3085 的 record 分支加 else 回退） +- Test: 单测 + `tests/regress_column_lineage.rs` + +**Step 1: 写失败测试（单测，Red）** + +```rust +/// #142: a `rec t%ROWTYPE` record (anchor is a TABLE, not a registered cursor) +/// resolves its fields to that table's columns. +#[test] +fn table_rowtype_record_field_resolves_to_table_column() { + let mut ctx = ProcedureVarContext::default(); + ctx.record_cursors.insert("r".to_string(), "t_src".to_string()); + let maps = column_mappings_of_with_context( + "INSERT INTO t_dst (id, amt) VALUES (r.id, r.amt)", + &ctx, + ); + assert_eq!( + find_mapping(&maps, "id").sources, + vec![col(Some("t_src"), "id")] + ); + assert_eq!( + find_mapping(&maps, "amt").sources, + vec![col(Some("t_src"), "amt")] + ); +} +``` + +新增测试 helper（放在 `column_mappings_of` L4929 之后）： + +```rust + /// Column mappings with a seeded procedure variable context (#142): lets a + /// standalone INSERT walk see cursor/record bindings that in real procedures + /// come from the DECLARE block. + fn column_mappings_of_with_context(sql: &str, ctx: &ProcedureVarContext) -> Vec { + let tokens = Tokenizer::new(sql).tokenize().unwrap(); + let mut parser = ogsql_parser::Parser::with_source(tokens, sql.to_string()); + let stmts = parser.parse_with_text(); + let mut result = Vec::new(); + for info in &stmts { + let mut extractor = ColumnAccessExtractor::new_with_context(ctx); + walk_statement(&mut extractor, &info.statement); + result.extend(extractor.finish().column_mappings); + } + result + } +``` + +**Step 2: 运行确认失败** + +Run: `cargo test --features full table_rowtype_record_field_resolves_to_table_column` +Expected: FAIL — `sources == [col(None, "id")]`（anchor 查 `cursor_sources` 落空，走兜底 `table: None` → `?.id`）。 + +**Step 3: 写端到端失败测试（Red）** + +```rust +/// #142: a table-anchored %ROWTYPE record (`r t_src%ROWTYPE`) written via +/// `VALUES (r.id, r.amt)` must resolve to t_src columns, not "?.id". +#[test] +fn table_rowtype_record_insert_values_resolves_to_table() { + let dir = TempDir::new().unwrap(); + let root = project_with_sql( + &dir, + r#" +CREATE TABLE t_src(id NUMBER, amt NUMBER); +CREATE TABLE t_dst(id NUMBER, amt NUMBER); +CREATE PROCEDURE p_table_rowtype AS + r t_src%ROWTYPE; + CURSOR cur IS SELECT id, amt FROM t_src; +BEGIN + OPEN cur; + LOOP + FETCH cur INTO r; + EXIT WHEN cur%NOTFOUND; + INSERT INTO t_dst (id, amt) VALUES (r.id, r.amt); + END LOOP; + CLOSE cur; +END; +"#, + ); + let out = lineage(&root, "t_dst.id", "upstream", "tree"); + assert!( + out.contains("t_src.id"), + "table-anchored record field must resolve:\n{out}" + ); + assert!( + !out.contains("?.id"), + "table-anchored record field must not stay unattributed:\n{out}" + ); +} +``` + +**Step 4: 运行确认失败** + +Run: `cargo test --features full --test regress_column_lineage table_rowtype_record_insert_values_resolves_to_table` +Expected: FAIL（输出含 `?.id`）。 + +**Step 5: 最小实现（Green）** + +改 `column_source`（L3085）的 record 分支——在 `cursor_sources.get(cursor)` 的 `Some` 分支之外加 `else` 回退： + +```rust + fn column_source(&self, names: &[ogsql_parser::Ident]) -> ColumnSource { + let (alias_prefix, column) = split_alias_column(names); + + // `%ROWTYPE` record field (issue #147 L2): `rec.id` where rec is a record + // resolves to the cursor's source column by output name. + if let Some(record) = &alias_prefix { + if let Some(cursor) = self.record_cursors.get(&record.to_lowercase()) { + if let Some(cols) = self.cursor_sources.get(cursor) { + if let Some(col) = cols + .iter() + .find(|c| c.output_name.eq_ignore_ascii_case(&column)) + { + if !col.source_col.is_empty() { + return ColumnSource::Column { + table: col.source_table.clone(), + column: col.source_col.clone(), + }; + } + } + } else { + // #142: the `%ROWTYPE` anchor is a TABLE, not a registered + // cursor (`rec t_src%ROWTYPE`): the record's fields are that + // table's columns. (A custom record TYPE anchor is rare; it + // would attribute the type name as a table — the field is + // still attributable, unlike the old `?.field`.) + return ColumnSource::Column { + table: Some(cursor.clone()), + column: column.clone(), + }; + } + } + } + + let table = match alias_prefix.as_ref() { + Some(a) => self.resolve_alias(a).map(|ta| ta.table.clone()), + None => self.scope_sole_table.clone(), + }; + ColumnSource::Column { table, column } + } +``` + +**Step 6: 运行确认通过** + +Run: `cargo test --features full table_rowtype_record_field_resolves_to_table_column` +Expected: PASS。 + +Run: `cargo test --features full --test regress_column_lineage table_rowtype_record_insert_values_resolves_to_table` +Expected: PASS。 + +**Step 7: 回归 Task 1 与游标路径** + +Run: `cargo test --features full scalar_subquery` +Run: `cargo test --features full cursor_fetch_insert_values_resolves_to_source_columns` +Expected: PASS（`else` 回退不影响已注册游标路径——`cursor_sources.get` 命中时走原逻辑）。 + +**Step 8: Commit** + +```bash +git add src/parser/extractor.rs tests/regress_column_lineage.rs +git commit -m "feat(lineage): 表锚定 %ROWTYPE 记录字段解析为表列 (fix #142 部分)" +``` + +--- + +## Task 3: SELECT * 游标 + 记录字段（Case 1b） + +**Files:** +- Modify: `src/parser/extractor.rs`（`column_source` L3085 record 分支内加 catch-all 匹配） +- Test: 单测 + `tests/regress_column_lineage.rs` + +**Step 1: 写失败测试（单测，Red）** + +```rust +/// #142: a `SELECT *` cursor produces a single catch-all cursor source (empty +/// output name, table attributed). Record fields over it attribute to the +/// cursor's table under the field's own name. +#[test] +fn star_cursor_rowtype_record_field_attributes_to_cursor_table() { + let mut ctx = ProcedureVarContext::default(); + ctx.cursor_sources.insert( + "cur".to_string(), + vec![CursorColumn { + output_name: String::new(), + source_table: Some("t_src".to_string()), + source_col: String::new(), + }], + ); + ctx.record_cursors.insert("r".to_string(), "cur".to_string()); + let maps = column_mappings_of_with_context( + "INSERT INTO t_dst (id, amt) VALUES (r.id, r.amt)", + &ctx, + ); + assert_eq!( + find_mapping(&maps, "id").sources, + vec![col(Some("t_src"), "id")] + ); + assert_eq!( + find_mapping(&maps, "amt").sources, + vec![col(Some("t_src"), "amt")] + ); +} +``` + +**Step 2: 运行确认失败** + +Run: `cargo test --features full star_cursor_rowtype_record_field_attributes_to_cursor_table` +Expected: FAIL — `sources == [col(None, "id")]`（catch-all 的 `output_name` 为空，`find` 按 output_name 匹配不到）。 + +**Step 3: 写端到端失败测试（Red）** + +```rust +/// #142: `SELECT *` cursor + `%ROWTYPE` record fields must resolve to the +/// cursor's table (columns attributed under the field names), not "?.id". +#[test] +fn star_cursor_rowtype_record_resolves_to_cursor_table() { + let dir = TempDir::new().unwrap(); + let root = project_with_sql( + &dir, + r#" +CREATE TABLE t_src(id NUMBER, amt NUMBER); +CREATE TABLE t_dst(id NUMBER, amt NUMBER); +CREATE PROCEDURE p_star_cursor AS + CURSOR cur IS SELECT * FROM t_src; + r cur%ROWTYPE; +BEGIN + OPEN cur; + LOOP + FETCH cur INTO r; + EXIT WHEN cur%NOTFOUND; + INSERT INTO t_dst (id, amt) VALUES (r.id, r.amt); + END LOOP; + CLOSE cur; +END; +"#, + ); + let out = lineage(&root, "t_dst.id", "upstream", "tree"); + assert!( + out.contains("t_src.id"), + "star-cursor record field must resolve:\n{out}" + ); + assert!( + !out.contains("?.id"), + "star-cursor record field must not stay unattributed:\n{out}" + ); +} +``` + +**Step 4: 运行确认失败** + +Run: `cargo test --features full --test regress_column_lineage star_cursor_rowtype_record_resolves_to_cursor_table` +Expected: FAIL（输出含 `?.id`）。 + +**Step 5: 最小实现（Green）** + +在 `column_source` record 分支的 `Some(cols)` 内、`find` 匹配失败之后追加 catch-all 匹配： + +```rust + if let Some(cols) = self.cursor_sources.get(cursor) { + if let Some(col) = cols + .iter() + .find(|c| c.output_name.eq_ignore_ascii_case(&column)) + { + if !col.source_col.is_empty() { + return ColumnSource::Column { + table: col.source_table.clone(), + column: col.source_col.clone(), + }; + } + } + // #142: a single catch-all cursor source (empty output name — + // `SELECT *` cursor, or dynamic-SQL attribution) covers every + // record field: the exact column is unknown, attribute to the + // cursor's table under the field's own name (same philosophy as + // `resolve_cursor_flows`). + if let [single] = cols.as_slice() { + if single.output_name.is_empty() { + if let Some(ref t) = single.source_table { + return ColumnSource::Column { + table: Some(t.clone()), + column: column.clone(), + }; + } + } + } + } else { + // #142: table-anchored %ROWTYPE (Task 2) + return ColumnSource::Column { + table: Some(cursor.clone()), + column: column.clone(), + }; + } +``` + +**Step 6: 运行确认通过** + +Run: `cargo test --features full star_cursor_rowtype_record_field_attributes_to_cursor_table` +Expected: PASS。 + +Run: `cargo test --features full --test regress_column_lineage star_cursor_rowtype_record_resolves_to_cursor_table` +Expected: PASS。 + +**Step 7: 回归** + +Run: `cargo test --features full table_rowtype_record_field_resolves_to_table_column` +Expected: PASS。 + +**Step 8: Commit** + +```bash +git add src/parser/extractor.rs tests/regress_column_lineage.rs +git commit -m "feat(lineage): SELECT * 游标 + %ROWTYPE 记录字段归因到游标表 (fix #142 部分)" +``` + +--- + +## Task 4: 整记录写入 INSERT ... VALUES r（Case 1c，游标锚定） + +**Files:** +- Modify: `src/parser/extractor.rs`（`visit_insert` L2928-2932，`RecordVariable` 分支拆出） +- Test: 单测 + `tests/regress_column_lineage.rs` + +**Step 1: 写失败测试（单测，Red）** + +```rust +/// #142: `INSERT INTO t (a, b) VALUES r` with a cursor-anchored %ROWTYPE record +/// expands the record's fields positionally through the cursor's SELECT sources. +#[test] +fn whole_record_insert_expands_cursor_rowtype_fields() { + let mut ctx = ProcedureVarContext::default(); + ctx.cursor_sources.insert( + "cur".to_string(), + vec![ + CursorColumn { + output_name: "id".to_string(), + source_table: Some("t_src".to_string()), + source_col: "id".to_string(), + }, + CursorColumn { + output_name: "amt".to_string(), + source_table: Some("t_src".to_string()), + source_col: "amt".to_string(), + }, + ], + ); + ctx.record_cursors.insert("r".to_string(), "cur".to_string()); + let maps = column_mappings_of_with_context( + "INSERT INTO t_dst (id, amt) VALUES r", + &ctx, + ); + assert_eq!( + find_mapping(&maps, "id").sources, + vec![col(Some("t_src"), "id")] + ); + assert_eq!( + find_mapping(&maps, "amt").sources, + vec![col(Some("t_src"), "amt")] + ); +} +``` + +> 实现时先验证 `INSERT INTO t (a,b) VALUES r` 解析为 `InsertSource::RecordVariable(Expr::PlVariable(["r"]))`（可用 `dbg!` 临时打印或先跑此测试看失败形态；若实际为 `Values([PlVariable])` 则改动点移到 Values 分支，逻辑相同——按实测调整）。 + +**Step 2: 运行确认失败** + +Run: `cargo test --features full whole_record_insert_expands_cursor_rowtype_fields` +Expected: FAIL — `find_mapping` panic "no mapping for id"（RecordVariable 分支当前被跳过，产生零映射）。 + +**Step 3: 写端到端失败测试（Red）** + +```rust +/// #142: whole-record insert `INSERT INTO t_dst (id, amt) VALUES r` (cursor-anchored +/// %ROWTYPE) must resolve positionally through the cursor's sources. +#[test] +fn whole_record_insert_values_r_resolves_through_cursor() { + let dir = TempDir::new().unwrap(); + let root = project_with_sql( + &dir, + r#" +CREATE TABLE t_src(id NUMBER, amt NUMBER); +CREATE TABLE t_dst(id NUMBER, amt NUMBER); +CREATE PROCEDURE p_rec_insert AS + CURSOR cur IS SELECT id, amt FROM t_src; + r cur%ROWTYPE; +BEGIN + OPEN cur; + LOOP + FETCH cur INTO r; + EXIT WHEN cur%NOTFOUND; + INSERT INTO t_dst (id, amt) VALUES r; + END LOOP; + CLOSE cur; +END; +"#, + ); + let out = lineage(&root, "t_dst.amt", "upstream", "tree"); + assert!( + out.contains("t_src.amt"), + "whole-record insert must resolve through the cursor:\n{out}" + ); +} +``` + +**Step 4: 运行确认失败** + +Run: `cargo test --features full --test regress_column_lineage whole_record_insert_values_r_resolves_through_cursor` +Expected: FAIL（无映射 → "No column lineage"）。 + +**Step 5: 最小实现（Green）** + +改 `visit_insert` L2928-2932，把 `RecordVariable` 拆出独立分支： + +```rust + // DEFAULT VALUES has no sources; `SET` is handled as assignments. + ogsql_parser::ast::InsertSource::DefaultValues + | ogsql_parser::ast::InsertSource::Set(_) => {} + // #142: `INSERT INTO t (a, b) VALUES r` — expand the record's + // fields through its `%ROWTYPE` anchor. Cursor-anchored records + // resolve positionally through the cursor's SELECT sources; a + // table-anchored record needs the table's column order (DDL), + // which is unavailable here, so it is left unresolved (documented + // limitation). + ogsql_parser::ast::InsertSource::RecordVariable(expr) => { + if let Expr::PlVariable(names) = peel_parenthesized(expr) { + let rec = names.join(".").to_lowercase(); + if let Some(anchor) = self.record_cursors.get(&rec) { + if let Some(cols) = self.cursor_sources.get(anchor) { + for (position, column) in insert.columns.iter().enumerate() { + let col = cols.get(position).or(match cols.as_slice() { + [single] if single.output_name.is_empty() => Some(single), + _ => None, + }); + let source = col.and_then(|c| { + if !c.source_col.is_empty() { + Some(ColumnSource::Column { + table: c.source_table.clone(), + column: c.source_col.clone(), + }) + } else if c.source_table.is_some() { + // Catch-all (`SELECT *` cursor): attribute + // under the target column's own name. + Some(ColumnSource::Column { + table: c.source_table.clone(), + column: column.clone(), + }) + } else { + None + } + }); + if let Some(source) = source { + self.column_mappings.push(ColumnMapping { + target_table: Some(table_name.clone()), + target_column: column.clone(), + position: Some(position), + sources: vec![source], + kind: MappingKind::Direct, + expression: None, + }); + } + } + } + } + } + } +``` + +**Step 6: 运行确认通过** + +Run: `cargo test --features full whole_record_insert_expands_cursor_rowtype_fields` +Expected: PASS。 + +Run: `cargo test --features full --test regress_column_lineage whole_record_insert_values_r_resolves_through_cursor` +Expected: PASS。 + +**Step 7: 回归** + +Run: `cargo test --features full scalar_subquery` +Run: `cargo test --features full table_rowtype_record_field` +Run: `cargo test --features full star_cursor_rowtype_record` +Expected: PASS。 + +**Step 8: Commit** + +```bash +git add src/parser/extractor.rs tests/regress_column_lineage.rs +git commit -m "feat(lineage): 整记录 INSERT VALUES r 按游标源列位置展开 (fix #142 部分)" +``` + +--- + +## Task 5: 特征测试锁定已修复的游标 %ROWTYPE 形态 + +**Files:** +- Test: `tests/regress_column_lineage.rs`（仅新增，无实现改动） + +**Step 1: 写特征测试（当前已 PASS，锁定 #148 行为防回归）** + +```rust +/// #142 characteristic test: cursor-anchored %ROWTYPE record written via +/// `VALUES (r.id, r.amt)` resolves to the cursor's source columns (fixed by #148; +/// this locks the behavior so later extraction changes cannot regress it). +#[test] +fn cursor_rowtype_record_insert_values_resolves_to_cursor_source() { + let dir = TempDir::new().unwrap(); + let root = project_with_sql( + &dir, + r#" +CREATE TABLE t_src(id NUMBER, amt NUMBER); +CREATE TABLE t_dst(id NUMBER, amt NUMBER); +CREATE PROCEDURE p_copy_cursor AS + CURSOR cur IS SELECT id, amt FROM t_src; + r cur%ROWTYPE; +BEGIN + OPEN cur; + LOOP + FETCH cur INTO r; + EXIT WHEN cur%NOTFOUND; + INSERT INTO t_dst (id, amt) VALUES (r.id, r.amt); + END LOOP; + CLOSE cur; +END; +"#, + ); + let out = lineage(&root, "t_dst.amt", "upstream", "tree"); + assert!( + out.contains("t_src.amt"), + "cursor %ROWTYPE record field must resolve:\n{out}" + ); +} +``` + +**Step 2: 运行确认通过** + +Run: `cargo test --features full --test regress_column_lineage cursor_rowtype_record_insert_values_resolves_to_cursor_source` +Expected: PASS（当前行为基线）。 + +**Step 3: Commit** + +```bash +git add tests/regress_column_lineage.rs +git commit -m "test(lineage): 锁定游标 %ROWTYPE 记录字段写入的穿透行为 (fix #142)" +``` + +--- + +## Task 6: 全量门禁与收尾 + +**Step 1: fmt** + +Run: `cargo fmt --all -- --check` +Expected: PASS。若有格式问题：`cargo fmt` 后重跑。 + +**Step 2: clippy（full）** + +Run: `cargo clippy --features full -- -D warnings` +Expected: PASS（零警告）。 + +**Step 3: 全量测试（full，跳过环境相关）** + +Run: `cargo test --features full -- --skip test_path_mapping_applied --skip test_serve_` +Expected: PASS（含新增 4 个单测 + 5 个端到端；无与本改动相关的既有失败）。 + +**Step 4: 检查 diff 范围** + +Run: `git diff --stat HEAD` 与 `git status` +Expected: 仅 `src/parser/extractor.rs`、`tests/regress_column_lineage.rs`、本计划文档；无调试输出/草稿。 + +**Step 5: 汇报** + +按 AGENTS.md「每个 TDD 循环汇报」格式输出：每个任务测试的行为（测试函数名）、最小实现改的文件、是否重构及边界、实际执行的命令与结果。 + +--- + +## 已知局限（有意不处理，记录备查） + +1. **整记录写入无列清单**：`INSERT INTO t VALUES r`（无 `(a,b)`）无法命名目标列——需要目标表 DDL 列序，超出静态解析能力，维持现状（不产生列映射）。 +2. **整记录写入 + `SELECT *` 游标**：catch-all 无精确列名，位置归因无法证明与目标列清单一致——不猜测列名（不产生映射），避免重排列清单时静默错归因。 +3. **`Expr::ScalarSublink`（`expr OP ANY/ALL/SOME (subquery)`）作为值源**：谓词形态非值源，不处理。 +4. **自定义 TYPE `%ROWTYPE` 锚定**：按类型表归因（无 DDL 列序可循）；真实数据源优先由 FETCH 重绑到实际游标（review #153-3）。 + +## 审核修订（review #153） + +- 标量子查询首表达式复用 `classify_value_expr` 分类，变换（UPPER/NVL/CAST…）标 Derived 并保留表达式文本，字面量标 Direct（review #153-1）。 +- 通用 walker 不再递归子查询 select（`visit_expr` 对 Subquery/Exists/InSubquery/ScalarSublink 返回 `SkipChildren`），子查询内 JOIN/filter 不泄漏到外层语句分析（review #153-2）。 +- `%ROWTYPE` 记录字段作为子查询首表达式**已穿透**：ogsql-parser v0.10 将 `rec.field` 解析为 dotted `ColumnRef`，经 `record_cursors` 解析到游标/表列（原「→ Variable」局限描述有误，review #153-4）。 + +## 验收标准 + +- [ ] Task 1-4 各有失败→通过的测试（单测 + 端到端） +- [ ] Task 5 特征测试锁定 #148 已修复行为 +- [ ] 未删除/跳过/改写人类已有测试 +- [ ] `cargo fmt`、`cargo clippy --features full -- -D warnings` 干净 +- [ ] `cargo test --features full -- --skip test_path_mapping_applied --skip test_serve_` 全绿 +- [ ] 仅改动 `src/parser/extractor.rs`、`tests/regress_column_lineage.rs`；无 store 结构变化（不 bump `STORE_VERSION`） diff --git a/src/parser/extractor.rs b/src/parser/extractor.rs index 658e393..236aab2 100644 --- a/src/parser/extractor.rs +++ b/src/parser/extractor.rs @@ -2716,7 +2716,18 @@ impl Visitor for ColumnAccessExtractor { _ => String::new(), }; let vars: Vec = fetch.node.into.iter().map(expr_var_name).collect(); - self.record_fetch(&cursor_name, vars); + self.record_fetch(&cursor_name, vars.clone()); + // Review #3: the record's data comes from the FETCHing cursor, not + // its declared %ROWTYPE type anchor — rebind so `column_source` + // resolves through the cursor's SELECT sources. + if !cursor_name.is_empty() && vars.len() == 1 { + if let Some(var) = vars.first() { + let key = var.to_lowercase(); + if self.record_cursors.contains_key(&key) { + self.record_cursors.insert(key, cursor_name.to_lowercase()); + } + } + } } // `OPEN c_fxj FOR v_sql_txt` / `FOR EXECUTE expr`: resolve the dynamic SQL to // the cursor's SELECT sources so FETCH-variable chains keep resolving. @@ -2925,11 +2936,52 @@ impl Visitor for ColumnAccessExtractor { } } } - // DEFAULT VALUES has no sources; `SET` is handled as assignments; a - // record variable needs the variable's own type to expand. + // DEFAULT VALUES has no sources; `SET` is handled as assignments. ogsql_parser::ast::InsertSource::DefaultValues - | ogsql_parser::ast::InsertSource::Set(_) - | ogsql_parser::ast::InsertSource::RecordVariable(_) => {} + | ogsql_parser::ast::InsertSource::Set(_) => {} + // `INSERT INTO t (a, b) VALUES r` expands positionally through the + // record's %ROWTYPE cursor sources. Whole-record inserts cannot be + // aligned without the target DDL column order. + ogsql_parser::ast::InsertSource::RecordVariable(expr) => { + if let Expr::ColumnRef(names) | Expr::PlVariable(names) = + peel_parenthesized(expr) + { + let rec = names.join(".").to_lowercase(); + if let Some(anchor) = self.record_cursors.get(&rec) { + if let Some(cols) = self.cursor_sources.get(anchor) { + for (position, column) in insert.columns.iter().enumerate() { + let col = cols.get(position).or(match cols.as_slice() { + [single] if single.output_name.is_empty() => Some(single), + _ => None, + }); + let source = col.and_then(|c| { + if !c.source_col.is_empty() { + Some(ColumnSource::Column { + table: c.source_table.clone(), + column: c.source_col.clone(), + }) + } else { + // A catch-all (`SELECT *`) cursor has no + // exact column; guessing under the target + // name would misattribute reordered lists. + None + } + }); + if let Some(source) = source { + self.column_mappings.push(ColumnMapping { + target_table: Some(table_name.clone()), + target_column: column.clone(), + position: Some(position), + sources: vec![source], + kind: MappingKind::Direct, + expression: None, + }); + } + } + } + } + } + } } } else if let ogsql_parser::ast::InsertSource::Select(select) = &insert.source { // No column list: name the target columns from the SELECT output (the first @@ -3068,6 +3120,18 @@ impl Visitor for ColumnAccessExtractor { } } } + // Subqueries carry their own scope; the generic walker would otherwise + // recurse into their SELECT and leak its alias/join/filter state into + // this statement's analysis (review #153-2). Exists/Subquery have no + // left operand, so skipping is complete. + Expr::Subquery(_) | Expr::Exists(_) => return VisitorResult::SkipChildren, + // InSubquery/ScalarSublink DO have a left operand (`t.x > ANY (...)`, + // `t.id IN (...)`): collect its column references first, then skip the + // nested SELECT (review 5136742683). + Expr::InSubquery { expr, .. } | Expr::ScalarSublink { expr, .. } => { + self.walk_expr_for_column_refs(expr); + return VisitorResult::SkipChildren; + } _ => {} } VisitorResult::Continue @@ -3101,6 +3165,28 @@ impl ColumnAccessExtractor { }; } } + // #142: a single catch-all cursor source (empty output name — + // `SELECT *` cursor, or dynamic-SQL attribution) covers every + // record field: the exact column is unknown, attribute to the + // cursor's table under the field's own name (same philosophy as + // `resolve_cursor_flows`). + if let [single] = cols.as_slice() { + if single.output_name.is_empty() { + if let Some(ref t) = single.source_table { + return ColumnSource::Column { + table: Some(t.clone()), + column: column.clone(), + }; + } + } + } + } else { + // A table-anchored %ROWTYPE record has no cursor_sources entry; + // its fields are the anchor table's columns. + return ColumnSource::Column { + table: Some(cursor.clone()), + column: column.clone(), + }; } } } @@ -3287,6 +3373,12 @@ impl ColumnAccessExtractor { position: Option, value: &Expr, ) { + // A scalar subquery as a value carries its own FROM scope; resolve its first + // output expression there rather than against the enclosing statement. + if let Expr::Subquery(select) = peel_parenthesized(value) { + self.push_subquery_column_mapping(target_table, target_column, position, select); + return; + } let (sources, kind, expression) = self.classify_value_expr(value); self.column_mappings.push(ColumnMapping { target_table, @@ -3298,6 +3390,63 @@ impl ColumnAccessExtractor { }); } + /// Column mapping for `target = (SELECT ... FROM ...)`: map the written column + /// to the subquery's select expression — a scalar subquery's single output, or + /// — for a multi-column subquery shared by an UPDATE SET list — the output at + /// the written column's position. Resolves against the subquery's own FROM + /// aliases, then restores the enclosing statement's scope. Correlated + /// references (`s.id` in the subquery's WHERE) are not value sources and are + /// intentionally not collected. + fn push_subquery_column_mapping( + &mut self, + target_table: Option, + target_column: String, + position: Option, + select: &SelectStatement, + ) { + // The first select-list expression is the value; classify it under the + // subquery's own FROM scope, then restore the enclosing scope. Alias + // collection doubles as join/filter extraction, so snapshot the + // statement-level accumulators too — a subquery JOIN must not leak into + // the parent analysis. + let saved_alias_map = self.alias_map.clone(); + let saved_joins = self.join_conditions.len(); + let saved_filters = self.hard_filters.len(); + let saved_refs = self.column_refs.len(); + self.collect_aliases_from_table_refs(&select.from); + let new_scope = self.scope_sole_table_of(&select.from); + let saved_scope = std::mem::replace(&mut self.scope_sole_table, new_scope); + + let target = if select.targets.len() <= 1 { + select.targets.first() + } else { + position.and_then(|p| select.targets.get(p)) + }; + let (sources, kind, expression) = match target { + Some(SelectTarget::Expr(first, _)) => { + self.classify_value_expr(peel_parenthesized(first)) + } + // A `SELECT *` target cannot be resolved to a column list without a + // schema; leave the mapping without sources. + _ => (Vec::new(), MappingKind::Derived, None), + }; + + self.scope_sole_table = saved_scope; + self.alias_map = saved_alias_map; + self.join_conditions.truncate(saved_joins); + self.hard_filters.truncate(saved_filters); + self.column_refs.truncate(saved_refs); + + self.column_mappings.push(ColumnMapping { + target_table, + target_column, + position, + sources, + kind, + expression, + }); + } + /// Extract column mappings from a MERGE statement's WHEN clauses. fn visit_merge_statement(&mut self, merge: &ogsql_parser::ast::MergeStatement) { let target_name = match &merge.target { @@ -4928,6 +5077,22 @@ mod column_tests { .collect() } + /// Column mappings with a seeded procedure variable context (#142): lets a + /// standalone INSERT walk see cursor/record bindings that in real procedures + /// come from the DECLARE block. + fn column_mappings_of_with_context(sql: &str, ctx: &ProcedureVarContext) -> Vec { + let tokens = Tokenizer::new(sql).tokenize().unwrap(); + let mut parser = ogsql_parser::Parser::with_source(tokens, sql.to_string()); + let stmts = parser.parse_with_text(); + let mut result = Vec::new(); + for info in &stmts { + let mut extractor = ColumnAccessExtractor::new_with_context(ctx); + walk_statement(&mut extractor, &info.statement); + result.extend(extractor.finish().column_mappings); + } + result + } + /// Column mappings of a view body, via the explicit `CREATE VIEW` entry point. fn view_column_mappings(view: &str, declared: &[&str], select_sql: &str) -> Vec { let tokens = Tokenizer::new(select_sql).tokenize().unwrap(); @@ -5108,6 +5273,301 @@ mod column_tests { ); } + /// #142: a scalar subquery as an INSERT..SELECT target contributes the inner + /// select's FIRST expression as the source, resolved in the subquery's own FROM + /// scope. Correlated refs (`s.id` in WHERE) must NOT leak as sources. + #[test] + fn scalar_subquery_target_resolves_its_first_column() { + let maps = column_mappings_of( + "INSERT INTO t_out (id, code) \ + SELECT s.id, (SELECT r.code FROM t_ref r WHERE r.id = s.id) FROM t_src s", + ); + let m = find_mapping(&maps, "code"); + assert_eq!(m.kind, MappingKind::Direct); + assert_eq!(m.sources, vec![col(Some("t_ref"), "code")]); + } + + /// #142: the choke point is push_column_mapping, so INSERT..VALUES subqueries + /// resolve too. + #[test] + fn scalar_subquery_in_insert_values_resolves() { + let maps = column_mappings_of( + "INSERT INTO t_out (code) VALUES ((SELECT r.code FROM t_ref r WHERE r.id = 1))", + ); + assert_eq!( + find_mapping(&maps, "code").sources, + vec![col(Some("t_ref"), "code")] + ); + } + + /// Review #1: a scalar subquery whose first expression is a TRANSFORMED column + /// (`UPPER`, `+1`, `NVL`, `CAST`, …) must classify as Derived and keep the + /// expression text — not masquerade as a Direct copy. + #[test] + fn scalar_subquery_with_transformed_first_expr_is_derived() { + let maps = column_mappings_of( + "INSERT INTO t_out (code) \ + SELECT (SELECT UPPER(r.code) FROM t_ref r WHERE r.id = 1)", + ); + let m = find_mapping(&maps, "code"); + assert_eq!(m.kind, MappingKind::Derived); + assert_eq!(m.sources, vec![col(Some("t_ref"), "code")]); + assert!( + m.expression.as_deref().unwrap_or("").contains("UPPER"), + "expression text must survive: {:?}", + m.expression + ); + } + + /// Review #1: a literal-only scalar subquery is a constant → Direct + Literal + /// source, consistent with how `classify_value_expr` treats a bare literal. + #[test] + fn literal_only_scalar_subquery_classifies_direct() { + let maps = column_mappings_of("INSERT INTO t_out (code) VALUES ((SELECT 'x' FROM dual))"); + let m = find_mapping(&maps, "code"); + assert_eq!(m.kind, MappingKind::Direct); + assert_eq!( + m.sources, + vec![ColumnSource::Literal { + value: "'x'".to_string() + }] + ); + } + + /// Review (5136742683): a multi-column subquery applied to a multi-column + /// UPDATE SET target must align each target column to its OWN select-list + /// position — not copy the first expression into every column. + #[test] + fn multi_column_set_subquery_aligns_by_position() { + let maps = column_mappings_of("UPDATE u_dst SET (a, b) = (SELECT x, y FROM u_src)"); + assert_eq!( + find_mapping(&maps, "a").sources, + vec![col(Some("u_src"), "x")] + ); + assert_eq!( + find_mapping(&maps, "b").sources, + vec![col(Some("u_src"), "y")], + "each SET column must pair with its own subquery output, not copy the first" + ); + } + + /// Review #2: a JOIN inside the scalar subquery's FROM must not leak its + /// join/filter state into the enclosing statement's analysis — the subquery + /// carries its own scope. + #[test] + fn scalar_subquery_join_does_not_leak_into_parent_analysis() { + let analyses = extract_column_analysis( + "INSERT INTO t_out (code) \ + SELECT (SELECT r.code FROM t_ref r JOIN t_other o ON r.id = o.id WHERE r.id = 1)", + ); + assert_eq!(analyses.len(), 1); + let a = &analyses[0]; + assert!( + a.join_conditions.is_empty(), + "subquery JOIN must not leak into the parent analysis: {:?}", + a.join_conditions + ); + } + + /// Review (5136742683): `t.x > ANY (SELECT ...)` — the left operand `t.x` is a + /// real column reference of the enclosing query and must still be collected; + /// only the nested SELECT's own scope must be skipped. + #[test] + fn scalar_sublink_left_operand_column_is_collected() { + let analyses = + extract_column_analysis("SELECT * FROM t WHERE t.x > ANY (SELECT y FROM t2)"); + assert_eq!(analyses.len(), 1); + let x_refs: Vec<&ColumnRef> = analyses[0] + .column_refs + .iter() + .filter(|r| r.column == "x") + .collect(); + assert_eq!( + x_refs.len(), + 1, + "left operand of ANY/ALL must be collected, got: {:?}", + analyses[0].column_refs + ); + assert_eq!(x_refs[0].resolved_table.as_deref(), Some("t")); + } + + /// Review (5136742683): the left operand of `IN (SELECT ...)` in an ON clause + /// must still be collected (the generic walker was its only collector). + #[test] + fn in_subquery_left_operand_in_join_condition_is_collected() { + let analyses = + extract_column_analysis("SELECT * FROM t1 JOIN t2 ON t1.id IN (SELECT id FROM t3)"); + assert_eq!(analyses.len(), 1); + let id_refs: Vec<&ColumnRef> = analyses[0] + .column_refs + .iter() + .filter(|r| r.column == "id" && r.alias_prefix.as_deref() == Some("t1")) + .collect(); + assert_eq!( + id_refs.len(), + 1, + "t1.id left operand of IN-subquery must be collected, got: {:?}", + analyses[0].column_refs + ); + } + + /// #142: a `rec t%ROWTYPE` record (anchor is a TABLE, not a registered cursor) + /// resolves its fields to that table's columns. + #[test] + fn table_rowtype_record_field_resolves_to_table_column() { + let mut ctx = ProcedureVarContext::default(); + ctx.record_cursors + .insert("r".to_string(), "t_src".to_string()); + let maps = column_mappings_of_with_context( + "INSERT INTO t_dst (id, amt) VALUES (r.id, r.amt)", + &ctx, + ); + assert_eq!( + find_mapping(&maps, "id").sources, + vec![col(Some("t_src"), "id")] + ); + assert_eq!( + find_mapping(&maps, "amt").sources, + vec![col(Some("t_src"), "amt")] + ); + } + + /// #142: a `SELECT *` cursor produces a single catch-all cursor source (empty + /// output name, table attributed). Record fields over it attribute to the + /// cursor's table under the field's own name. + #[test] + fn star_cursor_rowtype_record_field_attributes_to_cursor_table() { + let mut ctx = ProcedureVarContext::default(); + ctx.cursor_sources.insert( + "cur".to_string(), + vec![CursorColumn { + output_name: String::new(), + source_table: Some("t_src".to_string()), + source_col: String::new(), + }], + ); + ctx.record_cursors + .insert("r".to_string(), "cur".to_string()); + let maps = column_mappings_of_with_context( + "INSERT INTO t_dst (id, amt) VALUES (r.id, r.amt)", + &ctx, + ); + assert_eq!( + find_mapping(&maps, "id").sources, + vec![col(Some("t_src"), "id")] + ); + assert_eq!( + find_mapping(&maps, "amt").sources, + vec![col(Some("t_src"), "amt")] + ); + } + + /// #142: `INSERT INTO t (a, b) VALUES r` with a cursor-anchored %ROWTYPE record + /// expands the record's fields positionally through the cursor's SELECT sources. + #[test] + fn whole_record_insert_expands_cursor_rowtype_fields() { + let mut ctx = ProcedureVarContext::default(); + ctx.cursor_sources.insert( + "cur".to_string(), + vec![ + CursorColumn { + output_name: "id".to_string(), + source_table: Some("t_src".to_string()), + source_col: "id".to_string(), + }, + CursorColumn { + output_name: "amt".to_string(), + source_table: Some("t_src".to_string()), + source_col: "amt".to_string(), + }, + ], + ); + ctx.record_cursors + .insert("r".to_string(), "cur".to_string()); + let maps = column_mappings_of_with_context("INSERT INTO t_dst (id, amt) VALUES r", &ctx); + assert_eq!( + find_mapping(&maps, "id").sources, + vec![col(Some("t_src"), "id")] + ); + assert_eq!( + find_mapping(&maps, "amt").sources, + vec![col(Some("t_src"), "amt")] + ); + } + + /// Review #5: whole-record insert from a `SELECT *` cursor has no exact column + /// names — attributing each INSERT column under its own name would silently + /// misattribute a reordered column list. Leave such mappings unmapped instead. + #[test] + fn whole_record_insert_from_star_cursor_yields_no_guessed_mappings() { + let mut ctx = ProcedureVarContext::default(); + ctx.cursor_sources.insert( + "cur".to_string(), + vec![CursorColumn { + output_name: String::new(), + source_table: Some("t_src".to_string()), + source_col: String::new(), + }], + ); + ctx.record_cursors + .insert("r".to_string(), "cur".to_string()); + let maps = column_mappings_of_with_context("INSERT INTO t_dst (id, amt) VALUES r", &ctx); + assert!( + maps.is_empty(), + "a SELECT * catch-all must not fabricate column names: {maps:#?}" + ); + } + + /// Review #3: a `%ROWTYPE` record's data comes from the FETCH that fills it. + /// `r t_type%ROWTYPE` + `FETCH cur INTO r` (cur reads t_other) must resolve + /// `r.id` to t_other.id, not the declared type table. + #[test] + fn fetch_rebinds_rowtype_record_to_the_fetching_cursor() { + let maps = column_mappings_of( + "CREATE OR REPLACE PROCEDURE p AS\n\ + \x20 r t_type%ROWTYPE;\n\ + \x20 CURSOR cur IS SELECT id, amt FROM t_other;\n\ + BEGIN\n\ + \x20 OPEN cur;\n\ + \x20 FETCH cur INTO r;\n\ + \x20 INSERT INTO t_dst (id, amt) VALUES (r.id, r.amt);\n\ + END", + ); + assert_eq!( + find_mapping(&maps, "id").sources, + vec![col(Some("t_other"), "id")], + "record field must resolve to the FETCHing cursor's source, not the type table" + ); + } + + /// Review #4: a %ROWTYPE record field as a scalar subquery's first expression + /// penetrates through record_cursors to the cursor's source column — ogsql-parser + /// parses `rec.field` as a dotted ColumnRef, which column_source already resolves. + #[test] + fn record_field_in_scalar_subquery_resolves_to_column() { + let mut ctx = ProcedureVarContext::default(); + ctx.cursor_sources.insert( + "cur".to_string(), + vec![CursorColumn { + output_name: "CLIENT_ACNT_ID".to_string(), + source_table: Some("v_src".to_string()), + source_col: "CLIENT_ACNT_ID".to_string(), + }], + ); + ctx.record_cursors + .insert("v_fund_acnt_all".to_string(), "cur".to_string()); + let maps = column_mappings_of_with_context( + "INSERT INTO v_dst (acnt) \ + SELECT (SELECT v_fund_acnt_all.CLIENT_ACNT_ID FROM dual) FROM dual", + &ctx, + ); + assert_eq!( + find_mapping(&maps, "acnt").sources, + vec![col(Some("v_src"), "CLIENT_ACNT_ID")], + "record field in a scalar subquery must resolve to the cursor's column" + ); + } + /// Every union branch feeds the same target column. This needs both the extractor's /// set-operation walk and the parser's chain fix (c2j/ogsql-parser#318). #[test] diff --git a/tests/regress_column_lineage.rs b/tests/regress_column_lineage.rs index 23db8a6..af8dd5c 100644 --- a/tests/regress_column_lineage.rs +++ b/tests/regress_column_lineage.rs @@ -355,6 +355,238 @@ END; ); } +/// #142: a scalar subquery in the INSERT..SELECT target list must resolve to the +/// subquery's source column, not report "No column lineage". +#[test] +fn scalar_subquery_in_insert_select_target_resolves() { + let dir = TempDir::new().unwrap(); + let root = project_with_sql( + &dir, + r#" +CREATE TABLE t_src(id NUMBER, amt NUMBER); +CREATE TABLE t_ref(id NUMBER, code VARCHAR2(10)); +CREATE TABLE t_out(id NUMBER, code VARCHAR2(10)); +CREATE PROCEDURE p_copy_subquery AS BEGIN + INSERT INTO t_out (id, code) + SELECT s.id, (SELECT r.code FROM t_ref r WHERE r.id = s.id) FROM t_src s; +END; +"#, + ); + let out = lineage(&root, "t_out.code", "upstream", "tree"); + assert!( + !out.contains("No column lineage"), + "scalar subquery target must resolve:\n{out}" + ); + assert!( + out.contains("t_ref.code"), + "subquery source column missing:\n{out}" + ); +} + +/// #142: a table-anchored %ROWTYPE record (`r t_src%ROWTYPE`) written via +/// `VALUES (r.id, r.amt)` must resolve to t_src columns, not "?.id". +#[test] +fn table_rowtype_record_insert_values_resolves_to_table() { + let dir = TempDir::new().unwrap(); + let root = project_with_sql( + &dir, + r#" +CREATE TABLE t_src(id NUMBER, amt NUMBER); +CREATE TABLE t_dst(id NUMBER, amt NUMBER); +CREATE PROCEDURE p_table_rowtype AS + r t_src%ROWTYPE; + CURSOR cur IS SELECT id, amt FROM t_src; +BEGIN + OPEN cur; + LOOP + FETCH cur INTO r; + EXIT WHEN cur%NOTFOUND; + INSERT INTO t_dst (id, amt) VALUES (r.id, r.amt); + END LOOP; + CLOSE cur; +END; +"#, + ); + let out = lineage(&root, "t_dst.id", "upstream", "tree"); + assert!( + out.contains("t_src.id"), + "table-anchored record field must resolve:\n{out}" + ); + assert!( + !out.contains("?.id"), + "table-anchored record field must not stay unattributed:\n{out}" + ); +} + +/// #142: `SELECT *` cursor + `%ROWTYPE` record fields must resolve to the +/// cursor's table (columns attributed under the field names), not "?.id". +#[test] +fn star_cursor_rowtype_record_resolves_to_cursor_table() { + let dir = TempDir::new().unwrap(); + let root = project_with_sql( + &dir, + r#" +CREATE TABLE t_src(id NUMBER, amt NUMBER); +CREATE TABLE t_dst(id NUMBER, amt NUMBER); +CREATE PROCEDURE p_star_cursor AS + CURSOR cur IS SELECT * FROM t_src; + r cur%ROWTYPE; +BEGIN + OPEN cur; + LOOP + FETCH cur INTO r; + EXIT WHEN cur%NOTFOUND; + INSERT INTO t_dst (id, amt) VALUES (r.id, r.amt); + END LOOP; + CLOSE cur; +END; +"#, + ); + let out = lineage(&root, "t_dst.id", "upstream", "tree"); + assert!( + out.contains("t_src.id"), + "star-cursor record field must resolve:\n{out}" + ); + assert!( + !out.contains("?.id"), + "star-cursor record field must not stay unattributed:\n{out}" + ); +} + +/// #142: whole-record insert `INSERT INTO t_dst (id, amt) VALUES r` (cursor-anchored +/// %ROWTYPE) must resolve positionally through the cursor's sources. +#[test] +fn whole_record_insert_values_r_resolves_through_cursor() { + let dir = TempDir::new().unwrap(); + let root = project_with_sql( + &dir, + r#" +CREATE TABLE t_src(id NUMBER, amt NUMBER); +CREATE TABLE t_dst(id NUMBER, amt NUMBER); +CREATE PROCEDURE p_rec_insert AS + CURSOR cur IS SELECT id, amt FROM t_src; + r cur%ROWTYPE; +BEGIN + OPEN cur; + LOOP + FETCH cur INTO r; + EXIT WHEN cur%NOTFOUND; + INSERT INTO t_dst (id, amt) VALUES r; + END LOOP; + CLOSE cur; +END; +"#, + ); + let out = lineage(&root, "t_dst.amt", "upstream", "tree"); + assert!( + out.contains("t_src.amt"), + "whole-record insert must resolve through the cursor:\n{out}" + ); +} + +/// Review #5: whole-record insert over a `SELECT *` cursor with a REORDERED +/// column list must not silently misattribute — leave unmapped rather than +/// guess names. +#[test] +fn star_cursor_whole_record_insert_does_not_misattribute_reordered_columns() { + let dir = TempDir::new().unwrap(); + let root = project_with_sql( + &dir, + r#" +CREATE TABLE t_src(id NUMBER, amt NUMBER); +CREATE TABLE t_dst(amt NUMBER, id NUMBER); +CREATE PROCEDURE p_rec_reorder AS + CURSOR cur IS SELECT * FROM t_src; + r cur%ROWTYPE; +BEGIN + OPEN cur; + LOOP + FETCH cur INTO r; + EXIT WHEN cur%NOTFOUND; + INSERT INTO t_dst (amt, id) VALUES r; + END LOOP; + CLOSE cur; +END; +"#, + ); + // Positionally t_src.id (cursor col 0) fills t_dst.amt, but the SELECT * + // catch-all cannot prove that — emitting `t_dst.amt ← t_src.amt` would be a + // silent lie. Unmapped is correct. + let out = lineage(&root, "t_dst.amt", "upstream", "tree"); + assert!( + !out.contains("t_src.amt"), + "reordered whole-record insert must not fabricate a name match:\n{out}" + ); +} + +/// Review #3: FETCH fills the record, so `r t_type%ROWTYPE` + `FETCH cur INTO r` +/// (cur reads t_other) must resolve to t_other, not the declared type table. +#[test] +fn fetch_rebinding_overrides_rowtype_type_table() { + let dir = TempDir::new().unwrap(); + let root = project_with_sql( + &dir, + r#" +CREATE TABLE t_other(id NUMBER, amt NUMBER); +CREATE TABLE t_type(id NUMBER, amt NUMBER); +CREATE TABLE t_dst(id NUMBER, amt NUMBER); +CREATE PROCEDURE p_fetch_mismatch AS + r t_type%ROWTYPE; + CURSOR cur IS SELECT id, amt FROM t_other; +BEGIN + OPEN cur; + LOOP + FETCH cur INTO r; + EXIT WHEN cur%NOTFOUND; + INSERT INTO t_dst (id, amt) VALUES (r.id, r.amt); + END LOOP; + CLOSE cur; +END; +"#, + ); + let out = lineage(&root, "t_dst.id", "upstream", "tree"); + assert!( + out.contains("t_other.id"), + "record filled by FETCH must resolve to the cursor's source:\n{out}" + ); + assert!( + !out.contains("t_type.id"), + "declared %ROWTYPE type table must not be the data source:\n{out}" + ); +} + +/// #142 characteristic test: cursor-anchored %ROWTYPE record written via +/// `VALUES (r.id, r.amt)` resolves to the cursor's source columns (fixed by #148; +/// this locks the behavior so later extraction changes cannot regress it). +#[test] +fn cursor_rowtype_record_insert_values_resolves_to_cursor_source() { + let dir = TempDir::new().unwrap(); + let root = project_with_sql( + &dir, + r#" +CREATE TABLE t_src(id NUMBER, amt NUMBER); +CREATE TABLE t_dst(id NUMBER, amt NUMBER); +CREATE PROCEDURE p_copy_cursor AS + CURSOR cur IS SELECT id, amt FROM t_src; + r cur%ROWTYPE; +BEGIN + OPEN cur; + LOOP + FETCH cur INTO r; + EXIT WHEN cur%NOTFOUND; + INSERT INTO t_dst (id, amt) VALUES (r.id, r.amt); + END LOOP; + CLOSE cur; +END; +"#, + ); + let out = lineage(&root, "t_dst.amt", "upstream", "tree"); + assert!( + out.contains("t_src.amt"), + "cursor %ROWTYPE record field must resolve:\n{out}" + ); +} + /// Regression: a cursor declared with `SELECT *` resolves to zero source columns, so a /// later `FETCH` used to panic in `resolve_cursor_flows` — `bool::then_some` evaluates /// its argument eagerly, indexing `&cols[0]` on the empty list