Teradata
diff --git a/‎.github/labeler.yml
-20 b/‎.github/labeler.yml
-20
diff --git a/‎.github/workflows/assign-issue-to-project.yaml
-34 b/‎.github/workflows/assign-issue-to-project.yaml
-34
diff --git a/‎.github/workflows/community_ci.yml
+4-4 b/‎.github/workflows/community_ci.yml
+4-4
diff --git a/‎.github/workflows/label-github-issues-by-path.yml
-17 b/‎.github/workflows/label-github-issues-by-path.yml
-17
diff --git a/‎.github/workflows/notify-on-push-to-master.yml
-19 b/‎.github/workflows/notify-on-push-to-master.yml
-19
diff --git a/‎airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/config/SyncBeanFactory.kt
+2-2 b/‎airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/config/SyncBeanFactory.kt
+2-2
diff --git a/‎airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/data/AirbyteValue.kt
+77 b/‎airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/data/AirbyteValue.kt
+77
diff --git a/‎airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/message/DestinationMessage.kt
+41 b/‎airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/message/DestinationMessage.kt
+41
diff --git a/‎airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/pipeline/DirectLoadPipelineStep.kt
+2-3 b/‎airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/pipeline/DirectLoadPipelineStep.kt
+2-3
diff --git a/‎airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/pipeline/DirectLoadRecordAccumulator.kt
+3-6 b/‎airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/pipeline/DirectLoadRecordAccumulator.kt
+3-6
diff --git a/‎airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/pipeline/InputPartitioner.kt
+3-3 b/‎airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/pipeline/InputPartitioner.kt
+3-3
diff --git a/‎airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/task/DestinationTaskLauncher.kt
+2-2 b/‎airbyte-cdk/bulk/core/load/src/main/kotlin/io/airbyte/cdk/load/task/DestinationTaskLauncher.kt
+2-2
@@ -37,8 +37,8 @@ jobs:
           echo "The fork has changes in protected paths. This is not allowed."
           exit 1
 
-  connectors_early_ci:
-    name: Run connectors early CI on fork
+  connectors_pre_release_checks:
+    name: Run pre-release checks on fork
     if: github.event.pull_request.head.repo.fork == true
     needs: fail_on_protected_path_changes
     environment: community-ci-auto
@@ -75,7 +75,7 @@ jobs:
         with:
           context: "pull_request"
           sentry_dsn: ${{ secrets.SENTRY_AIRBYTE_CI_DSN }}
-          subcommand: "connectors --modified test --only-step=qa_checks --only-step=version_inc_check --global-status-check-context='Connectors early CI checks' --global-status-check-description='Running early CI checks on connectors'"
+          subcommand: "connectors --modified test --only-step=qa_checks --only-step=version_inc_check --global-status-check-context='Connector Pre-Release Checks' --global-status-check-description='Running pre-release checks...'"
           is_fork: "true"
           git_repo_url: ${{ github.event.pull_request.head.repo.clone_url }}
           git_branch: ${{ github.head_ref }}
@@ -144,7 +144,7 @@ jobs:
           github_token: ${{ secrets.GH_PAT_MAINTENANCE_OSS }}
           s3_build_cache_access_key_id: ${{ secrets.SELF_RUNNER_AWS_ACCESS_KEY_ID }}
           s3_build_cache_secret_key: ${{ secrets.SELF_RUNNER_AWS_SECRET_ACCESS_KEY }}
-          subcommand: "connectors --modified test"
+          subcommand: "connectors --modified test --skip-step=qa_checks --skip-step=version_inc_check"
           is_fork: "true"
       - name: Upload pipeline reports
         id: upload-artifact
 
@@ -9,7 +9,7 @@ import io.airbyte.cdk.load.command.DestinationConfiguration
 import io.airbyte.cdk.load.command.DestinationStream
 import io.airbyte.cdk.load.message.BatchEnvelope
 import io.airbyte.cdk.load.message.ChannelMessageQueue
-import io.airbyte.cdk.load.message.DestinationRecordAirbyteValue
+import io.airbyte.cdk.load.message.DestinationRecordRaw
 import io.airbyte.cdk.load.message.MultiProducerChannel
 import io.airbyte.cdk.load.message.PartitionedQueue
 import io.airbyte.cdk.load.message.PipelineEvent
@@ -120,7 +120,7 @@ class SyncBeanFactory {
     @Named("recordQueue")
     fun recordQueue(
         loadStrategy: LoadStrategy? = null,
-    ): PartitionedQueue<Reserved<PipelineEvent<StreamKey, DestinationRecordAirbyteValue>>> {
+    ): PartitionedQueue<Reserved<PipelineEvent<StreamKey, DestinationRecordRaw>>> {
         return PartitionedQueue(
             Array(loadStrategy?.inputPartitions ?: 1) {
                 ChannelMessageQueue(Channel(Channel.UNLIMITED))
 
@@ -11,6 +11,9 @@ import com.fasterxml.jackson.databind.JsonSerializer
 import com.fasterxml.jackson.databind.SerializerProvider
 import com.fasterxml.jackson.databind.annotation.JsonSerialize
 import com.fasterxml.jackson.databind.node.NullNode
+import io.airbyte.cdk.load.message.Meta
+import io.airbyte.protocol.models.v0.AirbyteRecordMessageMetaChange
+import io.airbyte.protocol.models.v0.AirbyteRecordMessageMetaChange.Reason
 import java.math.BigDecimal
 import java.math.BigInteger
 import java.time.LocalDate
@@ -153,3 +156,77 @@ private class ObjectValueSerializer : JsonSerializer<ObjectValue>() {
 }
 
 @JvmInline value class UnknownValue(val value: JsonNode) : AirbyteValue
+
+/**
+ * Represents an "enriched" (/augmented) Airbyte value with additional metadata.
+ *
+ * @property value The actual [AirbyteValue]
+ * @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
+ */
+data class EnrichedAirbyteValue(
+    val value: AirbyteValue,
+    val type: AirbyteType,
+    val changes: List<Meta.Change> = emptyList(),
+    val name: String,
+    val fieldCategory: FieldCategory
+) {
+    init {
+        require(name.isNotBlank()) { "Field name cannot be blank" }
+    }
+
+    /**
+     * Creates a nullified version of this value with the specified reason.
+     *
+     * @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 {
+        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)
+    }
+
+    /**
+     * Creates a truncated version of this value with the specified reason and new value.
+     *
+     * @param reason The [Reason] for truncation, defaults to DESTINATION_RECORD_SIZE_LIMITATION
+     * @param newValue The new (truncated) value to use
+     * @return A new [EnrichedAirbyteValue] with the truncated value and an additional change record
+     */
+    fun toTruncated(
+        reason: Reason = Reason.DESTINATION_RECORD_SIZE_LIMITATION,
+        newValue: AirbyteValue
+    ): EnrichedAirbyteValue {
+        val truncateChange =
+            Meta.Change(
+                field = name,
+                change = AirbyteRecordMessageMetaChange.Change.TRUNCATED,
+                reason = reason
+            )
+
+        // Return a copy with null value and the new change added to the changes list
+        return copy(value = newValue, changes = changes + 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
+}
@@ -11,6 +11,7 @@ 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.AirbyteValueDeepCoercingMapper
+import io.airbyte.cdk.load.data.EnrichedAirbyteValue
 import io.airbyte.cdk.load.data.IntegerValue
 import io.airbyte.cdk.load.data.StringValue
 import io.airbyte.cdk.load.data.TimestampWithTimezoneValue
@@ -143,6 +144,9 @@ data class DestinationRecord(
             serialized.length.toLong()
         )
     }
+    fun asDestinationRecordRaw(): DestinationRecordRaw {
+        return DestinationRecordRaw(stream, message, serialized)
+    }
 }
 
 /**
@@ -163,6 +167,43 @@ data class DestinationRecordAirbyteValue(
     val serializedSizeBytes: Long = 0L
 )
 
+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
+)
+
+data class DestinationRecordRaw(
+    val stream: DestinationStream.Descriptor,
+    private val rawData: AirbyteMessage,
+    private val serialized: String
+) {
+    fun asRawJson(): JsonNode {
+        return rawData.record.data
+    }
+
+    fun asDestinationRecordAirbyteValue(): DestinationRecordAirbyteValue {
+        return DestinationRecordAirbyteValue(
+            stream,
+            rawData.record.data.toAirbyteValue(),
+            rawData.record.emittedAt,
+            Meta(
+                rawData.record.meta?.changes?.map { Meta.Change(it.field, it.change, it.reason) }
+                    ?: emptyList()
+            ),
+            serialized.length.toLong()
+        )
+    }
+
+    fun asEnrichedDestinationRecordAirbyteValue(): EnrichedDestinationRecordAirbyteValue {
+        TODO()
+    }
+}
+
 data class DestinationFile(
     override val stream: DestinationStream.Descriptor,
     val emittedAtMs: Long,
 
@@ -4,7 +4,7 @@
 
 package io.airbyte.cdk.load.pipeline
 
-import io.airbyte.cdk.load.message.DestinationRecordAirbyteValue
+import io.airbyte.cdk.load.message.DestinationRecordRaw
 import io.airbyte.cdk.load.message.PartitionedQueue
 import io.airbyte.cdk.load.message.PipelineEvent
 import io.airbyte.cdk.load.message.QueueWriter
@@ -24,8 +24,7 @@ import jakarta.inject.Singleton
 class DirectLoadPipelineStep<S : DirectLoader>(
     val accumulator: DirectLoadRecordAccumulator<S, StreamKey>,
     @Named("recordQueue")
-    val inputQueue:
-        PartitionedQueue<Reserved<PipelineEvent<StreamKey, DestinationRecordAirbyteValue>>>,
+    val inputQueue: PartitionedQueue<Reserved<PipelineEvent<StreamKey, DestinationRecordRaw>>>,
     @Named("batchStateUpdateQueue") val batchQueue: QueueWriter<BatchUpdate>,
     @Value("\${airbyte.destination.core.record-batch-size-override:null}")
     val batchSizeOverride: Long? = null,
 
@@ -5,7 +5,7 @@
 package io.airbyte.cdk.load.pipeline
 
 import io.airbyte.cdk.load.message.Batch
-import io.airbyte.cdk.load.message.DestinationRecordAirbyteValue
+import io.airbyte.cdk.load.message.DestinationRecordRaw
 import io.airbyte.cdk.load.message.WithBatchState
 import io.airbyte.cdk.load.message.WithStream
 import io.airbyte.cdk.load.write.DirectLoader
@@ -25,15 +25,12 @@ data class DirectLoadAccResult(override val state: Batch.State) : WithBatchState
 @Requires(bean = DirectLoaderFactory::class)
 class DirectLoadRecordAccumulator<S : DirectLoader, K : WithStream>(
     val directLoaderFactory: DirectLoaderFactory<S>
-) : BatchAccumulator<S, K, DestinationRecordAirbyteValue, DirectLoadAccResult> {
+) : BatchAccumulator<S, K, DestinationRecordRaw, DirectLoadAccResult> {
     override fun start(key: K, part: Int): S {
         return directLoaderFactory.create(key.stream, part)
     }
 
-    override fun accept(
-        record: DestinationRecordAirbyteValue,
-        state: S
-    ): Pair<S, DirectLoadAccResult?> {
+    override fun accept(record: DestinationRecordRaw, state: S): Pair<S, DirectLoadAccResult?> {
         state.accept(record).let {
             return when (it) {
                 is Incomplete -> Pair(state, null)
 
@@ -4,7 +4,7 @@
 
 package io.airbyte.cdk.load.pipeline
 
-import io.airbyte.cdk.load.message.DestinationRecordAirbyteValue
+import io.airbyte.cdk.load.message.DestinationRecordRaw
 import io.micronaut.context.annotation.Secondary
 import jakarta.inject.Singleton
 import kotlin.math.abs
@@ -14,13 +14,13 @@ import kotlin.math.abs
  * partitioned by a hash of the stream name and namespace.
  */
 interface InputPartitioner {
-    fun getPartition(record: DestinationRecordAirbyteValue, numParts: Int): Int
+    fun getPartition(record: DestinationRecordRaw, numParts: Int): Int
 }
 
 @Singleton
 @Secondary
 class ByStreamInputPartitioner : InputPartitioner {
-    override fun getPartition(record: DestinationRecordAirbyteValue, numParts: Int): Int {
+    override fun getPartition(record: DestinationRecordRaw, numParts: Int): Int {
         return abs(record.stream.hashCode()) % numParts
     }
 }
@@ -11,7 +11,7 @@ import io.airbyte.cdk.load.command.DestinationStream
 import io.airbyte.cdk.load.message.BatchEnvelope
 import io.airbyte.cdk.load.message.ChannelMessageQueue
 import io.airbyte.cdk.load.message.CheckpointMessageWrapped
-import io.airbyte.cdk.load.message.DestinationRecordAirbyteValue
+import io.airbyte.cdk.load.message.DestinationRecordRaw
 import io.airbyte.cdk.load.message.DestinationStreamEvent
 import io.airbyte.cdk.load.message.MessageQueue
 import io.airbyte.cdk.load.message.MessageQueueSupplier
@@ -145,7 +145,7 @@ class DefaultDestinationTaskLauncher<K : WithStream>(
     // New interface shim
     @Named("recordQueue")
     private val recordQueueForPipeline:
-        PartitionedQueue<Reserved<PipelineEvent<StreamKey, DestinationRecordAirbyteValue>>>,
+        PartitionedQueue<Reserved<PipelineEvent<StreamKey, DestinationRecordRaw>>>,
     @Named("batchStateUpdateQueue") private val batchUpdateQueue: ChannelMessageQueue<BatchUpdate>,
     private val loadPipeline: LoadPipeline?,
     private val partitioner: InputPartitioner,