[source-postgres] Fix duplicate streams in postgres (#40719)

Author: xiaohansong

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

@@ -1 +1 @@
-version=0.40.9
+version=0.40.10

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

@@ -61,7 +61,8 @@ abstract class CdcSourceTest<S : Source, T : TestDatabase<*, T, *>> {
  )
  .withSourceDefinedPrimaryKey(
  java.util.List.of(java.util.List.of(COL_ID)),
- ),
+ )
+ .withIsResumable(true),
  ),
  )
 
@@ -688,7 +689,8 @@ abstract class CdcSourceTest<S : Source, T : TestDatabase<*, T, *>> {
  .withSupportedSyncModes(
  Lists.newArrayList(SyncMode.FULL_REFRESH, SyncMode.INCREMENTAL),
  )
- .withSourceDefinedPrimaryKey(java.util.List.of(java.util.List.of(COL_ID))),
+ .withSourceDefinedPrimaryKey(java.util.List.of(java.util.List.of(COL_ID)))
+ .withIsResumable(true),
  )
  airbyteStream.syncMode = SyncMode.FULL_REFRESH
 
@@ -770,6 +772,7 @@ abstract class CdcSourceTest<S : Source, T : TestDatabase<*, T, *>> {
 
  val recordMessages2 = extractRecordMessages(actualRecords2)
  val stateMessages2 = extractStateMessages(actualRecords2)
+ stateMessages2.map { state -> assertStateDoNotHaveDuplicateStreams(state) }
 
  assertExpectedStateMessagesFromIncrementalSync(stateMessages2)
  assertExpectedStateMessageCountMatches(stateMessages2, 1)
@@ -826,6 +829,7 @@ abstract class CdcSourceTest<S : Source, T : TestDatabase<*, T, *>> {
 
  val recordMessages2 = extractRecordMessages(actualRecords2)
  val stateMessages2 = extractStateMessages(actualRecords2)
+ stateMessages2.map { state -> assertStateDoNotHaveDuplicateStreams(state) }
 
  assertExpectedRecords(
  (MODEL_RECORDS_2 + listOf(puntoRecord)).toSet(),
@@ -905,7 +909,8 @@ abstract class CdcSourceTest<S : Source, T : TestDatabase<*, T, *>> {
  )
  .withSupportedSyncModes(
  Lists.newArrayList(SyncMode.FULL_REFRESH, SyncMode.INCREMENTAL),
- ),
+ )
+ .withIsResumable(false),
  )
  airbyteStream.syncMode = SyncMode.FULL_REFRESH
 
@@ -932,6 +937,7 @@ abstract class CdcSourceTest<S : Source, T : TestDatabase<*, T, *>> {
  stateMessages1,
  MODEL_RECORDS.size.toLong() + MODEL_RECORDS_2.size.toLong()
  )
+ stateMessages1.map { state -> assertStateDoNotHaveDuplicateStreams(state) }
  assertExpectedRecords(
  (MODEL_RECORDS_2 + MODEL_RECORDS).toSet(),
  recordMessages1,
@@ -950,6 +956,7 @@ abstract class CdcSourceTest<S : Source, T : TestDatabase<*, T, *>> {
 
  val recordMessages2 = extractRecordMessages(actualRecords2)
  val stateMessages2 = extractStateMessages(actualRecords2)
+ stateMessages2.map { state -> assertStateDoNotHaveDuplicateStreams(state) }
 
  assertExpectedStateMessageCountMatches(stateMessages2, 1 + MODEL_RECORDS_2.size.toLong())
  assertExpectedRecords(
@@ -1370,7 +1377,8 @@ abstract class CdcSourceTest<S : Source, T : TestDatabase<*, T, *>> {
  .withSupportedSyncModes(
  Lists.newArrayList(SyncMode.FULL_REFRESH, SyncMode.INCREMENTAL),
  )
- .withSourceDefinedPrimaryKey(java.util.List.of(java.util.List.of(COL_ID))),
+ .withSourceDefinedPrimaryKey(java.util.List.of(java.util.List.of(COL_ID)))
+ .withIsResumable(true),
  )
  airbyteStream.syncMode = SyncMode.FULL_REFRESH
 

airbyte-cdk/java/airbyte-cdk/db-sources/src/testFixtures/kotlin/io/airbyte/cdk/integrations/debezium/CdcSourceTest.kt

@@ -12,7 +12,7 @@ java {
 }
 
 airbyteJavaConnector {
- cdkVersionRequired = '0.40.7'
+ cdkVersionRequired = '0.40.10'
  features = ['db-sources', 'datastore-postgres']
  useLocalCdk = false
 }

airbyte-integrations/connectors/source-postgres/build.gradle

@@ -9,7 +9,7 @@ data:
  connectorSubtype: database
  connectorType: source
  definitionId: decd338e-5647-4c0b-adf4-da0e75f5a750
- dockerImageTag: 3.4.21
+ dockerImageTag: 3.4.22
  dockerRepository: airbyte/source-postgres
  documentationUrl: https://docs.airbyte.com/integrations/sources/postgres
  githubIssueLabel: source-postgres

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

@@ -68,8 +68,7 @@ private void initStream(final CtidStreams ctidStreams,
  }
 
  if (configuredAirbyteStream.getSyncMode() == SyncMode.FULL_REFRESH) {
- if (fileNodeHandler.hasFileNode(
- new AirbyteStreamNameNamespacePair(configuredAirbyteStream.getStream().getName(), configuredAirbyteStream.getStream().getNamespace()))) {
+ if (configuredAirbyteStream.getStream().getIsResumable()) {
  this.resumableFullRefreshStreams.add(pair);
  } else {
  this.nonResumableFullRefreshStreams.add(pair);
@@ -144,10 +143,14 @@ public CdcState getCdcState() {
 
  }
 
+ private boolean isIncrementalStream(final AirbyteStreamNameNamespacePair pair) {
+ return !resumableFullRefreshStreams.contains(pair) && !nonResumableFullRefreshStreams.contains(pair);
+ }
+
  @Override
  public AirbyteStateMessage createFinalStateMessage(final AirbyteStreamNameNamespacePair pair, final JsonNode streamStateForIncrementalRun) {
  // Only incremental streams can be transformed into the next phase.
- if (!resumableFullRefreshStreams.contains(pair)) {
+ if (isIncrementalStream(pair)) {
  streamsThatHaveCompletedSnapshot.add(pair);
  }
  final List<AirbyteStreamState> streamStates = new ArrayList<>();

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

@@ -86,7 +86,8 @@ protected ConfiguredAirbyteCatalog getConfiguredCatalog() {
  Field.of("name", JsonSchemaType.STRING))
  .withSourceDefinedCursor(true)
  .withSourceDefinedPrimaryKey(List.of(List.of("id")))
- .withSupportedSyncModes(Lists.newArrayList(SyncMode.FULL_REFRESH, SyncMode.INCREMENTAL))),
+ .withSupportedSyncModes(Lists.newArrayList(SyncMode.FULL_REFRESH, SyncMode.INCREMENTAL))
+ .withIsResumable(true)),
  new ConfiguredAirbyteStream()
  .withSyncMode(SyncMode.INCREMENTAL)
  .withDestinationSyncMode(DestinationSyncMode.APPEND)
@@ -97,7 +98,8 @@ protected ConfiguredAirbyteCatalog getConfiguredCatalog() {
  Field.of("name", JsonSchemaType.STRING))
  .withSourceDefinedCursor(true)
  .withSourceDefinedPrimaryKey(List.of(List.of("id")))
- .withSupportedSyncModes(Lists.newArrayList(SyncMode.FULL_REFRESH, SyncMode.INCREMENTAL)))));
+ .withSupportedSyncModes(Lists.newArrayList(SyncMode.FULL_REFRESH, SyncMode.INCREMENTAL))
+ .withIsResumable(true))));
  }
 
  protected ConfiguredAirbyteCatalog getConfiguredCatalogWithPartialColumns() {
@@ -112,7 +114,8 @@ protected ConfiguredAirbyteCatalog getConfiguredCatalogWithPartialColumns() {
  /* no name field */)
  .withSourceDefinedCursor(true)
  .withSourceDefinedPrimaryKey(List.of(List.of("id")))
- .withSupportedSyncModes(Lists.newArrayList(SyncMode.FULL_REFRESH, SyncMode.INCREMENTAL))),
+ .withSupportedSyncModes(Lists.newArrayList(SyncMode.FULL_REFRESH, SyncMode.INCREMENTAL))
+ .withIsResumable(true)),
  new ConfiguredAirbyteStream()
  .withSyncMode(SyncMode.INCREMENTAL)
  .withDestinationSyncMode(DestinationSyncMode.APPEND)
@@ -123,7 +126,8 @@ protected ConfiguredAirbyteCatalog getConfiguredCatalogWithPartialColumns() {
  Field.of("name", JsonSchemaType.STRING))
  .withSourceDefinedCursor(true)
  .withSourceDefinedPrimaryKey(List.of(List.of("id")))
- .withSupportedSyncModes(Lists.newArrayList(SyncMode.FULL_REFRESH, SyncMode.INCREMENTAL)))));
+ .withSupportedSyncModes(Lists.newArrayList(SyncMode.FULL_REFRESH, SyncMode.INCREMENTAL))
+ .withIsResumable(true))));
  }
 
  @Override

airbyte-integrations/connectors/source-postgres/src/test-integration/java/io/airbyte/integrations/io/airbyte/integration_tests/sources/CdcPostgresSourceAcceptanceTest.java

@@ -266,10 +266,6 @@ protected void validateStreamStateInResumableFullRefresh(final JsonNode streamSt
  assertEquals("ctid", streamStateToBeTested.get("state_type").asText());
  }
 
- @Override
- @Test
- protected void testCdcAndNonResumableFullRefreshInSameSync() throws Exception {}
-
  @Override
  protected void assertStateMessagesForNewTableSnapshotTest(final List<? extends AirbyteStateMessage> stateMessages,
  final AirbyteStateMessage stateMessageEmittedAfterFirstSyncCompletion) {