This is an automated email from the ASF dual-hosted git repository.
bamaer 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 4d0782bc92 Issue #8280 : Harden Neo4j execution information for Neo4j
5 (#8282)
4d0782bc92 is described below
commit 4d0782bc92487ca516b8273e64ee21ab35b7835b
Author: Matt Casters <[email protected]>
AuthorDate: Tue Sep 8 16:39:55 2026 +0200
Issue #8280 : Harden Neo4j execution information for Neo4j 5 (#8282)
* Issue #8280 : Harden Neo4j execution information for Neo4j 5
Log the real exception when sample-row registration fails, keep state
updates independent of data writes, coerce Neo4j-unsafe property types,
and fix lineage/error Cypher for Neo4j 5. Also paint nested pipeline
errors from stored metrics and fail index/constraint actions on Cypher
errors. See discussion #8204.
* Issue #8280 : Index ExecutionMetric and close Neo4j drivers on error
Point idx_execution_metric_id at :ExecutionMetric including metricKey.
Stop writeHierarchies from throwing out of finished listeners, and close
the logging session/driver in a finally block.
---
.../engines/local/LocalPipelineEngine.java | 34 ++---
.../engines/local/LocalWorkflowEngine.java | 8 +-
.../hop/spark/engines/SparkPipelineEngine.java | 6 +-
.../neo4j/actions/constraint/Neo4jConstraint.java | 9 +-
.../apache/hop/neo4j/actions/index/Neo4jIndex.java | 32 ++---
.../hop/neo4j/core/data/GraphPropertyDataType.java | 2 +-
.../neo4j/execution/NeoExecutionInfoLocation.java | 125 ++++++++++++-----
.../neo4j/execution/builder/BaseCypherBuilder.java | 151 +++++++++++++++++----
.../execution/builder/CypherCreateBuilder.java | 6 +-
.../execution/builder/CypherMatchBuilder.java | 2 +-
.../execution/builder/CypherMergeBuilder.java | 6 +-
.../execution/builder/CypherQueryBuilder.java | 6 +-
.../neo4j/execution/cache/NeoLocationCache.java | 90 +++++-------
.../path/base/NeoExecutionViewerErrorTab.java | 56 ++++----
.../path/base/NeoExecutionViewerLineageTab.java | 56 ++++----
.../path/base/NeoExecutionViewerTabBase.java | 51 ++++---
.../apache/hop/neo4j/logging/util/LoggingCore.java | 1 -
.../logging/xp/PipelineLoggingExtensionPoint.java | 38 +++---
.../logging/xp/WorkflowLoggingExtensionPoint.java | 36 ++---
.../apache/hop/neo4j/shared/DriverSingleton.java | 31 ++---
.../org/apache/hop/neo4j/shared/NeoConnection.java | 11 ++
.../hop/neo4j/transforms/importer/Importer.java | 5 +-
.../neo4j/transforms/importer/ImporterDialog.java | 6 +-
.../neo4j/actions/index/Neo4jIndexCypherTest.java | 55 ++++++++
.../neo4j/core/data/GraphPropertyDataTypeTest.java | 30 ++++
.../execution/builder/BaseCypherBuilderTest.java | 76 +++++++++++
.../execution/cache/NeoLocationCacheTest.java | 64 +++++++++
.../path/NeoExecutionViewerTabBaseTest.java | 56 ++++++++
.../hop/neo4j/perspective/ErrorPathCypherIT.java | 55 ++++++++
.../execution/PipelineExecutionViewer.java | 27 +++-
30 files changed, 822 insertions(+), 309 deletions(-)
diff --git
a/engine/src/main/java/org/apache/hop/pipeline/engines/local/LocalPipelineEngine.java
b/engine/src/main/java/org/apache/hop/pipeline/engines/local/LocalPipelineEngine.java
index fc24bcb602..8359f8c884 100644
---
a/engine/src/main/java/org/apache/hop/pipeline/engines/local/LocalPipelineEngine.java
+++
b/engine/src/main/java/org/apache/hop/pipeline/engines/local/LocalPipelineEngine.java
@@ -486,27 +486,29 @@ public class LocalPipelineEngine extends Pipeline
implements IPipelineEngine<Pip
new TimerTask() {
@Override
public void run() {
- try {
- // Collect data from all the sampler stores.
- //
- if (dataProfile != null) {
+ // Sample rows and execution state are written independently so a
conversion error
+ // on sampled data cannot skip the state update (and hide the real
exception).
+ //
+ if (dataProfile != null) {
+ try {
ExecutionDataBuilder dataBuilder =
ExecutionDataBuilder.fromAllTransformData(
LocalPipelineEngine.this, samplerStoresMap, false);
-
- // Send it to the location once
- //
iLocation.registerData(dataBuilder.build());
+ } catch (Exception e) {
+ log.logError(
+ "Warning: unable to register execution data at location "
+ + executionInfoLocation.getName()
+ + " (non-fatal)",
+ e);
}
+ }
- // Update the pipeline execution state regularly
- //
+ try {
ExecutionState pipelineState =
ExecutionStateBuilder.fromExecutor(LocalPipelineEngine.this,
-1).build();
iLocation.updateExecutionState(pipelineState);
- // Update the state of all the transforms
- //
for (IEngineComponent component : getComponents()) {
ExecutionState transformState =
ExecutionStateBuilder.fromTransform(LocalPipelineEngine.this, component)
@@ -514,13 +516,11 @@ public class LocalPipelineEngine extends Pipeline
implements IPipelineEngine<Pip
iLocation.updateExecutionState(transformState);
}
} catch (Exception e) {
- // This is probably cause by a race condition triggering this
code after the pipeline
- // finished. We're just going to log this as a warning.
- //
- log.logBasic(
- "Warning: unable to register execution info (data and state)
at location "
+ log.logError(
+ "Warning: unable to register execution state at location "
+ executionInfoLocation.getName()
- + "(non-fatal)");
+ + " (non-fatal)",
+ e);
}
}
};
diff --git
a/engine/src/main/java/org/apache/hop/workflow/engines/local/LocalWorkflowEngine.java
b/engine/src/main/java/org/apache/hop/workflow/engines/local/LocalWorkflowEngine.java
index 916ebd609b..632437e763 100644
---
a/engine/src/main/java/org/apache/hop/workflow/engines/local/LocalWorkflowEngine.java
+++
b/engine/src/main/java/org/apache/hop/workflow/engines/local/LocalWorkflowEngine.java
@@ -31,7 +31,6 @@ import org.apache.hop.core.database.Database;
import org.apache.hop.core.database.map.DatabaseConnectionMap;
import org.apache.hop.core.exception.HopDatabaseException;
import org.apache.hop.core.exception.HopException;
-import org.apache.hop.core.exception.HopRuntimeException;
import org.apache.hop.core.logging.ILogChannel;
import org.apache.hop.core.logging.ILoggingObject;
import org.apache.hop.core.util.ExecutorUtil;
@@ -401,9 +400,10 @@ public class LocalWorkflowEngine extends Workflow
implements IWorkflowEngine<Wor
lastLogLineNr.set(executionState.getLastLogLineNr());
}
} catch (Exception e) {
- throw new HopRuntimeException(
- "Error registering execution info data from transforms at
location "
- + executionInfoLocation.getName(),
+ log.logError(
+ "Warning: unable to register execution state at location "
+ + executionInfoLocation.getName()
+ + " (non-fatal)",
e);
}
}
diff --git
a/plugins/engines/spark/src/main/java/org/apache/hop/spark/engines/SparkPipelineEngine.java
b/plugins/engines/spark/src/main/java/org/apache/hop/spark/engines/SparkPipelineEngine.java
index f654f9dc9c..8c32a8024c 100644
---
a/plugins/engines/spark/src/main/java/org/apache/hop/spark/engines/SparkPipelineEngine.java
+++
b/plugins/engines/spark/src/main/java/org/apache/hop/spark/engines/SparkPipelineEngine.java
@@ -604,11 +604,11 @@ public class SparkPipelineEngine extends Variables
implements IPipelineEngine<Pi
updatePipelineState(iLocation);
} catch (Exception e) {
if (logChannel != null) {
- logChannel.logBasic(
+ logChannel.logError(
"Warning: unable to register execution info at location "
+ executionInfoLocation.getName()
- + " (non-fatal): "
- + e.getMessage());
+ + " (non-fatal)",
+ e);
}
}
}
diff --git
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/actions/constraint/Neo4jConstraint.java
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/actions/constraint/Neo4jConstraint.java
index 6447928df8..d7ab56ae47 100644
---
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/actions/constraint/Neo4jConstraint.java
+++
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/actions/constraint/Neo4jConstraint.java
@@ -23,6 +23,7 @@ import org.apache.commons.lang3.StringUtils;
import org.apache.hop.core.Result;
import org.apache.hop.core.annotations.Action;
import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.exception.HopRuntimeException;
import org.apache.hop.metadata.api.HopMetadataProperty;
import org.apache.hop.neo4j.shared.NeoConnection;
import org.apache.hop.workflow.action.ActionBase;
@@ -138,8 +139,8 @@ public class Neo4jConstraint extends ActionBase implements
IAction {
result.consume();
return true;
} catch (Throwable e) {
- logError("Error dropping constraint with cypher [" + _cypher +
"]", e);
- return false;
+ throw new HopRuntimeException(
+ "Error dropping constraint with cypher [" + _cypher + "]",
e);
}
});
}
@@ -249,8 +250,8 @@ public class Neo4jConstraint extends ActionBase implements
IAction {
result.consume();
return true;
} catch (Throwable e) {
- logError("Error creating constraint with cypher [" + _cypher +
"]", e);
- return false;
+ throw new HopRuntimeException(
+ "Error creating constraint with cypher [" + _cypher + "]",
e);
}
});
}
diff --git
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/actions/index/Neo4jIndex.java
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/actions/index/Neo4jIndex.java
index 75cff40926..5a48e78f0a 100644
---
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/actions/index/Neo4jIndex.java
+++
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/actions/index/Neo4jIndex.java
@@ -24,6 +24,7 @@ import org.apache.hop.core.Const;
import org.apache.hop.core.Result;
import org.apache.hop.core.annotations.Action;
import org.apache.hop.core.exception.HopException;
+import org.apache.hop.core.exception.HopRuntimeException;
import org.apache.hop.metadata.api.HopMetadataProperty;
import org.apache.hop.neo4j.shared.NeoConnection;
import org.apache.hop.workflow.action.ActionBase;
@@ -109,23 +110,14 @@ public class Neo4jIndex extends ActionBase implements
IAction {
public static String generateDropIndexCypher(IndexUpdate indexUpdate) throws
HopException {
String cypher = "DROP INDEX ";
- if (StringUtils.isNotEmpty(indexUpdate.getIndexName())) {
- cypher += indexUpdate.getIndexName();
- } else {
- cypher += " FOR ";
- switch (indexUpdate.getObjectType()) {
- case NODE:
- cypher +=
- ":" + indexUpdate.getObjectName() + "(" +
indexUpdate.getObjectProperties() + ")";
- break;
- case RELATIONSHIP:
- throw new HopException(
- "Please drop indexes on relationship properties with their name.
Relationship label: "
- + indexUpdate.getObjectName()
- + ", properties: "
- + indexUpdate.getObjectProperties());
- }
+ if (StringUtils.isEmpty(indexUpdate.getIndexName())) {
+ throw new HopException(
+ "Please drop indexes with the name of the index. Object: "
+ + indexUpdate.getObjectName()
+ + ", properties: "
+ + indexUpdate.getObjectProperties());
}
+ cypher += indexUpdate.getIndexName();
cypher += " IF EXISTS";
return cypher;
}
@@ -148,8 +140,8 @@ public class Neo4jIndex extends ActionBase implements
IAction {
result.consume();
return true;
} catch (Throwable e) {
- logError("Error dropping index with cypher [" + _cypher + "]",
e);
- return false;
+ throw new HopRuntimeException(
+ "Error dropping index with cypher [" + _cypher + "]", e);
}
});
}
@@ -216,8 +208,8 @@ public class Neo4jIndex extends ActionBase implements
IAction {
result.consume();
return true;
} catch (Throwable e) {
- logError("Error creating index with cypher [" + _cypher + "]",
e);
- return false;
+ throw new HopRuntimeException(
+ "Error creating index with cypher [" + _cypher + "]", e);
}
});
}
diff --git
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/core/data/GraphPropertyDataType.java
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/core/data/GraphPropertyDataType.java
index 817625870a..92616bb6bf 100644
---
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/core/data/GraphPropertyDataType.java
+++
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/core/data/GraphPropertyDataType.java
@@ -29,7 +29,7 @@ public enum GraphPropertyDataType {
String("string"),
Integer("long"),
Float("double"),
- Number("doubler"),
+ Number("double"),
Boolean("boolean"),
Date("date"),
LocalDateTime("localdatetime"),
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 b0445b34eb..ddc6516d2a 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
@@ -232,11 +232,31 @@ public class NeoExecutionInfoLocation implements
IExecutionInfoLocation {
@Override
public void close() throws HopException {
+ Exception first = null;
try {
- session.close();
- driver.close();
+ if (session != null) {
+ session.close();
+ }
+ } catch (Exception e) {
+ first = e;
+ } finally {
+ session = null;
+ }
+ try {
+ if (driver != null) {
+ driver.close();
+ }
} catch (Exception e) {
- throw new HopException("Error closing Neo4j execution information
location", e);
+ if (first == null) {
+ first = e;
+ } else {
+ first.addSuppressed(e);
+ }
+ } finally {
+ driver = null;
+ }
+ if (first != null) {
+ throw new HopException("Error closing Neo4j execution information
location", first);
}
}
@@ -247,7 +267,7 @@ public class NeoExecutionInfoLocation implements
IExecutionInfoLocation {
@Override
public void unBuffer(String executionId) {
- // There is nothing to remove from a buffer or cache.
+ NeoLocationCache.remove(executionId);
}
/** Simply show the DDL to create the indexes */
@@ -265,7 +285,14 @@ 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_metric_id", EL_EXECUTION, EP_ID, EP_NAME,
EP_COPY_NR);
+ addIndex(
+ cypher,
+ "idx_execution_metric_id",
+ CL_EXECUTION_METRIC,
+ CP_ID,
+ CP_NAME,
+ CP_COPY_NR,
+ CP_METRIC_KEY);
addIndex(cypher, "idx_execution_data_id", DL_EXECUTION_DATA, DP_PARENT_ID,
DP_OWNER_ID);
addIndex(
cypher,
@@ -348,9 +375,17 @@ public class NeoExecutionInfoLocation implements
IExecutionInfoLocation {
}
addIndex(neo4jIndex, "idx_execution_id", EL_EXECUTION, EP_ID);
+ 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_metric_id", EL_EXECUTION, EP_ID,
EP_NAME, EP_COPY_NR);
+ addIndex(
+ neo4jIndex,
+ "idx_execution_metric_id",
+ CL_EXECUTION_METRIC,
+ CP_ID,
+ CP_NAME,
+ CP_COPY_NR,
+ CP_METRIC_KEY);
addIndex(neo4jIndex, "idx_execution_data_id", DL_EXECUTION_DATA,
DP_PARENT_ID, DP_OWNER_ID);
addIndex(
neo4jIndex,
@@ -423,9 +458,12 @@ public class NeoExecutionInfoLocation implements
IExecutionInfoLocation {
public void registerExecution(Execution execution) throws HopException {
synchronized (this) {
try {
- assert execution.getName() != null : "Please register executions with
a name";
- assert execution.getExecutionType() != null
- : "Please register executions with an execution type";
+ if (StringUtils.isEmpty(execution.getName())) {
+ throw new HopException("Please register executions with a name");
+ }
+ if (execution.getExecutionType() == null) {
+ throw new HopException("Please register executions with an execution
type");
+ }
session.executeWrite(transaction ->
registerNeo4jExecution(transaction, execution));
} catch (Exception e) {
@@ -762,9 +800,12 @@ public class NeoExecutionInfoLocation implements
IExecutionInfoLocation {
public void updateExecutionState(ExecutionState executionState) throws
HopException {
synchronized (this) {
try {
- assert executionState.getName() != null : "Please update execution
states with a name";
- assert executionState.getExecutionType() != null
- : "Please update execution states with an execution type";
+ if (StringUtils.isEmpty(executionState.getName())) {
+ throw new HopException("Please update execution states with a name");
+ }
+ if (executionState.getExecutionType() == null) {
+ throw new HopException("Please update execution states with an
execution type");
+ }
session.executeWrite(transaction ->
updateNeo4jExecutionState(transaction, executionState));
} catch (Exception e) {
@@ -798,14 +839,13 @@ public class NeoExecutionInfoLocation implements
IExecutionInfoLocation {
// Save all the metrics in this map in there...
//
for (String metricKey : metric.getMetrics().keySet()) {
- Map<String, Object> metricKeys =
- Map.of(
- CP_ID, state.getId(),
- CP_NAME, metric.getComponentName(),
- CP_COPY_NR, metric.getComponentCopy(),
- CP_METRIC_KEY, metricKey);
- CypherCreateBuilder metricBuilder =
- CypherCreateBuilder.of()
+ Map<String, Object> metricKeys = new HashMap<>();
+ metricKeys.put(CP_ID, state.getId());
+ metricKeys.put(CP_NAME, Const.NVL(metric.getComponentName(), ""));
+ metricKeys.put(CP_COPY_NR, Const.NVL(metric.getComponentCopy(),
"0"));
+ metricKeys.put(CP_METRIC_KEY, metricKey);
+ CypherMergeBuilder metricBuilder =
+ CypherMergeBuilder.of()
.withLabelAndKeys(CL_EXECUTION_METRIC, metricKeys)
.withValue(CP_METRIC_VALUE,
metric.getMetrics().get(metricKey));
execute(transaction, metricBuilder);
@@ -813,20 +853,16 @@ public class NeoExecutionInfoLocation implements
IExecutionInfoLocation {
CypherRelationshipBuilder.of()
.withMatch(EL_EXECUTION, "e", EP_ID, state.getId())
.withMatch(CL_EXECUTION_METRIC, "m", metricKeys)
- .withCreate("e", "m", R_HAS_METRIC);
+ .withMerge("e", "m", R_HAS_METRIC);
execute(transaction, relationshipBuilder);
}
}
}
- // Transaction is automatically committed by executeWrite
+ NeoLocationCache.store(state);
return true;
} catch (Exception e) {
- // Transaction is automatically rolled back by executeWrite on exception
throw new HopRuntimeException("Error updating the state of an execution
in Neo4j", e);
- } finally {
- // Update the cache
- NeoLocationCache.store(state);
}
}
@@ -1113,8 +1149,20 @@ public class NeoExecutionInfoLocation implements
IExecutionInfoLocation {
private boolean registerNeo4jData(TransactionContext transaction,
ExecutionData data) {
try {
- assert data != null : "no execution data provided";
- assert data.getExecutionType() != null : "execution data has no type";
+ if (data == null) {
+ throw new HopRuntimeException("no execution data provided");
+ }
+ if (data.getExecutionType() == null) {
+ throw new HopRuntimeException("execution data has no type");
+ }
+ if (StringUtils.isEmpty(data.getParentId()) ||
StringUtils.isEmpty(data.getOwnerId())) {
+ throw new HopRuntimeException(
+ "Execution data is missing parentId or ownerId (parentId="
+ + data.getParentId()
+ + ", ownerId="
+ + data.getOwnerId()
+ + ")");
+ }
// We'll not cache this data as it can be a lot.
@@ -1175,6 +1223,10 @@ public class NeoExecutionInfoLocation implements
IExecutionInfoLocation {
for (String setKey : data.getDataSets().keySet()) {
RowBuffer rowBuffer = data.getDataSets().get(setKey);
ExecutionDataSetMeta setMeta = data.getSetMetaData().get(setKey);
+ if (setMeta == null) {
+ log.logError("Skipping execution data set '" + setKey + "': no
metadata");
+ continue;
+ }
saveNeo4jRowsAndMeta(
transaction, data.getParentId(), data.getOwnerId(), rowBuffer,
setMeta);
}
@@ -1195,8 +1247,14 @@ public class NeoExecutionInfoLocation implements
IExecutionInfoLocation {
ExecutionDataSetMeta setMeta) {
try {
+ if (rowBuffer == null || rowBuffer.getRowMeta() == null) {
+ log.logError(
+ "Skipping execution data set without row metadata: "
+ + (setMeta == null ? "?" : setMeta.getSetKey()));
+ return;
+ }
IRowMeta rowMeta = rowBuffer.getRowMeta();
- String rowMetaJson = rowMeta == null ? null :
JsonRowMeta.toJson(rowMeta);
+ String rowMetaJson = JsonRowMeta.toJson(rowMeta);
// Save the Execution Data Set node
//
@@ -1730,12 +1788,9 @@ public class NeoExecutionInfoLocation implements
IExecutionInfoLocation {
@Override
public String findParentId(String childId) throws HopException {
try {
- for (String id : getExecutionIds(true, 100)) {
- ExecutionState executionState = getExecutionState(id);
- List<String> childIds = executionState.getChildIds();
- if (childIds != null && childIds.contains(childId)) {
- return id;
- }
+ Execution child = getExecution(childId);
+ if (child != null) {
+ return child.getParentId();
}
return null;
} catch (Exception e) {
diff --git
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/builder/BaseCypherBuilder.java
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/builder/BaseCypherBuilder.java
index 345630e7ee..6db8f1836c 100644
---
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/builder/BaseCypherBuilder.java
+++
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/builder/BaseCypherBuilder.java
@@ -20,18 +20,31 @@ package org.apache.hop.neo4j.execution.builder;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.JsonNode;
+import java.lang.reflect.Array;
import java.math.BigDecimal;
+import java.math.BigInteger;
import java.sql.Timestamp;
-import java.text.SimpleDateFormat;
+import java.time.Duration;
+import java.time.LocalDate;
import java.time.LocalDateTime;
+import java.time.LocalTime;
+import java.time.OffsetDateTime;
+import java.time.OffsetTime;
import java.time.ZoneId;
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
+import java.util.List;
import java.util.Map;
-import org.apache.hop.core.exception.HopRuntimeException;
+import java.util.UUID;
import org.apache.hop.core.json.HopJson;
public abstract class BaseCypherBuilder implements ICypherBuilder {
+ private static final DateTimeFormatter TIMESTAMP_FORMAT =
+ DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSSSSSSSS");
+
protected StringBuilder cypher;
protected Map<String, Object> parameters;
@@ -46,16 +59,7 @@ public abstract class BaseCypherBuilder implements
ICypherBuilder {
}
protected void addParameter(String property, Object value) {
- if (value != null) {
- if (value instanceof Date date) {
- // Convert to LocalDateTime
- parameters.put(property, LocalDateTime.ofInstant(date.toInstant(),
ZoneId.systemDefault()));
- } else {
- parameters.put(property, value);
- }
- } else {
- parameters.put(property, null);
- }
+ parameters.put(property, mapTypes(value));
}
public void withExtraClause(String clause) {
@@ -70,27 +74,114 @@ public abstract class BaseCypherBuilder implements
ICypherBuilder {
return parameters;
}
- public static SimpleDateFormat timestampFormat =
- new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSSSSSSSS");
-
+ /**
+ * Convert a Hop / Java value into a type the Neo4j driver accepts as a node
property.
+ *
+ * <p>Unsupported values are coerced to String or JSON so a single sampled
field cannot abort an
+ * entire execution-data transaction.
+ */
protected Object mapTypes(Object value) {
- Object result = value;
- if (value instanceof BigDecimal) {
- result = value.toString();
- }
- if (value instanceof Timestamp) {
- result = timestampFormat.format((Timestamp) value);
- }
- if (value instanceof Map) {
- try {
- result = HopJson.newMapper().writeValueAsString(value);
- } catch (JsonProcessingException e) {
- throw new HopRuntimeException("Error converting Map to a JSON String",
e);
- }
+ if (value == null) {
+ return null;
+ }
+ if (value instanceof Timestamp timestamp) {
+ return TIMESTAMP_FORMAT.format(timestamp.toLocalDateTime());
+ }
+ if (value instanceof java.sql.Date sqlDate) {
+ return sqlDate.toLocalDate();
+ }
+ if (value instanceof java.sql.Time sqlTime) {
+ return sqlTime.toLocalTime();
+ }
+ if (value instanceof Date date) {
+ return LocalDateTime.ofInstant(date.toInstant(), ZoneId.systemDefault());
+ }
+ if (value instanceof BigDecimal || value instanceof BigInteger) {
+ return value.toString();
+ }
+ if (value instanceof Float number) {
+ return number.doubleValue();
+ }
+ if (value instanceof Integer number) {
+ return number.longValue();
+ }
+ if (value instanceof Short number) {
+ return number.longValue();
+ }
+ if (value instanceof Byte number) {
+ return number.longValue();
+ }
+ if (value instanceof Character || value instanceof UUID) {
+ return value.toString();
}
if (value instanceof JsonNode node) {
- result = node.toPrettyString();
+ return node.toPrettyString();
+ }
+ if (value instanceof Map<?, ?> map) {
+ return toJsonString(map);
}
- return result;
+ if (value instanceof byte[] bytes) {
+ return bytes;
+ }
+ if (value instanceof List<?> list) {
+ return mapList(list);
+ }
+ if (value.getClass().isArray()) {
+ int length = Array.getLength(value);
+ List<Object> list = new ArrayList<>(length);
+ for (int i = 0; i < length; i++) {
+ list.add(Array.get(value, i));
+ }
+ return mapList(list);
+ }
+ if (isNeo4jPropertyValue(value)) {
+ return value;
+ }
+ return String.valueOf(value);
+ }
+
+ private Object mapList(List<?> list) {
+ List<Object> mapped = new ArrayList<>(list.size());
+ Class<?> elementType = null;
+ for (Object element : list) {
+ Object mappedElement = mapTypes(element);
+ if (mappedElement == null) {
+ mapped.add(null);
+ continue;
+ }
+ if (!isNeo4jPropertyValue(mappedElement)) {
+ return toJsonString(list);
+ }
+ if (elementType == null) {
+ elementType = mappedElement.getClass();
+ } else if (!elementType.equals(mappedElement.getClass())) {
+ return toJsonString(list);
+ }
+ mapped.add(mappedElement);
+ }
+ return mapped;
+ }
+
+ private String toJsonString(Object value) {
+ try {
+ return HopJson.newMapper().writeValueAsString(value);
+ } catch (JsonProcessingException e) {
+ return String.valueOf(value);
+ }
+ }
+
+ private boolean isNeo4jPropertyValue(Object value) {
+ return value instanceof Boolean
+ || value instanceof Long
+ || value instanceof Double
+ || value instanceof String
+ || value instanceof byte[]
+ || value instanceof LocalDate
+ || value instanceof LocalDateTime
+ || value instanceof LocalTime
+ || value instanceof OffsetTime
+ || value instanceof OffsetDateTime
+ || value instanceof ZonedDateTime
+ || value instanceof Duration;
}
}
diff --git
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/builder/CypherCreateBuilder.java
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/builder/CypherCreateBuilder.java
index 18f86492e8..de72e59644 100644
---
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/builder/CypherCreateBuilder.java
+++
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/builder/CypherCreateBuilder.java
@@ -38,7 +38,7 @@ public class CypherCreateBuilder extends BaseCypherBuilder {
.append(" : $")
.append(key)
.append("}) ");
- parameters.put(key, value);
+ addParameter(key, value);
return this;
}
@@ -53,7 +53,7 @@ public class CypherCreateBuilder extends BaseCypherBuilder {
cypher.append(", ");
}
cypher.append(key).append(" : $").append(key);
- parameters.put(key, value);
+ addParameter(key, value);
}
cypher.append(" }) ");
return this;
@@ -68,8 +68,6 @@ public class CypherCreateBuilder extends BaseCypherBuilder {
}
cypher.append("n.").append(property).append("=$").append(property).append(" ");
- value = mapTypes(value);
-
addParameter(property, value);
return this;
diff --git
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/builder/CypherMatchBuilder.java
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/builder/CypherMatchBuilder.java
index 4040300bd3..c283b5e3ea 100644
---
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/builder/CypherMatchBuilder.java
+++
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/builder/CypherMatchBuilder.java
@@ -60,7 +60,7 @@ public class CypherMatchBuilder extends BaseCypherBuilder {
cypher.append(", ");
}
cypher.append(key).append(" : $").append(param);
- parameters.put(param, value);
+ addParameter(param, value);
}
cypher.append(" }) ");
return this;
diff --git
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/builder/CypherMergeBuilder.java
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/builder/CypherMergeBuilder.java
index 2c0c36da86..fef5a0de91 100644
---
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/builder/CypherMergeBuilder.java
+++
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/builder/CypherMergeBuilder.java
@@ -38,7 +38,7 @@ public class CypherMergeBuilder extends BaseCypherBuilder {
.append(" : $")
.append(key)
.append("}) ");
- parameters.put(key, value);
+ addParameter(key, value);
return this;
}
@@ -53,7 +53,7 @@ public class CypherMergeBuilder extends BaseCypherBuilder {
cypher.append(", ");
}
cypher.append(key).append(" : $").append(key);
- parameters.put(key, value);
+ addParameter(key, value);
}
cypher.append(" }) ");
@@ -69,7 +69,7 @@ public class CypherMergeBuilder extends BaseCypherBuilder {
}
cypher.append("n.").append(property).append("=$").append(property).append(" ");
- addParameter(property, mapTypes(value));
+ addParameter(property, value);
return this;
}
diff --git
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/builder/CypherQueryBuilder.java
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/builder/CypherQueryBuilder.java
index 16c90c01fa..4f9eb56531 100644
---
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/builder/CypherQueryBuilder.java
+++
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/builder/CypherQueryBuilder.java
@@ -44,7 +44,7 @@ public class CypherQueryBuilder extends BaseCypherBuilder {
.append(" : $")
.append(key)
.append(" }) ");
- parameters.put(key, keyValue);
+ addParameter(key, keyValue);
return this;
}
@@ -62,7 +62,7 @@ public class CypherQueryBuilder extends BaseCypherBuilder {
cypher.append(", ");
}
cypher.append(key).append(" : $").append(param);
- parameters.put(param, value);
+ addParameter(param, value);
}
cypher.append(" }) ");
@@ -175,7 +175,7 @@ public class CypherQueryBuilder extends BaseCypherBuilder {
String parameter = nodeAlias + "_" + otherKey;
Object value = nodeKeys.get(otherKey);
cypher.append(otherKey).append(" : $").append(parameter);
- parameters.put(parameter, value);
+ addParameter(parameter, value);
}
cypher.append(" }) ");
return this;
diff --git
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/cache/NeoLocationCache.java
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/cache/NeoLocationCache.java
index 46201689fd..3795324f44 100644
---
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/cache/NeoLocationCache.java
+++
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/cache/NeoLocationCache.java
@@ -6,7 +6,7 @@
* (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
+ * 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,
@@ -23,7 +23,6 @@ import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.concurrent.atomic.AtomicBoolean;
import lombok.Getter;
import lombok.Setter;
import org.apache.hop.core.Const;
@@ -38,19 +37,18 @@ import org.apache.hop.execution.caching.DatedId;
public class NeoLocationCache {
private static NeoLocationCache instance;
- private Map<String, CacheEntry> cache;
-
- private final AtomicBoolean locked;
+ private final Map<String, CacheEntry> cache;
private int maximumSize;
+ private int evictionBatchSize;
private NeoLocationCache() {
- this.locked = new AtomicBoolean(false);
this.cache = new HashMap<>();
this.maximumSize = 1000;
+ this.evictionBatchSize = 50;
}
- public static NeoLocationCache getInstance() {
+ public static synchronized NeoLocationCache getInstance() {
if (instance == null) {
instance = new NeoLocationCache();
}
@@ -58,17 +56,14 @@ public class NeoLocationCache {
}
public static void add(CacheEntry entry) {
- synchronized (getInstance().locked) {
- getInstance().locked.set(true);
- getInstance().cache.put(entry.getId(), entry);
- getInstance().locked.set(false);
+ NeoLocationCache lc = getInstance();
+ synchronized (lc.cache) {
+ lc.cache.put(entry.getId(), entry);
+ manageCacheSize();
}
- manageCacheSize();
}
public static void store(Execution execution) {
- // Add a new Cache Entry
- //
CacheEntry cacheEntry = new CacheEntry();
cacheEntry.setId(execution.getId());
cacheEntry.setExecution(execution);
@@ -77,8 +72,6 @@ public class NeoLocationCache {
}
public static void store(ExecutionState executionState) {
- // Update the cache entry
- //
CacheEntry cacheEntry = get(executionState.getId());
if (cacheEntry != null) {
cacheEntry.setExecutionState(executionState);
@@ -87,8 +80,6 @@ public class NeoLocationCache {
}
public static void store(String executionId, ExecutionData executionData) {
- // Update the cache entry
- //
CacheEntry cacheEntry = get(executionId);
if (cacheEntry != null) {
cacheEntry.addExecutionData(executionData);
@@ -97,8 +88,9 @@ public class NeoLocationCache {
}
public static CacheEntry get(String id) {
- synchronized (getInstance().locked) {
- CacheEntry cacheEntry = getInstance().cache.get(id);
+ NeoLocationCache lc = getInstance();
+ synchronized (lc.cache) {
+ CacheEntry cacheEntry = lc.cache.get(id);
if (cacheEntry != null) {
cacheEntry.setLastRead(new Date());
}
@@ -123,53 +115,37 @@ public class NeoLocationCache {
}
public static void remove(String executionId) {
- synchronized (getInstance().locked) {
- getInstance().cache.remove(executionId);
+ NeoLocationCache lc = getInstance();
+ synchronized (lc.cache) {
+ lc.cache.remove(executionId);
}
}
- private static synchronized void manageCacheSize() {
+ private static void manageCacheSize() {
NeoLocationCache lc = getInstance();
Map<String, CacheEntry> c = lc.cache;
- try {
- if (lc.locked.get()) {
- // Let the buffer overrun happen for a bit
- // We'll sweep it clean on the next one.
- return;
- }
- lc.locked.set(true);
- // The maximum size is by default 1000 entries
- if (c.size() >= lc.maximumSize + 50) {
- // Remove the last 50
- //
- List<DatedId> datedIds = new ArrayList<>();
- for (CacheEntry ce : c.values()) {
- Date date = ce.getLastRead();
- if (date == null) {
- // Never read? Perhaps it's time to get rid of it.
- //
- date = Const.MIN_DATE;
- }
- datedIds.add(new DatedId(ce.getId(), date));
- }
- // reverse sort by creation date of the cache entry
- //
- datedIds.sort(Comparator.comparing(DatedId::getDate).reversed());
-
- // Now delete the first 50 records in the cache
- //
- for (DatedId datedId : datedIds) {
- c.remove(datedId.getId());
- }
+ if (c.size() < lc.maximumSize + lc.evictionBatchSize) {
+ return;
+ }
+ List<DatedId> datedIds = new ArrayList<>();
+ for (CacheEntry ce : c.values()) {
+ Date date = ce.getLastRead();
+ if (date == null) {
+ date = Const.MIN_DATE;
}
- } finally {
- instance.locked.set(false);
+ datedIds.add(new DatedId(ce.getId(), date));
+ }
+ datedIds.sort(Comparator.comparing(DatedId::getDate));
+ int toRemove = Math.min(lc.evictionBatchSize, datedIds.size());
+ for (int i = 0; i < toRemove; i++) {
+ c.remove(datedIds.get(i).getId());
}
}
public static void clear() {
- synchronized (getInstance().locked) {
- getInstance().cache.clear();
+ NeoLocationCache lc = getInstance();
+ synchronized (lc.cache) {
+ lc.cache.clear();
}
}
}
diff --git
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/path/base/NeoExecutionViewerErrorTab.java
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/path/base/NeoExecutionViewerErrorTab.java
index 02200b384e..39dddd89d7 100644
---
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/path/base/NeoExecutionViewerErrorTab.java
+++
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/path/base/NeoExecutionViewerErrorTab.java
@@ -30,6 +30,7 @@ import org.apache.hop.i18n.BaseMessages;
import org.apache.hop.neo4j.execution.path.PathResult;
import org.apache.hop.neo4j.logging.util.LoggingCore;
import org.apache.hop.ui.core.PropsUi;
+import org.apache.hop.ui.core.dialog.ErrorDialog;
import org.apache.hop.ui.core.gui.GuiResource;
import org.apache.hop.ui.core.widget.TreeMemory;
import org.apache.hop.ui.core.widget.TreeUtil;
@@ -206,33 +207,36 @@ public class NeoExecutionViewerErrorTab extends
NeoExecutionViewerTabBase {
pathParams.put("executionId", executionId);
String pathCypher = getPathToFailedCypher();
- getSession()
- .executeRead(
- tx -> {
- Result pathResult = tx.run(pathCypher, pathParams);
-
- while (pathResult.hasNext()) {
- Record pathRecord = pathResult.next();
- Value pathValue = pathRecord.get(0);
- Path path = pathValue.asPath();
- List<PathResult> shortestPath = new ArrayList<>();
- for (Node node : path.nodes()) {
- PathResult nodeResult = new PathResult();
- nodeResult.setId(LoggingCore.getStringValue(node, "id"));
- nodeResult.setName(LoggingCore.getStringValue(node, "name"));
- nodeResult.setType(LoggingCore.getStringValue(node,
"executionType"));
- nodeResult.setFailed(LoggingCore.getBooleanValue(node,
"failed"));
- nodeResult.setRegistrationDate(
- LoggingCore.getDateValue(node, "registrationDate"));
- nodeResult.setCopy(LoggingCore.getStringValue(node,
"copyNr"));
-
- shortestPath.add(0, nodeResult);
+ try {
+ getSession()
+ .executeRead(
+ tx -> {
+ Result pathResult = tx.run(pathCypher, pathParams);
+
+ while (pathResult.hasNext()) {
+ Record pathRecord = pathResult.next();
+ Value pathValue = pathRecord.get(0);
+ Path path = pathValue.asPath();
+ List<PathResult> shortestPath = new ArrayList<>();
+ for (Node node : path.nodes()) {
+ PathResult nodeResult = new PathResult();
+ nodeResult.setId(LoggingCore.getStringValue(node, "id"));
+ nodeResult.setName(LoggingCore.getStringValue(node,
"name"));
+ nodeResult.setType(LoggingCore.getStringValue(node,
"executionType"));
+ nodeResult.setFailed(LoggingCore.getBooleanValue(node,
"failed"));
+ nodeResult.setRegistrationDate(
+ LoggingCore.getDateValue(node, "registrationDate"));
+ nodeResult.setCopy(LoggingCore.getStringValue(node,
"copyNr"));
+
+ shortestPath.add(0, nodeResult);
+ }
+ shortestPaths.add(shortestPath);
}
- shortestPaths.add(shortestPath);
- }
- //
- return null;
- });
+ return null;
+ });
+ } catch (Exception e) {
+ new ErrorDialog(viewer.getShell(), "Error", "Error loading the error
lineage from Neo4j", e);
+ }
return shortestPaths;
}
diff --git
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/path/base/NeoExecutionViewerLineageTab.java
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/path/base/NeoExecutionViewerLineageTab.java
index d61c6ddcad..85341e4af6 100644
---
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/path/base/NeoExecutionViewerLineageTab.java
+++
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/path/base/NeoExecutionViewerLineageTab.java
@@ -31,6 +31,7 @@ import org.apache.hop.i18n.BaseMessages;
import org.apache.hop.neo4j.execution.path.PathResult;
import org.apache.hop.neo4j.logging.util.LoggingCore;
import org.apache.hop.ui.core.PropsUi;
+import org.apache.hop.ui.core.dialog.ErrorDialog;
import org.apache.hop.ui.core.gui.GuiResource;
import org.apache.hop.ui.core.widget.TreeMemory;
import org.apache.hop.ui.core.widget.TreeUtil;
@@ -211,33 +212,36 @@ public class NeoExecutionViewerLineageTab extends
NeoExecutionViewerTabBase {
pathParams.put("executionId", executionId);
String pathCypher = getPathToRootCypher();
- getSession()
- .executeRead(
- tx -> {
- Result pathResult = tx.run(pathCypher, pathParams);
-
- while (pathResult.hasNext()) {
- Record pathRecord = pathResult.next();
- Value pathValue = pathRecord.get(0);
- Path path = pathValue.asPath();
- List<PathResult> shortestPath = new ArrayList<>();
- for (Node node : path.nodes()) {
- PathResult nodeResult = new PathResult();
- nodeResult.setId(LoggingCore.getStringValue(node, "id"));
- nodeResult.setName(LoggingCore.getStringValue(node, "name"));
- nodeResult.setType(LoggingCore.getStringValue(node,
"executionType"));
- nodeResult.setFailed(LoggingCore.getBooleanValue(node,
"failed"));
- nodeResult.setRegistrationDate(
- LoggingCore.getDateValue(node, "registrationDate"));
- nodeResult.setCopy(LoggingCore.getStringValue(node,
"copyNr"));
-
- shortestPath.add(0, nodeResult);
+ try {
+ getSession()
+ .executeRead(
+ tx -> {
+ Result pathResult = tx.run(pathCypher, pathParams);
+
+ while (pathResult.hasNext()) {
+ Record pathRecord = pathResult.next();
+ Value pathValue = pathRecord.get(0);
+ Path path = pathValue.asPath();
+ List<PathResult> shortestPath = new ArrayList<>();
+ for (Node node : path.nodes()) {
+ PathResult nodeResult = new PathResult();
+ nodeResult.setId(LoggingCore.getStringValue(node, "id"));
+ nodeResult.setName(LoggingCore.getStringValue(node,
"name"));
+ nodeResult.setType(LoggingCore.getStringValue(node,
"executionType"));
+ nodeResult.setFailed(LoggingCore.getBooleanValue(node,
"failed"));
+ nodeResult.setRegistrationDate(
+ LoggingCore.getDateValue(node, "registrationDate"));
+ nodeResult.setCopy(LoggingCore.getStringValue(node,
"copyNr"));
+
+ shortestPath.add(0, nodeResult);
+ }
+ shortestPaths.add(shortestPath);
}
- shortestPaths.add(shortestPath);
- }
- //
- return null;
- });
+ return null;
+ });
+ } catch (Exception e) {
+ new ErrorDialog(viewer.getShell(), "Error", "Error loading execution
lineage from Neo4j", e);
+ }
return shortestPaths;
}
diff --git
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/path/base/NeoExecutionViewerTabBase.java
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/path/base/NeoExecutionViewerTabBase.java
index 5a0d819f96..a2ff9d85bf 100644
---
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/path/base/NeoExecutionViewerTabBase.java
+++
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/execution/path/base/NeoExecutionViewerTabBase.java
@@ -67,32 +67,51 @@ public abstract class NeoExecutionViewerTabBase {
}
protected String getPathToRootCypher() {
- // Do we have a parent? If not we can just return the cypher to the
current execution node
- //
- if (StringUtils.isEmpty(viewer.getExecution().getParentId())) {
+ return
buildPathToRootCypher(StringUtils.isNotEmpty(viewer.getExecution().getParentId()));
+ }
+
+ /**
+ * Cypher that walks from a child execution to the root parent using
directed EXECUTES
+ * relationships. The cartesian {@code MATCH (top:Execution)} form is
avoided because it does not
+ * scale on a busy logging graph and is a common timeout on Neo4j 5.
+ */
+ public static String buildPathToRootCypher(boolean hasParent) {
+ if (!hasParent) {
return "MATCH(e:Execution {id: $executionId }) " + Const.CR + "RETURN e
" + Const.CR;
- } else {
- return "MATCH(top:Execution), (child:Execution {id: $executionId }),
p=shortestPath((top)-[:EXECUTES*]-(child)) "
- + Const.CR
- + "WHERE top.parentId IS NULL "
- + Const.CR
- + "RETURN p "
- + Const.CR
- + "ORDER BY size(RELATIONSHIPS(p)) DESC "
- + Const.CR
- + "LIMIT 10 "
- + Const.CR;
}
+ return "MATCH (child:Execution {id: $executionId }) "
+ + Const.CR
+ + "MATCH p = shortestPath((top:Execution)-[:EXECUTES*]->(child)) "
+ + Const.CR
+ + "WHERE top.parentId IS NULL "
+ + Const.CR
+ + "RETURN p "
+ + Const.CR
+ + "ORDER BY size(RELATIONSHIPS(p)) DESC "
+ + Const.CR
+ + "LIMIT 10 "
+ + Const.CR;
}
protected String getPathToFailedCypher() {
+ return buildPathToFailedCypher();
+ }
- return "MATCH(top:Execution {id: $executionId }), (child:Execution),
p=shortestPath((top)-[:EXECUTES*]-(child)) "
+ /**
+ * Cypher that walks from the current execution to failed leaf executions.
The leaf predicate uses
+ * a pattern predicate rather than {@code size((n)-[:EXECUTES]->())}, which
Neo4j 5 removed.
+ */
+ public static String buildPathToFailedCypher() {
+ return "MATCH (top:Execution {id: $executionId }) "
+ Const.CR
- + "WHERE child.failed "
+ + "MATCH p = shortestPath((top)-[:EXECUTES*]->(child:Execution)) "
+ + Const.CR
+ + "WHERE child.failed = true "
+ Const.CR
+ "AND child.id <> $executionId "
+ Const.CR
+ + "AND NOT (child)-[:EXECUTES]->() "
+ + Const.CR
+ "RETURN p "
+ Const.CR
+ "ORDER BY size(RELATIONSHIPS(p)) "
diff --git
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/logging/util/LoggingCore.java
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/logging/util/LoggingCore.java
index b1f0825315..2091eb82e5 100644
---
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/logging/util/LoggingCore.java
+++
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/logging/util/LoggingCore.java
@@ -131,7 +131,6 @@ public class LoggingCore {
// Transaction is automatically committed by executeWrite
} catch (Exception e) {
log.logError("Error logging hierarchies", e);
- // Transaction is automatically rolled back by executeWrite on exception
}
}
diff --git
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/logging/xp/PipelineLoggingExtensionPoint.java
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/logging/xp/PipelineLoggingExtensionPoint.java
index 09cddab8e7..42ad268d5b 100644
---
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/logging/xp/PipelineLoggingExtensionPoint.java
+++
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/logging/xp/PipelineLoggingExtensionPoint.java
@@ -95,25 +95,25 @@ public class PipelineLoggingExtensionPoint
pipeline.addExecutionFinishedListener(
pipelineEngine -> {
- logEndOfPipeline(log, session, connection, pipelineEngine);
-
- // If there are no other parents, we now have the complete log
channel hierarchy
- //
- if (pipelineEngine.getParentWorkflow() == null
- && pipelineEngine.getParentPipeline() == null) {
- String logChannelId = pipelineEngine.getLogChannelId();
- List<LoggingHierarchy> loggingHierarchy =
- LoggingCore.getLoggingHierarchy(logChannelId);
- logHierarchy(log, session, connection, loggingHierarchy,
logChannelId);
- }
-
- // Let's not forget to close the session and driver...
- //
- if (session != null) {
- session.close();
- }
- if (driver != null) {
- driver.close();
+ try {
+ logEndOfPipeline(log, session, connection, pipelineEngine);
+
+ // If there are no other parents, we now have the complete log
channel hierarchy
+ //
+ if (pipelineEngine.getParentWorkflow() == null
+ && pipelineEngine.getParentPipeline() == null) {
+ String logChannelId = pipelineEngine.getLogChannelId();
+ List<LoggingHierarchy> loggingHierarchy =
+ LoggingCore.getLoggingHierarchy(logChannelId);
+ logHierarchy(log, session, connection, loggingHierarchy,
logChannelId);
+ }
+ } finally {
+ if (session != null) {
+ session.close();
+ }
+ if (driver != null) {
+ driver.close();
+ }
}
});
} catch (Exception e) {
diff --git
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/logging/xp/WorkflowLoggingExtensionPoint.java
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/logging/xp/WorkflowLoggingExtensionPoint.java
index 4d9b36e225..fa9f8baac0 100644
---
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/logging/xp/WorkflowLoggingExtensionPoint.java
+++
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/logging/xp/WorkflowLoggingExtensionPoint.java
@@ -90,24 +90,24 @@ public class WorkflowLoggingExtensionPoint
workflow.addWorkflowFinishedListener(
workflowMetaIWorkflowEngine -> {
- logEndOfWorkflow(log, session, connection, workflow);
-
- // If there are no other parents, we now have the complete log
channel hierarchy
- //
- if (workflow.getParentWorkflow() == null &&
workflow.getParentPipeline() == null) {
- String logChannelId = workflow.getLogChannelId();
- List<LoggingHierarchy> loggingHierarchy =
- LoggingCore.getLoggingHierarchy(logChannelId);
- logHierarchy(log, session, connection, loggingHierarchy,
logChannelId);
- }
-
- // Let's not forget to close the session and driver...
- //
- if (session != null) {
- session.close();
- }
- if (driver != null) {
- driver.close();
+ try {
+ logEndOfWorkflow(log, session, connection, workflow);
+
+ // If there are no other parents, we now have the complete log
channel hierarchy
+ //
+ if (workflow.getParentWorkflow() == null &&
workflow.getParentPipeline() == null) {
+ String logChannelId = workflow.getLogChannelId();
+ List<LoggingHierarchy> loggingHierarchy =
+ LoggingCore.getLoggingHierarchy(logChannelId);
+ logHierarchy(log, session, connection, loggingHierarchy,
logChannelId);
+ }
+ } finally {
+ if (session != null) {
+ session.close();
+ }
+ if (driver != null) {
+ driver.close();
+ }
}
});
diff --git
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/shared/DriverSingleton.java
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/shared/DriverSingleton.java
index 99d96c746e..bdfe307a22 100644
---
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/shared/DriverSingleton.java
+++
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/shared/DriverSingleton.java
@@ -37,7 +37,7 @@ public class DriverSingleton {
driverMap = new HashMap<>();
}
- public static DriverSingleton getInstance() {
+ public static synchronized DriverSingleton getInstance() {
if (singleton == null) {
singleton = new DriverSingleton();
}
@@ -47,27 +47,26 @@ public class DriverSingleton {
public static Driver getDriver(ILogChannel log, IVariables variables,
NeoConnection connection)
throws HopConfigException {
DriverSingleton ds = getInstance();
-
String key = getDriverKey(connection, variables);
-
- Driver driver = ds.driverMap.get(key);
- if (driver == null) {
- driver = connection.getDriver(log, variables);
- ds.driverMap.put(key, driver);
+ synchronized (ds.driverMap) {
+ Driver driver = ds.driverMap.get(key);
+ if (driver == null) {
+ driver = connection.getDriver(log, variables);
+ ds.driverMap.put(key, driver);
+ }
+ return driver;
}
-
- return driver;
}
public static void closeAll() {
DriverSingleton ds = getInstance();
-
- List<String> keys = new ArrayList<>(ds.getDriverMap().keySet());
- for (String key : keys) {
- synchronized (ds.getDriverMap()) {
- Driver driver = ds.driverMap.get(key);
- driver.close();
- ds.driverMap.remove(key);
+ synchronized (ds.driverMap) {
+ List<String> keys = new ArrayList<>(ds.driverMap.keySet());
+ for (String key : keys) {
+ Driver driver = ds.driverMap.remove(key);
+ if (driver != null) {
+ driver.close();
+ }
}
}
}
diff --git
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/shared/NeoConnection.java
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/shared/NeoConnection.java
index f0be18b4d4..4de7a1a0bd 100644
---
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/shared/NeoConnection.java
+++
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/shared/NeoConnection.java
@@ -412,6 +412,17 @@ public class NeoConnection extends HopMetadataBase
implements IHopMetadata {
Config config = configBuilder.build();
+ if (uris.isEmpty()) {
+ throw new HopConfigException("No Neo4j URIs configured for connection
" + name);
+ }
+ if (uris.size() > 1) {
+ log.logDetailed(
+ "Neo4j Java Driver 6 accepts a single URI; using "
+ + uris.get(0)
+ + " and ignoring "
+ + (uris.size() - 1)
+ + " additional server(s). Use the neo4j:// scheme for cluster
routing.");
+ }
Driver driver;
// In Neo4j 5.x, routingDriver() was removed. Use driver() with neo4j://
URI scheme for
// routing
diff --git
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/transforms/importer/Importer.java
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/transforms/importer/Importer.java
index 8e6297b312..001986163a 100644
---
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/transforms/importer/Importer.java
+++
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/transforms/importer/Importer.java
@@ -149,7 +149,10 @@ public class Importer extends BaseTransform<ImporterMeta,
ImporterData> {
// Determine Neo4j version (default to 4.x for backward compatibility)
String neo4jVersion =
StringUtils.isNotEmpty(meta.getNeo4jVersion()) ?
meta.getNeo4jVersion() : "4.x";
- boolean isNeo4j5 = "5.x".equals(neo4jVersion) ||
neo4jVersion.startsWith("5.");
+ boolean isNeo4j5 =
+ neo4jVersion.startsWith("5")
+ || neo4jVersion.startsWith("2025")
+ || neo4jVersion.startsWith("2026");
// Build command based on Neo4j version
if (isNeo4j5) {
diff --git
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/transforms/importer/ImporterDialog.java
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/transforms/importer/ImporterDialog.java
index b553a86705..43c091b8bb 100644
---
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/transforms/importer/ImporterDialog.java
+++
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/transforms/importer/ImporterDialog.java
@@ -211,7 +211,7 @@ public class ImporterDialog extends BaseTransformDialog {
fdlNeo4jVersion.top = new FormAttachment(lastControl, margin);
wlNeo4jVersion.setLayoutData(fdlNeo4jVersion);
wNeo4jVersion = new CCombo(wComposite, SWT.BORDER | SWT.READ_ONLY);
- wNeo4jVersion.setItems(new String[] {"4.x", "5.x"});
+ wNeo4jVersion.setItems(new String[] {"4.x", "5.x", "2025.x"});
PropsUi.setLook(wNeo4jVersion);
wNeo4jVersion.addModifyListener(lsMod);
FormData fdNeo4jVersion = new FormData();
@@ -621,7 +621,9 @@ public class ImporterDialog extends BaseTransformDialog {
wBaseFolder.setText(Const.NVL(input.getBaseFolder(), ""));
String neo4jVersion = Const.NVL(input.getNeo4jVersion(), "4.x");
- if (neo4jVersion.equals("5.x") || neo4jVersion.startsWith("5.")) {
+ if (neo4jVersion.startsWith("2025") || neo4jVersion.startsWith("2026")) {
+ wNeo4jVersion.setText("2025.x");
+ } else if (neo4jVersion.equals("5.x") || neo4jVersion.startsWith("5.")) {
wNeo4jVersion.setText("5.x");
} else {
wNeo4jVersion.setText("4.x");
diff --git
a/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/actions/index/Neo4jIndexCypherTest.java
b/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/actions/index/Neo4jIndexCypherTest.java
new file mode 100644
index 0000000000..f8c14eadb8
--- /dev/null
+++
b/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/actions/index/Neo4jIndexCypherTest.java
@@ -0,0 +1,55 @@
+/*
+ * 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.neo4j.actions.index;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.apache.hop.core.exception.HopException;
+import org.junit.jupiter.api.Test;
+
+class Neo4jIndexCypherTest {
+
+ @Test
+ void dropRequiresAnIndexName() {
+ IndexUpdate update = new IndexUpdate(UpdateType.DROP, ObjectType.NODE, "",
"Person", "id");
+
+ HopException exception =
+ assertThrows(HopException.class, () ->
Neo4jIndex.generateDropIndexCypher(update));
+ assertTrue(exception.getMessage().contains("Please drop indexes with the
name of the index"));
+ }
+
+ @Test
+ void dropUsesIfExists() throws HopException {
+ IndexUpdate update =
+ new IndexUpdate(UpdateType.DROP, ObjectType.NODE, "idx_person_id",
"Person", "id");
+
+ assertEquals("DROP INDEX idx_person_id IF EXISTS",
Neo4jIndex.generateDropIndexCypher(update));
+ }
+
+ @Test
+ void createUsesForOnSyntax() {
+ IndexUpdate update =
+ new IndexUpdate(UpdateType.CREATE, ObjectType.NODE, "idx_person_id",
"Person", "id,name");
+
+ String cypher = Neo4jIndex.generateCreateIndexCypher(update);
+ assertTrue(cypher.contains("CREATE INDEX idx_person_id IF NOT EXISTS FOR
(n:Person)"));
+ assertTrue(cypher.contains("ON (n.id, n.name)"));
+ }
+}
diff --git
a/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/core/data/GraphPropertyDataTypeTest.java
b/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/core/data/GraphPropertyDataTypeTest.java
new file mode 100644
index 0000000000..5abf7d004d
--- /dev/null
+++
b/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/core/data/GraphPropertyDataTypeTest.java
@@ -0,0 +1,30 @@
+/*
+ * 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.neo4j.core.data;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import org.junit.jupiter.api.Test;
+
+class GraphPropertyDataTypeTest {
+
+ @Test
+ void numberImportTypeIsDouble() {
+ assertEquals("double", GraphPropertyDataType.Number.getImportType());
+ }
+}
diff --git
a/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/execution/builder/BaseCypherBuilderTest.java
b/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/execution/builder/BaseCypherBuilderTest.java
new file mode 100644
index 0000000000..8d2a59d2ee
--- /dev/null
+++
b/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/execution/builder/BaseCypherBuilderTest.java
@@ -0,0 +1,76 @@
+/*
+ * 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.neo4j.execution.builder;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import java.math.BigDecimal;
+import java.time.LocalDateTime;
+import java.util.Date;
+import java.util.List;
+import java.util.Map;
+import org.junit.jupiter.api.Test;
+
+class BaseCypherBuilderTest {
+
+ private static Object mapped(Object value) {
+ return CypherMergeBuilder.of().withValue("x", value).parameters().get("x");
+ }
+
+ @Test
+ void integerBecomesLong() {
+ assertEquals(5L, mapped(5));
+ }
+
+ @Test
+ void bigDecimalBecomesString() {
+ assertEquals("1.25", mapped(new BigDecimal("1.25")));
+ }
+
+ @Test
+ void dateBecomesLocalDateTime() {
+ Date date = new Date(1_700_000_000_000L);
+ assertInstanceOf(LocalDateTime.class, mapped(date));
+ }
+
+ @Test
+ void mapBecomesJsonString() {
+ Object value = mapped(Map.of("a", "b"));
+ assertInstanceOf(String.class, value);
+ assertTrue(value.toString().contains("\"a\""));
+ }
+
+ @Test
+ void mixedListBecomesJsonString() {
+ Object value = mapped(List.of("a", 1));
+ assertInstanceOf(String.class, value);
+ }
+
+ @Test
+ void homogeneousStringListIsKept() {
+ Object value = mapped(List.of("a", "b"));
+ assertEquals(List.of("a", "b"), value);
+ }
+
+ @Test
+ void unknownObjectBecomesString() {
+ assertEquals("hello", mapped(new StringBuilder("hello")));
+ }
+}
diff --git
a/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/execution/cache/NeoLocationCacheTest.java
b/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/execution/cache/NeoLocationCacheTest.java
new file mode 100644
index 0000000000..064f2fb787
--- /dev/null
+++
b/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/execution/cache/NeoLocationCacheTest.java
@@ -0,0 +1,64 @@
+/*
+ * 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.neo4j.execution.cache;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+import org.apache.hop.execution.caching.CacheEntry;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+class NeoLocationCacheTest {
+
+ @BeforeEach
+ @AfterEach
+ void reset() {
+ NeoLocationCache.clear();
+ NeoLocationCache cache = NeoLocationCache.getInstance();
+ cache.setMaximumSize(1000);
+ cache.setEvictionBatchSize(50);
+ }
+
+ @Test
+ void evictionRemovesABatchNotTheWholeCache() {
+ NeoLocationCache cache = NeoLocationCache.getInstance();
+ cache.setMaximumSize(5);
+ cache.setEvictionBatchSize(50);
+
+ for (int i = 0; i < 55; i++) {
+ CacheEntry entry = new CacheEntry();
+ entry.setId("id-" + i);
+ NeoLocationCache.add(entry);
+ }
+
+ assertEquals(5, cache.getCache().size());
+ }
+
+ @Test
+ void removeDropsASingleEntry() {
+ CacheEntry entry = new CacheEntry();
+ entry.setId("keep-me");
+ NeoLocationCache.add(entry);
+
+ NeoLocationCache.remove("keep-me");
+
+ assertNull(NeoLocationCache.get("keep-me"));
+ }
+}
diff --git
a/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/execution/path/NeoExecutionViewerTabBaseTest.java
b/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/execution/path/NeoExecutionViewerTabBaseTest.java
new file mode 100644
index 0000000000..2672cd3509
--- /dev/null
+++
b/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/execution/path/NeoExecutionViewerTabBaseTest.java
@@ -0,0 +1,56 @@
+/*
+ * 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.neo4j.execution.path;
+
+import static
org.apache.hop.neo4j.CypherAssertions.assertNoSizeOfPatternExpression;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+import org.apache.hop.neo4j.execution.path.base.NeoExecutionViewerTabBase;
+import org.junit.jupiter.api.Test;
+
+class NeoExecutionViewerTabBaseTest {
+
+ @Test
+ void lineageCypherIsDirectedAndBoundByChildId() {
+ String cypher = NeoExecutionViewerTabBase.buildPathToRootCypher(true);
+
+ assertTrue(cypher.contains("MATCH (child:Execution {id: $executionId })"));
+ assertTrue(cypher.contains("[:EXECUTES*]->(child)"));
+ assertFalse(cypher.contains("MATCH(top:Execution), (child:Execution"));
+ assertNoSizeOfPatternExpression(cypher);
+ }
+
+ @Test
+ void errorPathCypherUsesALeafPredicateAndBooleanFailed() {
+ String cypher = NeoExecutionViewerTabBase.buildPathToFailedCypher();
+
+ assertTrue(cypher.contains("child.failed = true"));
+ assertTrue(cypher.contains("AND NOT (child)-[:EXECUTES]->()"));
+ assertTrue(cypher.contains("[:EXECUTES*]->(child:Execution)"));
+ assertNoSizeOfPatternExpression(cypher);
+ }
+
+ @Test
+ void rootExecutionLineageDoesNotWalkAPath() {
+ String cypher = NeoExecutionViewerTabBase.buildPathToRootCypher(false);
+
+ assertTrue(cypher.contains("RETURN e"));
+ assertFalse(cypher.contains("shortestPath"));
+ }
+}
diff --git
a/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/perspective/ErrorPathCypherIT.java
b/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/perspective/ErrorPathCypherIT.java
index d02014638d..6296cefc5e 100644
---
a/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/perspective/ErrorPathCypherIT.java
+++
b/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/perspective/ErrorPathCypherIT.java
@@ -27,6 +27,7 @@ import java.time.Duration;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
+import org.apache.hop.neo4j.execution.path.base.NeoExecutionViewerTabBase;
import org.junit.jupiter.api.AfterAll;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
@@ -100,6 +101,17 @@ class ErrorPathCypherIT {
CREATE (workflow)-[:EXECUTES]->(pipeline)
CREATE (pipeline)-[:EXECUTES]->(failed)
CREATE (pipeline)-[:EXECUTES]->(ok)
+ CREATE (eiWorkflow:Execution { id : 'ei-workflow', name :
'main',\
+ executionType : 'Workflow', failed : true })
+ CREATE (eiPipeline:Execution { id : 'ei-pipeline', name :
'load',\
+ executionType : 'Pipeline', parentId : 'ei-workflow', failed
: true })
+ CREATE (eiFailed:Execution { id : 'ei-failed', name : 'Table
output',\
+ executionType : 'Transform', parentId : 'ei-pipeline', failed
: true })
+ CREATE (eiOk:Execution { id : 'ei-ok', name : 'Dummy',\
+ executionType : 'Transform', parentId : 'ei-pipeline', failed
: false })
+ CREATE (eiWorkflow)-[:EXECUTES]->(eiPipeline)
+ CREATE (eiPipeline)-[:EXECUTES]->(eiFailed)
+ CREATE (eiPipeline)-[:EXECUTES]->(eiOk)
""");
return null;
});
@@ -171,4 +183,47 @@ class ErrorPathCypherIT {
assertThrows(ClientException.class, () ->
runErrorPathCypher(removedSyntax));
}
+
+ @Test
+ void executionInfoErrorPathReturnsTheDeepestFailedLeaf() {
+ List<Record> records = runExecutionInfoErrorPath();
+
+ assertEquals(1, records.size());
+ assertEquals(List.of("ei-workflow", "ei-pipeline", "ei-failed"),
executionIds(records.get(0)));
+ }
+
+ @Test
+ void executionInfoErrorPathSkipsNonLeavesAndSuccessfulTransforms() {
+ List<String> ids = executionIds(runExecutionInfoErrorPath().get(0));
+
+ assertEquals("ei-failed", ids.get(ids.size() - 1));
+ assertFalse(ids.contains("ei-ok"));
+ }
+
+ @Test
+ void executionInfoLineageWalksDirectedExecutesToTheRoot() {
+ try (Session session = driver.session()) {
+ List<Record> records =
+ session.executeRead(
+ tx ->
+ tx.run(
+
NeoExecutionViewerTabBase.buildPathToRootCypher(true),
+ Map.of("executionId", "ei-failed"))
+ .list());
+ assertEquals(1, records.size());
+ assertEquals(
+ List.of("ei-workflow", "ei-pipeline", "ei-failed"),
executionIds(records.get(0)));
+ }
+ }
+
+ private List<Record> runExecutionInfoErrorPath() {
+ try (Session session = driver.session()) {
+ return session.executeRead(
+ tx ->
+ tx.run(
+ NeoExecutionViewerTabBase.buildPathToFailedCypher(),
+ Map.of("executionId", "ei-workflow"))
+ .list());
+ }
+ }
}
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 4ccb36ff59..1480226183 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
@@ -791,6 +791,7 @@ public class PipelineExecutionViewer extends
BaseExecutionViewer
pipelinePainter.setMaximum(maximum);
pipelinePainter.setShowingNavigationView(true);
pipelinePainter.setScreenMagnification(magnification);
+ pipelinePainter.setTransformLogMap(buildTransformErrorMap());
try {
pipelinePainter.drawPipelineImage();
@@ -808,6 +809,28 @@ public class PipelineExecutionViewer extends
BaseExecutionViewer
CanvasFacade.setData(canvas, magnification, offset, pipelineMeta);
}
+ /**
+ * Populate the painter error map from stored component metrics so failed
transforms get a red
+ * border even though this viewer has no live {@code IPipelineEngine}.
+ */
+ private Map<String, String> buildTransformErrorMap() {
+ Map<String, String> transformErrorMap = new HashMap<>();
+ if (executionState == null || executionState.getMetrics() == null) {
+ return transformErrorMap;
+ }
+ String errorHeader = Pipeline.METRIC_ERROR.getHeader();
+ for (ExecutionStateComponentMetrics metrics : executionState.getMetrics())
{
+ if (metrics.getMetrics() == null) {
+ continue;
+ }
+ Long errors = metrics.getMetrics().get(errorHeader);
+ if (errors != null && errors > 0 &&
StringUtils.isNotEmpty(metrics.getComponentName())) {
+ transformErrorMap.put(metrics.getComponentName(), errors + "
error(s)");
+ }
+ }
+ return transformErrorMap;
+ }
+
@Override
public Control getControl() {
return this;
@@ -1107,7 +1130,7 @@ public class PipelineExecutionViewer extends
BaseExecutionViewer
// Don't load logging text as that can be a lot of data.
// Lazily load that when the logging text comes into focus.
//
- ExecutionState executionState =
iLocation.getExecutionState(execution.getId(), false);
+ ExecutionState executionState =
iLocation.getExecutionState(child.getId(), false);
perspective.createExecutionViewer(locationName, child,
executionState);
return;
}
@@ -1158,7 +1181,7 @@ public class PipelineExecutionViewer extends
BaseExecutionViewer
}
// Don't load execution logging text to prevent memory issues.
//
- ExecutionState executionState =
iLocation.getExecutionState(execution.getId(), false);
+ ExecutionState executionState =
iLocation.getExecutionState(childExecution.getId(), false);
perspective.createExecutionViewer(locationName, childExecution,
executionState);
} catch (Exception e) {
