fix(MXP-4112): EnumerableBatchNestedLoopJoin causes infinit plan optimization for single sourced JDBC queries

Author: kramerul

core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableBatchNestedLoopJoinRule.java

@@ -16,9 +16,14 @@
  */
 package org.apache.calcite.adapter.enumerable;
 
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.apache.calcite.adapter.jdbc.JdbcConvention;
 import org.apache.calcite.plan.RelOptCluster;
 import org.apache.calcite.plan.RelOptRuleCall;
 import org.apache.calcite.plan.RelRule;
+import org.apache.calcite.plan.RelTrait;
+import org.apache.calcite.plan.RelTraitSet;
 import org.apache.calcite.rel.RelNode;
 import org.apache.calcite.rel.core.CorrelationId;
 import org.apache.calcite.rel.core.Join;
@@ -80,10 +85,38 @@ public EnumerableBatchNestedLoopJoinRule(RelBuilderFactory relBuilderFactory,
   @Override public boolean matches(RelOptRuleCall call) {
     Join join = call.rel(0);
     JoinRelType joinType = join.getJoinType();
-    return joinType == JoinRelType.INNER
+    return (joinType == JoinRelType.INNER
         || joinType == JoinRelType.LEFT
         || joinType == JoinRelType.ANTI
-        || joinType == JoinRelType.SEMI;
+        || joinType == JoinRelType.SEMI )
+           && !containsSingleJdbcSource(join, new AtomicReference<>());
+  }
+
+  private boolean containsSingleJdbcSource(RelNode node, AtomicReference<JdbcConvention> jdbcConvention) {
+      List<RelNode> inputs = node.getInputs();
+      if ( inputs.isEmpty()) {
+        RelTraitSet traitSet = node.getTraitSet();
+        for ( int i = 0; i < traitSet.size(); i++) {
+          RelTrait trait = traitSet.get(i);
+          if ( trait instanceof JdbcConvention ) {
+            JdbcConvention jdbcTrait = (JdbcConvention) trait;
+            if ( jdbcConvention.get() == null){
+              jdbcConvention.set(jdbcTrait);
+              return true;
+            } else {
+              return jdbcConvention.get() == jdbcTrait;
+            }
+          }
+        }
+        return false;
+      } else {
+        for ( RelNode input : inputs) {
+          if ( !containsSingleJdbcSource(input.stripped(), jdbcConvention)) {
+            return false;
+          }
+        }
+      }
+      return true;
   }
 
   @Override public void onMatch(RelOptRuleCall call) {