make it mutable

Author: edgao

airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/data/AirbyteValue.kt

@@ -166,12 +166,12 @@ private class ObjectValueSerializer : JsonSerializer<ObjectValue>() {
  * @property name Field name
  * @property fieldCategory [FieldCategory] of the field
  */
-data class EnrichedAirbyteValue(
-    val value: AirbyteValue,
+class EnrichedAirbyteValue(
+    var value: AirbyteValue,
     val type: AirbyteType,
     val name: String,
     val fieldCategory: FieldCategory,
-    val changes: List<Meta.Change> = emptyList()
+    val changes: MutableList<Meta.Change> = mutableListOf()
 ) {
     init {
         require(name.isNotBlank()) { "Field name cannot be blank" }
@@ -183,16 +183,16 @@ data class EnrichedAirbyteValue(
      * @param reason The [Reason] for nullification, defaults to DESTINATION_SERIALIZATION_ERROR
      * @return A new [EnrichedAirbyteValue] with a null value and an additional change record
      */
-    fun toNullified(reason: Reason = Reason.DESTINATION_SERIALIZATION_ERROR): EnrichedAirbyteValue {
+    fun nullify(reason: Reason = Reason.DESTINATION_SERIALIZATION_ERROR) {
         val nullChange =
             Meta.Change(
                 field = name,
                 change = AirbyteRecordMessageMetaChange.Change.NULLED,
                 reason = reason
             )
 
-        // Return a copy with null value and the new change added to the changes list
-        return copy(value = NullValue, changes = changes + nullChange)
+        value = NullValue
+        changes.add(nullChange)
     }
 
     /**
@@ -202,10 +202,10 @@ data class EnrichedAirbyteValue(
      * @param newValue The new (truncated) value to use
      * @return A new [EnrichedAirbyteValue] with the truncated value and an additional change record
      */
-    fun toTruncated(
+    fun truncate(
         reason: Reason = Reason.DESTINATION_RECORD_SIZE_LIMITATION,
         newValue: AirbyteValue
-    ): EnrichedAirbyteValue {
+    ) {
         val truncateChange =
             Meta.Change(
                 field = name,
@@ -214,7 +214,8 @@ data class EnrichedAirbyteValue(
             )
 
         // Return a copy with null value and the new change added to the changes list
-        return copy(value = newValue, changes = changes + truncateChange)
+        value = newValue
+        changes.add(truncateChange)
     }
 }
 

airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/message/DestinationMessage.kt

@@ -41,7 +41,6 @@ import java.math.BigInteger
 import java.time.Instant
 import java.time.OffsetDateTime
 import java.time.ZoneOffset
-import java.util.*
 
 /**
  * Internal representation of destination messages. These are intended to be specialized for
@@ -197,6 +196,8 @@ data class EnrichedDestinationRecordAirbyteValue(
             TODO("all the other efields"),
         )
     }
+
+    val allFields = declaredFields + airbyteMetaFields
 }
 
 data class DestinationRecordRaw(
@@ -246,26 +247,21 @@ data class DestinationRecordRaw(
                                 "Field '$fieldName' exists in schema keys but not in properties"
                             )
 
-                    val airbyteValue =
-                        if (fieldValue.isNull) NullValue else fieldValue.toAirbyteValue()
-                    declaredFields[fieldName] =
+                    val enrichedValue =
                         EnrichedAirbyteValue(
-                                value = airbyteValue,
-                                type = fieldType,
-                                name = fieldName,
-                                fieldCategory = FieldCategory.CLIENT_DATA,
-                            )
-                            .let {
-                                val coercedValue = AirbyteValueCoercer.coerce(airbyteValue, schema)
-                                if (coercedValue == null) {
-                                    it.toNullified(
-                                        AirbyteRecordMessageMetaChange.Reason
-                                            .DESTINATION_SERIALIZATION_ERROR
-                                    )
-                                } else {
-                                    it
-                                }
-                            }
+                            value = NullValue,
+                            type = fieldType,
+                            name = fieldName,
+                            fieldCategory = FieldCategory.CLIENT_DATA,
+                        )
+                    AirbyteValueCoercer.coerce(fieldValue.toAirbyteValue(), schema)?.let {
+                        enrichedValue.value = it
+                    }
+                        ?: enrichedValue.nullify(
+                            AirbyteRecordMessageMetaChange.Reason.DESTINATION_SERIALIZATION_ERROR
+                        )
+
+                    declaredFields[fieldName] = enrichedValue
                 }
                 else -> {
                     // Undeclared field (not in schema)