/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.solr.client.solrj.io.stream;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Optional;
import java.util.stream.Collectors;

import org.apache.solr.client.solrj.SolrRequest;
import org.apache.solr.client.solrj.impl.CloudSolrClient;
import org.apache.solr.client.solrj.impl.CloudSolrClient.Builder;
import org.apache.solr.client.solrj.impl.ClusterStateProvider;
import org.apache.solr.client.solrj.io.SolrClientCache;
import org.apache.solr.client.solrj.io.Tuple;
import org.apache.solr.client.solrj.io.comp.ComparatorOrder;
import org.apache.solr.client.solrj.io.comp.FieldComparator;
import org.apache.solr.client.solrj.io.comp.MultipleFieldComparator;
import org.apache.solr.client.solrj.io.comp.StreamComparator;
import org.apache.solr.client.solrj.io.stream.expr.Explanation;
import org.apache.solr.client.solrj.io.stream.expr.Explanation.ExpressionType;
import org.apache.solr.client.solrj.io.stream.expr.Expressible;
import org.apache.solr.client.solrj.io.stream.expr.StreamExplanation;
import org.apache.solr.client.solrj.io.stream.expr.StreamExpression;
import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionNamedParameter;
import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionParameter;
import org.apache.solr.client.solrj.io.stream.expr.StreamExpressionValue;
import org.apache.solr.client.solrj.io.stream.expr.StreamFactory;
import org.apache.solr.client.solrj.io.stream.metrics.Bucket;
import org.apache.solr.client.solrj.io.stream.metrics.CountMetric;
import org.apache.solr.client.solrj.io.stream.metrics.Metric;
import org.apache.solr.client.solrj.request.QueryRequest;
import org.apache.solr.common.params.ModifiableSolrParams;
import org.apache.solr.common.params.SolrParams;
import org.apache.solr.common.util.NamedList;

/**
 *  The FacetStream abstracts the output from the JSON facet API as a Stream of Tuples. This provides an alternative to the
 *  RollupStream which uses Map/Reduce to perform aggregations.
 * @since 6.0.0
 **/

public class FacetStream extends TupleStream implements Expressible, ParallelMetricsRollup {

  private static final long serialVersionUID = 1;

  // allow client apps to disable the auto-plist via system property if they want to turn it off globally
  private static final boolean defaultTieredEnabled =
      Boolean.parseBoolean(System.getProperty("solr.facet.stream.tiered", "false"));

  static final String TIERED_PARAM = "tiered";

  private Bucket[] buckets;
  private Metric[] metrics;
  private int rows;
  private int offset;
  private int overfetch;
  private int bucketSizeLimit;
  private boolean refine;
  private String method;
  private FieldComparator[] bucketSorts;
  private List<Tuple> tuples = new ArrayList<Tuple>();
  private int index;
  private String zkHost;
  private ModifiableSolrParams params;
  private String collection;
  private boolean resortNeeded;
  private boolean serializeBucketSizeLimit;

  protected transient SolrClientCache cache;
  protected transient CloudSolrClient cloudSolrClient;
  protected transient TupleStream parallelizedStream;
  protected transient StreamContext context;

  public FacetStream(String zkHost,
                     String collection,
                     SolrParams params,
                     Bucket[] buckets,
                     Metric[] metrics,
                     FieldComparator[] bucketSorts,
                     int bucketSizeLimit) throws IOException {

    if(bucketSizeLimit == -1) {
      bucketSizeLimit = Integer.MAX_VALUE;
    }
    init(collection, params, buckets, bucketSorts, metrics, bucketSizeLimit,0, bucketSizeLimit, false, null, true, 0, zkHost);
  }
  
  public FacetStream(StreamExpression expression, StreamFactory factory) throws IOException{   
    // grab all parameters out
    String collectionName = factory.getValueOperand(expression, 0);

    if(collectionName.indexOf('"') > -1) {
      collectionName = collectionName.replaceAll("\"", "").replaceAll(" ", "");
    }

    List<StreamExpressionNamedParameter> namedParams = factory.getNamedOperands(expression);
    StreamExpressionNamedParameter bucketExpression = factory.getNamedOperand(expression, "buckets");
    StreamExpressionNamedParameter bucketSortExpression = factory.getNamedOperand(expression, "bucketSorts");
    List<StreamExpression> metricExpressions = factory.getExpressionOperandsRepresentingTypes(expression, Expressible.class, Metric.class);
    StreamExpressionNamedParameter bucketLimitExpression = factory.getNamedOperand(expression, "bucketSizeLimit");
    StreamExpressionNamedParameter zkHostExpression = factory.getNamedOperand(expression, "zkHost");
    StreamExpressionNamedParameter rowsExpression = factory.getNamedOperand(expression, "rows");
    StreamExpressionNamedParameter offsetExpression = factory.getNamedOperand(expression, "offset");
    StreamExpressionNamedParameter overfetchExpression = factory.getNamedOperand(expression, "overfetch");
    StreamExpressionNamedParameter refineExpression = factory.getNamedOperand(expression, "refine");
    StreamExpressionNamedParameter methodExpression = factory.getNamedOperand(expression, "method");

    // Validate there are no unknown parameters
    if(expression.getParameters().size() != 1 + namedParams.size() + metricExpressions.size()){
      throw new IOException(String.format(Locale.ROOT,"invalid expression %s - unknown operands found",expression));
    }
    
    // Collection Name
    if(null == collectionName){
      throw new IOException(String.format(Locale.ROOT,"invalid expression %s - collectionName expected as first operand",expression));
    }
        
    // Named parameters - passed directly to solr as solrparams
    if(0 == namedParams.size()){
      throw new IOException(String.format(Locale.ROOT,"invalid expression %s - at least one named parameter expected. eg. 'q=*:*'",expression));
    }
    
    // pull out known named params
    ModifiableSolrParams params = new ModifiableSolrParams();
    for(StreamExpressionNamedParameter namedParam : namedParams){
      if(!namedParam.getName().equals("zkHost") &&
          !namedParam.getName().equals("buckets") &&
          !namedParam.getName().equals("bucketSorts") &&
          !namedParam.getName().equals("bucketSizeLimit") &&
          !namedParam.getName().equals("method") &&
          !namedParam.getName().equals("offset") &&
          !namedParam.getName().equals("rows") &&
          !namedParam.getName().equals("refine") &&
          !namedParam.getName().equals("overfetch")){
        params.add(namedParam.getName(), namedParam.getParameter().toString().trim());
      }
    }

    if(params.get("q") == null) {
      params.set("q", "*:*");
    }

    // buckets, required - comma separated
    Bucket[] buckets = null;
    if(null != bucketExpression){
      if(bucketExpression.getParameter() instanceof StreamExpressionValue){
        String[] keys = ((StreamExpressionValue)bucketExpression.getParameter()).getValue().split(",");
        if(0 != keys.length){
          buckets = new Bucket[keys.length];
          for(int idx = 0; idx < keys.length; ++idx){
            buckets[idx] = new Bucket(keys[idx].trim());
          }
        }
      }
    }

    if(null == buckets){      
      throw new IOException(String.format(Locale.ROOT,"invalid expression %s - at least one bucket expected. eg. 'buckets=\"name\"'",expression,collectionName));
    }


    // Construct the metrics
    Metric[] metrics = new Metric[metricExpressions.size()];
    for(int idx = 0; idx < metricExpressions.size(); ++idx) {
      metrics[idx] = factory.constructMetric(metricExpressions.get(idx));
    }

    if(metrics.length == 0) {
      metrics = new Metric[1];
      metrics[0] = new CountMetric();
    }

    String bucketSortString = null;

    if(bucketSortExpression == null) {
      bucketSortString = metrics[0].getIdentifier()+" desc";
    } else {
      bucketSortString = ((StreamExpressionValue)bucketSortExpression.getParameter()).getValue();
      if(bucketSortString.contains("(") &&
          metricExpressions.size() == 0 &&
          (!bucketSortExpression.equals("count(*) desc") &&
           !bucketSortExpression.equals("count(*) asc"))) {
      //Attempting bucket sort on a metric that is not going to be calculated.
        throw new IOException(String.format(Locale.ROOT,"invalid expression %s - the bucketSort is being performed on a metric that is not being calculated.",expression,collectionName));
      }
    }

    FieldComparator[] bucketSorts = parseBucketSorts(bucketSortString, buckets);

    if(null == bucketSorts || 0 == bucketSorts.length) {
      throw new IOException(String.format(Locale.ROOT,"invalid expression %s - at least one bucket sort expected. eg. 'bucketSorts=\"name asc\"'",expression,collectionName));
    }


    boolean refine = false;

    if(refineExpression != null) {
      String refineStr = ((StreamExpressionValue) refineExpression.getParameter()).getValue();
      if (refineStr != null) {
        refine = Boolean.parseBoolean(refineStr);
      }
    }

    if(bucketLimitExpression != null && (rowsExpression != null ||
                                         offsetExpression != null ||
                                         overfetchExpression != null)) {
      throw new IOException("bucketSizeLimit is incompatible with rows, offset and overfetch.");
    }

    String methodStr = null;
    if(methodExpression != null) {
      methodStr = ((StreamExpressionValue) methodExpression.getParameter()).getValue();
    }

    int overfetchInt = 250;
    if(overfetchExpression != null) {
      String overfetchStr = ((StreamExpressionValue) overfetchExpression.getParameter()).getValue();
      overfetchInt = Integer.parseInt(overfetchStr);
    }

    int offsetInt = 0;
    if(offsetExpression != null) {
      String offsetStr = ((StreamExpressionValue) offsetExpression.getParameter()).getValue();
      offsetInt = Integer.parseInt(offsetStr);
    }

    int rowsInt = Integer.MIN_VALUE;
    int bucketLimit = Integer.MIN_VALUE;
    boolean bucketLimitSet = false;

    if(null != rowsExpression) {
      String rowsStr = ((StreamExpressionValue)rowsExpression.getParameter()).getValue();
      try {
        rowsInt = Integer.parseInt(rowsStr);
        if (rowsInt <= 0 && rowsInt != -1) {
          throw new IOException(String.format(Locale.ROOT, "invalid expression %s - limit '%s' must be greater than 0 or -1.", expression, rowsStr));
        }
        //Rows is set so configure the bucketLimitSize
        if(rowsInt == -1) {
          bucketLimit = rowsInt = Integer.MAX_VALUE;
        } else if(overfetchInt == -1) {
          bucketLimit = Integer.MAX_VALUE;
        }else{
          bucketLimit = offsetInt+overfetchInt+rowsInt;
        }
      } catch (NumberFormatException e) {
        throw new IOException(String.format(Locale.ROOT, "invalid expression %s - limit '%s' is not a valid integer.", expression, rowsStr));
      }
    }

    if(bucketLimitExpression != null) {
      String bucketLimitStr = ((StreamExpressionValue) bucketLimitExpression.getParameter()).getValue();
      try {
        bucketLimit = Integer.parseInt(bucketLimitStr);
        bucketLimitSet = true;

        if (bucketLimit <= 0 && bucketLimit != -1) {
          throw new IOException(String.format(Locale.ROOT, "invalid expression %s - bucketSizeLimit '%s' must be greater than 0 or -1.", expression, bucketLimitStr));
        }

        // Bucket limit is set. So set rows.
        if(bucketLimit == -1) {
         rowsInt = bucketLimit = Integer.MAX_VALUE;
        } else {
          rowsInt = bucketLimit;
        }
      }  catch (NumberFormatException e) {
        throw new IOException(String.format(Locale.ROOT, "invalid expression %s - bucketSizeLimit '%s' is not a valid integer.", expression, bucketLimitStr));
      }
    }

    if(rowsExpression == null && bucketLimitExpression == null) {
      rowsInt = 10;
      if(overfetchInt == -1) {
        bucketLimit = Integer.MAX_VALUE;
      }else{
        bucketLimit = offsetInt+overfetchInt+rowsInt;
      }
    }

    // zkHost, optional - if not provided then will look into factory list to get
    String zkHost = null;
    if(null == zkHostExpression){
      zkHost = factory.getCollectionZkHost(collectionName);
      if(zkHost == null) {
        zkHost = factory.getDefaultZkHost();
      }
    } else if(zkHostExpression.getParameter() instanceof StreamExpressionValue) {
      zkHost = ((StreamExpressionValue)zkHostExpression.getParameter()).getValue();
    }

    if(null == zkHost){
      throw new IOException(String.format(Locale.ROOT,"invalid expression %s - zkHost not found for collection '%s'",expression,collectionName));
    }
    
    // We've got all the required items
    init(collectionName,
         params,
         buckets,
         bucketSorts,
         metrics,
         rowsInt,
         offsetInt,
         bucketLimit,
         refine,
         methodStr,
         bucketLimitSet,
         overfetchInt,
         zkHost);
  }

  // see usage in parallelize method
  private FacetStream() {
  }

  public int getBucketSizeLimit() {
    return this.bucketSizeLimit;
  }

  public int getRows() {
    return this.rows;
  }

  public int getOffset() {
    return this.offset;
  }

  public int getOverfetch() {
    return this.overfetch;
  }

  public Bucket[] getBuckets() {
    return this.buckets;
  }

  public String getCollection() {
    return this.collection;
  }

  private FieldComparator[] parseBucketSorts(String bucketSortString, Bucket[] buckets) throws IOException {

    String[] sorts = parseSorts(bucketSortString);

    FieldComparator[] comps = new FieldComparator[sorts.length];
    for(int i=0; i<sorts.length; i++) {
      String s = sorts[i];

      String fieldName = null;
      String order = null;

      if(s.endsWith("asc") || s.endsWith("ASC")) {
        order = "asc";
        fieldName = s.substring(0, s.length()-3).trim().replace(" ", "");
      } else if(s.endsWith("desc") || s.endsWith("DESC")) {
        order = "desc";
        fieldName = s.substring(0, s.length()-4).trim().replace(" ", "");
      } else {
        throw new IOException(String.format(Locale.ROOT,"invalid expression - bad bucketSort '%s'.",bucketSortString));
      }
            
      comps[i] = new FieldComparator(fieldName, order.equalsIgnoreCase("asc") ? ComparatorOrder.ASCENDING : ComparatorOrder.DESCENDING);
    }

    return comps;
  }

  private String[] parseSorts(String sortString) {
    List<String> sorts = new ArrayList<>();
    boolean inParam = false;
    StringBuilder buff = new StringBuilder();
    for(int i=0; i<sortString.length(); i++) {
      char c = sortString.charAt(i);
      if(c == '(') {
        inParam=true;
        buff.append(c);
      } else if (c == ')') {
        inParam = false;
        buff.append(c);
      } else if (c == ',' && !inParam) {
        sorts.add(buff.toString().trim());
        buff = new StringBuilder();
      } else {
        buff.append(c);
      }
    }

    if(buff.length() > 0) {
      sorts.add(buff.toString());
    }

    return sorts.toArray(new String[sorts.size()]);
  }


  private void init(String collection, SolrParams params, Bucket[] buckets, FieldComparator[] bucketSorts, Metric[] metrics, int rows, int offset, int bucketSizeLimit, boolean refine, String method, boolean serializeBucketSizeLimit, int overfetch, String zkHost) throws IOException {
    this.zkHost  = zkHost;
    this.params = new ModifiableSolrParams(params);
    this.buckets = buckets;
    this.metrics = metrics;
    this.rows = rows;
    this.offset = offset;
    this.refine = refine;
    this.bucketSizeLimit   = bucketSizeLimit;
    this.collection = collection;
    this.bucketSorts = bucketSorts;
    this.method = method;
    this.serializeBucketSizeLimit = serializeBucketSizeLimit;
    this.overfetch = overfetch;
    
    // In a facet world it only makes sense to have the same field name in all of the sorters
    // Because FieldComparator allows for left and right field names we will need to validate
    // that they are the same
    for(FieldComparator sort : bucketSorts){
      if(sort.hasDifferentFieldNames()){
        throw new IOException("Invalid FacetStream - all sorts must be constructed with a single field name.");
      }
    }
  }
  
  @Override
  public StreamExpressionParameter toExpression(StreamFactory factory) throws IOException {    
    // function name
    StreamExpression expression = new StreamExpression(factory.getFunctionName(this.getClass()));
    
    // collection
    if(collection.indexOf(',') > -1) {
      expression.addParameter("\""+collection+"\"");
    } else {
      expression.addParameter(collection);
    }
    
    // parameters

    for (Entry<String, String[]> param : params.getMap().entrySet()) {
      for (String val : param.getValue()) {
        expression.addParameter(new StreamExpressionNamedParameter(param.getKey(), val));
      }
    }
    
    // buckets
    {
      StringBuilder builder = new StringBuilder();
      for(Bucket bucket : buckets){
        if(0 != builder.length()){ builder.append(","); }
        builder.append(bucket.toString());
      }
      expression.addParameter(new StreamExpressionNamedParameter("buckets", builder.toString()));
    }
    
    // bucketSorts
    {
      StringBuilder builder = new StringBuilder();
      for(FieldComparator sort : bucketSorts){
        if(0 != builder.length()){ builder.append(","); }
        builder.append(sort.toExpression(factory));
      }
      expression.addParameter(new StreamExpressionNamedParameter("bucketSorts", builder.toString()));
    }
    
    // metrics
    for(Metric metric : metrics){
      expression.addParameter(metric.toExpression(factory));
    }
    
    if(serializeBucketSizeLimit) {
      if(bucketSizeLimit == Integer.MAX_VALUE) {
        expression.addParameter(new StreamExpressionNamedParameter("bucketSizeLimit", Integer.toString(-1)));
      } else {
        expression.addParameter(new StreamExpressionNamedParameter("bucketSizeLimit", Integer.toString(bucketSizeLimit)));
      }
    } else {
      if (rows == Integer.MAX_VALUE) {
        expression.addParameter(new StreamExpressionNamedParameter("rows", Integer.toString(-1)));
      } else{
        expression.addParameter(new StreamExpressionNamedParameter("rows", Integer.toString(rows)));
      }

      expression.addParameter(new StreamExpressionNamedParameter("offset", Integer.toString(offset)));

      if(overfetch == Integer.MAX_VALUE) {
        expression.addParameter(new StreamExpressionNamedParameter("overfetch", Integer.toString(-1)));
      } else {
        expression.addParameter(new StreamExpressionNamedParameter("overfetch", Integer.toString(overfetch)));
      }
    }

    if(method != null) {
      expression.addParameter(new StreamExpressionNamedParameter("method", this.method));
    }
        
    // zkHost
    expression.addParameter(new StreamExpressionNamedParameter("zkHost", zkHost));
        
    return expression;   
  }

  @Override
  public Explanation toExplanation(StreamFactory factory) throws IOException {

    StreamExplanation explanation = new StreamExplanation(getStreamNodeId().toString());
    
    explanation.setFunctionName(factory.getFunctionName(this.getClass()));
    explanation.setImplementingClass(this.getClass().getName());
    explanation.setExpressionType(ExpressionType.STREAM_SOURCE);
    explanation.setExpression(toExpression(factory).toString());
    
    // child is a datastore so add it at this point
    StreamExplanation child = new StreamExplanation(getStreamNodeId() + "-datastore");
    child.setFunctionName(String.format(Locale.ROOT, "solr (%s)", collection)); 
    // TODO: fix this so we know the # of workers - check with Joel about a Topic's ability to be in a
    // parallel stream.
    
    child.setImplementingClass("Solr/Lucene");
    child.setExpressionType(ExpressionType.DATASTORE);

    child.setExpression(params.stream().map(e -> String.format(Locale.ROOT, "%s=%s", e.getKey(), Arrays.toString(e.getValue()))).collect(Collectors.joining(",")));
    
    explanation.addChild(child);
    
    return explanation;
  }
  
  public void setStreamContext(StreamContext context) {
    this.context = context;
    cache = context.getSolrClientCache();
  }

  public List<TupleStream> children() {
    return new ArrayList<>();
  }

  public void open() throws IOException {
    if(cache != null) {
      cloudSolrClient = cache.getCloudSolrClient(zkHost);
    } else {
      final List<String> hosts = new ArrayList<>();
      hosts.add(zkHost);
      cloudSolrClient = new Builder(hosts, Optional.empty()).withSocketTimeout(30000).withConnectionTimeout(15000).build();
    }

    // Parallelize the facet expression across multiple collections for an alias using plist if possible
    if (params.getBool(TIERED_PARAM, defaultTieredEnabled)) {
      ClusterStateProvider clusterStateProvider = cloudSolrClient.getClusterStateProvider();
      final List<String> resolved = clusterStateProvider != null ? clusterStateProvider.resolveAlias(collection) : null;
      if (resolved != null && resolved.size() > 1) {
        Optional<TupleStream> maybeParallelize = openParallelStream(context, resolved, metrics);
        if (maybeParallelize.isPresent()) {
          this.parallelizedStream = maybeParallelize.get();
          return; // we're using a plist to parallelize the facet operation
        } // else, there's a metric that we can't rollup over the plist results safely ... no plist for you!
      }
    }

    FieldComparator[] adjustedSorts = adjustSorts(buckets, bucketSorts);
    this.resortNeeded = resortNeeded(adjustedSorts);

    String json = getJsonFacetString(buckets, metrics, adjustedSorts, method, refine, bucketSizeLimit);
    assert expectedJson(json);
    ModifiableSolrParams paramsLoc = new ModifiableSolrParams(params);
    paramsLoc.set("json.facet", json);
    paramsLoc.set("rows", "0");

    QueryRequest request = new QueryRequest(paramsLoc, SolrRequest.METHOD.POST);
    if (paramsLoc.get("lb.proxy") != null) {
      request.setPath("/"+collection+"/select");
    }

    try {
      @SuppressWarnings({"rawtypes"})
      NamedList response = cloudSolrClient.request(request, collection);
      getTuples(response, buckets, metrics);

      if(resortNeeded) {
        Collections.sort(tuples, getStreamSort());
      }

      index=this.offset;
    } catch (Exception e) {
      throw new IOException(e);
    }
  }

  private boolean expectedJson(String json) {
    if(this.method != null) {
      if(!json.contains("\"method\":\""+this.method+"\"")) {
        return false;
      }
    }

    if(this.refine) {
      if(!json.contains("\"refine\":true")) {
        return false;
      }
    }

    if(serializeBucketSizeLimit) {
      if(!json.contains("\"limit\":"+bucketSizeLimit)) {
        return false;
      }
    } else {
      if(!json.contains("\"limit\":"+(this.rows+this.offset+this.overfetch))) {
        return false;
      }
    }

    for(Bucket bucket : buckets) {
      if(!json.contains("\""+bucket.toString()+"\":")) {
        return false;
      }
    }

    for(Metric metric: metrics) {
      String func = metric.getFunctionName();
      if(!func.equals("count") && !func.equals("per") && !func.equals("std")) {
        if (!json.contains(metric.getIdentifier())) {
          return false;
        }
      }
    }

    return true;
  }

  public void close() throws IOException {
    if(cache == null) {
      if (cloudSolrClient != null) {
        cloudSolrClient.close();
      }
    }
  }

  public Tuple read() throws IOException {
    // if we're parallelizing the facet expression over multiple collections with plist,
    // then delegate the read operation to that stream instead
    if (parallelizedStream != null) {
      return parallelizedStream.read();
    }

    if(index < tuples.size() && index < (offset+rows)) {
      Tuple tuple = tuples.get(index);
      ++index;
      return tuple;
    } else {
      Tuple tuple = Tuple.EOF();

      if(bucketSizeLimit == Integer.MAX_VALUE) {
        tuple.put("totalRows", tuples.size());
      }
      return tuple;
    }
  }

  private String getJsonFacetString(Bucket[] _buckets,
                                    Metric[] _metrics,
                                    FieldComparator[] _sorts,
                                    String _method,
                                    boolean _refine,
                                    int _limit) {
    StringBuilder buf = new StringBuilder();
    appendJson(buf, _buckets, _metrics, _sorts, _limit, _method, _refine,0);
    return "{"+buf.toString()+"}";
  }

  private FieldComparator[] adjustSorts(Bucket[] _buckets, FieldComparator[] _sorts) throws IOException {
    if(_buckets.length == _sorts.length) {
      return _sorts;
    } else if(_sorts.length == 1) {
      FieldComparator[] adjustedSorts = new FieldComparator[_buckets.length];
      if (_sorts[0].getLeftFieldName().contains("(")) {
        //Its a metric sort so apply the same sort criteria at each level.
        for (int i = 0; i < adjustedSorts.length; i++) {
          adjustedSorts[i] = _sorts[0];
        }
      } else {
        //Its an index sort so apply an index sort at each level.
        for (int i = 0; i < adjustedSorts.length; i++) {
          adjustedSorts[i] = new FieldComparator(_buckets[i].toString(), _sorts[0].getOrder());
        }
      }
      return adjustedSorts;
    } else {
      throw new IOException("If multiple sorts are specified there must be a sort for each bucket.");
    }
  }

  private boolean resortNeeded(FieldComparator[] fieldComparators) {
    for(FieldComparator fieldComparator : fieldComparators) {
      if(fieldComparator.getLeftFieldName().contains("(")) {
        return true;
      }
    }
    return false;
  }

  private void appendJson(StringBuilder buf,
                          Bucket[] _buckets,
                          Metric[] _metrics,
                          FieldComparator[] _sorts,
                          int _limit,
                          String method,
                          boolean refine,
                          int level) {
    buf.append('"');
    buf.append(_buckets[level].toString());
    buf.append('"');
    buf.append(":{");
    buf.append("\"type\":\"terms\"");
    buf.append(",\"field\":\"").append(_buckets[level].toString()).append('"');
    buf.append(",\"limit\":").append(_limit);

    if(refine) {
      buf.append(",\"refine\":true");
    }

    if(method != null) {
      buf.append(",\"method\":\"").append(method).append('"');
    }

    String fsort = getFacetSort(_sorts[level].getLeftFieldName(), _metrics);

    buf.append(",\"sort\":{\"").append(fsort).append("\":\"").append(_sorts[level].getOrder()).append("\"}");

    buf.append(",\"facet\":{");
    int metricCount = 0;


    ++level;
    boolean comma = false;
    for(Metric metric : _metrics) {
      //Only compute the metric if it's a leaf node or if the branch level sort equals is the metric
      String facetKey = "facet_"+metricCount;
      String identifier = metric.getIdentifier();
      if (!identifier.startsWith("count(")) {
        if (comma) {
          buf.append(",");
        }

        if(level == _buckets.length || fsort.equals(facetKey) ) {
          comma = true;
          if (identifier.startsWith("per(")) {
            buf.append("\"facet_").append(metricCount).append("\":\"").append(identifier.replaceFirst("per", "percentile")).append('"');
          } else if (identifier.startsWith("std(")) {
            buf.append("\"facet_").append(metricCount).append("\":\"").append(identifier.replaceFirst("std", "stddev")).append('"');
          } else {
            buf.append('"').append(facetKey).append("\":\"").append(identifier).append('"');
          }
        }
        ++metricCount;
      }
    }

    if(level < _buckets.length) {
      if(metricCount>0) {
        buf.append(",");
      }
      appendJson(buf, _buckets, _metrics, _sorts, _limit, method, refine, level);
    }
    buf.append("}}");
  }

  private String getFacetSort(String id, Metric[] _metrics) {
    int index = 0;
    int metricCount=0;
    for(Metric metric : _metrics) {
      if(metric.getIdentifier().startsWith("count(")) {
        if(id.startsWith("count(")) {
          return "count";
        }
        ++index;
      } else {
        if (id.equals(_metrics[index].getIdentifier())) {
          return "facet_" + metricCount;
        }
        ++index;
        ++metricCount;
      }
    }
    return "index";
  }

  private void getTuples(@SuppressWarnings({"rawtypes"})NamedList response,
                                Bucket[] buckets,
                                Metric[] metrics) {

    Tuple tuple = new Tuple();
    @SuppressWarnings({"rawtypes"})
    NamedList facets = (NamedList)response.get("facets");
    fillTuples(0,
               tuples,
               tuple,
               facets,
               buckets,
               metrics);

  }

  private void fillTuples(int level,
                          List<Tuple> tuples,
                          Tuple currentTuple,
                          @SuppressWarnings({"rawtypes"}) NamedList facets,
                          Bucket[] _buckets,
                          Metric[] _metrics) {

    String bucketName = _buckets[level].toString();
    @SuppressWarnings({"rawtypes"})
    NamedList nl = (NamedList)facets.get(bucketName);
    if(nl == null) {
      return;
    }
    @SuppressWarnings({"rawtypes"})
    List allBuckets = (List)nl.get("buckets");
    for(int b=0; b<allBuckets.size(); b++) {
      @SuppressWarnings({"rawtypes"})
      NamedList bucket = (NamedList)allBuckets.get(b);
      Object val = bucket.get("val");
      if (val instanceof Integer) {
        val=((Integer)val).longValue();  // calcite currently expects Long values here
      }
      Tuple t = currentTuple.clone();
      t.put(bucketName, val);
      int nextLevel = level+1;
      if(nextLevel<_buckets.length) {
        fillTuples(nextLevel,
                   tuples,
                   t.clone(),
                   bucket,
                   _buckets,
                   _metrics);
      } else {
        int m = 0;
        for(Metric metric : _metrics) {
          String identifier = metric.getIdentifier();
          if(!identifier.startsWith("count(")) {
            Number d = ((Number)bucket.get("facet_"+m));
            if(metric.outputLong) {
              if (d instanceof Long || d instanceof Integer) {
                t.put(identifier, d.longValue());
              } else {
                t.put(identifier, Math.round(d.doubleValue()));
              }
            } else {
              t.put(identifier, d.doubleValue());
            }
            ++m;
          } else {
            long l = ((Number)bucket.get("count")).longValue();
            t.put("count(*)", l);
          }
        }
        tuples.add(t);
      }
    }
  }

  public int getCost() {
    return 0;
  }

  @Override
  public StreamComparator getStreamSort() {
    if(bucketSorts.length > 1) {
      return new MultipleFieldComparator(bucketSorts);
    } else {
      return bucketSorts[0];
    }
  }

  List<Metric> getMetrics() {
    return Arrays.asList(metrics);
  }

  @Override
  public TupleStream[] parallelize(List<String> partitions) throws IOException {

    final ModifiableSolrParams withoutTieredParam = new ModifiableSolrParams(params);
    withoutTieredParam.remove(TIERED_PARAM); // each individual facet request is not tiered

    TupleStream[] streams = new TupleStream[partitions.size()];
    for (int p = 0; p < streams.length; p++) {
      FacetStream cloned = new FacetStream();
      cloned.init(partitions.get(p), /* each collection */
                  withoutTieredParam, /* removes the tiered param */
                  buckets,
                  bucketSorts,
                  metrics,
                  rows,
                  offset,
                  bucketSizeLimit,
                  refine,
                  method,
                  serializeBucketSizeLimit,
                  overfetch,
                  zkHost);
      streams[p] = cloned;
    }
    return streams;
  }

  @Override
  public TupleStream getSortedRollupStream(ParallelListStream plist, Metric[] rollupMetrics) throws IOException {
    // using a hashRollup removes the need to sort the streams from the plist
    HashRollupStream rollup = new HashRollupStream(plist, buckets, rollupMetrics);
    SelectStream select = new SelectStream(rollup, getRollupSelectFields(rollupMetrics));
    // the final stream must be sorted based on the original stream sort
    return new SortStream(select, getStreamSort());
  }

  /**
   * The projection of dimensions and metrics from the rollup stream.
   *
   * @param rollupMetrics The metrics being rolled up.
   * @return A mapping of fields produced by the rollup stream to their output name.
   */
  protected Map<String, String> getRollupSelectFields(Metric[] rollupMetrics) {
    Map<String, String> map = new HashMap<>();
    for (Bucket b : buckets) {
      String key = b.toString();
      map.put(key, key);
    }
    for (Metric m : rollupMetrics) {
      String[] cols = m.getColumns();
      String col = cols != null && cols.length > 0 ? cols[0] : "*";
      map.put(m.getIdentifier(), col);
    }
    return map;
  }
}