>From Ian Maxon <[email protected]>:
Ian Maxon has uploaded this change for review. (
https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21373?usp=email )
Change subject: [NO ISSUE][RT] Move stat calculation to FrameTupleAppender
......................................................................
[NO ISSUE][RT] Move stat calculation to FrameTupleAppender
Currently ProfiledFrameWriter re-scans each frame to calculate
summary statistics for operators. However this isn't usually
needed- these could just be accumulated in FrameTupleAppender
and updated by it, which saves time.
Change-Id: If8539f3967b2d02a28f3a627f9e739c3caf7c1a9
---
M
hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/meta/AlgebricksMetaOperatorDescriptor.java
M
hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/dataflow/ITimedWriter.java
M
hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/dataflow/ProfiledFrameWriter.java
M
hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/AbstractFrameAppender.java
M
hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameTupleAppender.java
5 files changed, 56 insertions(+), 11 deletions(-)
git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb
refs/changes/73/21373/1
diff --git
a/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/meta/AlgebricksMetaOperatorDescriptor.java
b/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/meta/AlgebricksMetaOperatorDescriptor.java
index 75609c1..71153fe 100644
---
a/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/meta/AlgebricksMetaOperatorDescriptor.java
+++
b/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/meta/AlgebricksMetaOperatorDescriptor.java
@@ -277,6 +277,11 @@
}
@Override
+ public IOperatorStats getInputStats() {
+ return parentStats;
+ }
+
+ @Override
public long getTotalTime() {
return startOfPipeline instanceof ITimedWriter ?
((ITimedWriter) startOfPipeline).getTotalTime() : 0;
}
diff --git
a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/dataflow/ITimedWriter.java
b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/dataflow/ITimedWriter.java
index e4f990f..5454466 100644
---
a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/dataflow/ITimedWriter.java
+++
b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/dataflow/ITimedWriter.java
@@ -24,5 +24,8 @@
public interface ITimedWriter extends IFrameWriter {
void setInputStats(IOperatorStats stats);
+ IOperatorStats getInputStats();
+
long getTotalTime();
+
}
diff --git
a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/dataflow/ProfiledFrameWriter.java
b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/dataflow/ProfiledFrameWriter.java
index 447aa1d..9cf3164 100644
---
a/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/dataflow/ProfiledFrameWriter.java
+++
b/hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/dataflow/ProfiledFrameWriter.java
@@ -46,6 +46,7 @@
private int maxSz = -1;
protected final ICounter totalTime;
+ protected boolean callerCounted = false;
public ProfiledFrameWriter(IFrameWriter writer) {
this.writer = writer;
@@ -57,6 +58,12 @@
this.inputStats = stats;
}
+ @Override
+ public IOperatorStats getInputStats() {
+ callerCounted = true;
+ return inputStats;
+ }
+
public static void timeMethod(HyracksRunnable r, ICounter c) throws
HyracksDataException {
long nt = 0;
try {
@@ -106,7 +113,9 @@
@Override
public void nextFrame(ByteBuffer buffer) throws HyracksDataException {
- updateTupleStats(buffer);
+ if (!callerCounted) {
+ updateTupleStats(buffer);
+ }
timeMethod(writer::nextFrame, buffer);
}
diff --git
a/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/AbstractFrameAppender.java
b/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/AbstractFrameAppender.java
index 7338cc7..87cbfd4 100644
---
a/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/AbstractFrameAppender.java
+++
b/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/AbstractFrameAppender.java
@@ -26,7 +26,9 @@
import org.apache.hyracks.api.comm.IFrame;
import org.apache.hyracks.api.comm.IFrameAppender;
import org.apache.hyracks.api.comm.IFrameWriter;
+import org.apache.hyracks.api.dataflow.ITimedWriter;
import org.apache.hyracks.api.exceptions.HyracksDataException;
+import org.apache.hyracks.api.job.profiling.IOperatorStats;
import org.apache.hyracks.util.IntSerDeUtils;
import org.apache.hyracks.util.trace.ITracer;
@@ -46,7 +48,12 @@
protected IFrame frame;
protected byte[] array; // cached the getBuffer().array to speed up byte
array access a little
- protected int tupleCount;
+ protected int tupleCount = 0;
+ protected long totalTupleCount = 0;
+ protected long maxTupleSize = -1;
+ protected long minTupleSize = Long.MAX_VALUE;
+ protected long bytesProcessed = 0;
+ protected long frameCount = 0;
protected int tupleDataEndOffset;
protected boolean ignoreFailures = false;
@@ -59,6 +66,18 @@
reset(getBuffer(), clear);
}
+ protected void updateTupleStats(int length) {
+ ++tupleCount;
+ ++totalTupleCount;
+ bytesProcessed += length;
+ if (length > maxTupleSize) {
+ maxTupleSize = length;
+ }
+ if (length < minTupleSize) {
+ minTupleSize = length;
+ }
+ }
+
protected boolean hasEnoughSpace(int fieldCount, int tupleLength) {
return tupleDataEndOffset + FrameHelper.calcRequiredSpace(fieldCount,
tupleLength)
+ tupleCount * FrameConstants.SIZE_LEN <=
FrameHelper.getTupleCountOffset(frame.getFrameSize());
@@ -92,6 +111,15 @@
public void write(IFrameWriter outWriter, boolean clearFrame) throws
HyracksDataException {
failIfInterrupted();
getBuffer().clear();
+ ++frameCount;
+ if (outWriter instanceof ITimedWriter) {
+ IOperatorStats stats = ((ITimedWriter) outWriter).getInputStats();
+ stats.getTupleCounter().set(totalTupleCount);
+ stats.getTupleBytes().set(bytesProcessed);
+ stats.getMaxTupleSz().set(maxTupleSize);
+ stats.getMinTupleSz().set(minTupleSize);
+ stats.getTotalFrameCount().set(frameCount);
+ }
if (!ignoreFailures) {
outWriter.nextFrame(getBuffer());
if (clearFrame) {
diff --git
a/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameTupleAppender.java
b/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameTupleAppender.java
index 58194f4..4277aa6 100644
---
a/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameTupleAppender.java
+++
b/hyracks-fullstack/hyracks/hyracks-dataflow-common/src/main/java/org/apache/hyracks/dataflow/common/comm/io/FrameTupleAppender.java
@@ -54,7 +54,7 @@
tupleDataEndOffset += fieldEndOffsets.length * 4 + length;
IntSerDeUtils.putInt(getBuffer().array(),
FrameHelper.getTupleCountOffset(frame.getFrameSize()) - 4
* (tupleCount + 1), tupleDataEndOffset);
- ++tupleCount;
+ updateTupleStats(fieldEndOffsets.length * 4 + length);
IntSerDeUtils.putInt(getBuffer().array(),
FrameHelper.getTupleCountOffset(frame.getFrameSize()),
tupleCount);
return true;
@@ -83,7 +83,7 @@
tupleDataEndOffset += tuple.getFieldCount() * 4 + length;
IntSerDeUtils.putInt(getBuffer().array(),
FrameHelper.getTupleCountOffset(frame.getFrameSize()) - 4
* (tupleCount + 1), tupleDataEndOffset);
- ++tupleCount;
+ updateTupleStats(tuple.getFieldCount() * 4 + length);
IntSerDeUtils.putInt(getBuffer().array(),
FrameHelper.getTupleCountOffset(frame.getFrameSize()),
tupleCount);
return true;
@@ -98,7 +98,7 @@
tupleDataEndOffset += length;
IntSerDeUtils.putInt(array,
FrameHelper.getTupleCountOffset(frame.getFrameSize()) - 4 * (tupleCount + 1),
tupleDataEndOffset);
- ++tupleCount;
+ updateTupleStats(length);
IntSerDeUtils.putInt(array,
FrameHelper.getTupleCountOffset(frame.getFrameSize()), tupleCount);
return true;
}
@@ -120,7 +120,7 @@
tupleDataEndOffset += effectiveSlots * 4 + length;
IntSerDeUtils.putInt(array,
FrameHelper.getTupleCountOffset(frame.getFrameSize()) - 4 * (tupleCount + 1),
tupleDataEndOffset);
- ++tupleCount;
+ updateTupleStats(effectiveSlots * 4 + length);
IntSerDeUtils.putInt(array,
FrameHelper.getTupleCountOffset(frame.getFrameSize()), tupleCount);
return true;
}
@@ -137,7 +137,7 @@
tupleDataEndOffset += length;
IntSerDeUtils.putInt(array,
FrameHelper.getTupleCountOffset(frame.getFrameSize()) - 4 * (tupleCount + 1),
tupleDataEndOffset);
- ++tupleCount;
+ updateTupleStats(length);
IntSerDeUtils.putInt(array,
FrameHelper.getTupleCountOffset(frame.getFrameSize()), tupleCount);
return true;
}
@@ -185,7 +185,7 @@
tupleDataEndOffset += (length0 + length1);
IntSerDeUtils.putInt(array,
FrameHelper.getTupleCountOffset(frame.getFrameSize()) - 4 * (tupleCount + 1),
tupleDataEndOffset);
- ++tupleCount;
+ updateTupleStats(length0 + length1);
IntSerDeUtils.putInt(array,
FrameHelper.getTupleCountOffset(frame.getFrameSize()), tupleCount);
return true;
}
@@ -221,7 +221,7 @@
tupleDataEndOffset += (length0 + length1);
IntSerDeUtils.putInt(array,
FrameHelper.getTupleCountOffset(frame.getFrameSize()) - 4 * (tupleCount + 1),
tupleDataEndOffset);
- ++tupleCount;
+ updateTupleStats(length0 + length1);
IntSerDeUtils.putInt(array,
FrameHelper.getTupleCountOffset(frame.getFrameSize()), tupleCount);
return true;
}
@@ -258,7 +258,7 @@
tupleDataEndOffset += (length0 + length1);
IntSerDeUtils.putInt(array,
FrameHelper.getTupleCountOffset(frame.getFrameSize()) - 4 * (tupleCount + 1),
tupleDataEndOffset);
- ++tupleCount;
+ updateTupleStats(length0 + length1);
IntSerDeUtils.putInt(array,
FrameHelper.getTupleCountOffset(frame.getFrameSize()), tupleCount);
return true;
}
@@ -293,7 +293,7 @@
tupleDataEndOffset += length;
IntSerDeUtils.putInt(array,
FrameHelper.getTupleCountOffset(frame.getFrameSize()) - 4 * (tupleCount + 1),
tupleDataEndOffset);
- ++tupleCount;
+ updateTupleStats(length);
IntSerDeUtils.putInt(array,
FrameHelper.getTupleCountOffset(frame.getFrameSize()), tupleCount);
return true;
}
--
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/21373?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: If8539f3967b2d02a28f3a627f9e739c3caf7c1a9
Gerrit-Change-Number: 21373
Gerrit-PatchSet: 1
Gerrit-Owner: Ian Maxon <[email protected]>
