Add option to control join filter rewrites (#9472)

* Add option to control join filter rewrites

* Fix inspections

benchmarks/src/test/java/org/apache/druid/benchmark/JoinAndLookupBenchmark.java

@@ -139,7 +139,8 @@ public class JoinAndLookupBenchmark
                 )
             )
         ),
-        QueryContexts.DEFAULT_ENABLE_JOIN_FILTER_PUSH_DOWN
+        QueryContexts.DEFAULT_ENABLE_JOIN_FILTER_PUSH_DOWN,
+        QueryContexts.DEFAULT_ENABLE_JOIN_FILTER_REWRITE
     );
 
     hashJoinLookupLongKeySegment = new HashJoinSegment(
@@ -156,7 +157,8 @@ public class JoinAndLookupBenchmark
                 )
             )
         ),
-        QueryContexts.DEFAULT_ENABLE_JOIN_FILTER_PUSH_DOWN
+        QueryContexts.DEFAULT_ENABLE_JOIN_FILTER_PUSH_DOWN,
+        QueryContexts.DEFAULT_ENABLE_JOIN_FILTER_REWRITE
     );
 
     hashJoinIndexedTableStringKeySegment = new HashJoinSegment(
@@ -173,7 +175,8 @@ public class JoinAndLookupBenchmark
                 )
             )
         ),
-        QueryContexts.DEFAULT_ENABLE_JOIN_FILTER_PUSH_DOWN
+        QueryContexts.DEFAULT_ENABLE_JOIN_FILTER_PUSH_DOWN,
+        QueryContexts.DEFAULT_ENABLE_JOIN_FILTER_REWRITE
     );
 
     hashJoinIndexedTableLongKeySegment = new HashJoinSegment(
@@ -190,7 +193,8 @@ public class JoinAndLookupBenchmark
                 )
             )
         ),
-        QueryContexts.DEFAULT_ENABLE_JOIN_FILTER_PUSH_DOWN
+        QueryContexts.DEFAULT_ENABLE_JOIN_FILTER_PUSH_DOWN,
+        QueryContexts.DEFAULT_ENABLE_JOIN_FILTER_REWRITE
     );
 
     final Map<String, String> countryCodeToNameMap = JoinTestHelper.createCountryIsoCodeToNameLookup().getMap();

processing/src/main/java/org/apache/druid/query/QueryContexts.java

@@ -46,6 +46,7 @@ public class QueryContexts
   public static final String VECTORIZE_KEY = "vectorize";
   public static final String VECTOR_SIZE_KEY = "vectorSize";
   public static final String JOIN_FILTER_PUSH_DOWN_KEY = "enableJoinFilterPushDown";
+  public static final String JOIN_FILTER_REWRITE_KEY = "enableJoinFilterRewrite";
 
   public static final boolean DEFAULT_BY_SEGMENT = false;
   public static final boolean DEFAULT_POPULATE_CACHE = true;
@@ -59,6 +60,7 @@ public class QueryContexts
   public static final long NO_TIMEOUT = 0;
   public static final boolean DEFAULT_ENABLE_PARALLEL_MERGE = true;
   public static final boolean DEFAULT_ENABLE_JOIN_FILTER_PUSH_DOWN = true;
+  public static final boolean DEFAULT_ENABLE_JOIN_FILTER_REWRITE = false;
 
   @SuppressWarnings("unused") // Used by Jackson serialization
   public enum Vectorize
@@ -225,6 +227,11 @@ public class QueryContexts
     return parseBoolean(query, JOIN_FILTER_PUSH_DOWN_KEY, DEFAULT_ENABLE_JOIN_FILTER_PUSH_DOWN);
   }
 
+  public static <T> boolean getEnableJoinFilterRewrite(Query<T> query)
+  {
+    return parseBoolean(query, JOIN_FILTER_REWRITE_KEY, DEFAULT_ENABLE_JOIN_FILTER_REWRITE);
+  }
+
   public static <T> Query<T> withMaxScatterGatherBytes(Query<T> query, long maxScatterGatherBytesLimit)
   {
     Object obj = query.getContextValue(MAX_SCATTER_GATHER_BYTES_KEY);

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

@@ -41,6 +41,7 @@ public class HashJoinSegment extends AbstractSegment
   private final Segment baseSegment;
   private final List<JoinableClause> clauses;
   private final boolean enableFilterPushDown;
+  private final boolean enableFilterRewrite;
 
   /**
    * @param baseSegment          The left-hand side base segment
@@ -52,12 +53,14 @@ public class HashJoinSegment extends AbstractSegment
   public HashJoinSegment(
       Segment baseSegment,
       List<JoinableClause> clauses,
-      boolean enableFilterPushDown
+      boolean enableFilterPushDown,
+      boolean enableFilterRewrite
   )
   {
     this.baseSegment = baseSegment;
     this.clauses = clauses;
     this.enableFilterPushDown = enableFilterPushDown;
+    this.enableFilterRewrite = enableFilterRewrite;
 
     // Verify 'clauses' is nonempty (otherwise it's a waste to create this object, and the caller should know)
     if (clauses.isEmpty()) {
@@ -90,7 +93,7 @@ public class HashJoinSegment extends AbstractSegment
   @Override
   public StorageAdapter asStorageAdapter()
   {
-    return new HashJoinSegmentStorageAdapter(baseSegment.asStorageAdapter(), clauses, enableFilterPushDown);
+    return new HashJoinSegmentStorageAdapter(baseSegment.asStorageAdapter(), clauses, enableFilterPushDown, enableFilterRewrite);
   }
 
   @Override

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

@@ -56,6 +56,7 @@ public class HashJoinSegmentStorageAdapter implements StorageAdapter
   private final StorageAdapter baseAdapter;
   private final List<JoinableClause> clauses;
   private final boolean enableFilterPushDown;
+  private final boolean enableFilterRewrite;
 
   /**
    * @param baseAdapter          A StorageAdapter for the left-hand side base segment
@@ -66,12 +67,14 @@ public class HashJoinSegmentStorageAdapter implements StorageAdapter
   HashJoinSegmentStorageAdapter(
       StorageAdapter baseAdapter,
       List<JoinableClause> clauses,
-      final boolean enableFilterPushDown
+      final boolean enableFilterPushDown,
+      final boolean enableFilterRewrite
   )
   {
     this.baseAdapter = baseAdapter;
     this.clauses = clauses;
     this.enableFilterPushDown = enableFilterPushDown;
+    this.enableFilterRewrite = enableFilterRewrite;
   }
 
   @VisibleForTesting
@@ -80,7 +83,12 @@ public class HashJoinSegmentStorageAdapter implements StorageAdapter
       List<JoinableClause> clauses
   )
   {
-    this(baseAdapter, clauses, QueryContexts.DEFAULT_ENABLE_JOIN_FILTER_PUSH_DOWN);
+    this(
+        baseAdapter,
+        clauses,
+        QueryContexts.DEFAULT_ENABLE_JOIN_FILTER_PUSH_DOWN,
+        QueryContexts.DEFAULT_ENABLE_JOIN_FILTER_REWRITE
+    );
   }
 
   @Override
@@ -242,7 +250,8 @@ public class HashJoinSegmentStorageAdapter implements StorageAdapter
         this,
         baseColumns,
         filter,
-        enableFilterPushDown
+        enableFilterPushDown,
+        enableFilterRewrite
     );
     preJoinVirtualColumns.addAll(joinFilterSplit.getPushDownVirtualColumns());
 

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

@@ -82,7 +82,8 @@ public class Joinables
       final List<PreJoinableClause> clauses,
       final JoinableFactory joinableFactory,
       final AtomicLong cpuTimeAccumulator,
-      final boolean enableFilterPushDown
+      final boolean enableFilterPushDown,
+      final boolean enableFilterRewrite
   )
   {
     return JvmUtils.safeAccumulateThreadCpuTime(
@@ -92,7 +93,7 @@ public class Joinables
             return Function.identity();
           } else {
             final List<JoinableClause> joinableClauses = createJoinableClauses(clauses, joinableFactory);
-            return baseSegment -> new HashJoinSegment(baseSegment, joinableClauses, enableFilterPushDown);
+            return baseSegment -> new HashJoinSegment(baseSegment, joinableClauses, enableFilterPushDown, enableFilterRewrite);
           }
         }
     );

processing/src/main/java/org/apache/druid/segment/join/filter/JoinFilterAnalyzer.java

@@ -51,7 +51,7 @@ import java.util.Set;
  * When there is a filter in a join query, we can sometimes improve performance by applying parts of the filter
  * when we first read from the base table instead of after the join.
  *
- * This class provides a {@link #splitFilter(HashJoinSegmentStorageAdapter, Set, Filter, boolean)} method that
+ * This class provides a {@link #splitFilter(HashJoinSegmentStorageAdapter, Set, Filter, boolean, boolean)} method that
  * takes a filter and splits it into a portion that should be applied to the base table prior to the join, and a
  * portion that should be applied after the join.
  *
@@ -88,7 +88,8 @@ public class JoinFilterAnalyzer
       HashJoinSegmentStorageAdapter hashJoinSegmentStorageAdapter,
       Set<String> baseColumnNames,
       @Nullable Filter originalFilter,
-      boolean enableFilterPushDown
+      boolean enableFilterPushDown,
+      boolean enableFilterRewrite
   )
   {
     if (originalFilter == null) {
@@ -150,7 +151,8 @@ public class JoinFilterAnalyzer
           orClause,
           prefixes,
           equiconditions,
-          correlationCache
+          correlationCache,
+          enableFilterRewrite
       );
       if (joinFilterAnalysis.isCanPushDown()) {
         leftFilters.add(joinFilterAnalysis.getPushDownFilter().get());
@@ -189,7 +191,9 @@ public class JoinFilterAnalyzer
       Filter filterClause,
       Map<String, JoinableClause> prefixes,
       Map<String, Set<Expr>> equiconditions,
-      Map<String, Optional<List<JoinFilterColumnCorrelationAnalysis>>> correlationCache
+      Map<String, Optional<List<JoinFilterColumnCorrelationAnalysis>>> correlationCache,
+      boolean enableFilterRewrite
+
   )
   {
     // NULL matching conditions are not currently pushed down.
@@ -199,6 +203,27 @@ public class JoinFilterAnalyzer
       return JoinFilterAnalysis.createNoPushdownFilterAnalysis(filterClause);
     }
 
+    boolean baseTableOnly = true;
+    for (String requiredColumn : filterClause.getRequiredColumns()) {
+      if (!baseColumnNames.contains(requiredColumn)) {
+        baseTableOnly = false;
+        break;
+      }
+    }
+
+    if (baseTableOnly) {
+      return new JoinFilterAnalysis(
+          false,
+          filterClause,
+          filterClause,
+          ImmutableList.of()
+      );
+    }
+
+    if (!enableFilterRewrite) {
+      return JoinFilterAnalysis.createNoPushdownFilterAnalysis(filterClause);
+    }
+
     // Currently we only support rewrites of selector filters and selector filters within OR filters.
     if (filterClause instanceof SelectorFilter) {
       return rewriteSelectorFilter(
@@ -220,17 +245,7 @@ public class JoinFilterAnalyzer
       );
     }
 
-    for (String requiredColumn : filterClause.getRequiredColumns()) {
+    return JoinFilterAnalysis.createNoPushdownFilterAnalysis(filterClause);
-      if (!baseColumnNames.contains(requiredColumn)) {
-        return JoinFilterAnalysis.createNoPushdownFilterAnalysis(filterClause);
-      }
-    }
-    return new JoinFilterAnalysis(
-        false,
-        filterClause,
-        filterClause,
-        ImmutableList.of()
-    );
   }
 
   /**

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

@@ -78,7 +78,8 @@ public class HashJoinSegmentTest
                 JoinConditionAnalysis.forExpression("1", "j1.", ExprMacroTable.nil())
             )
         ),
-        QueryContexts.DEFAULT_ENABLE_JOIN_FILTER_PUSH_DOWN
+        QueryContexts.DEFAULT_ENABLE_JOIN_FILTER_PUSH_DOWN,
+        QueryContexts.DEFAULT_ENABLE_JOIN_FILTER_REWRITE
     );
   }
 
@@ -91,7 +92,8 @@ public class HashJoinSegmentTest
     final HashJoinSegment ignored = new HashJoinSegment(
         baseSegment,
         ImmutableList.of(),
-        QueryContexts.DEFAULT_ENABLE_JOIN_FILTER_PUSH_DOWN
+        QueryContexts.DEFAULT_ENABLE_JOIN_FILTER_PUSH_DOWN,
+        QueryContexts.DEFAULT_ENABLE_JOIN_FILTER_REWRITE
     );
   }
 

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

@@ -68,6 +68,7 @@ public class JoinFilterAnalyzerTest extends BaseHashJoinSegmentStorageAdapterTes
         adapter,
         adapter.determineBaseColumnsWithPreAndPostJoinVirtualColumns(VirtualColumns.EMPTY, null, null),
         originalFilter,
+        true,
         true
     );
     Assert.assertEquals(expectedFilterSplit, actualFilterSplit);
@@ -140,6 +141,7 @@ public class JoinFilterAnalyzerTest extends BaseHashJoinSegmentStorageAdapterTes
         adapter,
         adapter.determineBaseColumnsWithPreAndPostJoinVirtualColumns(VirtualColumns.EMPTY, null, null),
         originalFilter,
+        true,
         true
     );
     Assert.assertEquals(expectedFilterSplit, actualFilterSplit);
@@ -195,6 +197,7 @@ public class JoinFilterAnalyzerTest extends BaseHashJoinSegmentStorageAdapterTes
         adapter,
         adapter.determineBaseColumnsWithPreAndPostJoinVirtualColumns(VirtualColumns.EMPTY, null, null),
         originalFilter,
+        true,
         true
     );
     Assert.assertEquals(expectedFilterSplit, actualFilterSplit);
@@ -258,6 +261,7 @@ public class JoinFilterAnalyzerTest extends BaseHashJoinSegmentStorageAdapterTes
         adapter,
         adapter.determineBaseColumnsWithPreAndPostJoinVirtualColumns(VirtualColumns.EMPTY, null, null),
         originalFilter,
+        true,
         true
     );
     Assert.assertEquals(expectedFilterSplit, actualFilterSplit);
@@ -315,6 +319,7 @@ public class JoinFilterAnalyzerTest extends BaseHashJoinSegmentStorageAdapterTes
         adapter,
         adapter.determineBaseColumnsWithPreAndPostJoinVirtualColumns(VirtualColumns.EMPTY, null, null),
         originalFilter,
+        true,
         true
     );
     Assert.assertEquals(expectedFilterSplit, actualFilterSplit);
@@ -372,6 +377,7 @@ public class JoinFilterAnalyzerTest extends BaseHashJoinSegmentStorageAdapterTes
         adapter,
         adapter.determineBaseColumnsWithPreAndPostJoinVirtualColumns(virtualColumns, null, null),
         originalFilter,
+        true,
         true
     );
     Assert.assertEquals(expectedFilterSplit, actualFilterSplit);
@@ -442,6 +448,7 @@ public class JoinFilterAnalyzerTest extends BaseHashJoinSegmentStorageAdapterTes
         adapter,
         adapter.determineBaseColumnsWithPreAndPostJoinVirtualColumns(virtualColumns, null, null),
         originalFilter,
+        true,
         true
     );
     Assert.assertEquals(expectedFilterSplit, actualFilterSplit);
@@ -601,6 +608,7 @@ public class JoinFilterAnalyzerTest extends BaseHashJoinSegmentStorageAdapterTes
         adapter,
         adapter.determineBaseColumnsWithPreAndPostJoinVirtualColumns(VirtualColumns.EMPTY, null, null),
         originalFilter,
+        true,
         true
     );
     Assert.assertEquals(expectedFilterSplit, actualFilterSplit);
@@ -672,6 +680,7 @@ public class JoinFilterAnalyzerTest extends BaseHashJoinSegmentStorageAdapterTes
         adapter,
         adapter.determineBaseColumnsWithPreAndPostJoinVirtualColumns(VirtualColumns.EMPTY, null, null),
         originalFilter,
+        true,
         true
     );
     ExpressionVirtualColumn expectedVirtualColumn = new ExpressionVirtualColumn(
@@ -830,6 +839,7 @@ public class JoinFilterAnalyzerTest extends BaseHashJoinSegmentStorageAdapterTes
         adapter,
         adapter.determineBaseColumnsWithPreAndPostJoinVirtualColumns(VirtualColumns.EMPTY, null, null),
         originalFilter,
+        true,
         true
     );
     Assert.assertEquals(expectedFilterSplit, actualFilterSplit);
@@ -908,6 +918,7 @@ public class JoinFilterAnalyzerTest extends BaseHashJoinSegmentStorageAdapterTes
         adapter,
         adapter.determineBaseColumnsWithPreAndPostJoinVirtualColumns(VirtualColumns.EMPTY, null, null),
         filter,
+        true,
         true
     );
     Assert.assertEquals(
@@ -971,6 +982,7 @@ public class JoinFilterAnalyzerTest extends BaseHashJoinSegmentStorageAdapterTes
         adapter,
         adapter.determineBaseColumnsWithPreAndPostJoinVirtualColumns(VirtualColumns.EMPTY, null, null),
         originalFilter,
+        true,
         true
     );
     Assert.assertEquals(expectedFilterSplit, actualFilterSplit);
@@ -1026,6 +1038,7 @@ public class JoinFilterAnalyzerTest extends BaseHashJoinSegmentStorageAdapterTes
         adapter,
         adapter.determineBaseColumnsWithPreAndPostJoinVirtualColumns(VirtualColumns.EMPTY, null, null),
         originalFilter,
+        true,
         true
     );
     Assert.assertEquals(expectedFilterSplit, actualFilterSplit);
@@ -1079,6 +1092,7 @@ public class JoinFilterAnalyzerTest extends BaseHashJoinSegmentStorageAdapterTes
         adapter,
         adapter.determineBaseColumnsWithPreAndPostJoinVirtualColumns(VirtualColumns.EMPTY, null, null),
         originalFilter,
+        true,
         true
     );
     Assert.assertEquals(expectedFilterSplit, actualFilterSplit);
@@ -1140,6 +1154,7 @@ public class JoinFilterAnalyzerTest extends BaseHashJoinSegmentStorageAdapterTes
         adapter,
         adapter.determineBaseColumnsWithPreAndPostJoinVirtualColumns(VirtualColumns.EMPTY, null, null),
         originalFilter,
+        true,
         true
     );
     Assert.assertEquals(expectedFilterSplit, actualFilterSplit);
@@ -1192,6 +1207,7 @@ public class JoinFilterAnalyzerTest extends BaseHashJoinSegmentStorageAdapterTes
         adapter,
         adapter.determineBaseColumnsWithPreAndPostJoinVirtualColumns(VirtualColumns.EMPTY, null, null),
         filter,
+        true,
         true
     );
     Assert.assertEquals(expectedFilterSplit, actualFilterSplit);
@@ -1247,6 +1263,7 @@ public class JoinFilterAnalyzerTest extends BaseHashJoinSegmentStorageAdapterTes
         adapter,
         adapter.determineBaseColumnsWithPreAndPostJoinVirtualColumns(VirtualColumns.EMPTY, null, null),
         originalFilter,
+        true,
         true
     );
     Assert.assertEquals(expectedFilterSplit, actualFilterSplit);
@@ -1312,6 +1329,7 @@ public class JoinFilterAnalyzerTest extends BaseHashJoinSegmentStorageAdapterTes
         adapter,
         adapter.determineBaseColumnsWithPreAndPostJoinVirtualColumns(VirtualColumns.EMPTY, null, null),
         originalFilter,
+        true,
         true
     );
     Assert.assertEquals(expectedFilterSplit, actualFilterSplit);
@@ -1370,6 +1388,7 @@ public class JoinFilterAnalyzerTest extends BaseHashJoinSegmentStorageAdapterTes
         adapter,
         adapter.determineBaseColumnsWithPreAndPostJoinVirtualColumns(VirtualColumns.EMPTY, null, null),
         originalFilter,
+        true,
         true
     );
     Assert.assertEquals(expectedFilterSplit, actualFilterSplit);
@@ -1402,7 +1421,8 @@ public class JoinFilterAnalyzerTest extends BaseHashJoinSegmentStorageAdapterTes
             factToRegion(JoinType.LEFT),
             regionToCountry(JoinType.LEFT)
         ),
-        false
+        false,
+        true
     );
     Filter originalFilter = new SelectorFilter("page", "Peremptory norm");
 
@@ -1415,7 +1435,8 @@ public class JoinFilterAnalyzerTest extends BaseHashJoinSegmentStorageAdapterTes
         adapter,
         adapter.determineBaseColumnsWithPreAndPostJoinVirtualColumns(VirtualColumns.EMPTY, null, null),
         originalFilter,
-        adapter.isEnableFilterPushDown()
+        adapter.isEnableFilterPushDown(),
+        true
     );
     Assert.assertEquals(expectedFilterSplit, actualFilterSplit);
 
@@ -1439,6 +1460,101 @@ public class JoinFilterAnalyzerTest extends BaseHashJoinSegmentStorageAdapterTes
     );
   }
 
+  @Test
+  public void test_filterPushDown_factToRegionToCountryLeftEnablePushDownDisableRewrite()
+  {
+    HashJoinSegmentStorageAdapter adapter = new HashJoinSegmentStorageAdapter(
+        factSegment.asStorageAdapter(),
+        ImmutableList.of(
+            factToRegion(JoinType.LEFT),
+            regionToCountry(JoinType.LEFT)
+        )
+    );
+    Filter originalFilter = new AndFilter(
+        ImmutableList.of(
+            new SelectorFilter("channel", "#en.wikipedia"),
+            new SelectorFilter("rtc.countryName", "United States"),
+            new OrFilter(
+                ImmutableList.of(
+                    new SelectorFilter("page", "DirecTV"),
+                    new SelectorFilter("rtc.countryIsoCode", "US")
+                )
+            ),
+            new BoundFilter(new BoundDimFilter(
+                "namespace",
+                "Main",
+                "Main",
+                false,
+                false,
+                null,
+                null,
+                null
+            ))
+        )
+    );
+
+    JoinFilterSplit expectedFilterSplit = new JoinFilterSplit(
+        new AndFilter(
+            ImmutableList.of(
+                new SelectorFilter("channel", "#en.wikipedia"),
+                new BoundFilter(new BoundDimFilter(
+                    "namespace",
+                    "Main",
+                    "Main",
+                    false,
+                    false,
+                    null,
+                    null,
+                    null
+                ))
+            )
+        ),
+        new AndFilter(
+            ImmutableList.of(
+                new SelectorFilter("rtc.countryName", "United States"),
+                new OrFilter(
+                    ImmutableList.of(
+                        new SelectorFilter("page", "DirecTV"),
+                        new SelectorFilter("rtc.countryIsoCode", "US")
+                    )
+                )
+            )
+        ),
+        ImmutableList.of()
+    );
+    JoinFilterSplit actualFilterSplit = JoinFilterAnalyzer.splitFilter(
+        adapter,
+        adapter.determineBaseColumnsWithPreAndPostJoinVirtualColumns(VirtualColumns.EMPTY, null, null),
+        originalFilter,
+        true,
+        false
+    );
+    Assert.assertEquals(expectedFilterSplit, actualFilterSplit);
+
+    JoinTestHelper.verifyCursors(
+        adapter.makeCursors(
+            originalFilter,
+            Intervals.ETERNITY,
+            VirtualColumns.EMPTY,
+            Granularities.ALL,
+            false,
+            null
+        ),
+        ImmutableList.of(
+            "page",
+            FACT_TO_REGION_PREFIX + "regionName",
+            REGION_TO_COUNTRY_PREFIX + "countryName"
+        ),
+        ImmutableList.of(
+            new Object[]{"President of India", "California", "United States"},
+            new Object[]{"Otjiwarongo Airport", "California", "United States"},
+            new Object[]{"DirecTV", "North Carolina", "United States"},
+            new Object[]{"Carlo Curti", "California", "United States"},
+            new Object[]{"Old Anatolian Turkish", "Virginia", "United States"}
+        )
+    );
+  }
+
   @Test
   public void test_JoinFilterSplit_equals()
   {

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

@@ -98,7 +98,8 @@ public class JoinablesTest
         ImmutableList.of(),
         NoopJoinableFactory.INSTANCE,
         new AtomicLong(),
-        QueryContexts.DEFAULT_ENABLE_JOIN_FILTER_PUSH_DOWN
+        QueryContexts.DEFAULT_ENABLE_JOIN_FILTER_PUSH_DOWN,
+        QueryContexts.DEFAULT_ENABLE_JOIN_FILTER_REWRITE
     );
 
     Assert.assertSame(Function.identity(), segmentMapFn);
@@ -122,7 +123,8 @@ public class JoinablesTest
         ImmutableList.of(clause),
         NoopJoinableFactory.INSTANCE,
         new AtomicLong(),
-        QueryContexts.DEFAULT_ENABLE_JOIN_FILTER_PUSH_DOWN
+        QueryContexts.DEFAULT_ENABLE_JOIN_FILTER_PUSH_DOWN,
+        QueryContexts.DEFAULT_ENABLE_JOIN_FILTER_REWRITE
     );
   }
 
@@ -154,7 +156,8 @@ public class JoinablesTest
           }
         },
         new AtomicLong(),
-        QueryContexts.DEFAULT_ENABLE_JOIN_FILTER_PUSH_DOWN
+        QueryContexts.DEFAULT_ENABLE_JOIN_FILTER_PUSH_DOWN,
+        QueryContexts.DEFAULT_ENABLE_JOIN_FILTER_REWRITE
     );
 
     Assert.assertNotSame(Function.identity(), segmentMapFn);

server/src/main/java/org/apache/druid/segment/realtime/appenderator/SinkQuerySegmentWalker.java

@@ -175,7 +175,8 @@ public class SinkQuerySegmentWalker implements QuerySegmentWalker
         analysis.getPreJoinableClauses(),
         joinableFactory,
         cpuTimeAccumulator,
-        QueryContexts.getEnableJoinFilterPushDown(query)
+        QueryContexts.getEnableJoinFilterPushDown(query),
+        QueryContexts.getEnableJoinFilterRewrite(query)
     );
 
     Iterable<QueryRunner<T>> perSegmentRunners = Iterables.transform(

server/src/main/java/org/apache/druid/server/coordination/ServerManager.java

@@ -196,7 +196,8 @@ public class ServerManager implements QuerySegmentWalker
         analysis.getPreJoinableClauses(),
         joinableFactory,
         cpuTimeAccumulator,
-        QueryContexts.getEnableJoinFilterPushDown(query)
+        QueryContexts.getEnableJoinFilterPushDown(query),
+        QueryContexts.getEnableJoinFilterRewrite(query)
     );
 
     FunctionalIterable<QueryRunner<T>> queryRunners = FunctionalIterable

sql/src/test/java/org/apache/druid/sql/calcite/util/SpecificSegmentsQuerySegmentWalker.java

@@ -374,7 +374,8 @@ public class SpecificSegmentsQuerySegmentWalker implements QuerySegmentWalker, C
           analysis.getPreJoinableClauses(),
           joinableFactory,
           new AtomicLong(),
-          QueryContexts.getEnableJoinFilterPushDown(query)
+          QueryContexts.getEnableJoinFilterPushDown(query),
+          QueryContexts.getEnableJoinFilterRewrite(query)
       );
 
       final QueryRunner<T> baseRunner = new FinalizeResultsQueryRunner<>(