[CALCITE-7356] The MARK JOIN generated by TopDownGeneralDecorrelator needs to be adapted to RelFieldTrimmer

Author: xiedeyantu

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

@@ -957,17 +957,25 @@ public TrimResult trimFields(
     switch (join.getJoinType()) {
     case SEMI:
     case ANTI:
-      // For SemiJoins and AntiJoins only map fields from the left-side
+    case LEFT_MARK:
+      // For SemiJoins, AntiJoins and LeftMarkJoins only map fields from the left-side.
+      // For LeftMarkJoins, the mark column is also mapped.
       if (join.getJoinType() == JoinRelType.SEMI) {
         relBuilder.semiJoin(newConditionExpr);
-      } else {
+      } else if (join.getJoinType() == JoinRelType.ANTI) {
         relBuilder.antiJoin(newConditionExpr);
+      } else {
+        relBuilder.join(join.getJoinType(), newConditionExpr, join.getVariablesSet());
       }
       Mapping inputMapping = inputMappings.get(0);
+      int targetCount = newSystemFieldCount + inputMapping.getTargetCount();
+      if (join.getJoinType() == JoinRelType.LEFT_MARK) {
+        targetCount++;
+      }
       mapping =
           Mappings.create(MappingType.INVERSE_SURJECTION,
               join.getRowType().getFieldCount(),
-              newSystemFieldCount + inputMapping.getTargetCount());
+              targetCount);
       for (int i = 0; i < newSystemFieldCount; ++i) {
         mapping.set(i, i);
       }
@@ -976,6 +984,9 @@ public TrimResult trimFields(
       for (IntPair pair : inputMapping) {
         mapping.set(pair.source + offset, pair.target + newOffset);
       }
+      if (join.getJoinType() == JoinRelType.LEFT_MARK) {
+        mapping.set(join.getRowType().getFieldCount() - 1, targetCount - 1);
+      }
       break;
     case ASOF:
     case LEFT_ASOF:

core/src/test/resources/sql/new-decorr.iq

@@ -42,4 +42,58 @@ SELECT * FROM t0 WHERE t0a <
 
 !ok
 
+# [CALCITE-7356] The MARK JOIN generated by TopDownGeneralDecorrelator needs to be adapted to RelFieldTrimmer
+!use blank
+CREATE TABLE emps (
+  empid INTEGER NOT NULL,
+  deptno INTEGER NOT NULL,
+  name VARCHAR(10) NOT NULL,
+  salary DECIMAL(10, 2) NOT NULL,
+  commission INTEGER);
+(0 rows modified)
+
+!update
+
+INSERT INTO emps (empid, deptno, name, salary, commission) VALUES
+(100, 10, 'Bill', 10000.00, 1000),
+(200, 20, 'Eric', 8000.00, 500),
+(150, 10, 'Sebastian', 7000.00, NULL),
+(110, 10, 'Theodore', 11500.00, 250),
+(170, 30, 'Theodore', 11500.00, 250),
+(140, 10, 'Sebastian', 7000.00, NULL);
+(6 rows modified)
+
+!update
+
+SELECT *, EXISTS(select * from (
+  SELECT e2.deptno FROM emps e2 where e1.commission = e2.commission) as table3
+  where table3.deptno <> e1.deptno)
+from emps e1 order by empid;
++-------+--------+-----------+----------+------------+--------+
+| EMPID | DEPTNO | NAME      | SALARY   | COMMISSION | EXPR$5 |
++-------+--------+-----------+----------+------------+--------+
+|   100 |     10 | Bill      | 10000.00 |       1000 | false  |
+|   110 |     10 | Theodore  | 11500.00 |        250 | true   |
+|   140 |     10 | Sebastian |  7000.00 |            | false  |
+|   150 |     10 | Sebastian |  7000.00 |            | false  |
+|   170 |     30 | Theodore  | 11500.00 |        250 | true   |
+|   200 |     20 | Eric      |  8000.00 |        500 | false  |
++-------+--------+-----------+----------+------------+--------+
+(6 rows)
+
+!ok
+
+!if (use_new_decorr) {
+EnumerableSort(sort0=[$0], dir0=[ASC])
+  EnumerableHashJoin(condition=[AND(IS NOT DISTINCT FROM($1, $6), IS NOT DISTINCT FROM($4, $7))], joinType=[left_mark])
+    EnumerableTableScan(table=[[BLANK, EMPS]])
+    EnumerableCalc(expr#0..3=[{inputs}], DEPTNO=[$t2], DEPTNO0=[$t0], COMMISSION0=[$t1])
+      EnumerableHashJoin(condition=[AND(=($1, $3), <>($2, $0))], joinType=[inner])
+        EnumerableAggregate(group=[{1, 4}])
+          EnumerableTableScan(table=[[BLANK, EMPS]])
+        EnumerableCalc(expr#0..4=[{inputs}], DEPTNO=[$t1], COMMISSION=[$t4])
+          EnumerableTableScan(table=[[BLANK, EMPS]])
+!plan
+!}
+
 # End new-decorr.iq