This is an automated email from the ASF dual-hosted git repository.
hansva 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 79cfa6aa70 Fixes #8993 : Run CALL { } IN TRANSACTIONS in auto-commit
on Neo4j (#8994)
79cfa6aa70 is described below
commit 79cfa6aa709e79f39d6938cee60062607c2e38f0
Author: Bart Maertens <[email protected]>
AuthorDate: Mon Oct 19 09:25:45 2026 +0200
Fixes #8993 : Run CALL { } IN TRANSACTIONS in auto-commit on Neo4j (#8994)
---
.../pages/pipeline/transforms/graph-query.adoc | 20 ++++
.../ROOT/pages/workflow/actions/graph-script.adoc | 9 ++
.../neo4j/actions/cypherscript/CypherScript.java | 82 +++++++++++++---
.../apache/hop/neo4j/bolt/Neo4jGraphDialect.java | 106 +++++++++++++++++++++
.../apache/hop/neo4j/transforms/cypher/Cypher.java | 47 ++++++++-
.../cypherscript/CypherScriptTransactionTest.java | 92 +++++++++++++++++-
.../hop/neo4j/bolt/BoltGraphDialectTest.java | 53 +++++++++++
.../neo4j/transforms/cypher/CypherRetryTest.java | 100 +++++++++++++++++++
8 files changed, 491 insertions(+), 18 deletions(-)
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/graph-query.adoc
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/graph-query.adoc
index 0938834f07..84ef1d97c0 100644
---
a/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/graph-query.adoc
+++
b/docs/hop-user-manual/modules/ROOT/pages/pipeline/transforms/graph-query.adoc
@@ -81,3 +81,23 @@ Give a return value the type `Vector` to read a stored
embedding into a Vector f
On Gremlin servers a traversal returns a row per result: maps, from
`project()` or `elementMap()` for example, are rows of their own, anything else
is a value called `result`.
|===
+
+== Transactions
+
+The statements of a batch of input rows, or the UNWIND statement of a batch,
run in one transaction.
+
+On Neo4j, a statement with `CALL { ... } IN TRANSACTIONS` (or `USING PERIODIC
COMMIT` on Neo4j 4) runs on its own in an auto-commit transaction: Neo4j
commits its batches of rows itself.
+This is the alternative to `apoc.periodic.iterate`.
+With UNWIND, the transform passes a batch of input rows and Neo4j commits them
in smaller batches:
+
+[source,cypher]
+----
+UNWIND $rows AS row
+CALL (row) {
+ MERGE (c:Customer { id : row.id })
+ SET c.name = row.name
+} IN TRANSACTIONS OF 1000 ROWS
+----
+
+When such a statement fails, the batches Neo4j committed before the error stay
committed.
+With retries configured, a retry runs the whole statement again and applies
those batches a second time: use MERGE rather than CREATE to make the statement
safe to repeat, or don't configure retries.
diff --git
a/docs/hop-user-manual/modules/ROOT/pages/workflow/actions/graph-script.adoc
b/docs/hop-user-manual/modules/ROOT/pages/workflow/actions/graph-script.adoc
index 07aeafbe23..87399be7a3 100644
--- a/docs/hop-user-manual/modules/ROOT/pages/workflow/actions/graph-script.adoc
+++ b/docs/hop-user-manual/modules/ROOT/pages/workflow/actions/graph-script.adoc
@@ -34,6 +34,15 @@ The result of this action is a `true` or `false` exit code
that determines if a
**To allow you to enter semi-colons (;) in commands, strings and so on we
require you to separate commands with a semi-colon on a new line**
+== Transactions
+
+On a database with transactions the script runs in one transaction: when a
statement fails, none of the statements are kept.
+
+On Neo4j, a statement with `CALL { ... } IN TRANSACTIONS` (or `USING PERIODIC
COMMIT` on Neo4j 4) runs on its own in an auto-commit transaction, and the
statements before and after it run in a transaction each.
+The script then no longer runs in a single transaction: when a statement
fails, the statements before it stay committed.
+
+On Memgraph every statement runs on its own in an auto-commit transaction.
+
== Configuration options
|===
diff --git
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/actions/cypherscript/CypherScript.java
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/actions/cypherscript/CypherScript.java
index 076088eaef..09f4bbec06 100644
---
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/actions/cypherscript/CypherScript.java
+++
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/actions/cypherscript/CypherScript.java
@@ -95,23 +95,22 @@ public class CypherScript extends ActionBase implements
IAction {
try (IGraphConnection connection =
graphConnection.connect(getLogChannel(), this)) {
IGraphDialect dialect = connection.getGraphDialect();
if (dialect.isSupportingSchemaChangesInTransactions()
+ && connection.isSupportingTransactions()
+ && statements.stream().anyMatch(dialect::isRequiringAutoCommit)) {
+ // Statements like CALL { ... } IN TRANSACTIONS can't run in an
explicit transaction
+ //
+ if (isBasic()) {
+ logBasic(
+ "Some statements run on their own in auto-commit transactions:
the script doesn't"
+ + " run in a single transaction");
+ }
+ nrExecuted = executeAroundAutoCommit(connection, dialect, statements,
result);
+ } else if (dialect.isSupportingSchemaChangesInTransactions()
&& connection.isSupportingTransactions()) {
try {
- nrExecuted =
- connection.executeWrite(
- transaction -> {
- int executed = 0;
- for (String cypher : statements) {
- transaction.execute(cypher, Map.of());
- executed++;
- if (isDetailed()) {
- logDetailed("Executed cypher statement: " + cypher);
- }
- }
- // The transaction is committed when this work returns,
and rolled back when it
- // throws: a failed script keeps none of its statements.
- return executed;
- });
+ // The transaction is committed when the work returns, and rolled
back when it throws: a
+ // failed script keeps none of its statements.
+ nrExecuted = executeInTransaction(connection, statements);
} catch (Exception e) {
logError("Error executing cypher statements, the transaction is
rolled back", e);
result.setNrErrors(1);
@@ -211,6 +210,59 @@ public class CypherScript extends ActionBase implements
IAction {
this.replacingVariables = replacingVariables;
}
+ /** Run the statements in one transaction, none if there are no statements.
*/
+ private int executeInTransaction(IGraphConnection connection, List<String>
statements)
+ throws HopException {
+ if (statements.isEmpty()) {
+ return 0;
+ }
+ return connection.executeWrite(
+ transaction -> {
+ int executed = 0;
+ for (String cypher : statements) {
+ transaction.execute(cypher, Map.of());
+ executed++;
+ if (isDetailed()) {
+ logDetailed("Executed cypher statement: " + cypher);
+ }
+ }
+ return executed;
+ });
+ }
+
+ /**
+ * Run the statements which need an auto-commit transaction on their own,
and the statements
+ * between them in one transaction each, in the order of the script. An
error stops the script:
+ * what ran before the failed statement or transaction stays committed.
+ */
+ private int executeAroundAutoCommit(
+ IGraphConnection connection, IGraphDialect dialect, List<String>
statements, Result result) {
+ int executed = 0;
+ List<String> inTransaction = new ArrayList<>();
+ try {
+ for (String cypher : statements) {
+ if (dialect.isRequiringAutoCommit(cypher)) {
+ executed += executeInTransaction(connection, inTransaction);
+ inTransaction.clear();
+ connection.execute(cypher, Map.of());
+ executed++;
+ if (isDetailed()) {
+ logDetailed("Executed cypher statement in an auto-commit
transaction: " + cypher);
+ }
+ } else {
+ inTransaction.add(cypher);
+ }
+ }
+ executed += executeInTransaction(connection, inTransaction);
+ } catch (Exception e) {
+ logError(
+ "Error executing cypher statements, the statements before the error
stay committed", e);
+ result.setNrErrors(1);
+ result.setResult(false);
+ }
+ return executed;
+ }
+
private int executeAutoCommit(
IGraphConnection connection, List<String> statements, Result result) {
int executed = 0;
diff --git
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/bolt/Neo4jGraphDialect.java
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/bolt/Neo4jGraphDialect.java
index 6d7fef8b24..33ff7ee31c 100644
---
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/bolt/Neo4jGraphDialect.java
+++
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/bolt/Neo4jGraphDialect.java
@@ -19,6 +19,8 @@ package org.apache.hop.neo4j.bolt;
import java.util.List;
import java.util.Map;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
import org.apache.hop.core.exception.HopException;
import org.apache.hop.core.graph.GraphIndex;
import org.apache.hop.core.graph.GraphSchema;
@@ -29,10 +31,114 @@ public class Neo4jGraphDialect extends BoltGraphDialect {
public static final Neo4jGraphDialect INSTANCE = new Neo4jGraphDialect();
+ /** A CALL keyword, not a property or a part of a longer name. */
+ private static final Pattern CALL =
+ Pattern.compile("(?<![\\w.$])CALL\\b", Pattern.CASE_INSENSITIVE);
+
+ /** What follows the subquery of CALL { ... } IN [n] [CONCURRENT]
TRANSACTIONS. */
+ private static final Pattern IN_TRANSACTIONS =
+ Pattern.compile(
+ "\\s*IN\\s+(?:[^\\s{}()]+\\s+)?(?:CONCURRENT\\s+)?TRANSACTIONS\\b",
+ Pattern.CASE_INSENSITIVE);
+
+ /** USING PERIODIC COMMIT, before LOAD CSV in Neo4j 4. */
+ private static final Pattern PERIODIC_COMMIT =
+ Pattern.compile("(?<![\\w.$])USING\\s+PERIODIC\\s+COMMIT\\b",
Pattern.CASE_INSENSITIVE);
+
public Neo4jGraphDialect() {
super("NEO4J");
}
+ /**
+ * Neo4j only runs CALL { ... } IN TRANSACTIONS and USING PERIODIC COMMIT in
an implicit
+ * transaction: they commit batches of rows themselves.
+ */
+ @Override
+ public boolean isRequiringAutoCommit(String statement) {
+ if (statement == null) {
+ return false;
+ }
+ String code = blankLiteralsAndComments(statement);
+ return PERIODIC_COMMIT.matcher(code).find() || isCallInTransactions(code);
+ }
+
+ /** True if the code has a CALL subquery followed by IN TRANSACTIONS. */
+ private static boolean isCallInTransactions(String code) {
+ Matcher call = CALL.matcher(code);
+ while (call.find()) {
+ int index = skipWhitespace(code, call.end());
+ // The variables of the subquery: CALL (n, m) { ... } or CALL (*) { ... }
+ if (index < code.length() && code.charAt(index) == '(') {
+ index = skipWhitespace(code, closingIndex(code, index, '(', ')') + 1);
+ }
+ if (index >= code.length() || code.charAt(index) != '{') {
+ // A procedure call
+ continue;
+ }
+ int end = closingIndex(code, index, '{', '}');
+ if (end < code.length()
+ && IN_TRANSACTIONS.matcher(code).region(end + 1,
code.length()).lookingAt()) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private static int skipWhitespace(String code, int index) {
+ while (index < code.length() &&
Character.isWhitespace(code.charAt(index))) {
+ index++;
+ }
+ return index;
+ }
+
+ /** The index of the bracket which closes the one at the index, the length
if there is none. */
+ private static int closingIndex(String code, int index, char open, char
close) {
+ int depth = 0;
+ for (int i = index; i < code.length(); i++) {
+ char c = code.charAt(i);
+ if (c == open) {
+ depth++;
+ } else if (c == close && --depth == 0) {
+ return i;
+ }
+ }
+ return code.length();
+ }
+
+ /**
+ * The statement with the string literals, quoted names and comments
replaced by spaces, so that
+ * the words and brackets in them don't count.
+ */
+ static String blankLiteralsAndComments(String statement) {
+ StringBuilder code = new StringBuilder(statement.length());
+ int i = 0;
+ while (i < statement.length()) {
+ char c = statement.charAt(i);
+ int end;
+ if (c == '\'' || c == '"' || c == '`') {
+ end = i + 1;
+ while (end < statement.length() && statement.charAt(end) != c) {
+ // A backslash escapes the next character in a string, not in a
quoted name
+ end += c != '`' && statement.charAt(end) == '\\' ? 2 : 1;
+ }
+ end = Math.min(end + 1, statement.length());
+ } else if (statement.startsWith("//", i)) {
+ int newline = statement.indexOf('\n', i);
+ end = newline < 0 ? statement.length() : newline;
+ } else if (statement.startsWith("/*", i)) {
+ int close = statement.indexOf("*/", i + 2);
+ end = close < 0 ? statement.length() : close + 2;
+ } else {
+ code.append(c);
+ i++;
+ continue;
+ }
+ code.append(" ".repeat(end - i));
+ i = end;
+ }
+ return code.toString();
+ }
+
@Override
public List<GraphIndex> getIndexes(IGraphConnection connection) throws
HopException {
try {
diff --git
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/transforms/cypher/Cypher.java
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/transforms/cypher/Cypher.java
index 8e831921bc..e90f4ddc09 100644
---
a/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/transforms/cypher/Cypher.java
+++
b/plugins/tech/neo4j/src/main/java/org/apache/hop/neo4j/transforms/cypher/Cypher.java
@@ -481,12 +481,53 @@ public class Cypher extends BaseTransform<CypherMeta,
CypherData> {
throw new HopException(
"Unable to execute cypher statement '" +
cypherStatement.getCypher() + "'", e);
}
- logBasic("Retrying after attempt #" + (attempt + 1) + " with error : "
+ e.getMessage());
+ logAutoCommitRetry(attempt, e);
}
}
flushOutputRows();
}
+ /**
+ * Run the unwind statement in an auto-commit transaction, with the
configured retries. A
+ * statement like CALL { ... } IN TRANSACTIONS commits its own batches of
rows.
+ */
+ private void writeAutoCommitUnwindList(Map<String, Object> unwindMap) throws
HopException {
+ beginWork();
+ for (int attempt = 0; attempt < data.attempts; attempt++) {
+ try {
+ startAttempt();
+ Result result = data.session.run(data.cypher, unwindMap);
+ getResultRows(result, new Object[0], true);
+ // Wait for the end of the statement: an error in one of its batches
is raised here
+ result.consume();
+ break;
+ } catch (Exception e) {
+ dropAttemptRows();
+ if (attempt + 1 >= data.attempts) {
+ throw new HopException("Unexpected error writing unwind list to
Neo4j", e);
+ }
+ logAutoCommitRetry(attempt, e);
+ }
+ }
+ flushOutputRows();
+ setLinesOutput(getLinesOutput() + data.unwindList.size());
+ data.unwindList.clear();
+ data.outputCount = 0;
+ }
+
+ /**
+ * Log the retry of a statement in an auto-commit transaction. A statement
which commits its own
+ * batches, like CALL { ... } IN TRANSACTIONS, keeps the batches committed
before the error.
+ */
+ private void logAutoCommitRetry(int attempt, Exception e) {
+ logBasic("Retrying after attempt #" + (attempt + 1) + " with error : " +
e.getMessage());
+ if (!meta.isReadOnly()) {
+ logBasic(
+ "Warning: the statement runs in an auto-commit transaction. Batches
it committed before"
+ + " the error stay committed and the retry applies them again");
+ }
+ }
+
private List<Object[]> writeUnwindList() throws HopException {
if (data.graphConnection != null) {
writeGenericUnwindList();
@@ -494,6 +535,10 @@ public class Cypher extends BaseTransform<CypherMeta,
CypherData> {
}
HashMap<String, Object> unwindMap = new HashMap<>();
unwindMap.put(data.unwindMapName, data.unwindList);
+ if (data.neoConnection.getDialect().isRequiringAutoCommit(data.cypher)) {
+ writeAutoCommitUnwindList(unwindMap);
+ return null;
+ }
List<Object[]> resultRows = null;
CypherTransactionWork cypherTransactionWork =
new CypherTransactionWork(this, new Object[0], true, data.cypher,
unwindMap);
diff --git
a/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/actions/cypherscript/CypherScriptTransactionTest.java
b/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/actions/cypherscript/CypherScriptTransactionTest.java
index 22e184b4af..03000df0d7 100644
---
a/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/actions/cypherscript/CypherScriptTransactionTest.java
+++
b/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/actions/cypherscript/CypherScriptTransactionTest.java
@@ -91,15 +91,96 @@ class CypherScriptTransactionTest {
assertTrue(database.committed);
}
+ @Test
+ void autoCommitStatementRunsOnItsOwn() throws Exception {
+ TransactionalDatabase database = new TransactionalDatabase(new
AutoCommitDialect());
+ CypherScript action = action(database);
+ action.setScript("CREATE (:A)\n;\nCREATE (:B)\n;\nAUTO\n;\nCREATE (:C)");
+
+ Result result = action.execute(new Result(), 0);
+
+ assertTrue(result.getResult());
+ assertEquals(0, result.getNrErrors());
+ assertEquals(List.of("CREATE (:A)", "CREATE (:B)", "AUTO", "CREATE (:C)"),
database.executed);
+ assertEquals(List.of("AUTO"), database.autoCommitted);
+ // The statements before and after the auto-commit statement: one
transaction each
+ assertEquals(2, database.commits);
+ }
+
+ @Test
+ void failureAfterAutoCommitKeepsWhatRanBefore() throws Exception {
+ TransactionalDatabase database = new TransactionalDatabase(new
AutoCommitDialect());
+ CypherScript action = action(database);
+ action.setScript("CREATE (:A)\n;\nAUTO\n;\nFAIL\n;\nCREATE (:B)");
+
+ Result result = action.execute(new Result(), 0);
+
+ assertFalse(result.getResult());
+ assertEquals(1, result.getNrErrors());
+ assertEquals(List.of("CREATE (:A)", "AUTO", "FAIL"), database.executed);
+ assertEquals(List.of("AUTO"), database.autoCommitted);
+ assertEquals(1, database.commits);
+ assertTrue(
+ database.rolledBack, "The transaction with the failed statement was
not rolled back");
+ }
+
+ @Test
+ void scriptWithoutAutoCommitStatementsKeepsOneTransaction() throws Exception
{
+ TransactionalDatabase database = new TransactionalDatabase(new
AutoCommitDialect());
+ CypherScript action = action(database);
+ action.setScript("CREATE (:A)\n;\nCREATE (:B)");
+
+ Result result = action.execute(new Result(), 0);
+
+ assertTrue(result.getResult());
+ assertEquals(List.of(), database.autoCommitted);
+ assertEquals(1, database.commits);
+ }
+
+ private static CypherScript action(TransactionalDatabase database) throws
HopException {
+ MemoryMetadataProvider metadataProvider = new MemoryMetadataProvider();
+ metadataProvider
+ .getSerializer(GraphDatabaseMeta.class)
+ .save(new GraphDatabaseMeta("tx", database));
+ CypherScript action = new CypherScript("script");
+ action.setMetadataProvider(metadataProvider);
+ action.setConnectionName("tx");
+ return action;
+ }
+
+ /** Statements starting with AUTO can't run in an explicit transaction. */
+ private static class AutoCommitDialect extends CypherGraphDialect {
+ AutoCommitDialect() {
+ super("AUTO_COMMIT");
+ }
+
+ @Override
+ public boolean isRequiringAutoCommit(String statement) {
+ return statement.startsWith("AUTO");
+ }
+ }
+
/** Runs write work in a transaction that commits on return and rolls back
on an exception. */
private static class TransactionalDatabase extends BaseGraphDatabase {
final List<String> executed = new ArrayList<>();
+ final List<String> autoCommitted = new ArrayList<>();
+ private final IGraphDialect dialect;
boolean committed;
boolean rolledBack;
+ int commits;
+ boolean inTransaction;
+
+ TransactionalDatabase() {
+ this(CypherGraphDialect.DEFAULT);
+ }
+
+ TransactionalDatabase(IGraphDialect dialect) {
+ this.dialect = dialect;
+ }
@Override
public IGraphDialect getGraphDialect() {
- return CypherGraphDialect.DEFAULT;
+ return dialect;
}
@Override
@@ -109,6 +190,9 @@ class CypherScriptTransactionTest {
public List<Map<String, Object>> execute(String statement, Map<String,
Object> parameters)
throws HopException {
executed.add(statement);
+ if (!inTransaction) {
+ autoCommitted.add(statement);
+ }
if (statement.startsWith("FAIL")) {
throw new HopException("Syntax error in " + statement);
}
@@ -145,19 +229,23 @@ class CypherScriptTransactionTest {
// Nothing to close
}
};
+ inTransaction = true;
try {
T value = work.execute(transaction);
committed = true;
+ commits++;
return value;
} catch (HopException | RuntimeException e) {
rolledBack = true;
throw e;
+ } finally {
+ inTransaction = false;
}
}
@Override
public IGraphDialect getGraphDialect() {
- return CypherGraphDialect.DEFAULT;
+ return dialect;
}
@Override
diff --git
a/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/bolt/BoltGraphDialectTest.java
b/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/bolt/BoltGraphDialectTest.java
index 6bfe2a91ac..3abb7f18e4 100644
---
a/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/bolt/BoltGraphDialectTest.java
+++
b/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/bolt/BoltGraphDialectTest.java
@@ -135,6 +135,59 @@ class BoltGraphDialectTest {
assertFalse(Neo4jGraphDialect.INSTANCE.isRequiringAutoCommit("SHOW
INDEXES"));
}
+ /** Neo4j only runs statements which commit their own batches in an implicit
transaction. */
+ @Test
+ void testNeo4jAutoCommitStatements() {
+ Neo4jGraphDialect neo4j = Neo4jGraphDialect.INSTANCE;
+ assertTrue(
+ neo4j.isRequiringAutoCommit(
+ "MATCH (n:Person) CALL (n) { SET n.processed = true } IN
TRANSACTIONS OF 10 ROWS"));
+ assertTrue(
+ neo4j.isRequiringAutoCommit(
+ "MATCH (n:Person) CALL { WITH n SET n.processed = true } IN
TRANSACTIONS"));
+ assertTrue(neo4j.isRequiringAutoCommit("call () {\n create (:A)\n}\nin
transactions"));
+ assertTrue(
+ neo4j.isRequiringAutoCommit(
+ "UNWIND $rows AS row CALL (*) { MERGE (p:Person {id: row.id}) SET
p += {name: row.name}"
+ + " } IN 4 CONCURRENT TRANSACTIONS OF 1000 ROWS ON ERROR
CONTINUE"));
+ assertTrue(
+ neo4j.isRequiringAutoCommit(
+ "CALL (row) { CREATE (:A {text: '}'}) } IN CONCURRENT
TRANSACTIONS"));
+ assertTrue(
+ neo4j.isRequiringAutoCommit(
+ "// batches\nCALL () { CREATE (:A) } /* of rows */ IN TRANSACTIONS
OF 5 ROWS"));
+ assertTrue(
+ neo4j.isRequiringAutoCommit(
+ "USING PERIODIC COMMIT 500 LOAD CSV FROM 'file:///a.csv' AS line
CREATE (:A)"));
+
+ // Statements which run in a transaction as before
+ assertFalse(neo4j.isRequiringAutoCommit(null));
+ assertFalse(neo4j.isRequiringAutoCommit("SHOW INDEXES"));
+ assertFalse(neo4j.isRequiringAutoCommit("MATCH (n) CALL (n) { SET n.x = 1
} RETURN n"));
+ assertFalse(neo4j.isRequiringAutoCommit("CALL db.labels() YIELD label
RETURN label"));
+ assertFalse(
+ neo4j.isRequiringAutoCommit(
+ "CALL apoc.periodic.iterate('MATCH (n) RETURN n', 'SET n.x = 1',
{batchSize: 100})"));
+ assertFalse(
+ neo4j.isRequiringAutoCommit(
+ "MATCH (n) WHERE n.note = 'CALL { } IN TRANSACTIONS' RETURN n"));
+ assertFalse(neo4j.isRequiringAutoCommit("// CALL () { } IN
TRANSACTIONS\nMATCH (n) RETURN n"));
+ assertFalse(neo4j.isRequiringAutoCommit("MATCH (n) WHERE {id: n.id} IN
transactions RETURN n"));
+ assertFalse(
+ neo4j.isRequiringAutoCommit(
+ "MATCH (n) WHERE EXISTS { MATCH (n)--() } AND n.id IN transactions
RETURN n"));
+ assertFalse(neo4j.isRequiringAutoCommit("MATCH (n {call: 1}) RETURN
n.call"));
+ assertFalse(neo4j.isRequiringAutoCommit("MATCH (n:`CALL {} IN
TRANSACTIONS`) RETURN n"));
+
+ // The other dialects are unchanged
+ assertFalse(
+ NeptuneGraphDialect.INSTANCE.isRequiringAutoCommit(
+ "CALL () { CREATE (:A) } IN TRANSACTIONS"));
+ assertFalse(
+ MemgraphGraphDialect.INSTANCE.isRequiringAutoCommit(
+ "CALL () { CREATE (:A) } IN TRANSACTIONS"));
+ }
+
/** Memgraph has no IF [NOT] EXISTS: its errors about existing or missing
schema are tolerated. */
@Test
void testMemgraphExistingOrMissing() {
diff --git
a/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/transforms/cypher/CypherRetryTest.java
b/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/transforms/cypher/CypherRetryTest.java
index ec82c0602b..27f5774fc1 100644
---
a/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/transforms/cypher/CypherRetryTest.java
+++
b/plugins/tech/neo4j/src/test/java/org/apache/hop/neo4j/transforms/cypher/CypherRetryTest.java
@@ -24,8 +24,12 @@ import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyMap;
+import static org.mockito.ArgumentMatchers.contains;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
import java.util.ArrayList;
@@ -212,6 +216,102 @@ class CypherRetryTest {
assertArrayEquals(new Object[] {"b"}, cypher.rows.get(1));
}
+ @Test
+ void boltInTransactionsStatementRunsInAutoCommit() throws Exception {
+ data.attempts = 1;
+ setUpBolt(tx -> {}, 1);
+ when(data.session.run(any(String.class), anyMap())).thenAnswer(invocation
-> emptyResult());
+ String inTransactions = "CALL () { CREATE (:C) } IN TRANSACTIONS";
+ data.cypherStatements.add(1, new CypherStatement(new Object[] {"c"},
inTransactions, Map.of()));
+
+ cypher.batchComplete();
+
+ // CREATE (:A) and CREATE (:B) in a transaction each, around the
auto-commit statement
+ verify(data.session).run(eq(inTransactions), anyMap());
+ verify(data.session, times(2)).executeWrite(any());
+ assertEquals(3, cypher.rows.size());
+ assertArrayEquals(new Object[] {"c"}, cypher.rows.get(1));
+ }
+
+ @Test
+ void boltUnwindInTransactionsRunsInAutoCommit() throws Exception {
+ data.attempts = 1;
+ setUpBolt(tx -> {}, 1);
+ when(data.session.run(any(String.class), anyMap())).thenAnswer(invocation
-> emptyResult());
+ meta.setUsingUnwind(true);
+ data.cypher =
+ "UNWIND $rows AS row CALL (row) { CREATE (:A {id: row.id}) } IN
TRANSACTIONS OF 2 ROWS";
+ data.unwindMapName = "rows";
+ data.unwindList = new ArrayList<>(List.of(Map.of("id", 1), Map.of("id",
2), Map.of("id", 3)));
+ data.outputCount = 3;
+
+ cypher.batchComplete();
+
+ verify(data.session).run(eq(data.cypher), anyMap());
+ verify(data.session, never()).executeWrite(any());
+ assertEquals(3, cypher.getLinesOutput());
+ assertTrue(data.unwindList.isEmpty());
+ }
+
+ @Test
+ void boltUnwindInTransactionsErrorAtTheEndFailsTheBatch() {
+ data.attempts = 1;
+ setUpBolt(tx -> {}, 1);
+ // The statement starts, a later batch of it fails
+ Result failing = emptyResult();
+ when(failing.consume())
+ .thenThrow(new org.neo4j.driver.exceptions.ClientException("Batch 4
failed"));
+ when(data.session.run(any(String.class), anyMap())).thenReturn(failing);
+ unwindInTransactions();
+
+ assertThrows(HopException.class, () -> cypher.batchComplete());
+ assertEquals(0, cypher.getLinesOutput());
+ assertTrue(cypher.rows.isEmpty());
+ }
+
+ @Test
+ void boltAutoCommitRetryOfChangesIsLogged() throws Exception {
+ data.attempts = 2;
+ setUpBolt(tx -> {}, 1);
+ Result failing = emptyResult();
+ when(failing.consume())
+ .thenThrow(new org.neo4j.driver.exceptions.ClientException("Batch 4
failed"));
+ Result succeeding = emptyResult();
+ when(data.session.run(any(String.class), anyMap())).thenReturn(failing,
succeeding);
+ unwindInTransactions();
+
+ cypher.batchComplete();
+
+ verify(data.session, times(2)).run(eq(data.cypher), anyMap());
+ verify(helper.iLogChannel).logBasic(contains("the retry applies them
again"));
+ assertEquals(3, cypher.getLinesOutput());
+ }
+
+ private void unwindInTransactions() {
+ meta.setUsingUnwind(true);
+ data.cypher =
+ "UNWIND $rows AS row CALL (row) { CREATE (:A {id: row.id}) } IN
TRANSACTIONS OF 2 ROWS";
+ data.unwindMapName = "rows";
+ data.unwindList = new ArrayList<>(List.of(Map.of("id", 1), Map.of("id",
2), Map.of("id", 3)));
+ data.outputCount = 3;
+ }
+
+ @Test
+ void boltUnwindKeepsTheManagedTransaction() throws Exception {
+ data.attempts = 1;
+ setUpBolt(tx -> {}, 1);
+ meta.setUsingUnwind(true);
+ data.cypher = "UNWIND $rows AS row CREATE (:A {id: row.id})";
+ data.unwindMapName = "rows";
+ data.unwindList = new ArrayList<>(List.of(Map.of("id", 1)));
+ data.outputCount = 1;
+
+ cypher.batchComplete();
+
+ verify(data.session).executeWrite(any());
+ verify(data.session, never()).run(any(String.class), anyMap());
+ }
+
/**
* A Bolt session whose managed transactions call the work the given number
of times, as the
* driver does when it retries a transient error.
