fix: rewrite denormalized node aliases to edge aliases in WITH CTE body

When a denormalized schema embeds node properties in the edge table,
the FROM clause aliases the table with the edge alias (e.g., `flights AS r`).
Property access on denormalized nodes (e.g., `a.city`) must be rewritten
to use the edge alias (`r.OriginCityName`). The property name mapping was
already handled by `rewrite_expression_with_property_mapping`, but the
table alias was not being rewritten in the WITH CTE rendering path.

Add `rewrite_denormalized_aliases_in_expr()` that uses
`get_properties_with_table_alias()` to detect denormalized nodes and
rewrite their aliases in SELECT items and GROUP BY expressions during
CTE body construction.

Fixes 3 denormalized WITH aggregation integration tests.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: genezhang

src/render_plan/plan_builder_utils.rs

@@ -8284,6 +8284,43 @@ pub(crate) fn build_chained_with_match_cte_plan(
                         // Performance optimization: Wrap non-ID columns with ANY() when aggregating
                         // This allows GROUP BY to only include ID column (more efficient)
 
+                        // Rewrite denormalized node aliases to edge aliases in RenderExpr.
+                        // For denormalized schemas (e.g., Airport in flights table),
+                        // PropertyAccessExp(a, OriginCityName) must become
+                        // PropertyAccessExp(r, OriginCityName) because FROM is `flights AS r`.
+                        fn rewrite_denormalized_aliases_in_expr(expr: &mut RenderExpr, plan: &LogicalPlan) {
+                            match expr {
+                                RenderExpr::PropertyAccessExp(prop) => {
+                                    if let Ok((_, Some(edge_alias))) = plan.get_properties_with_table_alias(&prop.table_alias.0) {
+                                        if edge_alias != prop.table_alias.0 {
+                                            log::info!(
+                                                "🔧 Denormalized alias rewrite in WITH: '{}.{}' → '{}.{}'",
+                                                prop.table_alias.0, prop.column.raw(),
+                                                edge_alias, prop.column.raw()
+                                            );
+                                            prop.table_alias = crate::render_plan::render_expr::TableAlias(edge_alias);
+                                        }
+                                    }
+                                }
+                                RenderExpr::AggregateFnCall(agg) => {
+                                    for arg in &mut agg.args {
+                                        rewrite_denormalized_aliases_in_expr(arg, plan);
+                                    }
+                                }
+                                RenderExpr::ScalarFnCall(f) => {
+                                    for arg in &mut f.args {
+                                        rewrite_denormalized_aliases_in_expr(arg, plan);
+                                    }
+                                }
+                                RenderExpr::OperatorApplicationExp(op) => {
+                                    for operand in &mut op.operands {
+                                        rewrite_denormalized_aliases_in_expr(operand, plan);
+                                    }
+                                }
+                                _ => {}
+                            }
+                        }
+
                         // Extract UNWIND alias from plan if present — UNWIND aliases are simple
                         // ARRAY JOIN column references, not table aliases to expand.
                         // Must recurse through wrapping nodes (Filter, Projection, etc.)
@@ -8438,6 +8475,9 @@ pub(crate) fn build_chained_with_match_cte_plan(
 
                                                 let expr_result: Result<RenderExpr, _> = expanded_expr.try_into();
                                                 expr_result.ok().map(|mut expr| {
+                                                    // Rewrite denormalized node aliases (e.g., a → r)
+                                                    rewrite_denormalized_aliases_in_expr(&mut expr, plan_to_render);
+
                                                     // 🔧 FIX: VLP CTE column rewriting for non-TableAlias WITH items
                                                     // When FROM is a VLP/multi-type CTE, PropertyAccess references
                                                     // (e.g., message.content) must be rewritten to CTE columns
@@ -8740,7 +8780,10 @@ pub(crate) fn build_chained_with_match_cte_plan(
                                                             ExpressionRewriteContext::new(plan_to_render)
                                                         };
                                                         let rewritten = rewrite_expression_with_property_mapping(&item.expression, &rewrite_ctx);
-                                                        let expr_vec: Vec<RenderExpr> = rewritten.try_into().ok().into_iter().collect();
+                                                        let expr_vec: Vec<RenderExpr> = rewritten.try_into().ok().map(|mut expr: RenderExpr| {
+                                                            rewrite_denormalized_aliases_in_expr(&mut expr, plan_to_render);
+                                                            expr
+                                                        }).into_iter().collect();
                                                         expr_vec
                                                     }
                                                 }

tests/integration/query_patterns/test_pattern_matrix.py

@@ -817,7 +817,7 @@ def test_group_by_2(self):
         result = execute_query(query, "ontime_flights")
         assert "error" not in result, f"Query failed: {result}"
 
-    @pytest.mark.xfail(reason="Denormalized schema: node property resolution or VLP CTE gap")
+    @pytest.mark.xfail(reason="Denormalized schema: 'airport' property not in schema mapping")
     def test_with_agg_0(self):
         """
         WITH clause aggregation
@@ -827,7 +827,6 @@ def test_with_agg_0(self):
         result = execute_query(query, "ontime_flights")
         assert "error" not in result, f"Query failed: {result}"
 
-    @pytest.mark.xfail(reason="Denormalized schema: node property resolution or VLP CTE gap")
     def test_with_agg_1(self):
         """
         WITH clause aggregation
@@ -837,7 +836,7 @@ def test_with_agg_1(self):
         result = execute_query(query, "ontime_flights")
         assert "error" not in result, f"Query failed: {result}"
 
-    @pytest.mark.xfail(reason="Denormalized schema: node property resolution or VLP CTE gap")
+    @pytest.mark.xfail(reason="Denormalized schema: 'airport' property not in schema mapping")
     def test_with_agg_2(self):
         """
         WITH clause aggregation

tests/integration/query_patterns/test_pattern_schema_matrix.py

@@ -836,7 +836,6 @@ def test_group_by_2(self):
         result = execute_query(query, "ontime_flights")
         assert "error" not in result, f"Query failed: {result}"
 
-    @pytest.mark.xfail(reason="Bug: WITH aggregation generates incorrect SQL")
     def test_with_agg_0(self):
         """
         WITH clause aggregation
@@ -846,7 +845,6 @@ def test_with_agg_0(self):
         result = execute_query(query, "ontime_flights")
         assert "error" not in result, f"Query failed: {result}"
 
-    @pytest.mark.xfail(reason="Bug: WITH aggregation generates incorrect SQL")
     def test_with_agg_1(self):
         """
         WITH clause aggregation
@@ -856,7 +854,7 @@ def test_with_agg_1(self):
         result = execute_query(query, "ontime_flights")
         assert "error" not in result, f"Query failed: {result}"
 
-    @pytest.mark.xfail(reason="Bug: WITH aggregation generates incorrect SQL")
+    @pytest.mark.xfail(reason="Bug: denormalized node_id property not mapped to DB column in WITH")
     def test_with_agg_2(self):
         """
         WITH clause aggregation