You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@drill.apache.org by ja...@apache.org on 2014/08/31 19:28:07 UTC
[08/15] git commit: DRILL-1365: output unified coordinates at profile
view
DRILL-1365: output unified coordinates at profile view
Project: http://git-wip-us.apache.org/repos/asf/incubator-drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-drill/commit/ffbf3ae1
Tree: http://git-wip-us.apache.org/repos/asf/incubator-drill/tree/ffbf3ae1
Diff: http://git-wip-us.apache.org/repos/asf/incubator-drill/diff/ffbf3ae1
Branch: refs/heads/master
Commit: ffbf3ae15a2752acc60693972b9a37eb948ef872
Parents: c1fcb65
Author: Hanifi Gunes <hg...@maprtech.com>
Authored: Fri Aug 29 18:34:07 2014 -0700
Committer: Jacques Nadeau <ja...@apache.org>
Committed: Fri Aug 29 22:16:40 2014 -0700
----------------------------------------------------------------------
.../drill/exec/server/rest/ProfileWrapper.java | 229 ++++++++++++-------
.../physical/impl/writer/TestParquetWriter.java | 5 +-
2 files changed, 152 insertions(+), 82 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/ffbf3ae1/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/ProfileWrapper.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/ProfileWrapper.java b/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/ProfileWrapper.java
index 777779f..63ecb98 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/ProfileWrapper.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/ProfileWrapper.java
@@ -17,6 +17,7 @@
*/
package org.apache.drill.exec.server.rest;
+import com.google.common.base.Preconditions;
import org.apache.commons.lang3.tuple.ImmutablePair;
import org.apache.commons.lang3.tuple.Pair;
import org.apache.drill.exec.proto.UserBitShared.CoreOperatorType;
@@ -30,7 +31,6 @@ import org.apache.drill.exec.proto.helper.QueryIdHelper;
import com.google.common.base.Predicate;
import com.google.common.base.Predicates;
import com.google.common.collect.Collections2;
-import org.apache.drill.exec.proto.helper.QueryIdHelper;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
@@ -45,6 +45,7 @@ import java.util.Locale;
import java.util.Map;
public class ProfileWrapper {
+
public QueryProfile profile;
public String id;
@@ -64,23 +65,23 @@ public class ProfileWrapper {
public String getQueryId() {
return QueryIdHelper.getQueryId(profile.getId());
}
-
+
public List<OperatorWrapper> getOperatorProfiles() {
List<OperatorWrapper> ows = Lists.newArrayList();
Map<ImmutablePair<Integer, Integer>, List<ImmutablePair<OperatorProfile, Integer>>> opmap = Maps.newHashMap();
-
+
List<MajorFragmentProfile> majors = new ArrayList<>(profile.getFragmentProfileList());
Collections.sort(majors, Comparators.majorIdCompare);
for (MajorFragmentProfile major : majors) {
-
+
List<MinorFragmentProfile> minors = new ArrayList<>(major.getMinorFragmentProfileList());
Collections.sort(minors, Comparators.minorIdCompare);
for (MinorFragmentProfile minor : minors) {
-
+
List<OperatorProfile> ops = new ArrayList<>(minor.getOperatorProfileList());
Collections.sort(ops, Comparators.operatorIdCompare);
for (OperatorProfile op : ops) {
-
+
ImmutablePair<Integer, Integer> ip = new ImmutablePair<>(
major.getMajorFragmentId(), op.getOperatorId());
if (!opmap.containsKey(ip)) {
@@ -91,30 +92,31 @@ public class ProfileWrapper {
}
}
}
-
+
List<ImmutablePair<Integer, Integer>> keys = new ArrayList<>(opmap.keySet());
Collections.sort(keys);
+ ImmutablePair<OperatorProfile, Integer> val;
for (ImmutablePair<Integer, Integer> ip : keys) {
ows.add(new OperatorWrapper(ip.getLeft(), opmap.get(ip)));
}
-
+
return ows;
}
-
+
public List<FragmentWrapper> getFragmentProfiles() {
List<FragmentWrapper> fws = Lists.newArrayList();
-
+
List<MajorFragmentProfile> majors = new ArrayList<>(profile.getFragmentProfileList());
Collections.sort(majors, Comparators.majorIdCompare);
for (MajorFragmentProfile major : majors) {
fws.add(new FragmentWrapper(major));
}
-
+
return fws;
}
public String getFragmentsOverview() {
- final String[] columns = {"Fragment", "Minor Fragments Reporting", "First Start", "Last Start", "First End", "Last End", "tmin", "tavg", "tmax"};
+ final String[] columns = {"Major Fragment", "Minor Fragments Reporting", "First Start", "Last Start", "First End", "Last End", "tmin", "tavg", "tmax"};
TableBuilder tb = new TableBuilder(columns);
for (FragmentWrapper fw : getFragmentProfiles()) {
fw.addSummary(tb);
@@ -122,19 +124,19 @@ public class ProfileWrapper {
return tb.toString();
}
- public String majorFragmentTimingProfile(MajorFragmentProfile majorFragmentProfile) {
- final String[] columns = {"Minor", "Start", "End", "Total Time", "Max Records", "Max Batches"};
+ public String majorFragmentTimingProfile(MajorFragmentProfile major) {
+ final String[] columns = {"Minor Fragment", "Start", "End", "Total Time", "Max Records", "Max Batches"};
TableBuilder builder = new TableBuilder(columns);
ArrayList<MinorFragmentProfile> complete, incomplete;
complete = new ArrayList<MinorFragmentProfile>(Collections2.filter(
- majorFragmentProfile.getMinorFragmentProfileList(), Filters.hasOperatorsAndTimes));
+ major.getMinorFragmentProfileList(), Filters.hasOperatorsAndTimes));
incomplete = new ArrayList<MinorFragmentProfile>(Collections2.filter(
- majorFragmentProfile.getMinorFragmentProfileList(), Filters.missingOperatorsOrTimes));
+ major.getMinorFragmentProfileList(), Filters.missingOperatorsOrTimes));
Collections.sort(complete, Comparators.minorIdCompare);
- for (MinorFragmentProfile m : complete) {
- ArrayList<OperatorProfile> ops = new ArrayList<OperatorProfile>(m.getOperatorProfileList());
+ for (MinorFragmentProfile minor : complete) {
+ ArrayList<OperatorProfile> ops = new ArrayList<OperatorProfile>(minor.getOperatorProfileList());
long t0 = profile.getStart();
long biggestIncomingRecords = 0;
@@ -151,23 +153,23 @@ public class ProfileWrapper {
biggestBatches = Math.max(biggestBatches, batches);
}
- builder.appendCell(String.format("#%d - %d", majorFragmentProfile.getMajorFragmentId(), m.getMinorFragmentId()), null);
- builder.appendMillis(m.getStartTime() - t0, null);
- builder.appendMillis(m.getEndTime() - t0, null);
- builder.appendMillis(m.getEndTime() - m.getStartTime(), null);
-
+ builder.appendCell(new OperatorPathBuilder().setMajor(major).setMinor(minor).build(), null);
+ builder.appendMillis(minor.getStartTime() - t0, null);
+ builder.appendMillis(minor.getEndTime() - t0, null);
+ builder.appendMillis(minor.getEndTime() - minor.getStartTime(), null);
+
builder.appendInteger(biggestIncomingRecords, null);
builder.appendInteger(biggestBatches, null);
}
for (MinorFragmentProfile m : incomplete) {
builder.appendCell(
- majorFragmentProfile.getMajorFragmentId() + "-"
+ major.getMajorFragmentId() + "-"
+ m.getMinorFragmentId(), null);
builder.appendRepeated(m.getState().toString(), null, 5);
}
return builder.toString();
}
-
+
public String getOperatorsOverview() {
final String [] columns = {"Operator", "Type", "Setup (min)", "Setup (avg)", "Setup (max)", "Process (min)", "Process (avg)", "Process (max)", "Wait (min)", "Wait (avg)", "Wait (max)"};
TableBuilder tb = new TableBuilder(columns);
@@ -176,7 +178,7 @@ public class ProfileWrapper {
}
return tb.toString();
}
-
+
public String getOperatorsJSON() {
StringBuilder sb = new StringBuilder("{");
String sep = "";
@@ -186,31 +188,99 @@ public class ProfileWrapper {
}
return sb.append("}").toString();
}
-
+
+ private static class OperatorPathBuilder {
+ private static final String OPERATOR_PATH_PATTERN = "%s-%s-%s";
+ private static final String DEFAULT = "xx";
+ private String major;
+ private String minor;
+ private String operator;
+
+ public OperatorPathBuilder() {
+ clear();
+ }
+
+ public void clear() {
+ major = DEFAULT;
+ minor = DEFAULT;
+ operator = DEFAULT;
+ }
+
+ // Utility to left pad strings
+ protected String leftPad(String text) {
+ return String.format("00%s", text).substring(text.length());
+ }
+
+ public OperatorPathBuilder setMajor(MajorFragmentProfile major) {
+ if (major!=null) {
+ return setMajor(major.getMajorFragmentId());
+ }
+ return this;
+ }
+
+ public OperatorPathBuilder setMajor(int newMajor) {
+ major = leftPad(String.valueOf(newMajor));
+ return this;
+ }
+
+ public OperatorPathBuilder setMinor(MinorFragmentProfile minor) {
+ if (minor!=null) {
+ return setMinor(minor.getMinorFragmentId());
+ }
+ return this;
+ }
+
+ public OperatorPathBuilder setMinor(int newMinor) {
+ minor = leftPad(String.valueOf(newMinor));
+ return this;
+ }
+
+ public OperatorPathBuilder setOperator(OperatorProfile op) {
+ if (op!=null) {
+ return setOperator(op.getOperatorId());
+ }
+ return this;
+ }
+
+ public OperatorPathBuilder setOperator(int newOp) {
+ operator = leftPad(String.valueOf(newOp));
+ return this;
+ }
+
+ public String build() {
+ StringBuffer sb = new StringBuffer();
+ return sb.append(major).append("-")
+ .append(minor).append("-")
+ .append(operator)
+ .toString();
+ }
+ }
+
public class FragmentWrapper {
- MajorFragmentProfile major;
+ private final MajorFragmentProfile major;
+
public FragmentWrapper(MajorFragmentProfile major) {
- this.major = major;
+ this.major = Preconditions.checkNotNull(major);
}
-
+
public String getDisplayName() {
- return "Fragment #" + major.getMajorFragmentId();
+ return String.format("Major Fragment: %s", new OperatorPathBuilder().setMajor(major).build());
}
-
+
public String getId() {
- return "fragment-" + major.getMajorFragmentId();
+ return String.format("fragment-%s", major.getMajorFragmentId());
}
-
+
public void addSummary(TableBuilder tb) {
final String fmt = " (%d)";
long t0 = profile.getStart();
-
+
ArrayList<MinorFragmentProfile> complete = new ArrayList<MinorFragmentProfile>(
Collections2.filter(major.getMinorFragmentProfileList(), Filters.hasOperatorsAndTimes));
- tb.appendCell("#" + major.getMajorFragmentId(), null);
+ tb.appendCell(new OperatorPathBuilder().setMajor(major).build(), null);
tb.appendCell(complete.size() + " / " + major.getMinorFragmentProfileCount(), null);
-
+
if (complete.size() < 1) {
tb.appendRepeated("", null, 7);
return;
@@ -225,7 +295,7 @@ public class ProfileWrapper {
Collections.sort(complete, Comparators.endTimeCompare);
tb.appendMillis(complete.get(0).getEndTime() - t0, String.format(fmt, complete.get(0).getMinorFragmentId()));
tb.appendMillis(complete.get(li).getEndTime() - t0, String.format(fmt, complete.get(li).getMinorFragmentId()));
-
+
long total = 0;
for (MinorFragmentProfile p : complete) {
total += p.getEndTime() - p.getStartTime();
@@ -237,65 +307,68 @@ public class ProfileWrapper {
tb.appendMillis(complete.get(li).getEndTime() - complete.get(li).getStartTime(),
String.format(fmt, complete.get(li).getMinorFragmentId()));
}
-
+
public String getContent() {
return majorFragmentTimingProfile(major);
}
}
-
+
public class OperatorWrapper {
- int major;
- List<ImmutablePair<OperatorProfile, Integer>> ops;
-
+ private final int major;
+ private List<ImmutablePair<OperatorProfile, Integer>> ops;
+
public OperatorWrapper(int major, List<ImmutablePair<OperatorProfile, Integer>> ops) {
assert ops.size() > 0;
this.major = major;
this.ops = ops;
}
-
+
public String getDisplayName() {
- return String.format("Fragment #%d - Operator %d (%s)",
- major, ops.get(0).getLeft().getOperatorId(),
- CoreOperatorType.valueOf(ops.get(0).getLeft().getOperatorType() ).toString());
+ OperatorProfile op = ops.get(0).getLeft();
+ String path = new OperatorPathBuilder().setMajor(major).setOperator(op).build();
+ return String.format("%s - %s", path, CoreOperatorType.valueOf(op.getOperatorType()).toString());
}
-
+
public String getId() {
return String.format("operator-%d-%d", major, ops.get(0).getLeft().getOperatorId());
}
-
+
public String getContent() {
- final String [] columns = {"Fragment", "Setup", "Process", "Wait", "Max Batches", "Max Records"};
+ final String [] columns = {"Minor Fragment", "Setup", "Process", "Wait", "Max Batches", "Max Records"};
TableBuilder builder = new TableBuilder(columns);
-
+
for (ImmutablePair<OperatorProfile, Integer> ip : ops) {
int minor = ip.getRight();
OperatorProfile op = ip.getLeft();
-
- builder.appendCell(String.format("#%d - %d", major, minor), null);
+
+ String path = new OperatorPathBuilder().setMajor(major).setMinor(minor).setOperator(op).build();
+ builder.appendCell(path, null);
builder.appendNanos(op.getSetupNanos(), null);
builder.appendNanos(op.getProcessNanos(), null);
builder.appendNanos(op.getWaitNanos(), null);
-
+
long maxBatches = Long.MIN_VALUE;
long maxRecords = Long.MIN_VALUE;
for (StreamProfile sp : op.getInputProfileList()) {
maxBatches = Math.max(sp.getBatches(), maxBatches);
maxRecords = Math.max(sp.getRecords(), maxRecords);
}
-
+
builder.appendInteger(maxBatches, null);
builder.appendInteger(maxRecords, null);
}
return builder.toString();
}
-
+
public void addSummary(TableBuilder tb) {
- tb.appendCell(String.format("#%d - Op %d", major, ops.get(0).getLeft().getOperatorId()), null);
+ OperatorProfile op = ops.get(0).getLeft();
+ String path = new OperatorPathBuilder().setMajor(major).setOperator(op).build();
+ tb.appendCell(path, null);
tb.appendCell(CoreOperatorType.valueOf(ops.get(0).getLeft().getOperatorType() ).toString(), null);
int li = ops.size() - 1;
String fmt = " (%s)";
-
+
double setupSum = 0.0;
double processSum = 0.0;
double waitSum = 0.0;
@@ -304,17 +377,17 @@ public class ProfileWrapper {
processSum += ip.getLeft().getProcessNanos();
waitSum += ip.getLeft().getWaitNanos();
}
-
+
Collections.sort(ops, Comparators.setupTimeSort);
tb.appendNanos(ops.get(0).getLeft().getSetupNanos(), String.format(fmt, ops.get(0).getRight()));
tb.appendNanos((long) (setupSum / ops.size()), null);
tb.appendNanos(ops.get(li).getLeft().getSetupNanos(), String.format(fmt, ops.get(li).getRight()));
-
+
Collections.sort(ops, Comparators.processTimeSort);
tb.appendNanos(ops.get(0).getLeft().getProcessNanos(), String.format(fmt, ops.get(0).getRight()));
tb.appendNanos((long) (processSum / ops.size()), null);
tb.appendNanos(ops.get(li).getLeft().getProcessNanos(), String.format(fmt, ops.get(li).getRight()));
-
+
Collections.sort(ops, Comparators.waitTimeSort);
tb.appendNanos(ops.get(0).getLeft().getWaitNanos(), String.format(fmt, ops.get(0).getRight()));
tb.appendNanos((long) (waitSum / ops.size()), null);
@@ -328,13 +401,13 @@ public class ProfileWrapper {
return Long.compare(o1.getMajorFragmentId(), o2.getMajorFragmentId());
}
};
-
+
final static Comparator<MinorFragmentProfile> minorIdCompare = new Comparator<MinorFragmentProfile>() {
public int compare(MinorFragmentProfile o1, MinorFragmentProfile o2) {
return Long.compare(o1.getMinorFragmentId(), o2.getMinorFragmentId());
}
};
-
+
final static Comparator<MinorFragmentProfile> startTimeCompare = new Comparator<MinorFragmentProfile>() {
public int compare(MinorFragmentProfile o1, MinorFragmentProfile o2) {
return Long.compare(o1.getStartTime(), o2.getStartTime());
@@ -352,25 +425,25 @@ public class ProfileWrapper {
return Long.compare(o1.getEndTime() - o1.getStartTime(), o2.getEndTime() - o2.getStartTime());
}
};
-
+
final static Comparator<OperatorProfile> operatorIdCompare = new Comparator<OperatorProfile>() {
public int compare(OperatorProfile o1, OperatorProfile o2) {
return Long.compare(o1.getOperatorId(), o2.getOperatorId());
}
};
-
+
final static Comparator<Pair<OperatorProfile, Integer>> setupTimeSort = new Comparator<Pair<OperatorProfile, Integer>>() {
public int compare(Pair<OperatorProfile, Integer> o1, Pair<OperatorProfile, Integer> o2) {
return Long.compare(o1.getLeft().getSetupNanos(), o2.getLeft().getSetupNanos());
}
};
-
+
final static Comparator<Pair<OperatorProfile, Integer>> processTimeSort = new Comparator<Pair<OperatorProfile, Integer>>() {
public int compare(Pair<OperatorProfile, Integer> o1, Pair<OperatorProfile, Integer> o2) {
return Long.compare(o1.getLeft().getProcessNanos(), o2.getLeft().getProcessNanos());
}
};
-
+
final static Comparator<Pair<OperatorProfile, Integer>> waitTimeSort = new Comparator<Pair<OperatorProfile, Integer>>() {
public int compare(Pair<OperatorProfile, Integer> o1, Pair<OperatorProfile, Integer> o2) {
return Long.compare(o1.getLeft().getWaitNanos(), o2.getLeft().getWaitNanos());
@@ -395,29 +468,29 @@ public class ProfileWrapper {
final static Predicate<MinorFragmentProfile> missingOperatorsOrTimes = Predicates.not(hasOperatorsAndTimes);
}
-
+
class TableBuilder {
NumberFormat format = NumberFormat.getInstance(Locale.US);
DateFormat dateFormat = new SimpleDateFormat("HH:mm:ss.SSS");
-
+
StringBuilder sb;
int w = 0;
int width;
-
+
public TableBuilder(String[] columns) {
sb = new StringBuilder();
width = columns.length;
-
+
format.setMaximumFractionDigits(3);
format.setMinimumFractionDigits(3);
-
+
sb.append("<table class=\"table table-bordered text-right\">\n<tr>");
for (String cn : columns) {
sb.append("<th>" + cn + "</th>");
}
sb.append("</tr>\n");
}
-
+
public void appendCell(String s, String link) {
if (w == 0) {
sb.append("<tr>");
@@ -428,7 +501,7 @@ public class ProfileWrapper {
w = 0;
}
}
-
+
public void appendRepeated(String s, String link, int n) {
for (int i = 0; i < n; i++) {
appendCell(s, link);
@@ -438,15 +511,15 @@ public class ProfileWrapper {
public void appendTime(long d, String link) {
appendCell(dateFormat.format(d), link);
}
-
+
public void appendMillis(long p, String link) {
appendCell(format.format(p / 1000.0), link);
}
-
+
public void appendNanos(long p, String link) {
appendMillis((long) (p / 1000.0 / 1000.0), link);
}
-
+
public void appendFormattedNumber(Number n, String link) {
appendCell(format.format(n), link);
}
@@ -454,7 +527,7 @@ public class ProfileWrapper {
public void appendInteger(long l, String link) {
appendCell(Long.toString(l), link);
}
-
+
public String toString() {
String rv;
rv = sb.append("\n</table>").toString();
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/ffbf3ae1/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/writer/TestParquetWriter.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/writer/TestParquetWriter.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/writer/TestParquetWriter.java
index cdae265..a61b40f 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/writer/TestParquetWriter.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/writer/TestParquetWriter.java
@@ -26,8 +26,6 @@ import java.util.Map;
import org.apache.drill.BaseTestQuery;
import org.apache.drill.exec.ExecConstants;
import org.apache.drill.exec.exception.SchemaChangeException;
-import org.apache.drill.exec.memory.TopLevelAllocator;
-import org.apache.drill.exec.planner.physical.PlannerSettings;
import org.apache.drill.exec.proto.UserBitShared;
import org.apache.drill.exec.record.BatchSchema;
import org.apache.drill.exec.record.RecordBatchLoader;
@@ -42,8 +40,6 @@ import org.junit.BeforeClass;
import org.junit.Ignore;
import org.junit.Test;
-import java.io.UnsupportedEncodingException;
-
public class TestParquetWriter extends BaseTestQuery {
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestParquetWriter.class);
@@ -225,6 +221,7 @@ public class TestParquetWriter extends BaseTestQuery {
}
@Test
+ @Ignore
public void testParquetRead() throws Exception {
test("alter system set `store.parquet.use_new_reader` = true");
List<QueryResultBatch> expected = testSqlWithResults("select * from dfs.`/tmp/voter`");