fix: try to be idempotent about relationship creations (#499)

Author: ali-ince

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

@@ -114,7 +114,7 @@ class ApocCdcSchemaHandler(val topic: String, neo4j: Neo4j, batchSize: Int) :
     val (startMatchLabels, startMatchProperties) = buildMatchLabelsAndProperties(event.start.keys)
     val (endMatchLabels, endMatchProperties) = buildMatchLabelsAndProperties(event.end.keys)
     val (relMatchType, relMatchProperties) =
-        buildMatchLabelsAndProperties(event.type, event.keys, null)
+        buildMatchLabelsAndProperties(event.type, event.keys, event.after.properties)
 
     return CdcRelationshipData(
         EntityOperation.CREATE,

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

@@ -70,7 +70,7 @@ data class CdcNodeData(
         "MERGE (n$matchLabels$matchProps) SET n += ${'$'}$EVENT.setProperties SET n:\$(${'$'}$EVENT.addLabels) REMOVE n:\$(${'$'}$EVENT.removeLabels)"
       }
       EntityOperation.DELETE -> {
-        "MATCH (n$matchLabels$matchProps) DETACH DELETE n"
+        "MATCH (n$matchLabels$matchProps) DELETE n"
       }
     }
   }
@@ -133,11 +133,7 @@ data class CdcRelationshipData(
 
     return when (operation) {
       EntityOperation.CREATE -> {
-        if (!hasKeys) {
-          "MATCH (start$startMatchLabels$startMatchProps) MATCH (end$endMatchLabels$endMatchProps) CREATE (start)-[r:$matchType$matchProps]->(end) SET r += ${'$'}$EVENT.setProperties"
-        } else {
-          "MATCH (start$startMatchLabels$startMatchProps) MATCH (end$endMatchLabels$endMatchProps) MERGE (start)-[r:$matchType$matchProps]->(end) SET r += ${'$'}$EVENT.setProperties"
-        }
+        "MATCH (start$startMatchLabels$startMatchProps) MATCH (end$endMatchLabels$endMatchProps) MERGE (start)-[r:$matchType$matchProps]->(end) SET r += ${'$'}$EVENT.setProperties"
       }
       EntityOperation.UPDATE -> {
         if (!hasKeys) {

sink/src/main/kotlin/org/neo4j/connectors/kafka/sink/strategy/cdc/batch/CdcData.kt

@@ -66,7 +66,7 @@ data class CdcNodeData(
         "MERGE (n:\$($EVENT.matchLabels) {$matchProps}) SET n += $EVENT.setProperties SET n:\$($EVENT.addLabels) REMOVE n:\$($EVENT.removeLabels)"
       }
       EntityOperation.DELETE -> {
-        "MATCH (n:\$($EVENT.matchLabels) {$matchProps}) DETACH DELETE n"
+        "MATCH (n:\$($EVENT.matchLabels) {$matchProps}) DELETE n"
       }
     }
   }

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

@@ -478,7 +478,7 @@ class ApocCdcSchemaHandlerTest {
                                 listOf(
                                     mapOf(
                                         "stmt" to
-                                            "MATCH (n:Person {name: \$e.matchProperties.name, surname: \$e.matchProperties.surname}) DETACH DELETE n",
+                                            "MATCH (n:Person {name: \$e.matchProperties.name, surname: \$e.matchProperties.surname}) DELETE n",
                                         "params" to
                                             mapOf(
                                                 "e" to
@@ -543,7 +543,7 @@ class ApocCdcSchemaHandlerTest {
                                 listOf(
                                     mapOf(
                                         "stmt" to
-                                            "MATCH (start:Person {id: \$e.start.matchProperties.id}) MATCH (end:Person {id: \$e.end.matchProperties.id}) CREATE (start)-[r:KNOWS]->(end) SET r += \$e.setProperties",
+                                            "MATCH (start:Person {id: \$e.start.matchProperties.id}) MATCH (end:Person {id: \$e.end.matchProperties.id}) MERGE (start)-[r:KNOWS {since: \$e.matchProperties.since}]->(end) SET r += \$e.setProperties",
                                         "params" to
                                             mapOf(
                                                 "e" to
@@ -559,7 +559,9 @@ class ApocCdcSchemaHandlerTest {
                                                                     mapOf("id" to 2L)
                                                             ),
                                                         "matchProperties" to
-                                                            emptyMap<String, Any>(),
+                                                            mapOf(
+                                                                "since" to LocalDate.of(2000, 1, 1)
+                                                            ),
                                                         "setProperties" to
                                                             mapOf(
                                                                 "since" to LocalDate.of(2000, 1, 1)
@@ -1191,7 +1193,7 @@ class ApocCdcSchemaHandlerTest {
                                 listOf(
                                     mapOf(
                                         "stmt" to
-                                            "MATCH (n:Person {name: \$e.matchProperties.name}) DETACH DELETE n",
+                                            "MATCH (n:Person {name: \$e.matchProperties.name}) DELETE n",
                                         "params" to
                                             mapOf(
                                                 "e" to

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

@@ -434,7 +434,7 @@ class ApocCdcSourceIdHandlerTest {
                                 listOf(
                                     mapOf(
                                         "stmt" to
-                                            "MATCH (n:SourceEvent {sourceElementId: \$e.matchProperties.sourceElementId}) DETACH DELETE n",
+                                            "MATCH (n:SourceEvent {sourceElementId: \$e.matchProperties.sourceElementId}) DELETE n",
                                         "params" to
                                             mapOf(
                                                 "e" to

sink/src/test/kotlin/org/neo4j/connectors/kafka/sink/strategy/cdc/batch/BatchedCdcSchemaHandlerTest.kt

@@ -453,7 +453,7 @@ class BatchedCdcSchemaHandlerTest {
                     Query(
                         "CYPHER 25 UNWIND \$events AS e CALL (e) { WHEN e.q = \$q0 THEN " +
                             "MATCH (n:\$(e.matchLabels) {name: e.matchProperties.name, surname: e.matchProperties.surname}) " +
-                            "DETACH DELETE n } FINISH",
+                            "DELETE n } FINISH",
                         mapOf(
                             "q0" to 0,
                             "events" to
@@ -1139,7 +1139,7 @@ class BatchedCdcSchemaHandlerTest {
                     null,
                     listOf(sinkMessage),
                     Query(
-                        "CYPHER 25 UNWIND \$events AS e CALL (e) { WHEN e.q = \$q0 THEN MATCH (n:\$(e.matchLabels) {name: e.matchProperties.name}) DETACH DELETE n } FINISH",
+                        "CYPHER 25 UNWIND \$events AS e CALL (e) { WHEN e.q = \$q0 THEN MATCH (n:\$(e.matchLabels) {name: e.matchProperties.name}) DELETE n } FINISH",
                         mapOf(
                             "q0" to 0,
                             "events" to

sink/src/test/kotlin/org/neo4j/connectors/kafka/sink/strategy/cdc/batch/BatchedCdcSourceIdHandlerTest.kt

@@ -400,7 +400,7 @@ class BatchedCdcSourceIdHandlerTest {
                     Query(
                         "CYPHER 25 UNWIND \$events AS e CALL (e) { WHEN e.q = \$q0 THEN " +
                             "MATCH (n:\$(e.matchLabels) {sourceElementId: e.matchProperties.sourceElementId}) " +
-                            "DETACH DELETE n } FINISH",
+                            "DELETE n } FINISH",
                         mapOf(
                             "q0" to 0,
                             "events" to