major implementation

Author: edgao

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

@@ -5,55 +5,10 @@
 package io.airbyte.cdk.load.data
 
 import io.airbyte.cdk.load.message.Meta
+import io.airbyte.cdk.load.message.Meta.AirbyteMetaFields
 
 class AirbyteTypeToAirbyteTypeWithMeta(private val flatten: Boolean) {
     fun convert(schema: AirbyteType): ObjectType {
-        val properties =
-            linkedMapOf(
-                Meta.COLUMN_NAME_AB_RAW_ID to FieldType(StringType, nullable = false),
-                Meta.COLUMN_NAME_AB_EXTRACTED_AT to FieldType(IntegerType, nullable = false),
-                Meta.COLUMN_NAME_AB_META to
-                    FieldType(
-                        nullable = false,
-                        type =
-                            ObjectType(
-                                linkedMapOf(
-                                    "sync_id" to FieldType(IntegerType, nullable = false),
-                                    "changes" to
-                                        FieldType(
-                                            nullable = false,
-                                            type =
-                                                ArrayType(
-                                                    FieldType(
-                                                        nullable = false,
-                                                        type =
-                                                            ObjectType(
-                                                                linkedMapOf(
-                                                                    "field" to
-                                                                        FieldType(
-                                                                            StringType,
-                                                                            nullable = false
-                                                                        ),
-                                                                    "change" to
-                                                                        FieldType(
-                                                                            StringType,
-                                                                            nullable = false
-                                                                        ),
-                                                                    "reason" to
-                                                                        FieldType(
-                                                                            StringType,
-                                                                            nullable = false
-                                                                        ),
-                                                                )
-                                                            )
-                                                    )
-                                                )
-                                        )
-                                )
-                            )
-                    ),
-                Meta.COLUMN_NAME_AB_GENERATION_ID to FieldType(IntegerType, nullable = false)
-            )
         if (flatten) {
             if (schema is ObjectType) {
                 schema.properties.forEach { (name, field) -> properties[name] = field }
@@ -69,6 +24,20 @@ class AirbyteTypeToAirbyteTypeWithMeta(private val flatten: Boolean) {
         }
         return ObjectType(properties)
     }
+
+    companion object {
+        val properties =
+            linkedMapOf(
+                Meta.COLUMN_NAME_AB_RAW_ID to
+                    FieldType(AirbyteMetaFields.RAW_ID.type, nullable = false),
+                Meta.COLUMN_NAME_AB_EXTRACTED_AT to
+                    FieldType(AirbyteMetaFields.EXTRACTED_AT.type, nullable = false),
+                Meta.COLUMN_NAME_AB_META to
+                    FieldType(AirbyteMetaFields.META.type, nullable = false),
+                Meta.COLUMN_NAME_AB_GENERATION_ID to
+                    FieldType(AirbyteMetaFields.GENERATION_ID.type, nullable = false),
+            )
+    }
 }
 
 fun AirbyteType.withAirbyteMeta(flatten: Boolean = false): ObjectType =

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

@@ -163,13 +163,12 @@ private class ObjectValueSerializer : JsonSerializer<ObjectValue>() {
  * @property type The type ([AirbyteType]) of the [AirbyteValue]
  * @property changes List of [Meta.Change]s that have been applied to this value
  * @property name Field name
- * @property fieldCategory [FieldCategory] of the field
+ * @property fieldCategory [AirbyteMetaFields] of the field
  */
 class EnrichedAirbyteValue(
     var value: AirbyteValue,
     val type: AirbyteType,
     val name: String,
-    val fieldCategory: FieldCategory,
     val changes: MutableList<Meta.Change> = mutableListOf()
 ) {
     init {
@@ -207,16 +206,3 @@ class EnrichedAirbyteValue(
         changes.add(truncateChange)
     }
 }
-
-/**
- * The [EnrichedAirbyteValue] category allows us to quickly understand if the field is an Airbyte
- * controlled field or if it is declared by the source.
- */
-enum class FieldCategory {
-    RAW_ID,
-    EXTRACTED_AT,
-    META,
-    GENERATION_ID,
-    // For fields that don't match any of the predefined Airbyte columns
-    CLIENT_DATA
-}

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

@@ -63,3 +63,14 @@ fun DestinationRecordAirbyteValue.dataWithAirbyteMeta(
     stream: DestinationStream,
     flatten: Boolean = false
 ) = DestinationRecordToAirbyteValueWithMeta(stream, flatten).convert(data, emittedAtMs, meta)
+
+fun Meta.Change.toAirbyteValue(): ObjectValue =
+    ObjectValue(
+        linkedMapOf(
+            "field" to StringValue(field),
+            "change" to StringValue(change.name),
+            "reason" to StringValue(reason.name)
+        )
+    )
+
+fun List<Meta.Change>.toAirbyteValues(): List<ObjectValue> = map { it.toAirbyteValue() }