coerce immediately + kill meta fields

Author: edgao

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

@@ -169,9 +169,9 @@ private class ObjectValueSerializer : JsonSerializer<ObjectValue>() {
 data class EnrichedAirbyteValue(
     val value: AirbyteValue,
     val type: AirbyteType,
-    val changes: List<Meta.Change> = emptyList(),
     val name: String,
-    val fieldCategory: FieldCategory
+    val fieldCategory: FieldCategory,
+    val changes: List<Meta.Change> = emptyList()
 ) {
     init {
         require(name.isNotBlank()) { "Field name cannot be blank" }

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

@@ -25,6 +25,30 @@ import java.time.format.DateTimeFormatter
  * common-sense conversions among numeric types, as well as upcasting any value to StringValue.
  */
 object AirbyteValueCoercer {
+    fun coerce(value: AirbyteValue, type: AirbyteType): AirbyteValue? =
+        when (type) {
+            BooleanType -> coerceBoolean(value)
+            DateType -> coerceDate(value)
+            IntegerType -> coerceInt(value)
+            NumberType -> coerceNumber(value)
+            StringType -> coerceString(value)
+            TimeTypeWithTimezone -> coerceTimeTz(value)
+            TimeTypeWithoutTimezone -> coerceTimeNtz(value)
+            TimestampTypeWithTimezone -> coerceTimestampTz(value)
+            TimestampTypeWithoutTimezone -> coerceTimestampNtz(value)
+            is ArrayType,
+            ArrayTypeWithoutSchema -> coerceArray(value)
+            is ObjectType,
+            ObjectTypeWithEmptySchema,
+            ObjectTypeWithoutSchema -> coerceObject(value)
+
+            // Don't touch unions, just pass it through
+            is UnionType -> value
+            // Similarly, if we don't know what type it's supposed to be,
+            // leave it unchanged.
+            is UnknownType -> value
+        }
+
     fun coerceBoolean(value: AirbyteValue): BooleanValue? = requireType<BooleanValue>(value)
 
     fun coerceInt(value: AirbyteValue): IntegerValue? =

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

@@ -10,13 +10,15 @@ import io.airbyte.cdk.load.command.DestinationCatalog
 import io.airbyte.cdk.load.command.DestinationStream
 import io.airbyte.cdk.load.data.AirbyteType
 import io.airbyte.cdk.load.data.AirbyteValue
+import io.airbyte.cdk.load.data.AirbyteValueCoercer
 import io.airbyte.cdk.load.data.AirbyteValueDeepCoercingMapper
 import io.airbyte.cdk.load.data.EnrichedAirbyteValue
 import io.airbyte.cdk.load.data.FieldCategory
 import io.airbyte.cdk.load.data.IntegerValue
 import io.airbyte.cdk.load.data.NullValue
 import io.airbyte.cdk.load.data.ObjectType
 import io.airbyte.cdk.load.data.StringValue
+import io.airbyte.cdk.load.data.TimestampTypeWithTimezone
 import io.airbyte.cdk.load.data.TimestampWithTimezoneValue
 import io.airbyte.cdk.load.data.json.toAirbyteValue
 import io.airbyte.cdk.load.message.CheckpointMessage.Checkpoint
@@ -36,7 +38,9 @@ import io.airbyte.protocol.models.v0.AirbyteTraceMessage
 import io.micronaut.context.annotation.Value
 import jakarta.inject.Singleton
 import java.math.BigInteger
+import java.time.Instant
 import java.time.OffsetDateTime
+import java.time.ZoneOffset
 import java.util.*
 
 /**
@@ -174,12 +178,26 @@ data class DestinationRecordAirbyteValue(
 data class EnrichedDestinationRecordAirbyteValue(
     val stream: DestinationStream.Descriptor,
     val declaredFields: Map<String, EnrichedAirbyteValue>,
-    val airbyteMetaFields: Map<String, EnrichedAirbyteValue>,
     val undeclaredFields: Map<String, JsonNode>,
     val emittedAtMs: Long,
     val meta: Meta?,
     val serializedSizeBytes: Long = 0L
-)
+) {
+    val airbyteMetaFields: Map<String, EnrichedAirbyteValue> by lazy {
+        mapOf(
+            "_airbyte_extracted_at" to
+                EnrichedAirbyteValue(
+                    TimestampWithTimezoneValue(
+                        OffsetDateTime.ofInstant(Instant.ofEpochMilli(emittedAtMs), ZoneOffset.UTC)
+                    ),
+                    TimestampTypeWithTimezone,
+                    name = "_airbyte_extracted_at",
+                    FieldCategory.EXTRACTED_AT,
+                ),
+            TODO("all the other efields"),
+        )
+    }
+}
 
 data class DestinationRecordRaw(
     val stream: DestinationStream.Descriptor,
@@ -215,7 +233,6 @@ data class DestinationRecordRaw(
             }
 
         val declaredFields = mutableMapOf<String, EnrichedAirbyteValue>()
-        val airbyteMetaFields = mutableMapOf<String, EnrichedAirbyteValue>()
         val undeclaredFields = mutableMapOf<String, JsonNode>()
 
         // Process fields from the raw JSON
@@ -233,11 +250,22 @@ data class DestinationRecordRaw(
                         if (fieldValue.isNull) NullValue else fieldValue.toAirbyteValue()
                     declaredFields[fieldName] =
                         EnrichedAirbyteValue(
-                            value = airbyteValue,
-                            type = fieldType,
-                            name = fieldName,
-                            fieldCategory = FieldCategory.CLIENT_DATA
-                        )
+                                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
+                                }
+                            }
                 }
                 else -> {
                     // Undeclared field (not in schema)
@@ -249,7 +277,6 @@ data class DestinationRecordRaw(
         return EnrichedDestinationRecordAirbyteValue(
             stream = stream,
             declaredFields = declaredFields,
-            airbyteMetaFields = airbyteMetaFields,
             undeclaredFields = undeclaredFields,
             emittedAtMs = rawData.record.emittedAt,
             meta =