You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@asterixdb.apache.org by AsterixDB Code Review <do...@asterix-gerrit.ics.uci.edu> on 2025/07/10 23:23:34 UTC
Change in asterixdb[ionic]: [ASTERIXDB-3628][RT] Correct subplan profiles
From Ian Maxon <im...@apache.org>:
Ian Maxon has uploaded this change for review. ( https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/20069 )
Change subject: [ASTERIXDB-3628][RT] Correct subplan profiles
......................................................................
[ASTERIXDB-3628][RT] Correct subplan profiles
- user model changes: no
- storage format changes: no
- interface changes: no
Details:
Currently subplan timing information is calculated during close().
This can lead to inaccuracies in the timing, especially when the
subplan is near the end of a task pipeline, because other
operator's times are calculated after close, during deinitialize().
Correct this by adding the capability in the MetaOperator to invoke a
method on each ProfiledPushRuntime SubplanPushRuntime to calculate the
timings, so it can be invoked at the proper time as to retrieve the
correct time from its downstream operators.
Ext-ref: MB-67499
Change-Id: I4d38710626778ef52abb1cd4fe1e0b26e0d74bec
---
A hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/base/IProfiledPushRuntime.java
M hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/meta/SubplanRuntimeFactory.java
M hyracks-fullstack/hyracks/hyracks-api/src/main/java/org/apache/hyracks/api/dataflow/ProfiledFrameWriter.java
M hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/base/ProfiledPushRuntime.java
M hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/meta/PipelineAssembler.java
M hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/meta/AlgebricksMetaOperatorDescriptor.java
6 files changed, 101 insertions(+), 13 deletions(-)
git pull ssh://asterix-gerrit.ics.uci.edu:29418/asterixdb refs/changes/69/20069/1
diff --git a/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/base/IProfiledPushRuntime.java b/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/base/IProfiledPushRuntime.java
new file mode 100644
index 0000000..9875a2c
--- /dev/null
+++ b/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/base/IProfiledPushRuntime.java
@@ -0,0 +1,25 @@
+/*
+ * 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.hyracks.algebricks.runtime.base;
+
+public interface IProfiledPushRuntime extends IPushRuntime {
+
+ public void computeTimings();
+
+}
diff --git a/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/base/ProfiledPushRuntime.java b/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/base/ProfiledPushRuntime.java
index bf4533f..ff9a58b 100644
--- a/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/base/ProfiledPushRuntime.java
+++ b/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/base/ProfiledPushRuntime.java
@@ -28,7 +28,7 @@
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.job.profiling.IOperatorStats;
-public class ProfiledPushRuntime extends ProfiledFrameWriter implements IPushRuntime {
+public class ProfiledPushRuntime extends ProfiledFrameWriter implements IProfiledPushRuntime {
private final IPushRuntime wrapped;
private final IOperatorStats stats;
@@ -45,9 +45,16 @@
this.last = last;
}
+ public IOperatorStats getStats() {
+ return stats;
+ }
+
@Override
- public void close() throws HyracksDataException {
- super.close();
+ public void computeTimings() {
+ //mainly to push through to subplans
+ if (wrapped instanceof IProfiledPushRuntime) {
+ ((IProfiledPushRuntime) wrapped).computeTimings();
+ }
long ownTime = getTotalTime();
//for micro union all. accumulate the time of each input into the counter.
//then, on input 0, subtract the output from the accumulated time.
@@ -62,10 +69,6 @@
stats.getTimeCounter().set(ownTime);
}
- public IOperatorStats getStats() {
- return stats;
- }
-
@Override
public void setOutputFrameWriter(int index, IFrameWriter writer, RecordDescriptor recordDesc) {
if (writer instanceof ITimedWriter) {
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 560f817..149aa24 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
@@ -26,6 +26,7 @@
import java.util.Map;
import org.apache.hyracks.algebricks.runtime.base.AlgebricksPipeline;
+import org.apache.hyracks.algebricks.runtime.base.IProfiledPushRuntime;
import org.apache.hyracks.algebricks.runtime.base.IPushRuntimeFactory;
import org.apache.hyracks.api.comm.IFrameWriter;
import org.apache.hyracks.api.context.IHyracksTaskContext;
@@ -35,6 +36,7 @@
import org.apache.hyracks.api.dataflow.value.RecordDescriptor;
import org.apache.hyracks.api.exceptions.HyracksDataException;
import org.apache.hyracks.api.job.IOperatorDescriptorRegistry;
+import org.apache.hyracks.api.job.JobFlag;
import org.apache.hyracks.api.job.profiling.IOperatorStats;
import org.apache.hyracks.api.job.profiling.NoOpOperatorStats;
import org.apache.hyracks.api.job.profiling.OperatorStats;
@@ -204,6 +206,7 @@
private boolean opened = false;
private IOperatorStats parentStats = NoOpOperatorStats.INSTANCE;
private Map<IPushRuntimeFactory, IOperatorStats> microOpStats = new HashMap<>();
+ private PipelineAssembler assembler;
public void open() throws HyracksDataException {
if (startOfPipeline == null) {
@@ -211,9 +214,9 @@
outputArity > 0 ? AlgebricksMetaOperatorDescriptor.this.outRecDescs[0] : null;
RecordDescriptor pipelineInputRecordDescriptor = recordDescProvider
.getInputRecordDescriptor(AlgebricksMetaOperatorDescriptor.this.getActivityId(), 0);
- PipelineAssembler pa = new PipelineAssembler(pipeline, inputArity, outputArity,
- pipelineInputRecordDescriptor, pipelineOutputRecordDescriptor);
- startOfPipeline = pa.assemblePipeline(writer, ctx, microOpStats);
+ assembler = new PipelineAssembler(pipeline, inputArity, outputArity, pipelineInputRecordDescriptor,
+ pipelineOutputRecordDescriptor);
+ startOfPipeline = assembler.assemblePipeline(writer, ctx, microOpStats);
}
opened = true;
startOfPipeline.open();
@@ -245,7 +248,9 @@
@Override
public void deinitialize() throws HyracksDataException {
- super.deinitialize();
+ if (ctx.getJobFlags().contains(JobFlag.PROFILE_RUNTIME)) {
+ assembler.getProfiledPushRuntimes().forEach(IProfiledPushRuntime::computeTimings);
+ }
}
@Override
diff --git a/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/meta/PipelineAssembler.java b/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/meta/PipelineAssembler.java
index c2e8473..0f9355b 100644
--- a/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/meta/PipelineAssembler.java
+++ b/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/meta/PipelineAssembler.java
@@ -18,11 +18,15 @@
*/
package org.apache.hyracks.algebricks.runtime.operators.meta;
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import org.apache.hyracks.algebricks.runtime.base.AlgebricksPipeline;
import org.apache.hyracks.algebricks.runtime.base.EnforcePushRuntime;
+import org.apache.hyracks.algebricks.runtime.base.IProfiledPushRuntime;
import org.apache.hyracks.algebricks.runtime.base.IPushRuntime;
import org.apache.hyracks.algebricks.runtime.base.IPushRuntimeFactory;
import org.apache.hyracks.algebricks.runtime.base.ProfiledPushRuntime;
@@ -47,6 +51,7 @@
private final int outputArity;
private final AlgebricksPipeline pipeline;
private final Map<IPushRuntimeFactory, IPushRuntime[]> runtimeMap;
+ private final List<IProfiledPushRuntime> profiledRuntimes;
private final boolean ignoreFailures;
public PipelineAssembler(AlgebricksPipeline pipeline, int inputArity, int outputArity,
@@ -63,6 +68,7 @@
this.inputArity = inputArity;
this.outputArity = outputArity;
this.runtimeMap = new HashMap<>();
+ this.profiledRuntimes = new ArrayList<>();
this.ignoreFailures = ignoreFailures;
}
@@ -97,6 +103,7 @@
profiled = (ProfiledPushRuntime) ProfiledPushRuntime.time(newRuntimes[j],
microOpStats.get(runtimeFactory), false);
}
+ profiledRuntimes.add(profiled);
newRuntimes[j] = profiled;
} else if (enforce && !profile) {
newRuntimes[j] = EnforcePushRuntime.enforce(newRuntimes[j]);
@@ -129,6 +136,10 @@
return runtimeMap.get(runtimeFactory);
}
+ public List<IProfiledPushRuntime> getProfiledPushRuntimes() {
+ return Collections.unmodifiableList(profiledRuntimes);
+ }
+
//TODO: refactoring is needed
public static IFrameWriter assemblePipeline(AlgebricksPipeline subplan, IFrameWriter writer,
IHyracksTaskContext ctx, Map<IPushRuntimeFactory, IPushRuntime> outRuntimeMap) throws HyracksDataException {
diff --git a/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/meta/SubplanRuntimeFactory.java b/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/meta/SubplanRuntimeFactory.java
index 242c603..766bfac 100644
--- a/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/meta/SubplanRuntimeFactory.java
+++ b/hyracks-fullstack/algebricks/algebricks-runtime/src/main/java/org/apache/hyracks/algebricks/runtime/operators/meta/SubplanRuntimeFactory.java
@@ -20,12 +20,14 @@
import java.io.DataOutput;
import java.nio.ByteBuffer;
+import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.hyracks.algebricks.common.exceptions.NotImplementedException;
import org.apache.hyracks.algebricks.runtime.base.AlgebricksPipeline;
+import org.apache.hyracks.algebricks.runtime.base.IProfiledPushRuntime;
import org.apache.hyracks.algebricks.runtime.base.IPushRuntime;
import org.apache.hyracks.algebricks.runtime.base.IPushRuntimeFactory;
import org.apache.hyracks.algebricks.runtime.operators.base.AbstractOneInputOneOutputOneFramePushRuntime;
@@ -116,7 +118,8 @@
return new SubplanPushRuntime(ctx, false);
}
- public class SubplanPushRuntime extends AbstractOneInputOneOutputOneFramePushRuntime {
+ public class SubplanPushRuntime extends AbstractOneInputOneOutputOneFramePushRuntime
+ implements IProfiledPushRuntime {
protected final IHyracksTaskContext ctx;
@@ -126,9 +129,16 @@
boolean profile;
+ List<IProfiledPushRuntime> timedMicroOps;
+
+ PipelineAssembler[] pipelineAssemblers;
+
protected SubplanPushRuntime(IHyracksTaskContext ctx, boolean ignoreFailures) throws HyracksDataException {
this.ctx = ctx;
this.profile = ctx.getJobFlags().contains(JobFlag.PROFILE_RUNTIME);
+ if (profile) {
+ timedMicroOps = new ArrayList<>();
+ }
this.first = true;
IMissingWriter[] missingWriters = new IMissingWriter[missingWriterFactories.length];
@@ -138,7 +148,7 @@
int pipelineCount = pipelines.size();
startOfPipelines = new NestedTupleSourceRuntime[pipelineCount];
- PipelineAssembler[] pipelineAssemblers = new PipelineAssembler[pipelineCount];
+ pipelineAssemblers = new PipelineAssembler[pipelineCount];
for (int i = 0; i < pipelineCount; i++) {
AlgebricksPipeline pipeline = pipelines.get(i);
RecordDescriptor pipelineLastRecordDescriptor =
@@ -174,6 +184,12 @@
}
}
+ public void computeTimings() {
+ for (PipelineAssembler assembler : pipelineAssemblers) {
+ assembler.getProfiledPushRuntimes().forEach(IProfiledPushRuntime::computeTimings);
+ }
+ }
+
@Override
public void open() throws HyracksDataException {
// writer opened many times?
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 53de340..881fdba 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
@@ -123,6 +123,7 @@
@Override
public void close() throws HyracksDataException {
timeMethod(writer::close, totalTime);
+
}
private int getTupleStartOffset(int tupleIndex, int tupleCountOffset, ByteBuffer buffer) {
--
To view, visit https://asterix-gerrit.ics.uci.edu/c/asterixdb/+/20069
To unsubscribe, or for help writing mail filters, visit https://asterix-gerrit.ics.uci.edu/settings
Gerrit-Project: asterixdb
Gerrit-Branch: ionic
Gerrit-Change-Id: I4d38710626778ef52abb1cd4fe1e0b26e0d74bec
Gerrit-Change-Number: 20069
Gerrit-PatchSet: 1
Gerrit-Owner: Ian Maxon <im...@apache.org>
Gerrit-MessageType: newchange