This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new a6d64c6b25f [refine](profilev2) format profilev2 (#25963)
a6d64c6b25f is described below
commit a6d64c6b25f27537b8e140e26d1e3d8c684e1553
Author: Mryange <[email protected]>
AuthorDate: Fri Oct 27 14:26:38 2023 +0800
[refine](profilev2) format profilev2 (#25963)
---
be/src/pipeline/exec/exchange_sink_operator.cpp | 8 ++
be/src/pipeline/exec/exchange_sink_operator.h | 1 +
.../org/apache/doris/common/profile/Profile.java | 2 +-
.../org/apache/doris/common/util/AggCounter.java | 66 +++++++++
.../java/org/apache/doris/common/util/Counter.java | 26 ++--
.../apache/doris/common/util/RuntimeProfile.java | 154 +++++++++++++++++++--
.../org/apache/doris/planner/PlanFragment.java | 8 ++
.../java/org/apache/doris/planner/PlanNode.java | 22 +++
.../java/org/apache/doris/planner/Planner.java | 11 ++
9 files changed, 278 insertions(+), 20 deletions(-)
diff --git a/be/src/pipeline/exec/exchange_sink_operator.cpp
b/be/src/pipeline/exec/exchange_sink_operator.cpp
index 0988052a981..0ca38d81c13 100644
--- a/be/src/pipeline/exec/exchange_sink_operator.cpp
+++ b/be/src/pipeline/exec/exchange_sink_operator.cpp
@@ -235,6 +235,14 @@ Status ExchangeSinkLocalState::open(RuntimeState* state) {
return Status::OK();
}
+std::string ExchangeSinkLocalState::id_name() {
+ std::string name = " (id=" + std::to_string(_parent->node_id());
+ auto& p = _parent->cast<ExchangeSinkOperatorX>();
+ name += ",dest_id=" + std::to_string(p._dest_node_id);
+ name += ")";
+ return name;
+}
+
segment_v2::CompressionTypePB& ExchangeSinkLocalState::compression_type() {
return _parent->cast<ExchangeSinkOperatorX>()._compression_type;
}
diff --git a/be/src/pipeline/exec/exchange_sink_operator.h
b/be/src/pipeline/exec/exchange_sink_operator.h
index 69fa4da4be4..05fa799e5cf 100644
--- a/be/src/pipeline/exec/exchange_sink_operator.h
+++ b/be/src/pipeline/exec/exchange_sink_operator.h
@@ -159,6 +159,7 @@ public:
[[nodiscard]] int sender_id() const { return _sender_id; }
+ std::string id_name() override;
segment_v2::CompressionTypePB& compression_type();
std::vector<vectorized::PipChannel<ExchangeSinkLocalState>*> channels;
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/common/profile/Profile.java
b/fe/fe-core/src/main/java/org/apache/doris/common/profile/Profile.java
index e75082ddb42..2d4424177a3 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/common/profile/Profile.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/common/profile/Profile.java
@@ -72,7 +72,7 @@ public class Profile {
executionProfile.update(startTime, isFinished);
}
rootProfile.computeTimeInProfile();
- rootProfile.setPlaner(planner);
+ rootProfile.setFragmentPlanInfo(planner);
rootProfile.setProfileLevel(profileLevel);
rootProfile.setIsPipelineX(isPipelineX);
ProfileManager.getInstance().pushProfile(rootProfile);
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/common/util/AggCounter.java
b/fe/fe-core/src/main/java/org/apache/doris/common/util/AggCounter.java
new file mode 100644
index 00000000000..ed8f91533fb
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/common/util/AggCounter.java
@@ -0,0 +1,66 @@
+// 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.doris.common.util;
+
+import java.util.LinkedList;
+
+// Counter means indicators field. The counter's name is key, the counter
itself is value.
+public class AggCounter extends Counter {
+ Counter max;
+ Counter sum;
+ Counter min;
+ int number;
+
+ public AggCounter(org.apache.doris.thrift.TUnit type, long value) {
+ super(type, value);
+ max = new Counter(type, value);
+ sum = new Counter(type, value);
+ min = new Counter(type, value);
+ number = 1;
+ }
+
+ public void addCounter(Counter counter) {
+ max.maxValue(counter);
+ sum.addValue(counter);
+ min.minValue(counter);
+ number++;
+ }
+
+ public void addCounters(LinkedList<Counter> rhsCounter) {
+ for (Counter counter : rhsCounter) {
+ addCounter(counter);
+ }
+ }
+
+ public String print() {
+ if (isTimeType()) {
+ Counter avg = new Counter(sum.getType(), sum.getValue());
+ avg.divValue(number);
+ String infoString = RuntimeProfile.AVG_TIME_PRE
+ + RuntimeProfile.printCounter(avg.getValue(),
avg.getType()) + ", "
+ + RuntimeProfile.MAX_TIME_PRE
+ + RuntimeProfile.printCounter(max.getValue(),
max.getType()) + ", "
+ + RuntimeProfile.MIN_TIME_PRE
+ + RuntimeProfile.printCounter(min.getValue(),
min.getType());
+ return infoString;
+ } else {
+ String infoString = RuntimeProfile.printCounter(sum.getValue(),
sum.getType());
+ return infoString;
+ }
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/util/Counter.java
b/fe/fe-core/src/main/java/org/apache/doris/common/util/Counter.java
index 2beb845438a..af9cd1e6cd6 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/common/util/Counter.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/common/util/Counter.java
@@ -23,7 +23,6 @@ import org.apache.doris.thrift.TUnit;
public class Counter {
private volatile long value;
private volatile int type;
- private volatile boolean remove = false;
private volatile long level;
public long getValue() {
@@ -70,6 +69,20 @@ public class Counter {
this.value += other.value;
}
+ public void minValue(Counter other) {
+ if (other == null) {
+ return;
+ }
+ this.value = Math.min(this.value, other.value);
+ }
+
+ public void maxValue(Counter other) {
+ if (other == null) {
+ return;
+ }
+ this.value = Math.max(this.value, other.value);
+ }
+
public void divValue(long div) {
if (div <= 0) {
return;
@@ -82,15 +95,12 @@ public class Counter {
return ttype == TUnit.TIME_MS || ttype == TUnit.TIME_NS || ttype ==
TUnit.TIME_S;
}
- public void setCanRemove() {
- this.remove = true;
+ public long getLevel() {
+ return this.level;
}
- public boolean isRemove() {
- return this.remove;
+ public String print() {
+ return RuntimeProfile.printCounter(value, getType());
}
- public long getLevel() {
- return this.level;
- }
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/common/util/RuntimeProfile.java
b/fe/fe-core/src/main/java/org/apache/doris/common/util/RuntimeProfile.java
index 3519ebce254..4e0602e8079 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/common/util/RuntimeProfile.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/common/util/RuntimeProfile.java
@@ -69,6 +69,7 @@ public class RuntimeProfile {
private LinkedList<Pair<RuntimeProfile, Boolean>> childList =
Lists.newLinkedList();
private ReentrantReadWriteLock childLock = new ReentrantReadWriteLock();
+ private List<String> planNodeInfos = Lists.newArrayList();
private String name;
private Long timestamp = -1L;
@@ -83,7 +84,7 @@ public class RuntimeProfile {
private Boolean isSinkOperator = false;
private int profileLevel = 3;
- private Planner planner = null;
+ private Map<Integer, String> planNodeMap = null;
private int nodeid = -1;
public RuntimeProfile(String name) {
@@ -92,6 +93,13 @@ public class RuntimeProfile {
this.counterTotalTime = new Counter(TUnit.TIME_NS, 0, 1);
}
+ public RuntimeProfile(String name, int nodeId) {
+ this();
+ this.name = name;
+ this.counterTotalTime = new Counter(TUnit.TIME_NS, 0, 3);
+ this.nodeid = nodeId;
+ }
+
public RuntimeProfile() {
this.counterTotalTime = new Counter(TUnit.TIME_NS, 0, 1);
this.localTimePercent = 0;
@@ -134,6 +142,10 @@ public class RuntimeProfile {
this.isPipelineX = isPipelineX;
}
+ public boolean getIsPipelineX() {
+ return this.isPipelineX;
+ }
+
public Map<String, Counter> getCounterMap() {
return counterMap;
}
@@ -179,6 +191,29 @@ public class RuntimeProfile {
}
}
+ public void addCounter(String name, Counter newCounter, String
parentCounterName) {
+ counterLock.writeLock().lock();
+ try {
+ Counter counter = this.counterMap.get(name);
+ if (counter != null) {
+ return;
+ } else {
+ Preconditions.checkState(parentCounterName.equals(ROOT_COUNTER)
+ || this.counterMap.containsKey(parentCounterName));
+ this.counterMap.put(name, newCounter);
+
+ Set<String> childCounters =
childCounterMap.get(parentCounterName);
+ if (childCounters == null) {
+ childCounterMap.put(parentCounterName, new
TreeSet<String>());
+ childCounters = childCounterMap.get(parentCounterName);
+ }
+ childCounters.add(name);
+ }
+ } finally {
+ counterLock.writeLock().unlock();
+ }
+ }
+
public void update(final TRuntimeProfileTree thriftProfile) {
Reference<Integer> idx = new Reference<Integer>(0);
update(thriftProfile.nodes, idx);
@@ -339,6 +374,9 @@ public class RuntimeProfile {
}
builder.append("\n");
+ // plan node info
+ printPlanNodeInfo(prefix + " ", builder);
+
// 2. info String
infoStringsLock.readLock().lock();
try {
@@ -373,6 +411,17 @@ public class RuntimeProfile {
}
}
+ private void printPlanNodeInfo(String prefix, StringBuilder builder) {
+ if (planNodeInfos.isEmpty()) {
+ return;
+ }
+ builder.append(prefix + "- " + "PlanInfo\n");
+
+ for (String info : planNodeInfos) {
+ builder.append(prefix + " - " + info + "\n");
+ }
+ }
+
public void simpleProfile(int depth, int childIdx, ProfileStatistics
statistics) {
if (depth == FRAGMENT_DEPTH) {
statistics.setFragmentId(childIdx);
@@ -528,15 +577,69 @@ public class RuntimeProfile {
if (this.profileLevel == 3) {
return toString();
}
- if (this.planner == null) {
+ if (this.planNodeMap == null) {
return toString();
}
- StringBuilder builder = new StringBuilder();
- prettyPrint(builder, "");
- ProfileStatistics statistics = new ProfileStatistics(this.isPipelineX);
- simpleProfile(0, 0, statistics);
- String planerStr = this.planner.getExplainStringToProfile(statistics);
- return "Simple profile \n \n " + planerStr + "\n \n \n" +
builder.toString();
+ RuntimeProfile simpleProfile = new RuntimeProfile("SimpleProfile");
+ getSimpleProfile(0, simpleProfile, this.planNodeMap);
+ return simpleProfile.toString() + " \n \n " + toString();
+ }
+
+ public void getSimpleProfile(int depth, RuntimeProfile simpleProfile,
Map<Integer, String> planNodeMap) {
+ if (depth == FRAGMENT_DEPTH) {
+ gettSimpleProfileFromMutiInstance(childList, simpleProfile,
planNodeMap);
+ return;
+ }
+ for (int i = 0; i < childList.size(); i++) {
+ Pair<RuntimeProfile, Boolean> pair = childList.get(i);
+ RuntimeProfile profile = pair.first;
+ profile.getSimpleProfile(depth + 1, simpleProfile, planNodeMap);
+ }
+ }
+
+ public static void
gettSimpleProfileFromMutiInstance(LinkedList<Pair<RuntimeProfile, Boolean>>
childList,
+ RuntimeProfile simpleProfile, Map<Integer, String> planNodeMap) {
+ RuntimeProfile oneProfile = childList.get(0).first;
+ int instanceNum = childList.size();
+ RuntimeProfile mergedProfile = new RuntimeProfile("Instance" + "(" +
instanceNum + ")", oneProfile.nodeid);
+ LinkedList<RuntimeProfile> other = new LinkedList<RuntimeProfile>();
+ for (int i = 1; i < childList.size(); i++) {
+ other.add(childList.get(i).first);
+ }
+ simpleProfile.addChildWithCheck(mergedProfile, planNodeMap);
+ collecteProfile(oneProfile, other, mergedProfile, planNodeMap);
+ }
+
+ public static void collecteProfile(RuntimeProfile src,
LinkedList<RuntimeProfile> others,
+ RuntimeProfile simpleProfile, Map<Integer, String> planNodeMap) {
+ collecteProfileCounter(src, ROOT_COUNTER, others, simpleProfile);
+ for (int i = 0; i < src.childList.size(); i++) {
+ RuntimeProfile srcChild = src.childList.get(i).first;
+ LinkedList<RuntimeProfile> rhsChild = getChildListFromLists(i,
others);
+ RuntimeProfile childProfile = new RuntimeProfile(srcChild.name,
srcChild.nodeId());
+ simpleProfile.addChildWithCheck(childProfile, planNodeMap);
+ collecteProfile(srcChild, rhsChild, childProfile, planNodeMap);
+ }
+ }
+
+ private static void collecteProfileCounter(RuntimeProfile src, String
counterName, LinkedList<RuntimeProfile> rhs,
+ RuntimeProfile simpleProfile) {
+ Set<String> childCounterSet = src.childCounterMap.get(counterName);
+ if (childCounterSet == null) {
+ return;
+ }
+ List<String> childCounterList = new LinkedList<>(childCounterSet);
+ for (String childCounterName : childCounterList) {
+ Counter counter = src.counterMap.get(childCounterName);
+ collecteProfileCounter(src, childCounterName, rhs, simpleProfile);
+ if (counter.getLevel() == 1) {
+ LinkedList<Counter> rhsCounter =
getCounterListFromLists(childCounterName, rhs);
+ // String info = getMergeString(counter, rhsCounter);
+ AggCounter aggCounter = new AggCounter(counter.getType(),
counter.getValue());
+ aggCounter.addCounters(rhsCounter);
+ simpleProfile.addCounter(childCounterName, aggCounter,
ROOT_COUNTER);
+ }
+ }
}
private void printChildCounters(String prefix, String counterName,
StringBuilder builder) {
@@ -551,7 +654,7 @@ public class RuntimeProfile {
Counter counter = this.counterMap.get(childCounterName);
Preconditions.checkState(counter != null);
builder.append(prefix).append(" -
").append(childCounterName).append(": ")
- .append(printCounter(counter.getValue(),
counter.getType())).append("\n");
+ .append(counter.print()).append("\n");
this.printChildCounters(prefix + " ", childCounterName,
builder);
}
} finally {
@@ -657,6 +760,33 @@ public class RuntimeProfile {
}
}
+ public void addChildWithCheck(RuntimeProfile child, Map<Integer, String>
planNodeMap) {
+ // check name
+ if (child.name.startsWith("PipelineTask") ||
child.name.startsWith("PipelineContext")) {
+ return;
+ }
+ childLock.writeLock().lock();
+ try {
+ Pair<RuntimeProfile, Boolean> pair = Pair.of(child, true);
+ this.childList.add(pair);
+ } finally {
+ childLock.writeLock().unlock();
+ }
+ // insert plan node info to profile strinfo
+ if (!planNodeMap.containsKey(child.nodeId())) {
+ return;
+ }
+ child.addPlanNodeInfos(planNodeMap.get(child.nodeId()));
+ planNodeMap.remove(child.nodeId());
+ }
+
+ public void addPlanNodeInfos(String infos) {
+ String[] infoList = infos.split("\n");
+ for (String info : infoList) {
+ planNodeInfos.add(info);
+ }
+ }
+
public void addFirstChild(RuntimeProfile child) {
if (child == null) {
return;
@@ -689,8 +819,10 @@ public class RuntimeProfile {
this.profileLevel = profileLevel;
}
- public void setPlaner(Planner planner) {
- this.planner = planner;
+ public void setFragmentPlanInfo(Planner planner) {
+ if (planner != null) {
+ this.planNodeMap = planner.getExplainStringMap();
+ }
}
private void computeTimeInProfile(long total) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/planner/PlanFragment.java
b/fe/fe-core/src/main/java/org/apache/doris/planner/PlanFragment.java
index 26c09e439e6..3324f9de1b1 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/PlanFragment.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/PlanFragment.java
@@ -43,6 +43,7 @@ import org.apache.logging.log4j.Logger;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
@@ -355,6 +356,13 @@ public class PlanFragment extends TreeNode<PlanFragment> {
return str.toString();
}
+ public void getExplainStringMap(Map<Integer, String> planNodeMap) {
+ org.apache.doris.thrift.TExplainLevel explainLevel =
org.apache.doris.thrift.TExplainLevel.NORMAL;
+ if (planRoot != null) {
+ planRoot.getExplainStringMap(explainLevel, planNodeMap);
+ }
+ }
+
/**
* Returns true if this fragment is partitioned.
*/
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java
b/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java
index 754b9fcfcb2..c323b3160fb 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java
@@ -63,6 +63,7 @@ import java.util.Collection;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
+import java.util.Map;
import java.util.Set;
/**
@@ -631,6 +632,27 @@ public abstract class PlanNode extends TreeNode<PlanNode>
implements PlanStats {
return expBuilder.toString();
}
+ private String getplanNodeExplainString(String prefix, TExplainLevel
detailLevel) {
+ StringBuilder expBuilder = new StringBuilder();
+ expBuilder.append(getNodeExplainString(prefix, detailLevel));
+ if (limit != -1) {
+ expBuilder.append(prefix + "limit: " + limit + "\n");
+ }
+ if (!CollectionUtils.isEmpty(projectList)) {
+ expBuilder.append(prefix).append("projections:
").append(getExplainString(projectList)).append("\n");
+ expBuilder.append(prefix).append("project output tuple id: ")
+ .append(outputTupleDesc.getId().asInt()).append("\n");
+ }
+ return expBuilder.toString();
+ }
+
+ public void getExplainStringMap(TExplainLevel detailLevel, Map<Integer,
String> planNodeMap) {
+ planNodeMap.put(id.asInt(), getplanNodeExplainString("", detailLevel));
+ for (int i = 0; i < children.size(); ++i) {
+ children.get(i).getExplainStringMap(detailLevel, planNodeMap);
+ }
+ }
+
/**
* Return the node-specific details.
* Subclass should override this function.
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/Planner.java
b/fe/fe-core/src/main/java/org/apache/doris/planner/Planner.java
index 8681455a399..e03c696dfd9 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/Planner.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/Planner.java
@@ -39,7 +39,9 @@ import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.util.ArrayList;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.Optional;
public abstract class Planner {
@@ -105,6 +107,15 @@ public abstract class Planner {
return str.toString();
}
+ public Map<Integer, String> getExplainStringMap() {
+ Map<Integer, String> planNodeMap = new HashMap<Integer, String>();
+ for (int i = 0; i < fragments.size(); ++i) {
+ PlanFragment fragment = fragments.get(i);
+ fragment.getExplainStringMap(planNodeMap);
+ }
+ return planNodeMap;
+ }
+
protected void handleLiteralInFe(LiteralExpr literalExpr, List<String>
data) {
if (literalExpr instanceof NullLiteral) {
data.add(null);
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
