Merge branch 'master' into 11873-ai-agent-in-docs

Author: ian-at-airbyte

.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())
             }

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

@@ -9,7 +9,6 @@ import io.airbyte.cdk.load.command.azureBlobStorage.AzureBlobStorageConfiguratio
 import io.airbyte.cdk.load.file.object_storage.StreamingUpload
 import io.airbyte.cdk.load.util.setOnce
 import io.github.oshai.kotlinlogging.KotlinLogging
-import java.io.BufferedInputStream
 import java.nio.ByteBuffer
 import java.util.Base64
 import java.util.concurrent.ConcurrentSkipListMap
@@ -42,14 +41,16 @@ class AzureBlobStreamingUpload(
 
         // The stageBlock call can be done asynchronously or blocking.
         // Here we use the blocking call in a coroutine context.
-        BufferedInputStream(part.inputStream()).use {
+        part.inputStream().use {
             blockBlobClient.stageBlock(
                 blockId,
                 it,
                 part.size.toLong(),
             )
         }
 
+        log.info { "Staged block #$index => $rawBlockId (encoded = $blockId)" }
+
         // Keep track of the blocks in the order they arrived (or the index).
         blockIds[index] = blockId
     }
@@ -77,6 +78,8 @@ class AzureBlobStreamingUpload(
                     blockBlobClient.setMetadata(filteredMetadata)
                 }
             }
+        } else {
+            log.warn { "Complete called multiple times for ${blockBlobClient.blobName}" }
         }
 
         return AzureBlob(blockBlobClient.blobName, config)

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

@@ -117,7 +117,6 @@ class MSSQLCSVFormattingWriter(
     private val printer = finalSchema.toCsvPrinterWithHeader(outputStream)
     private val mssqlRowValidator = MSSQLCsvRowValidator(validateValuesPreLoad)
     override fun accept(record: DestinationRecordAirbyteValue) {
-
         printer.printRecord(
             mssqlRowValidator
                 .validate(record, this.finalSchema)

airbyte-cdk/bulk/core/extract/src/testFixtures/kotlin/io/airbyte/cdk/discover/TestMetaFieldDecorator.kt

@@ -85,7 +85,7 @@ If you are developing on pipelines, we recommend installing airbyte-ci with poet
 ```bash
 cd airbyte-ci/connectors/pipelines/
 poetry install
-poetry shell
+poetry env activate
 cd ../../
 ```
 

airbyte-cdk/bulk/toolkits/load-azure-blob-storage/src/main/kotlin/io/airbyte/cdk/load/file/azureBlobStorage/AzureBlobStreamingUpload.kt

@@ -64,7 +64,7 @@ class ManifestOnlyConnectorUnitTests(PytestStep):
 
     title = "Manifest-only unit tests"
     test_directory_name = "unit_tests"
-    common_test_dependencies = ["pytest"]
+    common_test_dependencies = ["pytest", "requests-mock"]
 
     async def install_testing_environment(
         self,

airbyte-cdk/bulk/toolkits/load-object-storage/src/main/kotlin/io/airbyte/cdk/load/file/object_storage/ObjectStorageFormattingWriter.kt

@@ -168,7 +168,7 @@ async def install_testing_environment(
         )
         if self.common_test_dependencies:
             container_with_test_deps = container_with_test_deps.with_user("root").with_exec(
-                ["pip", "install", f'{" ".join(self.common_test_dependencies)}']
+                ["pip", "install"] + self.common_test_dependencies
             )
 
         container_with_test_deps = (

airbyte-ci/connectors/pipelines/README.md

@@ -35,7 +35,7 @@
 MANIFEST_FILE_NAME = "manifest.yaml"
 METADATA_ICON_FILE_NAME = "icon.svg"
 DIFF_FILTER = "MADRT"  # Modified, Added, Deleted, Renamed, Type changed
-IGNORED_FILE_EXTENSIONS: List[str] = []
+IGNORED_FILE_EXTENSIONS: List[str] = [".md"]
 
 
 # This utils will probably be redundant once https://github.com/dagger/dagger/issues/3764 is implemented