fix: handle heterogeneous object list (#349)

Fixes #343

This applies only to the source connector, with the query strategy.

Note that maps with heterogeneous value types will still be
encoded as structs.

Some converters (like Protobuf) do not support map message types by default.

Author: fbiville

common/src/main/resources/neo4j-source-configuration.properties

@@ -14,7 +14,7 @@
 neo4j.source-strategy=Type: Enum<QUERY, CDC>;\nDescription: Source strategy for this connector.
 neo4j.batch-size=Type: Integer;\nDescription: Maximum number of change events to publish for each poll cycle.
 neo4j.start-from=Type: Enum<EARLIEST, NOW, USER_PROVIDED>;\nDescription: A time anchor to start streaming from.
-neo4j.start-from.value=Type: STRING|LONG;\nDescription: Custom value to use as a starting offset. \
+neo4j.start-from.value=Type: String|Long;\nDescription: Custom value to use as a starting offset. \
   Used once during the initial run of the connector, and will be ignored if there is an offset stored in Kafka Connect.
 neo4j.ignore-stored-offset=Type: Boolean;\nDescription: Whether to ignore any offset value retrieved from the offset storage saved by a previous run.
 neo4j.query=Type: String;\nDescription: Cypher query to use for gathering changes. \
@@ -24,6 +24,8 @@ neo4j.query.streaming-property=Type: String;\nDescription: Property name that is
 neo4j.query.poll-duration=Type: String;\nDescription: Maximum amount of time Kafka Connect poll request will wait for a change to be received from the database (valid units are: `ms`, `s`, `m`, `h` and `d`; default unit is `s`).
 neo4j.query.poll-interval=Type: String;\nDescription: Interval in which the query is executed during `neo4j.query.poll-duration` timespan (valid units are: `ms`, `s`, `m`, `h` and `d`; default unit is `s`).
 neo4j.query.topic=Type: String;\nDescription: Kafka topic to publish change events gathered through provided query.
+neo4j.query.force-maps-as-struct=Type: Boolean;\nDescription: Whether the schema of maps with homogeneous value types is encoded as struct or map (default is: `true`). \
+  Not all converters support map schemas.
 neo4j.query.timeout=Type: Duration;\nDescription: Maximum amount of time query is allowed to run (valid units are: `ms`, `s`, `m`, `h` and `d`; default unit is `s`).
 neo4j.cdc.poll-duration=Type: Duration;\nDescription: Maximum amount of time Kafka Connect poll request will wait for a change to be received from the database (valid units are: `ms`, `s`, `m`, `h` and `d`; default unit is `s`).
 neo4j.cdc.poll-interval=Type: Duration;\nDescription: The interval in which the database will be polled for changes (valid units are: `ms`, `s`, `m`, `h` and `d`; default unit is `s`).

source-connector/src/test/kotlin/org/neo4j/connectors/kafka/source/Neo4jSourceQueryIT.kt

@@ -27,6 +27,7 @@ import org.neo4j.connectors.kafka.testing.TestSupport.runTest
 import org.neo4j.connectors.kafka.testing.assertions.TopicVerifier
 import org.neo4j.connectors.kafka.testing.format.KafkaConverter.AVRO
 import org.neo4j.connectors.kafka.testing.format.KafkaConverter.JSON_EMBEDDED
+import org.neo4j.connectors.kafka.testing.format.KafkaConverter.JSON_RAW
 import org.neo4j.connectors.kafka.testing.format.KafkaConverter.JSON_SCHEMA
 import org.neo4j.connectors.kafka.testing.format.KafkaConverter.PROTOBUF
 import org.neo4j.connectors.kafka.testing.format.KeyValueConverter
@@ -39,7 +40,7 @@ import org.neo4j.driver.Session
 abstract class Neo4jSourceQueryIT {
 
   companion object {
-    private const val TOPIC = "neo4j-source-topic"
+    const val TOPIC = "neo4j-source-topic"
   }
 
   @Neo4jSource(
@@ -219,6 +220,54 @@ class Neo4jSourceJsonEmbeddedExtendedIT : Neo4jSourceQueryIT()
 @KeyValueConverter(key = JSON_EMBEDDED, value = JSON_EMBEDDED, payloadMode = PayloadMode.COMPACT)
 class Neo4jSourceJsonEmbeddedCompactIT : Neo4jSourceQueryIT()
 
+// it doesn't make sense to add EXTENDED mode for JSON_RAW since it's not a schema supporting
+// converter
+@KeyValueConverter(key = JSON_RAW, value = JSON_RAW, payloadMode = PayloadMode.COMPACT)
+class Neo4jSourceJsonRawCompactIT : Neo4jSourceQueryIT() {
+
+  @Neo4jSource(
+      topic = TOPIC,
+      strategy = SourceStrategy.QUERY,
+      streamingProperty = "timestamp",
+      startFrom = "USER_PROVIDED",
+      startFromValue = "1704067200000", // 2024-01-01T00:00:00
+      query =
+          "WITH {id: 'ROOT_ID', list: [{ property1: 'value1' }, { property2: 'value2' }]} AS data RETURN data, data.id AS guid, dateTime().epochMillis AS timestamp")
+  @Test
+  fun `serializes list of heterogeneous objects as map by default`(
+      @TopicConsumer(topic = TOPIC, offset = "earliest") consumer: ConvertingKafkaConsumer
+  ) = runTest {
+    TopicVerifier.createForMap(consumer)
+        .assertMessageValue { value ->
+          val list = (value["data"] as Map<*, *>)["list"]
+          list shouldBe
+              mapOf("e0" to mapOf("property1" to "value1"), "e1" to mapOf("property2" to "value2"))
+        }
+        .verifyWithin(Duration.ofSeconds(30))
+  }
+
+  @Neo4jSource(
+      topic = TOPIC,
+      strategy = SourceStrategy.QUERY,
+      streamingProperty = "timestamp",
+      startFrom = "USER_PROVIDED",
+      startFromValue = "1704067200000", // 2024-01-01T00:00:00
+      forceMapsAsStruct = false,
+      query =
+          "WITH {id: 'ROOT_ID', list: [{ property1: 'value1' }, { property2: 'value2' }]} AS data RETURN data, data.id AS guid, dateTime().epochMillis AS timestamp")
+  @Test
+  fun `serializes list of heterogeneous objects as list when not forcing structs for map values with homogeneous value types`(
+      @TopicConsumer(topic = TOPIC, offset = "earliest") consumer: ConvertingKafkaConsumer
+  ) = runTest {
+    TopicVerifier.createForMap(consumer)
+        .assertMessageValue { value ->
+          val list = (value["data"] as Map<*, *>)["list"]
+          list shouldBe listOf(mapOf("property1" to "value1"), mapOf("property2" to "value2"))
+        }
+        .verifyWithin(Duration.ofSeconds(30))
+  }
+}
+
 @KeyValueConverter(key = PROTOBUF, value = PROTOBUF, payloadMode = PayloadMode.EXTENDED)
 class Neo4jSourceProtobufExtendedIT : Neo4jSourceQueryIT()
 

source/src/main/kotlin/org/neo4j/connectors/kafka/source/Neo4jQueryTask.kt

@@ -97,7 +97,10 @@ class Neo4jQueryTask : SourceTask() {
     val recordAsMap = record.asMap()
     val schema =
         DynamicTypes.toConnectSchema(
-            config.payloadMode, recordAsMap, optional = true, forceMapsAsStruct = true)
+            config.payloadMode,
+            recordAsMap,
+            optional = true,
+            forceMapsAsStruct = config.forceMapsAsStruct)
     val value = DynamicTypes.toConnectValue(schema, recordAsMap)
 
     return SourceRecord(

source/src/main/kotlin/org/neo4j/connectors/kafka/source/SourceConfiguration.kt

@@ -58,6 +58,9 @@ enum class StartFrom {
 class SourceConfiguration(originals: Map<*, *>) :
     Neo4jConfiguration(config(), originals, ConnectorType.SOURCE) {
 
+  val forceMapsAsStruct: Boolean
+    get(): Boolean = getBoolean(QUERY_FORCE_MAPS_AS_STRUCT)
+
   val startFrom
     get(): StartFrom = StartFrom.valueOf(getString(START_FROM))
 
@@ -456,6 +459,7 @@ class SourceConfiguration(originals: Map<*, *>) :
     const val QUERY_POLL_DURATION = "neo4j.query.poll-duration"
     const val QUERY_TIMEOUT = "neo4j.query.timeout"
     const val QUERY_TOPIC = "neo4j.query.topic"
+    const val QUERY_FORCE_MAPS_AS_STRUCT = "neo4j.query.force-maps-as-struct"
     const val CDC_POLL_INTERVAL = "neo4j.cdc.poll-interval"
     const val CDC_POLL_DURATION = "neo4j.cdc.poll-duration"
     const val PAYLOAD_MODE = "neo4j.payload-mode"
@@ -489,6 +493,8 @@ class SourceConfiguration(originals: Map<*, *>) :
     private val DEFAULT_QUERY_POLL_DURATION = 5.seconds
     private const val DEFAULT_BATCH_SIZE = 1000
     private val DEFAULT_QUERY_TIMEOUT = 0.seconds
+    private const val DEFAULT_QUERY_FORCE_MAPS_AS_STRUCT = true
+
     private val DEFAULT_CDC_POLL_INTERVAL = 1.seconds
     private val DEFAULT_CDC_POLL_DURATION = 5.seconds
     private const val DEFAULT_STREAMING_PROPERTY = "timestamp"
@@ -643,6 +649,14 @@ class SourceConfiguration(originals: Map<*, *>) :
                       Recommenders.visibleIf(STRATEGY, Predicate.isEqual(SourceType.QUERY.name))
                   validator = Validators.pattern(SIMPLE_DURATION_PATTERN)
                 })
+            .define(
+                ConfigKeyBuilder.of(QUERY_FORCE_MAPS_AS_STRUCT, ConfigDef.Type.BOOLEAN) {
+                  importance = ConfigDef.Importance.LOW
+                  defaultValue = DEFAULT_QUERY_FORCE_MAPS_AS_STRUCT
+                  group = Groups.CONNECTOR_ADVANCED.title
+                  recommender =
+                      Recommenders.visibleIf(STRATEGY, Predicate.isEqual(SourceType.QUERY.name))
+                })
             .define(
                 ConfigKeyBuilder.of(BATCH_SIZE, ConfigDef.Type.INT) {
                   importance = ConfigDef.Importance.MEDIUM

testing/src/main/kotlin/org/neo4j/connectors/kafka/testing/assertions/TopicVerifier.kt

@@ -159,14 +159,23 @@ class TopicVerifier<K, V>(
       topic: String,
       value: ByteArray?,
   ): Any? {
-    return when (val sourceValue = converter.toConnectData(topic, value).value()) {
+    val connectData = converter.toConnectData(topic, value)
+    return when (val sourceValue = connectData.value()) {
       is Struct ->
           when (assertionClass) {
             ChangeEvent::class.java -> sourceValue.toChangeEvent()
             Map::class.java ->
                 DynamicTypes.fromConnectValue(sourceValue.schema(), sourceValue, true)
             else -> sourceValue as V
           }
+      is Map<*, *> -> {
+        val schema = connectData.schema()
+        if (schema == null) {
+          sourceValue
+        } else {
+          DynamicTypes.fromConnectValue(schema, sourceValue, true)
+        }
+      }
       else -> sourceValue
     }
   }
@@ -179,6 +188,7 @@ class TopicVerifier<K, V>(
       when (consumer.keyConverter) {
         KafkaConverter.JSON_EMBEDDED ->
             keyConverter.configure(mapOf("schemas.enable" to true), true)
+        KafkaConverter.JSON_RAW -> keyConverter.configure(mapOf("schemas.enable" to false), true)
         else ->
             keyConverter.configure(
                 mapOf(
@@ -193,6 +203,7 @@ class TopicVerifier<K, V>(
       when (consumer.valueConverter) {
         KafkaConverter.JSON_EMBEDDED ->
             valueConverter.configure(mapOf("schemas.enable" to true), false)
+        KafkaConverter.JSON_RAW -> valueConverter.configure(mapOf("schemas.enable" to false), false)
         else ->
             valueConverter.configure(
                 mapOf(

testing/src/main/kotlin/org/neo4j/connectors/kafka/testing/format/KafkaConverter.kt

@@ -76,6 +76,7 @@ enum class KafkaConverter(
       converterProvider = { JsonConverter() },
       serializerClass = KafkaJsonSerializer::class.java,
       testShimSerializer = JsonRawSerializer,
+      supportsSchemaRegistry = false,
       additionalProperties = JSON_RAW_OPTIONS,
   ),
   PROTOBUF(

testing/src/main/kotlin/org/neo4j/connectors/kafka/testing/source/Neo4jSource.kt

@@ -40,6 +40,7 @@ annotation class Neo4jSource(
     val topic: String = "",
     val streamingProperty: String = "",
     val query: String = "",
+    val forceMapsAsStruct: Boolean = true,
 
     // CDC strategy
     val cdc: CdcSource = CdcSource()

testing/src/main/kotlin/org/neo4j/connectors/kafka/testing/source/Neo4jSourceExtension.kt

@@ -184,6 +184,7 @@ internal class Neo4jSourceExtension(
             startFrom = sourceAnnotation.startFrom,
             startFromValue = sourceAnnotation.startFromValue,
             query = sourceAnnotation.query,
+            forceMapsAsStruct = sourceAnnotation.forceMapsAsStruct,
             strategy = sourceAnnotation.strategy,
             keyConverter = keyValueConverterResolver.resolveKeyConverter(context),
             valueConverter = keyValueConverterResolver.resolveValueConverter(context),

testing/src/main/kotlin/org/neo4j/connectors/kafka/testing/source/Neo4jSourceRegistration.kt

@@ -49,7 +49,8 @@ internal class Neo4jSourceRegistration(
     cdcMetadata: Map<String, List<Map<String, String>>>,
     cdcKeySerializations: Map<String, String>,
     cdcValueSerializations: Map<String, String>,
-    payloadMode: PayloadMode
+    payloadMode: PayloadMode,
+    forceMapsAsStruct: Boolean
 ) {
 
   val name: String = randomizedName("Neo4jSourceConnector")
@@ -92,6 +93,7 @@ internal class Neo4jSourceRegistration(
       if (strategy == QUERY) {
         put("neo4j.query.topic", topic)
         put("neo4j.query", query)
+        put("neo4j.query.force-maps-as-struct", forceMapsAsStruct)
         put("neo4j.query.streaming-property", streamingProperty)
         put("neo4j.query.poll-interval", "${pollInterval.toMillis()}ms")
         put("neo4j.query.poll-duration", "${pollDuration.toMillis()}ms")