fix: use an aggregate to make sure subquery has a result

Author: ali-ince

sink/src/main/kotlin/org/neo4j/connectors/kafka/sink/strategy/cdc/apoc/ApocCdcHandler.kt

@@ -105,7 +105,8 @@ abstract class ApocCdcHandler(
   }
 
   private fun batchedStatement(topic: String, partition: Int, events: List<MessageToEvent>): Query {
-    val termination = if (neo4j.version >= Neo4jVersion(5, 19, 0)) "FINISH" else "RETURN 1"
+    val termination =
+        if (neo4j.version >= Neo4jVersion(5, 19, 0)) "FINISH" else "RETURN COUNT(1) AS total"
 
     val query =
         if (eosOffsetLabel.isNotBlank()) {
@@ -118,15 +119,15 @@ abstract class ApocCdcHandler(
             )
             appendLine("WITH k, ${EVENT} WHERE ${EVENT}.offset > k.offset")
             appendLine("WITH k, ${EVENT} ORDER BY ${EVENT}.offset ASC")
-            appendCallSubquery(termination)
+            appendCallSubquery()
             appendLine("WITH k, max(${EVENT}.offset) AS newOffset SET k.offset = newOffset")
             append(termination)
           }
         } else {
           buildString {
             appendLine("UNWIND \$events AS ${EVENT}")
             appendLine("WITH ${EVENT} ORDER BY ${EVENT}.offset ASC")
-            appendCallSubquery(termination)
+            appendCallSubquery()
             append(termination)
           }
         }
@@ -153,11 +154,13 @@ abstract class ApocCdcHandler(
     )
   }
 
-  private fun StringBuilder.appendCallSubquery(termination: String) {
+  private fun StringBuilder.appendCallSubquery() {
     if (canIUse(Cypher.callSubqueryWithVariableScopeClause()).withNeo4j(neo4j))
         appendLine("CALL (${EVENT}) {")
     else appendLine("CALL { WITH ${EVENT}")
-    appendLine("  CALL apoc.cypher.doIt(${EVENT}.stmt, ${EVENT}.params) YIELD value $termination")
+    appendLine(
+        "  CALL apoc.cypher.doIt(${EVENT}.stmt, ${EVENT}.params) YIELD value RETURN COUNT(1) AS total"
+    )
     appendLine("}")
   }
 

sink/src/test/kotlin/org/neo4j/connectors/kafka/sink/strategy/cdc/apoc/ApocCdcSchemaHandlerTest.kt

@@ -54,7 +54,7 @@ class ApocCdcSchemaHandlerWithEOSTest :
         |WITH k, e WHERE e.offset > k.offset
         |WITH k, e ORDER BY e.offset ASC
         |CALL (e) {
-        |  CALL apoc.cypher.doIt(e.stmt, e.params) YIELD value FINISH
+        |  CALL apoc.cypher.doIt(e.stmt, e.params) YIELD value RETURN COUNT(1) AS total
         |}
         |WITH k, max(e.offset) AS newOffset SET k.offset = newOffset
         |FINISH
@@ -69,7 +69,7 @@ class ApocCdcSchemaHandlerWithoutEOSTest :
         |UNWIND ${'$'}events AS e
         |WITH e ORDER BY e.offset ASC
         |CALL (e) {
-        |  CALL apoc.cypher.doIt(e.stmt, e.params) YIELD value FINISH
+        |  CALL apoc.cypher.doIt(e.stmt, e.params) YIELD value RETURN COUNT(1) AS total
         |}
         |FINISH
         """

sink/src/test/kotlin/org/neo4j/connectors/kafka/sink/strategy/cdc/apoc/ApocCdcSourceIdHandlerTest.kt

@@ -49,7 +49,7 @@ class ApocCdcSourceIdHandlerWithEOSTest :
         |WITH k, e WHERE e.offset > k.offset
         |WITH k, e ORDER BY e.offset ASC
         |CALL (e) {
-        |  CALL apoc.cypher.doIt(e.stmt, e.params) YIELD value FINISH
+        |  CALL apoc.cypher.doIt(e.stmt, e.params) YIELD value RETURN COUNT(1) AS total
         |}
         |WITH k, max(e.offset) AS newOffset SET k.offset = newOffset
         |FINISH
@@ -64,7 +64,7 @@ class ApocCdcSourceIdHandlerWithoutEOSTest :
         |UNWIND ${'$'}events AS e
         |WITH e ORDER BY e.offset ASC
         |CALL (e) {
-        |  CALL apoc.cypher.doIt(e.stmt, e.params) YIELD value FINISH
+        |  CALL apoc.cypher.doIt(e.stmt, e.params) YIELD value RETURN COUNT(1) AS total
         |}
         |FINISH
         """