[Source-postgres] Set default cursor value for cdc mode

airbyte-integrations/bases/debezium/src/testFixtures/java/io/airbyte/integrations/debezium/CdcSourceTest.java

@@ -770,6 +770,7 @@ protected AirbyteCatalog expectedCatalogForDiscover() {
     // stream with PK
     streams.get(0).setSourceDefinedCursor(true);
     addCdcMetadataColumns(streams.get(0));
+    addCdcDefaultCursorField(streams.get(0));
 
     final AirbyteStream streamWithoutPK = CatalogHelpers.createAirbyteStream(
         MODELS_STREAM_NAME + "_2",
@@ -779,6 +780,7 @@ protected AirbyteCatalog expectedCatalogForDiscover() {
         Field.of(COL_MODEL, JsonSchemaType.STRING));
     streamWithoutPK.setSourceDefinedPrimaryKey(Collections.emptyList());
     streamWithoutPK.setSupportedSyncModes(List.of(SyncMode.FULL_REFRESH));
+    addCdcDefaultCursorField(streamWithoutPK);
     addCdcMetadataColumns(streamWithoutPK);
 
     final AirbyteStream randomStream = CatalogHelpers.createAirbyteStream(
@@ -790,6 +792,8 @@ protected AirbyteCatalog expectedCatalogForDiscover() {
         .withSourceDefinedCursor(true)
         .withSupportedSyncModes(Lists.newArrayList(SyncMode.FULL_REFRESH, SyncMode.INCREMENTAL))
         .withSourceDefinedPrimaryKey(List.of(List.of(COL_ID + "_random")));
+
+    addCdcDefaultCursorField(randomStream);
     addCdcMetadataColumns(randomStream);
 
     streams.add(streamWithoutPK);
@@ -815,6 +819,8 @@ protected AirbyteCatalog expectedCatalogForDiscover() {
 
   protected abstract void addCdcMetadataColumns(final AirbyteStream stream);
 
+  protected abstract void addCdcDefaultCursorField(final AirbyteStream stream);
+
   protected abstract Source getSource();
 
   protected abstract JsonNode getConfig();

airbyte-integrations/connectors/source-mssql/src/test/java/io/airbyte/integrations/source/mssql/CdcMssqlSourceTest.java

@@ -436,6 +436,11 @@ protected void addCdcMetadataColumns(final AirbyteStream stream) {
 
   }
 
+  @Override
+  protected void addCdcDefaultCursorField(final AirbyteStream stream) {
+    // Leaving empty until cdc default cursor is implemented for MSSQL
+  }
+
   @Override
   protected Source getSource() {
     return new MssqlSource();

airbyte-integrations/connectors/source-mysql/src/test/java/io/airbyte/integrations/source/mysql/CdcMysqlSourceTest.java

@@ -191,6 +191,11 @@ protected void addCdcMetadataColumns(final AirbyteStream stream) {
     properties.set(CDC_DELETED_AT, stringType);
   }
 
+  @Override
+  protected void addCdcDefaultCursorField(final AirbyteStream stream) {
+    // Leaving empty until cdc default cursor is implemented for MySQL
+  }
+
   @Override
   protected Source getSource() {
     return source;

airbyte-integrations/connectors/source-postgres-strict-encrypt/Dockerfile

@@ -24,5 +24,5 @@ ENV APPLICATION source-postgres-strict-encrypt
 
 COPY --from=build /airbyte /airbyte
 
-LABEL io.airbyte.version=2.1.1
+LABEL io.airbyte.version=3.0.0
 LABEL io.airbyte.name=airbyte/source-postgres-strict-encrypt

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

@@ -12,7 +12,7 @@ data:
   connectorType: source
   definitionId: decd338e-5647-4c0b-adf4-da0e75f5a750
   maxSecondsBetweenMessages: 7200
-  dockerImageTag: 2.1.1
+  dockerImageTag: 3.0.0
   dockerRepository: airbyte/source-postgres-strict-encrypt
   githubIssueLabel: source-postgres
   icon: postgresql.svg

airbyte-integrations/connectors/source-postgres/Dockerfile

@@ -24,5 +24,5 @@ ENV APPLICATION source-postgres
 
 COPY --from=build /airbyte /airbyte
 
-LABEL io.airbyte.version=2.1.1
+LABEL io.airbyte.version=3.0.0
 LABEL io.airbyte.name=airbyte/source-postgres

airbyte-integrations/connectors/source-postgres/acceptance-test-config.yml

@@ -24,6 +24,8 @@ acceptance_tests:
     tests:
       - config_path: "secrets/config.json"
       - config_path: "secrets/config_cdc.json"
+        backward_compatibility_tests_config:
+          disable_for_version: "2.1.1"
   basic_read:
     tests:
       - config_path: "secrets/config.json"

airbyte-integrations/connectors/source-postgres/metadata.yaml

@@ -6,7 +6,7 @@ data:
   connectorSubtype: database
   connectorType: source
   definitionId: decd338e-5647-4c0b-adf4-da0e75f5a750
-  dockerImageTag: 2.1.1
+  dockerImageTag: 3.0.0
   maxSecondsBetweenMessages: 7200
   dockerRepository: airbyte/source-postgres
   githubIssueLabel: source-postgres

airbyte-integrations/connectors/source-postgres/src/main/java/io/airbyte/integrations/source/postgres/PostgresCatalogHelper.java

@@ -4,8 +4,11 @@
 
 package io.airbyte.integrations.source.postgres;
 
+import static io.airbyte.integrations.debezium.internals.DebeziumEventUtils.CDC_LSN;
+
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Lists;
 import io.airbyte.commons.json.Jsons;
@@ -123,7 +126,7 @@ public static Set<AirbyteStreamNameNamespacePair> getPublicizedTables(final Jdbc
   }
 
   /**
-   * This method is used for xmin synsc in order to overwrite sync modes for cursor fields. For xmin, we want streams to only have incremental mode
+   * This method is used for xmin syncs in order to overwrite sync modes for cursor fields. For xmin, we want streams to only have incremental mode
    * enabled.
    *
    * @param stream - airbyte stream
@@ -133,4 +136,12 @@ public static AirbyteStream overrideSyncModesForXmin(final AirbyteStream stream)
     return stream.withSupportedSyncModes(Lists.newArrayList(SyncMode.INCREMENTAL));
   }
 
+  /*
+   * To prepare for Destination v2, cdc streams must have a default cursor field
+   * this defaults to lsn as a cursor as it is monotonically increasing and unique
+   */
+  public static AirbyteStream setDefaultCursorFieldForCdc(final AirbyteStream stream) {
+    stream.setDefaultCursorField(ImmutableList.of(CDC_LSN));
+    return stream;
+  }
 }

airbyte-integrations/connectors/source-postgres/src/main/java/io/airbyte/integrations/source/postgres/PostgresSource.java

@@ -275,6 +275,7 @@ public AirbyteCatalog discover(final JsonNode config) throws Exception {
           .map(PostgresCatalogHelper::overrideSyncModes)
           .map(PostgresCatalogHelper::removeIncrementalWithoutPk)
           .map(PostgresCatalogHelper::setIncrementalToSourceDefined)
+          .map(PostgresCatalogHelper::setDefaultCursorFieldForCdc)
           .map(PostgresCatalogHelper::addCdcMetadataColumns)
           // If we're in CDC mode and a stream is not in the publication, the user should only be able to sync
           // this in FULL_REFRESH mode

airbyte-integrations/connectors/source-postgres/src/test/java/io/airbyte/integrations/source/postgres/CdcPostgresSourceTest.java

@@ -293,6 +293,11 @@ protected void addCdcMetadataColumns(final AirbyteStream stream) {
 
   }
 
+  @Override
+  protected void addCdcDefaultCursorField(final AirbyteStream stream) {
+    stream.setDefaultCursorField(ImmutableList.of(CDC_LSN));
+  }
+
   @Override
   protected Source getSource() {
     return source;

airbyte-integrations/connectors/source-postgres/src/test/java/io/airbyte/integrations/source/postgres/PostgresCatalogHelperTest.java

@@ -4,6 +4,7 @@
 
 package io.airbyte.integrations.source.postgres;
 
+import static io.airbyte.integrations.debezium.internals.DebeziumEventUtils.CDC_LSN;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -16,6 +17,7 @@
 import java.util.List;
 import java.util.Map;
 import org.junit.jupiter.api.Test;
+import org.testcontainers.shaded.com.google.common.collect.ImmutableList;
 
 class PostgresCatalogHelperTest {
 
@@ -64,6 +66,16 @@ public void testSetIncrementalToSourceDefined() {
         .setIncrementalToSourceDefined(noIncremental)
         .getSourceDefinedCursor());
   }
+  @Test
+  public void testSetDefaultCursorFieldForCdc() {
+    final AirbyteStream cdcIncrementalStream = new AirbyteStream()
+        .withSourceDefinedCursor(true)
+        .withSupportedSyncModes(List.of(SyncMode.FULL_REFRESH));
+    PostgresCatalogHelper.setDefaultCursorFieldForCdc(cdcIncrementalStream);
+
+    assertTrue(cdcIncrementalStream.getSourceDefinedCursor());
+    assertEquals(cdcIncrementalStream.getDefaultCursorField(), ImmutableList.of(CDC_LSN));
+  }
 
   @Test
   public void testAddCdcMetadataColumns() {

docs/integrations/sources/postgres.md

@@ -407,6 +407,7 @@ Some larger tables may encounter an error related to the temporary file size lim
 
 | Version | Date       | Pull Request                                             | Subject                                                                                                                                                                    |
 | :------ | :--------- | :------------------------------------------------------- | :------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | --- |
+| 3.0.0   | 2023-07-12 | [27442](https://github.com/airbytehq/airbyte/pull/27442) | Set _ab_cdc_lsn as the source defined cursor for CDC mode to prepare for Destination v2 normalization                                                                                                                                           |
 | 2.1.1   | 2023-07-06 | [26723](https://github.com/airbytehq/airbyte/pull/26723) | Add new xmin replication method.                                                                                                                                           |
 | 2.1.0   | 2023-06-26 | [27737](https://github.com/airbytehq/airbyte/pull/27737) | License Update: Elv2                                                                                                                                                       |
 | 2.0.34  | 2023-06-20 | [27212](https://github.com/airbytehq/airbyte/pull/27212) | Fix silent exception swallowing in StreamingJdbcDatabase                                                                                                                   |