Improve javadoc and introduce leftMarkJoinInternal

Author: xiedeyantu

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

@@ -24,7 +24,8 @@
 import org.immutables.value.Value;
 
 /**
- * Implementation of nested loops with conditional correlates over enumerable inputs.
+ * Rule which converts a {@link LogicalConditionalCorrelate} into its enumerable implementation,
+ * implementing conditional correlates via nested loops over enumerable inputs.
  *
  * @see EnumerableRules#ENUMERABLE_CONDITIONAL_CORRELATE_RULE
  */

linq4j/src/main/java/org/apache/calcite/linq4j/EnumerableDefaults.java

@@ -1966,13 +1966,19 @@ static <TSource, TInner, TKey, TNsKey, TResult> Enumerable<TResult> leftMarkHash
   }
 
   /**
-   * For each row of the {@code outer} enumerable returns correlated rows
-   * from the inner enumerable generated for each outer row, filtered by a predicate
-   * (correlated LEFT_MARK join).
+   * The implementation of left mark join based on nested loop.
+   * This is a unified implementation supporting both correlated and non-correlated cases.
+   *
+   * @param outer           Left input
+   * @param innerProvider   Function that provides inner enumerable for each outer row
+   *                        (for correlated join, pass outerRow -> correlatedInner)
+   *                        (for non-correlated join, pass outerRow -> staticInner)
+   * @param predicate       Non-equi predicate that can return NULL
+   * @param resultSelector  Function that concats the row of left input and marker
    */
-  public static <TSource, TInner, TResult> Enumerable<TResult> correlateLeftMarkJoin(
+  private static <TSource, TInner, TResult> Enumerable<TResult> leftMarkJoinInternal(
       final Enumerable<TSource> outer,
-      final Function1<TSource, Enumerable<TInner>> inner,
+      final Function1<TSource, Enumerable<TInner>> innerProvider,
       final NullablePredicate2<TSource, TInner> predicate,
       final Function2<TSource, @Nullable Boolean, TResult> resultSelector) {
     return new AbstractEnumerable<TResult>() {
@@ -1991,7 +1997,7 @@ public static <TSource, TInner, TResult> Enumerable<TResult> correlateLeftMarkJo
             }
             marker = false;
             final TSource outerRow = outers.current();
-            Enumerable<TInner> innerEnumerable = inner.apply(outerRow);
+            Enumerable<TInner> innerEnumerable = innerProvider.apply(outerRow);
             if (innerEnumerable != null) {
               try (Enumerator<TInner> inners = innerEnumerable.enumerator()) {
                 while (inners.moveNext()) {
@@ -2021,6 +2027,19 @@ public static <TSource, TInner, TResult> Enumerable<TResult> correlateLeftMarkJo
     };
   }
 
+  /**
+   * For each row of the {@code outer} enumerable returns correlated rows
+   * from the inner enumerable generated for each outer row, filtered by a predicate
+   * (correlated LEFT_MARK join).
+   */
+  public static <TSource, TInner, TResult> Enumerable<TResult> correlateLeftMarkJoin(
+      final Enumerable<TSource> outer,
+      final Function1<TSource, Enumerable<TInner>> inner,
+      final NullablePredicate2<TSource, TInner> predicate,
+      final Function2<TSource, @Nullable Boolean, TResult> resultSelector) {
+    return leftMarkJoinInternal(outer, inner, predicate, resultSelector);
+  }
+
   /**
    * The implementation of left mark join based on nested loop.
    *
@@ -2033,47 +2052,7 @@ public static <TSource, TInner, TResult> Enumerable<TResult> leftMarkNestedLoopJ
       final Enumerable<TSource> outer, final Enumerable<TInner> inner,
       final NullablePredicate2<TSource, TInner> predicate,
       final Function2<TSource, @Nullable Boolean, TResult> resultSelector) {
-    return new AbstractEnumerable<TResult>() {
-      @Override public Enumerator<TResult> enumerator() {
-        return new Enumerator<TResult>() {
-          Enumerator<TSource> outers = outer.enumerator();
-          @Nullable Boolean marker = false;
-
-          @Override public TResult current() {
-            return resultSelector.apply(outers.current(), marker);
-          }
-
-          @Override public boolean moveNext() {
-            if (!outers.moveNext()) {
-              return false;
-            }
-            marker = false;
-            final TSource outerRow = outers.current();
-            try (Enumerator<TInner> inners = inner.enumerator()) {
-              while (inners.moveNext()) {
-                final TInner innerRow = inners.current();
-                Boolean predicateMatched = predicate.apply(outerRow, innerRow);
-                if (predicateMatched == null) {
-                  marker = null;
-                } else if (predicateMatched) {
-                  marker = true;
-                  break;
-                }
-              }
-            }
-            return true;
-          }
-
-          @Override public void reset() {
-            outers.reset();
-          }
-
-          @Override public void close() {
-            outers.close();
-          }
-        };
-      }
-    };
+    return leftMarkJoinInternal(outer, ignored -> inner, predicate, resultSelector);
   }
 
   /**