diff --git a/benchmarks/src/test/java/org/apache/druid/benchmark/JoinAndLookupBenchmark.java b/benchmarks/src/test/java/org/apache/druid/benchmark/JoinAndLookupBenchmark.java index 5f818acbbe7..146daffd58d 100644 --- a/benchmarks/src/test/java/org/apache/druid/benchmark/JoinAndLookupBenchmark.java +++ b/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 countryCodeToNameMap = JoinTestHelper.createCountryIsoCodeToNameLookup().getMap(); diff --git a/processing/src/main/java/org/apache/druid/query/QueryContexts.java b/processing/src/main/java/org/apache/druid/query/QueryContexts.java index 70cd1e61ea5..2af3b95156b 100644 --- a/processing/src/main/java/org/apache/druid/query/QueryContexts.java +++ b/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 boolean getEnableJoinFilterRewrite(Query query) + { + return parseBoolean(query, JOIN_FILTER_REWRITE_KEY, DEFAULT_ENABLE_JOIN_FILTER_REWRITE); + } + public static Query withMaxScatterGatherBytes(Query query, long maxScatterGatherBytesLimit) { Object obj = query.getContextValue(MAX_SCATTER_GATHER_BYTES_KEY); diff --git a/processing/src/main/java/org/apache/druid/segment/join/HashJoinSegment.java b/processing/src/main/java/org/apache/druid/segment/join/HashJoinSegment.java index b19bc0f56d8..6ad0aa9f38e 100644 --- a/processing/src/main/java/org/apache/druid/segment/join/HashJoinSegment.java +++ b/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 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 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 diff --git a/processing/src/main/java/org/apache/druid/segment/join/HashJoinSegmentStorageAdapter.java b/processing/src/main/java/org/apache/druid/segment/join/HashJoinSegmentStorageAdapter.java index 99cf4a33cb0..f5900995330 100644 --- a/processing/src/main/java/org/apache/druid/segment/join/HashJoinSegmentStorageAdapter.java +++ b/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 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 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 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()); diff --git a/processing/src/main/java/org/apache/druid/segment/join/Joinables.java b/processing/src/main/java/org/apache/druid/segment/join/Joinables.java index 1d9dae6245c..792eed0420e 100644 --- a/processing/src/main/java/org/apache/druid/segment/join/Joinables.java +++ b/processing/src/main/java/org/apache/druid/segment/join/Joinables.java @@ -82,7 +82,8 @@ public class Joinables final List 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 joinableClauses = createJoinableClauses(clauses, joinableFactory); - return baseSegment -> new HashJoinSegment(baseSegment, joinableClauses, enableFilterPushDown); + return baseSegment -> new HashJoinSegment(baseSegment, joinableClauses, enableFilterPushDown, enableFilterRewrite); } } ); diff --git a/processing/src/main/java/org/apache/druid/segment/join/filter/JoinFilterAnalyzer.java b/processing/src/main/java/org/apache/druid/segment/join/filter/JoinFilterAnalyzer.java index dad06865e06..574446decfa 100644 --- a/processing/src/main/java/org/apache/druid/segment/join/filter/JoinFilterAnalyzer.java +++ b/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 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 prefixes, Map> equiconditions, - Map>> correlationCache + Map>> 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()) { - if (!baseColumnNames.contains(requiredColumn)) { - return JoinFilterAnalysis.createNoPushdownFilterAnalysis(filterClause); - } - } - return new JoinFilterAnalysis( - false, - filterClause, - filterClause, - ImmutableList.of() - ); + return JoinFilterAnalysis.createNoPushdownFilterAnalysis(filterClause); } /** diff --git a/processing/src/test/java/org/apache/druid/segment/join/HashJoinSegmentTest.java b/processing/src/test/java/org/apache/druid/segment/join/HashJoinSegmentTest.java index 19a4842b6bf..4950e841e5b 100644 --- a/processing/src/test/java/org/apache/druid/segment/join/HashJoinSegmentTest.java +++ b/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 ); } diff --git a/processing/src/test/java/org/apache/druid/segment/join/JoinFilterAnalyzerTest.java b/processing/src/test/java/org/apache/druid/segment/join/JoinFilterAnalyzerTest.java index 05b93de6892..2241452f8c9 100644 --- a/processing/src/test/java/org/apache/druid/segment/join/JoinFilterAnalyzerTest.java +++ b/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() { diff --git a/processing/src/test/java/org/apache/druid/segment/join/JoinablesTest.java b/processing/src/test/java/org/apache/druid/segment/join/JoinablesTest.java index 7ad3503f1f4..3bd120b42b3 100644 --- a/processing/src/test/java/org/apache/druid/segment/join/JoinablesTest.java +++ b/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); diff --git a/server/src/main/java/org/apache/druid/segment/realtime/appenderator/SinkQuerySegmentWalker.java b/server/src/main/java/org/apache/druid/segment/realtime/appenderator/SinkQuerySegmentWalker.java index ae337a1c1b9..0fa8e39acc8 100644 --- a/server/src/main/java/org/apache/druid/segment/realtime/appenderator/SinkQuerySegmentWalker.java +++ b/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> perSegmentRunners = Iterables.transform( diff --git a/server/src/main/java/org/apache/druid/server/coordination/ServerManager.java b/server/src/main/java/org/apache/druid/server/coordination/ServerManager.java index e93badb2b08..313eb158f6f 100644 --- a/server/src/main/java/org/apache/druid/server/coordination/ServerManager.java +++ b/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> queryRunners = FunctionalIterable diff --git a/sql/src/test/java/org/apache/druid/sql/calcite/util/SpecificSegmentsQuerySegmentWalker.java b/sql/src/test/java/org/apache/druid/sql/calcite/util/SpecificSegmentsQuerySegmentWalker.java index 46f8dad9a23..70ed9cff838 100644 --- a/sql/src/test/java/org/apache/druid/sql/calcite/util/SpecificSegmentsQuerySegmentWalker.java +++ b/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 baseRunner = new FinalizeResultsQueryRunner<>(