fix airbytevalue serialize

airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/util/JsonUtils.kt

@@ -13,6 +13,8 @@ import com.fasterxml.jackson.databind.ObjectMapper
 import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule
 import com.fasterxml.jackson.module.afterburner.AfterburnerModule
 import com.fasterxml.jackson.module.kotlin.registerKotlinModule
+import io.airbyte.cdk.load.data.NullValue
+import io.airbyte.cdk.load.data.ObjectValue
 import java.io.InputStream
 
 object Jsons : ObjectMapper() {
@@ -35,7 +37,15 @@ object Jsons : ObjectMapper() {
 }
 
 fun <T> T.serializeToString(): String {
-    return Jsons.writeValueAsString(this)
+    return when (this) {
+        // Handle null specially - otherwise jackson sees it as an empty object
+        // and returns "{}"
+        is NullValue -> return "null"
+        // no idea why this is necessary, but jackson throws an exception if you try to serialize
+        // the ObjectValue directly
+        is ObjectValue -> return Jsons.writeValueAsString(this.values)
+        else -> Jsons.writeValueAsString(this)
+    }
 }
 
 fun <T> InputStream.readIntoClass(klass: Class<T>): T =
@@ -54,5 +64,10 @@ fun <T> String.deserializeToClass(klass: Class<T>): T {
 }
 
 fun Any.serializeToJsonBytes(): ByteArray {
-    return Jsons.writeValueAsBytes(this)
+    // same special cases as serializeToString()
+    return when (this) {
+        is NullValue -> return "null".toByteArray(Charsets.UTF_8)
+        is ObjectValue -> return Jsons.writeValueAsBytes(this.values)
+        else -> Jsons.writeValueAsBytes(this)
+    }
 }