✨ Destination Postgres: DV2 GA (#35042)

Co-authored-by: Marius Posta <[email protected]>
Co-authored-by: Evan Tahler <[email protected]>

airbyte-cdk/java/airbyte-cdk/README.md

@@ -166,6 +166,7 @@ MavenLocal debugging steps:
 
 | Version | Date       | Pull Request                                               | Subject                                                                                                                                                        |
 |:--------|:-----------|:-----------------------------------------------------------|:---------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| 0.20.4  | 2024-02-12 | [\#35042](https://github.com/airbytehq/airbyte/pull/35042) | Use delegate's isDestinationV2 invocation in SshWrappedDestination.                                                                                            |
 | 0.20.3  | 2024-02-09 | [\#34580](https://github.com/airbytehq/airbyte/pull/34580) | Support special chars in mysql/mssql database name.                                                                                                            |
 | 0.20.2  | 2024-02-12 | [\#35111](https://github.com/airbytehq/airbyte/pull/35144) | Make state emission from async framework synchronized.                                                                                                         |
 | 0.20.1  | 2024-02-11 | [\#35111](https://github.com/airbytehq/airbyte/pull/35111) | Fix GlobalAsyncStateManager stats counting logic.                                                                                                              |

airbyte-cdk/java/airbyte-cdk/core/src/main/java/io/airbyte/cdk/integrations/base/ssh/SshWrappedDestination.java

@@ -129,4 +129,9 @@ protected SshTunnel getTunnelInstance(final JsonNode config) throws Exception {
         : getInstance(config, hostKey, portKey);
   }
 
+  @Override
+  public boolean isV2Destination() {
+    return delegate.isV2Destination();
+  }
+
 }

airbyte-cdk/java/airbyte-cdk/core/src/main/resources/version.properties

@@ -1 +1 @@
-version=0.20.3
+version=0.20.4

airbyte-integrations/connectors/destination-postgres-strict-encrypt/build.gradle

@@ -1,48 +1,20 @@
 plugins {
-    id 'application'
     id 'airbyte-java-connector'
 }
 
 airbyteJavaConnector {
-    cdkVersionRequired = '0.16.3'
-    features = [
-            'db-sources', // required for tests
-            'db-destinations'
-    ]
+    cdkVersionRequired = '0.20.4'
+    features = ['db-destinations', 'typing-deduping', 'datastore-postgres']
     useLocalCdk = false
 }
 
-//remove once upgrading the CDK version to 0.4.x or later
-java {
-    compileJava {
-        options.compilerArgs.remove("-Werror")
-    }
-}
-
-airbyteJavaConnector.addCdkDependencies()
-
 application {
     mainClass = 'io.airbyte.integrations.destination.postgres.PostgresDestinationStrictEncrypt'
     applicationDefaultJvmArgs = ['-XX:+ExitOnOutOfMemoryError', '-XX:MaxRAMPercentage=75.0']
 }
 
 dependencies {
     implementation project(':airbyte-integrations:connectors:destination-postgres')
-    // TODO: declare typing-deduping as a CDK feature instead of importing from source.
-    implementation project(':airbyte-cdk:java:airbyte-cdk:typing-deduping')
-
-    testImplementation libs.testcontainers.jdbc
-    testImplementation libs.testcontainers.postgresql
-
-    testFixturesImplementation libs.testcontainers.jdbc
-    testFixturesImplementation libs.testcontainers.postgresql
 
     integrationTestJavaImplementation testFixtures(project(':airbyte-integrations:connectors:destination-postgres'))
-    integrationTestJavaImplementation testFixtures(project(':airbyte-cdk:java:airbyte-cdk:typing-deduping'))
-}
-
-configurations.all {
-    resolutionStrategy {
-        force libs.jooq
-    }
 }

airbyte-integrations/connectors/destination-postgres-strict-encrypt/metadata.yaml

@@ -2,22 +2,27 @@ data:
   connectorSubtype: database
   connectorType: destination
   definitionId: 25c5221d-dce2-4163-ade9-739ef790f503
-  dockerImageTag: 0.6.3
+  dockerImageTag: 2.0.0
   dockerRepository: airbyte/destination-postgres-strict-encrypt
   documentationUrl: https://docs.airbyte.com/integrations/destinations/postgres
   githubIssueLabel: destination-postgres
   icon: postgresql.svg
   license: ELv2
   name: Postgres
-  normalizationConfig:
-    normalizationIntegrationType: postgres
-    normalizationRepository: airbyte/normalization
-    normalizationTag: 0.4.1
   registries:
     cloud:
       enabled: false
     oss:
       enabled: false
+  releases:
+    breakingChanges:
+      2.0.0:
+        message: >
+          This version introduces [Destinations V2](https://docs.airbyte.com/release_notes/upgrading_to_destinations_v2/#what-is-destinations-v2), which provides better error handling, incremental delivery of data for large syncs, and improved final table structures. 
+          To review the breaking changes, and how to upgrade, see [here](https://docs.airbyte.com/release_notes/upgrading_to_destinations_v2/#quick-start-to-upgrading). 
+          These changes will likely require updates to downstream dbt / SQL models, which we walk through [here](https://docs.airbyte.com/release_notes/upgrading_to_destinations_v2/#updating-downstream-transformations).
+          Selecting `Upgrade` will upgrade **all** connections using this destination at their next sync. For more controlled upgrade [see instructions](https://docs.airbyte.com/release_notes/upgrading_to_destinations_v2/#upgrading-connections-one-by-one-with-dual-writing).
+        upgradeDeadline: "2024-05-31"
   releaseStage: alpha
   supportsDbt: true
   tags:

airbyte-integrations/connectors/destination-postgres-strict-encrypt/src/main/java/io/airbyte/integrations/destination/postgres/PostgresDestinationStrictEncrypt.java

@@ -60,6 +60,11 @@ public AirbyteConnectionStatus check(final JsonNode config) throws Exception {
     return super.check(config);
   }
 
+  @Override
+  public boolean isV2Destination() {
+    return true;
+  }
+
   public static void main(final String[] args) throws Exception {
     final Destination destination = new PostgresDestinationStrictEncrypt();
     LOGGER.info("starting destination: {}", PostgresDestinationStrictEncrypt.class);

airbyte-integrations/connectors/destination-postgres-strict-encrypt/src/test-integration/java/io/airbyte/integrations/destination/postgres/PostgresDestinationStrictEncryptAcceptanceTest.java

@@ -16,6 +16,7 @@
 import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 
+@Disabled("Disabled after DV2 migration. Re-enable with fixtures updated to DV2.")
 public class PostgresDestinationStrictEncryptAcceptanceTest extends AbstractPostgresDestinationAcceptanceTest {
 
   private PostgresTestDatabase testDb;

airbyte-integrations/connectors/destination-postgres-strict-encrypt/src/test-integration/java/io/airbyte/integrations/destination/postgres/typing_deduping/PostgresStrictEncryptTypingDedupingTest.java

@@ -0,0 +1,68 @@
+/*
+ * Copyright (c) 2023 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.destination.postgres.typing_deduping;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.collect.ImmutableMap;
+import io.airbyte.integrations.destination.postgres.PostgresDestination;
+import io.airbyte.integrations.destination.postgres.PostgresTestDatabase;
+import io.airbyte.integrations.destination.postgres.PostgresTestDatabase.BaseImage;
+import javax.sql.DataSource;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.BeforeAll;
+
+// TODO: This test is added to ensure coverage missed by disabling DATs. Redundant when DATs
+// enabled.
+public class PostgresStrictEncryptTypingDedupingTest extends AbstractPostgresTypingDedupingTest {
+
+  protected static PostgresTestDatabase testContainer;
+
+  @BeforeAll
+  public static void setupPostgres() {
+    // Postgres-13 is alpine image and SSL conf is failing to load, intentionally using 12:bullseye
+    testContainer = PostgresTestDatabase.in(BaseImage.POSTGRES_12, PostgresTestDatabase.ContainerModifier.CERT);
+  }
+
+  @AfterAll
+  public static void teardownPostgres() {
+    testContainer.close();
+  }
+
+  @Override
+  protected ObjectNode getBaseConfig() {
+    return (ObjectNode) testContainer.configBuilder()
+        .with("schema", "public")
+        .withDatabase()
+        .withResolvedHostAndPort()
+        .withCredentials()
+        .withSsl(ImmutableMap.builder()
+            .put("mode", "verify-ca") // verify-full will not work since the spawned container is only allowed for 127.0.0.1/32 CIDRs
+            .put("ca_certificate", testContainer.getCertificates().caCertificate())
+            .build())
+        .build();
+  }
+
+  @Override
+  protected DataSource getDataSource(final JsonNode config) {
+    // Intentionally ignore the config and rebuild it.
+    // The config param has the resolved (i.e. in-docker) host/port.
+    // We need the unresolved host/port since the test wrapper code is running from the docker host
+    // rather than in a container.
+    return new PostgresDestination().getDataSource(testContainer.configBuilder()
+        .with("schema", "public")
+        .withDatabase()
+        .withHostAndPort()
+        .withCredentials()
+        .withoutSsl()
+        .build());
+  }
+
+  @Override
+  protected String getImageName() {
+    return "airbyte/destination-postgres-strict-encrypt:dev";
+  }
+
+}

airbyte-integrations/connectors/destination-postgres-strict-encrypt/src/test-integration/resources/dat/sync1_cursorchange_expectedrecords_dedup_final.jsonl

@@ -0,0 +1,3 @@
+{"_airbyte_extracted_at": "1970-01-01T00:00:01.000000Z", "_airbyte_meta": {"errors":[]}, "id1": 1, "id2": 200, "old_cursor": 1, "name": "Alice", "address": {"city": "Los Angeles", "state": "CA"}}
+{"_airbyte_extracted_at": "1970-01-01T00:00:01.000000Z", "_airbyte_meta": {"errors":[]}, "id1": 1, "id2": 201, "old_cursor": 2, "name": "Bob", "address": {"city": "Boston", "state": "MA"}}
+{"_airbyte_extracted_at": "1970-01-01T00:00:01.000000Z", "_airbyte_meta": {"errors":["Problem with `age`", "Problem with `registration_date`"]}, "id1": 2, "id2": 200, "old_cursor": 3, "name": "Charlie"}

airbyte-integrations/connectors/destination-postgres-strict-encrypt/src/test-integration/resources/dat/sync1_cursorchange_expectedrecords_dedup_raw.jsonl

@@ -0,0 +1,4 @@
+{"_airbyte_extracted_at": "1970-01-01T00:00:01.000000Z", "_airbyte_data": {"id1": 1, "id2": 200, "old_cursor": 0, "_ab_cdc_deleted_at": null, "name" :"Alice", "address": {"city": "San Francisco", "state": "CA"}}}
+{"_airbyte_extracted_at": "1970-01-01T00:00:01.000000Z", "_airbyte_data": {"id1": 1, "id2": 200, "old_cursor": 1, "_ab_cdc_deleted_at": null, "name": "Alice", "address": {"city": "Los Angeles", "state": "CA"}}}
+{"_airbyte_extracted_at": "1970-01-01T00:00:01.000000Z", "_airbyte_data": {"id1": 1, "id2": 201, "old_cursor": 2, "name": "Bob", "address": {"city": "Boston", "state": "MA"}}}
+{"_airbyte_extracted_at": "1970-01-01T00:00:01.000000Z", "_airbyte_data": {"id1": 2, "id2": 200, "old_cursor": 3, "name": "Charlie", "age": "this is not an integer", "registration_date": "this is not a date"}}

airbyte-integrations/connectors/destination-postgres-strict-encrypt/src/test-integration/resources/dat/sync1_expectedrecords_dedup_final.jsonl

@@ -0,0 +1,5 @@
+// Keep the Alice record with more recent updated_at
+{"_airbyte_extracted_at": "1970-01-01T00:00:01.000000Z", "_airbyte_meta": {"errors":[]}, "id1": 1, "id2": 200, "updated_at": "2000-01-01T00:01:00.000000Z", "name": "Alice", "address": {"city": "Los Angeles", "state": "CA"}}
+{"_airbyte_extracted_at": "1970-01-01T00:00:01.000000Z", "_airbyte_meta": {"errors":[]}, "id1": 1, "id2": 201, "updated_at": "2000-01-01T00:02:00.000000Z", "name": "Bob", "address": {"city": "Boston", "state": "MA"}}
+{"_airbyte_extracted_at": "1970-01-01T00:00:01.000000Z", "_airbyte_meta": {"errors":["Problem with `age`", "Problem with `registration_date`"]}, "id1": 2, "id2": 200, "updated_at": "2000-01-01T00:03:00.000000Z", "name": "Charlie"}
+{"_airbyte_extracted_at": "1970-01-01T00:00:01.000000Z", "_airbyte_meta": {"errors":[]}, "id1": 3, "id2": 200, "updated_at": "2000-01-01T00:04:00.000000Z", "name": "a\bb\fc\nd\re\tf`~!@#$%^&*()_+-=[]\\{}|'\",./<>?"}

airbyte-integrations/connectors/destination-postgres-strict-encrypt/src/test-integration/resources/dat/sync1_expectedrecords_dedup_final2.jsonl

@@ -0,0 +1 @@
+{"_airbyte_extracted_at": "1970-01-01T00:00:01.000000Z", "_airbyte_meta": {"errors":[]}, "id1": 1, "id2": 200, "updated_at": "2001-01-01T00:00:00.000000Z", "name": "Someone completely different"}

airbyte-integrations/connectors/destination-postgres-strict-encrypt/src/test-integration/resources/dat/sync1_expectedrecords_nondedup_final.jsonl

@@ -0,0 +1,6 @@
+{"_airbyte_extracted_at": "1970-01-01T00:00:01.000000Z", "_airbyte_meta": {"errors":[]}, "id1": 1, "id2": 200, "updated_at": "2000-01-01T00:00:00.000000Z", "name": "Alice", "address": {"city": "San Francisco", "state": "CA"}}
+{"_airbyte_extracted_at": "1970-01-01T00:00:01.000000Z", "_airbyte_meta": {"errors":[]}, "id1": 1, "id2": 200, "updated_at": "2000-01-01T00:01:00.000000Z", "name": "Alice", "address": {"city": "Los Angeles", "state": "CA"}}
+{"_airbyte_extracted_at": "1970-01-01T00:00:01.000000Z", "_airbyte_meta": {"errors":[]}, "id1": 1, "id2": 201, "updated_at": "2000-01-01T00:02:00.000000Z", "name": "Bob", "address": {"city": "Boston", "state": "MA"}}
+// Invalid columns are nulled out (i.e. SQL null, not JSON null)
+{"_airbyte_extracted_at": "1970-01-01T00:00:01.000000Z", "_airbyte_meta": {"errors":["Problem with `age`", "Problem with `registration_date`"]}, "id1": 2, "id2": 200, "updated_at": "2000-01-01T00:03:00.000000Z", "name": "Charlie"}
+{"_airbyte_extracted_at": "1970-01-01T00:00:01.000000Z", "_airbyte_meta": {"errors":[]}, "id1": 3, "id2": 200, "updated_at": "2000-01-01T00:04:00.000000Z", "name": "a\bb\fc\nd\re\tf`~!@#$%^&*()_+-=[]\\{}|'\",./<>?"}

airbyte-integrations/connectors/destination-postgres-strict-encrypt/src/test-integration/resources/dat/sync1_expectedrecords_raw.jsonl

@@ -0,0 +1,6 @@
+{"_airbyte_extracted_at": "1970-01-01T00:00:01.000000Z", "_airbyte_data": {"id1": 1, "id2": 200, "updated_at": "2000-01-01T00:00:00Z", "_ab_cdc_deleted_at": null, "name": "Alice", "address": {"city": "San Francisco", "state": "CA"}}}
+{"_airbyte_extracted_at": "1970-01-01T00:00:01.000000Z", "_airbyte_data": {"id1": 1, "id2": 200, "updated_at": "2000-01-01T00:01:00Z", "_ab_cdc_deleted_at": null, "name": "Alice", "address": {"city": "Los Angeles", "state": "CA"}}}
+{"_airbyte_extracted_at": "1970-01-01T00:00:01.000000Z", "_airbyte_data": {"id1": 1, "id2": 201, "updated_at": "2000-01-01T00:02:00Z", "name": "Bob", "address": {"city": "Boston", "state": "MA"}}}
+// Invalid data is still allowed in the raw table.
+{"_airbyte_extracted_at": "1970-01-01T00:00:01.000000Z", "_airbyte_data": {"id1": 2, "id2": 200, "updated_at": "2000-01-01T00:03:00Z", "name": "Charlie", "age": "this is not an integer", "registration_date": "this is not a date"}}
+{"_airbyte_extracted_at": "1970-01-01T00:00:01.000000Z", "_airbyte_data": {"id1": 3, "id2": 200, "updated_at": "2000-01-01T00:04:00Z", "name": "a\bb\fc\nd\re\tf`~!@#$%^&*()_+-=[]\\{}|'\",./<>?"}}

airbyte-integrations/connectors/destination-postgres-strict-encrypt/src/test-integration/resources/dat/sync1_expectedrecords_raw2.jsonl

@@ -0,0 +1 @@
+{"_airbyte_extracted_at": "1970-01-01T00:00:01.000000Z", "_airbyte_data": {"id1": 1, "id2": 200, "updated_at": "2001-01-01T00:00:00Z", "_ab_cdc_deleted_at": null, "name": "Someone completely different"}}

airbyte-integrations/connectors/destination-postgres-strict-encrypt/src/test-integration/resources/dat/sync2_cursorchange_expectedrecords_incremental_dedup_final.jsonl

@@ -0,0 +1,3 @@
+{"_airbyte_extracted_at": "1970-01-01T00:00:02.000000Z", "_airbyte_meta":{"errors":[]}, "id1": 1, "id2": 200, "updated_at": "2000-01-02T00:00:00.000000Z", "name": "Alice", "address": {"city": "Seattle", "state": "WA"}}
+// Charlie wasn't reemitted with updated_at, so it still has a null cursor
+{"_airbyte_extracted_at": "1970-01-01T00:00:01.000000Z", "_airbyte_meta": {"errors":["Problem with `age`", "Problem with `registration_date`"]}, "id1": 2, "id2": 200, "name": "Charlie"}

airbyte-integrations/connectors/destination-postgres-strict-encrypt/src/test-integration/resources/dat/sync2_cursorchange_expectedrecords_incremental_dedup_raw.jsonl

@@ -0,0 +1,7 @@
+{"_airbyte_extracted_at": "1970-01-01T00:00:01.000000Z", "_airbyte_data": {"id1": 1, "id2": 200, "old_cursor": 0, "_ab_cdc_deleted_at": null, "name" :"Alice", "address": {"city": "San Francisco", "state": "CA"}}}
+{"_airbyte_extracted_at": "1970-01-01T00:00:01.000000Z", "_airbyte_data": {"id1": 1, "id2": 200, "old_cursor": 1, "_ab_cdc_deleted_at": null, "name": "Alice", "address": {"city": "Los Angeles", "state": "CA"}}}
+{"_airbyte_extracted_at": "1970-01-01T00:00:01.000000Z", "_airbyte_data": {"id1": 1, "id2": 201, "old_cursor": 2, "name": "Bob", "address": {"city": "Boston", "state": "MA"}}}
+{"_airbyte_extracted_at": "1970-01-01T00:00:01.000000Z", "_airbyte_data": {"id1": 2, "id2": 200, "old_cursor": 3, "name": "Charlie", "age": "this is not an integer", "registration_date": "this is not a date"}}
+{"_airbyte_extracted_at": "1970-01-01T00:00:02.000000Z", "_airbyte_data": {"id1": 1, "id2": 200, "updated_at": "2000-01-02T00:00:00Z", "_ab_cdc_deleted_at": null, "name": "Alice", "address": {"city": "Seattle", "state": "WA"}}}
+{"_airbyte_extracted_at": "1970-01-01T00:00:02.000000Z", "_airbyte_data": {"id1": 1, "id2": 201, "updated_at": "2000-01-02T00:00:00Z", "_ab_cdc_deleted_at": null, "name": "Bob", "address": {"city": "New York", "state": "NY"}}}
+{"_airbyte_extracted_at": "1970-01-01T00:00:02.000000Z", "_airbyte_data": {"id1": 1, "id2": 201, "updated_at": "2000-01-02T00:01:00Z", "_ab_cdc_deleted_at": "1970-01-01T00:00:00Z"}}

airbyte-integrations/connectors/destination-postgres-strict-encrypt/src/test-integration/resources/dat/sync2_expectedrecords_fullrefresh_append_final.jsonl

@@ -0,0 +1,9 @@
+{"_airbyte_extracted_at": "1970-01-01T00:00:01.000000Z", "_airbyte_meta": {"errors":[]}, "id1": 1, "id2": 200, "updated_at": "2000-01-01T00:00:00.000000Z", "name": "Alice", "address": {"city": "San Francisco", "state": "CA"}}
+{"_airbyte_extracted_at": "1970-01-01T00:00:01.000000Z", "_airbyte_meta": {"errors":[]}, "id1": 1, "id2": 200, "updated_at": "2000-01-01T00:01:00.000000Z", "name": "Alice", "address": {"city": "Los Angeles", "state": "CA"}}
+{"_airbyte_extracted_at": "1970-01-01T00:00:01.000000Z", "_airbyte_meta": {"errors":[]}, "id1": 1, "id2": 201, "updated_at": "2000-01-01T00:02:00.000000Z", "name": "Bob", "address": {"city": "Boston", "state": "MA"}}
+{"_airbyte_extracted_at": "1970-01-01T00:00:01.000000Z", "_airbyte_meta": {"errors":["Problem with `age`", "Problem with `registration_date`"]}, "id1": 2, "id2": 200, "updated_at": "2000-01-01T00:03:00.000000Z", "name": "Charlie"}
+{"_airbyte_extracted_at": "1970-01-01T00:00:01.000000Z", "_airbyte_meta": {"errors":[]}, "id1": 3, "id2": 200, "updated_at": "2000-01-01T00:04:00.000000Z", "name": "a\bb\fc\nd\re\tf`~!@#$%^&*()_+-=[]\\{}|'\",./<>?"}
+
+{"_airbyte_extracted_at": "1970-01-01T00:00:02.000000Z", "_airbyte_meta":{"errors":[]}, "id1": 1, "id2": 200, "updated_at": "2000-01-02T00:00:00.000000Z", "name": "Alice", "address": {"city": "Seattle", "state": "WA"}}
+{"_airbyte_extracted_at": "1970-01-01T00:00:02.000000Z", "_airbyte_meta":{"errors":[]}, "id1": 1, "id2": 201, "updated_at": "2000-01-02T00:00:00.000000Z", "name": "Bob", "address": {"city": "New York", "state": "NY"}}
+{"_airbyte_extracted_at": "1970-01-01T00:00:02.000000Z", "_airbyte_meta":{"errors":[]}, "id1": 1, "id2": 201, "updated_at": "2000-01-02T00:01:00.000000Z", "_ab_cdc_deleted_at": "1970-01-01T00:00:00.000000Z"}

airbyte-integrations/connectors/destination-postgres-strict-encrypt/src/test-integration/resources/dat/sync2_expectedrecords_fullrefresh_overwrite_final.jsonl

@@ -0,0 +1,3 @@
+{"_airbyte_extracted_at": "1970-01-01T00:00:02.000000Z", "_airbyte_meta":{"errors":[]}, "id1": 1, "id2": 200, "updated_at": "2000-01-02T00:00:00.000000Z", "name": "Alice", "address": {"city": "Seattle", "state": "WA"}}
+{"_airbyte_extracted_at": "1970-01-01T00:00:02.000000Z", "_airbyte_meta":{"errors":[]}, "id1": 1, "id2": 201, "updated_at": "2000-01-02T00:00:00.000000Z", "name": "Bob", "address": {"city": "New York", "state": "NY"}}
+{"_airbyte_extracted_at": "1970-01-01T00:00:02.000000Z", "_airbyte_meta":{"errors":[]}, "id1": 1, "id2": 201, "updated_at": "2000-01-02T00:01:00.000000Z", "_ab_cdc_deleted_at": "1970-01-01T00:00:00.000000Z"}

airbyte-integrations/connectors/destination-postgres-strict-encrypt/src/test-integration/resources/dat/sync2_expectedrecords_fullrefresh_overwrite_raw.jsonl

@@ -0,0 +1,3 @@
+{"_airbyte_extracted_at": "1970-01-01T00:00:02.000000Z", "_airbyte_data": {"id1": 1, "id2": 200, "updated_at": "2000-01-02T00:00:00Z", "_ab_cdc_deleted_at": null, "name": "Alice", "address": {"city": "Seattle", "state": "WA"}}}
+{"_airbyte_extracted_at": "1970-01-01T00:00:02.000000Z", "_airbyte_data": {"id1": 1, "id2": 201, "updated_at": "2000-01-02T00:00:00Z", "_ab_cdc_deleted_at": null, "name": "Bob", "address": {"city": "New York", "state": "NY"}}}
+{"_airbyte_extracted_at": "1970-01-01T00:00:02.000000Z", "_airbyte_data": {"id1": 1, "id2": 201, "updated_at": "2000-01-02T00:01:00Z", "_ab_cdc_deleted_at": "1970-01-01T00:00:00Z"}}

airbyte-integrations/connectors/destination-postgres-strict-encrypt/src/test-integration/resources/dat/sync2_expectedrecords_incremental_dedup_final.jsonl

@@ -0,0 +1,4 @@
+{"_airbyte_extracted_at": "1970-01-01T00:00:02.000000Z", "_airbyte_meta":{"errors":[]}, "id1": 1, "id2": 200, "updated_at": "2000-01-02T00:00:00.000000Z", "name": "Alice", "address": {"city": "Seattle", "state": "WA"}}
+// Delete Bob, keep Charlie
+{"_airbyte_extracted_at": "1970-01-01T00:00:01.000000Z", "_airbyte_meta": {"errors":["Problem with `age`", "Problem with `registration_date`"]}, "id1": 2, "id2": 200, "updated_at": "2000-01-01T00:03:00.000000Z", "name": "Charlie"}
+{"_airbyte_extracted_at": "1970-01-01T00:00:01.000000Z", "_airbyte_meta": {"errors":[]}, "id1": 3, "id2": 200, "updated_at": "2000-01-01T00:04:00.000000Z", "name": "a\bb\fc\nd\re\tf`~!@#$%^&*()_+-=[]\\{}|'\",./<>?"}

airbyte-integrations/connectors/destination-postgres-strict-encrypt/src/test-integration/resources/dat/sync2_expectedrecords_incremental_dedup_final2.jsonl

@@ -0,0 +1 @@
+{"_airbyte_extracted_at": "1970-01-01T00:00:02.000000Z", "_airbyte_meta":{"errors":[]}, "id1": 1, "id2": 200, "updated_at": "2001-01-02T00:00:00.000000Z", "name": "Someone completely different v2"}