<commit_before>package org.apache.solr.request;

import org.apache.lucene.search.DocIdSet;
import org.apache.lucene.search.DocIdSetIterator;
import org.apache.lucene.search.FieldCache;
import org.apache.lucene.search.Filter;
import org.apache.lucene.util.PriorityQueue;
import org.apache.lucene.util.BytesRef;
import org.apache.solr.common.SolrException;
import org.apache.solr.common.params.FacetParams;
import org.apache.solr.common.util.NamedList;
import org.apache.solr.schema.FieldType;
import org.apache.solr.search.DocSet;
import org.apache.solr.search.SolrIndexReader;
import org.apache.solr.search.SolrIndexSearcher;
import org.apache.solr.util.BoundedTreeSet;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.*;


class PerSegmentSingleValuedFaceting {

  // input params
  SolrIndexSearcher searcher;
  DocSet docs;
  String fieldName;
  int offset;
  int limit;
  int mincount;
  boolean missing;
  String sort;
  String prefix;

  Filter baseSet;

  int nThreads;

  public PerSegmentSingleValuedFaceting(SolrIndexSearcher searcher, DocSet docs, String fieldName, int offset, int limit, int mincount, boolean missing, String sort, String prefix) {
    this.searcher = searcher;
    this.docs = docs;
    this.fieldName = fieldName;
    this.offset = offset;
    this.limit = limit;
    this.mincount = mincount;
    this.missing = missing;
    this.sort = sort;
    this.prefix = prefix;
  }

  public void setNumThreads(int threads) {
    nThreads = threads;
  }


  NamedList getFacetCounts(Executor executor) throws IOException {

    CompletionService<SegFacet> completionService = new ExecutorCompletionService<SegFacet>(executor);

    // reuse the translation logic to go from top level set to per-segment set
    baseSet = docs.getTopFilter();

    SolrIndexReader topReader = searcher.getReader();
    final SolrIndexReader[] leafReaders = topReader.getLeafReaders();
    int[] offsets = topReader.getLeafOffsets();

    // The list of pending tasks that aren't immediately submitted
    // TODO: Is there a completion service, or a delegating executor that can
    // limit the number of concurrent tasks submitted to a bigger executor?
    LinkedList<Callable<SegFacet>> pending = new LinkedList<Callable<SegFacet>>();

    int threads = nThreads <= 0 ? Integer.MAX_VALUE : nThreads;

    for (int i=0; i<leafReaders.length; i++) {
      final SegFacet segFacet = new SegFacet(leafReaders[i], offsets[i]);

      Callable<SegFacet> task = new Callable<SegFacet>() {
        public SegFacet call() throws Exception {
          segFacet.countTerms();
          return segFacet;
        }
      };

      // TODO: if limiting threads, submit by largest segment first?

      if (--threads >= 0) {
        completionService.submit(task);
      } else {
        pending.add(task);
      }
    }


    // now merge the per-segment results
    PriorityQueue<SegFacet> queue = new PriorityQueue<SegFacet>() {
      {
        initialize(leafReaders.length);
      }
      @Override
      protected boolean lessThan(SegFacet a, SegFacet b) {
        return a.si.lookup(a.pos, a.tempBR).compareTo(b.si.lookup(b.pos, b.tempBR)) < 0;
      }
    };


    boolean hasMissingCount=false;
    int missingCount=0;
    for (int i=0; i<leafReaders.length; i++) {
      SegFacet seg = null;

      try {
        Future<SegFacet> future = completionService.take();        
        seg = future.get();
        if (!pending.isEmpty()) {
          completionService.submit(pending.removeFirst());
        }
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, e);
      } catch (ExecutionException e) {
        Throwable cause = e.getCause();
        if (cause instanceof RuntimeException) {
          throw (RuntimeException)cause;
        } else {
          throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, "Error in per-segment faceting on field: " + fieldName, cause);
        }
      }


      if (seg.startTermIndex < seg.endTermIndex) {
        if (seg.startTermIndex==0) {
          hasMissingCount=true;
          missingCount += seg.counts[0];
          seg.pos = 1;
        } else {
          seg.pos = seg.startTermIndex;
        }
        if (seg.pos < seg.endTermIndex) {
          queue.add(seg);
        }
      }
    }

    FacetCollector collector;
    if (sort.equals(FacetParams.FACET_SORT_COUNT) || sort.equals(FacetParams.FACET_SORT_COUNT_LEGACY)) {
      collector = new CountSortedFacetCollector(offset, limit, mincount);
    } else {
      collector = new IndexSortedFacetCollector(offset, limit, mincount);
    }

    final BytesRef tempBR = new BytesRef();

    while (queue.size() > 0) {
      SegFacet seg = queue.top();
      BytesRef val = seg.si.lookup(seg.pos, tempBR);
      int count = 0;

      do {
        count += seg.counts[seg.pos - seg.startTermIndex];

        // TODO: OPTIMIZATION...
        // if mincount>0 then seg.pos++ can skip ahead to the next non-zero entry.
        seg.pos++;
        if (seg.pos >= seg.endTermIndex) {
          queue.pop();
          seg = queue.top();
        }  else {
          seg = queue.updateTop();
        }
      } while (seg != null && val.compareTo(seg.si.lookup(seg.pos, seg.tempBR)) == 0);

      boolean stop = collector.collect(val, count);
      if (stop) break;
    }

    NamedList res = collector.getFacetCounts();

    // convert labels to readable form    
    FieldType ft = searcher.getSchema().getFieldType(fieldName);
    int sz = res.size();
    for (int i=0; i<sz; i++) {
      res.setName(i, ft.indexedToReadable(res.getName(i)));
    }

    if (missing) {
      if (!hasMissingCount) {
        missingCount = SimpleFacets.getFieldMissingCount(searcher,docs,fieldName);
      }
      res.add(null, missingCount);
    }<commit_msg>LUCENE-2380: work on faceting speedups after fieldcache conversion - store BytesRef per seg instead of looking it up each comparison<commit_after>package org.apache.solr.request;

import org.apache.lucene.search.DocIdSet;
import org.apache.lucene.search.DocIdSetIterator;
import org.apache.lucene.search.FieldCache;
import org.apache.lucene.search.Filter;
import org.apache.lucene.util.PriorityQueue;
import org.apache.lucene.util.BytesRef;
import org.apache.solr.common.SolrException;
import org.apache.solr.common.params.FacetParams;
import org.apache.solr.common.util.NamedList;
import org.apache.solr.schema.FieldType;
import org.apache.solr.search.DocSet;
import org.apache.solr.search.SolrIndexReader;
import org.apache.solr.search.SolrIndexSearcher;
import org.apache.solr.util.BoundedTreeSet;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.*;


class PerSegmentSingleValuedFaceting {

  // input params
  SolrIndexSearcher searcher;
  DocSet docs;
  String fieldName;
  int offset;
  int limit;
  int mincount;
  boolean missing;
  String sort;
  String prefix;

  Filter baseSet;

  int nThreads;

  public PerSegmentSingleValuedFaceting(SolrIndexSearcher searcher, DocSet docs, String fieldName, int offset, int limit, int mincount, boolean missing, String sort, String prefix) {
    this.searcher = searcher;
    this.docs = docs;
    this.fieldName = fieldName;
    this.offset = offset;
    this.limit = limit;
    this.mincount = mincount;
    this.missing = missing;
    this.sort = sort;
    this.prefix = prefix;
  }

  public void setNumThreads(int threads) {
    nThreads = threads;
  }


  NamedList getFacetCounts(Executor executor) throws IOException {

    CompletionService<SegFacet> completionService = new ExecutorCompletionService<SegFacet>(executor);

    // reuse the translation logic to go from top level set to per-segment set
    baseSet = docs.getTopFilter();

    SolrIndexReader topReader = searcher.getReader();
    final SolrIndexReader[] leafReaders = topReader.getLeafReaders();
    int[] offsets = topReader.getLeafOffsets();

    // The list of pending tasks that aren't immediately submitted
    // TODO: Is there a completion service, or a delegating executor that can
    // limit the number of concurrent tasks submitted to a bigger executor?
    LinkedList<Callable<SegFacet>> pending = new LinkedList<Callable<SegFacet>>();

    int threads = nThreads <= 0 ? Integer.MAX_VALUE : nThreads;

    for (int i=0; i<leafReaders.length; i++) {
      final SegFacet segFacet = new SegFacet(leafReaders[i], offsets[i]);

      Callable<SegFacet> task = new Callable<SegFacet>() {
        public SegFacet call() throws Exception {
          segFacet.countTerms();
          return segFacet;
        }
      };

      // TODO: if limiting threads, submit by largest segment first?

      if (--threads >= 0) {
        completionService.submit(task);
      } else {
        pending.add(task);
      }
    }


    // now merge the per-segment results
    PriorityQueue<SegFacet> queue = new PriorityQueue<SegFacet>() {
      {
        initialize(leafReaders.length);
      }
      @Override
      protected boolean lessThan(SegFacet a, SegFacet b) {
        return a.tempBR.compareTo(b.tempBR) < 0;
      }
    };


    boolean hasMissingCount=false;
    int missingCount=0;
    for (int i=0; i<leafReaders.length; i++) {
      SegFacet seg = null;

      try {
        Future<SegFacet> future = completionService.take();        
        seg = future.get();
        if (!pending.isEmpty()) {
          completionService.submit(pending.removeFirst());
        }
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
        throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, e);
      } catch (ExecutionException e) {
        Throwable cause = e.getCause();
        if (cause instanceof RuntimeException) {
          throw (RuntimeException)cause;
        } else {
          throw new SolrException(SolrException.ErrorCode.SERVER_ERROR, "Error in per-segment faceting on field: " + fieldName, cause);
        }
      }


      if (seg.startTermIndex < seg.endTermIndex) {
        if (seg.startTermIndex==0) {
          hasMissingCount=true;
          missingCount += seg.counts[0];
          seg.pos = 1;
        } else {
          seg.pos = seg.startTermIndex;
        }
        if (seg.pos < seg.endTermIndex) {
          seg.si.lookup(seg.pos, seg.tempBR);
          queue.add(seg);
        }
      }
    }

    FacetCollector collector;
    if (sort.equals(FacetParams.FACET_SORT_COUNT) || sort.equals(FacetParams.FACET_SORT_COUNT_LEGACY)) {
      collector = new CountSortedFacetCollector(offset, limit, mincount);
    } else {
      collector = new IndexSortedFacetCollector(offset, limit, mincount);
    }

    BytesRef val = new BytesRef();

    while (queue.size() > 0) {
      SegFacet seg = queue.top();

      // make a shallow copy
      // Is this always safe? Or could the byte[] be changed?
      val.bytes = seg.tempBR.bytes;
      val.offset = seg.tempBR.offset;
      val.length = seg.tempBR.length;

      int count = 0;

      do {
        count += seg.counts[seg.pos - seg.startTermIndex];

        // TODO: OPTIMIZATION...
        // if mincount>0 then seg.pos++ can skip ahead to the next non-zero entry.
        seg.pos++;
        if (seg.pos >= seg.endTermIndex) {
          queue.pop();
          seg = queue.top();
        }  else {
          seg.si.lookup(seg.pos, seg.tempBR);          
          seg = queue.updateTop();
        }
      } while (seg != null && val.compareTo(seg.tempBR) == 0);

      boolean stop = collector.collect(val, count);
      if (stop) break;
    }

    NamedList res = collector.getFacetCounts();

    // convert labels to readable form    
    FieldType ft = searcher.getSchema().getFieldType(fieldName);
    int sz = res.size();
    for (int i=0; i<sz; i++) {
      res.setName(i, ft.indexedToReadable(res.getName(i)));
    }

    if (missing) {
      if (!hasMissingCount) {
        missingCount = SimpleFacets.getFieldMissingCount(searcher,docs,fieldName);
      }
      res.add(null, missingCount);
    }
{'fraction_non_alphanumeric': 0.11039176626826029, 'fraction_numerical': 0.0026560424966799467, 'mean_word_length': 2.7712050078247263, 'pattern_counts': {'":': 0, '<': 39, '<?xml version=': 0, '>': 33, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7251726', 'n_tokens_mistral': 3797, 'n_tokens_neox': 3666, 'n_words': 864}