Merge branch 'master' into daryna/source-greenhouce/milliseconds-cursor

Author: darynaishchenko

.github/labeler.yml

@@ -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

.github/workflows/assign-issue-to-project.yaml

@@ -1,19 +0,0 @@
-name: Trigger action in cloud based on push
-on:
-  push:
-    branches:
-      - master
-  workflow_dispatch:
-
-jobs:
-  repo-sync:
-    name: "Fire a Repo Dispatch event to airbyte-cloud"
-    runs-on: ubuntu-24.04
-    steps:
-      - name: Repository Dispatch
-        uses: peter-evans/repository-dispatch@v2
-        with:
-          token: ${{ secrets.GH_PAT_MAINTENANCE_OCTAVIA }}
-          repository: airbytehq/airbyte-cloud
-          event-type: oss-push-to-master
-          client-payload: '{"ref": "${{ github.ref }}", "sha": "${{ github.sha }}"}'

.github/workflows/community_ci.yml

@@ -118,13 +118,25 @@ sealed class FeedBootstrap<T : Feed>(
                 .filter { it.streams.contains(stream) }
                 .firstOrNull()
 
+        // Ideally we should check if sync is trigger-based CDC by checking source connector
+        // configuration. But we don't have that information here. So this is just a hacky solution
+        private val isTriggerBasedCdc: Boolean =
+            precedingGlobalFeed == null &&
+                metaFieldDecorator.globalCursor != null &&
+                stream.schema.none { it.id == metaFieldDecorator.globalCursor?.id } &&
+                stream.configuredCursor?.id == metaFieldDecorator.globalCursor?.id &&
+                stream.configuredSyncMode == ConfiguredSyncMode.INCREMENTAL
+
         private val defaultRecordData: ObjectNode =
             Jsons.objectNode().also { recordData: ObjectNode ->
                 stream.schema.forEach { recordData.putNull(it.id) }
-                if (feed is Stream && precedingGlobalFeed != null) {
+                if (feed is Stream && precedingGlobalFeed != null || isTriggerBasedCdc) {
                     metaFieldDecorator.decorateRecordData(
                         timestamp = outputConsumer.recordEmittedAt.atOffset(ZoneOffset.UTC),
-                        globalStateValue = stateManager.scoped(precedingGlobalFeed).current(),
+                        globalStateValue =
+                            if (precedingGlobalFeed != null)
+                                stateManager.scoped(precedingGlobalFeed).current()
+                            else null,
                         stream,
                         recordData,
                     )

.github/workflows/label-github-issues-by-path.yml

@@ -184,6 +184,52 @@ class FeedBootstrapTest {
         )
     }
 
+    @Test
+    fun testTriggerBasedCdcMetadataDecoration() {
+        // Create a stream without the global cursor in its schema to simulate trigger-based CDC
+        val triggerBasedStream =
+            Stream(
+                id = StreamIdentifier.from(StreamDescriptor().withName("tbl").withNamespace("ns")),
+                schema =
+                    setOf(
+                        k,
+                        v,
+                    ),
+                configuredSyncMode = ConfiguredSyncMode.INCREMENTAL,
+                configuredPrimaryKey = listOf(k),
+                configuredCursor =
+                    GlobalCursor, // For trigger based CDC the cursor is uniquely defined, we just
+                // use this object for test case
+                )
+
+        // Create state manager and bootstrap without a global feed
+        val stateManager =
+            StateManager(initialStreamStates = mapOf(triggerBasedStream to Jsons.arrayNode()))
+        val bootstrap = triggerBasedStream.bootstrap(stateManager)
+        val consumer = bootstrap.streamRecordConsumers().toList().first().second
+
+        // Test that a record gets CDC metadata decoration even without a global feed
+        consumer.accept(Jsons.readTree("""{"k":3,"v":"trigger"}""") as ObjectNode, changes = null)
+
+        val recordOutput = outputConsumer.records().map(Jsons::writeValueAsString).first()
+        val recordJson = Jsons.readTree(recordOutput)
+        val data = recordJson.get("data")
+
+        // Verify CDC metadata fields are present and properly decorated
+        Assertions.assertNotNull(data.get("_ab_cdc_lsn"))
+        Assertions.assertNotNull(data.get("_ab_cdc_updated_at"))
+        Assertions.assertNotNull(data.get("_ab_cdc_deleted_at"))
+
+        // The _ab_cdc_lsn should be decorated with the transaction timestamp
+        Assertions.assertTrue(data.get("_ab_cdc_lsn").isTextual)
+
+        // _ab_cdc_updated_at should be a timestamp string
+        Assertions.assertTrue(data.get("_ab_cdc_updated_at").isTextual)
+
+        // _ab_cdc_deleted_at should be null for non-deleted records
+        Assertions.assertTrue(data.get("_ab_cdc_deleted_at").isNull)
+    }
+
     companion object {
         const val GLOBAL_RECORD_DATA =
             """{"k":1,"v":"foo","_ab_cdc_lsn":123,"_ab_cdc_updated_at":"2024-03-01T01:02:03.456789","_ab_cdc_deleted_at":null}"""

.github/workflows/notify-on-push-to-master.yml

@@ -8,7 +8,6 @@ import com.fasterxml.jackson.databind.JsonNode
 import com.fasterxml.jackson.databind.node.ObjectNode
 import io.airbyte.cdk.command.OpaqueStateValue
 import io.airbyte.cdk.read.Stream
-import io.airbyte.cdk.util.Jsons
 import io.micronaut.context.annotation.Requires
 import io.micronaut.context.env.Environment
 import jakarta.inject.Singleton
@@ -43,7 +42,7 @@ class TestMetaFieldDecorator : MetaFieldDecorator {
         recordData.set<JsonNode>(
             GlobalCursor.id,
             if (globalStateValue == null) {
-                Jsons.nullNode()
+                CdcOffsetDateTimeMetaFieldType.jsonEncoder.encode(timestamp)
             } else {
                 CdcStringMetaFieldType.jsonEncoder.encode(globalStateValue.toString())
             }

.secrets

@@ -35,6 +35,11 @@ data class DestinationCatalog(val streams: List<DestinationStream> = emptyList()
             ?: throw IllegalArgumentException("Stream not found: namespace=$namespace, name=$name")
     }
 
+    fun getStream(descriptor: DestinationStream.Descriptor): DestinationStream {
+        return byDescriptor[descriptor]
+            ?: throw IllegalArgumentException("Stream not found: $descriptor")
+    }
+
     fun asProtocolObject(): ConfiguredAirbyteCatalog =
         ConfiguredAirbyteCatalog().withStreams(streams.map { it.asProtocolObject() })
 

airbyte-cdk/bulk/core/extract/src/main/kotlin/io/airbyte/cdk/read/FeedBootstrap.kt

@@ -9,10 +9,17 @@ 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.MultiProducerChannel
+import io.airbyte.cdk.load.message.PartitionedQueue
+import io.airbyte.cdk.load.message.PipelineEvent
+import io.airbyte.cdk.load.message.StreamKey
+import io.airbyte.cdk.load.pipeline.BatchUpdate
 import io.airbyte.cdk.load.state.ReservationManager
+import io.airbyte.cdk.load.state.Reserved
 import io.airbyte.cdk.load.task.implementor.FileAggregateMessage
 import io.airbyte.cdk.load.task.implementor.FileTransferQueueMessage
+import io.airbyte.cdk.load.write.LoadStrategy
 import io.github.oshai.kotlinlogging.KotlinLogging
 import io.micronaut.context.annotation.Factory
 import io.micronaut.context.annotation.Value
@@ -94,5 +101,37 @@ class SyncBeanFactory {
 
     @Singleton
     @Named("openStreamQueue")
-    class OpenStreamQueue : ChannelMessageQueue<DestinationStream>()
+    class OpenStreamQueue : ChannelMessageQueue<DestinationStream>(Channel(Channel.UNLIMITED))
+
+    /**
+     * If the client uses a new-style LoadStrategy, then we need to checkpoint by checkpoint id
+     * instead of record index.
+     */
+    @Singleton
+    @Named("checkpointById")
+    fun isCheckpointById(loadStrategy: LoadStrategy? = null): Boolean = loadStrategy != null
+
+    /**
+     * A single record queue for the whole sync, containing all streams, optionally partitioned by a
+     * configurable number of partitions. Number of partitions is controlled by the specified
+     * LoadStrategy, if any.
+     */
+    @Singleton
+    @Named("recordQueue")
+    fun recordQueue(
+        loadStrategy: LoadStrategy? = null,
+    ): PartitionedQueue<Reserved<PipelineEvent<StreamKey, DestinationRecordAirbyteValue>>> {
+        return PartitionedQueue(
+            Array(loadStrategy?.inputPartitions ?: 1) {
+                ChannelMessageQueue(Channel(Channel.UNLIMITED))
+            }
+        )
+    }
+
+    /** A queue for updating batch states, which is not partitioned. */
+    @Singleton
+    @Named("batchStateUpdateQueue")
+    fun batchStateUpdateQueue(): ChannelMessageQueue<BatchUpdate> {
+        return ChannelMessageQueue(Channel(100))
+    }
 }

airbyte-cdk/bulk/core/extract/src/test/kotlin/io/airbyte/cdk/read/FeedBootstrapTest.kt

@@ -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
+}