#3944 fix: unit CALL subquery with UNWIND must not multiply outer rows (#3945)

(cherry picked from commit bae159c)

Author: robfrank

engine/src/main/java/com/arcadedb/query/opencypher/executor/steps/SubqueryStep.java

@@ -325,6 +325,10 @@ private int resolveBatchSize(final CommandContext context) {
   /**
    * Executes the inner query with the outer row as the initial seed.
    * The seed row provides variables for the inner query's WITH clause.
+   * <p>
+   * For unit subqueries (no RETURN clause), the inner query is consumed entirely
+   * for side effects (e.g. CREATE, DELETE) and a single empty result is returned
+   * so the outer row count is preserved exactly once per input row.
    */
   private List<Result> executeInnerQuery(final Result outerRow, final CommandContext context) {
     final CypherStatement innerStatement = subqueryClause.getInnerStatement();
@@ -334,6 +338,14 @@ private List<Result> executeInnerQuery(final Result outerRow, final CommandConte
 
     final ResultSet resultSet = innerPlan.executeWithSeedRow(outerRow);
 
+    // Unit subquery: no RETURN clause — consume all inner rows for side effects only.
+    // The outer row count must be preserved (one output row per outer row).
+    if (innerStatement.getReturnClause() == null) {
+      while (resultSet.hasNext())
+        resultSet.next();
+      return List.of(new ResultInternal());
+    }
+
     final List<Result> results = new ArrayList<>();
     while (resultSet.hasNext())
       results.add(resultSet.next());

engine/src/test/java/com/arcadedb/query/opencypher/OpenCypherSubqueryTest.java

@@ -301,4 +301,74 @@ void callSubqueryWithUnionFallsThrough() {
     assertThat(((Number) row2.getProperty("success2")).intValue()).isEqualTo(2);
     assertThat(result2.hasNext()).isFalse();
   }
+
+  /**
+   * Issue #3944: Unit CALL subquery (no RETURN) with inner UNWIND must not multiply outer rows.
+   * Each outer row should appear exactly once regardless of inner UNWIND cardinality.
+   */
+  @Test
+  void unitCallSubqueryWithUnwindPreservesOuterRowCount() {
+    database.getSchema().createVertexType("Person3944");
+    database.getSchema().createVertexType("Clone3944");
+
+    database.transaction(() -> {
+      database.command("opencypher", "CREATE (:Person3944 {name: 'Alice'})");
+      database.command("opencypher", "CREATE (:Person3944 {name: 'Bob'})");
+      database.command("opencypher", "CREATE (:Person3944 {name: 'Charlie'})");
+    });
+
+    database.transaction(() -> {
+      final ResultSet result = database.command("opencypher",
+          "MATCH (p:Person3944) " +
+          "CALL { " +
+          "  WITH p " +
+          "  UNWIND range(1, 2) AS i " +
+          "  CREATE (:Clone3944 {name: p.name, id: i}) " +
+          "} " +
+          "RETURN p.name AS person " +
+          "ORDER BY person");
+
+      final List<Result> rows = new ArrayList<>();
+      while (result.hasNext())
+        rows.add(result.next());
+
+      assertThat(rows).as("Unit CALL subquery must not multiply outer rows by inner UNWIND cardinality").hasSize(3);
+      assertThat((String) rows.get(0).getProperty("person")).isEqualTo("Alice");
+      assertThat((String) rows.get(1).getProperty("person")).isEqualTo("Bob");
+      assertThat((String) rows.get(2).getProperty("person")).isEqualTo("Charlie");
+    });
+  }
+
+  /**
+   * Issue #3944 control: Unit CALL subquery without UNWIND should still work correctly (one row per outer row).
+   */
+  @Test
+  void unitCallSubqueryWithoutUnwindPreservesOuterRowCount() {
+    database.getSchema().createVertexType("Person3944b");
+    database.getSchema().createVertexType("Clone3944b");
+
+    database.transaction(() -> {
+      database.command("opencypher", "CREATE (:Person3944b {name: 'Alice'})");
+      database.command("opencypher", "CREATE (:Person3944b {name: 'Bob'})");
+    });
+
+    database.transaction(() -> {
+      final ResultSet result = database.command("opencypher",
+          "MATCH (p:Person3944b) " +
+          "CALL { " +
+          "  WITH p " +
+          "  CREATE (:Clone3944b {name: p.name}) " +
+          "} " +
+          "RETURN p.name AS person " +
+          "ORDER BY person");
+
+      final List<Result> rows = new ArrayList<>();
+      while (result.hasNext())
+        rows.add(result.next());
+
+      assertThat(rows).as("Unit CALL subquery must produce exactly one row per outer row").hasSize(2);
+      assertThat((String) rows.get(0).getProperty("person")).isEqualTo("Alice");
+      assertThat((String) rows.get(1).getProperty("person")).isEqualTo("Bob");
+    });
+  }
 }

grpcw/src/main/java/com/arcadedb/server/grpc/ArcadeDbGrpcService.java

@@ -1726,10 +1726,10 @@ public void onNext(final GraphBatchChunk chunk) {
           }
         } catch (final Exception e) {
           errorSent[0] = true;
+          // Null batchRef so onCompleted skips processing; skip closeQuietly to avoid blocking the
+          // gRPC thread via async.waitCompletion() (buffered edges have no open transaction).
+          batchRef.set(null);
           resp.onError(Status.INTERNAL.withDescription("graphBatchLoad: " + e.getMessage()).asException());
-          // Note: closeQuietly may flush/commit partial data. GraphBatch has no rollback path by design
-          // (same as the HTTP batch endpoint). Callers should treat batch loading as non-atomic.
-          closeQuietly(batchRef.get());
           return;
         } finally {
           if (!cancelled.get() && !errorSent[0])
@@ -1739,11 +1739,13 @@ public void onNext(final GraphBatchChunk chunk) {
 
       @Override
       public void onError(final Throwable t) {
-        closeQuietly(batchRef.get());
+        closeQuietly(batchRef.getAndSet(null));
       }
 
       @Override
       public void onCompleted() {
+        if (errorSent[0])
+          return;
         try {
           final GraphBatch batch = batchRef.get();
           if (batch == null) {