Try converting all inner joins to filters (#13201)

processing/src/main/java/org/apache/druid/segment/join/JoinableFactoryWrapper.java

@@ -79,34 +79,40 @@ public class JoinableFactoryWrapper
     }
 
     Set<String> rightPrefixes = clauses.stream().map(JoinableClause::getPrefix).collect(Collectors.toSet());
-    // Walk through the list of clauses, picking off any from the start of the list that can be converted to filters.
+    boolean isRightyJoinSeen = false;
-    boolean atStart = true;
     for (JoinableClause clause : clauses) {
-      if (atStart) {
+      // Incase we find a RIGHT/OUTER join, we shouldn't convert join conditions to left column filters for any join
-        // Remove this clause from columnsRequiredByJoinClauses. It's ok if it relies on itself.
+      // afterwards because the values of left colmun might change to NULL after the RIGHT/OUTER join. We should only
-        for (String column : clause.getCondition().getRequiredColumns()) {
+      // consider cases where the values of the filter columns do not change after the join.
-          columnsRequiredByJoinClauses.remove(column, 1);
+      isRightyJoinSeen = isRightyJoinSeen || clause.getJoinType().isRighty();
-        }
+      if (isRightyJoinSeen) {
-
-        final JoinClauseToFilterConversion joinClauseToFilterConversion =
-            convertJoinToFilter(
-                clause,
-                Sets.union(requiredColumns, columnsRequiredByJoinClauses.elementSet()),
-                maxNumFilterValues,
-                rightPrefixes
-            );
-
-        // add the converted filter to the filter list
-        if (joinClauseToFilterConversion.getConvertedFilter() != null) {
-          filterList.add(joinClauseToFilterConversion.getConvertedFilter());
-        }
-        // if the converted filter is partial, keep the join clause too
-        if (!joinClauseToFilterConversion.isJoinClauseFullyConverted()) {
-          clausesToUse.add(clause);
-          atStart = false;
-        }
-      } else {
         clausesToUse.add(clause);
+        continue;
+      }
+      // Remove this clause from columnsRequiredByJoinClauses. It's ok if it relies on itself.
+      for (String column : clause.getCondition().getRequiredColumns()) {
+        columnsRequiredByJoinClauses.remove(column, 1);
+      }
+
+      final JoinClauseToFilterConversion joinClauseToFilterConversion =
+          convertJoinToFilter(
+              clause,
+              Sets.union(requiredColumns, columnsRequiredByJoinClauses.elementSet()),
+              maxNumFilterValues,
+              rightPrefixes
+          );
+
+      // add the converted filter to the filter list
+      if (joinClauseToFilterConversion.getConvertedFilter() != null) {
+        filterList.add(joinClauseToFilterConversion.getConvertedFilter());
+      }
+      // if the converted filter is partial, keep the join clause too
+      if (!joinClauseToFilterConversion.isJoinClauseFullyConverted()) {
+        clausesToUse.add(clause);
+        // add back the required columns by this join since it wasn't converted fully
+        for (String column : clause.getCondition().getRequiredColumns()) {
+          columnsRequiredByJoinClauses.add(column, 1);
+        }
       }
     }
 

processing/src/test/java/org/apache/druid/segment/join/JoinableFactoryWrapperTest.java

@@ -506,7 +506,7 @@ public class JoinableFactoryWrapperTest extends NullHandlingTest
 
     Assert.assertEquals(
         Pair.of(
-            ImmutableList.of(),
+            ImmutableList.of(new InDimFilter("x", TEST_LOOKUP_KEYS)),
             clauses
         ),
         conversion
@@ -590,4 +590,39 @@ public class JoinableFactoryWrapperTest extends NullHandlingTest
         conversion
     );
   }
+
+  @Test
+  public void test_convertJoinsToFilters_dontConvertJoinsDependedOnByPreviousJoins()
+  {
+    // upon discovering a RIGHT/OUTER join, all conversions for subsequent joins should stop since the output of left
+    // table columns might change to NULL after the RIGHT/OUTER join.
+    final ImmutableList<JoinableClause> clauses = ImmutableList.of(
+        new JoinableClause(
+            "j.",
+            LookupJoinable.wrap(new MapLookupExtractor(TEST_LOOKUP, false)),
+            JoinType.RIGHT,
+            JoinConditionAnalysis.forExpression("x == \"j.k\"", "j.", ExprMacroTable.nil())
+        ),
+        new JoinableClause(
+            "_j.",
+            LookupJoinable.wrap(new MapLookupExtractor(TEST_LOOKUP, false)),
+            JoinType.INNER,
+            JoinConditionAnalysis.forExpression("x == \"_j.k\"", "_j.", ExprMacroTable.nil())
+        )
+    );
+
+    final Pair<List<Filter>, List<JoinableClause>> conversion = JoinableFactoryWrapper.convertJoinsToFilters(
+        clauses,
+        ImmutableSet.of("x"),
+        Integer.MAX_VALUE
+    );
+
+    Assert.assertEquals(
+        Pair.of(
+            ImmutableList.of(),
+            clauses
+        ),
+        conversion
+    );
+  }
 }