Skip to content

Commit

Permalink
Invoke postCollection on aggregation collectors.
Browse files Browse the repository at this point in the history 
Also cleanup how facet and aggs collector are used inside the QueryCollector

Closes #5387
  • Loading branch information
@martijnvg
martijnvg committed Mar 13, 2014
1 parent 44ecb3f commit 9af3664
Show file tree
Hide file tree
Showing 2 changed files with 19 additions and 29 deletions.
2 changes: 1 addition & 1 deletion src/main/java/org/elasticsearch/percolator/PercolatorService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -777,7 +777,7 @@ private void queryBasedPercolating(Engine.Searcher percolatorSearcher, Percolate
FilteredQuery query = new FilteredQuery(context.percolateQuery(), percolatorTypeFilter);
percolatorSearcher.searcher().search(query, percolateCollector);

for (Collector queryCollector : percolateCollector.facetCollectors) {
for (Collector queryCollector : percolateCollector.facetAndAggregatorCollector) {
if (queryCollector instanceof XCollector) {
((XCollector) queryCollector).postCollection();
}
Expand Down
46 changes: 18 additions & 28 deletions src/main/java/org/elasticsearch/percolator/QueryCollector.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@
package org.elasticsearch.percolator;

import com.carrotsearch.hppc.FloatArrayList;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import org.apache.lucene.index.AtomicReaderContext;
import org.apache.lucene.search.*;
Expand Down Expand Up @@ -61,8 +62,7 @@ abstract class QueryCollector extends Collector {

BytesValues values;

final List<Collector> facetCollectors = new ArrayList<Collector>();
final Collector facetAndAggregatorCollector;
final List<Collector> facetAndAggregatorCollector;

QueryCollector(ESLogger logger, PercolateContext context) {
this.logger = logger;
Expand All @@ -71,6 +71,7 @@ abstract class QueryCollector extends Collector {
final FieldMapper<?> idMapper = context.mapperService().smartNameFieldMapper(IdFieldMapper.NAME);
this.idFieldData = context.fieldData().getForField(idMapper);

ImmutableList.Builder<Collector> facetAggCollectorBuilder = ImmutableList.builder();
if (context.facets() != null) {
for (SearchContextFacets.Entry entry : context.facets().entries()) {
if (entry.isGlobal()) {
Expand All @@ -84,11 +85,10 @@ abstract class QueryCollector extends Collector {
collector = new FilteredCollector(collector, entry.getFilter());
}
}
facetCollectors.add(collector);
facetAggCollectorBuilder.add(collector);
}
}

List<Collector> collectors = new ArrayList<Collector>(facetCollectors);
if (context.aggregations() != null) {
AggregationContext aggregationContext = new AggregationContext(context);
context.aggregations().aggregationContext(aggregationContext);
Expand All @@ -105,33 +105,31 @@ abstract class QueryCollector extends Collector {
}
context.aggregations().aggregators(aggregators);
if (!aggregatorCollectors.isEmpty()) {
collectors.add(new AggregationPhase.AggregationsCollector(aggregatorCollectors, aggregationContext));
facetAggCollectorBuilder.add(new AggregationPhase.AggregationsCollector(aggregatorCollectors, aggregationContext));
}
}
facetAndAggregatorCollector = facetAggCollectorBuilder.build();
}

int size = collectors.size();
if (size == 0) {
facetAndAggregatorCollector = null;
} else if (size == 1) {
facetAndAggregatorCollector = collectors.get(0);
} else {
facetAndAggregatorCollector = MultiCollector.wrap(collectors.toArray(new Collector[collectors.size()]));
public void postMatch(int doc) throws IOException {
for (Collector collector : facetAndAggregatorCollector) {
collector.collect(doc);
}
}

@Override
public void setScorer(Scorer scorer) throws IOException {
if (facetAndAggregatorCollector != null) {
facetAndAggregatorCollector.setScorer(scorer);
for (Collector collector : facetAndAggregatorCollector) {
collector.setScorer(scorer);
}
}

@Override
public void setNextReader(AtomicReaderContext context) throws IOException {
// we use the UID because id might not be indexed
values = idFieldData.load(context).getBytesValues(true);
if (facetAndAggregatorCollector != null) {
facetAndAggregatorCollector.setNextReader(context);
for (Collector collector : facetAndAggregatorCollector) {
collector.setNextReader(context);
}
}

Expand Down Expand Up @@ -214,9 +212,7 @@ public void collect(int doc) throws IOException {
}
}
counter++;
if (facetAndAggregatorCollector != null) {
facetAndAggregatorCollector.collect(doc);
}
postMatch(doc);
}
} catch (IOException e) {
logger.warn("[" + spare.bytes.utf8ToString() + "] failed to execute query", e);
Expand Down Expand Up @@ -259,9 +255,7 @@ public void collect(int doc) throws IOException {
searcher.search(query, collector);
if (collector.exists()) {
topDocsCollector.collect(doc);
if (facetAndAggregatorCollector != null) {
facetAndAggregatorCollector.collect(doc);
}
postMatch(doc);
}
} catch (IOException e) {
logger.warn("[" + spare.bytes.utf8ToString() + "] failed to execute query", e);
Expand Down Expand Up @@ -333,9 +327,7 @@ public void collect(int doc) throws IOException {
}
}
counter++;
if (facetAndAggregatorCollector != null) {
facetAndAggregatorCollector.collect(doc);
}
postMatch(doc);
}
} catch (IOException e) {
logger.warn("[" + spare.bytes.utf8ToString() + "] failed to execute query", e);
Expand Down Expand Up @@ -385,9 +377,7 @@ public void collect(int doc) throws IOException {
searcher.search(query, collector);
if (collector.exists()) {
counter++;
if (facetAndAggregatorCollector != null) {
facetAndAggregatorCollector.collect(doc);
}
postMatch(doc);
}
} catch (IOException e) {
logger.warn("[" + spare.bytes.utf8ToString() + "] failed to execute query", e);
Expand Down

0 comments on commit 9af3664

Please sign in to comment.