Do no create filters on right side table columns while join to filter conversion (#12899)

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

@@ -58,6 +58,7 @@ import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.function.Function;
+import java.util.stream.Collectors;
 
 /**
  * A wrapper class over {@link JoinableFactory} for working with {@link Joinable} related classes.
@@ -233,6 +234,7 @@ 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 atStart = true;
     for (JoinableClause clause : clauses) {
@@ -246,7 +248,8 @@ public class JoinableFactoryWrapper
             convertJoinToFilter(
                 clause,
                 Sets.union(requiredColumns, columnsRequiredByJoinClauses.elementSet()),
-                maxNumFilterValues
+                maxNumFilterValues,
+                rightPrefixes
             );
 
         // add the converted filter to the filter list
@@ -287,7 +290,8 @@ public class JoinableFactoryWrapper
   static JoinClauseToFilterConversion convertJoinToFilter(
       final JoinableClause clause,
       final Set<String> requiredColumns,
-      final int maxNumFilterValues
+      final int maxNumFilterValues,
+      final Set<String> rightPrefixes
   )
   {
     if (clause.getJoinType() == JoinType.INNER
@@ -305,6 +309,12 @@ public class JoinableFactoryWrapper
           return new JoinClauseToFilterConversion(null, false);
         }
 
+        // don't add a filter on any right side table columns. only filter on left base table is supported as of now.
+        if (rightPrefixes.stream().anyMatch(leftColumn::startsWith)) {
+          joinClauseFullyConverted = false;
+          continue;
+        }
+
         Joinable.ColumnValuesWithUniqueFlag columnValuesWithUniqueFlag =
             clause.getJoinable().getNonNullColumnValues(condition.getRightColumn(), numValues);
         // For an empty values set, isAllUnique flag will be true only if the column had no non-null values.

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

@@ -833,6 +833,42 @@ public class JoinableFactoryWrapperTest extends NullHandlingTest
     );
   }
 
+  @Test
+  public void test_convertJoinsToFilters_dontConvertJoinsDependedOnByLaterJoins()
+  {
+    // in this multi-join, a join matching two right sides is kept first to ensure :
+    // 1. there is no filter on the right side table column j.k
+    // 2. the right side matching join gets considered for join conversion always (since it is the first join clause)
+    final ImmutableList<JoinableClause> clauses = ImmutableList.of(
+        new JoinableClause(
+            "_j.",
+            LookupJoinable.wrap(new MapLookupExtractor(TEST_LOOKUP, false)),
+            JoinType.INNER,
+            JoinConditionAnalysis.forExpression("\"j.k\" == \"_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
+    );
+  }
+
   @Test
   public void test_convertJoinsToFilters_partialConvertJoinsDependedOnByLaterJoins()
   {