fix(destination-redshift): use anti-join instead of correlated NOT EXISTS in dedup insert

airbyte-integrations/connectors/destination-redshift/metadata.yaml

@@ -55,7 +55,7 @@ data:
             type: GSM
   connectorType: destination
   definitionId: f7a7d195-377f-cf5b-70a5-be6b819019dc
-  dockerImageTag: 4.0.4
+  dockerImageTag: 4.0.5
   dockerRepository: airbyte/destination-redshift
   documentationUrl: https://docs.airbyte.com/integrations/destinations/redshift
   githubIssueLabel: destination-redshift

airbyte-integrations/connectors/destination-redshift/src/main/kotlin/io/airbyte/integrations/destination/redshift/sql/RedshiftSqlGenerator.kt

@@ -345,8 +345,9 @@ class RedshiftSqlGenerator(private val config: RedshiftConfiguration) {
      * Generates an INSERT statement for new rows from the deduped source that don't exist in the
      * target.
      *
-     * Uses `NOT EXISTS` to check for existing records by primary key. CDC-deleted rows are skipped
-     * if CDC hard delete is enabled.
+     * Uses a LEFT JOIN anti-join to check for existing records by primary key because Redshift
+     * rejects the correlated NOT EXISTS form. CDC-deleted rows are skipped if CDC hard delete is
+     * enabled.
      */
     internal fun insertNewRows(
         dedupTableAlias: String,
@@ -355,12 +356,14 @@ class RedshiftSqlGenerator(private val config: RedshiftConfiguration) {
         primaryKeyTargetColumns: List<String>,
         cdcHardDeleteEnabled: Boolean,
     ): String {
+        val targetAlias = quoteIdentifier("_airbyte_target")
         val primaryKeysConditions =
             buildNullSafePkMatch(
                 primaryKeyTargetColumns,
-                getFullyQualifiedName(targetTableName),
+                targetAlias,
                 dedupTableAlias,
             )
+        val rawIdColumn = quoteIdentifier(Meta.COLUMN_NAME_AB_RAW_ID)
 
         val skipCdcDeletedClause =
             if (cdcHardDeleteEnabled) {
@@ -374,14 +377,12 @@ class RedshiftSqlGenerator(private val config: RedshiftConfiguration) {
             |  ${allTargetColumns.joinToString(",\n  ")}
             |)
             |SELECT
-            |  ${allTargetColumns.joinToString(",\n  ")}
+            |  ${allTargetColumns.joinToString(",\n  ") { "$dedupTableAlias.$it" }}
             |FROM $dedupTableAlias
+            |LEFT JOIN ${getFullyQualifiedName(targetTableName)} AS $targetAlias
+            |  ON $primaryKeysConditions
             |WHERE
-            |  NOT EXISTS (
-            |    SELECT 1
-            |    FROM ${getFullyQualifiedName(targetTableName)}
-            |    WHERE $primaryKeysConditions
-            |  )$skipCdcDeletedClause
+            |  $targetAlias.$rawIdColumn IS NULL$skipCdcDeletedClause
         """.trimMargin()
     }
 

airbyte-integrations/connectors/destination-redshift/src/test/kotlin/io/airbyte/integrations/destination/redshift/sql/RedshiftSqlGeneratorTest.kt

@@ -425,7 +425,8 @@ internal class RedshiftSqlGeneratorTest {
             )
 
         assertTrue(sql.contains("INSERT INTO"))
-        assertTrue(sql.contains("NOT EXISTS"))
+        assertTrue(sql.contains("LEFT JOIN"))
+        assertTrue(sql.contains(""""_airbyte_target"."_airbyte_raw_id" IS NULL"""))
         assertFalse(sql.contains("_ab_cdc_deleted_at"))
     }
 
@@ -486,12 +487,40 @@ internal class RedshiftSqlGeneratorTest {
                 cdcHardDeleteEnabled = false,
             )
 
-        // NOT EXISTS subquery should use NULL-safe PK matching
+        assertFalse(sql.contains("NOT EXISTS"))
+        assertTrue(sql.contains("LEFT JOIN"))
         assertTrue(
             sql.contains(
-                """("ns"."final"."id" = deduped_source."id" OR ("ns"."final"."id" IS NULL AND deduped_source."id" IS NULL))"""
+                """("_airbyte_target"."id" = deduped_source."id" OR ("_airbyte_target"."id" IS NULL AND deduped_source."id" IS NULL))"""
+            )
+        )
+        assertTrue(sql.contains("deduped_source."))
+    }
+
+    @Test
+    fun `insertNewRows uses NULL-safe PK matching for compound keys`() {
+        val target = TableName(namespace = "ns", name = "final")
+        val sql =
+            sqlGenerator.insertNewRows(
+                dedupTableAlias = "deduped_source",
+                targetTableName = target,
+                allTargetColumns = listOf(""""id"""", """"org_id"""", """"name""""),
+                primaryKeyTargetColumns = listOf(""""id"""", """"org_id""""),
+                cdcHardDeleteEnabled = false,
+            )
+
+        assertFalse(sql.contains("NOT EXISTS"))
+        assertTrue(
+            sql.contains(
+                """("_airbyte_target"."id" = deduped_source."id" OR ("_airbyte_target"."id" IS NULL AND deduped_source."id" IS NULL))"""
+            )
+        )
+        assertTrue(
+            sql.contains(
+                """("_airbyte_target"."org_id" = deduped_source."org_id" OR ("_airbyte_target"."org_id" IS NULL AND deduped_source."org_id" IS NULL))"""
             )
         )
+        assertTrue(sql.contains(""""_airbyte_target"."_airbyte_raw_id" IS NULL"""))
     }
 
     // ================================================================
@@ -536,7 +565,9 @@ internal class RedshiftSqlGeneratorTest {
         assertTrue(sql.contains("FROM $dedup"))
         // Insert new rows
         assertTrue(sql.contains("INSERT INTO \"ns\".\"final\""))
-        assertTrue(sql.contains("NOT EXISTS"))
+        assertFalse(sql.contains("NOT EXISTS"))
+        assertTrue(sql.contains("LEFT JOIN"))
+        assertTrue(sql.contains(""""_airbyte_target"."_airbyte_raw_id" IS NULL"""))
         assertTrue(sql.contains("\"_ab_cdc_deleted_at\" IS NULL"))
         // Cleanup
         assertTrue(sql.contains("DROP TABLE IF EXISTS $dedup;"))

docs/integrations/destinations/redshift.md

@@ -277,6 +277,7 @@ This destination supports [namespaces](https://docs.airbyte.com/platform/using-a
 
 | Version | Date       | Pull Request                                               | Subject                                                                                                                                                                                                          |
 |:--------|:-----------|:-----------------------------------------------------------|:-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| 4.0.5 | 2026-07-30 | [83261](https://github.com/airbytehq/airbyte/pull/83261) | Fix dedup sync failure on Redshift by replacing the correlated NOT EXISTS subquery with an anti-join |
 | 4.0.4 | 2026-07-29 | [83245](https://github.com/airbytehq/airbyte/pull/83245) | fix: use NULL-safe primary key matching in upsert to prevent silent mismatches when PK columns contain NULL values |
 | 4.0.3 | 2026-07-14 | [81552](https://github.com/airbytehq/airbyte/pull/81552) | fix: narrow SQLException handling to only treat table-not-found as missing |
 | 4.0.2 | 2026-06-08 | [79161](https://github.com/airbytehq/airbyte/pull/79161) | fix: validate nested string sizes within SUPER columns to prevent COPY error 1224 |