Skip to content

Commit 818b6fc

Browse files
JackieTien97JackieTien97
authored
[To dev/1.3] Optimize aggregation metric recording
1 parent 073625a commit 818b6fc

9 files changed

Lines changed: 434 additions & 144 deletions

File tree

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/aggregation/Aggregator.java‎

Lines changed: 34 additions & 58 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,6 @@
1919

2020
package org.apache.iotdb.db.queryengine.execution.aggregation;
2121

22-
import org.apache.iotdb.db.queryengine.metric.QueryExecutionMetricSet;
2322
import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.AggregationStep;
2423
import org.apache.iotdb.db.queryengine.plan.planner.plan.parameter.InputLocation;
2524

@@ -34,17 +33,13 @@
3433
import java.util.List;
3534

3635
import static com.google.common.base.Preconditions.checkArgument;
37-
import static org.apache.iotdb.db.queryengine.metric.QueryExecutionMetricSet.AGGREGATION_FROM_RAW_DATA;
38-
import static org.apache.iotdb.db.queryengine.metric.QueryExecutionMetricSet.AGGREGATION_FROM_STATISTICS;
3936

4037
public class Aggregator {
4138

4239
protected final Accumulator accumulator;
4340
// In some intermediate result input, inputLocation[] should include two columns
4441
protected List<InputLocation[]> inputLocationList;
4542
protected final AggregationStep step;
46-
protected static final QueryExecutionMetricSet QUERY_EXECUTION_METRICS =
47-
QueryExecutionMetricSet.getInstance();
4843

4944
// Used for SeriesAggregateScanOperator
5045
public Aggregator(Accumulator accumulator, AggregationStep step) {
@@ -64,57 +59,44 @@ public Aggregator(
6459

6560
// Used for SeriesAggregateScanOperator and RawDataAggregateOperator
6661
public void processTsBlock(TsBlock tsBlock, BitMap bitMap) {
67-
long startTime = System.nanoTime();
68-
try {
69-
checkArgument(
70-
step.isInputRaw(),
71-
"Step in SeriesAggregateScanOperator and RawDataAggregateOperator can only process raw input");
72-
for (InputLocation[] inputLocations : inputLocationList) {
73-
Column[] timeAndValueColumn = new Column[1 + inputLocations.length];
74-
timeAndValueColumn[0] = tsBlock.getTimeColumn();
75-
for (int i = 0; i < inputLocations.length; i++) {
76-
checkArgument(
77-
inputLocations[i].getTsBlockIndex() == 0,
78-
"RawDataAggregateOperator can only process one tsBlock input.");
79-
int index = inputLocations[i].getValueColumnIndex();
80-
// for count_time, time column is also its value column
81-
// for max_by, the input column can also be time column.
82-
timeAndValueColumn[1 + i] =
83-
index == -1 ? timeAndValueColumn[0] : tsBlock.getColumn(index);
84-
}
85-
accumulator.addInput(timeAndValueColumn, bitMap);
62+
checkArgument(
63+
step.isInputRaw(),
64+
"Step in SeriesAggregateScanOperator and RawDataAggregateOperator can only process raw input");
65+
for (InputLocation[] inputLocations : inputLocationList) {
66+
Column[] timeAndValueColumn = new Column[1 + inputLocations.length];
67+
timeAndValueColumn[0] = tsBlock.getTimeColumn();
68+
for (int i = 0; i < inputLocations.length; i++) {
69+
checkArgument(
70+
inputLocations[i].getTsBlockIndex() == 0,
71+
"RawDataAggregateOperator can only process one tsBlock input.");
72+
int index = inputLocations[i].getValueColumnIndex();
73+
// for count_time, time column is also its value column
74+
// for max_by, the input column can also be time column.
75+
timeAndValueColumn[1 + i] = index == -1 ? timeAndValueColumn[0] : tsBlock.getColumn(index);
8676
}
87-
} finally {
88-
QUERY_EXECUTION_METRICS.recordExecutionCost(
89-
AGGREGATION_FROM_RAW_DATA, System.nanoTime() - startTime);
77+
accumulator.addInput(timeAndValueColumn, bitMap);
9078
}
9179
}
9280

9381
// Used for AggregateOperator
9482
public void processTsBlocks(TsBlock[] tsBlock) {
95-
long startTime = System.nanoTime();
96-
try {
97-
checkArgument(!step.isInputRaw(), "Step in AggregateOperator cannot process raw input");
98-
if (step.isInputFinal()) {
99-
checkArgument(inputLocationList.size() == 1, "Final output can only be single column");
100-
Column finalResult =
101-
tsBlock[inputLocationList.get(0)[0].getTsBlockIndex()].getColumn(
102-
inputLocationList.get(0)[0].getValueColumnIndex());
103-
accumulator.setFinal(finalResult);
104-
} else {
105-
for (InputLocation[] inputLocations : inputLocationList) {
106-
Column[] columns = new Column[inputLocations.length];
107-
for (int i = 0; i < inputLocations.length; i++) {
108-
columns[i] =
109-
tsBlock[inputLocations[i].getTsBlockIndex()].getColumn(
110-
inputLocations[i].getValueColumnIndex());
111-
}
112-
accumulator.addIntermediate(columns);
83+
checkArgument(!step.isInputRaw(), "Step in AggregateOperator cannot process raw input");
84+
if (step.isInputFinal()) {
85+
checkArgument(inputLocationList.size() == 1, "Final output can only be single column");
86+
Column finalResult =
87+
tsBlock[inputLocationList.get(0)[0].getTsBlockIndex()].getColumn(
88+
inputLocationList.get(0)[0].getValueColumnIndex());
89+
accumulator.setFinal(finalResult);
90+
} else {
91+
for (InputLocation[] inputLocations : inputLocationList) {
92+
Column[] columns = new Column[inputLocations.length];
93+
for (int i = 0; i < inputLocations.length; i++) {
94+
columns[i] =
95+
tsBlock[inputLocations[i].getTsBlockIndex()].getColumn(
96+
inputLocations[i].getValueColumnIndex());
11397
}
98+
accumulator.addIntermediate(columns);
11499
}
115-
} finally {
116-
QUERY_EXECUTION_METRICS.recordExecutionCost(
117-
AGGREGATION_FROM_RAW_DATA, System.nanoTime() - startTime);
118100
}
119101
}
120102

@@ -128,16 +110,10 @@ public void outputResult(ColumnBuilder[] columnBuilder) {
128110

129111
/** Used for SeriesAggregateScanOperator. */
130112
public void processStatistics(Statistics timeStatistics, Statistics[] valueStatistics) {
131-
long startTime = System.nanoTime();
132-
try {
133-
for (InputLocation[] inputLocations : inputLocationList) {
134-
int valueIndex = inputLocations[0].getValueColumnIndex();
135-
// valueIndex == -1 means it is count_time, we need to use timeStatistics
136-
accumulator.addStatistics(valueIndex == -1 ? timeStatistics : valueStatistics[valueIndex]);
137-
}
138-
} finally {
139-
QUERY_EXECUTION_METRICS.recordExecutionCost(
140-
AGGREGATION_FROM_STATISTICS, System.nanoTime() - startTime);
113+
for (InputLocation[] inputLocations : inputLocationList) {
114+
int valueIndex = inputLocations[0].getValueColumnIndex();
115+
// valueIndex == -1 means it is count_time, we need to use timeStatistics
116+
accumulator.addStatistics(valueIndex == -1 ? timeStatistics : valueStatistics[valueIndex]);
141117
}
142118
}
143119

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/fragment/FragmentInstanceContext.java‎

Lines changed: 52 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -34,6 +34,7 @@
3434
import org.apache.iotdb.db.queryengine.common.SessionInfo;
3535
import org.apache.iotdb.db.queryengine.exception.MemoryNotEnoughException;
3636
import org.apache.iotdb.db.queryengine.metric.DriverSchedulerMetricSet;
37+
import org.apache.iotdb.db.queryengine.metric.QueryExecutionMetricSet;
3738
import org.apache.iotdb.db.queryengine.metric.QueryRelatedResourceMetricSet;
3839
import org.apache.iotdb.db.queryengine.metric.QueryResourceMetricSet;
3940
import org.apache.iotdb.db.queryengine.metric.SeriesScanCostMetricSet;
@@ -77,6 +78,9 @@
7778

7879
import static org.apache.iotdb.db.queryengine.metric.DriverSchedulerMetricSet.BLOCK_QUEUED_TIME;
7980
import static org.apache.iotdb.db.queryengine.metric.DriverSchedulerMetricSet.READY_QUEUED_TIME;
81+
import static org.apache.iotdb.db.queryengine.metric.QueryExecutionMetricSet.AGGREGATION_FROM_RAW_DATA;
82+
import static org.apache.iotdb.db.queryengine.metric.QueryExecutionMetricSet.AGGREGATION_FROM_STATISTICS;
83+
import static org.apache.iotdb.db.queryengine.metric.QueryExecutionMetricSet.AGGREGATION_OPERATOR_FROM_RAW_DATA;
8084
import static org.apache.iotdb.db.storageengine.dataregion.VirtualDataRegion.EMPTY_QUERY_DATA_SOURCE;
8185
import static org.apache.iotdb.db.storageengine.dataregion.VirtualDataRegion.UNFINISHED_QUERY_DATA_SOURCE;
8286

@@ -149,6 +153,9 @@ public class FragmentInstanceContext extends QueryContext {
149153
private int initQueryDataSourceRetryCount = 0;
150154
private final AtomicLong readyQueueTime = new AtomicLong(0);
151155
private final AtomicLong blockQueueTime = new AtomicLong(0);
156+
private final AtomicLong scanAggregationFromRawDataCost = new AtomicLong(0);
157+
private final AtomicLong scanAggregationFromStatisticsCost = new AtomicLong(0);
158+
private final AtomicLong aggregationOperatorFromRawDataCost = new AtomicLong(0);
152159
private long unclosedSeqFileNum = 0;
153160
private long unclosedUnseqFileNum = 0;
154161
private long closedSeqFileNum = 0;
@@ -970,6 +977,8 @@ public synchronized void releaseResource() {
970977

971978
QueryRelatedResourceMetricSet.getInstance().updateFragmentInstanceTime(durationTime);
972979

980+
recordAggregationCostToMetric();
981+
973982
QueryResourceMetricSet.getInstance()
974983
.recordInitQueryResourceRetryCount(getInitQueryDataSourceRetryCount());
975984

@@ -1115,6 +1124,49 @@ public void addBlockQueuedTime(long time) {
11151124
blockQueueTime.addAndGet(time);
11161125
}
11171126

1127+
public void recordScanAggregationFromRawDataCost(long costTimeInNanos) {
1128+
addCost(scanAggregationFromRawDataCost, costTimeInNanos);
1129+
}
1130+
1131+
public void recordScanAggregationFromStatisticsCost(long costTimeInNanos) {
1132+
addCost(scanAggregationFromStatisticsCost, costTimeInNanos);
1133+
}
1134+
1135+
public void recordAggregationOperatorFromRawDataCost(long costTimeInNanos) {
1136+
addCost(aggregationOperatorFromRawDataCost, costTimeInNanos);
1137+
}
1138+
1139+
private void addCost(AtomicLong cost, long costTimeInNanos) {
1140+
if (costTimeInNanos > 0) {
1141+
cost.addAndGet(costTimeInNanos);
1142+
}
1143+
}
1144+
1145+
long drainScanAggregationFromRawDataCost() {
1146+
return scanAggregationFromRawDataCost.getAndSet(0);
1147+
}
1148+
1149+
long drainScanAggregationFromStatisticsCost() {
1150+
return scanAggregationFromStatisticsCost.getAndSet(0);
1151+
}
1152+
1153+
long drainAggregationOperatorFromRawDataCost() {
1154+
return aggregationOperatorFromRawDataCost.getAndSet(0);
1155+
}
1156+
1157+
void recordAggregationCostToMetric() {
1158+
recordAggregationCost(AGGREGATION_FROM_RAW_DATA, drainScanAggregationFromRawDataCost());
1159+
recordAggregationCost(AGGREGATION_FROM_STATISTICS, drainScanAggregationFromStatisticsCost());
1160+
recordAggregationCost(
1161+
AGGREGATION_OPERATOR_FROM_RAW_DATA, drainAggregationOperatorFromRawDataCost());
1162+
}
1163+
1164+
private void recordAggregationCost(String stage, long costTimeInNanos) {
1165+
if (costTimeInNanos > 0) {
1166+
QueryExecutionMetricSet.getInstance().recordExecutionCost(stage, costTimeInNanos);
1167+
}
1168+
}
1169+
11181170
public long getReadyQueueTime() {
11191171
return readyQueueTime.get();
11201172
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/OperatorContext.java‎

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -119,6 +119,30 @@ public SessionInfo getSessionInfo() {
119119
return getInstanceContext().getSessionInfo();
120120
}
121121

122+
public void recordScanAggregationFromRawDataCost(long costTimeInNanos) {
123+
if (driverContext != null && driverContext.getFragmentInstanceContext() != null) {
124+
driverContext
125+
.getFragmentInstanceContext()
126+
.recordScanAggregationFromRawDataCost(costTimeInNanos);
127+
}
128+
}
129+
130+
public void recordScanAggregationFromStatisticsCost(long costTimeInNanos) {
131+
if (driverContext != null && driverContext.getFragmentInstanceContext() != null) {
132+
driverContext
133+
.getFragmentInstanceContext()
134+
.recordScanAggregationFromStatisticsCost(costTimeInNanos);
135+
}
136+
}
137+
138+
public void recordAggregationOperatorFromRawDataCost(long costTimeInNanos) {
139+
if (driverContext != null && driverContext.getFragmentInstanceContext() != null) {
140+
driverContext
141+
.getFragmentInstanceContext()
142+
.recordAggregationOperatorFromRawDataCost(costTimeInNanos);
143+
}
144+
}
145+
122146
public PlanNodeId getPlanNodeId() {
123147
return planNodeId;
124148
}

‎iotdb-core/datanode/src/main/java/org/apache/iotdb/db/queryengine/execution/operator/process/AggregationOperator.java‎

Lines changed: 15 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -153,19 +153,24 @@ public boolean isFinished() throws Exception {
153153

154154
private void calculateNextAggregationResult() {
155155
// Consume current input tsBlocks
156-
for (Aggregator aggregator : aggregators) {
157-
aggregator.processTsBlocks(inputTsBlocks);
158-
}
156+
long startTime = System.nanoTime();
157+
try {
158+
for (Aggregator aggregator : aggregators) {
159+
aggregator.processTsBlocks(inputTsBlocks);
160+
}
159161

160-
for (int i = 0; i < inputOperatorsCount; i++) {
161-
inputTsBlocks[i] = inputTsBlocks[i].skipFirst();
162-
if (inputTsBlocks[i].isEmpty()) {
163-
inputTsBlocks[i] = null;
162+
for (int i = 0; i < inputOperatorsCount; i++) {
163+
inputTsBlocks[i] = inputTsBlocks[i].skipFirst();
164+
if (inputTsBlocks[i].isEmpty()) {
165+
inputTsBlocks[i] = null;
166+
}
164167
}
165-
}
166168

167-
// Update result using aggregators
168-
updateResultTsBlock();
169+
// Update result using aggregators
170+
updateResultTsBlock();
171+
} finally {
172+
operatorContext.recordAggregationOperatorFromRawDataCost(System.nanoTime() - startTime);
173+
}
169174
}
170175

171176
private void updateResultTsBlock() {

0 commit comments

Comments
 (0)