/*
 * 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.drill.test;

import java.io.File;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import javax.json.Json;
import javax.json.JsonArray;
import javax.json.JsonNumber;
import javax.json.JsonObject;
import javax.json.JsonReader;
import javax.json.JsonValue;

import org.apache.drill.exec.server.rest.profile.CoreOperatorType;
import com.google.common.base.Preconditions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Parses a query profile and provides access to various bits of the profile
 * for diagnostic purposes during tests.
 */

public class ProfileParser {
  private static final Logger logger = LoggerFactory.getLogger(ProfileParser.class);

  /**
   * The original JSON profile.
   */

  JsonObject profile;

  /**
   * Query text parsed out of the profile.
   */

  String query;

  /**
   * List of operator plans in the order in which they appear in the query
   * plan section of the profile. This is an intermediate representation used
   * to create the more fully analyzed structures.
   */

  List<String> plans;

  /**
   * Operations sorted by operator ID. The Operator ID serves as
   * an index into the list to get the information for that operator.
   * Operator ID is the one shown in the plan: xx-nn, where nn is the
   * operator ID. This is NOT the same as the operator type.
   */

  List<OperatorSummary> operations;

  /**
   * Map from major fragment number to fragment information. The major
   * fragment number is the nn in the nn-xx notation in the plan.
   */

  Map<Integer,FragInfo> fragments = new HashMap<>();

  /**
   * Operations in the original topological order as shown in the text
   * version of the query plan in the query profile.
   */
  private List<OperatorSummary> topoOrder;

  public ProfileParser(File file) throws IOException {
    try (FileReader fileReader = new FileReader(file);
         JsonReader reader = Json.createReader(fileReader)) {
      profile = (JsonObject) reader.read();
    }

    parse();
  }

  private void parse() {
    parseQuery();
    parsePlans();
    buildFrags();
    parseFragProfiles();
    mapOpProfiles();
    aggregateOpers();
    buildTree();
  }

  private void parseQuery() {
    query = profile.getString("query");
    query = query.replace("//n", "\n");
  }

  /**
   * Parse a text version of the plan as it appears in the JSON
   * query profile.
   */

  private static class PlanParser {

    List<String> plans = new ArrayList<>();
    List<OperatorSummary> operations = new ArrayList<>();
    List<OperatorSummary> sorted = new ArrayList<>();

    public void parsePlans(String plan) {
      plans = new ArrayList<>();
      String[] parts = plan.split("\n");
      for (String part : parts) {
        plans.add(part);
        OperatorSummary opDef = new OperatorSummary(part);
        operations.add(opDef);
      }
      sortList();
    }

    private void sortList() {
      List<OperatorSummary> raw = new ArrayList<>();
      raw.addAll(operations);
      Collections.sort(raw, new Comparator<OperatorSummary>() {
        @Override
        public int compare(OperatorSummary o1, OperatorSummary o2) {
          int result = Integer.compare(o1.majorId, o2.majorId);
          if (result == 0) {
            result = Integer.compare(o1.stepId, o2.stepId);
          }
          return result;
        }
      });
      int currentFrag = 0;
      int currentStep = 0;
      for (OperatorSummary opDef : raw) {
        if (currentFrag < opDef.majorId) {
          currentFrag++;
          OperatorSummary sender = new OperatorSummary(currentFrag, 0);
          sender.isInferred = true;
          sender.name = "Sender";
          sorted.add(sender);
          currentStep = 1;
          opDef.inferredParent = sender;
          sender.children.add(opDef);
        }
        if (opDef.stepId > currentStep) {
          OperatorSummary unknown = new OperatorSummary(currentFrag, currentStep);
          unknown.isInferred = true;
          unknown.name = "Unknown";
          sorted.add(unknown);
          opDef.inferredParent = unknown;
          unknown.children.add(opDef);
        }
        sorted.add(opDef);
        currentStep = opDef.stepId + 1;
      }
    }
  }

  /**
   * Parse the plan portion of the query profile. Unfortunately,
   * the plan is in text form an is awkward to parse. Also, there is no ID
   * to correlate operators shown in the plan with those referenced in the
   * profile JSON. Inference is needed.
   */

  private void parsePlans() {
    PlanParser parser = new PlanParser();
    String plan = getPlan();
    parser.parsePlans(plan);
    plans = parser.plans;
    topoOrder = parser.operations;
    operations = parser.sorted;
  }

  private void buildFrags() {
    for (OperatorSummary opDef : operations) {
      FragInfo major = fragments.get(opDef.majorId);
      if (major == null) {
        major = new FragInfo(opDef.majorId);
        fragments.put(opDef.majorId, major);
      }
      major.ops.add(opDef);
    }
  }

  private static List<FieldDef> parseCols(String cols) {
    String[] parts = cols.split(", ");
    List<FieldDef> fields = new ArrayList<>();
    for (String part : parts) {
      String[] halves = part.split(" ");
      fields.add(new FieldDef(halves[1], halves[0]));
    }
    return fields;
  }

  private void parseFragProfiles() {
    JsonArray frags = getFragmentProfile();
    for (JsonObject fragProfile : frags.getValuesAs(JsonObject.class)) {
      int mId = fragProfile.getInt("majorFragmentId");
      FragInfo major = fragments.get(mId);
      major.parse(fragProfile);
    }
  }

  private void mapOpProfiles() {
    for (FragInfo major : fragments.values()) {
      for (MinorFragInfo minor : major.minors) {
        minor.mapOpProfiles(major);
      }
    }
  }

  /**
   * A typical plan has many operator details across multiple
   * minor fragments. Aggregate these totals to the "master"
   * definition of each operator.
   */

  private void aggregateOpers() {
    for (FragInfo major : fragments.values()) {
      for (OperatorSummary opDef : major.ops) {
        long sumPeak = 0;
        opDef.execCount = opDef.opExecs.size();
        for (OperatorProfile op : opDef.opExecs) {
          Preconditions.checkState(major.id == op.majorFragId);
          Preconditions.checkState(opDef.stepId == op.opId);
          opDef.actualRows += op.records;
          opDef.actualBatches += op.batches;
          opDef.setupMs += op.setupMs;
          opDef.processMs += op.processMs;
          sumPeak += op.peakMem;
        }
        opDef.actualMemory = sumPeak * 1024 * 1024;
      }
    }
  }

  /**
   * Reconstruct the operator tree from parsed information.
   */

  public void buildTree() {
    int currentLevel = 0;
    OperatorSummary[] opStack = new OperatorSummary[topoOrder.size()];
    for (OperatorSummary opDef : topoOrder) {
      currentLevel = opDef.globalLevel;
      opStack[currentLevel] = opDef;
      if (opDef.inferredParent == null) {
        if (currentLevel > 0) {
          opStack[currentLevel-1].children.add(opDef);
        }
      } else {
        opStack[currentLevel-1].children.add(opDef.inferredParent);
      }
    }
  }


  public String getQuery() {
    return profile.getString("query");
  }

  public String getPlan() {
    return profile.getString("plan");
  }

  public List<String> getPlans() {
    return plans;
  }

  public List<String> getScans() {
    List<String> scans = new ArrayList<>();
    int n = getPlans().size();
    for (int i = n-1; i >= 0;  i--) {
      String plan = plans.get(i);
      if (plan.contains(" Scan(")) {
        scans.add(plan);
      }
    }
    return scans;
  }

  public List<FieldDef> getColumns(String plan) {
    Pattern p = Pattern.compile("RecordType\\((.*)\\):");
    Matcher m = p.matcher(plan);
    if (! m.find()) { return null; }
    String frag = m.group(1);
    String[] parts = frag.split(", ");
    List<FieldDef> fields = new ArrayList<>();
    for (String part : parts) {
      String[] halves = part.split(" ");
      fields.add(new FieldDef(halves[1], halves[0]));
    }
    return fields;
  }

  public Map<Integer,String> getOperators() {
    Map<Integer,String> ops = new HashMap<>();
    int n = getPlans().size();
    Pattern p = Pattern.compile("\\d+-(\\d+)\\s+(\\w+)");
    for (int i = n-1; i >= 0;  i--) {
      String plan = plans.get(i);
      Matcher m = p.matcher(plan);
      if (! m.find()) { continue; }
      int index = Integer.parseInt(m.group(1));
      String op = m.group(2);
      ops.put(index,op);
    }
    return ops;
  }

  public JsonArray getFragmentProfile() {
    return profile.getJsonArray("fragmentProfile");
  }

  /**
   * Information for a fragment, including the operators
   * in that fragment and the set of minor fragments.
   */

  public static class FragInfo {
    public int baseLevel;
    public int id;
    public List<OperatorSummary> ops = new ArrayList<>();
    public List<MinorFragInfo> minors = new ArrayList<>();

    public FragInfo(int majorId) {
      this.id = majorId;
    }

    public OperatorSummary getRootOperator() {
      return ops.get(0);
    }

    public void parse(JsonObject fragProfile) {
      JsonArray minorList = fragProfile.getJsonArray("minorFragmentProfile");
      for (JsonObject minorProfile : minorList.getValuesAs(JsonObject.class)) {
        minors.add(new MinorFragInfo(id, minorProfile));
      }
    }
  }

  /**
   * Information about a minor fragment as parsed from the profile.
   */

  public static class MinorFragInfo {
    public final int majorId;
    public final int id;
    public final List<OperatorProfile> ops = new ArrayList<>();

    public MinorFragInfo(int majorId, JsonObject minorProfile) {
      this.majorId = majorId;
      id = minorProfile.getInt("minorFragmentId");
      JsonArray opList = minorProfile.getJsonArray("operatorProfile");
      for (JsonObject opProfile : opList.getValuesAs(JsonObject.class)) {
        ops.add(new OperatorProfile(majorId, id, opProfile));
      }
    }

    /**
     * Map each operator execution profiles back to the definition of that
     * operator. The only common key is the xx-yy value where xx is the fragment
     * number and yy is the operator ID.
     *
     * @param major major fragment that corresponds to the xx portion of the
     * operator id
     */

    public void mapOpProfiles(FragInfo major) {
      for (OperatorProfile op : ops) {
        OperatorSummary opDef = major.ops.get(op.opId);
        if (opDef == null) {
          logger.info("Can't find operator def: {}-{}", major.id, op.opId);
          continue;
        }
        op.opName = op.type.replace("_", " ");
        op.name = opDef.name;
        if (op.name.equalsIgnoreCase(op.opName)) {
          op.opName = null;
        }
        op.defn = opDef;
        opDef.opName = op.opName;
        opDef.type = op.type;
        opDef.opExecs.add(op);
      }
    }
  }

  /**
   * Detailed information about each operator within a minor fragment
   * for a major fragment. Gathers the detailed information from
   * the profile.
   */

  public static class OperatorProfile {
    public OperatorSummary defn;
    public String opName;
    public int majorFragId;
    public int minorFragId;
    public int opId;
    public String type;
    public String name;
    public long processMs;
    public long waitMs;
    public long setupMs;
    public long peakMem;
    public Map<Integer,JsonNumber> metrics = new HashMap<>();
    public long records;
    public int batches;
    public int schemas;

    public OperatorProfile(int majorId, int minorId, JsonObject opProfile) {
      majorFragId = majorId;
      minorFragId = minorId;
      opId = opProfile.getInt("operatorId");
      JsonValue.ValueType valueType = opProfile.get("operatorType").getValueType();
      type = valueType == JsonValue.ValueType.STRING
          ? opProfile.getString("operatorType")
          : Objects.requireNonNull(CoreOperatorType.valueOf(opProfile.getInt("operatorType"))).name();
      processMs = opProfile.getJsonNumber("processNanos").longValue() / 1_000_000;
      waitMs = opProfile.getJsonNumber("waitNanos").longValue() / 1_000_000;
      setupMs = opProfile.getJsonNumber("setupNanos").longValue() / 1_000_000;
      peakMem = opProfile.getJsonNumber("peakLocalMemoryAllocated").longValue() / (1024 * 1024);
      JsonArray array = opProfile.getJsonArray("inputProfile");
      if (array != null) {
        for (int i = 0; i < array.size(); i++) {
          JsonObject obj = array.getJsonObject(i);
          records += obj.getJsonNumber("records").longValue();
          batches += obj.getInt("batches");
          schemas += obj.getInt("schemas");
        }
      }
      array = opProfile.getJsonArray("metric");
      if (array != null) {
        for (int i = 0; i < array.size(); i++) {
          JsonObject metric = array.getJsonObject(i);
          metrics.put(metric.getJsonNumber("metricId").intValue(), metric.getJsonNumber("longValue"));
        }
      }
    }

    public long getMetric(int id) {
      JsonNumber value = metrics.get(id);
      if (value == null) {
        return 0;
      }
      return value.longValue();
    }

    @Override
    public String toString() {
      return String.format("[OperatorProfile %02d-%02d-%02d, type: %s, name: %s]",
          majorFragId, opId, minorFragId, type,
          (name == null) ? "null" : name);
    }
  }

  /**
   * Information about an operator definition: the plan-time information
   * that appears in the plan portion of the profile. Also holds the
   * "actuals" from the minor fragment portion of the profile.
   * Allows integrating the "planned" vs. "actual" performance of the
   * query.
   * <p>
   * There is one operator definition (represented here), each of which may
   * give rise to multiple operator executions (housed in minor fragments.)
   * The {@link #opExecs} field provides the list of operator executions
   * (which provides access to operator metrics.)
   */

  public static class OperatorSummary {
    public String type;
    public long processMs;
    public long setupMs;
    public int execCount;
    public String opName;
    public boolean isInferred;
    public int majorId;
    public int stepId;
    public String args;
    public List<FieldDef> columns;
    public int globalLevel;
    public int localLevel;
    public int id;
    public int branchId;
    public boolean isBranchRoot;
    public double estMemoryCost;
    public double estNetCost;
    public double estIOCost;
    public double estCpuCost;
    public double estRowCost;
    public double estRows;
    public String name;
    public long actualMemory;
    public int actualBatches;
    public long actualRows;
    public OperatorSummary inferredParent;
    public List<OperatorProfile> opExecs = new ArrayList<>();
    public List<OperatorSummary> children = new ArrayList<>();

    // 00-00    Screen : rowType = RecordType(VARCHAR(10) Year, VARCHAR(65536) Month, VARCHAR(100) Devices, VARCHAR(100) Tier, VARCHAR(100) LOB, CHAR(10) Gateway, BIGINT Day, BIGINT Hour, INTEGER Week, VARCHAR(100) Week_end_date, BIGINT Usage_Cnt): \
    // rowcount = 100.0, cumulative cost = {7.42124276972414E9 rows, 7.663067406383167E10 cpu, 0.0 io, 2.24645048816E10 network, 2.692766612982188E8 memory}, id = 129302
    //
    // 00-01      Project(Year=[$0], Month=[$1], Devices=[$2], Tier=[$3], LOB=[$4], Gateway=[$5], Day=[$6], Hour=[$7], Week=[$8], Week_end_date=[$9], Usage_Cnt=[$10]) :
    // rowType = RecordType(VARCHAR(10) Year, VARCHAR(65536) Month, VARCHAR(100) Devices, VARCHAR(100) Tier, VARCHAR(100) LOB, CHAR(10) Gateway, BIGINT Day, BIGINT Hour, INTEGER Week, VARCHAR(100) Week_end_date, BIGINT Usage_Cnt): rowcount = 100.0, cumulative cost = {7.42124275972414E9 rows, 7.663067405383167E10 cpu, 0.0 io, 2.24645048816E10 network, 2.692766612982188E8 memory}, id = 129301

    public OperatorSummary(String plan) {
      Pattern p = Pattern.compile("^(\\d+)-(\\d+)(\\s+)(\\w+)(?:\\((.*)\\))?\\s*:\\s*(.*)$");
      Matcher m = p.matcher(plan);
      if (!m.matches()) {
        throw new IllegalStateException("Could not parse plan: " + plan);
      }
      majorId = Integer.parseInt(m.group(1));
      stepId = Integer.parseInt(m.group(2));
      name = m.group(4);
      args = m.group(5);
      String tail = m.group(6);
      String indent = m.group(3);
      globalLevel = (indent.length() - 4) / 2;

      p = Pattern.compile("rowType = RecordType\\((.*)\\): (rowcount .*)");
      m = p.matcher(tail);
      if (m.matches()) {
        columns = parseCols(m.group(1));
        tail = m.group(2);
      }

      p = Pattern.compile("rowcount = ([\\d.E]+), cumulative cost = \\{([\\d.E]+) rows, ([\\d.E]+) cpu, ([\\d.E]+) io, ([\\d.E]+) network, ([\\d.E]+) memory\\}, id = (\\d+)");
      m = p.matcher(tail);
      if (! m.find()) {
        throw new IllegalStateException("Could not parse costs: " + tail);
      }
      estRows = Double.parseDouble(m.group(1));
      estRowCost = Double.parseDouble(m.group(2));
      estCpuCost = Double.parseDouble(m.group(3));
      estIOCost = Double.parseDouble(m.group(4));
      estNetCost = Double.parseDouble(m.group(5));
      estMemoryCost = Double.parseDouble(m.group(6));
      id = Integer.parseInt(m.group(7));
    }

    public void printTree(String indent) {
      new TreePrinter().visit(this);
    }

    public OperatorSummary(int major, int id) {
      majorId = major;
      stepId = id;
    }

    @Override
    public String toString() {
      String head = "[OpDefInfo " + majorId + "-" + stepId + ": " + name;
      if (isInferred) {
        head += " (" + opName + ")";
      }
      return head + "]";
    }
  }

  /**
   * Visit a tree of operator definitions to support printing,
   * analysis and other tasks.
   */

  public static class TreeVisitor
  {
    public void visit(OperatorSummary root) {
      visit(root, 0);
    }

    public void visit(OperatorSummary node, int indent) {
      visitOp(node, indent);
      if (node.children.isEmpty()) {
        return;
      }
      if (node.children.size() == 1) {
        visit(node.children.get(0), indent);
        return;
      }
      indent++;
      int i = 0;
      for (OperatorSummary child : node.children) {
        visitSubtree(node, i++, indent);
        visit(child, indent+1);
      }
    }

    protected void visitOp(OperatorSummary node, int indent) {
    }

    protected void visitSubtree(OperatorSummary node, int i, int indent) {
    }

    public String indentString(int indent, String pad) {
      StringBuilder buf = new StringBuilder();
      for (int i = 0; i < indent; i++) {
        buf.append(pad);
      }
      return buf.toString();
    }

    public String indentString(int indent) {
      return indentString(indent, "  ");
    }

    public String subtreeLabel(OperatorSummary node, int branch) {
      if (node.name.equals("HashJoin")) {
        return (branch == 0) ? "Probe" : "Build";
      } else {
        return "Input " + (branch + 1);
      }
    }
  }

  /**
   * Print the operator tree for analysis.
   */

  public static class TreePrinter extends TreeVisitor
  {
    @Override
    protected void visitOp(OperatorSummary node, int indent) {
      logger.info("{}{}", indentString(indent), node.toString());
    }

    @Override
    protected void visitSubtree(OperatorSummary node, int i, int indent) {
      logger.info("{}{}", indentString(indent), subtreeLabel(node, i));
    }
  }

  /**
   * Print out the tree showing a comparison of estimated vs.
   * actual costs. Example:
   * <p><pre>
   * 03-05 HashJoin (HASH JOIN)
   *                                 Estimate:       2,521,812 rows,      1 MB
   *                                 Actual:           116,480 rows,     52 MB
   *         Probe
   * 03-07 . . Project
   *                                 Estimate:       2,521,812 rows,      1 MB
   *                                 Actual:                 0 rows,      0 MB
   * </pre>
   */

  public static class CostPrinter extends TreeVisitor
  {
    @Override
    protected void visitOp(OperatorSummary node, int indentLevel) {
      final StringBuilder nodeBuilder = new StringBuilder();
      nodeBuilder.append(String.format("%02d-%02d ", node.majorId, node.stepId));
      String indent = indentString(indentLevel, ". ");
      nodeBuilder.append(indent).append(node.name);
      if (node.opName != null) {
        nodeBuilder.append(" (").append(node.opName).append(")");
      }
      logger.info(nodeBuilder.toString());

      final StringBuilder costBuilder = new StringBuilder();
      indent = indentString(15);
      costBuilder.append(indent);
      costBuilder.append(String.format("  Estimate: %,15.0f rows, %,7.0f MB",
                         node.estRows, node.estMemoryCost / 1024 / 1024));
      costBuilder.append(indent);
      costBuilder.append(String.format("  Actual:   %,15d rows, %,7d MB",
                         node.actualRows, node.actualMemory / 1024 / 1024));
      logger.info(nodeBuilder.toString());
    }

    @Override
    protected void visitSubtree(OperatorSummary node, int i, int indent) {
      logger.info("{}      {}", indentString(indent), subtreeLabel(node, i));
    }
  }

  public static class FindOpVisitor extends TreeVisitor
  {
    private List<OperatorSummary> ops;
    private String type;

    public List<OperatorSummary> find(String type, OperatorSummary node) {
      ops = new ArrayList<>();
      this.type = type;
      visit(node);
      return ops;
    }

    @Override
    protected void visitOp(OperatorSummary node, int indentLevel) {
      if (type.equals(node.type)) {
        ops.add(node);
      }
    }
  }

  /**
   * We often run test queries single threaded to make analysis of the profile
   * easier. For a single-threaded (single slice) query, get a map from
   * operator ID to operator information as preparation for additional
   * analysis.
   *
   * @return
   */

  public Map<Integer,OperatorProfile> getOpInfo() {
    Map<Integer,String> ops = getOperators();
    Map<Integer,OperatorProfile> info = new HashMap<>();
    JsonArray frags = getFragmentProfile();
    JsonObject fragProfile = frags.getJsonObject(0).getJsonArray("minorFragmentProfile").getJsonObject(0);
    JsonArray opList = fragProfile.getJsonArray("operatorProfile");
    for (JsonObject opProfile : opList.getValuesAs(JsonObject.class)) {
      parseOpProfile(ops, info, opProfile);
    }
    return info;
  }

  /**
   * For a single-slice query, get all operators of a given numeric operator
   * type.
   * @param type the operator type
   * @return a list of operators of the given type
   */

  public List<OperatorProfile> getOpsOfType(String type) {
    List<OperatorProfile> ops = new ArrayList<>();
    List<OperatorSummary> opDefs = getOpDefsOfType(type);
    for (OperatorSummary opDef : opDefs) {
      ops.addAll(opDef.opExecs);
    }
    return ops;
  }

  public List<OperatorSummary> getOpDefsOfType(String type) {
    return new FindOpVisitor().find(type, topoOrder.get(0));
  }

  private void parseOpProfile(Map<Integer, String> ops,
      Map<Integer, OperatorProfile> info, JsonObject opProfile) {
    OperatorProfile opInfo = new OperatorProfile(0, 0, opProfile);
    opInfo.name = ops.get(opInfo.opId);
    info.put(opInfo.opId, opInfo);
  }

  public void printPlan() {
    new CostPrinter().visit(topoOrder.get(0));
  }

  public void printTime() {
    new TimePrinter().visit(topoOrder.get(0));
  }

  public static class Aggregator extends TreeVisitor
  {
    protected int n;
    protected long totalSetup;
    protected long totalProcess;
    protected long total;
    protected int maxFrag;
    protected boolean isTree;

    @Override
    public void visit(OperatorSummary root) {
      super.visit(root, 0);
      total = totalSetup + totalProcess;
    }

    @Override
    protected void visitOp(OperatorSummary node, int indentLevel) {
      n++;
      totalSetup += node.setupMs;
      totalProcess += node.processMs;
      maxFrag = Math.max(maxFrag, node.majorId);
      isTree |= (node.children.size() > 1);
    }
  }

  public static class TimePrinter extends TreeVisitor
  {
    private Aggregator totals;
    private boolean singleThread;
    private boolean singleFragment;

    @Override
    public void visit(OperatorSummary root) {
      totals = new Aggregator();
      totals.visit(root);
      singleThread = ! totals.isTree;
      singleFragment = (totals.maxFrag == 0);
      super.visit(root, 0);
      logger.info("Total:");
      String indent = singleThread? "  " : indentString(15);
      logger.info("{}{}", indent, String.format("Setup:   %,6d ms", totals.totalSetup));
      logger.info("{}{}", indent, String.format("Process: %,6d ms", totals.totalProcess));
    }

    @Override
    protected void visitOp(OperatorSummary node, int indentLevel) {
      if (singleThread) {
        printSimpleFormat(node);
      } else {
        printTreeFormat(node, indentLevel);
      }
    }

    private void printSimpleFormat(OperatorSummary node) {
      final StringBuilder sb = new StringBuilder();

      if (singleFragment) {
        sb.append(String.format("%02d ", node.stepId));
      } else {
        sb.append(String.format("%02d-%02d ", node.majorId, node.stepId));
      }
      sb.append(node.name);
      if (node.opName != null) {
        sb.append(" (").append(node.opName).append(")");
      }
      logger.info(sb.toString());
      printTimes(node, "  ");
    }

    private void printTimes(OperatorSummary node, String indent) {
      logger.info("{}{}", indent, String.format("Setup:   %,6d ms - %3d%%, %3d%%", node.setupMs,
                         percent(node.setupMs, totals.totalSetup),
                         percent(node.setupMs, totals.total)));
      logger.info("{}{}", indent, String.format("Process: %,6d ms - %3d%%, %3d%%", node.processMs,
                         percent(node.processMs, totals.totalProcess),
                         percent(node.processMs, totals.total)));
    }

    private void printTreeFormat(OperatorSummary node, int indentLevel) {
      final StringBuilder sb = new StringBuilder();
      sb.append(String.format("%02d-%02d ", node.majorId, node.stepId));
      String indent = indentString(indentLevel, ". ");
      sb.append(indent).append(node.name);
      if (node.opName != null) {
        sb.append(" (").append(node.opName).append(")");
      }
      logger.info(sb.toString());
      indent = indentString(15);
      printTimes(node, indent);
    }
  }

  /**
   * For a single-slice query, print a summary of the operator stack
   * and costs. At present, works for a linear query with on single-input
   * operators.
   */

  public void print() {
    printTime();
  }

  public void simplePrint() {
    Map<Integer, OperatorProfile> opInfo = getOpInfo();
    int n = opInfo.size();
    long totalSetup = 0;
    long totalProcess = 0;
    for (int i = 0;  i <= n;  i++) {
      OperatorProfile op = opInfo.get(i);
      if (op == null) { continue; }
      totalSetup += op.setupMs;
      totalProcess += op.processMs;
    }
    long total = totalSetup + totalProcess;
    for (int i = 0;  i <= n;  i++) {
      OperatorProfile op = opInfo.get(i);
      if (op == null) { continue; }
      logger.info("Op: {} {}", op.opId, op.name);
      logger.info("Setup:   {} - {}%, {}%", op.setupMs, percent(op.setupMs, totalSetup), percent(op.setupMs, total));
      logger.info("Process: {} - {}%, {}%", op.processMs, percent(op.processMs, totalProcess), percent(op.processMs, total));
      if (op.type.equals("EXTERNAL_SORT")) {
        long value = op.getMetric(0);
        logger.info("  Spills: {}", value);
      }
      if (op.waitMs > 0) {
        logger.info("  Wait:    {}", op.waitMs);
      }
      if (op.peakMem > 0) {
        logger.info("  Memory: {}", op.peakMem);
      }
    }
    logger.info("Total:");
    logger.info("  Setup:   {}", totalSetup);
    logger.info("  Process: {}", totalProcess);
  }

  public static long percent(long value, long total) {
    if (total == 0) {
      return 0; }
    return Math.round(value * 100 / (double)total);
  }

  public List<OperatorSummary> getOpDefn(String target) {
    List<OperatorSummary> ops = new ArrayList<>();
    for (OperatorSummary opDef : operations) {
      if (opDef.name.startsWith(target)) {
        ops.add(opDef);
      }
    }
    return ops;
  }
}