[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

@@ -836,6 +836,11 @@ public TrimResult trimFields(
       Join join,
       ImmutableBitSet fieldsUsed,
       Set<RelDataTypeField> extraFields) {
+    // If the column "mark" is included, it needs to be excluded first.
+    if (join.getJoinType() == JoinRelType.LEFT_MARK) {
+      int markIndex = join.getRowType().getFieldCount() - 1;
+      fieldsUsed = fieldsUsed.except(ImmutableBitSet.of(markIndex));
+    }
     final int fieldCount = join.getSystemFieldList().size()
         + join.getLeft().getRowType().getFieldCount()
         + join.getRight().getRowType().getFieldCount();
@@ -957,17 +962,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 +989,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,60 @@ 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 empid, 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 | EXPR$1 |
++-------+--------+
+|   100 | false  |
+|   110 | true   |
+|   140 | false  |
+|   150 | false  |
+|   170 | true   |
+|   200 | false  |
++-------+--------+
+(6 rows)
+
+!ok
+
+!if (use_new_decorr) {
+EnumerableSort(sort0=[$0], dir0=[ASC])
+  EnumerableCalc(expr#0..3=[{inputs}], EMPID=[$t0], EXPR$1=[$t3])
+    EnumerableHashJoin(condition=[AND(IS NOT DISTINCT FROM($1, $3), IS NOT DISTINCT FROM($2, $4))], joinType=[left_mark])
+      EnumerableCalc(expr#0..4=[{inputs}], proj#0..1=[{exprs}], COMMISSION=[$t4])
+        EnumerableTableScan(table=[[BLANK, EMPS]])
+      EnumerableCalc(expr#0..3=[{inputs}], proj#0..1=[{exprs}])
+        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