From d044cf53098516ef2936cefaf1cd6907134f461f Mon Sep 17 00:00:00 2001 From: Kunal Khatua Date: Tue, 21 Feb 2017 23:06:54 -0800 Subject: [PATCH] DRILL-5190: Display planning and queued time for a query's profile page Modified UserSharedBit protobuf for marking planning and wait-in-queue end times. This will allow for accurately reporting the planning, queued and actual execution times of a query. Planning Time: In the absence of the planning time's end, for older profiles, the root fragment's (i.e. SCREEN operator) start time is taken as the estimated end of planning time, and as the estimated start time of the execution phase. QueueWait Time: We do not estimate the queue time if the planning end time is not available. Execution Time: We calculate the execution time based on the availability of these 2 planning time. The computation is done the following way, and reflects a decreasing level of accuracy 1. Execution time = [end(QueueWait) - endTime(Query)] 2. Execution time = [end(Planning) - endTime(Query)] 3. Execution time = [start(rootFragment) - endTime(Query)] - {Estimated} --- .../server/rest/profile/ProfileWrapper.java | 86 ++++++ .../drill/exec/work/foreman/Foreman.java | 5 + .../drill/exec/work/foreman/QueryManager.java | 12 + .../main/resources/rest/profile/profile.ftl | 3 + .../drill/exec/proto/SchemaUserBitShared.java | 14 + .../drill/exec/proto/UserBitShared.java | 271 ++++++++++++++---- .../drill/exec/proto/beans/QueryProfile.java | 44 +++ .../src/main/protobuf/UserBitShared.proto | 2 + 8 files changed, 383 insertions(+), 54 deletions(-) diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/profile/ProfileWrapper.java b/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/profile/ProfileWrapper.java index 20de752deab..cb80f760b48 100644 --- a/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/profile/ProfileWrapper.java +++ b/exec/java-exec/src/main/java/org/apache/drill/exec/server/rest/profile/ProfileWrapper.java @@ -31,6 +31,7 @@ import org.apache.drill.exec.proto.UserBitShared.MinorFragmentProfile; import org.apache.drill.exec.proto.UserBitShared.OperatorProfile; import org.apache.drill.exec.proto.UserBitShared.QueryProfile; +import org.apache.drill.exec.proto.UserBitShared.QueryResult.QueryState; import org.apache.drill.exec.proto.helper.QueryIdHelper; import org.apache.drill.exec.server.options.OptionList; import org.apache.drill.exec.server.options.OptionValue; @@ -41,6 +42,8 @@ * Wrapper class for a {@link #profile query profile}, so it to be presented through web UI. */ public class ProfileWrapper { + private static final String ESTIMATED_LABEL = " (Estimated)"; + private static final String NOT_AVAILABLE_LABEL = "Not Available"; private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ProfileWrapper.class); private static final ObjectMapper mapper = new ObjectMapper().enable(INDENT_OUTPUT); @@ -122,6 +125,89 @@ public String getQueryId() { return id; } + public String getPlanningDuration() { + //Check if Planning End is known + if (profile.getPlanEnd() > 0L) { + return (new SimpleDurationFormat(profile.getStart(), profile.getPlanEnd())).verbose(); + } + + //Check if any fragments have started + if (profile.getFragmentProfileCount() > 0) { + //Init Planning End Time + long estimatedPlanEnd = Long.MAX_VALUE; + //Using Screen MajorFragment as reference + MajorFragmentProfile majorFrag0 = profile.getFragmentProfile(0); + //Searching for earliest starting fragment + for (MinorFragmentProfile fragmentWrapper : majorFrag0.getMinorFragmentProfileList()) { + long minorFragmentStart = fragmentWrapper.getStartTime(); + if (minorFragmentStart > 0 && minorFragmentStart < estimatedPlanEnd) { + estimatedPlanEnd = minorFragmentStart; + } + } + //Provide estimated plan time + return (new SimpleDurationFormat(profile.getStart(), estimatedPlanEnd)).verbose() + ESTIMATED_LABEL; + } + + //Unable to estimate/calculate Specific Time spent in Planning + return NOT_AVAILABLE_LABEL; + } + + public String getQueuedDuration() { + //Check if State is ENQUEUED + if (profile.getState() == QueryState.ENQUEUED) { + return (new SimpleDurationFormat(profile.getPlanEnd(), System.currentTimeMillis())).verbose(); + } + + //Check if Queue Wait End is known + if (profile.getQueueWaitEnd() > 0L) { + return (new SimpleDurationFormat(profile.getPlanEnd(), profile.getQueueWaitEnd())).verbose(); + } + + //Unable to estimate/calculate Specific Time spent in Queue + return NOT_AVAILABLE_LABEL; + } + + public String getExecutionDuration() { + //Check if State is STARTING or RUNNING + if (profile.getState() == QueryState.STARTING || + profile.getState() == QueryState.ENQUEUED || + profile.getState() == QueryState.RUNNING) { + return NOT_AVAILABLE_LABEL; + } + + //Check if QueueEnd is known + if (profile.getQueueWaitEnd() > 0L) { + //Execution time [end(QueueWait) - endTime(Query)] + return (new SimpleDurationFormat(profile.getQueueWaitEnd(), profile.getEnd())).verbose(); + } + + //Check if Plan End is known + if (profile.getPlanEnd() > 0L) { + //Execution time [end(Planning) - endTime(Query)] + return (new SimpleDurationFormat(profile.getPlanEnd(), profile.getEnd())).verbose(); + } + + //Check if any fragments have started + if (profile.getFragmentProfileCount() > 0) { + //Providing Invalid Planning End Time (Will update later) + long estimatedPlanEnd = Long.MAX_VALUE; + //Using Screen MajorFragment as reference + MajorFragmentProfile majorFrag0 = profile.getFragmentProfile(0); + //Searching for earliest starting fragment + for (MinorFragmentProfile fragmentWrapper : majorFrag0.getMinorFragmentProfileList()) { + long minorFragmentStart = fragmentWrapper.getStartTime(); + if (minorFragmentStart > 0 && minorFragmentStart < estimatedPlanEnd) { + estimatedPlanEnd = minorFragmentStart; + } + } + //Execution time [start(rootFragment) - endTime(Query)] + return (new SimpleDurationFormat(estimatedPlanEnd, profile.getEnd())).verbose() + ESTIMATED_LABEL; + } + + //Unable to estimate/calculate Specific Execution Time + return NOT_AVAILABLE_LABEL; + } + public List getFragmentProfiles() { return fragmentProfiles; } diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/Foreman.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/Foreman.java index 30718b6ef94..4dd6731d332 100644 --- a/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/Foreman.java +++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/Foreman.java @@ -418,9 +418,14 @@ private void parseAndRunPhysicalPlan(final String json) throws ExecutionSetupExc private void runPhysicalPlan(final PhysicalPlan plan) throws ExecutionSetupException { validatePlan(plan); MemoryAllocationUtilities.setupSortMemoryAllocations(plan, queryContext); + //Marking endTime of Planning + queryManager.markPlanningEndTime(); + if (queuingEnabled) { acquireQuerySemaphore(plan); moveToState(QueryState.STARTING, null); + //Marking endTime of Waiting in Queue + queryManager.markQueueWaitEndTime(); } final QueryWorkUnit work = getQueryWorkUnit(plan); diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/QueryManager.java b/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/QueryManager.java index 25b88b80e2a..c3bde6e7157 100644 --- a/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/QueryManager.java +++ b/exec/java-exec/src/main/java/org/apache/drill/exec/work/foreman/QueryManager.java @@ -100,6 +100,8 @@ public class QueryManager implements AutoCloseable { private String planText; private long startTime = System.currentTimeMillis(); private long endTime; + private long planningEndTime; + private long queueWaitEndTime; // How many nodes have finished their execution. Query is complete when all nodes are complete. private final AtomicInteger finishedNodes = new AtomicInteger(0); @@ -341,6 +343,8 @@ private QueryProfile getQueryProfile(UserException ex) { .setForeman(foreman.getQueryContext().getCurrentEndpoint()) .setStart(startTime) .setEnd(endTime) + .setPlanEnd(planningEndTime) + .setQueueWaitEnd(queueWaitEndTime) .setTotalFragments(fragmentDataSet.size()) .setFinishedFragments(finishedFragments.get()) .setOptionsJson(getQueryOptionsAsJson()); @@ -416,6 +420,14 @@ void markEndTime() { endTime = System.currentTimeMillis(); } + void markPlanningEndTime() { + planningEndTime = System.currentTimeMillis(); + } + + void markQueueWaitEndTime() { + queueWaitEndTime = System.currentTimeMillis(); + } + /** * Internal class used to track the number of pending completion messages required from particular node. This allows * to know for each node that is part of this query, what portion of fragments are still outstanding. In the case that diff --git a/exec/java-exec/src/main/resources/rest/profile/profile.ftl b/exec/java-exec/src/main/resources/rest/profile/profile.ftl index e9a8632dda8..39561950fd8 100644 --- a/exec/java-exec/src/main/resources/rest/profile/profile.ftl +++ b/exec/java-exec/src/main/resources/rest/profile/profile.ftl @@ -107,6 +107,9 @@

FOREMAN: ${model.getProfile().getForeman().getAddress()}

TOTAL FRAGMENTS: ${model.getProfile().getTotalFragments()}

DURATION: ${model.getProfileDuration()}

+

    PLANNING: ${model.getPlanningDuration()}

+

    QUEUED: ${model.getQueuedDuration()}

+

    EXECUTION: ${model.getExecutionDuration()}

<#assign options = model.getOptions()> <#if (options?keys?size > 0)> diff --git a/protocol/src/main/java/org/apache/drill/exec/proto/SchemaUserBitShared.java b/protocol/src/main/java/org/apache/drill/exec/proto/SchemaUserBitShared.java index f31110d91c9..ef061b44690 100644 --- a/protocol/src/main/java/org/apache/drill/exec/proto/SchemaUserBitShared.java +++ b/protocol/src/main/java/org/apache/drill/exec/proto/SchemaUserBitShared.java @@ -1790,6 +1790,10 @@ public void writeTo(com.dyuproject.protostuff.Output output, org.apache.drill.ex output.writeString(16, message.getErrorNode(), false); if(message.hasOptionsJson()) output.writeString(17, message.getOptionsJson(), false); + if(message.hasPlanEnd()) + output.writeInt64(18, message.getPlanEnd(), false); + if(message.hasQueueWaitEnd()) + output.writeInt64(19, message.getQueueWaitEnd(), false); } public boolean isInitialized(org.apache.drill.exec.proto.UserBitShared.QueryProfile message) { @@ -1883,6 +1887,12 @@ public void mergeFrom(com.dyuproject.protostuff.Input input, org.apache.drill.ex case 17: builder.setOptionsJson(input.readString()); break; + case 18: + builder.setPlanEnd(input.readInt64()); + break; + case 19: + builder.setQueueWaitEnd(input.readInt64()); + break; default: input.handleUnknownField(number, this); } @@ -1940,6 +1950,8 @@ public static java.lang.String getFieldName(int number) case 15: return "errorId"; case 16: return "errorNode"; case 17: return "optionsJson"; + case 18: return "planEnd"; + case 19: return "queueWaitEnd"; default: return null; } } @@ -1968,6 +1980,8 @@ public static int getFieldNumber(java.lang.String name) fieldMap.put("errorId", 15); fieldMap.put("errorNode", 16); fieldMap.put("optionsJson", 17); + fieldMap.put("planEnd", 18); + fieldMap.put("queueWaitEnd", 19); } } diff --git a/protocol/src/main/java/org/apache/drill/exec/proto/UserBitShared.java b/protocol/src/main/java/org/apache/drill/exec/proto/UserBitShared.java index 9a61cbb2a2d..c3eb7a0d57e 100644 --- a/protocol/src/main/java/org/apache/drill/exec/proto/UserBitShared.java +++ b/protocol/src/main/java/org/apache/drill/exec/proto/UserBitShared.java @@ -13314,6 +13314,26 @@ org.apache.drill.exec.proto.UserBitShared.MajorFragmentProfileOrBuilder getFragm */ com.google.protobuf.ByteString getOptionsJsonBytes(); + + // optional int64 planEnd = 18; + /** + * optional int64 planEnd = 18; + */ + boolean hasPlanEnd(); + /** + * optional int64 planEnd = 18; + */ + long getPlanEnd(); + + // optional int64 queueWaitEnd = 19; + /** + * optional int64 queueWaitEnd = 19; + */ + boolean hasQueueWaitEnd(); + /** + * optional int64 queueWaitEnd = 19; + */ + long getQueueWaitEnd(); } /** * Protobuf type {@code exec.shared.QueryProfile} @@ -13482,6 +13502,16 @@ private QueryProfile( optionsJson_ = input.readBytes(); break; } + case 144: { + bitField0_ |= 0x00010000; + planEnd_ = input.readInt64(); + break; + } + case 152: { + bitField0_ |= 0x00020000; + queueWaitEnd_ = input.readInt64(); + break; + } } } } catch (com.google.protobuf.InvalidProtocolBufferException e) { @@ -14045,6 +14075,38 @@ public java.lang.String getOptionsJson() { } } + // optional int64 planEnd = 18; + public static final int PLANEND_FIELD_NUMBER = 18; + private long planEnd_; + /** + * optional int64 planEnd = 18; + */ + public boolean hasPlanEnd() { + return ((bitField0_ & 0x00010000) == 0x00010000); + } + /** + * optional int64 planEnd = 18; + */ + public long getPlanEnd() { + return planEnd_; + } + + // optional int64 queueWaitEnd = 19; + public static final int QUEUEWAITEND_FIELD_NUMBER = 19; + private long queueWaitEnd_; + /** + * optional int64 queueWaitEnd = 19; + */ + public boolean hasQueueWaitEnd() { + return ((bitField0_ & 0x00020000) == 0x00020000); + } + /** + * optional int64 queueWaitEnd = 19; + */ + public long getQueueWaitEnd() { + return queueWaitEnd_; + } + private void initFields() { id_ = org.apache.drill.exec.proto.UserBitShared.QueryId.getDefaultInstance(); type_ = org.apache.drill.exec.proto.UserBitShared.QueryType.SQL; @@ -14063,6 +14125,8 @@ private void initFields() { errorId_ = ""; errorNode_ = ""; optionsJson_ = ""; + planEnd_ = 0L; + queueWaitEnd_ = 0L; } private byte memoizedIsInitialized = -1; public final boolean isInitialized() { @@ -14127,6 +14191,12 @@ public void writeTo(com.google.protobuf.CodedOutputStream output) if (((bitField0_ & 0x00008000) == 0x00008000)) { output.writeBytes(17, getOptionsJsonBytes()); } + if (((bitField0_ & 0x00010000) == 0x00010000)) { + output.writeInt64(18, planEnd_); + } + if (((bitField0_ & 0x00020000) == 0x00020000)) { + output.writeInt64(19, queueWaitEnd_); + } getUnknownFields().writeTo(output); } @@ -14204,6 +14274,14 @@ public int getSerializedSize() { size += com.google.protobuf.CodedOutputStream .computeBytesSize(17, getOptionsJsonBytes()); } + if (((bitField0_ & 0x00010000) == 0x00010000)) { + size += com.google.protobuf.CodedOutputStream + .computeInt64Size(18, planEnd_); + } + if (((bitField0_ & 0x00020000) == 0x00020000)) { + size += com.google.protobuf.CodedOutputStream + .computeInt64Size(19, queueWaitEnd_); + } size += getUnknownFields().getSerializedSize(); memoizedSerializedSize = size; return size; @@ -14369,6 +14447,10 @@ public Builder clear() { bitField0_ = (bitField0_ & ~0x00008000); optionsJson_ = ""; bitField0_ = (bitField0_ & ~0x00010000); + planEnd_ = 0L; + bitField0_ = (bitField0_ & ~0x00020000); + queueWaitEnd_ = 0L; + bitField0_ = (bitField0_ & ~0x00040000); return this; } @@ -14478,6 +14560,14 @@ public org.apache.drill.exec.proto.UserBitShared.QueryProfile buildPartial() { to_bitField0_ |= 0x00008000; } result.optionsJson_ = optionsJson_; + if (((from_bitField0_ & 0x00020000) == 0x00020000)) { + to_bitField0_ |= 0x00010000; + } + result.planEnd_ = planEnd_; + if (((from_bitField0_ & 0x00040000) == 0x00040000)) { + to_bitField0_ |= 0x00020000; + } + result.queueWaitEnd_ = queueWaitEnd_; result.bitField0_ = to_bitField0_; onBuilt(); return result; @@ -14584,6 +14674,12 @@ public Builder mergeFrom(org.apache.drill.exec.proto.UserBitShared.QueryProfile optionsJson_ = other.optionsJson_; onChanged(); } + if (other.hasPlanEnd()) { + setPlanEnd(other.getPlanEnd()); + } + if (other.hasQueueWaitEnd()) { + setQueueWaitEnd(other.getQueueWaitEnd()); + } this.mergeUnknownFields(other.getUnknownFields()); return this; } @@ -15881,6 +15977,72 @@ public Builder setOptionsJsonBytes( return this; } + // optional int64 planEnd = 18; + private long planEnd_ ; + /** + * optional int64 planEnd = 18; + */ + public boolean hasPlanEnd() { + return ((bitField0_ & 0x00020000) == 0x00020000); + } + /** + * optional int64 planEnd = 18; + */ + public long getPlanEnd() { + return planEnd_; + } + /** + * optional int64 planEnd = 18; + */ + public Builder setPlanEnd(long value) { + bitField0_ |= 0x00020000; + planEnd_ = value; + onChanged(); + return this; + } + /** + * optional int64 planEnd = 18; + */ + public Builder clearPlanEnd() { + bitField0_ = (bitField0_ & ~0x00020000); + planEnd_ = 0L; + onChanged(); + return this; + } + + // optional int64 queueWaitEnd = 19; + private long queueWaitEnd_ ; + /** + * optional int64 queueWaitEnd = 19; + */ + public boolean hasQueueWaitEnd() { + return ((bitField0_ & 0x00040000) == 0x00040000); + } + /** + * optional int64 queueWaitEnd = 19; + */ + public long getQueueWaitEnd() { + return queueWaitEnd_; + } + /** + * optional int64 queueWaitEnd = 19; + */ + public Builder setQueueWaitEnd(long value) { + bitField0_ |= 0x00040000; + queueWaitEnd_ = value; + onChanged(); + return this; + } + /** + * optional int64 queueWaitEnd = 19; + */ + public Builder clearQueueWaitEnd() { + bitField0_ = (bitField0_ & ~0x00040000); + queueWaitEnd_ = 0L; + onChanged(); + return this; + } + // @@protoc_insertion_point(builder_scope:exec.shared.QueryProfile) } @@ -22535,7 +22697,7 @@ public Builder addFunctionSignatureBytes( "rt\030\002 \001(\003\0222\n\005state\030\003 \001(\0162#.exec.shared.Qu", "eryResult.QueryState\022\017\n\004user\030\004 \001(\t:\001-\022\'\n" + "\007foreman\030\005 \001(\0132\026.exec.DrillbitEndpoint\022\024" + - "\n\014options_json\030\006 \001(\t\"\320\003\n\014QueryProfile\022 \n" + + "\n\014options_json\030\006 \001(\t\"\367\003\n\014QueryProfile\022 \n" + "\002id\030\001 \001(\0132\024.exec.shared.QueryId\022$\n\004type\030" + "\002 \001(\0162\026.exec.shared.QueryType\022\r\n\005start\030\003" + " \001(\003\022\013\n\003end\030\004 \001(\003\022\r\n\005query\030\005 \001(\t\022\014\n\004plan" + @@ -22547,58 +22709,59 @@ public Builder addFunctionSignatureBytes( "agmentProfile\022\017\n\004user\030\014 \001(\t:\001-\022\r\n\005error\030" + "\r \001(\t\022\024\n\014verboseError\030\016 \001(\t\022\020\n\010error_id\030" + "\017 \001(\t\022\022\n\nerror_node\030\020 \001(\t\022\024\n\014options_jso" + - "n\030\021 \001(\t\"t\n\024MajorFragmentProfile\022\031\n\021major" + - "_fragment_id\030\001 \001(\005\022A\n\026minor_fragment_pro" + - "file\030\002 \003(\0132!.exec.shared.MinorFragmentPr" + - "ofile\"\350\002\n\024MinorFragmentProfile\022)\n\005state\030" + - "\001 \001(\0162\032.exec.shared.FragmentState\022(\n\005err" + - "or\030\002 \001(\0132\031.exec.shared.DrillPBError\022\031\n\021m", - "inor_fragment_id\030\003 \001(\005\0226\n\020operator_profi" + - "le\030\004 \003(\0132\034.exec.shared.OperatorProfile\022\022" + - "\n\nstart_time\030\005 \001(\003\022\020\n\010end_time\030\006 \001(\003\022\023\n\013" + - "memory_used\030\007 \001(\003\022\027\n\017max_memory_used\030\010 \001" + - "(\003\022(\n\010endpoint\030\t \001(\0132\026.exec.DrillbitEndp" + - "oint\022\023\n\013last_update\030\n \001(\003\022\025\n\rlast_progre" + - "ss\030\013 \001(\003\"\377\001\n\017OperatorProfile\0221\n\rinput_pr" + - "ofile\030\001 \003(\0132\032.exec.shared.StreamProfile\022" + - "\023\n\013operator_id\030\003 \001(\005\022\025\n\roperator_type\030\004 " + - "\001(\005\022\023\n\013setup_nanos\030\005 \001(\003\022\025\n\rprocess_nano", - "s\030\006 \001(\003\022#\n\033peak_local_memory_allocated\030\007" + - " \001(\003\022(\n\006metric\030\010 \003(\0132\030.exec.shared.Metri" + - "cValue\022\022\n\nwait_nanos\030\t \001(\003\"B\n\rStreamProf" + - "ile\022\017\n\007records\030\001 \001(\003\022\017\n\007batches\030\002 \001(\003\022\017\n" + - "\007schemas\030\003 \001(\003\"J\n\013MetricValue\022\021\n\tmetric_" + - "id\030\001 \001(\005\022\022\n\nlong_value\030\002 \001(\003\022\024\n\014double_v" + - "alue\030\003 \001(\001\")\n\010Registry\022\035\n\003jar\030\001 \003(\0132\020.ex" + - "ec.shared.Jar\"/\n\003Jar\022\014\n\004name\030\001 \001(\t\022\032\n\022fu" + - "nction_signature\030\002 \003(\t*5\n\nRpcChannel\022\017\n\013" + - "BIT_CONTROL\020\000\022\014\n\010BIT_DATA\020\001\022\010\n\004USER\020\002*V\n", - "\tQueryType\022\007\n\003SQL\020\001\022\013\n\007LOGICAL\020\002\022\014\n\010PHYS" + - "ICAL\020\003\022\r\n\tEXECUTION\020\004\022\026\n\022PREPARED_STATEM" + - "ENT\020\005*\207\001\n\rFragmentState\022\013\n\007SENDING\020\000\022\027\n\023" + - "AWAITING_ALLOCATION\020\001\022\013\n\007RUNNING\020\002\022\014\n\010FI" + - "NISHED\020\003\022\r\n\tCANCELLED\020\004\022\n\n\006FAILED\020\005\022\032\n\026C" + - "ANCELLATION_REQUESTED\020\006*\335\005\n\020CoreOperator" + - "Type\022\021\n\rSINGLE_SENDER\020\000\022\024\n\020BROADCAST_SEN" + - "DER\020\001\022\n\n\006FILTER\020\002\022\022\n\016HASH_AGGREGATE\020\003\022\r\n" + - "\tHASH_JOIN\020\004\022\016\n\nMERGE_JOIN\020\005\022\031\n\025HASH_PAR" + - "TITION_SENDER\020\006\022\t\n\005LIMIT\020\007\022\024\n\020MERGING_RE", - "CEIVER\020\010\022\034\n\030ORDERED_PARTITION_SENDER\020\t\022\013" + - "\n\007PROJECT\020\n\022\026\n\022UNORDERED_RECEIVER\020\013\022\020\n\014R" + - "ANGE_SENDER\020\014\022\n\n\006SCREEN\020\r\022\034\n\030SELECTION_V" + - "ECTOR_REMOVER\020\016\022\027\n\023STREAMING_AGGREGATE\020\017" + - "\022\016\n\nTOP_N_SORT\020\020\022\021\n\rEXTERNAL_SORT\020\021\022\t\n\005T" + - "RACE\020\022\022\t\n\005UNION\020\023\022\014\n\010OLD_SORT\020\024\022\032\n\026PARQU" + - "ET_ROW_GROUP_SCAN\020\025\022\021\n\rHIVE_SUB_SCAN\020\026\022\025" + - "\n\021SYSTEM_TABLE_SCAN\020\027\022\021\n\rMOCK_SUB_SCAN\020\030" + - "\022\022\n\016PARQUET_WRITER\020\031\022\023\n\017DIRECT_SUB_SCAN\020" + - "\032\022\017\n\013TEXT_WRITER\020\033\022\021\n\rTEXT_SUB_SCAN\020\034\022\021\n", - "\rJSON_SUB_SCAN\020\035\022\030\n\024INFO_SCHEMA_SUB_SCAN" + - "\020\036\022\023\n\017COMPLEX_TO_JSON\020\037\022\025\n\021PRODUCER_CONS" + - "UMER\020 \022\022\n\016HBASE_SUB_SCAN\020!\022\n\n\006WINDOW\020\"\022\024" + - "\n\020NESTED_LOOP_JOIN\020#\022\021\n\rAVRO_SUB_SCAN\020$B" + - ".\n\033org.apache.drill.exec.protoB\rUserBitS" + - "haredH\001" + "n\030\021 \001(\t\022\017\n\007planEnd\030\022 \001(\003\022\024\n\014queueWaitEnd" + + "\030\023 \001(\003\"t\n\024MajorFragmentProfile\022\031\n\021major_" + + "fragment_id\030\001 \001(\005\022A\n\026minor_fragment_prof" + + "ile\030\002 \003(\0132!.exec.shared.MinorFragmentPro" + + "file\"\350\002\n\024MinorFragmentProfile\022)\n\005state\030\001" + + " \001(\0162\032.exec.shared.FragmentState\022(\n\005erro", + "r\030\002 \001(\0132\031.exec.shared.DrillPBError\022\031\n\021mi" + + "nor_fragment_id\030\003 \001(\005\0226\n\020operator_profil" + + "e\030\004 \003(\0132\034.exec.shared.OperatorProfile\022\022\n" + + "\nstart_time\030\005 \001(\003\022\020\n\010end_time\030\006 \001(\003\022\023\n\013m" + + "emory_used\030\007 \001(\003\022\027\n\017max_memory_used\030\010 \001(" + + "\003\022(\n\010endpoint\030\t \001(\0132\026.exec.DrillbitEndpo" + + "int\022\023\n\013last_update\030\n \001(\003\022\025\n\rlast_progres" + + "s\030\013 \001(\003\"\377\001\n\017OperatorProfile\0221\n\rinput_pro" + + "file\030\001 \003(\0132\032.exec.shared.StreamProfile\022\023" + + "\n\013operator_id\030\003 \001(\005\022\025\n\roperator_type\030\004 \001", + "(\005\022\023\n\013setup_nanos\030\005 \001(\003\022\025\n\rprocess_nanos" + + "\030\006 \001(\003\022#\n\033peak_local_memory_allocated\030\007 " + + "\001(\003\022(\n\006metric\030\010 \003(\0132\030.exec.shared.Metric" + + "Value\022\022\n\nwait_nanos\030\t \001(\003\"B\n\rStreamProfi" + + "le\022\017\n\007records\030\001 \001(\003\022\017\n\007batches\030\002 \001(\003\022\017\n\007" + + "schemas\030\003 \001(\003\"J\n\013MetricValue\022\021\n\tmetric_i" + + "d\030\001 \001(\005\022\022\n\nlong_value\030\002 \001(\003\022\024\n\014double_va" + + "lue\030\003 \001(\001\")\n\010Registry\022\035\n\003jar\030\001 \003(\0132\020.exe" + + "c.shared.Jar\"/\n\003Jar\022\014\n\004name\030\001 \001(\t\022\032\n\022fun" + + "ction_signature\030\002 \003(\t*5\n\nRpcChannel\022\017\n\013B", + "IT_CONTROL\020\000\022\014\n\010BIT_DATA\020\001\022\010\n\004USER\020\002*V\n\t" + + "QueryType\022\007\n\003SQL\020\001\022\013\n\007LOGICAL\020\002\022\014\n\010PHYSI" + + "CAL\020\003\022\r\n\tEXECUTION\020\004\022\026\n\022PREPARED_STATEME" + + "NT\020\005*\207\001\n\rFragmentState\022\013\n\007SENDING\020\000\022\027\n\023A" + + "WAITING_ALLOCATION\020\001\022\013\n\007RUNNING\020\002\022\014\n\010FIN" + + "ISHED\020\003\022\r\n\tCANCELLED\020\004\022\n\n\006FAILED\020\005\022\032\n\026CA" + + "NCELLATION_REQUESTED\020\006*\335\005\n\020CoreOperatorT" + + "ype\022\021\n\rSINGLE_SENDER\020\000\022\024\n\020BROADCAST_SEND" + + "ER\020\001\022\n\n\006FILTER\020\002\022\022\n\016HASH_AGGREGATE\020\003\022\r\n\t" + + "HASH_JOIN\020\004\022\016\n\nMERGE_JOIN\020\005\022\031\n\025HASH_PART", + "ITION_SENDER\020\006\022\t\n\005LIMIT\020\007\022\024\n\020MERGING_REC" + + "EIVER\020\010\022\034\n\030ORDERED_PARTITION_SENDER\020\t\022\013\n" + + "\007PROJECT\020\n\022\026\n\022UNORDERED_RECEIVER\020\013\022\020\n\014RA" + + "NGE_SENDER\020\014\022\n\n\006SCREEN\020\r\022\034\n\030SELECTION_VE" + + "CTOR_REMOVER\020\016\022\027\n\023STREAMING_AGGREGATE\020\017\022" + + "\016\n\nTOP_N_SORT\020\020\022\021\n\rEXTERNAL_SORT\020\021\022\t\n\005TR" + + "ACE\020\022\022\t\n\005UNION\020\023\022\014\n\010OLD_SORT\020\024\022\032\n\026PARQUE" + + "T_ROW_GROUP_SCAN\020\025\022\021\n\rHIVE_SUB_SCAN\020\026\022\025\n" + + "\021SYSTEM_TABLE_SCAN\020\027\022\021\n\rMOCK_SUB_SCAN\020\030\022" + + "\022\n\016PARQUET_WRITER\020\031\022\023\n\017DIRECT_SUB_SCAN\020\032", + "\022\017\n\013TEXT_WRITER\020\033\022\021\n\rTEXT_SUB_SCAN\020\034\022\021\n\r" + + "JSON_SUB_SCAN\020\035\022\030\n\024INFO_SCHEMA_SUB_SCAN\020" + + "\036\022\023\n\017COMPLEX_TO_JSON\020\037\022\025\n\021PRODUCER_CONSU" + + "MER\020 \022\022\n\016HBASE_SUB_SCAN\020!\022\n\n\006WINDOW\020\"\022\024\n" + + "\020NESTED_LOOP_JOIN\020#\022\021\n\rAVRO_SUB_SCAN\020$B." + + "\n\033org.apache.drill.exec.protoB\rUserBitSh" + + "aredH\001" }; com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner assigner = new com.google.protobuf.Descriptors.FileDescriptor.InternalDescriptorAssigner() { @@ -22688,7 +22851,7 @@ public com.google.protobuf.ExtensionRegistry assignDescriptors( internal_static_exec_shared_QueryProfile_fieldAccessorTable = new com.google.protobuf.GeneratedMessage.FieldAccessorTable( internal_static_exec_shared_QueryProfile_descriptor, - new java.lang.String[] { "Id", "Type", "Start", "End", "Query", "Plan", "Foreman", "State", "TotalFragments", "FinishedFragments", "FragmentProfile", "User", "Error", "VerboseError", "ErrorId", "ErrorNode", "OptionsJson", }); + new java.lang.String[] { "Id", "Type", "Start", "End", "Query", "Plan", "Foreman", "State", "TotalFragments", "FinishedFragments", "FragmentProfile", "User", "Error", "VerboseError", "ErrorId", "ErrorNode", "OptionsJson", "PlanEnd", "QueueWaitEnd", }); internal_static_exec_shared_MajorFragmentProfile_descriptor = getDescriptor().getMessageTypes().get(14); internal_static_exec_shared_MajorFragmentProfile_fieldAccessorTable = new diff --git a/protocol/src/main/java/org/apache/drill/exec/proto/beans/QueryProfile.java b/protocol/src/main/java/org/apache/drill/exec/proto/beans/QueryProfile.java index 150db2671dd..216ce6313f8 100644 --- a/protocol/src/main/java/org/apache/drill/exec/proto/beans/QueryProfile.java +++ b/protocol/src/main/java/org/apache/drill/exec/proto/beans/QueryProfile.java @@ -68,6 +68,8 @@ public static QueryProfile getDefaultInstance() private String errorId; private String errorNode; private String optionsJson; + private long planEnd; + private long queueWaitEnd; public QueryProfile() { @@ -297,6 +299,32 @@ public QueryProfile setOptionsJson(String optionsJson) return this; } + // planEnd + + public long getPlanEnd() + { + return planEnd; + } + + public QueryProfile setPlanEnd(long planEnd) + { + this.planEnd = planEnd; + return this; + } + + // queueWaitEnd + + public long getQueueWaitEnd() + { + return queueWaitEnd; + } + + public QueryProfile setQueueWaitEnd(long queueWaitEnd) + { + this.queueWaitEnd = queueWaitEnd; + return this; + } + // java serialization public void readExternal(ObjectInput in) throws IOException @@ -407,6 +435,12 @@ public void mergeFrom(Input input, QueryProfile message) throws IOException case 17: message.optionsJson = input.readString(); break; + case 18: + message.planEnd = input.readInt64(); + break; + case 19: + message.queueWaitEnd = input.readInt64(); + break; default: input.handleUnknownField(number, this); } @@ -475,6 +509,12 @@ public void writeTo(Output output, QueryProfile message) throws IOException if(message.optionsJson != null) output.writeString(17, message.optionsJson, false); + + if(message.planEnd != 0) + output.writeInt64(18, message.planEnd, false); + + if(message.queueWaitEnd != 0) + output.writeInt64(19, message.queueWaitEnd, false); } public String getFieldName(int number) @@ -498,6 +538,8 @@ public String getFieldName(int number) case 15: return "errorId"; case 16: return "errorNode"; case 17: return "optionsJson"; + case 18: return "planEnd"; + case 19: return "queueWaitEnd"; default: return null; } } @@ -528,6 +570,8 @@ public int getFieldNumber(String name) __fieldMap.put("errorId", 15); __fieldMap.put("errorNode", 16); __fieldMap.put("optionsJson", 17); + __fieldMap.put("planEnd", 18); + __fieldMap.put("queueWaitEnd", 19); } } diff --git a/protocol/src/main/protobuf/UserBitShared.proto b/protocol/src/main/protobuf/UserBitShared.proto index 5efca2d20ed..0ba7e060331 100644 --- a/protocol/src/main/protobuf/UserBitShared.proto +++ b/protocol/src/main/protobuf/UserBitShared.proto @@ -207,6 +207,8 @@ message QueryProfile { optional string error_id = 15; optional string error_node = 16; optional string options_json = 17; + optional int64 planEnd = 18; + optional int64 queueWaitEnd = 19; } message MajorFragmentProfile {