Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Lazily build Filter in FilteredAggregatorFactory to avoid parsing exceptions in Router #15526

Merged
merged 3 commits into from
Dec 9, 2023
Merged

Lazily build Filter in FilteredAggregatorFactory to avoid parsing exceptions in Router #15526

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
2fb4a32
Memoize dimFilter::toFilter call
findingrish Dec 8, 2023
356b884
Add UT
findingrish Dec 8, 2023
a110b63
Fix checkstyle
findingrish Dec 8, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 9 additions & 7 deletions processing/src/main/java/org/apache/druid/query/aggregation/FilteredAggregatorFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,8 @@
import com.fasterxml.jackson.annotation.JsonProperty;
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import com.google.common.base.Supplier;
import com.google.common.base.Suppliers;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
import org.apache.druid.query.PerSegmentQueryOptimizationContext;
Expand All @@ -49,7 +51,7 @@ public class FilteredAggregatorFactory extends AggregatorFactory
{
private final AggregatorFactory delegate;
private final DimFilter dimFilter;
private final Filter filter;
private final Supplier<Filter> filterSupplier;

@Nullable
private final String name;
Expand All @@ -75,14 +77,14 @@ public FilteredAggregatorFactory(

this.delegate = delegate;
this.dimFilter = dimFilter;
this.filter = dimFilter.toFilter();
this.filterSupplier = Suppliers.memoize(dimFilter::toFilter);
this.name = name;
}

@Override
public Aggregator factorize(ColumnSelectorFactory columnSelectorFactory)
{
final ValueMatcher valueMatcher = filter.makeMatcher(columnSelectorFactory);
final ValueMatcher valueMatcher = filterSupplier.get().makeMatcher(columnSelectorFactory);
return new FilteredAggregator(
valueMatcher,
delegate.factorize(columnSelectorFactory)
Expand All @@ -92,7 +94,7 @@ public Aggregator factorize(ColumnSelectorFactory columnSelectorFactory)
@Override
public BufferAggregator factorizeBuffered(ColumnSelectorFactory columnSelectorFactory)
{
final ValueMatcher valueMatcher = filter.makeMatcher(columnSelectorFactory);
final ValueMatcher valueMatcher = filterSupplier.get().makeMatcher(columnSelectorFactory);
return new FilteredBufferAggregator(
valueMatcher,
delegate.factorizeBuffered(columnSelectorFactory)
Expand All @@ -103,7 +105,7 @@ public BufferAggregator factorizeBuffered(ColumnSelectorFactory columnSelectorFa
public VectorAggregator factorizeVector(VectorColumnSelectorFactory columnSelectorFactory)
{
Preconditions.checkState(canVectorize(columnSelectorFactory), "Cannot vectorize");
final VectorValueMatcher valueMatcher = filter.makeVectorMatcher(columnSelectorFactory);
final VectorValueMatcher valueMatcher = filterSupplier.get().makeVectorMatcher(columnSelectorFactory);
return new FilteredVectorAggregator(
valueMatcher,
delegate.factorizeVector(columnSelectorFactory)
Expand All @@ -113,7 +115,7 @@ public VectorAggregator factorizeVector(VectorColumnSelectorFactory columnSelect
@Override
public boolean canVectorize(ColumnInspector columnInspector)
{
return delegate.canVectorize(columnInspector) && filter.canVectorizeMatcher(columnInspector);
return delegate.canVectorize(columnInspector) && filterSupplier.get().canVectorizeMatcher(columnInspector);
}

@Override
Expand Down Expand Up @@ -176,7 +178,7 @@ public List<String> requiredFields()
{
return ImmutableList.copyOf(
// use a set to get rid of dupes
ImmutableSet.<String>builder().addAll(delegate.requiredFields()).addAll(filter.getRequiredColumns()).build()
ImmutableSet.<String>builder().addAll(delegate.requiredFields()).addAll(filterSupplier.get().getRequiredColumns()).build()
);
}

Expand Down
87 changes: 85 additions & 2 deletions services/src/test/java/org/apache/druid/server/AsyncQueryForwardingServletTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
package org.apache.druid.server;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.InjectableValues;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
Expand Down Expand Up @@ -54,6 +55,13 @@
import org.apache.druid.query.MapQueryToolChestWarehouse;
import org.apache.druid.query.Query;
import org.apache.druid.query.QueryException;
import org.apache.druid.query.aggregation.FilteredAggregatorFactory;
import org.apache.druid.query.aggregation.any.StringAnyAggregatorFactory;
import org.apache.druid.query.filter.SelectorDimFilter;
import org.apache.druid.query.lookup.LookupExtractorFactoryContainer;
import org.apache.druid.query.lookup.LookupExtractorFactoryContainerProvider;
import org.apache.druid.query.lookup.LookupModule;
import org.apache.druid.query.lookup.RegisteredLookupExtractionFn;
import org.apache.druid.query.timeseries.TimeseriesQuery;
import org.apache.druid.segment.TestHelper;
import org.apache.druid.server.initialization.BaseJettyTest;
Expand Down Expand Up @@ -105,9 +113,12 @@
import java.net.URI;
import java.net.URL;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicLong;
import java.util.zip.Deflater;
Expand Down Expand Up @@ -557,6 +568,78 @@ public void testSqlQueryProxyFailure() throws Exception
verifyServletCallsForQuery(query, true, false, hostFinder, properties, true);
}

@Test
public void testNoParseExceptionOnGroupByWithFilteredAggregationOnLookups() throws Exception
{
class TestLookupReferenceManager implements LookupExtractorFactoryContainerProvider
{
@Override
public Set<String> getAllLookupNames()
{
return null;
}

@Override
public Optional<LookupExtractorFactoryContainer> get(String lookupName)
{
return Optional.empty();
}
}

final TimeseriesQuery query =
Druids.newTimeseriesQueryBuilder()
.dataSource("foo")
.intervals("2000/P1D")
.aggregators(
Collections.singletonList(
new FilteredAggregatorFactory(
new StringAnyAggregatorFactory("stringAny", "col", 1024, true),
new SelectorDimFilter(
"test",
"1",
new RegisteredLookupExtractionFn(
new TestLookupReferenceManager(),
"somelookup",
false,
null,
null,
false
)
),
"agg"
)))
.granularity(Granularities.ALL)
.context(ImmutableMap.of("queryId", "dummy"))
.build();

final QueryHostFinder hostFinder = EasyMock.createMock(QueryHostFinder.class);
EasyMock.expect(hostFinder.pickServer(query)).andReturn(new TestServer("http", "1.2.3.4", 9999)).once();
EasyMock.replay(hostFinder);

final ObjectMapper jsonMapper =
TestHelper.makeJsonMapper()
.registerModules(new LookupModule().getJacksonModules())
.setInjectableValues(
new InjectableValues.Std().addValue(
LookupExtractorFactoryContainerProvider.class,
new TestLookupReferenceManager()
)
);
verifyServletCallsForQuery(query, false, false, hostFinder, new Properties(), false, jsonMapper);
}

private void verifyServletCallsForQuery(
Object query,
boolean isNativeSql,
boolean isJDBCSql,
QueryHostFinder hostFinder,
Properties properties,
boolean isFailure
) throws Exception
{
verifyServletCallsForQuery(query, isNativeSql, isJDBCSql, hostFinder, properties, isFailure, TestHelper.makeJsonMapper());
}

/**
* Verifies that the Servlet calls the right methods the right number of times.
*/
Expand All @@ -566,10 +649,10 @@ private void verifyServletCallsForQuery(
boolean isJDBCSql,
QueryHostFinder hostFinder,
Properties properties,
boolean isFailure
boolean isFailure,
ObjectMapper jsonMapper
) throws Exception
{
final ObjectMapper jsonMapper = TestHelper.makeJsonMapper();
final ByteArrayInputStream inputStream = new ByteArrayInputStream(jsonMapper.writeValueAsBytes(query));
final ServletInputStream servletInputStream = new ServletInputStream()
{
Expand Down
Loading