[source-postgres] Disable incremental sync for view in Postgres xmin replication mode

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

@@ -12,9 +12,9 @@ java {
 }
 
 airbyteJavaConnector {
- cdkVersionRequired = '0.42.1'
+ cdkVersionRequired = '0.42.2'
  features = ['db-sources', 'datastore-postgres']
- useLocalCdk = true
+ useLocalCdk = false
 }
 
 application {

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

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

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

@@ -18,10 +18,7 @@
 import io.airbyte.protocol.models.v0.AirbyteStreamNameNamespacePair;
 import io.airbyte.protocol.models.v0.SyncMode;
 import java.sql.SQLException;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.List;
-import java.util.Set;
+import java.util.*;
 import java.util.stream.Collectors;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -61,6 +58,42 @@ public static AirbyteStream overrideSyncModes(final AirbyteStream stream) {
  return stream.withSupportedSyncModes(Lists.newArrayList(SyncMode.FULL_REFRESH, SyncMode.INCREMENTAL));
  }
 
+ public static List<String> getViewsInSchema(final JdbcDatabase database, final String schema) throws SQLException {
+ final String query = "SELECT viewname FROM pg_catalog.pg_views WHERE schemaname = ?";
+ final List<JsonNode> result = database.queryJsons(query, schema);
+ return result.stream()
+ .map(jsonNode -> jsonNode.get("viewname").asText())
+ .collect(Collectors.toList());
+ }
+
+ public static Map<String, List<String>> getViewsForAllSchemas(final JdbcDatabase database, final List<String> schemas) throws SQLException {
+ Map<String, List<String>> viewsBySchema = new HashMap<>();
+ for (String schema : schemas) {
+ List<String> views = getViewsInSchema(database, schema);
+ viewsBySchema.put(schema, views);
+ }
+ return viewsBySchema;
+ }
+
+ public static boolean isStreamAView(Map<String, List<String>> viewsBySchema, final AirbyteStream stream) {
+ return viewsBySchema.getOrDefault(stream.getNamespace(), Collections.emptyList()).contains(stream.getName());
+ }
+
+ /**
+ * This method is used for CDC sync in order to overwrite sync modes for cursor fields cause cdc use
+ * another cursor logic. Used in discovery when xmin is configured.
+ *
+ * @param stream - airbyte stream
+ * @param viewsBySchema - view names by schema
+ * @return will return list of sync modes where view streams will have only FULL_REFRESH sync mode
+ */
+ public static AirbyteStream overrideSyncModes(final AirbyteStream stream, Map<String, List<String>> viewsBySchema) {
+ if (isStreamAView(viewsBySchema, stream)) {
+ return stream.withSupportedSyncModes(Lists.newArrayList(SyncMode.FULL_REFRESH));
+ }
+ return stream.withSupportedSyncModes(Lists.newArrayList(SyncMode.FULL_REFRESH, SyncMode.INCREMENTAL));
+ }
+
  /*
  * Set all streams that do have incremental to sourceDefined, so that the user cannot set or
  * override a cursor field.

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

@@ -312,16 +312,20 @@ public AirbyteCatalog discover(final JsonNode config) {
 
  catalog.setStreams(streams);
  } else if (isXmin(config)) {
- // Xmin replication has a source-defined cursor (the xmin column). This is done to prevent the user
- // from being able to pick their own cursor.
- final List<AirbyteStream> streams = catalog.getStreams().stream()
- .map(PostgresCatalogHelper::overrideSyncModes)
- .map(PostgresCatalogHelper::setIncrementalToSourceDefined)
- .collect(toList());
-
- catalog.setStreams(streams);
+ try {
+ JdbcDatabase database = createDatabase(config);
+ Map<String, List<String>> viewsBySchema = PostgresCatalogHelper.getViewsForAllSchemas(database, schemas);
+ // Xmin replication has a source-defined cursor (the xmin column). This is done to prevent the user
+ // from being able to pick their own cursor.
+ final List<AirbyteStream> streams = catalog.getStreams().stream()
+ .map(stream -> PostgresCatalogHelper.overrideSyncModes(stream, viewsBySchema))
+ .map(PostgresCatalogHelper::setIncrementalToSourceDefined)
+ .collect(toList());
+ catalog.setStreams(streams);
+ } catch (SQLException e) {
+ LOGGER.error("Error checking if stream is a view", e);
+ }
  }
-
  return catalog;
  }
 

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

@@ -141,6 +141,25 @@ void testDiscover() throws Exception {
  });
  }
 
+ @Test
+ void testDiscoverDisableIncrementalSyncForView() throws Exception {
+ testdb.query(ctx -> {
+ ctx.fetch("CREATE VIEW id_and_name_view AS SELECT * FROM id_and_name;");
+ return null;
+ });
+ final AirbyteCatalog actual = source().discover(getXminConfig());
+ actual.getStreams().forEach(actualStream -> {
+ if (actualStream.getName().equals("id_and_name_view")) {
+ assertTrue(!actualStream.getSupportedSyncModes().contains(SyncMode.INCREMENTAL));
+ assertTrue(actualStream.getSupportedSyncModes().contains(SyncMode.FULL_REFRESH));
+ }
+ });
+ testdb.query(ctx -> {
+ ctx.fetch("DROP VIEW id_and_name_view;");
+ return null;
+ });
+ }
+
  @Test
  void testReadSuccess() throws Exception {
  // Perform an initial sync with the configured catalog, which is set up to use xmin_replication.

docs/integrations/sources/postgres.md

@@ -311,6 +311,7 @@ According to Postgres [documentation](https://www.postgresql.org/docs/14/datatyp
 
 | Version | Date | Pull Request | Subject |
 |---------|------------|----------------------------------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
+| 3.6.2 | 2024-07-18 | [42108](https://github.com/airbytehq/airbyte/pull/42108) | Disable incremental sync for view streams in xmin replication mode |
 | 3.6.1 | 2024-07-05 | [40716](https://github.com/airbytehq/airbyte/pull/40716) | Fix typo in connector specification |
 | 3.6.0 | 2024-07-17 | [40208](https://github.com/airbytehq/airbyte/pull/40208) | Start using the new error Postgres source error handler that comes with a new error translation layer. | 
 | 3.5.2 | 2024-07-17 | [42068](https://github.com/airbytehq/airbyte/pull/42068) | Add analytics for WASS case occurrence. |