[CALCITE-7405] Pre-process expressions for correlations before building projection in SqlToRelConverter

core/src/main/java/org/apache/calcite/sql2rel/SqlToRelConverter.java

@@ -3774,9 +3774,10 @@ private void createAggImpl(Blackboard bb,
       final RelNode inputRel = bb.root();
 
       // Project the expressions required by agg and having.
-      RelNode intermediateProject = relBuilder.push(inputRel)
-          .projectNamed(preExprs.leftList(), preExprs.rightList(), false)
-          .build();
+      // Using LogicalProject.create to avoid bloat optimizations in RelBuilder.
+      RelNode intermediateProject =
+          LogicalProject.create(inputRel, ImmutableList.of(), preExprs.leftList(),
+              preExprs.rightList(), ImmutableSet.of());
       final RelNode r2;
       // deal with correlation
       final CorrelationUse p = getCorrelationUse(bb, intermediateProject);
@@ -3790,7 +3791,9 @@ private void createAggImpl(Blackboard bb,
                 true, ImmutableSet.of(p.id))
             .build();
       } else {
-        r2 = intermediateProject;
+        r2 = relBuilder.push(inputRel)
+                  .projectNamed(preExprs.leftList(), preExprs.rightList(), false)
+                  .build();
       }
       bb.setRoot(r2, false);
       bb.mapRootRelToFieldProjection.put(bb.root(), r.groupExprProjection);
@@ -5025,10 +5028,10 @@ private void convertNonAggregateSelectList(
         SqlValidatorUtil.uniquify(fieldNames,
             catalogReader.nameMatcher().isCaseSensitive());
 
-    relBuilder.push(bb.root())
-        .projectNamed(exprs, uniqueFieldNames, true);
-
-    RelNode project = relBuilder.build();
+    // Using LogicalProject.create to avoid bloat optimizations in RelBuilder.
+    RelNode project =
+        LogicalProject.create(bb.root(), ImmutableList.of(), exprs, uniqueFieldNames,
+            ImmutableSet.of());
 
     final RelNode r;
     final CorrelationUse p = getCorrelationUse(bb, project);
@@ -5042,7 +5045,7 @@ private void convertNonAggregateSelectList(
               ImmutableSet.of(p.id))
           .build();
     } else {
-      r = project;
+      r = relBuilder.push(bb.root()).projectNamed(exprs, uniqueFieldNames, true).build();
     }
 
     bb.setRoot(r, false);

core/src/test/java/org/apache/calcite/test/SqlToRelConverterTest.java

@@ -4209,6 +4209,41 @@ void checkCorrelatedMapSubQuery(boolean expand) {
     sql(sql).withExpand(false).withDecorrelate(false).ok();
   }
 
+  /** Test case for
+   * <a href="https://issues.apache.org/jira/browse/CALCITE-7405">[CALCITE-7405]
+   * Correlated subquery where outer tree is affected by bloat optimization
+   * causing project fusion</a>. */
+  @Test void testCorrelationInProjectionWithBloatFusion() {
+    final String sql = "select e1.sal + "
+        + "(select sum(e2.sal) from emp e2 where e2.deptno = d.deptno)\n"
+        + "from dept d join emp e1 on  d.deptno + 1 = e1.deptno";
+    sql(sql)
+        .withExpand(false)
+        .withDecorrelate(false)
+        .withConfig(srcc ->
+            srcc.addRelBuilderConfigTransform(rbc ->
+                rbc.withBloat(100).withPushJoinCondition(true)))
+        .ok();
+  }
+
+  /** Test case for
+   * <a href="https://issues.apache.org/jira/browse/CALCITE-7405">[CALCITE-7405]
+   * Correlated subquery using a compound expression from input projection that is not ultimately
+   * projected by query, and where outer tree is affected by bloat optimization causing project
+   * fusion and elimination of compound expression required by correlation</a>. */
+  @Test void testCorrelationInProjectionWithBloatFusionAndCompoundNonProjectedCorrelation() {
+    final String sql = "select d2.name, "
+        + "(select sum(e.sal) from emp e where e.deptno = d2.compound)\n"
+        + "from (select d.name, d.deptno + 1 as compound from dept d) as d2";
+    sql(sql)
+        .withExpand(false)
+        .withDecorrelate(false)
+        .withConfig(srcc ->
+            srcc.addRelBuilderConfigTransform(rbc ->
+                rbc.withBloat(100)))
+        .ok();
+  }
+
   @Test void testCustomColumnResolving() {
     final String sql = "select k0 from struct.t";
     sql(sql).ok();

core/src/test/resources/org/apache/calcite/test/SqlToRelConverterTest.xml

@@ -1333,6 +1333,45 @@ LogicalAggregate(group=[{}], EXPR$0=[SUM($0)])
       LogicalTableScan(table=[[CATALOG, SALES, EMP]])
 })])
   LogicalTableScan(table=[[CATALOG, SALES, EMP]])
+]]>
+    </Resource>
+  </TestCase>
+  <TestCase name="testCorrelationInProjectionWithBloatFusion">
+    <Resource name="sql">
+      <![CDATA[select e1.sal + (select sum(e2.sal) from emp e2 where e2.deptno = d.deptno)
+from dept d join emp e1 on  d.deptno + 1 = e1.deptno]]>
+    </Resource>
+    <Resource name="plan">
+      <![CDATA[
+LogicalProject(variablesSet=[[$cor0]], EXPR$0=[+($7, $SCALAR_QUERY({
+LogicalAggregate(group=[{}], EXPR$0=[SUM($0)])
+  LogicalProject(SAL=[$5])
+    LogicalFilter(condition=[=($7, $cor0.DEPTNO)])
+      LogicalTableScan(table=[[CATALOG, SALES, EMP]])
+}))])
+  LogicalProject(DEPTNO=[$0], NAME=[$1], EMPNO=[$3], ENAME=[$4], JOB=[$5], MGR=[$6], HIREDATE=[$7], SAL=[$8], COMM=[$9], DEPTNO0=[$10], SLACKER=[$11])
+    LogicalJoin(condition=[=($2, $10)], joinType=[inner])
+      LogicalProject(DEPTNO=[$0], NAME=[$1], $f2=[+($0, 1)])
+        LogicalTableScan(table=[[CATALOG, SALES, DEPT]])
+      LogicalTableScan(table=[[CATALOG, SALES, EMP]])
+]]>
+    </Resource>
+  </TestCase>
+  <TestCase name="testCorrelationInProjectionWithBloatFusionAndCompoundNonProjectedCorrelation">
+    <Resource name="sql">
+      <![CDATA[select d2.name, (select sum(e.sal) from emp e where e.deptno = d2.compound)
+from (select d.name, d.deptno + 1 as compound from dept d) as d2]]>
+    </Resource>
+    <Resource name="plan">
+      <![CDATA[
+LogicalProject(variablesSet=[[$cor0]], NAME=[$0], EXPR$1=[$SCALAR_QUERY({
+LogicalAggregate(group=[{}], EXPR$0=[SUM($0)])
+  LogicalProject(SAL=[$5])
+    LogicalFilter(condition=[=($7, $cor0.COMPOUND)])
+      LogicalTableScan(table=[[CATALOG, SALES, EMP]])
+})])
+  LogicalProject(NAME=[$1], COMPOUND=[+($0, 1)])
+    LogicalTableScan(table=[[CATALOG, SALES, DEPT]])
 ]]>
     </Resource>
   </TestCase>