This is an automated email from the ASF dual-hosted git repository.
mattcasters pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/hop.git
The following commit(s) were added to refs/heads/main by this push:
new 341c28a87b Issue #8863 : Write the parent execution link when a child
starts (#8865)
341c28a87b is described below
commit 341c28a87b7c2e33b12236ed1899260d8c1ec886
Author: Matt Casters <[email protected]>
AuthorDate: Wed Oct 14 16:11:50 2026 +0200
Issue #8863 : Write the parent execution link when a child starts (#8865)
* Issue #8863 : Write the parent execution link when a child starts
Store the owning workflow or pipeline as parentId and the action or
transform
that started the child as launchedById. Drop the file and database scans
that
rediscovered that link during refresh. List the execution tree from the
summary
columns so a large location does not load every document.
* Issue #8863 : Find launched children on file and graph locations
File and graph locations match the action or transform that started a
child, and the graph links that launcher when its node is already stored. A
child started before the launcher is flushed still records the link. Database
summary rows load the update time of a run that has not ended, and copying a
filename loads the stored execution. Execution information appends launchedById.
---
.../modules/ROOT/pages/hop-server/json-api.adoc | 2 +-
.../modules/ROOT/pages/hop-server/rest-api.adoc | 2 +-
.../ROOT/pages/pipeline/transforms/execinfo.adoc | 8 +-
.../java/org/apache/hop/execution/Execution.java | 53 ++++-
.../org/apache/hop/execution/ExecutionBuilder.java | 47 +++-
.../org/apache/hop/execution/ExecutionState.java | 5 +-
.../hop/execution/ExecutionStateBuilder.java | 10 +-
.../hop/execution/IExecutionInfoLocation.java | 28 +++
.../org/apache/hop/execution/ListedExecution.java | 23 ++
.../caching/BaseCachingExecutionInfoLocation.java | 106 ++++++---
.../apache/hop/execution/caching/CacheEntry.java | 4 +
.../caching/CachingFileExecutionInfoLocation.java | 81 ++-----
.../execution/local/FileExecutionInfoLocation.java | 8 +-
.../execution/ExecutionBuilderParentLinkTest.java | 104 +++++++++
.../BaseCachingExecutionInfoLocationTest.java | 6 +-
.../CachingFileExecutionInfoLocationTest.java | 106 ++++++++-
.../local/FileExecutionInfoLocationTest.java | 18 ++
integration-tests/exec-info/002-validation.hpl | 14 +-
.../workflow/actions/pipeline/ActionPipeline.java | 2 +
.../workflow/actions/workflow/ActionWorkflow.java | 4 +-
.../CachingDatabaseExecutionInfoLocation.java | 239 ++++++++++++++-------
.../CachingDatabaseExecutionInfoLocationTest.java | 89 +++++++-
.../neo4j/execution/NeoExecutionInfoLocation.java | 69 ++++--
.../NeoExecutionInfoLocationDeleteTest.java | 12 ++
.../hop/pipeline/transforms/execinfo/ExecInfo.java | 6 +-
.../pipeline/transforms/execinfo/ExecInfoMeta.java | 1 +
.../execution/ExecutionPerspective.java | 126 +++++++----
.../execution/PipelineExecutionViewer.java | 23 +-
.../execution/WorkflowExecutionViewer.java | 21 +-
.../hop/ui/hopgui/shared/BaseExecutionViewer.java | 50 +++++
30 files changed, 980 insertions(+), 287 deletions(-)
diff --git a/docs/hop-user-manual/modules/ROOT/pages/hop-server/json-api.adoc
b/docs/hop-user-manual/modules/ROOT/pages/hop-server/json-api.adoc
index b4bbd94782..60b0935329 100644
--- a/docs/hop-user-manual/modules/ROOT/pages/hop-server/json-api.adoc
+++ b/docs/hop-user-manual/modules/ROOT/pages/hop-server/json-api.adoc
@@ -229,7 +229,7 @@ They are the JSON equivalent of the `hop/getExecInfo` and
`hop/registerExecInfo`
|`GET`
|`+location/{location}/executions/{id}/parent+`
-|Find the parent execution ID.
+|Find the parent execution ID. This is the workflow or pipeline that owns the
execution. Before Hop 2.20 it was the action or transform that started it.
|`GET`
|`+location/{location}/executions/{id}/data?parentId=+`
diff --git a/docs/hop-user-manual/modules/ROOT/pages/hop-server/rest-api.adoc
b/docs/hop-user-manual/modules/ROOT/pages/hop-server/rest-api.adoc
index d2d623f5e2..32b56d08a4 100644
--- a/docs/hop-user-manual/modules/ROOT/pages/hop-server/rest-api.adoc
+++ b/docs/hop-user-manual/modules/ROOT/pages/hop-server/rest-api.adoc
@@ -1077,7 +1077,7 @@ The `type` parameter accepts the following values:
|`DATA`|Get the execution data of an execution|parentId
|`LAST_EXECUTION`|Get the last execution with a given name and type|name,
execType
|`CHILD_IDS`|Get the child ids of an execution|id, execType
-|`PARENT_ID`|Get the parent id of a child execution|id
+|`PARENT_ID`|Get the id of the workflow or pipeline that owns the execution.
Before Hop 2.20 this was the action or transform that started it.|id
|`DELETE`|Delete an execution|id
|===
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/execinfo.adoc
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/execinfo.adoc
index f4b12cd4f3..366536f0cf 100644
--- a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/execinfo.adoc
+++ b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/execinfo.adoc
@@ -86,7 +86,7 @@ The transform always needs input rows. Which input fields are
required, and whic
|Find executions
|Execution parent ID
-|One output row per child execution, with execution and state columns
+|One output row per child execution, with execution and state columns. The id
may be the owning workflow or pipeline, or the action or transform that started
the child.
|Find last execution
|Execution name, Execution type
@@ -102,7 +102,7 @@ The transform always needs input rows. Which input fields
are required, and whic
|Find parent ID
|Execution ID
-|`id`
+|`id` of the workflow or pipeline that owns the execution. Before Hop 2.20
this was the action or transform that started it. That id is now `launchedById`
on the execution.
|Get execution data
|Execution ID, Execution parent ID
@@ -117,7 +117,9 @@ Execution *type* is the `ExecutionType` name: `Pipeline`,
`Workflow`, `Transform
Find last execution / Find previous successful execution / Get execution and
state add:
-`executionId`, `parentId`, `name`, `executionType`, `filename`, `executorXml`,
`metadataJson`, `registrationDate`, `executionStartDate`,
`runConfigurationName`, `logLevel`, `updateTime`, `loggingText`, `failed`,
`statusDescription`, `executionEndDate`.
+`executionId`, `parentId`, `name`, `executionType`, `filename`, `executorXml`,
`metadataJson`, `registrationDate`, `executionStartDate`,
`runConfigurationName`, `logLevel`, `updateTime`, `loggingText`, `failed`,
`statusDescription`, `executionEndDate`, `launchedById`.
+
+`launchedById` is the action or transform that started the execution. It is
empty when nothing did, and it is the last column so a pipeline that reads
these fields by position keeps the columns it already uses.
If no execution matches, the transform writes no output row for that input.
diff --git a/engine/src/main/java/org/apache/hop/execution/Execution.java
b/engine/src/main/java/org/apache/hop/execution/Execution.java
index 9fc41b59ef..8960a62167 100644
--- a/engine/src/main/java/org/apache/hop/execution/Execution.java
+++ b/engine/src/main/java/org/apache/hop/execution/Execution.java
@@ -20,11 +20,14 @@ package org.apache.hop.execution;
import com.fasterxml.jackson.annotation.JsonIgnore;
import com.fasterxml.jackson.annotation.JsonInclude;
+import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
import lombok.Getter;
import lombok.Setter;
+import org.apache.commons.lang3.StringUtils;
import org.apache.hop.core.logging.LogLevel;
/**
@@ -70,9 +73,27 @@ public class Execution {
/** The unique ID of a pipeline or workflow that is about to get executed */
private String id;
- /** The unique ID of the parent executor (pipeline or workflow) */
+ /**
+ * The pipeline or workflow execution that owns this one. When an action or
transform starts it,
+ * this is that workflow or pipeline, not the action or transform log
channel id.
+ */
private String parentId;
+ /**
+ * Log channel id of the action or transform that started this pipeline or
workflow. Empty when
+ * this execution was not started by one. Omitted when empty so a reader
that rejects unknown
+ * properties can still open a document that has no launcher.
+ */
+ @JsonInclude(JsonInclude.Include.NON_EMPTY)
+ private String launchedById;
+
+ /**
+ * Pipeline and workflow execution ids started by this action or transform.
Written when the child
+ * is registered so a refresh does not have to search for them.
+ */
+ @JsonInclude(JsonInclude.Include.NON_EMPTY)
+ private List<String> childExecutionIds;
+
/** The type of execution to register: Pipeline or Workflow only */
private ExecutionType executionType;
@@ -114,4 +135,34 @@ public class Execution {
this.parameterValues = new HashMap<>();
this.environmentDetails = new HashMap<>();
}
+
+ /**
+ * @return true when {@code childId} was not already recorded
+ */
+ public boolean addChildExecutionId(String childId) {
+ if (StringUtils.isEmpty(childId)) {
+ return false;
+ }
+ if (childExecutionIds == null) {
+ childExecutionIds = new ArrayList<>();
+ }
+ if (childExecutionIds.contains(childId)) {
+ return false;
+ }
+ childExecutionIds.add(childId);
+ return true;
+ }
+
+ /**
+ * Keep every child id from {@code other}. Two location instances can
persist the same action, and
+ * the later write must not drop ids the other one recorded.
+ */
+ public void unionChildExecutionIds(Execution other) {
+ if (other == null || other.childExecutionIds == null) {
+ return;
+ }
+ for (String childId : other.childExecutionIds) {
+ addChildExecutionId(childId);
+ }
+ }
}
diff --git
a/engine/src/main/java/org/apache/hop/execution/ExecutionBuilder.java
b/engine/src/main/java/org/apache/hop/execution/ExecutionBuilder.java
index 0b00ea8422..ca41be26a0 100644
--- a/engine/src/main/java/org/apache/hop/execution/ExecutionBuilder.java
+++ b/engine/src/main/java/org/apache/hop/execution/ExecutionBuilder.java
@@ -25,6 +25,7 @@ import java.util.HashMap;
import java.util.Map;
import org.apache.commons.lang3.StringUtils;
import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.logging.ILoggingObject;
import org.apache.hop.core.logging.LogLevel;
import org.apache.hop.core.metadata.SerializableMetadataProvider;
import org.apache.hop.core.parameters.INamedParameters;
@@ -44,6 +45,7 @@ public final class ExecutionBuilder {
public String filename;
public String id;
public String parentId;
+ public String launchedById;
public ExecutionType executionType;
public String executorXml;
public String metadataJson;
@@ -68,6 +70,38 @@ public final class ExecutionBuilder {
return new ExecutionBuilder();
}
+ /**
+ * The pipeline or workflow that owns an execution. An action's logging
parent is the action, and
+ * a transform's logging parent is the transform. The stored parent is the
workflow or pipeline
+ * that launcher belongs to.
+ */
+ public static String executionParentId(ILoggingObject loggingParent) {
+ if (loggingParent == null) {
+ return null;
+ }
+ if (loggingParent instanceof IAction action && action.getParentWorkflow()
!= null) {
+ return action.getParentWorkflow().getLogChannelId();
+ }
+ if (loggingParent instanceof ITransform transform &&
transform.getPipeline() != null) {
+ return transform.getPipeline().getLogChannelId();
+ }
+ return loggingParent.getLogChannelId();
+ }
+
+ /**
+ * Log channel id of the action or transform that started an execution. Null
when the logging
+ * parent is already the pipeline or workflow, or when there is no parent.
+ */
+ public static String launchedById(ILoggingObject loggingParent) {
+ if (loggingParent instanceof IAction action && action.getParentWorkflow()
!= null) {
+ return loggingParent.getLogChannelId();
+ }
+ if (loggingParent instanceof ITransform transform &&
transform.getPipeline() != null) {
+ return loggingParent.getLogChannelId();
+ }
+ return null;
+ }
+
public static ExecutionBuilder fromExecutor(IPipelineEngine<PipelineMeta>
pipeline)
throws HopException {
ExecutionBuilder builder = of();
@@ -75,7 +109,8 @@ public final class ExecutionBuilder {
.withFilename(pipeline.getPipelineMeta().getFilename())
.withName(pipeline.getPipelineMeta().getName())
.withId(pipeline.getLogChannelId())
- .withParentId(pipeline.getParent() == null ? null :
pipeline.getParent().getLogChannelId())
+ .withParentId(executionParentId(pipeline.getParent()))
+ .withLaunchedById(launchedById(pipeline.getParent()))
.withExecutorType(ExecutionType.Pipeline)
.withExecutorXml(pipeline.getPipelineMeta().getXml(pipeline))
.withMetadataJson(new
SerializableMetadataProvider(pipeline.getMetadataProvider()).toJson())
@@ -99,8 +134,8 @@ public final class ExecutionBuilder {
of().withFilename(workflow.getWorkflowMeta().getFilename())
.withName(workflow.getWorkflowMeta().getName())
.withId(workflow.getLogChannelId())
- .withParentId(
- workflow.getParent() == null ? null :
workflow.getParent().getLogChannelId())
+ .withParentId(executionParentId(workflow.getParent()))
+ .withLaunchedById(launchedById(workflow.getParent()))
.withExecutorType(ExecutionType.Workflow)
.withExecutorXml(workflow.getWorkflowMeta().getXml(workflow))
.withMetadataJson(
@@ -222,6 +257,11 @@ public final class ExecutionBuilder {
return this;
}
+ public ExecutionBuilder withLaunchedById(String launchedById) {
+ this.launchedById = launchedById;
+ return this;
+ }
+
public ExecutionBuilder withExecutorType(ExecutionType executionType) {
this.executionType = executionType;
return this;
@@ -293,6 +333,7 @@ public final class ExecutionBuilder {
executionRegistration.setFilename(filename);
executionRegistration.setId(id);
executionRegistration.setParentId(parentId);
+ executionRegistration.setLaunchedById(launchedById);
executionRegistration.setExecutionType(executionType);
executionRegistration.setExecutorXml(executorXml);
executionRegistration.setMetadataJson(metadataJson);
diff --git a/engine/src/main/java/org/apache/hop/execution/ExecutionState.java
b/engine/src/main/java/org/apache/hop/execution/ExecutionState.java
index 98362a9c3f..9b78307682 100644
--- a/engine/src/main/java/org/apache/hop/execution/ExecutionState.java
+++ b/engine/src/main/java/org/apache/hop/execution/ExecutionState.java
@@ -38,8 +38,9 @@ public class ExecutionState {
private ExecutionType executionType;
/**
- * The unique ID of the parent executor. If this is an action or transform
update, this is the ID
- * of that executor.
+ * The pipeline or workflow that owns this update. For an action or
transform this is that
+ * executor. For a pipeline or workflow started by an action or transform,
this is the same owner
+ * stored on {@link Execution#getParentId()}.
*/
private String parentId;
diff --git
a/engine/src/main/java/org/apache/hop/execution/ExecutionStateBuilder.java
b/engine/src/main/java/org/apache/hop/execution/ExecutionStateBuilder.java
index 07e5552b17..1c578897f7 100644
--- a/engine/src/main/java/org/apache/hop/execution/ExecutionStateBuilder.java
+++ b/engine/src/main/java/org/apache/hop/execution/ExecutionStateBuilder.java
@@ -97,9 +97,6 @@ public final class ExecutionStateBuilder {
public static ExecutionStateBuilder fromExecutor(
IPipelineEngine<PipelineMeta> pipeline, Integer lastLogLineNr) {
- String parentLogChannelId =
- pipeline.getParent() == null ? null :
pipeline.getParent().getLogChannelId();
-
// The last log line nr for this pipeline?
// Look it up before querying the store to avoid missing lines
//
@@ -107,7 +104,7 @@ public final class ExecutionStateBuilder {
ExecutionStateBuilder builder =
of().withExecutionType(ExecutionType.Pipeline)
- .withParentId(parentLogChannelId)
+
.withParentId(ExecutionBuilder.executionParentId(pipeline.getParent()))
.withId(pipeline.getLogChannelId())
.withName(pipeline.getPipelineMeta().getName())
.withLoggingText(getLoggingText(pipeline.getLogChannelId(),
lastLogLineNr))
@@ -261,9 +258,6 @@ public final class ExecutionStateBuilder {
public static ExecutionStateBuilder fromExecutor(
IWorkflowEngine<WorkflowMeta> workflow, Integer lastLogLineNr) {
- String parentLogChannelId =
- workflow.getParent() == null ? null :
workflow.getParent().getLogChannelId();
-
// The last log line nr for this workflow?
// Look it up before querying the store to avoid missing lines
//
@@ -271,7 +265,7 @@ public final class ExecutionStateBuilder {
Result result = workflow.getResult();
return of().withExecutionType(ExecutionType.Workflow)
- .withParentId(parentLogChannelId)
+ .withParentId(ExecutionBuilder.executionParentId(workflow.getParent()))
.withId(workflow.getLogChannelId())
.withName(workflow.getWorkflowMeta().getName())
.withLoggingText(getLoggingText(workflow.getLogChannelId(),
lastLogLineNr))
diff --git
a/engine/src/main/java/org/apache/hop/execution/IExecutionInfoLocation.java
b/engine/src/main/java/org/apache/hop/execution/IExecutionInfoLocation.java
index 8f0bb42b5f..9bb2fefb71 100644
--- a/engine/src/main/java/org/apache/hop/execution/IExecutionInfoLocation.java
+++ b/engine/src/main/java/org/apache/hop/execution/IExecutionInfoLocation.java
@@ -18,6 +18,7 @@
package org.apache.hop.execution;
+import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import org.apache.hop.core.IProgressMonitor;
@@ -227,6 +228,28 @@ public interface IExecutionInfoLocation extends Cloneable {
*/
List<String> findExecutionIDs(IExecutionSelector selector) throws
HopException;
+ /**
+ * Rows for the execution tree. Name, type, dates, and status are enough.
The pipeline or workflow
+ * document stays unread until that execution is opened. The default loads
each id in full;
+ * locations that store a summary separately override this.
+ */
+ default List<ListedExecution> listExecutions(IExecutionSelector selector)
throws HopException {
+ IExecutionSelector active = selector == null ? IExecutionSelector.ALL :
selector;
+ List<ListedExecution> listed = new ArrayList<>();
+ for (String id : findExecutionIDs(active)) {
+ Execution execution = getExecution(id);
+ if (execution == null || !active.isSelected(execution)) {
+ continue;
+ }
+ ExecutionState state = getExecutionState(id);
+ if (!active.isSelected(state)) {
+ continue;
+ }
+ listed.add(new ListedExecution(execution, state));
+ }
+ return listed;
+ }
+
/**
* Get execution data for transforms or an action. The parent ID would
typically be a pipeline ID,
* and you'd get data for all the transforms. You can also get the execution
data for specific
@@ -262,6 +285,11 @@ public interface IExecutionInfoLocation extends Cloneable {
List<String> findChildIds(ExecutionType parentExecutionType, String
parentExecutionId)
throws HopException;
+ /**
+ * The workflow or pipeline that owns {@code childId}. An execution written
before Hop 2.20 stored
+ * the action or transform that started it here. That id is now {@code
launchedById} on {@link
+ * Execution}.
+ */
String findParentId(String childId) throws HopException;
/**
diff --git a/engine/src/main/java/org/apache/hop/execution/ListedExecution.java
b/engine/src/main/java/org/apache/hop/execution/ListedExecution.java
new file mode 100644
index 0000000000..6cfdb0520c
--- /dev/null
+++ b/engine/src/main/java/org/apache/hop/execution/ListedExecution.java
@@ -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.
+ */
+
+package org.apache.hop.execution;
+
+/**
+ * One row in the execution perspective tree: name, dates, and status, without
the stored document.
+ */
+public record ListedExecution(Execution execution, ExecutionState state) {}
diff --git
a/engine/src/main/java/org/apache/hop/execution/caching/BaseCachingExecutionInfoLocation.java
b/engine/src/main/java/org/apache/hop/execution/caching/BaseCachingExecutionInfoLocation.java
index 21425f8cce..6db70834f4 100644
---
a/engine/src/main/java/org/apache/hop/execution/caching/BaseCachingExecutionInfoLocation.java
+++
b/engine/src/main/java/org/apache/hop/execution/caching/BaseCachingExecutionInfoLocation.java
@@ -350,11 +350,82 @@ public abstract class BaseCachingExecutionInfoLocation
implements IExecutionInfo
if (entry != null) {
persistCacheEntry(entry);
}
+ rememberLaunchedChild(execution);
} else {
addChildExecutionToCache(execution);
}
}
+ /**
+ * Record this pipeline or workflow on the action or transform that started
it, inside the parent
+ * document. A launcher that is not stored yet is written as a minimal
entry; the parent's next
+ * save keeps that child id when it merges. The child document still has
{@code parentId} and
+ * {@code launchedById}.
+ */
+ private void rememberLaunchedChild(Execution child) {
+ if (child == null
+ || StringUtils.isEmpty(child.getId())
+ || StringUtils.isEmpty(child.getParentId())
+ || StringUtils.isEmpty(child.getLaunchedById())) {
+ return;
+ }
+ // This location is often a different instance from the one running the
parent. Do not keep the
+ // loaded parent: a later state update must not match it, and close() must
not write it back.
+ boolean loadedForLink = !cache.containsKey(child.getParentId());
+ try {
+ CacheEntry parent = findCacheEntry(child.getParentId());
+ if (parent == null || parent.getExecution() == null) {
+ LogChannel.GENERAL.logBasic(
+ "Child execution '"
+ + child.getId()
+ + "' was not linked: parent '"
+ + child.getParentId()
+ + "' is not stored yet");
+ return;
+ }
+ Execution launcher = parent.peekChildExecution(child.getLaunchedById());
+ if (launcher == null) {
+ launcher = minimalLauncher(child, parent.getExecution());
+ parent.addChildExecution(launcher);
+ }
+ if (!launcher.addChildExecutionId(child.getId())) {
+ return;
+ }
+ parent.setDirty(true);
+ persistCacheEntry(parent);
+ parent.setDirty(false);
+ } catch (Exception e) {
+ LogChannel.GENERAL.logError(
+ "Unable to record child execution '"
+ + child.getId()
+ + "' on '"
+ + child.getParentId()
+ + "' (non-fatal)",
+ e);
+ } finally {
+ if (loadedForLink) {
+ cache.remove(child.getParentId());
+ }
+ }
+ }
+
+ /**
+ * An action or transform that has not been flushed yet. The id and the
child link are enough for
+ * drill-down. The parent's next save replaces this with the real launcher
and keeps the child id.
+ */
+ private static Execution minimalLauncher(Execution child, Execution
parentExecution) {
+ Execution launcher = new Execution();
+ launcher.setId(child.getLaunchedById());
+ launcher.setParentId(child.getParentId());
+ launcher.setExecutionType(
+ parentExecution.getExecutionType() == ExecutionType.Workflow
+ ? ExecutionType.Action
+ : ExecutionType.Transform);
+ launcher.setRegistrationDate(
+ child.getRegistrationDate() != null ? child.getRegistrationDate() :
new Date());
+ return launcher;
+ }
+
@Override
public List<String> getExecutionIds(boolean includeChildren, int limit)
throws HopException {
Set<DatedId> ids = new HashSet<>();
@@ -536,6 +607,10 @@ public abstract class BaseCachingExecutionInfoLocation
implements IExecutionInfo
if (cacheEntry.getExecution() == null) {
continue;
}
+ // The parent document is not a place to store this execution's state.
+ if (parentId.equals(cacheEntry.getId())) {
+ continue;
+ }
String execParent = cacheEntry.getExecution().getParentId();
if (parentId.equals(execParent)) {
found = cacheEntry;
@@ -798,9 +873,6 @@ public abstract class BaseCachingExecutionInfoLocation
implements IExecutionInfo
@Override
public synchronized Execution getExecution(String executionId) throws
HopException {
CacheEntry entry = findCacheEntry(executionId);
- if (entry == null) {
- entry = findCacheEntryWithChild(executionId);
- }
if (entry == null) {
return null;
}
@@ -837,9 +909,6 @@ public abstract class BaseCachingExecutionInfoLocation
implements IExecutionInfo
if (StringUtils.isNotEmpty(parentExecutionId)) {
CacheEntry parentEntry = findCacheEntry(parentExecutionId);
- if (parentEntry == null) {
- parentEntry = findCacheEntryWithChild(parentExecutionId);
- }
if (parentEntry != null && parentEntry.getChildExecutions() != null) {
for (Execution childExecution :
parentEntry.getChildExecutions().values()) {
if (childExecution != null &&
parentExecutionId.equals(childExecution.getParentId())) {
@@ -853,7 +922,8 @@ public abstract class BaseCachingExecutionInfoLocation
implements IExecutionInfo
Execution execution = getExecution(id);
if (execution != null
&& parentExecutionId != null
- && parentExecutionId.equals(execution.getParentId())) {
+ && (parentExecutionId.equals(execution.getParentId())
+ || parentExecutionId.equals(execution.getLaunchedById()))) {
executions.add(execution);
}
}
@@ -1000,9 +1070,6 @@ public abstract class BaseCachingExecutionInfoLocation
implements IExecutionInfo
return null;
}
CacheEntry cacheEntry = findCacheEntry(childId);
- if (cacheEntry == null) {
- cacheEntry = findCacheEntryWithChild(childId);
- }
if (cacheEntry == null) {
return null;
}
@@ -1019,23 +1086,4 @@ public abstract class BaseCachingExecutionInfoLocation
implements IExecutionInfo
}
return null;
}
-
- protected synchronized CacheEntry findCacheEntryWithChild(String childId)
throws HopException {
- if (StringUtils.isEmpty(childId)) {
- return null;
- }
- for (CacheEntry cacheEntry : cache.values()) {
- if (cacheEntry.peekChildExecution(childId) != null
- || (cacheEntry.getChildIds() != null &&
cacheEntry.getChildIds().contains(childId))) {
- cache.get(cacheEntry.getId());
- cacheEntry.markRead();
- return cacheEntry;
- }
- }
- return loadCacheEntryWithChild(childId);
- }
-
- protected CacheEntry loadCacheEntryWithChild(String childId) throws
HopException {
- return null;
- }
}
diff --git
a/engine/src/main/java/org/apache/hop/execution/caching/CacheEntry.java
b/engine/src/main/java/org/apache/hop/execution/caching/CacheEntry.java
index f5caf13fa9..eae0067049 100644
--- a/engine/src/main/java/org/apache/hop/execution/caching/CacheEntry.java
+++ b/engine/src/main/java/org/apache/hop/execution/caching/CacheEntry.java
@@ -254,6 +254,10 @@ public class CacheEntry {
}
public void addChildExecution(Execution childExecution) {
+ Execution existing = childExecutions.get(childExecution.getId());
+ if (existing != null) {
+ childExecution.unionChildExecutionIds(existing);
+ }
childExecutions.put(childExecution.getId(), childExecution);
flagDirty();
}
diff --git
a/engine/src/main/java/org/apache/hop/execution/caching/CachingFileExecutionInfoLocation.java
b/engine/src/main/java/org/apache/hop/execution/caching/CachingFileExecutionInfoLocation.java
index cd5728c324..da976fbf61 100644
---
a/engine/src/main/java/org/apache/hop/execution/caching/CachingFileExecutionInfoLocation.java
+++
b/engine/src/main/java/org/apache/hop/execution/caching/CachingFileExecutionInfoLocation.java
@@ -23,14 +23,11 @@ import com.fasterxml.jackson.core.JsonToken;
import com.fasterxml.jackson.databind.ObjectMapper;
import java.io.IOException;
import java.io.InputStream;
-import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.ZonedDateTime;
-import java.util.ArrayList;
import java.util.Date;
-import java.util.List;
import java.util.Map;
import java.util.Set;
import lombok.Getter;
@@ -51,6 +48,7 @@ import org.apache.hop.core.json.HopJson;
import org.apache.hop.core.logging.LogChannel;
import org.apache.hop.core.variables.IVariables;
import org.apache.hop.core.vfs.HopVfs;
+import org.apache.hop.execution.Execution;
import org.apache.hop.execution.ExecutionDeleter;
import org.apache.hop.execution.ExecutionInfoLocation;
import org.apache.hop.execution.IExecutionInfoLocation;
@@ -183,7 +181,7 @@ public class CachingFileExecutionInfoLocation extends
BaseCachingExecutionInfoLo
if (onDisk == null) {
return;
}
- mergeMap(onDisk.getChildExecutions(), cacheEntry.getChildExecutions());
+ mergeChildExecutions(onDisk.getChildExecutions(),
cacheEntry.getChildExecutions());
mergeMap(onDisk.getChildExecutionStates(),
cacheEntry.getChildExecutionStates());
mergeMap(onDisk.getChildExecutionData(),
cacheEntry.getChildExecutionData());
cacheEntry.keepStoredProjectId(onDisk);
@@ -203,6 +201,24 @@ public class CachingFileExecutionInfoLocation extends
BaseCachingExecutionInfoLo
}
}
+ /**
+ * Same as {@link #mergeMap}, and when both sides already have an action or
transform, keep the
+ * child execution ids recorded by either writer.
+ */
+ private static void mergeChildExecutions(
+ Map<String, Execution> fromDisk, Map<String, Execution> into) {
+ if (fromDisk == null || fromDisk.isEmpty() || into == null) {
+ return;
+ }
+ for (Map.Entry<String, Execution> entry : fromDisk.entrySet()) {
+ Execution stored = entry.getValue();
+ Execution kept = into.putIfAbsent(entry.getKey(), stored);
+ if (kept != null && stored != null) {
+ kept.unionChildExecutionIds(stored);
+ }
+ }
+ }
+
@Override
public void deleteCacheEntry(CacheEntry cacheEntry) throws HopException {
try {
@@ -422,63 +438,6 @@ public class CachingFileExecutionInfoLocation extends
BaseCachingExecutionInfoLo
}
}
- public static final int MAX_CHILD_SEARCH_SCAN = 100;
-
- @Override
- protected synchronized CacheEntry loadCacheEntryWithChild(String childId)
throws HopException {
- if (StringUtils.isEmpty(childId)) {
- return null;
- }
- try {
- FileObject[] files = getAllFileObjects(actualRootFolder);
- if (files == null || files.length == 0) {
- return null;
- }
-
- try {
- record FileWithTimestamp(FileObject file, long lastModified) {}
- List<FileWithTimestamp> fileList = new ArrayList<>(files.length);
- for (FileObject file : files) {
- long lastModified = 0L;
- try {
- lastModified = file.getContent().getLastModifiedTime();
- } catch (Exception ignored) {
- }
- fileList.add(new FileWithTimestamp(file, lastModified));
- }
- fileList.sort((a, b) -> Long.compare(b.lastModified, a.lastModified));
-
- int maxScan = Math.min(fileList.size(),
Math.max(MAX_CHILD_SEARCH_SCAN, maxSize * 2));
- for (int i = 0; i < maxScan; i++) {
- FileObject file = fileList.get(i).file();
- String id = getIdFromFileName(file);
- if (id.equals(childId)) {
- continue;
- }
- String content =
- HopVfs.getTextFileContent(file.getName().getURI(),
StandardCharsets.UTF_8);
- if (content != null && content.contains(childId)) {
- CacheEntry entry = HopJson.newMapper().readValue(content,
CacheEntry.class);
- if (entry != null
- && (entry.peekChildExecution(childId) != null
- || (entry.getChildIds() != null &&
entry.getChildIds().contains(childId)))) {
- cache.put(entry.getId(), entry);
- enforceMaxCacheSize();
- return entry;
- }
- }
- }
- } finally {
- for (FileObject file : files) {
- closeQuietly(file);
- }
- }
- } catch (Exception e) {
- LogChannel.GENERAL.logError("Error searching for cache entry containing
child " + childId, e);
- }
- return null;
- }
-
@Override
protected void retrieveIds(
boolean includeChildren, Set<DatedId> ids, int limit, final
IExecutionSelector selector)
diff --git
a/engine/src/main/java/org/apache/hop/execution/local/FileExecutionInfoLocation.java
b/engine/src/main/java/org/apache/hop/execution/local/FileExecutionInfoLocation.java
index beecae6a42..81ae8c73c3 100644
---
a/engine/src/main/java/org/apache/hop/execution/local/FileExecutionInfoLocation.java
+++
b/engine/src/main/java/org/apache/hop/execution/local/FileExecutionInfoLocation.java
@@ -729,7 +729,13 @@ public class FileExecutionInfoLocation implements
IExecutionInfoLocation {
for (String id : getExecutionIds(true, 10000)) {
Execution execution = getExecution(id);
- if (parentExecutionId.equals(execution.getParentId())) {
+ if (execution == null) {
+ continue;
+ }
+ // parentId is the owning workflow or pipeline. launchedById is the
action or transform
+ // that started this execution. Callers pass either one.
+ if (parentExecutionId.equals(execution.getParentId())
+ || parentExecutionId.equals(execution.getLaunchedById())) {
executions.add(execution);
}
}
diff --git
a/engine/src/test/java/org/apache/hop/execution/ExecutionBuilderParentLinkTest.java
b/engine/src/test/java/org/apache/hop/execution/ExecutionBuilderParentLinkTest.java
new file mode 100644
index 0000000000..a5de84807d
--- /dev/null
+++
b/engine/src/test/java/org/apache/hop/execution/ExecutionBuilderParentLinkTest.java
@@ -0,0 +1,104 @@
+/*
+ * 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.hop.execution;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.withSettings;
+
+import org.apache.hop.core.logging.ILoggingObject;
+import org.apache.hop.pipeline.PipelineMeta;
+import org.apache.hop.pipeline.engine.IPipelineEngine;
+import org.apache.hop.pipeline.transform.ITransform;
+import org.apache.hop.workflow.WorkflowMeta;
+import org.apache.hop.workflow.action.IAction;
+import org.apache.hop.workflow.engine.IWorkflowEngine;
+import org.junit.jupiter.api.Test;
+
+class ExecutionBuilderParentLinkTest {
+
+ @Test
+ void workflowActionStoresTheWorkflowAsParent() {
+ @SuppressWarnings("unchecked")
+ IWorkflowEngine<WorkflowMeta> workflow = mock(IWorkflowEngine.class);
+ when(workflow.getLogChannelId()).thenReturn("workflow-1");
+
+ ILoggingObject loggingParent =
+ mock(ILoggingObject.class,
withSettings().extraInterfaces(IAction.class));
+ IAction action = (IAction) loggingParent;
+ when(action.getParentWorkflow()).thenReturn(workflow);
+ when(loggingParent.getLogChannelId()).thenReturn("action-1");
+
+ assertEquals("workflow-1",
ExecutionBuilder.executionParentId(loggingParent));
+ assertEquals("action-1", ExecutionBuilder.launchedById(loggingParent));
+ }
+
+ @Test
+ void transformStoresThePipelineAsParent() {
+ @SuppressWarnings("unchecked")
+ IPipelineEngine<PipelineMeta> pipeline = mock(IPipelineEngine.class);
+ when(pipeline.getLogChannelId()).thenReturn("pipeline-1");
+
+ ILoggingObject loggingParent =
+ mock(ILoggingObject.class,
withSettings().extraInterfaces(ITransform.class));
+ ITransform transform = (ITransform) loggingParent;
+ when(transform.getPipeline()).thenReturn(pipeline);
+ when(loggingParent.getLogChannelId()).thenReturn("transform-1");
+
+ assertEquals("pipeline-1",
ExecutionBuilder.executionParentId(loggingParent));
+ assertEquals("transform-1", ExecutionBuilder.launchedById(loggingParent));
+ }
+
+ @Test
+ void rootExecutorHasNoLauncher() {
+ ILoggingObject root = mock(ILoggingObject.class);
+ when(root.getLogChannelId()).thenReturn("root-1");
+
+ assertEquals("root-1", ExecutionBuilder.executionParentId(root));
+ assertNull(ExecutionBuilder.launchedById(root));
+ assertNull(ExecutionBuilder.executionParentId(null));
+ assertNull(ExecutionBuilder.launchedById(null));
+ }
+
+ @Test
+ void actionWithoutAWorkflowKeepsItsOwnId() {
+ ILoggingObject loggingParent =
+ mock(ILoggingObject.class,
withSettings().extraInterfaces(IAction.class));
+ when(((IAction) loggingParent).getParentWorkflow()).thenReturn(null);
+ when(loggingParent.getLogChannelId()).thenReturn("action-1");
+
+ assertEquals("action-1",
ExecutionBuilder.executionParentId(loggingParent));
+ assertNull(ExecutionBuilder.launchedById(loggingParent));
+ }
+
+ @Test
+ void buildCopiesTheLauncher() {
+ Execution execution =
+ ExecutionBuilder.of()
+ .withName("pipeline")
+ .withId("pipe-1")
+ .withParentId("workflow-1")
+ .withLaunchedById("action-1")
+ .build();
+
+ assertEquals("workflow-1", execution.getParentId());
+ assertEquals("action-1", execution.getLaunchedById());
+ }
+}
diff --git
a/engine/src/test/java/org/apache/hop/execution/caching/BaseCachingExecutionInfoLocationTest.java
b/engine/src/test/java/org/apache/hop/execution/caching/BaseCachingExecutionInfoLocationTest.java
index 0c7d9962ef..6bbba58d8d 100644
---
a/engine/src/test/java/org/apache/hop/execution/caching/BaseCachingExecutionInfoLocationTest.java
+++
b/engine/src/test/java/org/apache/hop/execution/caching/BaseCachingExecutionInfoLocationTest.java
@@ -180,7 +180,8 @@ class BaseCachingExecutionInfoLocationTest {
Execution pipe = new Execution();
pipe.setId(pipelineId);
pipe.setName("Pipeline");
- pipe.setParentId(actionId);
+ pipe.setParentId(workflowId);
+ pipe.setLaunchedById(actionId);
pipe.setExecutionType(ExecutionType.Pipeline);
pipe.setRegistrationDate(new Date());
@@ -188,7 +189,7 @@ class BaseCachingExecutionInfoLocationTest {
location.registerExecution(action);
location.registerExecution(pipe);
- assertEquals(actionId, location.findParentId(pipelineId));
+ assertEquals(workflowId, location.findParentId(pipelineId));
assertEquals(workflowId, location.findParentId(actionId));
assertNull(location.findParentId(workflowId));
@@ -196,6 +197,7 @@ class BaseCachingExecutionInfoLocationTest {
assertNotNull(retrievedAction);
assertEquals(actionId, retrievedAction.getId());
assertEquals(ExecutionType.Action, retrievedAction.getExecutionType());
+ assertEquals(java.util.List.of(pipelineId),
retrievedAction.getChildExecutionIds());
java.util.List<Execution> actionChildren =
location.findExecutions(actionId);
assertEquals(1, actionChildren.size());
diff --git
a/engine/src/test/java/org/apache/hop/execution/caching/CachingFileExecutionInfoLocationTest.java
b/engine/src/test/java/org/apache/hop/execution/caching/CachingFileExecutionInfoLocationTest.java
index 252c2f39ba..e32912cf49 100644
---
a/engine/src/test/java/org/apache/hop/execution/caching/CachingFileExecutionInfoLocationTest.java
+++
b/engine/src/test/java/org/apache/hop/execution/caching/CachingFileExecutionInfoLocationTest.java
@@ -46,6 +46,7 @@ import org.apache.hop.execution.Execution;
import org.apache.hop.execution.ExecutionData;
import org.apache.hop.execution.ExecutionDataBuilder;
import org.apache.hop.execution.ExecutionDataSetMeta;
+import org.apache.hop.execution.ExecutionState;
import org.apache.hop.execution.ExecutionType;
import org.apache.hop.execution.IExecutionSelector;
import org.junit.jupiter.api.AfterEach;
@@ -549,7 +550,8 @@ class CachingFileExecutionInfoLocationTest {
Execution pipe = new Execution();
pipe.setId(pipelineId);
pipe.setName("Pipeline");
- pipe.setParentId(actionId);
+ pipe.setParentId(workflowId);
+ pipe.setLaunchedById(actionId);
pipe.setExecutionType(ExecutionType.Pipeline);
pipe.setRegistrationDate(new Date());
@@ -557,20 +559,32 @@ class CachingFileExecutionInfoLocationTest {
location.registerExecution(action);
location.registerExecution(pipe);
+ // A later write of the workflow must keep the child id recorded on the
action.
+ Execution cachedAction = location.getExecution(actionId);
+ assertEquals(List.of(pipelineId), cachedAction.getChildExecutionIds());
+ cachedAction.getChildExecutionIds().clear();
+ location.persistCacheEntry(location.getCache().get(workflowId));
+
// Persist to disk and clear cache so nothing is in memory
for (CacheEntry entry : location.getCache().values()) {
location.persistCacheEntry(entry);
}
location.clearCaches();
- assertEquals(actionId, location.findParentId(pipelineId));
- assertEquals(workflowId, location.findParentId(actionId));
- assertNull(location.findParentId(workflowId));
+ assertEquals(workflowId, location.findParentId(pipelineId));
+ // The action lives in the workflow file. It is not a file of its own,
and loading the
+ // pipeline does not load it.
+ assertNull(location.findParentId(actionId));
+ assertNull(location.getExecution(actionId));
+ assertNull(location.findParentId(workflowId));
+ Execution reloadedWorkflow = location.getExecution(workflowId);
+ assertNotNull(reloadedWorkflow);
Execution retrievedAction = location.getExecution(actionId);
assertNotNull(retrievedAction);
assertEquals(actionId, retrievedAction.getId());
assertEquals(ExecutionType.Action, retrievedAction.getExecutionType());
+ assertEquals(List.of(pipelineId),
retrievedAction.getChildExecutionIds());
assertNull(location.getExecution("non-existent-id"));
@@ -585,6 +599,90 @@ class CachingFileExecutionInfoLocationTest {
}
}
+ @Test
+ void aSecondLocationRecordsTheChildWhenTheLauncherIsNotOnDiskYet() throws
Exception {
+ Path root = tempDir.resolve("two-locations");
+ String rootFolder = root.toAbsolutePath().toString();
+ CachingFileExecutionInfoLocation parentLocation = open(rootFolder);
+ CachingFileExecutionInfoLocation childLocation = open(rootFolder);
+ try {
+ Execution pipeline = new Execution();
+ pipeline.setId("pipe");
+ pipeline.setName("Pipe");
+ pipeline.setExecutionType(ExecutionType.Pipeline);
+ pipeline.setRegistrationDate(new Date());
+ parentLocation.registerExecution(pipeline);
+
+ ExecutionState parentState = new ExecutionState();
+ parentState.setId("pipe");
+ parentState.setName("Pipe");
+ parentState.setExecutionType(ExecutionType.Pipeline);
+ parentState.setStatusDescription("parent-status");
+ parentState.setUpdateTime(new Date());
+ parentLocation.updateExecutionState(parentState);
+ parentLocation.persistCacheEntry(parentLocation.getCache().get("pipe"));
+
+ // The transform is in this process only. The 5s timer has not written
it yet.
+ Execution transform = new Execution();
+ transform.setId("transform");
+ transform.setName("Executor");
+ transform.setParentId("pipe");
+ transform.setExecutionType(ExecutionType.Transform);
+ transform.setRegistrationDate(new Date());
+ parentLocation.registerExecution(transform);
+
+ Execution child = new Execution();
+ child.setId("child");
+ child.setName("Child");
+ child.setExecutionType(ExecutionType.Pipeline);
+ child.setParentId("pipe");
+ child.setLaunchedById("transform");
+ child.setRegistrationDate(new Date());
+ childLocation.registerExecution(child);
+
+ assertNull(childLocation.getCache().get("pipe"));
+
+ // The child's own file is gone, so a state update can only fall back to
a cached parent.
+ childLocation.getCache().remove("child");
+ Files.deleteIfExists(root.resolve("child.json"));
+ ExecutionState childState = new ExecutionState();
+ childState.setId("child");
+ childState.setName("Child");
+ childState.setExecutionType(ExecutionType.Pipeline);
+ childState.setParentId("pipe");
+ childState.setStatusDescription("child-status");
+ childState.setUpdateTime(new Date());
+ childLocation.updateExecutionState(childState);
+ childLocation.close();
+
+ CachingFileExecutionInfoLocation reader = open(rootFolder);
+ try {
+ assertEquals("parent-status",
reader.getExecutionState("pipe").getStatusDescription());
+ } finally {
+ reader.close();
+ }
+
+ parentLocation.persistCacheEntry(parentLocation.getCache().get("pipe"));
+ parentLocation.clearCaches();
+
+ assertNotNull(parentLocation.getExecution("pipe"));
+ Execution launcher = parentLocation.getExecution("transform");
+ assertNotNull(launcher);
+ assertEquals("Executor", launcher.getName());
+ assertEquals(List.of("child"), launcher.getChildExecutionIds());
+ } finally {
+ childLocation.close();
+ parentLocation.close();
+ }
+ }
+
+ private static CachingFileExecutionInfoLocation open(String rootFolder)
throws Exception {
+ CachingFileExecutionInfoLocation location = new
CachingFileExecutionInfoLocation();
+ location.setRootFolder(rootFolder);
+ location.initialize(new Variables(), null);
+ return location;
+ }
+
private static Execution executionAt(String id, Date start) {
Execution execution = execution(id, null);
execution.setExecutionStartDate(start);
diff --git
a/engine/src/test/java/org/apache/hop/execution/local/FileExecutionInfoLocationTest.java
b/engine/src/test/java/org/apache/hop/execution/local/FileExecutionInfoLocationTest.java
index 56c174d326..2e83edb20a 100644
---
a/engine/src/test/java/org/apache/hop/execution/local/FileExecutionInfoLocationTest.java
+++
b/engine/src/test/java/org/apache/hop/execution/local/FileExecutionInfoLocationTest.java
@@ -26,6 +26,7 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Date;
+import java.util.List;
import java.util.UUID;
import org.apache.commons.io.FileUtils;
import org.apache.hop.core.variables.Variables;
@@ -127,6 +128,23 @@ class FileExecutionInfoLocationTest {
assertNull(location.getExecution("grandchild"));
}
+ @Test
+ void findExecutionsMatchesTheLauncherAsWellAsTheOwner() throws Exception {
+ FileExecutionInfoLocation location = openLocation();
+ Execution pipe = execution("pipe", "workflow", new Date(1_000L));
+ pipe.setLaunchedById("action");
+ location.registerExecution(pipe);
+
+ List<Execution> byOwner = location.findExecutions("workflow");
+ List<Execution> byLauncher = location.findExecutions("action");
+
+ assertEquals(1, byOwner.size());
+ assertEquals("pipe", byOwner.get(0).getId());
+ assertEquals(1, byLauncher.size());
+ assertEquals("pipe", byLauncher.get(0).getId());
+ assertEquals("workflow", location.findParentId("pipe"));
+ }
+
private FileExecutionInfoLocation openLocation() throws Exception {
FileExecutionInfoLocation location =
new FileExecutionInfoLocation(tempDir.resolve("root").toString());
diff --git a/integration-tests/exec-info/002-validation.hpl
b/integration-tests/exec-info/002-validation.hpl
index 1ad604f812..274bc8057f 100644
--- a/integration-tests/exec-info/002-validation.hpl
+++ b/integration-tests/exec-info/002-validation.hpl
@@ -218,8 +218,8 @@ limitations under the License.
<jsScript_type>0</jsScript_type>
<jsScript_name>Script 1</jsScript_name>
<jsScript_script>
-if (executionId == null || parentId == null) {
- throw "Pipeline execution ID or parent action ID is null!";
+if (executionId == null || parentId == null || launchedById == null) {
+ throw "Pipeline execution is missing its parent workflow or the action that
started it!";
}
</jsScript_script>
</jsScript>
@@ -274,7 +274,7 @@ if (executionId == null || parentId == null) {
<jsScript_name>Script 1</jsScript_name>
<jsScript_script>
if (id == null || !id.equals(parentId)) {
- throw "findParentId(pipelineId) [" + id + "] did not match parent action ID
[" + parentId + "]!";
+ throw "findParentId(pipelineId) [" + id + "] did not match parent workflow
ID [" + parentId + "]!";
}
</jsScript_script>
</jsScript>
@@ -366,7 +366,7 @@ if (id == null || !id.equals(parentId)) {
<jsScript_name>Script 1</jsScript_name>
<jsScript_script>
if (id == null || id.equals("") || id.equals(parentId)) {
- throw "findParentId(actionId) [" + id + "] is null or equal to actionId
itself!";
+ throw "findParentId(workflowId) [" + id + "] is null or equal to the
workflow itself!";
}
</jsScript_script>
</jsScript>
@@ -391,7 +391,7 @@ if (id == null || id.equals("") || id.equals(parentId)) {
</partitioning>
<fields>
<field>
- <name>parentId</name>
+ <name>launchedById</name>
</field>
<field>
<name>executionId</name>
@@ -419,7 +419,7 @@ if (id == null || id.equals("") || id.equals(parentId)) {
<location>caching-file</location>
<operationType>FindExecutions</operationType>
<idFieldName/>
- <parentIdFieldName>parentId</parentIdFieldName>
+ <parentIdFieldName>launchedById</parentIdFieldName>
<typeFieldName/>
<nameFieldName/>
<includeChildrenFieldName/>
@@ -448,7 +448,7 @@ if (id == null || id.equals("") || id.equals(parentId)) {
<jsScript_name>Script 1</jsScript_name>
<jsScript_script>
if (executionId == null || !executionId.equals(origPipelineId)) {
- throw "findExecutions(actionId) did not return expected pipeline execution
ID [" + origPipelineId + "]! Got: [" + executionId + "]";
+ throw "findExecutions(launchedById) did not return expected pipeline
execution ID [" + origPipelineId + "]! Got: [" + executionId + "]";
}
</jsScript_script>
</jsScript>
diff --git
a/plugins/actions/pipeline/src/main/java/org/apache/hop/workflow/actions/pipeline/ActionPipeline.java
b/plugins/actions/pipeline/src/main/java/org/apache/hop/workflow/actions/pipeline/ActionPipeline.java
index ab9bf5c2aa..c19623cdf2 100644
---
a/plugins/actions/pipeline/src/main/java/org/apache/hop/workflow/actions/pipeline/ActionPipeline.java
+++
b/plugins/actions/pipeline/src/main/java/org/apache/hop/workflow/actions/pipeline/ActionPipeline.java
@@ -520,6 +520,8 @@ public class ActionPipeline extends ActionBase implements
Cloneable, IAction {
pipeline =
PipelineEngineFactory.createPipelineEngine(
this, runConfiguration, getMetadataProvider(), pipelineMeta);
+ // Logging parent stays this action so log lines nest under it.
ExecutionBuilder stores the
+ // workflow log channel id as the execution parent.
pipeline.setParent(this);
// set the parent workflow on the pipeline, variables are taken from
here...
diff --git
a/plugins/actions/workflow/src/main/java/org/apache/hop/workflow/actions/workflow/ActionWorkflow.java
b/plugins/actions/workflow/src/main/java/org/apache/hop/workflow/actions/workflow/ActionWorkflow.java
index 6d9504b60b..7fcaf40a7b 100644
---
a/plugins/actions/workflow/src/main/java/org/apache/hop/workflow/actions/workflow/ActionWorkflow.java
+++
b/plugins/actions/workflow/src/main/java/org/apache/hop/workflow/actions/workflow/ActionWorkflow.java
@@ -456,7 +456,9 @@ public class ActionWorkflow extends ActionBase implements
Cloneable, IAction {
}
}
- // Create a new workflow
+ // Create a new workflow. The last argument is the logging parent
(this action), so log
+ // lines nest under the action. ExecutionBuilder stores the workflow
log channel id as the
+ // execution parent.
//
workflow =
WorkflowEngineFactory.createWorkflowEngine(
diff --git
a/plugins/misc/execution-database/src/main/java/org/apache/hop/execution/database/CachingDatabaseExecutionInfoLocation.java
b/plugins/misc/execution-database/src/main/java/org/apache/hop/execution/database/CachingDatabaseExecutionInfoLocation.java
index 12dc7a1649..9ae805489b 100644
---
a/plugins/misc/execution-database/src/main/java/org/apache/hop/execution/database/CachingDatabaseExecutionInfoLocation.java
+++
b/plugins/misc/execution-database/src/main/java/org/apache/hop/execution/database/CachingDatabaseExecutionInfoLocation.java
@@ -53,9 +53,11 @@ import org.apache.hop.core.variables.IVariables;
import org.apache.hop.execution.Execution;
import org.apache.hop.execution.ExecutionInfoLocation;
import org.apache.hop.execution.ExecutionState;
+import org.apache.hop.execution.ExecutionType;
import org.apache.hop.execution.IExecutionInfoLocation;
import org.apache.hop.execution.IExecutionSelector;
import org.apache.hop.execution.LastPeriod;
+import org.apache.hop.execution.ListedExecution;
import org.apache.hop.execution.caching.BaseCachingExecutionInfoLocation;
import org.apache.hop.execution.caching.CacheEntry;
import org.apache.hop.execution.caching.DatedId;
@@ -502,7 +504,7 @@ public class CachingDatabaseExecutionInfoLocation extends
BaseCachingExecutionIn
if (existing == null) {
return;
}
- mergeMap(existing.getChildExecutions(), cacheEntry.getChildExecutions());
+ mergeChildExecutions(existing.getChildExecutions(),
cacheEntry.getChildExecutions());
mergeMap(existing.getChildExecutionStates(),
cacheEntry.getChildExecutionStates());
mergeMap(existing.getChildExecutionData(),
cacheEntry.getChildExecutionData());
cacheEntry.keepStoredProjectId(existing);
@@ -521,6 +523,24 @@ public class CachingDatabaseExecutionInfoLocation extends
BaseCachingExecutionIn
}
}
+ /**
+ * Same as {@link #mergeMap}, and when both sides already have an action or
transform, keep the
+ * child execution ids recorded by either writer.
+ */
+ private static void mergeChildExecutions(
+ Map<String, Execution> fromStore, Map<String, Execution> into) {
+ if (fromStore == null || fromStore.isEmpty() || into == null) {
+ return;
+ }
+ for (Map.Entry<String, Execution> entry : fromStore.entrySet()) {
+ Execution stored = entry.getValue();
+ Execution kept = into.putIfAbsent(entry.getKey(), stored);
+ if (kept != null && stored != null) {
+ kept.unionChildExecutionIds(stored);
+ }
+ }
+ }
+
private boolean rowExists(String id) throws HopException {
String sql =
"SELECT 1 FROM "
@@ -720,112 +740,183 @@ public class CachingDatabaseExecutionInfoLocation
extends BaseCachingExecutionIn
}
@Override
- protected CacheEntry loadCacheEntryWithChild(String childId) throws
HopException {
- if (StringUtils.isEmpty(childId)) {
- return null;
+ public void deleteCacheEntry(CacheEntry cacheEntry) throws HopException {
+ if (cacheEntry == null || StringUtils.isEmpty(cacheEntry.getId())) {
+ return;
}
try {
- String columns = databaseMeta.quoteField(COL_JSON);
- if (stateJsonColumnAvailable) {
- columns += ", " + databaseMeta.quoteField(COL_STATE_JSON);
- }
+ String sql =
+ "DELETE FROM "
+ + getQuotedSchemaTable()
+ + " WHERE "
+ + databaseMeta.quoteField(COL_ID)
+ + " = ?";
+ IRowMeta paramMeta = new RowMeta();
+ paramMeta.addValueMeta(new ValueMetaString(COL_ID, 100, -1));
+ callWithDatabase(
+ () -> {
+ database.execStatement(sql, paramMeta, new Object[]
{cacheEntry.getId()});
+ return null;
+ });
+ } catch (Exception e) {
+ throw new HopException(
+ "Error deleting execution information from database for id '" +
cacheEntry.getId() + "'",
+ e);
+ }
+ }
+
+ /** Same cap as {@link BaseCachingExecutionInfoLocation#findExecutionIDs}. */
+ private static final int LIST_LIMIT = 50;
+
+ /**
+ * One query of the summary columns. The {@code json} and {@code state_json}
documents, which hold
+ * the pipeline XML, metadata, and sample rows, are not read.
+ */
+ @Override
+ public List<ListedExecution> listExecutions(IExecutionSelector selector)
throws HopException {
+ final IExecutionSelector active = selector == null ?
IExecutionSelector.ALL : selector;
+ try {
StringBuilder sql = new StringBuilder();
sql.append("SELECT ")
- .append(columns)
- .append(" FROM ")
- .append(getQuotedSchemaTable())
- .append(" WHERE (")
- .append(databaseMeta.quoteField(COL_JSON))
- .append(" LIKE ?");
- if (stateJsonColumnAvailable) {
- sql.append(" OR
").append(databaseMeta.quoteField(COL_STATE_JSON)).append(" LIKE ?");
- }
- sql.append(") AND ")
.append(databaseMeta.quoteField(COL_ID))
- .append(" != ?")
- .append(" ORDER BY ")
+ .append(", ")
+ .append(databaseMeta.quoteField(COL_NAME))
+ .append(", ")
+ .append(databaseMeta.quoteField(COL_EXECUTION_TYPE))
+ .append(", ")
+ .append(databaseMeta.quoteField(COL_PARENT_ID))
+ .append(", ")
+ .append(databaseMeta.quoteField(COL_REGISTRATION_DATE))
+ .append(", ")
.append(databaseMeta.quoteField(COL_EXECUTION_START_DATE))
- .append(" DESC")
- .append(databaseMeta.getLimitClause(20));
+ .append(", ")
+ .append(databaseMeta.quoteField(COL_EXECUTION_END_DATE))
+ .append(", ")
+ .append(databaseMeta.quoteField(COL_FAILED))
+ .append(", ")
+ .append(databaseMeta.quoteField(COL_STATUS_DESCRIPTION))
+ .append(" FROM ")
+ .append(getQuotedSchemaTable());
- IRowMeta paramMeta = new RowMeta();
- paramMeta.addValueMeta(new ValueMetaString("p1", 100, -1));
List<Object> params = new ArrayList<>();
- params.add("%" + childId + "%");
- if (stateJsonColumnAvailable) {
- paramMeta.addValueMeta(new ValueMetaString("p2", 100, -1));
- params.add("%" + childId + "%");
+ IRowMeta paramMeta = new RowMeta();
+ List<String> where = new ArrayList<>();
+ appendSelectorFilters(active, where, paramMeta, params);
+ if (!where.isEmpty()) {
+ sql.append(" WHERE ");
+ sql.append(String.join(" AND ", where));
}
- paramMeta.addValueMeta(new ValueMetaString("p_not_id", 100, -1));
- params.add(childId);
+ sql.append(" ORDER BY ")
+ .append(databaseMeta.quoteField(COL_EXECUTION_START_DATE))
+ .append(" DESC");
+ sql.append(databaseMeta.getLimitClause(LIST_LIMIT));
return callWithDatabase(
() -> {
+ List<ListedExecution> listed = new ArrayList<>();
ResultSet rs = database.openQuery(sql.toString(), paramMeta,
params.toArray());
try {
Object[] row = database.getRow(rs);
while (row != null) {
- if (row[0] != null) {
- try {
- CacheEntry entry =
JSON_MAPPER.readValue(row[0].toString(), CacheEntry.class);
- if (entry != null && !entry.getId().equals(childId)) {
- if (stateJsonColumnAvailable
- && row.length > 1
- && row[1] != null
- && StringUtils.isNotEmpty(row[1].toString())) {
- applyStateOverlay(
- entry, JSON_MAPPER.readValue(row[1].toString(),
CacheEntry.class));
- }
- if (entry.peekChildExecution(childId) != null
- || (entry.getChildIds() != null
- && entry.getChildIds().contains(childId))) {
- cache.put(entry.getId(), entry);
- enforceMaxCacheSize();
- return entry;
- }
- }
- } catch (Exception e) {
- // Ignore corrupted candidate and continue checking other
rows
- }
+ ListedExecution item = toListedExecution(row);
+ if (item != null
+ && active.isSelected(item.execution())
+ && active.isSelected(item.state())) {
+ listed.add(item);
}
row = database.getRow(rs);
}
- return null;
} finally {
database.closeQuery(rs);
}
+ for (ListedExecution item : listed) {
+ copyUpdateTimeWhenUnfinished(item);
+ }
+ return listed;
});
} catch (Exception e) {
- LogChannel.GENERAL.logError(
- "Error finding cache entry containing child " + childId + " from
database", e);
+ throw new HopException(
+ "Error listing executions from database table " +
getQuotedSchemaTable(), e);
+ }
+ }
+
+ private static ListedExecution toListedExecution(Object[] row) {
+ String id = asString(row[0]);
+ if (StringUtils.isEmpty(id)) {
+ return null;
+ }
+ ExecutionType type;
+ try {
+ type = ExecutionType.valueOf(asString(row[2]));
+ } catch (Exception e) {
return null;
}
+ Execution execution = new Execution();
+ execution.setId(id);
+ execution.setName(asString(row[1]));
+ execution.setExecutionType(type);
+ execution.setParentId(asString(row[3]));
+ execution.setRegistrationDate(asDate(row[4]));
+ execution.setExecutionStartDate(asDate(row[5]));
+
+ ExecutionState state = new ExecutionState();
+ state.setId(id);
+ state.setName(execution.getName());
+ state.setExecutionType(type);
+ state.setParentId(execution.getParentId());
+ state.setExecutionEndDate(asDate(row[6]));
+ state.setFailed(asBoolean(row[7]));
+ state.setStatusDescription(Const.NVL(asString(row[8]), ""));
+ // The summary columns have no last-update time. listExecutions fills it
for a row that has
+ // not ended, so a run whose process died is stale instead of running
forever.
+ state.setUpdateTime(null);
+ return new ListedExecution(execution, state);
}
- @Override
- public void deleteCacheEntry(CacheEntry cacheEntry) throws HopException {
- if (cacheEntry == null || StringUtils.isEmpty(cacheEntry.getId())) {
+ /**
+ * A finished row does not go stale. An unfinished row needs the update time
stored in its state.
+ */
+ private void copyUpdateTimeWhenUnfinished(ListedExecution item) {
+ ExecutionState state = item == null ? null : item.state();
+ if (state == null
+ || state.getExecutionEndDate() != null
+ || StringUtils.isEmpty(state.getId())) {
return;
}
try {
- String sql =
- "DELETE FROM "
- + getQuotedSchemaTable()
- + " WHERE "
- + databaseMeta.quoteField(COL_ID)
- + " = ?";
- IRowMeta paramMeta = new RowMeta();
- paramMeta.addValueMeta(new ValueMetaString(COL_ID, 100, -1));
- callWithDatabase(
- () -> {
- database.execStatement(sql, paramMeta, new Object[]
{cacheEntry.getId()});
- return null;
- });
+ CacheEntry stored = loadCacheEntry(state.getId());
+ if (stored != null && stored.getExecutionState() != null) {
+ state.setUpdateTime(stored.getExecutionState().getUpdateTime());
+ }
} catch (Exception e) {
- throw new HopException(
- "Error deleting execution information from database for id '" +
cacheEntry.getId() + "'",
- e);
+ LogChannel.GENERAL.logError(
+ "Unable to read the update time of execution '" + state.getId() +
"'", e);
+ }
+ }
+
+ private static String asString(Object value) {
+ return value == null ? null : value.toString();
+ }
+
+ private static Date asDate(Object value) {
+ if (value instanceof Date date) {
+ return date;
+ }
+ return null;
+ }
+
+ private static boolean asBoolean(Object value) {
+ if (value instanceof Boolean bool) {
+ return bool;
+ }
+ if (value instanceof Number number) {
+ return number.intValue() != 0;
+ }
+ if (value == null) {
+ return false;
}
+ String text = value.toString();
+ return "Y".equalsIgnoreCase(text) || "true".equalsIgnoreCase(text);
}
@Override
diff --git
a/plugins/misc/execution-database/src/test/java/org/apache/hop/execution/database/CachingDatabaseExecutionInfoLocationTest.java
b/plugins/misc/execution-database/src/test/java/org/apache/hop/execution/database/CachingDatabaseExecutionInfoLocationTest.java
index 34b4ee794f..e83737870b 100644
---
a/plugins/misc/execution-database/src/test/java/org/apache/hop/execution/database/CachingDatabaseExecutionInfoLocationTest.java
+++
b/plugins/misc/execution-database/src/test/java/org/apache/hop/execution/database/CachingDatabaseExecutionInfoLocationTest.java
@@ -52,6 +52,7 @@ import org.apache.hop.execution.ExecutionState;
import org.apache.hop.execution.ExecutionType;
import org.apache.hop.execution.IExecutionSelector;
import org.apache.hop.execution.LastPeriod;
+import org.apache.hop.execution.ListedExecution;
import org.apache.hop.execution.caching.CacheEntry;
import org.apache.hop.execution.caching.DatedId;
import org.apache.hop.metadata.serializer.memory.MemoryMetadataProvider;
@@ -146,6 +147,75 @@ class CachingDatabaseExecutionInfoLocationTest {
assertNull(location.loadCacheEntry(id));
}
+ @Test
+ void listExecutionsDoesNotReadTheJsonDocument() throws Exception {
+ CountingLocation counting = new CountingLocation();
+ counting.setConnectionName("h2-exec");
+
counting.setTableName(CachingDatabaseExecutionInfoLocation.DEFAULT_TABLE_NAME);
+ counting.setPersistenceDelay("60000");
+ counting.setMaxCacheAge("86400000");
+ counting.setDatabaseMeta(databaseMeta);
+ counting.initialize(variables, metadataProvider);
+ try {
+ counting.persistCacheEntry(
+ sampleEntry(
+ UUID.randomUUID().toString(), "Listed", ExecutionType.Pipeline,
false, "Finished"));
+ counting.clearCaches();
+ counting.loads = 0;
+
+ List<ListedExecution> listed =
counting.listExecutions(IExecutionSelector.ALL);
+
+ assertEquals(1, listed.size());
+ assertEquals("Listed", listed.getFirst().execution().getName());
+ assertEquals("Finished",
listed.getFirst().state().getStatusDescription());
+ assertFalse(listed.getFirst().state().isFailed());
+ assertNull(listed.getFirst().execution().getExecutorXml());
+ assertEquals(0, counting.loads);
+ } finally {
+ counting.close();
+ }
+ }
+
+ @Test
+ void listExecutionsLoadsUpdateTimeOnlyForUnfinishedRows() throws Exception {
+ CountingLocation counting = new CountingLocation();
+ counting.setConnectionName("h2-exec");
+
counting.setTableName(CachingDatabaseExecutionInfoLocation.DEFAULT_TABLE_NAME);
+ counting.setPersistenceDelay("60000");
+ counting.setMaxCacheAge("86400000");
+ counting.setDatabaseMeta(databaseMeta);
+ counting.initialize(variables, metadataProvider);
+ try {
+ String runningId = UUID.randomUUID().toString();
+ Date stale = new Date(1_000L);
+ CacheEntry running = sampleEntry(runningId, "Live",
ExecutionType.Pipeline, false, "Running");
+ running.getExecution().setFilename("/files/live.hpl");
+ running.getExecutionState().setExecutionEndDate(null);
+ running.getExecutionState().setUpdateTime(stale);
+ counting.persistCacheEntry(running);
+
+ String finishedId = UUID.randomUUID().toString();
+ counting.persistCacheEntry(
+ sampleEntry(finishedId, "Done", ExecutionType.Pipeline, false,
"Finished"));
+ counting.clearCaches();
+ counting.loads = 0;
+
+ List<ListedExecution> listed =
counting.listExecutions(IExecutionSelector.ALL);
+
+ ListedExecution live =
+ listed.stream()
+ .filter(item -> runningId.equals(item.execution().getId()))
+ .findFirst()
+ .orElseThrow();
+ assertEquals(stale, live.state().getUpdateTime());
+ assertTrue(live.state().isStale(20_000L));
+ assertNull(live.execution().getFilename());
+ assertEquals(1, counting.loads);
+ } finally {
+ counting.close();
+ }
+ }
+
@Test
void upsertUpdatesFilterColumnsAndJson() throws Exception {
String id = UUID.randomUUID().toString();
@@ -782,7 +852,8 @@ class CachingDatabaseExecutionInfoLocationTest {
Execution pipe = new Execution();
pipe.setId(pipelineId);
pipe.setName("Pipeline");
- pipe.setParentId(actionId);
+ pipe.setParentId(workflowId);
+ pipe.setLaunchedById(actionId);
pipe.setExecutionType(ExecutionType.Pipeline);
pipe.setRegistrationDate(new Date());
@@ -790,20 +861,30 @@ class CachingDatabaseExecutionInfoLocationTest {
location.registerExecution(action);
location.registerExecution(pipe);
+ Execution cachedAction = location.getExecution(actionId);
+ assertEquals(List.of(pipelineId), cachedAction.getChildExecutionIds());
+ cachedAction.getChildExecutionIds().clear();
+ location.persistCacheEntry(location.getCache().get(workflowId));
+
// Persist to database and clear in-memory cache so nothing is in memory
for (CacheEntry entry : location.getCache().values()) {
location.persistCacheEntry(entry);
}
location.clearCaches();
- assertEquals(actionId, location.findParentId(pipelineId));
- assertEquals(workflowId, location.findParentId(actionId));
- assertNull(location.findParentId(workflowId));
+ assertEquals(workflowId, location.findParentId(pipelineId));
+ // The action lives in the workflow row. It is not a row of its own, and
loading the pipeline
+ // does not load it.
+ assertNull(location.findParentId(actionId));
+ assertNull(location.getExecution(actionId));
+ assertNull(location.findParentId(workflowId));
+ assertNotNull(location.getExecution(workflowId));
Execution retrievedAction = location.getExecution(actionId);
assertNotNull(retrievedAction);
assertEquals(actionId, retrievedAction.getId());
assertEquals(ExecutionType.Action, retrievedAction.getExecutionType());
+ assertEquals(List.of(pipelineId),
retrievedAction.getChildExecutionIds());
assertNull(location.getExecution("non-existent-id"));
diff --git
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/NeoExecutionInfoLocation.java
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/NeoExecutionInfoLocation.java
index feebd25a0a..5b87e0d4d9 100644
---
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/NeoExecutionInfoLocation.java
+++
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/NeoExecutionInfoLocation.java
@@ -114,6 +114,7 @@ public class NeoExecutionInfoLocation implements
IExecutionInfoLocation {
public static final String EP_COPY_NR = "copyNr";
public static final String EP_FILENAME = "filename";
public static final String EP_PARENT_ID = "parentId";
+ public static final String EP_LAUNCHED_BY_ID = "launchedById";
public static final String EP_EXECUTION_TYPE = "executionType";
public static final String EP_EXECUTOR_XML = "executorXml";
public static final String EP_METADATA_JSON = "metadataJson";
@@ -293,6 +294,7 @@ public class NeoExecutionInfoLocation implements
IExecutionInfoLocation {
addIndex(cypher, "idx_execution_start_date", EL_EXECUTION,
EP_EXECUTION_START_DATE);
addIndex(cypher, "idx_execution_failed", EL_EXECUTION, EP_FAILED);
addIndex(cypher, "idx_execution_parent_id", EL_EXECUTION, EP_PARENT_ID);
+ addIndex(cypher, "idx_execution_launched_by_id", EL_EXECUTION,
EP_LAUNCHED_BY_ID);
addIndex(
cypher,
"idx_execution_metric_id",
@@ -382,6 +384,7 @@ public class NeoExecutionInfoLocation implements
IExecutionInfoLocation {
addIndex(neo4jIndex, "idx_execution_start_date", EL_EXECUTION,
EP_EXECUTION_START_DATE);
addIndex(neo4jIndex, "idx_execution_failed", EL_EXECUTION, EP_FAILED);
addIndex(neo4jIndex, "idx_execution_parent_id", EL_EXECUTION,
EP_PARENT_ID);
+ addIndex(neo4jIndex, "idx_execution_launched_by_id", EL_EXECUTION,
EP_LAUNCHED_BY_ID);
addIndex(
neo4jIndex,
"idx_execution_metric_id",
@@ -485,6 +488,7 @@ public class NeoExecutionInfoLocation implements
IExecutionInfoLocation {
.withValue(EP_COPY_NR, execution.getCopyNr())
.withValue(EP_FILENAME, execution.getFilename())
.withValue(EP_PARENT_ID, execution.getParentId())
+ .withValue(EP_LAUNCHED_BY_ID, execution.getLaunchedById())
.withValue(EP_EXECUTION_TYPE,
execution.getExecutionType().name())
.withValue(EP_EXECUTOR_XML, execution.getExecutorXml())
.withValue(EP_METADATA_JSON, execution.getMetadataJson())
@@ -497,12 +501,14 @@ public class NeoExecutionInfoLocation implements
IExecutionInfoLocation {
execute(transaction, builder);
if (StringUtils.isNotEmpty(execution.getParentId())) {
- CypherRelationshipBuilder selfRelationshipBuilder =
- CypherRelationshipBuilder.of()
- .withMatch(EL_EXECUTION, "e1", EP_ID, execution.getId())
- .withMatch(EL_EXECUTION, "e2", EP_ID, execution.getParentId())
- .withMerge("e2", "e1", R_EXECUTES);
- execute(transaction, selfRelationshipBuilder);
+ linkExecutes(transaction, execution.getParentId(), execution.getId());
+ }
+ // The action or transform that started this execution. Link it when
that node is already
+ // stored. A missing launcher must not roll back the execution itself.
+ if (StringUtils.isNotEmpty(execution.getLaunchedById())
+ && !execution.getLaunchedById().equals(execution.getParentId())
+ && executionNodeExists(transaction, execution.getLaunchedById())) {
+ linkExecutes(transaction, execution.getLaunchedById(),
execution.getId());
}
// See if we need to cache this
@@ -517,6 +523,42 @@ public class NeoExecutionInfoLocation implements
IExecutionInfoLocation {
}
}
+ /** {@code (from)-[:EXECUTES]->(to)} when both nodes are already stored. */
+ private void linkExecutes(IGraphTransaction transaction, String fromId,
String toId) {
+ execute(
+ transaction,
+ CypherRelationshipBuilder.of()
+ .withMatch(EL_EXECUTION, "e2", EP_ID, fromId)
+ .withMatch(EL_EXECUTION, "e1", EP_ID, toId)
+ .withMerge("e2", "e1", R_EXECUTES));
+ }
+
+ private boolean executionNodeExists(IGraphTransaction transaction, String
executionId) {
+ List<Map<String, Object>> found =
+ execute(
+ transaction,
+ CypherQueryBuilder.of()
+ .withLabelAndKey("n", EL_EXECUTION, EP_ID, executionId)
+ .withReturnValues("n", EP_ID));
+ return !found.isEmpty();
+ }
+
+ /**
+ * Executions whose owner or launcher is {@code executionId}. {@code
parentId} is the workflow or
+ * pipeline. {@code launchedById} is the action or transform that started
the execution.
+ */
+ static CypherQueryBuilder executionsWithParentOrLauncher(String executionId)
{
+ CypherQueryBuilder builder =
CypherQueryBuilder.of().withLabelWithoutKey("n", EL_EXECUTION);
+ builder.withExtraClause(
+ "WHERE (n."
+ + EP_PARENT_ID
+ + " = $executionId OR n."
+ + EP_LAUNCHED_BY_ID
+ + " = $executionId) ");
+ builder.parameters().put("executionId", executionId);
+ return builder;
+ }
+
/**
* Nodes removed per Neo4j transaction so one large sample does not exhaust
transaction memory.
*/
@@ -652,9 +694,7 @@ public class NeoExecutionInfoLocation implements
IExecutionInfoLocation {
List<Map<String, Object>> result =
execute(
transaction,
- CypherQueryBuilder.of()
- .withLabelAndKey("n", EL_EXECUTION, EP_PARENT_ID,
parentExecutionId)
- .withReturnValues("n", EP_ID));
+
executionsWithParentOrLauncher(parentExecutionId).withReturnValues("n", EP_ID));
for (Map<String, Object> record : result) {
ids.add(getString(record, EP_ID));
}
@@ -719,6 +759,7 @@ public class NeoExecutionInfoLocation implements
IExecutionInfoLocation {
EP_COPY_NR,
EP_FILENAME,
EP_PARENT_ID,
+ EP_LAUNCHED_BY_ID,
EP_EXECUTION_TYPE,
EP_EXECUTOR_XML,
EP_METADATA_JSON,
@@ -747,6 +788,7 @@ public class NeoExecutionInfoLocation implements
IExecutionInfoLocation {
return ExecutionBuilder.of()
.withId(executionId)
.withParentId(getString(record, EP_PARENT_ID))
+ .withLaunchedById(getString(record, EP_LAUNCHED_BY_ID))
.withName(getString(record, EP_NAME))
.withCopyNr(getString(record, EP_COPY_NR))
.withFilename(getString(record, EP_FILENAME))
@@ -866,6 +908,7 @@ public class NeoExecutionInfoLocation implements
IExecutionInfoLocation {
EP_COPY_NR,
EP_FILENAME,
EP_PARENT_ID,
+ EP_LAUNCHED_BY_ID,
EP_EXECUTION_TYPE,
EP_EXECUTOR_XML,
EP_METADATA_JSON,
@@ -1181,10 +1224,10 @@ public class NeoExecutionInfoLocation implements
IExecutionInfoLocation {
}
/**
- * Find those Executions that have a matching parentId
+ * Executions owned by this id, or started by the action or transform with
this id.
*
* @param transaction The read transaction to use
- * @param parentExecutionId The parent execution ID
+ * @param parentExecutionId The owner or launcher execution ID
* @return The list of executions or an empty list if nothing was found
*/
private List<Execution> findNeo4jExecutions(
@@ -1194,9 +1237,7 @@ public class NeoExecutionInfoLocation implements
IExecutionInfoLocation {
List<Map<String, Object>> result =
execute(
transaction,
- CypherQueryBuilder.of()
- .withLabelAndKey("n", EL_EXECUTION, EP_PARENT_ID,
parentExecutionId)
- .withReturnValues("n", EP_ID));
+
executionsWithParentOrLauncher(parentExecutionId).withReturnValues("n", EP_ID));
for (Map<String, Object> record : result) {
String executionId = getString(record, EP_ID);
diff --git
a/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/execution/NeoExecutionInfoLocationDeleteTest.java
b/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/execution/NeoExecutionInfoLocationDeleteTest.java
index 0a62ee2224..2be8c7f0db 100644
---
a/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/execution/NeoExecutionInfoLocationDeleteTest.java
+++
b/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/execution/NeoExecutionInfoLocationDeleteTest.java
@@ -17,10 +17,12 @@
package org.apache.hop.neo4j.execution;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
import java.util.Date;
+import org.apache.hop.neo4j.execution.builder.CypherQueryBuilder;
import org.junit.jupiter.api.Test;
class NeoExecutionInfoLocationDeleteTest {
@@ -46,6 +48,16 @@ class NeoExecutionInfoLocationDeleteTest {
assertTrue(cypher.contains("LIMIT 200"));
}
+ @Test
+ void childLookupMatchesTheOwnerAndTheLauncher() {
+ CypherQueryBuilder query =
NeoExecutionInfoLocation.executionsWithParentOrLauncher("action-1");
+
+ assertTrue(query.cypher().contains("parentId = $executionId"));
+ assertTrue(query.cypher().contains("launchedById = $executionId"));
+ assertTrue(query.cypher().contains(" OR "));
+ assertEquals("action-1", query.parameters().get("executionId"));
+ }
+
@Test
void batchDeleteLimitsEachTransaction() {
String cypher =
diff --git
a/plugins/transforms/execinfo/src/main/java/org/apache/hop/pipeline/transforms/execinfo/ExecInfo.java
b/plugins/transforms/execinfo/src/main/java/org/apache/hop/pipeline/transforms/execinfo/ExecInfo.java
index 6e37286ddc..86218c39f3 100644
---
a/plugins/transforms/execinfo/src/main/java/org/apache/hop/pipeline/transforms/execinfo/ExecInfo.java
+++
b/plugins/transforms/execinfo/src/main/java/org/apache/hop/pipeline/transforms/execinfo/ExecInfo.java
@@ -284,8 +284,12 @@ public class ExecInfo extends BaseTransform<ExecInfoMeta,
ExecInfoData> {
outputRow[index++] = executionState.getLoggingText();
outputRow[index++] = executionState.isFailed();
outputRow[index++] = executionState.getStatusDescription();
- outputRow[index] = executionState.getExecutionEndDate();
+ outputRow[index++] = executionState.getExecutionEndDate();
+ } else {
+ // updateTime, loggingText, failed, statusDescription, executionEndDate
+ index += 5;
}
+ outputRow[index] = execution.getLaunchedById();
putRow(data.outputRowMeta, outputRow);
}
diff --git
a/plugins/transforms/execinfo/src/main/java/org/apache/hop/pipeline/transforms/execinfo/ExecInfoMeta.java
b/plugins/transforms/execinfo/src/main/java/org/apache/hop/pipeline/transforms/execinfo/ExecInfoMeta.java
index 0b1ef09090..235844f0f6 100644
---
a/plugins/transforms/execinfo/src/main/java/org/apache/hop/pipeline/transforms/execinfo/ExecInfoMeta.java
+++
b/plugins/transforms/execinfo/src/main/java/org/apache/hop/pipeline/transforms/execinfo/ExecInfoMeta.java
@@ -111,6 +111,7 @@ public class ExecInfoMeta extends
BaseTransformMeta<ExecInfo, ExecInfoData> {
.addBoolean("failed")
.addString("statusDescription")
.addDate("executionEndDate")
+ .addString("launchedById")
.build());
return;
case DeleteExecution:
diff --git
a/ui/src/main/java/org/apache/hop/ui/hopgui/perspective/execution/ExecutionPerspective.java
b/ui/src/main/java/org/apache/hop/ui/hopgui/perspective/execution/ExecutionPerspective.java
index cd89bafe50..204d38540e 100644
---
a/ui/src/main/java/org/apache/hop/ui/hopgui/perspective/execution/ExecutionPerspective.java
+++
b/ui/src/main/java/org/apache/hop/ui/hopgui/perspective/execution/ExecutionPerspective.java
@@ -54,6 +54,7 @@ import org.apache.hop.execution.ExecutionType;
import org.apache.hop.execution.IExecutionInfoLocation;
import org.apache.hop.execution.IExecutionSelector;
import org.apache.hop.execution.LastPeriod;
+import org.apache.hop.execution.ListedExecution;
import org.apache.hop.execution.caching.CachingFileExecutionInfoLocation;
import org.apache.hop.execution.local.FileExecutionInfoLocation;
import org.apache.hop.history.AuditList;
@@ -647,9 +648,14 @@ public class ExecutionPerspective implements
IHopPerspective, TabClosable {
} else if (treeItem.getData() instanceof Execution execution) {
ExecutionInfoLocation location =
(ExecutionInfoLocation) treeItem.getParentItem().getData();
- ExecutionState executionState =
-
location.getExecutionInfoLocation().getExecutionState(execution.getId());
- createExecutionViewer(location.getName(), execution, executionState);
+ // The tree row has no pipeline XML. Load the stored document to
open the viewer.
+ IExecutionInfoLocation iLocation =
location.getExecutionInfoLocation();
+ Execution full = iLocation.getExecution(execution.getId());
+ if (full == null) {
+ full = execution;
+ }
+ ExecutionState executionState =
iLocation.getExecutionState(execution.getId());
+ createExecutionViewer(location.getName(), full, executionState);
} else if (treeItem.getData(CONST_ERROR) instanceof Exception
exception) {
new ErrorDialog(getShell(), CONST_ERROR1, "Location error:",
exception);
}
@@ -905,7 +911,9 @@ public class ExecutionPerspective implements
IHopPerspective, TabClosable {
Cursor busyCursor = getBusyCursor();
try {
- getShell().setCursor(busyCursor);
+ // On GTK the tree and toolbar windows do not use the shell cursor. Show
the wait cursor
+ // before the list query blocks this UI thread.
+ showBusyCursor(busyCursor);
// If there are any cached locations we want to close them before
initializing new ones
//
@@ -981,37 +989,29 @@ public class ExecutionPerspective implements
IHopPerspective, TabClosable {
onlyShowingPipelines,
filterText,
timeFilter);
- List<String> ids = iLocation.findExecutionIDs(executionSelector);
-
- // Display the executions
+ // Name, dates, and status only. The stored document is loaded
when the row is opened.
//
- for (String id : ids) {
+ List<ListedExecution> listed =
iLocation.listExecutions(executionSelector);
+
+ for (ListedExecution listedExecution : listed) {
+ Execution execution = listedExecution.execution();
+ String id = execution == null ? "" :
Const.NVL(execution.getId(), "");
try {
- Execution execution = iLocation.getExecution(id);
- if (execution != null) {
- // Apply an extra filter to make sure
- //
- if (!executionSelector.isSelected(execution)) {
- continue;
- }
- // We only need to consider the state after the previous
filtering
- //
- ExecutionState state = iLocation.getExecutionState(id);
- if (!executionSelector.isSelected(state)) {
- continue;
- }
-
- TreeItem executionItem = new TreeItem(locationItem,
SWT.NONE);
- switch (execution.getExecutionType()) {
- case Pipeline:
- decoratePipelineTreeItem(executionItem, location,
execution, state);
- break;
- case Workflow:
- decorateWorkflowTreeItem(executionItem, location,
execution, state);
- break;
- default:
- break;
- }
+ if (execution == null || execution.getExecutionType() ==
null) {
+ continue;
+ }
+ ExecutionState state = listedExecution.state();
+ TreeItem executionItem = new TreeItem(locationItem,
SWT.NONE);
+ switch (execution.getExecutionType()) {
+ case Pipeline:
+ decoratePipelineTreeItem(executionItem, location,
execution, state);
+ break;
+ case Workflow:
+ decorateWorkflowTreeItem(executionItem, location,
execution, state);
+ break;
+ default:
+ executionItem.dispose();
+ break;
}
} catch (Exception e) {
TreeItem errorItem = new TreeItem(locationItem, SWT.NONE);
@@ -1051,14 +1051,32 @@ public class ExecutionPerspective implements
IHopPerspective, TabClosable {
tree.setRedraw(true);
updateSelection();
} catch (Exception e) {
- getShell().setCursor(null);
+ showBusyCursor(null);
new ErrorDialog(
getShell(),
BaseMessages.getString(PKG,
"ExecutionPerspective.Refresh.Error.Header"),
BaseMessages.getString(PKG,
"ExecutionPerspective.Refresh.Error.Message"),
e);
} finally {
- getShell().setCursor(null);
+ showBusyCursor(null);
+ busyCursor.dispose();
+ }
+ }
+
+ /** Wait cursor on the shell, the tree, and the toolbar. {@code null}
restores the default. */
+ private void showBusyCursor(Cursor cursor) {
+ Shell shell = getShell();
+ if (shell != null && !shell.isDisposed()) {
+ shell.setCursor(cursor);
+ }
+ if (tree != null && !tree.isDisposed()) {
+ tree.setCursor(cursor);
+ }
+ if (toolBar != null && !toolBar.isDisposed()) {
+ toolBar.setCursor(cursor);
+ }
+ if (shell != null && !shell.isDisposed()) {
+ shell.getDisplay().update();
}
}
@@ -1350,10 +1368,12 @@ public class ExecutionPerspective implements
IHopPerspective, TabClosable {
toolTip = "i18n::ExecutionPerspective.ToolbarElement.Copy.Tooltip",
image = "ui/images/copy.svg")
public void copyToClipboard() {
- Execution execution = getSelectedExecution();
- if (execution == null) {
+ Execution selected = getSelectedExecution();
+ if (selected == null) {
return;
}
+ // The tree row is a summary and has no filename. The stored execution
does.
+ Execution execution = executionForCopy(selected);
String filename = execution.getFilename();
String relativeFilename = toProjectRelativePath(filename);
@@ -1394,6 +1414,38 @@ public class ExecutionPerspective implements
IHopPerspective, TabClosable {
menu.setVisible(true);
}
+ /**
+ * Load the stored execution when the selected row has no filename. A
summary row from the
+ * database location omits it. A failure keeps the summary, so the id can
still be copied.
+ */
+ private Execution executionForCopy(Execution selected) {
+ if (StringUtils.isNotEmpty(selected.getFilename()) ||
StringUtils.isEmpty(selected.getId())) {
+ return selected;
+ }
+ ExecutionInfoLocation location = getSelectedExecutionLocation();
+ if (location == null) {
+ return selected;
+ }
+ try {
+ IExecutionInfoLocation iLocation = location.getExecutionInfoLocation();
+ if (iLocation == null) {
+ return selected;
+ }
+ Execution stored = iLocation.getExecution(selected.getId());
+ if (stored != null) {
+ return stored;
+ }
+ } catch (Exception e) {
+ if (hopGui != null && hopGui.getLog() != null) {
+ hopGui
+ .getLog()
+ .logError(
+ "Unable to load execution '" + selected.getId() + "' to copy
its filename", e);
+ }
+ }
+ return selected;
+ }
+
private void addCopyMenuItem(Menu menu, String label, String value, boolean
enabled) {
MenuItem item = new MenuItem(menu, SWT.PUSH);
item.setText(label);
diff --git
a/ui/src/main/java/org/apache/hop/ui/hopgui/perspective/execution/PipelineExecutionViewer.java
b/ui/src/main/java/org/apache/hop/ui/hopgui/perspective/execution/PipelineExecutionViewer.java
index 1480226183..0b6ae7c756 100644
---
a/ui/src/main/java/org/apache/hop/ui/hopgui/perspective/execution/PipelineExecutionViewer.java
+++
b/ui/src/main/java/org/apache/hop/ui/hopgui/perspective/execution/PipelineExecutionViewer.java
@@ -1121,7 +1121,7 @@ public class PipelineExecutionViewer extends
BaseExecutionViewer
for (ExecutionDataSetMeta setMeta :
selectedTransformData.getSetMetaData().values()) {
if (setMeta.getName().equals(selectedTransform.getName())) {
String parentId = setMeta.getLogChannelId();
- List<Execution> childExecutions = iLocation.findExecutions(parentId);
+ List<Execution> childExecutions = launchedExecutions(iLocation,
parentId);
if (childExecutions.isEmpty()) {
break;
}
@@ -1169,7 +1169,7 @@ public class PipelineExecutionViewer extends
BaseExecutionViewer
// Now we have the transform execution that we want we can look for the
execution(s) with this
// as parent
//
- List<Execution> childExecutions =
iLocation.findExecutions(transformExecution.getId());
+ List<Execution> childExecutions = launchedExecutions(iLocation,
transformExecution.getId());
if (childExecutions.isEmpty()) {
return;
}
@@ -1230,34 +1230,23 @@ public class PipelineExecutionViewer extends
BaseExecutionViewer
image = "ui/images/up.svg")
public void goUp() {
try {
- String parentId = execution.getParentId();
- if (parentId == null) {
- return;
- }
-
ExecutionInfoLocation location =
perspective.getLocationMap().get(locationName);
if (location == null) {
return;
}
IExecutionInfoLocation iLocation = location.getExecutionInfoLocation();
-
- // This parent ID is the ID of the transform or action.
- // We need to find the parent of this child
- //
- String grandParentId = iLocation.findParentId(parentId);
- if (grandParentId == null) {
+ Execution parent = parentExecutor(iLocation);
+ if (parent == null) {
return;
}
- Execution grandParent = iLocation.getExecution(grandParentId);
-
// Don't load execution logging text to prevent memory issues.
//
- ExecutionState executionState =
iLocation.getExecutionState(grandParent.getId(), false);
+ ExecutionState executionState =
iLocation.getExecutionState(parent.getId(), false);
// Open this one
//
- perspective.createExecutionViewer(locationName, grandParent,
executionState);
+ perspective.createExecutionViewer(locationName, parent, executionState);
} catch (Exception e) {
new ErrorDialog(getShell(), CONST_ERROR, "Error navigating up to parent
execution", e);
}
diff --git
a/ui/src/main/java/org/apache/hop/ui/hopgui/perspective/execution/WorkflowExecutionViewer.java
b/ui/src/main/java/org/apache/hop/ui/hopgui/perspective/execution/WorkflowExecutionViewer.java
index f5f0a48885..d6697d68ba 100644
---
a/ui/src/main/java/org/apache/hop/ui/hopgui/perspective/execution/WorkflowExecutionViewer.java
+++
b/ui/src/main/java/org/apache/hop/ui/hopgui/perspective/execution/WorkflowExecutionViewer.java
@@ -983,7 +983,7 @@ public class WorkflowExecutionViewer extends
BaseExecutionViewer
return;
}
- List<Execution> childExecutions = iLocation.findExecutions(id);
+ List<Execution> childExecutions = launchedExecutions(iLocation, id);
if (childExecutions.isEmpty()) {
return;
}
@@ -1039,31 +1039,20 @@ public class WorkflowExecutionViewer extends
BaseExecutionViewer
image = "ui/images/up.svg")
public void goUp() {
try {
- String parentId = execution.getParentId();
- if (parentId == null) {
- return;
- }
-
ExecutionInfoLocation location =
perspective.getLocationMap().get(locationName);
if (location == null) {
return;
}
IExecutionInfoLocation iLocation = location.getExecutionInfoLocation();
-
- // This parent ID is the ID of the transform or action.
- // We need to find the parent of this child
- //
- String grandParentId = iLocation.findParentId(parentId);
- if (grandParentId == null) {
+ Execution parent = parentExecutor(iLocation);
+ if (parent == null) {
return;
}
- Execution grandParent = iLocation.getExecution(grandParentId);
-
// Open this execution with state. Don't load the logging text.
//
- ExecutionState state = iLocation.getExecutionState(grandParent.getId(),
false);
- perspective.createExecutionViewer(locationName, grandParent, state);
+ ExecutionState state = iLocation.getExecutionState(parent.getId(),
false);
+ perspective.createExecutionViewer(locationName, parent, state);
} catch (Exception e) {
new ErrorDialog(getShell(), CONST_ERROR, "Error navigating up to parent
execution", e);
}
diff --git
a/ui/src/main/java/org/apache/hop/ui/hopgui/shared/BaseExecutionViewer.java
b/ui/src/main/java/org/apache/hop/ui/hopgui/shared/BaseExecutionViewer.java
index 3c5efefe69..3e62ef0f3d 100644
--- a/ui/src/main/java/org/apache/hop/ui/hopgui/shared/BaseExecutionViewer.java
+++ b/ui/src/main/java/org/apache/hop/ui/hopgui/shared/BaseExecutionViewer.java
@@ -28,6 +28,7 @@ import java.util.Map;
import lombok.Getter;
import lombok.Setter;
import org.apache.hop.core.Const;
+import org.apache.hop.core.exception.HopException;
import org.apache.hop.core.gui.AreaOwner;
import org.apache.hop.core.gui.DPoint;
import org.apache.hop.core.gui.Point;
@@ -38,6 +39,7 @@ import org.apache.hop.core.variables.Variables;
import org.apache.hop.execution.Execution;
import org.apache.hop.execution.ExecutionInfoLocation;
import org.apache.hop.execution.ExecutionState;
+import org.apache.hop.execution.ExecutionType;
import org.apache.hop.execution.IExecutionInfoLocation;
import org.apache.hop.metadata.api.IHopMetadata;
import org.apache.hop.metadata.api.IHopMetadataSerializer;
@@ -135,6 +137,54 @@ public abstract class BaseExecutionViewer extends
DragViewZoomBase
return null;
}
+ /**
+ * The pipeline or workflow that owns this execution. A document written
before that link was
+ * stored points {@code parentId} at the action or transform, so open that
executor's parent.
+ */
+ protected Execution parentExecutor(IExecutionInfoLocation location) throws
HopException {
+ if (location == null || execution == null ||
Utils.isEmpty(execution.getParentId())) {
+ return null;
+ }
+ Execution parent = location.getExecution(execution.getParentId());
+ if (parent == null) {
+ return null;
+ }
+ if (parent.getExecutionType() == ExecutionType.Action
+ || parent.getExecutionType() == ExecutionType.Transform) {
+ if (Utils.isEmpty(parent.getParentId())) {
+ return null;
+ }
+ return location.getExecution(parent.getParentId());
+ }
+ return parent;
+ }
+
+ /**
+ * Pipeline or workflow executions started by an action or transform. Uses
the ids stored on that
+ * launcher, then executions whose parent or launcher id matches for
documents that predate the
+ * stored link.
+ */
+ protected List<Execution> launchedExecutions(IExecutionInfoLocation
location, String launcherId)
+ throws HopException {
+ if (location == null || Utils.isEmpty(launcherId)) {
+ return List.of();
+ }
+ Execution launcher = location.getExecution(launcherId);
+ if (launcher != null && !Utils.isEmpty(launcher.getChildExecutionIds())) {
+ List<Execution> children = new ArrayList<>();
+ for (String childId : launcher.getChildExecutionIds()) {
+ Execution child = location.getExecution(childId);
+ if (child != null) {
+ children.add(child);
+ }
+ }
+ if (!children.isEmpty()) {
+ return children;
+ }
+ }
+ return location.findExecutions(launcherId);
+ }
+
protected String formatDate(Date date) {
if (date == null) {
return "";
