>From Ian Maxon <[email protected]>:
Ian Maxon has uploaded this change for review. (
https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21367?usp=email )
Change subject: [WIP] Add bytes written/spilled on sort to profile
......................................................................
[WIP] Add bytes written/spilled on sort to profile
Change-Id: Ia6e37080a581292744ddc90b0b8149d7c13c19a3
---
M asterixdb/asterix-app/src/test/resources/runtimets/profiled.xml
A
asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/profile/sort-spill/sort-spill.1.ddl.sqlpp
A
asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/profile/sort-spill/sort-spill.2.update.sqlpp
A
asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/profile/sort-spill/sort-spill.3.plans.sqlpp
A
asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/profile/sort-spill/sort-spill.90.ddl.sqlpp
A
asterixdb/asterix-app/src/test/resources/runtimets/results/profile/sort-spill/sort-spill.3.regex
M
hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/prettyprint/LogicalOperatorPrettyPrintVisitorJson.java
M
hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/AbstractSortRunGenerator.java
M
hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/AbstractSorterOperatorDescriptor.java
M
hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/IRunGenerator.java
10 files changed, 192 insertions(+), 5 deletions(-)
git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb
refs/changes/67/21367/1
diff --git a/asterixdb/asterix-app/src/test/resources/runtimets/profiled.xml
b/asterixdb/asterix-app/src/test/resources/runtimets/profiled.xml
index f05e5a0..aadd9fd 100644
--- a/asterixdb/asterix-app/src/test/resources/runtimets/profiled.xml
+++ b/asterixdb/asterix-app/src/test/resources/runtimets/profiled.xml
@@ -91,5 +91,13 @@
<output-dir compare="Clean-JSON">groupby</output-dir>
</compilation-unit>
</test-case>
+ <test-case FilePath="profile">
+ <compilation-unit name="sort-spill">
+ <parameter name="profile" value="timings" type="string"/>
+ <parameter name="optimized-logical-plan" value="true" type="string"/>
+ <parameter name="plan-format" value="json" type="string"/>
+ <output-dir compare="Clean-JSON">sort-spill</output-dir>
+ </compilation-unit>
+ </test-case>
</test-group>
</test-suite>
diff --git
a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/profile/sort-spill/sort-spill.1.ddl.sqlpp
b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/profile/sort-spill/sort-spill.1.ddl.sqlpp
new file mode 100644
index 0000000..b7632d2
--- /dev/null
+++
b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/profile/sort-spill/sort-spill.1.ddl.sqlpp
@@ -0,0 +1,46 @@
+/*
+ * 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.
+ */
+
+drop dataverse tpch if exists;
+create dataverse tpch;
+
+use tpch;
+
+
+create type tpch.LineItemType as
+ closed {
+ l_orderkey : bigint,
+ l_partkey : bigint,
+ l_suppkey : bigint,
+ l_linenumber : bigint,
+ l_quantity : double,
+ l_extendedprice : double,
+ l_discount : double,
+ l_tax : double,
+ l_returnflag : string,
+ l_linestatus : string,
+ l_shipdate : string,
+ l_commitdate : string,
+ l_receiptdate : string,
+ l_shipinstruct : string,
+ l_shipmode : string,
+ l_comment : string
+};
+
+create dataset LineItem(LineItemType) primary key l_orderkey,l_linenumber;
diff --git
a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/profile/sort-spill/sort-spill.2.update.sqlpp
b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/profile/sort-spill/sort-spill.2.update.sqlpp
new file mode 100644
index 0000000..f0c2d09
--- /dev/null
+++
b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/profile/sort-spill/sort-spill.2.update.sqlpp
@@ -0,0 +1,23 @@
+/*
+ * 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.
+ */
+
+use tpch;
+
+
+load dataset LineItem using localfs
((`path`=`asterix_nc1://data/tpch0.001/lineitem.tbl`),(`format`=`delimited-text`),(`delimiter`=`|`))
pre-sorted;
diff --git
a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/profile/sort-spill/sort-spill.3.plans.sqlpp
b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/profile/sort-spill/sort-spill.3.plans.sqlpp
new file mode 100644
index 0000000..1695448
--- /dev/null
+++
b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/profile/sort-spill/sort-spill.3.plans.sqlpp
@@ -0,0 +1,26 @@
+/*
+ * 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.
+ */
+
+use tpch;
+
+set `compiler.sortmemory` "128KB";
+
+select value l
+from LineItem as l
+order by l.l_comment;
diff --git
a/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/profile/sort-spill/sort-spill.90.ddl.sqlpp
b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/profile/sort-spill/sort-spill.90.ddl.sqlpp
new file mode 100644
index 0000000..23f4949
--- /dev/null
+++
b/asterixdb/asterix-app/src/test/resources/runtimets/queries_sqlpp/profile/sort-spill/sort-spill.90.ddl.sqlpp
@@ -0,0 +1,20 @@
+/*
+ * 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.
+ */
+
+drop dataverse tpch;
diff --git
a/asterixdb/asterix-app/src/test/resources/runtimets/results/profile/sort-spill/sort-spill.3.regex
b/asterixdb/asterix-app/src/test/resources/runtimets/results/profile/sort-spill/sort-spill.3.regex
new file mode 100644
index 0000000..9767897
--- /dev/null
+++
b/asterixdb/asterix-app/src/test/resources/runtimets/results/profile/sort-spill/sort-spill.3.regex
@@ -0,0 +1,2 @@
+/"operator"\s*:\s*"order"/
+/"bytes-spilled"\s*:\s*"[0-9.]+ (?:bytes|[KMGT]B)"/
diff --git
a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/prettyprint/LogicalOperatorPrettyPrintVisitorJson.java
b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/prettyprint/LogicalOperatorPrettyPrintVisitorJson.java
index 9052bfa..6699f6a 100644
---
a/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/prettyprint/LogicalOperatorPrettyPrintVisitorJson.java
+++
b/hyracks-fullstack/algebricks/algebricks-core/src/main/java/org/apache/hyracks/algebricks/core/algebra/prettyprint/LogicalOperatorPrettyPrintVisitorJson.java
@@ -130,6 +130,10 @@
private static final String TUPLE_BYTES = "tuple-bytes";
+ private static final String BYTES_SPILLED = "bytes-spilled";
+
+ private static final String BYTES_READ_FROM_DISK = "bytes-read-from-disk";
+
private static final String AVG_TUPLES_PER_FRAME = "avg-tuples-per-frame";
private static final String NAME = "name";
private static final String ID = "id";
@@ -306,6 +310,8 @@
Map<String, Long> activityFrameCountTotal;
Map<String, WeightedAvg> activityAvgTuplesPerFrame;
Map<String, Long> activityTupleBytes;
+ Map<String, Long> activityBytesSpilled;
+ Map<String, Long> activityBytesReadFromDisk;
OperatorProfile() {
activityNames = new HashMap<>();
@@ -317,6 +323,8 @@
activityFrameCountTotal = new HashMap<>();
activityAvgTuplesPerFrame = new HashMap<>();
activityTupleBytes = new HashMap<>();
+ activityBytesSpilled = new HashMap<>();
+ activityBytesReadFromDisk = new HashMap<>();
}
void updateOperator(String extendedOpId, String name, double time,
long cardinality, long minSize, long maxSize,
@@ -342,6 +350,14 @@
updateMinMax(time, extendedOpId, activityTimes);
}
+ void updateBytesSpilled(String extendedOpId, long bytes) {
+ activityBytesSpilled.compute(extendedOpId, (id, total) -> total ==
null ? bytes : total + bytes);
+ }
+
+ void updateBytesReadFromDisk(String extendedOpId, long bytes) {
+ activityBytesReadFromDisk.compute(extendedOpId, (id, total) ->
total == null ? bytes : total + bytes);
+ }
+
private <T extends Comparable<T>> void updateMinMax(T comp, String id,
Map<String, MinMax<T>> opMap) {
MinMax<T> stat = opMap.computeIfAbsent(id, i -> new
MinMax<>(comp));
stat.update(comp);
@@ -386,6 +402,15 @@
info.updateOperator(identities.first.getActivityAndLocalId(), identities.second,
counters.get("run-time").asDouble());
}
+ JsonNode bytesSpilled = counters.get("bytes-written");
+ if (bytesSpilled != null) {
+
info.updateBytesSpilled(identities.first.getActivityAndLocalId(),
bytesSpilled.asLong());
+ }
+ JsonNode bytesReadFromDisk = counters.get("bytes-read");
+ if (bytesReadFromDisk != null) {
+
info.updateBytesReadFromDisk(identities.first.getActivityAndLocalId(),
+ bytesReadFromDisk.asLong());
+ }
}
for (JsonNode partition : task.get("partition-send-profile")) {
String id =
partition.get("partition-id").get("connector-id").asText();
@@ -460,8 +485,8 @@
private void printActivityStats(Optional<MinMax<Double>> time,
Optional<MinMax<Long>> card,
Optional<Long> totalCard, Optional<MinMax<Long>> tupleSize,
Optional<WeightedAvg> avgTupleSize,
- Optional<Long> frameCount, Optional<WeightedAvg>
avgTuplesPerFrame, Optional<Long> tupleBytes)
- throws IOException {
+ Optional<Long> frameCount, Optional<WeightedAvg>
avgTuplesPerFrame, Optional<Long> tupleBytes,
+ Optional<Long> bytesSpilled, Optional<Long> bytesReadFromDisk)
throws IOException {
if (time.isPresent()) {
jsonGenerator.writeNumberField(MIN_TIME, time.get().min);
jsonGenerator.writeNumberField(MAX_TIME, time.get().max);
@@ -489,6 +514,13 @@
if (tupleBytes.isPresent()) {
jsonGenerator.writeStringField(TUPLE_BYTES,
FileUtils.byteCountToDisplaySize(tupleBytes.get()));
}
+ if (bytesSpilled.isPresent()) {
+ jsonGenerator.writeStringField(BYTES_SPILLED,
FileUtils.byteCountToDisplaySize(bytesSpilled.get()));
+ }
+ if (bytesReadFromDisk.isPresent()) {
+ jsonGenerator.writeStringField(BYTES_READ_FROM_DISK,
+ FileUtils.byteCountToDisplaySize(bytesReadFromDisk.get()));
+ }
}
private void printOperatorStats(OperatorProfile info) throws IOException {
@@ -501,8 +533,10 @@
Optional<Long> frameCount =
info.activityFrameCountTotal.values().stream().findFirst();
Optional<WeightedAvg> avgTuplesPerFrame =
info.activityAvgTuplesPerFrame.values().stream().findFirst();
Optional<Long> tupleBytes =
info.activityTupleBytes.values().stream().findFirst();
- printActivityStats(times, cards, total, tupleSizes, avgTupleSize,
frameCount, avgTuplesPerFrame,
- tupleBytes);
+ Optional<Long> bytesSpilled =
info.activityBytesSpilled.values().stream().findFirst();
+ Optional<Long> bytesReadFromDisk =
info.activityBytesReadFromDisk.values().stream().findFirst();
+ printActivityStats(times, cards, total, tupleSizes, avgTupleSize,
frameCount, avgTuplesPerFrame, tupleBytes,
+ bytesSpilled, bytesReadFromDisk);
} else if (info.activityTimes.size() > 1) {
jsonGenerator.writeArrayFieldStart("activity-stats");
for (String acId : info.activityTimes.keySet()) {
@@ -519,7 +553,9 @@
Optional.ofNullable(info.activityAvgTupleSizes.get(acId)),
Optional.ofNullable(info.activityFrameCountTotal.get(acId)),
Optional.ofNullable(info.activityAvgTuplesPerFrame.get(acId)),
-
Optional.ofNullable(info.activityTupleBytes.get(acId)));
+ Optional.ofNullable(info.activityTupleBytes.get(acId)),
+
Optional.ofNullable(info.activityBytesSpilled.get(acId)),
+
Optional.ofNullable(info.activityBytesReadFromDisk.get(acId)));
jsonGenerator.writeEndObject();
}
jsonGenerator.writeEndArray();
diff --git
a/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/AbstractSortRunGenerator.java
b/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/AbstractSortRunGenerator.java
index 3f0b7c7..fe7cb51 100644
---
a/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/AbstractSortRunGenerator.java
+++
b/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/AbstractSortRunGenerator.java
@@ -24,17 +24,25 @@
import org.apache.hyracks.api.comm.IFrameWriter;
import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.job.profiling.IOperatorStats;
+import org.apache.hyracks.api.job.profiling.NoOpOperatorStats;
import org.apache.hyracks.dataflow.common.io.GeneratedRunFileReader;
import org.apache.hyracks.dataflow.common.io.RunFileWriter;
public abstract class AbstractSortRunGenerator implements IRunGenerator {
private final List<GeneratedRunFileReader> generatedRunFileReaders;
+ private IOperatorStats stats = NoOpOperatorStats.INSTANCE;
public AbstractSortRunGenerator() {
generatedRunFileReaders = new LinkedList<>();
}
+ @Override
+ public void setStats(IOperatorStats stats) {
+ this.stats = stats;
+ }
+
/**
* Null could be returned. Caller should check if it not null.
* @return the sorter associated with the run generator or null if there
is no sorter.
@@ -77,6 +85,9 @@
} finally {
flushWriter.close();
}
+ if (stats != null) {
+ stats.getBytesWritten().update(runWriter.getFileSize());
+ }
generatedRunFileReaders.add(runWriter.createDeleteOnCloseReader());
sorter.reset();
}
diff --git
a/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/AbstractSorterOperatorDescriptor.java
b/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/AbstractSorterOperatorDescriptor.java
index 6ff79b7..f285d5d4 100644
---
a/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/AbstractSorterOperatorDescriptor.java
+++
b/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/AbstractSorterOperatorDescriptor.java
@@ -37,6 +37,7 @@
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.job.IOperatorDescriptorRegistry;
import org.apache.hyracks.api.job.JobId;
+import org.apache.hyracks.api.job.profiling.IOperatorStats;
import org.apache.hyracks.dataflow.common.io.GeneratedRunFileReader;
import org.apache.hyracks.dataflow.std.base.AbstractActivityNode;
import org.apache.hyracks.dataflow.std.base.AbstractOperatorDescriptor;
@@ -110,14 +111,21 @@
final IRecordDescriptorProvider recordDescProvider, final int
partition, int nPartitions) {
return new AbstractUnaryInputSinkOperatorNodePushable() {
private IRunGenerator runGen;
+ private IOperatorStats stats;
@Override
public void open() throws HyracksDataException {
runGen = getRunGenerator(ctx, recordDescProvider);
+ runGen.setStats(stats);
runGen.open();
}
@Override
+ public void setOperatorStats(IOperatorStats stats) {
+ this.stats = stats;
+ }
+
+ @Override
public void nextFrame(ByteBuffer buffer) throws
HyracksDataException {
runGen.nextFrame(buffer);
}
diff --git
a/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/IRunGenerator.java
b/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/IRunGenerator.java
index d43d02d..5d3ec67 100644
---
a/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/IRunGenerator.java
+++
b/hyracks-fullstack/hyracks/hyracks-dataflow-std/src/main/java/org/apache/hyracks/dataflow/std/sort/IRunGenerator.java
@@ -21,6 +21,7 @@
import java.util.List;
import org.apache.hyracks.api.comm.IFrameWriter;
+import org.apache.hyracks.api.job.profiling.IOperatorStats;
import org.apache.hyracks.dataflow.common.io.GeneratedRunFileReader;
/**
@@ -35,4 +36,10 @@
List<GeneratedRunFileReader> getRuns();
ISorter getSorter();
+
+ /**
+ * Sets the operator stats used to track the number of bytes spilled to
disk while generating runs.
+ */
+ default void setStats(IOperatorStats stats) {
+ }
}
--
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21367?usp=email
To unsubscribe, or for help writing mail filters, visit
https://asterix-gerrit.ics.uci.edu/settings?usp=email
Gerrit-MessageType: newchange
Gerrit-Project: asterixdb
Gerrit-Branch: master
Gerrit-Change-Id: Ia6e37080a581292744ddc90b0b8149d7c13c19a3
Gerrit-Change-Number: 21367
Gerrit-PatchSet: 1
Gerrit-Owner: Ian Maxon <[email protected]>
