diff --git a/airbyte-api/src/main/openapi/config.yaml b/airbyte-api/src/main/openapi/config.yaml index d28c818016e1..baaa16a5ab3b 100644 --- a/airbyte-api/src/main/openapi/config.yaml +++ b/airbyte-api/src/main/openapi/config.yaml @@ -1887,10 +1887,12 @@ components: description: the mutable part of the stream to configure the destination type: object additionalProperties: false + required: + - syncMode + - destinationSyncMode properties: syncMode: $ref: "#/components/schemas/SyncMode" - default: full_refresh cursorField: description: Path to the field that will be used to determine if a record is new or modified since the last sync. This field is REQUIRED if `sync_mode` is `incremental`. Otherwise it is ignored. type: array @@ -1898,7 +1900,6 @@ components: type: string destinationSyncMode: $ref: "#/components/schemas/DestinationSyncMode" - default: append primaryKey: description: Paths to the fields that will be used as primary key. This field is REQUIRED if `destination_sync_mode` is `*_dedup`. Otherwise it is ignored. type: array @@ -2180,7 +2181,7 @@ components: - append - overwrite #- upsert_dedup # TODO chris: SCD Type 1 can be implemented later - - append_dedup # SCD Type 2 + - append_dedup # SCD Type 1 & 2 AirbyteArchive: type: string format: binary diff --git a/airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/22f6c74f-5699-40ff-833c-4a879ea40133.json b/airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/22f6c74f-5699-40ff-833c-4a879ea40133.json index 7e7a55eacace..3f6b703073e5 100644 --- a/airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/22f6c74f-5699-40ff-833c-4a879ea40133.json +++ b/airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/22f6c74f-5699-40ff-833c-4a879ea40133.json @@ -2,6 +2,6 @@ "destinationDefinitionId": "22f6c74f-5699-40ff-833c-4a879ea40133", "name": "BigQuery", "dockerRepository": "airbyte/destination-bigquery", - "dockerImageTag": "0.2.0", + "dockerImageTag": "0.2.2", "documentationUrl": "https://docs.airbyte.io/integrations/destinations/bigquery" } diff --git a/airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/25c5221d-dce2-4163-ade9-739ef790f503.json b/airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/25c5221d-dce2-4163-ade9-739ef790f503.json index 2bd62e324b94..a73ee7fd9bc7 100644 --- a/airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/25c5221d-dce2-4163-ade9-739ef790f503.json +++ b/airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/25c5221d-dce2-4163-ade9-739ef790f503.json @@ -2,6 +2,6 @@ "destinationDefinitionId": "25c5221d-dce2-4163-ade9-739ef790f503", "name": "Postgres", "dockerRepository": "airbyte/destination-postgres", - "dockerImageTag": "0.2.0", + "dockerImageTag": "0.2.2", "documentationUrl": "https://docs.airbyte.io/integrations/destinations/postgres" } diff --git a/airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/424892c4-daac-4491-b35d-c6688ba547ba.json b/airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/424892c4-daac-4491-b35d-c6688ba547ba.json index 8d679bb252c8..def1e27edf10 100644 --- a/airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/424892c4-daac-4491-b35d-c6688ba547ba.json +++ b/airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/424892c4-daac-4491-b35d-c6688ba547ba.json @@ -2,6 +2,6 @@ "destinationDefinitionId": "424892c4-daac-4491-b35d-c6688ba547ba", "name": "Snowflake", "dockerRepository": "airbyte/destination-snowflake", - "dockerImageTag": "0.2.0", + "dockerImageTag": "0.2.2", "documentationUrl": "https://docs.airbyte.io/integrations/destinations/snowflake" } diff --git a/airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/8be1cf83-fde1-477f-a4ad-318d23c9f3c6.json b/airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/8be1cf83-fde1-477f-a4ad-318d23c9f3c6.json index 3d1a8ea6f848..a4d086b0cb43 100644 --- a/airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/8be1cf83-fde1-477f-a4ad-318d23c9f3c6.json +++ b/airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/8be1cf83-fde1-477f-a4ad-318d23c9f3c6.json @@ -2,6 +2,6 @@ "destinationDefinitionId": "8be1cf83-fde1-477f-a4ad-318d23c9f3c6", "name": "Local CSV", "dockerRepository": "airbyte/destination-csv", - "dockerImageTag": "0.2.0", + "dockerImageTag": "0.2.2", "documentationUrl": "https://docs.airbyte.io/integrations/destinations/local-csv" } diff --git a/airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/a625d593-bba5-4a1c-a53d-2d246268a816.json b/airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/a625d593-bba5-4a1c-a53d-2d246268a816.json index d7ecf5e27e83..d97d3af4dd19 100644 --- a/airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/a625d593-bba5-4a1c-a53d-2d246268a816.json +++ b/airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/a625d593-bba5-4a1c-a53d-2d246268a816.json @@ -2,6 +2,6 @@ "destinationDefinitionId": "a625d593-bba5-4a1c-a53d-2d246268a816", "name": "Local JSON", "dockerRepository": "airbyte/destination-local-json", - "dockerImageTag": "0.2.0", + "dockerImageTag": "0.2.2", "documentationUrl": "https://docs.airbyte.io/integrations/destinations/local-json" } diff --git a/airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/af7c921e-5892-4ff2-b6c1-4a5ab258fb7e.json b/airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/af7c921e-5892-4ff2-b6c1-4a5ab258fb7e.json index 21018601a713..7888cb0d67cf 100644 --- a/airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/af7c921e-5892-4ff2-b6c1-4a5ab258fb7e.json +++ b/airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/af7c921e-5892-4ff2-b6c1-4a5ab258fb7e.json @@ -2,6 +2,6 @@ "destinationDefinitionId": "af7c921e-5892-4ff2-b6c1-4a5ab258fb7e", "name": "MeiliSearch", "dockerRepository": "airbyte/destination-meilisearch", - "dockerImageTag": "0.2.0", + "dockerImageTag": "0.2.2", "documentationUrl": "https://docs.airbyte.io/integrations/destinations/meilisearch" } diff --git a/airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/f7a7d195-377f-cf5b-70a5-be6b819019dc.json b/airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/f7a7d195-377f-cf5b-70a5-be6b819019dc.json index f686204fee6b..4cbcc1dc550d 100644 --- a/airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/f7a7d195-377f-cf5b-70a5-be6b819019dc.json +++ b/airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/f7a7d195-377f-cf5b-70a5-be6b819019dc.json @@ -2,6 +2,6 @@ "destinationDefinitionId": "f7a7d195-377f-cf5b-70a5-be6b819019dc", "name": "Redshift", "dockerRepository": "airbyte/destination-redshift", - "dockerImageTag": "0.2.0", + "dockerImageTag": "0.2.2", "documentationUrl": "https://docs.airbyte.io/integrations/destinations/redshift" } diff --git a/airbyte-config/init/src/main/resources/config/STANDARD_SOURCE_DEFINITION/435bb9a5-7887-4809-aa58-28c27df0d7ad.json b/airbyte-config/init/src/main/resources/config/STANDARD_SOURCE_DEFINITION/435bb9a5-7887-4809-aa58-28c27df0d7ad.json index 718350942fa9..51d5a72bb4e0 100644 --- a/airbyte-config/init/src/main/resources/config/STANDARD_SOURCE_DEFINITION/435bb9a5-7887-4809-aa58-28c27df0d7ad.json +++ b/airbyte-config/init/src/main/resources/config/STANDARD_SOURCE_DEFINITION/435bb9a5-7887-4809-aa58-28c27df0d7ad.json @@ -2,6 +2,6 @@ "sourceDefinitionId": "435bb9a5-7887-4809-aa58-28c27df0d7ad", "name": "MySQL", "dockerRepository": "airbyte/source-mysql", - "dockerImageTag": "0.2.1", + "dockerImageTag": "0.2.2", "documentationUrl": "https://docs.airbyte.io/integrations/sources/mysql" } diff --git a/airbyte-config/init/src/main/resources/config/STANDARD_SOURCE_DEFINITION/b5ea17b1-f170-46dc-bc31-cc744ca984c1.json b/airbyte-config/init/src/main/resources/config/STANDARD_SOURCE_DEFINITION/b5ea17b1-f170-46dc-bc31-cc744ca984c1.json index afb18ddb4931..8bed5a41b174 100644 --- a/airbyte-config/init/src/main/resources/config/STANDARD_SOURCE_DEFINITION/b5ea17b1-f170-46dc-bc31-cc744ca984c1.json +++ b/airbyte-config/init/src/main/resources/config/STANDARD_SOURCE_DEFINITION/b5ea17b1-f170-46dc-bc31-cc744ca984c1.json @@ -2,6 +2,6 @@ "sourceDefinitionId": "b5ea17b1-f170-46dc-bc31-cc744ca984c1", "name": "Microsoft SQL Server (MSSQL)", "dockerRepository": "airbyte/source-mssql", - "dockerImageTag": "0.2.1", + "dockerImageTag": "0.2.2", "documentationUrl": "https://hub.docker.com/r/airbyte/source-mssql" } diff --git a/airbyte-config/init/src/main/resources/config/STANDARD_SOURCE_DEFINITION/decd338e-5647-4c0b-adf4-da0e75f5a750.json b/airbyte-config/init/src/main/resources/config/STANDARD_SOURCE_DEFINITION/decd338e-5647-4c0b-adf4-da0e75f5a750.json index f1eeaed10da3..d10928ab7774 100644 --- a/airbyte-config/init/src/main/resources/config/STANDARD_SOURCE_DEFINITION/decd338e-5647-4c0b-adf4-da0e75f5a750.json +++ b/airbyte-config/init/src/main/resources/config/STANDARD_SOURCE_DEFINITION/decd338e-5647-4c0b-adf4-da0e75f5a750.json @@ -2,6 +2,6 @@ "sourceDefinitionId": "decd338e-5647-4c0b-adf4-da0e75f5a750", "name": "Postgres", "dockerRepository": "airbyte/source-postgres", - "dockerImageTag": "0.2.1", + "dockerImageTag": "0.2.2", "documentationUrl": "https://hub.docker.com/r/airbyte/source-postgres" } diff --git a/airbyte-config/init/src/main/resources/config/STANDARD_SOURCE_DEFINITION/e87ffa8e-a3b5-f69c-9076-6011339de1f6.json b/airbyte-config/init/src/main/resources/config/STANDARD_SOURCE_DEFINITION/e87ffa8e-a3b5-f69c-9076-6011339de1f6.json index 3075ad381c4d..52c2dbcca8d2 100644 --- a/airbyte-config/init/src/main/resources/config/STANDARD_SOURCE_DEFINITION/e87ffa8e-a3b5-f69c-9076-6011339de1f6.json +++ b/airbyte-config/init/src/main/resources/config/STANDARD_SOURCE_DEFINITION/e87ffa8e-a3b5-f69c-9076-6011339de1f6.json @@ -2,6 +2,6 @@ "sourceDefinitionId": "e87ffa8e-a3b5-f69c-9076-6011339de1f6", "name": "Redshift", "dockerRepository": "airbyte/source-redshift", - "dockerImageTag": "0.2.1", + "dockerImageTag": "0.2.2", "documentationUrl": "https://hub.docker.com/repository/docker/airbyte/source-redshift" } diff --git a/airbyte-config/init/src/main/resources/seed/destination_definitions.yaml b/airbyte-config/init/src/main/resources/seed/destination_definitions.yaml index 36fd3026e537..4f6b907da328 100644 --- a/airbyte-config/init/src/main/resources/seed/destination_definitions.yaml +++ b/airbyte-config/init/src/main/resources/seed/destination_definitions.yaml @@ -1,35 +1,35 @@ - destinationDefinitionId: a625d593-bba5-4a1c-a53d-2d246268a816 name: Local JSON dockerRepository: airbyte/destination-local-json - dockerImageTag: 0.2.0 + dockerImageTag: 0.2.2 documentationUrl: https://docs.airbyte.io/integrations/destinations/local-json - destinationDefinitionId: 8be1cf83-fde1-477f-a4ad-318d23c9f3c6 name: Local CSV dockerRepository: airbyte/destination-csv - dockerImageTag: 0.2.0 + dockerImageTag: 0.2.2 documentationUrl: https://docs.airbyte.io/integrations/destinations/local-csv - destinationDefinitionId: 25c5221d-dce2-4163-ade9-739ef790f503 name: Postgres dockerRepository: airbyte/destination-postgres - dockerImageTag: 0.2.0 + dockerImageTag: 0.2.2 documentationUrl: https://docs.airbyte.io/integrations/destinations/postgres - destinationDefinitionId: 22f6c74f-5699-40ff-833c-4a879ea40133 name: BigQuery dockerRepository: airbyte/destination-bigquery - dockerImageTag: 0.2.0 + dockerImageTag: 0.2.2 documentationUrl: https://docs.airbyte.io/integrations/destinations/bigquery - destinationDefinitionId: 424892c4-daac-4491-b35d-c6688ba547ba name: Snowflake dockerRepository: airbyte/destination-snowflake - dockerImageTag: 0.2.0 + dockerImageTag: 0.2.2 documentationUrl: https://docs.airbyte.io/integrations/destinations/snowflake - destinationDefinitionId: f7a7d195-377f-cf5b-70a5-be6b819019dc name: Redshift dockerRepository: airbyte/destination-redshift - dockerImageTag: 0.2.0 + dockerImageTag: 0.2.2 documentationUrl: https://docs.airbyte.io/integrations/destinations/redshift - destinationDefinitionId: af7c921e-5892-4ff2-b6c1-4a5ab258fb7e name: MeiliSearch dockerRepository: airbyte/destination-meilisearch - dockerImageTag: 0.2.0 + dockerImageTag: 0.2.2 documentationUrl: https://docs.airbyte.io/integrations/destinations/meilisearch diff --git a/airbyte-config/init/src/main/resources/seed/source_definitions.yaml b/airbyte-config/init/src/main/resources/seed/source_definitions.yaml index ea9ca5a0a151..e4b5dd350279 100644 --- a/airbyte-config/init/src/main/resources/seed/source_definitions.yaml +++ b/airbyte-config/init/src/main/resources/seed/source_definitions.yaml @@ -21,12 +21,12 @@ - sourceDefinitionId: b5ea17b1-f170-46dc-bc31-cc744ca984c1 name: Microsoft SQL Server (MSSQL) dockerRepository: airbyte/source-mssql - dockerImageTag: 0.2.1 + dockerImageTag: 0.2.2 documentationUrl: https://hub.docker.com/r/airbyte/source-mssql - sourceDefinitionId: decd338e-5647-4c0b-adf4-da0e75f5a750 name: Postgres dockerRepository: airbyte/source-postgres - dockerImageTag: 0.2.1 + dockerImageTag: 0.2.2 documentationUrl: https://hub.docker.com/r/airbyte/source-postgres - sourceDefinitionId: cd42861b-01fc-4658-a8ab-5d11d0510f01 name: Recurly @@ -51,7 +51,7 @@ - sourceDefinitionId: 435bb9a5-7887-4809-aa58-28c27df0d7ad name: MySQL dockerRepository: airbyte/source-mysql - dockerImageTag: 0.2.1 + dockerImageTag: 0.2.2 documentationUrl: https://docs.airbyte.io/integrations/sources/mysql - sourceDefinitionId: 2470e835-feaf-4db6-96f3-70fd645acc77 name: Salesforce @@ -96,7 +96,7 @@ - sourceDefinitionId: e87ffa8e-a3b5-f69c-9076-6011339de1f6 name: Redshift dockerRepository: airbyte/source-redshift - dockerImageTag: 0.2.1 + dockerImageTag: 0.2.2 documentationUrl: https://hub.docker.com/repository/docker/airbyte/source-redshift - sourceDefinitionId: 932e6363-d006-4464-a9f5-102b82e07c06 name: Twilio diff --git a/airbyte-config/models/src/main/resources/types/DestinationSyncMode.yaml b/airbyte-config/models/src/main/resources/types/DestinationSyncMode.yaml index eea319053044..0abb8f4920dd 100644 --- a/airbyte-config/models/src/main/resources/types/DestinationSyncMode.yaml +++ b/airbyte-config/models/src/main/resources/types/DestinationSyncMode.yaml @@ -8,4 +8,4 @@ enum: - append - overwrite #- upsert_dedup # TODO chris: SCD Type 1 can be implemented later - - append_dedup # SCD Type 2 + - append_dedup # SCD Type 1 & 2 diff --git a/airbyte-integrations/bases/airbyte-protocol/airbyte_protocol/models/airbyte_protocol.py b/airbyte-integrations/bases/airbyte-protocol/airbyte_protocol/models/airbyte_protocol.py index 71d509a7cce9..c73b38499ac3 100644 --- a/airbyte-integrations/bases/airbyte-protocol/airbyte_protocol/models/airbyte_protocol.py +++ b/airbyte-integrations/bases/airbyte-protocol/airbyte_protocol/models/airbyte_protocol.py @@ -141,12 +141,12 @@ class Config: extra = Extra.allow stream: AirbyteStream - sync_mode: Optional[SyncMode] = "full_refresh" + sync_mode: SyncMode cursor_field: Optional[List[str]] = Field( None, description="Path to the field that will be used to determine if a record is new or modified since the last sync. This field is REQUIRED if `sync_mode` is `incremental`. Otherwise it is ignored.", ) - destination_sync_mode: Optional[DestinationSyncMode] = "append" + destination_sync_mode: DestinationSyncMode primary_key: Optional[List[str]] = Field( None, description="Paths to the fields that will be used as primary key. This field is REQUIRED if `destination_sync_mode` is `*_dedup`. Otherwise it is ignored.", diff --git a/airbyte-integrations/bases/base-java/src/main/java/io/airbyte/integrations/destination/WriteConfig.java b/airbyte-integrations/bases/base-java/src/main/java/io/airbyte/integrations/destination/WriteConfig.java index 9b377c075523..685e494e3e64 100644 --- a/airbyte-integrations/bases/base-java/src/main/java/io/airbyte/integrations/destination/WriteConfig.java +++ b/airbyte-integrations/bases/base-java/src/main/java/io/airbyte/integrations/destination/WriteConfig.java @@ -24,7 +24,7 @@ package io.airbyte.integrations.destination; -import io.airbyte.protocol.models.SyncMode; +import io.airbyte.protocol.models.ConfiguredAirbyteStream.DestinationSyncMode; public class WriteConfig { @@ -32,9 +32,9 @@ public class WriteConfig { private final String outputNamespaceName; private final String tmpTableName; private final String outputTableName; - private final SyncMode syncMode; + private final DestinationSyncMode syncMode; - public WriteConfig(String streamName, String outputNamespaceName, String tmpTableName, String outputTableName, SyncMode syncMode) { + public WriteConfig(String streamName, String outputNamespaceName, String tmpTableName, String outputTableName, DestinationSyncMode syncMode) { this.streamName = streamName; this.outputNamespaceName = outputNamespaceName; this.tmpTableName = tmpTableName; @@ -58,7 +58,7 @@ public String getOutputTableName() { return outputTableName; } - public SyncMode getSyncMode() { + public DestinationSyncMode getSyncMode() { return syncMode; } diff --git a/airbyte-integrations/bases/standard-destination-test/src/main/java/io/airbyte/integrations/standardtest/destination/TestDestination.java b/airbyte-integrations/bases/standard-destination-test/src/main/java/io/airbyte/integrations/standardtest/destination/TestDestination.java index 6ff266eaf6e6..cc5eeffa104f 100644 --- a/airbyte-integrations/bases/standard-destination-test/src/main/java/io/airbyte/integrations/standardtest/destination/TestDestination.java +++ b/airbyte-integrations/bases/standard-destination-test/src/main/java/io/airbyte/integrations/standardtest/destination/TestDestination.java @@ -48,6 +48,7 @@ import io.airbyte.protocol.models.AirbyteStream; import io.airbyte.protocol.models.CatalogHelpers; import io.airbyte.protocol.models.ConfiguredAirbyteCatalog; +import io.airbyte.protocol.models.ConfiguredAirbyteStream.DestinationSyncMode; import io.airbyte.protocol.models.ConnectorSpecification; import io.airbyte.protocol.models.SyncMode; import io.airbyte.workers.DefaultCheckConnectionWorker; @@ -292,7 +293,10 @@ public void testIncrementalSync(String messagesFilename, String catalogFilename) final AirbyteCatalog catalog = Jsons.deserialize(MoreResources.readResource(catalogFilename), AirbyteCatalog.class); final ConfiguredAirbyteCatalog configuredCatalog = CatalogHelpers.toDefaultConfiguredCatalog(catalog); - configuredCatalog.getStreams().forEach(s -> s.withSyncMode(SyncMode.INCREMENTAL)); + configuredCatalog.getStreams().forEach(s -> { + s.withSyncMode(SyncMode.INCREMENTAL); + s.withDestinationSyncMode(DestinationSyncMode.APPEND); + }); final List firstSyncMessages = MoreResources.readResource(messagesFilename).lines() .map(record -> Jsons.deserialize(record, AirbyteMessage.class)).collect(Collectors.toList()); runSync(getConfig(), firstSyncMessages, configuredCatalog); diff --git a/airbyte-integrations/bases/standard-source-test/src/main/java/io/airbyte/integrations/standardtest/source/StandardSourceTest.java b/airbyte-integrations/bases/standard-source-test/src/main/java/io/airbyte/integrations/standardtest/source/StandardSourceTest.java index a5ce8512435d..f18cdf3467a1 100644 --- a/airbyte-integrations/bases/standard-source-test/src/main/java/io/airbyte/integrations/standardtest/source/StandardSourceTest.java +++ b/airbyte-integrations/bases/standard-source-test/src/main/java/io/airbyte/integrations/standardtest/source/StandardSourceTest.java @@ -48,6 +48,7 @@ import io.airbyte.protocol.models.AirbyteStateMessage; import io.airbyte.protocol.models.ConfiguredAirbyteCatalog; import io.airbyte.protocol.models.ConfiguredAirbyteStream; +import io.airbyte.protocol.models.ConfiguredAirbyteStream.DestinationSyncMode; import io.airbyte.protocol.models.ConnectorSpecification; import io.airbyte.workers.DefaultCheckConnectionWorker; import io.airbyte.workers.DefaultDiscoverCatalogWorker; @@ -378,6 +379,7 @@ private ConfiguredAirbyteCatalog withFullRefreshSyncModes(ConfiguredAirbyteCatal for (ConfiguredAirbyteStream configuredStream : clone.getStreams()) { if (configuredStream.getStream().getSupportedSyncModes().contains(FULL_REFRESH)) { configuredStream.setSyncMode(FULL_REFRESH); + configuredStream.setDestinationSyncMode(DestinationSyncMode.OVERWRITE); } } return clone; diff --git a/airbyte-integrations/connectors/destination-bigquery/Dockerfile b/airbyte-integrations/connectors/destination-bigquery/Dockerfile index 8dde87b24737..5412f0366844 100644 --- a/airbyte-integrations/connectors/destination-bigquery/Dockerfile +++ b/airbyte-integrations/connectors/destination-bigquery/Dockerfile @@ -8,5 +8,5 @@ COPY build/distributions/${APPLICATION}*.tar ${APPLICATION}.tar RUN tar xf ${APPLICATION}.tar --strip-components=1 -LABEL io.airbyte.version=0.2.0 +LABEL io.airbyte.version=0.2.2 LABEL io.airbyte.name=airbyte/destination-bigquery diff --git a/airbyte-integrations/connectors/destination-bigquery/src/main/java/io/airbyte/integrations/destination/bigquery/BigQueryDestination.java b/airbyte-integrations/connectors/destination-bigquery/src/main/java/io/airbyte/integrations/destination/bigquery/BigQueryDestination.java index f9fa5c3571cb..24f8f296d233 100644 --- a/airbyte-integrations/connectors/destination-bigquery/src/main/java/io/airbyte/integrations/destination/bigquery/BigQueryDestination.java +++ b/airbyte-integrations/connectors/destination-bigquery/src/main/java/io/airbyte/integrations/destination/bigquery/BigQueryDestination.java @@ -64,8 +64,8 @@ import io.airbyte.protocol.models.AirbyteMessage; import io.airbyte.protocol.models.ConfiguredAirbyteCatalog; import io.airbyte.protocol.models.ConfiguredAirbyteStream; +import io.airbyte.protocol.models.ConfiguredAirbyteStream.DestinationSyncMode; import io.airbyte.protocol.models.ConnectorSpecification; -import io.airbyte.protocol.models.SyncMode; import java.io.ByteArrayInputStream; import java.io.IOException; import java.nio.ByteBuffer; @@ -231,7 +231,7 @@ public DestinationConsumer write(JsonNode config, ConfiguredAirb .setFormatOptions(FormatOptions.json()).build(); // new-line delimited json. final TableDataWriteChannel writer = bigquery.writer(JobId.of(UUID.randomUUID().toString()), writeChannelConfiguration); - final WriteDisposition syncMode = getWriteDisposition(stream.getSyncMode()); + final WriteDisposition syncMode = getWriteDisposition(stream.getDestinationSyncMode()); writeConfigs.put(stream.getStream().getName(), new WriteConfig(TableId.of(schemaName, tableName), TableId.of(schemaName, tmpTableName), writer, syncMode)); @@ -242,13 +242,18 @@ public DestinationConsumer write(JsonNode config, ConfiguredAirb return new RecordConsumer(bigquery, writeConfigs, catalog); } - private static WriteDisposition getWriteDisposition(SyncMode syncMode) { - if (syncMode == null || syncMode == SyncMode.FULL_REFRESH) { - return WriteDisposition.WRITE_TRUNCATE; - } else if (syncMode == SyncMode.INCREMENTAL) { - return WriteDisposition.WRITE_APPEND; - } else { - throw new IllegalStateException("Unrecognized sync mode: " + syncMode); + private static WriteDisposition getWriteDisposition(DestinationSyncMode syncMode) { + if (syncMode == null) { + throw new IllegalStateException("Undefined destination sync mode"); + } + switch (syncMode) { + case OVERWRITE -> { + return WriteDisposition.WRITE_TRUNCATE; + } + case APPEND, APPEND_DEDUP -> { + return WriteDisposition.WRITE_APPEND; + } + default -> throw new IllegalStateException("Unrecognized destination sync mode: " + syncMode); } } diff --git a/airbyte-integrations/connectors/destination-csv/Dockerfile b/airbyte-integrations/connectors/destination-csv/Dockerfile index 954db370615d..703bf1679067 100644 --- a/airbyte-integrations/connectors/destination-csv/Dockerfile +++ b/airbyte-integrations/connectors/destination-csv/Dockerfile @@ -7,5 +7,5 @@ COPY build/distributions/${APPLICATION}*.tar ${APPLICATION}.tar RUN tar xf ${APPLICATION}.tar --strip-components=1 -LABEL io.airbyte.version=0.2.0 +LABEL io.airbyte.version=0.2.2 LABEL io.airbyte.name=airbyte/destination-csv diff --git a/airbyte-integrations/connectors/destination-csv/src/main/java/io/airbyte/integrations/destination/csv/CsvDestination.java b/airbyte-integrations/connectors/destination-csv/src/main/java/io/airbyte/integrations/destination/csv/CsvDestination.java index 1f6e4f17b08f..c3697be60877 100644 --- a/airbyte-integrations/connectors/destination-csv/src/main/java/io/airbyte/integrations/destination/csv/CsvDestination.java +++ b/airbyte-integrations/connectors/destination-csv/src/main/java/io/airbyte/integrations/destination/csv/CsvDestination.java @@ -39,8 +39,8 @@ import io.airbyte.protocol.models.AirbyteMessage; import io.airbyte.protocol.models.ConfiguredAirbyteCatalog; import io.airbyte.protocol.models.ConfiguredAirbyteStream; +import io.airbyte.protocol.models.ConfiguredAirbyteStream.DestinationSyncMode; import io.airbyte.protocol.models.ConnectorSpecification; -import io.airbyte.protocol.models.SyncMode; import java.io.FileWriter; import java.io.IOException; import java.nio.file.Files; @@ -105,12 +105,16 @@ public DestinationConsumer write(JsonNode config, ConfiguredAirb final Path finalPath = destinationDir.resolve(tableName + ".csv"); CSVFormat csvFormat = CSVFormat.DEFAULT.withHeader(JavaBaseConstants.COLUMN_NAME_AB_ID, JavaBaseConstants.COLUMN_NAME_EMITTED_AT, JavaBaseConstants.COLUMN_NAME_DATA); - final boolean isIncremental = stream.getSyncMode() == SyncMode.INCREMENTAL; - if (isIncremental && finalPath.toFile().exists()) { + final DestinationSyncMode syncMode = stream.getDestinationSyncMode(); + if (syncMode == null) { + throw new IllegalStateException("Undefined destination sync mode"); + } + final boolean isAppendMode = syncMode != DestinationSyncMode.OVERWRITE; + if (isAppendMode && finalPath.toFile().exists()) { Files.copy(finalPath, tmpPath, StandardCopyOption.REPLACE_EXISTING); csvFormat = csvFormat.withSkipHeaderRecord(); } - final FileWriter fileWriter = new FileWriter(tmpPath.toFile(), isIncremental); + final FileWriter fileWriter = new FileWriter(tmpPath.toFile(), isAppendMode); final CSVPrinter printer = new CSVPrinter(fileWriter, csvFormat); writeConfigs.put(stream.getStream().getName(), new WriteConfig(printer, tmpPath, finalPath)); } diff --git a/airbyte-integrations/connectors/destination-jdbc/Dockerfile b/airbyte-integrations/connectors/destination-jdbc/Dockerfile index 60827e128748..2f3681605029 100644 --- a/airbyte-integrations/connectors/destination-jdbc/Dockerfile +++ b/airbyte-integrations/connectors/destination-jdbc/Dockerfile @@ -8,5 +8,5 @@ COPY build/distributions/${APPLICATION}*.tar ${APPLICATION}.tar RUN tar xf ${APPLICATION}.tar --strip-components=1 -LABEL io.airbyte.version=0.2.0 +LABEL io.airbyte.version=0.2.1 LABEL io.airbyte.name=airbyte/destination-jdbc diff --git a/airbyte-integrations/connectors/destination-jdbc/src/main/java/io/airbyte/integrations/destination/jdbc/JdbcBufferedConsumerFactory.java b/airbyte-integrations/connectors/destination-jdbc/src/main/java/io/airbyte/integrations/destination/jdbc/JdbcBufferedConsumerFactory.java index acc6dfacdb2a..7812811a2546 100644 --- a/airbyte-integrations/connectors/destination-jdbc/src/main/java/io/airbyte/integrations/destination/jdbc/JdbcBufferedConsumerFactory.java +++ b/airbyte-integrations/connectors/destination-jdbc/src/main/java/io/airbyte/integrations/destination/jdbc/JdbcBufferedConsumerFactory.java @@ -38,7 +38,7 @@ import io.airbyte.integrations.destination.buffered_stream_consumer.BufferedStreamConsumer.RecordWriter; import io.airbyte.protocol.models.AirbyteMessage; import io.airbyte.protocol.models.ConfiguredAirbyteCatalog; -import io.airbyte.protocol.models.SyncMode; +import io.airbyte.protocol.models.ConfiguredAirbyteStream.DestinationSyncMode; import java.time.Instant; import java.util.List; import java.util.Map; @@ -84,7 +84,10 @@ private static List createWriteConfigs(NamingConventionTransformer final String schemaName = namingResolver.getIdentifier(config.get("schema").asText()); final String tableName = Names.concatQuotedNames("_airbyte_raw_", namingResolver.getIdentifier(streamName)); final String tmpTableName = Names.concatQuotedNames("_airbyte_" + now.toEpochMilli() + "_", tableName); - final SyncMode syncMode = stream.getSyncMode() != null ? stream.getSyncMode() : SyncMode.FULL_REFRESH; + final DestinationSyncMode syncMode = stream.getDestinationSyncMode(); + if (syncMode == null) { + throw new IllegalStateException("Undefined destination sync mode"); + } return new WriteConfig(streamName, schemaName, tmpTableName, tableName, syncMode); }).collect(Collectors.toList()); } @@ -138,8 +141,9 @@ private static OnCloseFunction onCloseFunction(JdbcDatabase database, SqlOperati sqlOperations.createTableIfNotExists(database, schemaName, dstTableName); switch (writeConfig.getSyncMode()) { - case FULL_REFRESH -> queries.append(sqlOperations.truncateTableQuery(schemaName, dstTableName)); - case INCREMENTAL -> {} + case OVERWRITE -> queries.append(sqlOperations.truncateTableQuery(schemaName, dstTableName)); + case APPEND -> {} + case APPEND_DEDUP -> {} default -> throw new IllegalStateException("Unrecognized sync mode: " + writeConfig.getSyncMode()); } queries.append(sqlOperations.copyTableQuery(schemaName, srcTableName, dstTableName)); diff --git a/airbyte-integrations/connectors/destination-jdbc/src/test/java/io/airbyte/integrations/destination/jdbc/JdbcDestinationTest.java b/airbyte-integrations/connectors/destination-jdbc/src/test/java/io/airbyte/integrations/destination/jdbc/JdbcDestinationTest.java index f8ff30c74a10..fad6a91c8f6d 100644 --- a/airbyte-integrations/connectors/destination-jdbc/src/test/java/io/airbyte/integrations/destination/jdbc/JdbcDestinationTest.java +++ b/airbyte-integrations/connectors/destination-jdbc/src/test/java/io/airbyte/integrations/destination/jdbc/JdbcDestinationTest.java @@ -53,6 +53,7 @@ import io.airbyte.protocol.models.CatalogHelpers; import io.airbyte.protocol.models.ConfiguredAirbyteCatalog; import io.airbyte.protocol.models.ConfiguredAirbyteStream; +import io.airbyte.protocol.models.ConfiguredAirbyteStream.DestinationSyncMode; import io.airbyte.protocol.models.ConnectorSpecification; import io.airbyte.protocol.models.Field; import io.airbyte.protocol.models.Field.JsonSchemaPrimitive; @@ -183,7 +184,10 @@ void testWriteSuccess() throws Exception { @Test void testWriteIncremental() throws Exception { final ConfiguredAirbyteCatalog catalog = Jsons.clone(CATALOG); - catalog.getStreams().forEach(stream -> stream.withSyncMode(SyncMode.INCREMENTAL)); + catalog.getStreams().forEach(stream -> { + stream.withSyncMode(SyncMode.INCREMENTAL); + stream.withDestinationSyncMode(DestinationSyncMode.APPEND); + }); final JdbcDestination destination = new JdbcDestination(); final DestinationConsumer consumer = destination.write(config, catalog); diff --git a/airbyte-integrations/connectors/destination-local-json/Dockerfile b/airbyte-integrations/connectors/destination-local-json/Dockerfile index 3a485d8198ab..ba73a9eda9ed 100644 --- a/airbyte-integrations/connectors/destination-local-json/Dockerfile +++ b/airbyte-integrations/connectors/destination-local-json/Dockerfile @@ -7,5 +7,5 @@ COPY build/distributions/${APPLICATION}*.tar ${APPLICATION}.tar RUN tar xf ${APPLICATION}.tar --strip-components=1 -LABEL io.airbyte.version=0.2.0 +LABEL io.airbyte.version=0.2.2 LABEL io.airbyte.name=airbyte/destination-local-json diff --git a/airbyte-integrations/connectors/destination-local-json/src/main/java/io/airbyte/integrations/destination/local_json/LocalJsonDestination.java b/airbyte-integrations/connectors/destination-local-json/src/main/java/io/airbyte/integrations/destination/local_json/LocalJsonDestination.java index 96542a4f1750..087e3d47f8d6 100644 --- a/airbyte-integrations/connectors/destination-local-json/src/main/java/io/airbyte/integrations/destination/local_json/LocalJsonDestination.java +++ b/airbyte-integrations/connectors/destination-local-json/src/main/java/io/airbyte/integrations/destination/local_json/LocalJsonDestination.java @@ -40,8 +40,8 @@ import io.airbyte.protocol.models.AirbyteMessage; import io.airbyte.protocol.models.ConfiguredAirbyteCatalog; import io.airbyte.protocol.models.ConfiguredAirbyteStream; +import io.airbyte.protocol.models.ConfiguredAirbyteStream.DestinationSyncMode; import io.airbyte.protocol.models.ConnectorSpecification; -import io.airbyte.protocol.models.SyncMode; import java.io.FileWriter; import java.io.IOException; import java.io.Writer; @@ -101,13 +101,16 @@ public DestinationConsumer write(JsonNode config, ConfiguredAirb final String streamName = stream.getStream().getName(); final Path finalPath = destinationDir.resolve(namingResolver.getRawTableName(streamName) + ".jsonl"); final Path tmpPath = destinationDir.resolve(namingResolver.getTmpTableName(streamName) + ".jsonl"); - - final boolean isIncremental = stream.getSyncMode() == SyncMode.INCREMENTAL; - if (isIncremental && finalPath.toFile().exists()) { + final DestinationSyncMode syncMode = stream.getDestinationSyncMode(); + if (syncMode == null) { + throw new IllegalStateException("Undefined destination sync mode"); + } + final boolean isAppendMode = syncMode != DestinationSyncMode.OVERWRITE; + if (isAppendMode && finalPath.toFile().exists()) { Files.copy(finalPath, tmpPath, StandardCopyOption.REPLACE_EXISTING); } - final Writer writer = new FileWriter(tmpPath.toFile(), isIncremental); + final Writer writer = new FileWriter(tmpPath.toFile(), isAppendMode); writeConfigs.put(stream.getStream().getName(), new WriteConfig(writer, tmpPath, finalPath)); } diff --git a/airbyte-integrations/connectors/destination-meilisearch/Dockerfile b/airbyte-integrations/connectors/destination-meilisearch/Dockerfile index 6a7d69e4d2ce..3b9617b5c7db 100644 --- a/airbyte-integrations/connectors/destination-meilisearch/Dockerfile +++ b/airbyte-integrations/connectors/destination-meilisearch/Dockerfile @@ -8,5 +8,5 @@ COPY build/distributions/${APPLICATION}*.tar ${APPLICATION}.tar RUN tar xf ${APPLICATION}.tar --strip-components=1 -LABEL io.airbyte.version=0.2.0 +LABEL io.airbyte.version=0.2.2 LABEL io.airbyte.name=airbyte/destination-meilisearch diff --git a/airbyte-integrations/connectors/destination-meilisearch/src/main/java/io/airbyte/integrations/destination/meilisearch/MeiliSearchDestination.java b/airbyte-integrations/connectors/destination-meilisearch/src/main/java/io/airbyte/integrations/destination/meilisearch/MeiliSearchDestination.java index b20a930683cd..7bb84df3c773 100644 --- a/airbyte-integrations/connectors/destination-meilisearch/src/main/java/io/airbyte/integrations/destination/meilisearch/MeiliSearchDestination.java +++ b/airbyte-integrations/connectors/destination-meilisearch/src/main/java/io/airbyte/integrations/destination/meilisearch/MeiliSearchDestination.java @@ -44,7 +44,7 @@ import io.airbyte.protocol.models.CatalogHelpers; import io.airbyte.protocol.models.ConfiguredAirbyteCatalog; import io.airbyte.protocol.models.ConfiguredAirbyteStream; -import io.airbyte.protocol.models.SyncMode; +import io.airbyte.protocol.models.ConfiguredAirbyteStream.DestinationSyncMode; import java.time.Instant; import java.util.Arrays; import java.util.HashMap; @@ -119,8 +119,11 @@ private static Map createIndices(ConfiguredAirbyteCatalog catalog final Map map = new HashMap<>(); for (final ConfiguredAirbyteStream stream : catalog.getStreams()) { final String indexName = getIndexName(stream); - - if (stream.getSyncMode() == SyncMode.FULL_REFRESH && indexExists(client, indexName)) { + final DestinationSyncMode syncMode = stream.getDestinationSyncMode(); + if (syncMode == null) { + throw new IllegalStateException("Undefined destination sync mode"); + } + if (syncMode == DestinationSyncMode.OVERWRITE && indexExists(client, indexName)) { client.deleteIndex(indexName); } diff --git a/airbyte-integrations/connectors/destination-postgres/Dockerfile b/airbyte-integrations/connectors/destination-postgres/Dockerfile index 4eeada7bca97..6cdf71571a1c 100644 --- a/airbyte-integrations/connectors/destination-postgres/Dockerfile +++ b/airbyte-integrations/connectors/destination-postgres/Dockerfile @@ -8,5 +8,5 @@ COPY build/distributions/${APPLICATION}*.tar ${APPLICATION}.tar RUN tar xf ${APPLICATION}.tar --strip-components=1 -LABEL io.airbyte.version=0.2.0 +LABEL io.airbyte.version=0.2.2 LABEL io.airbyte.name=airbyte/destination-postgres diff --git a/airbyte-integrations/connectors/destination-postgres/src/test/java/io/airbyte/integrations/destination/postgres/PostgresDestinationTest.java b/airbyte-integrations/connectors/destination-postgres/src/test/java/io/airbyte/integrations/destination/postgres/PostgresDestinationTest.java index a375d1e300a0..9bfcaa78a8db 100644 --- a/airbyte-integrations/connectors/destination-postgres/src/test/java/io/airbyte/integrations/destination/postgres/PostgresDestinationTest.java +++ b/airbyte-integrations/connectors/destination-postgres/src/test/java/io/airbyte/integrations/destination/postgres/PostgresDestinationTest.java @@ -52,6 +52,7 @@ import io.airbyte.protocol.models.CatalogHelpers; import io.airbyte.protocol.models.ConfiguredAirbyteCatalog; import io.airbyte.protocol.models.ConfiguredAirbyteStream; +import io.airbyte.protocol.models.ConfiguredAirbyteStream.DestinationSyncMode; import io.airbyte.protocol.models.ConnectorSpecification; import io.airbyte.protocol.models.Field; import io.airbyte.protocol.models.Field.JsonSchemaPrimitive; @@ -186,7 +187,10 @@ void testWriteSuccess() throws Exception { @Test void testWriteIncremental() throws Exception { final ConfiguredAirbyteCatalog catalog = Jsons.clone(CATALOG); - catalog.getStreams().forEach(stream -> stream.withSyncMode(SyncMode.INCREMENTAL)); + catalog.getStreams().forEach(stream -> { + stream.withSyncMode(SyncMode.INCREMENTAL); + stream.withDestinationSyncMode(DestinationSyncMode.APPEND); + }); final PostgresDestination destination = new PostgresDestination(); final DestinationConsumer consumer = destination.write(config, catalog); diff --git a/airbyte-integrations/connectors/destination-redshift/Dockerfile b/airbyte-integrations/connectors/destination-redshift/Dockerfile index d630505655ab..57953c58eaaa 100644 --- a/airbyte-integrations/connectors/destination-redshift/Dockerfile +++ b/airbyte-integrations/connectors/destination-redshift/Dockerfile @@ -8,5 +8,5 @@ COPY build/distributions/${APPLICATION}*.tar ${APPLICATION}.tar RUN tar xf ${APPLICATION}.tar --strip-components=1 -LABEL io.airbyte.version=0.2.0 +LABEL io.airbyte.version=0.2.2 LABEL io.airbyte.name=airbyte/destination-redshift diff --git a/airbyte-integrations/connectors/destination-snowflake/Dockerfile b/airbyte-integrations/connectors/destination-snowflake/Dockerfile index 27485fcaa37f..658bba2e508b 100644 --- a/airbyte-integrations/connectors/destination-snowflake/Dockerfile +++ b/airbyte-integrations/connectors/destination-snowflake/Dockerfile @@ -8,5 +8,5 @@ COPY build/distributions/${APPLICATION}*.tar ${APPLICATION}.tar RUN tar xf ${APPLICATION}.tar --strip-components=1 -LABEL io.airbyte.version=0.2.0 +LABEL io.airbyte.version=0.2.2 LABEL io.airbyte.name=airbyte/destination-snowflake diff --git a/airbyte-integrations/connectors/source-google-sheets/unit_tests/test_helpers.py b/airbyte-integrations/connectors/source-google-sheets/unit_tests/test_helpers.py index 88f773b1783a..9443590bb39f 100644 --- a/airbyte-integrations/connectors/source-google-sheets/unit_tests/test_helpers.py +++ b/airbyte-integrations/connectors/source-google-sheets/unit_tests/test_helpers.py @@ -25,7 +25,8 @@ import unittest from unittest.mock import Mock, patch -from airbyte_protocol import AirbyteRecordMessage, AirbyteStream, ConfiguredAirbyteCatalog, ConfiguredAirbyteStream +from airbyte_protocol import AirbyteRecordMessage, AirbyteStream, ConfiguredAirbyteCatalog, ConfiguredAirbyteStream, SyncMode +from airbyte_protocol.models.airbyte_protocol import DestinationSyncMode from google_sheets_source.client import GoogleSheetsClient from google_sheets_source.helpers import Helpers from google_sheets_source.models import CellData, GridData, RowData, Sheet, SheetProperties, Spreadsheet @@ -103,8 +104,16 @@ def test_parse_sheet_and_column_names_from_catalog(self): catalog = ConfiguredAirbyteCatalog( streams=[ - ConfiguredAirbyteStream(stream=AirbyteStream(name=sheet1, json_schema=sheet1_schema)), - ConfiguredAirbyteStream(stream=AirbyteStream(name=sheet2, json_schema=sheet2_schema)), + ConfiguredAirbyteStream( + stream=AirbyteStream(name=sheet1, json_schema=sheet1_schema), + sync_mode=SyncMode.full_refresh, + destination_sync_mode=DestinationSyncMode.overwrite, + ), + ConfiguredAirbyteStream( + stream=AirbyteStream(name=sheet2, json_schema=sheet2_schema), + sync_mode=SyncMode.full_refresh, + destination_sync_mode=DestinationSyncMode.overwrite, + ), ] ) diff --git a/airbyte-integrations/connectors/source-jdbc/src/main/java/io/airbyte/integrations/source/jdbc/AbstractJdbcSource.java b/airbyte-integrations/connectors/source-jdbc/src/main/java/io/airbyte/integrations/source/jdbc/AbstractJdbcSource.java index f72be1ba4de4..bb2485060025 100644 --- a/airbyte-integrations/connectors/source-jdbc/src/main/java/io/airbyte/integrations/source/jdbc/AbstractJdbcSource.java +++ b/airbyte-integrations/connectors/source-jdbc/src/main/java/io/airbyte/integrations/source/jdbc/AbstractJdbcSource.java @@ -274,10 +274,12 @@ private AutoCloseableIterator createReadIterator(JdbcDatabase da cursorOptional.orElse(null), cursorType), airbyteMessageIterator); - } else if (airbyteStream.getSyncMode() == SyncMode.FULL_REFRESH || airbyteStream.getSyncMode() == null) { + } else if (airbyteStream.getSyncMode() == SyncMode.FULL_REFRESH) { iterator = getFullRefreshStream(database, streamName, selectedDatabaseFields, table, emittedAt); + } else if (airbyteStream.getSyncMode() == null) { + throw new IllegalArgumentException(String.format("%s requires a source sync mode", AbstractJdbcSource.class)); } else { - throw new IllegalArgumentException(String.format("%s does not support sync mode: %s.", airbyteStream.getSyncMode(), AbstractJdbcSource.class)); + throw new IllegalArgumentException(String.format("%s does not support sync mode: %s.", AbstractJdbcSource.class, airbyteStream.getSyncMode())); } final AtomicLong recordCount = new AtomicLong(); diff --git a/airbyte-integrations/connectors/source-jdbc/src/testFixtures/java/io/airbyte/integrations/source/jdbc/test/JdbcSourceStandardTest.java b/airbyte-integrations/connectors/source-jdbc/src/testFixtures/java/io/airbyte/integrations/source/jdbc/test/JdbcSourceStandardTest.java index 17b17fbcf9aa..87e234821aa4 100644 --- a/airbyte-integrations/connectors/source-jdbc/src/testFixtures/java/io/airbyte/integrations/source/jdbc/test/JdbcSourceStandardTest.java +++ b/airbyte-integrations/connectors/source-jdbc/src/testFixtures/java/io/airbyte/integrations/source/jdbc/test/JdbcSourceStandardTest.java @@ -56,6 +56,7 @@ import io.airbyte.protocol.models.CatalogHelpers; import io.airbyte.protocol.models.ConfiguredAirbyteCatalog; import io.airbyte.protocol.models.ConfiguredAirbyteStream; +import io.airbyte.protocol.models.ConfiguredAirbyteStream.DestinationSyncMode; import io.airbyte.protocol.models.ConnectorSpecification; import io.airbyte.protocol.models.Field; import io.airbyte.protocol.models.Field.JsonSchemaPrimitive; @@ -437,6 +438,7 @@ void testReadOneTableIncrementallyTwice() throws Exception { configuredCatalog.getStreams().forEach(airbyteStream -> { airbyteStream.setSyncMode(SyncMode.INCREMENTAL); airbyteStream.setCursorField(Lists.newArrayList("id")); + airbyteStream.setDestinationSyncMode(DestinationSyncMode.APPEND); }); final JdbcState state = new JdbcState().withStreams(Lists.newArrayList(new JdbcStreamState().withStreamName(streamName))); @@ -492,6 +494,7 @@ void testReadMultipleTablesIncrementally() throws Exception { configuredCatalog.getStreams().forEach(airbyteStream -> { airbyteStream.setSyncMode(SyncMode.INCREMENTAL); airbyteStream.setCursorField(Lists.newArrayList("id")); + airbyteStream.setDestinationSyncMode(DestinationSyncMode.APPEND); }); final JdbcState state = new JdbcState().withStreams(Lists.newArrayList(new JdbcStreamState().withStreamName(streamName))); @@ -576,6 +579,7 @@ private void incrementalCursorCheck( throws Exception { airbyteStream.setSyncMode(SyncMode.INCREMENTAL); airbyteStream.setCursorField(Lists.newArrayList(cursorField)); + airbyteStream.setDestinationSyncMode(DestinationSyncMode.APPEND); final JdbcState state = new JdbcState() .withStreams(Lists.newArrayList(new JdbcStreamState() diff --git a/airbyte-integrations/connectors/source-jdbc/src/testFixtures/java/io/airbyte/integrations/source/jdbc/test/JdbcStressTest.java b/airbyte-integrations/connectors/source-jdbc/src/testFixtures/java/io/airbyte/integrations/source/jdbc/test/JdbcStressTest.java index a7d38a06cbe1..94970a91a9ae 100644 --- a/airbyte-integrations/connectors/source-jdbc/src/testFixtures/java/io/airbyte/integrations/source/jdbc/test/JdbcStressTest.java +++ b/airbyte-integrations/connectors/source-jdbc/src/testFixtures/java/io/airbyte/integrations/source/jdbc/test/JdbcStressTest.java @@ -42,6 +42,7 @@ import io.airbyte.protocol.models.CatalogHelpers; import io.airbyte.protocol.models.ConfiguredAirbyteCatalog; import io.airbyte.protocol.models.ConfiguredAirbyteStream; +import io.airbyte.protocol.models.ConfiguredAirbyteStream.DestinationSyncMode; import io.airbyte.protocol.models.Field; import io.airbyte.protocol.models.Field.JsonSchemaPrimitive; import io.airbyte.protocol.models.SyncMode; @@ -199,7 +200,8 @@ private static ConfiguredAirbyteCatalog getConfiguredCatalogIncremental() { return new ConfiguredAirbyteCatalog() .withStreams(Collections.singletonList(new ConfiguredAirbyteStream().withStream(getCatalog().getStreams().get(0)) .withCursorField(Collections.singletonList("id")) - .withSyncMode(SyncMode.INCREMENTAL))); + .withSyncMode(SyncMode.INCREMENTAL) + .withDestinationSyncMode(DestinationSyncMode.APPEND))); } private static AirbyteCatalog getCatalog() { diff --git a/airbyte-integrations/connectors/source-mssql/Dockerfile b/airbyte-integrations/connectors/source-mssql/Dockerfile index 9f9095f781df..fdedfcf23543 100644 --- a/airbyte-integrations/connectors/source-mssql/Dockerfile +++ b/airbyte-integrations/connectors/source-mssql/Dockerfile @@ -8,5 +8,5 @@ COPY build/distributions/${APPLICATION}*.tar ${APPLICATION}.tar RUN tar xf ${APPLICATION}.tar --strip-components=1 -LABEL io.airbyte.version=0.2.1 +LABEL io.airbyte.version=0.2.2 LABEL io.airbyte.name=airbyte/source-mssql diff --git a/airbyte-integrations/connectors/source-mysql/Dockerfile b/airbyte-integrations/connectors/source-mysql/Dockerfile index ed98d10c64bf..3043865f5d1b 100644 --- a/airbyte-integrations/connectors/source-mysql/Dockerfile +++ b/airbyte-integrations/connectors/source-mysql/Dockerfile @@ -8,5 +8,5 @@ COPY build/distributions/${APPLICATION}*.tar ${APPLICATION}.tar RUN tar xf ${APPLICATION}.tar --strip-components=1 -LABEL io.airbyte.version=0.2.1 +LABEL io.airbyte.version=0.2.2 LABEL io.airbyte.name=airbyte/source-mysql diff --git a/airbyte-integrations/connectors/source-mysql/src/test-integration/java/io/airbyte/integrations/source/mysql/MySqlStandardTest.java b/airbyte-integrations/connectors/source-mysql/src/test-integration/java/io/airbyte/integrations/source/mysql/MySqlStandardTest.java index e8e00f410dae..803b12b15ce7 100644 --- a/airbyte-integrations/connectors/source-mysql/src/test-integration/java/io/airbyte/integrations/source/mysql/MySqlStandardTest.java +++ b/airbyte-integrations/connectors/source-mysql/src/test-integration/java/io/airbyte/integrations/source/mysql/MySqlStandardTest.java @@ -35,6 +35,7 @@ import io.airbyte.protocol.models.CatalogHelpers; import io.airbyte.protocol.models.ConfiguredAirbyteCatalog; import io.airbyte.protocol.models.ConfiguredAirbyteStream; +import io.airbyte.protocol.models.ConfiguredAirbyteStream.DestinationSyncMode; import io.airbyte.protocol.models.ConnectorSpecification; import io.airbyte.protocol.models.Field; import io.airbyte.protocol.models.Field.JsonSchemaPrimitive; @@ -113,6 +114,7 @@ protected ConfiguredAirbyteCatalog getConfiguredCatalog() { new ConfiguredAirbyteStream() .withSyncMode(SyncMode.INCREMENTAL) .withCursorField(Lists.newArrayList("id")) + .withDestinationSyncMode(DestinationSyncMode.APPEND) .withStream(CatalogHelpers.createAirbyteStream( String.format("%s.%s", config.get("database").asText(), STREAM_NAME), Field.of("id", JsonSchemaPrimitive.NUMBER), @@ -121,6 +123,7 @@ protected ConfiguredAirbyteCatalog getConfiguredCatalog() { new ConfiguredAirbyteStream() .withSyncMode(SyncMode.INCREMENTAL) .withCursorField(Lists.newArrayList("id")) + .withDestinationSyncMode(DestinationSyncMode.APPEND) .withStream(CatalogHelpers.createAirbyteStream( String.format("%s.%s", config.get("database").asText(), STREAM_NAME2), Field.of("id", JsonSchemaPrimitive.NUMBER), diff --git a/airbyte-integrations/connectors/source-postgres/Dockerfile b/airbyte-integrations/connectors/source-postgres/Dockerfile index 8bf9208b97ef..75b4c6265a35 100644 --- a/airbyte-integrations/connectors/source-postgres/Dockerfile +++ b/airbyte-integrations/connectors/source-postgres/Dockerfile @@ -8,5 +8,5 @@ COPY build/distributions/${APPLICATION}*.tar ${APPLICATION}.tar RUN tar xf ${APPLICATION}.tar --strip-components=1 -LABEL io.airbyte.version=0.2.1 +LABEL io.airbyte.version=0.2.2 LABEL io.airbyte.name=airbyte/source-postgres diff --git a/airbyte-integrations/connectors/source-postgres/src/test-integration/java/io/airbyte/integrations/io/airbyte/integration_tests/sources/PostgresSourceStandardTest.java b/airbyte-integrations/connectors/source-postgres/src/test-integration/java/io/airbyte/integrations/io/airbyte/integration_tests/sources/PostgresSourceStandardTest.java index 5aba7e2228db..568ce9baa959 100644 --- a/airbyte-integrations/connectors/source-postgres/src/test-integration/java/io/airbyte/integrations/io/airbyte/integration_tests/sources/PostgresSourceStandardTest.java +++ b/airbyte-integrations/connectors/source-postgres/src/test-integration/java/io/airbyte/integrations/io/airbyte/integration_tests/sources/PostgresSourceStandardTest.java @@ -35,6 +35,7 @@ import io.airbyte.protocol.models.CatalogHelpers; import io.airbyte.protocol.models.ConfiguredAirbyteCatalog; import io.airbyte.protocol.models.ConfiguredAirbyteStream; +import io.airbyte.protocol.models.ConfiguredAirbyteStream.DestinationSyncMode; import io.airbyte.protocol.models.ConnectorSpecification; import io.airbyte.protocol.models.Field; import io.airbyte.protocol.models.Field.JsonSchemaPrimitive; @@ -113,6 +114,7 @@ protected ConfiguredAirbyteCatalog getConfiguredCatalog() { new ConfiguredAirbyteStream() .withSyncMode(SyncMode.INCREMENTAL) .withCursorField(Lists.newArrayList("id")) + .withDestinationSyncMode(DestinationSyncMode.APPEND) .withStream(CatalogHelpers.createAirbyteStream( STREAM_NAME, Field.of("id", JsonSchemaPrimitive.NUMBER), @@ -121,6 +123,7 @@ protected ConfiguredAirbyteCatalog getConfiguredCatalog() { new ConfiguredAirbyteStream() .withSyncMode(SyncMode.INCREMENTAL) .withCursorField(Lists.newArrayList("id")) + .withDestinationSyncMode(DestinationSyncMode.APPEND) .withStream(CatalogHelpers.createAirbyteStream( STREAM_NAME2, Field.of("id", JsonSchemaPrimitive.NUMBER), diff --git a/airbyte-integrations/connectors/source-redshift/Dockerfile b/airbyte-integrations/connectors/source-redshift/Dockerfile index faa4d5262ffd..494d0bf340bd 100644 --- a/airbyte-integrations/connectors/source-redshift/Dockerfile +++ b/airbyte-integrations/connectors/source-redshift/Dockerfile @@ -8,5 +8,5 @@ COPY build/distributions/${APPLICATION}*.tar ${APPLICATION}.tar RUN tar xf ${APPLICATION}.tar --strip-components=1 -LABEL io.airbyte.version=0.2.1 +LABEL io.airbyte.version=0.2.2 LABEL io.airbyte.name=airbyte/source-redshift diff --git a/airbyte-integrations/connectors/source-redshift/src/test-integration/java/io/airbyte/integrations/io/airbyte/integration_tests/sources/RedshiftStandardSourceTest.java b/airbyte-integrations/connectors/source-redshift/src/test-integration/java/io/airbyte/integrations/io/airbyte/integration_tests/sources/RedshiftStandardSourceTest.java index 4612257a4453..361daa78a2d4 100644 --- a/airbyte-integrations/connectors/source-redshift/src/test-integration/java/io/airbyte/integrations/io/airbyte/integration_tests/sources/RedshiftStandardSourceTest.java +++ b/airbyte-integrations/connectors/source-redshift/src/test-integration/java/io/airbyte/integrations/io/airbyte/integration_tests/sources/RedshiftStandardSourceTest.java @@ -41,14 +41,15 @@ import java.util.Collections; import java.util.HashMap; import java.util.List; +import org.apache.commons.lang3.RandomStringUtils; public class RedshiftStandardSourceTest extends StandardSourceTest { // This test case expects an active redshift cluster that is useable from outside of vpc - private static final String SCHEMA_NAME = "integration_test"; - private static final String STREAM_NAME = SCHEMA_NAME + ".customer"; private JsonNode config; private JdbcDatabase database; + private String schemaName; + private String streamName; private static JsonNode getStaticConfig() { return Jsons.deserialize(IOs.readFile(Path.of("secrets/config.json"))); @@ -56,7 +57,9 @@ private static JsonNode getStaticConfig() { @Override protected void setup(TestDestinationEnv testEnv) throws Exception { - final String createSchemaQuery = String.format("CREATE SCHEMA %s", SCHEMA_NAME); + schemaName = ("integration_test_" + RandomStringUtils.randomAlphanumeric(5)).toLowerCase(); + final String createSchemaQuery = String.format("CREATE SCHEMA %s", schemaName); + streamName = schemaName + ".customer"; config = getStaticConfig(); database = Databases.createJdbcDatabase( @@ -70,9 +73,9 @@ protected void setup(TestDestinationEnv testEnv) throws Exception { database.execute(connection -> connection.createStatement().execute(createSchemaQuery)); String createTestTable = - String.format("CREATE TABLE IF NOT EXISTS %s (c_custkey INTEGER, c_name VARCHAR(16), c_nation VARCHAR(16));\n", STREAM_NAME); + String.format("CREATE TABLE IF NOT EXISTS %s (c_custkey INTEGER, c_name VARCHAR(16), c_nation VARCHAR(16));\n", streamName); database.execute(connection -> connection.createStatement().execute(createTestTable)); - String insertTestData = String.format("insert into %s values (1, 'Chris', 'France');\n", STREAM_NAME); + String insertTestData = String.format("insert into %s values (1, 'Chris', 'France');\n", streamName); database.execute(connection -> { connection.createStatement().execute(insertTestData); System.out.println("more to be done."); @@ -81,7 +84,7 @@ protected void setup(TestDestinationEnv testEnv) throws Exception { @Override protected void tearDown(TestDestinationEnv testEnv) throws SQLException { - final String dropSchemaQuery = String.format("DROP SCHEMA IF EXISTS %s CASCADE", SCHEMA_NAME); + final String dropSchemaQuery = String.format("DROP SCHEMA IF EXISTS %s CASCADE", schemaName); database.execute(connection -> connection.createStatement().execute(dropSchemaQuery)); } @@ -103,7 +106,7 @@ protected JsonNode getConfig() { @Override protected ConfiguredAirbyteCatalog getConfiguredCatalog() { return CatalogHelpers.createConfiguredAirbyteCatalog( - STREAM_NAME, + streamName, Field.of("c_custkey", Field.JsonSchemaPrimitive.NUMBER), Field.of("c_name", Field.JsonSchemaPrimitive.STRING), Field.of("c_nation", Field.JsonSchemaPrimitive.STRING)); diff --git a/airbyte-migration/src/main/java/io/airbyte/migrate/Migrations.java b/airbyte-migration/src/main/java/io/airbyte/migrate/Migrations.java index a1947e14b09a..27b821887299 100644 --- a/airbyte-migration/src/main/java/io/airbyte/migrate/Migrations.java +++ b/airbyte-migration/src/main/java/io/airbyte/migrate/Migrations.java @@ -28,6 +28,7 @@ import io.airbyte.migrate.migrations.MigrationV0_14_0; import io.airbyte.migrate.migrations.MigrationV0_14_3; import io.airbyte.migrate.migrations.MigrationV0_17_0; +import io.airbyte.migrate.migrations.MigrationV0_18_0; import io.airbyte.migrate.migrations.NoOpMigration; import java.util.List; @@ -38,6 +39,7 @@ public class Migrations { private static final Migration MIGRATION_V_0_15_0 = new NoOpMigration(MIGRATION_V_0_14_3, "0.15.0-alpha"); private static final Migration MIGRATION_V_0_16_0 = new NoOpMigration(MIGRATION_V_0_15_0, "0.16.0-alpha"); private static final Migration MIGRATION_V_0_17_0 = new MigrationV0_17_0(MIGRATION_V_0_16_0); + private static final Migration MIGRATION_V_0_18_0 = new MigrationV0_18_0(MIGRATION_V_0_17_0); // all migrations must be added to the list in the order that they should be applied. public static final List MIGRATIONS = ImmutableList.of( @@ -45,6 +47,7 @@ public class Migrations { MIGRATION_V_0_14_3, MIGRATION_V_0_15_0, MIGRATION_V_0_16_0, - MIGRATION_V_0_17_0); + MIGRATION_V_0_17_0, + MIGRATION_V_0_18_0); } diff --git a/airbyte-migration/src/main/java/io/airbyte/migrate/migrations/MigrationV0_18_0.java b/airbyte-migration/src/main/java/io/airbyte/migrate/migrations/MigrationV0_18_0.java new file mode 100644 index 000000000000..bdece5c93275 --- /dev/null +++ b/airbyte-migration/src/main/java/io/airbyte/migrate/migrations/MigrationV0_18_0.java @@ -0,0 +1,190 @@ +/* + * MIT License + * + * Copyright (c) 2020 Airbyte + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package io.airbyte.migrate.migrations; + +import com.fasterxml.jackson.databind.JsonNode; +import com.fasterxml.jackson.databind.node.ObjectNode; +import com.google.common.collect.ImmutableMap; +import io.airbyte.commons.json.Jsons; +import io.airbyte.commons.util.MoreIterators; +import io.airbyte.migrate.Migration; +import io.airbyte.migrate.ResourceId; +import io.airbyte.migrate.ResourceType; +import java.util.Collections; +import java.util.List; +import java.util.Map; +import java.util.function.Consumer; +import java.util.stream.Collectors; +import java.util.stream.Stream; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +/** + * This migration makes sure that ConfiguredAirbyteCatalog always have values for the now required + * fields: syncMode (used by source to specify full_refresh/incremental) and destinationSyncMode + * (used by destination to specify append/overwrite/append_dedup) + * + * The primaryKey column is filled if available from the stream if defined by source + */ +public class MigrationV0_18_0 extends BaseMigration implements Migration { + + private static final Logger LOGGER = LoggerFactory.getLogger(MigrationV0_18_0.class); + + private static final ResourceId STANDARD_SYNC_RESOURCE_ID = ResourceId.fromConstantCase(ResourceType.CONFIG, "STANDARD_SYNC"); + + private static final String MIGRATION_VERSION = "0.18.0-alpha"; + + private final Migration previousMigration; + + public MigrationV0_18_0(Migration previousMigration) { + super(previousMigration); + this.previousMigration = previousMigration; + } + + @Override + public String getVersion() { + return MIGRATION_VERSION; + } + + @Override + public Map getOutputSchema() { + return previousMigration.getOutputSchema(); + } + + @Override + public void migrate(Map> inputData, Map> outputData) { + for (final Map.Entry> entry : inputData.entrySet()) { + final Consumer recordConsumer = outputData.get(entry.getKey()); + + entry.getValue().forEach(r -> { + if (entry.getKey().equals(STANDARD_SYNC_RESOURCE_ID)) { + ((ObjectNode) r).set("catalog", migrateCatalog(r.get("catalog"))); + } + recordConsumer.accept(r); + }); + } + } + + private JsonNode migrateCatalog(JsonNode catalog) { + final List> configuredStreams = MoreIterators.toList(catalog.get("streams").elements()) + .stream() + .map(stream -> { + final JsonNode airbyteStream = stream.get("stream"); + assert airbyteStream != null; + JsonNode syncMode = stream.get("sync_mode"); + if (syncMode == null) { + syncMode = Jsons.jsonNode(SyncMode.FULL_REFRESH.toString()); + LOGGER.info("Migrating {} to default source sync_mode: {}", airbyteStream.get("name"), syncMode); + } + JsonNode destinationSyncMode = stream.get("destination_sync_mode"); + if (destinationSyncMode == null) { + if (SyncMode.fromValue(syncMode.asText()) == SyncMode.FULL_REFRESH) { + destinationSyncMode = Jsons.jsonNode(DestinationSyncMode.OVERWRITE.toString()); + LOGGER.debug("Migrating {} to source sync_mode: {} destination_sync_mode: {}", airbyteStream.get("name"), syncMode, + destinationSyncMode); + } else if (SyncMode.fromValue(syncMode.asText()) == SyncMode.INCREMENTAL) { + destinationSyncMode = Jsons.jsonNode(DestinationSyncMode.APPEND.toString()); + LOGGER.debug("Migrating {} to source sync_mode: {} destination_sync_mode: {}", airbyteStream.get("name"), syncMode, + destinationSyncMode); + } else { + syncMode = Jsons.jsonNode(SyncMode.FULL_REFRESH.toString()); + destinationSyncMode = Jsons.jsonNode(DestinationSyncMode.OVERWRITE.toString()); + LOGGER.info("Migrating {} to default source sync_mode: {} destination_sync_mode: {}", airbyteStream.get("name"), syncMode, + destinationSyncMode); + } + } + JsonNode primaryKey = stream.get("primary_key"); + if (primaryKey == null) { + JsonNode sourceDefinedPrimaryKey = airbyteStream.get("source_defined_primary_key"); + primaryKey = sourceDefinedPrimaryKey != null ? sourceDefinedPrimaryKey : Jsons.jsonNode(Collections.emptyList()); + } + // configured catalog fields + return (Map) ImmutableMap.builder() + .put("stream", airbyteStream) + .put("sync_mode", syncMode) + .put("cursor_field", stream.get("cursor_field") != null ? stream.get("cursor_field") : Jsons.jsonNode(Collections.emptyList())) + .put("destination_sync_mode", destinationSyncMode) + .put("primary_key", primaryKey) + .build(); + }) + .collect(Collectors.toList()); + return Jsons.jsonNode(ImmutableMap.of("streams", configuredStreams)); + } + + public enum SyncMode { + + FULL_REFRESH("full_refresh"), + + INCREMENTAL("incremental"); + + private String value; + + SyncMode(String value) { + this.value = value; + } + + public String toString() { + return String.valueOf(value); + } + + public static SyncMode fromValue(String value) { + for (SyncMode b : SyncMode.values()) { + if (b.value.equals(value)) { + return b; + } + } + throw new IllegalArgumentException("Unexpected value '" + value + "'"); + } + + } + + public enum DestinationSyncMode { + + APPEND("append"), + OVERWRITE("overwrite"), + APPEND_DEDUP("append_dedup"); + + private final String value; + + private DestinationSyncMode(String value) { + this.value = value; + } + + public String toString() { + return String.valueOf(value); + } + + public static DestinationSyncMode fromValue(String value) { + for (DestinationSyncMode b : DestinationSyncMode.values()) { + if (b.value.equals(value)) { + return b; + } + } + throw new IllegalArgumentException("Unexpected value '" + value + "'"); + } + + } + +} diff --git a/airbyte-migration/src/main/resources/migrations/migrationV0_14_3/airbyte_config/AirbyteProtocolAbbreviated.yaml b/airbyte-migration/src/main/resources/migrations/migrationV0_14_3/airbyte_config/AirbyteProtocolAbbreviated.yaml deleted file mode 100644 index f3147a44ea6d..000000000000 --- a/airbyte-migration/src/main/resources/migrations/migrationV0_14_3/airbyte_config/AirbyteProtocolAbbreviated.yaml +++ /dev/null @@ -1,84 +0,0 @@ ---- -"$schema": http://json-schema.org/draft-07/schema# -"$id": https://github.com/airbytehq/airbyte/blob/master/airbyte-protocol/models/src/main/resources/airbyte_protocol/airbyte_protocol.yaml -title: AirbyteProtocol -type: object -description: AirbyteProtocol structs -properties: - airbyte_message: - "$ref": "#/definitions/AirbyteMessage" - configured_airbyte_catalog: - "$ref": "#/definitions/ConfiguredAirbyteCatalog" - -definitions: - AirbyteCatalog: - description: Airbyte stream schema catalog - type: object - additionalProperties: false - required: - - streams - properties: - streams: - type: array - items: - "$ref": "#/definitions/AirbyteStream" - AirbyteStream: - type: object - additionalProperties: false - required: - - name - - json_schema - # todo (cgardens) - make required once sources are migrated - # - supported_sync_modes - properties: - name: - type: string - description: Stream's name. - json_schema: - description: Stream schema using Json Schema specs. - type: object - existingJavaType: com.fasterxml.jackson.databind.JsonNode - supported_sync_modes: - type: array - items: - "$ref": "#/definitions/SyncMode" - source_defined_cursor: - description: If the source defines the cursor field, then it does any other cursor field inputs will be ignored. If it does not either the user_provided one is used or as a backup the default one is used. - type: boolean - default_cursor_field: - description: Path to the field that will be used to determine if a record is new or modified since the last sync. If not provided by the source, the end user will have to specify the comparable themselves. - type: array - items: - type: string - ConfiguredAirbyteCatalog: - description: Airbyte stream schema catalog - type: object - additionalProperties: false - required: - - streams - properties: - streams: - type: array - items: - "$ref": "#/definitions/ConfiguredAirbyteStream" - ConfiguredAirbyteStream: - type: object - additionalProperties: false - required: - - stream - properties: - stream: - "$ref": "#/definitions/AirbyteStream" - sync_mode: - "$ref": "#/definitions/SyncMode" - default: full_refresh - cursor_field: - description: Path to the field that will be used to determine if a record is new or modified since the last sync. This field is REQUIRED if `sync_mode` is `incremental`. Otherwise it is ignored. - type: array - items: - type: string - SyncMode: - type: string - enum: - - full_refresh - - incremental diff --git a/airbyte-migration/src/test/java/io/airbyte/migrate/migrations/MigrateV0_14_13Test.java b/airbyte-migration/src/test/java/io/airbyte/migrate/migrations/MigrateV0_14_3Test.java similarity index 99% rename from airbyte-migration/src/test/java/io/airbyte/migrate/migrations/MigrateV0_14_13Test.java rename to airbyte-migration/src/test/java/io/airbyte/migrate/migrations/MigrateV0_14_3Test.java index e3af5be1c0a3..e0ac669c895d 100644 --- a/airbyte-migration/src/test/java/io/airbyte/migrate/migrations/MigrateV0_14_13Test.java +++ b/airbyte-migration/src/test/java/io/airbyte/migrate/migrations/MigrateV0_14_3Test.java @@ -44,7 +44,7 @@ import java.util.stream.Stream; import org.junit.jupiter.api.Test; -class MigrateV0_14_13Test { +class MigrateV0_14_3Test { private static final ResourceId SYNC_RESOURCE_ID = ResourceId.fromConstantCase(ResourceType.CONFIG, "STANDARD_SYNC"); diff --git a/airbyte-migration/src/test/java/io/airbyte/migrate/migrations/MigrateV0_18_0Test.java b/airbyte-migration/src/test/java/io/airbyte/migrate/migrations/MigrateV0_18_0Test.java new file mode 100644 index 000000000000..7324921adaa1 --- /dev/null +++ b/airbyte-migration/src/test/java/io/airbyte/migrate/migrations/MigrateV0_18_0Test.java @@ -0,0 +1,96 @@ +/* + * MIT License + * + * Copyright (c) 2020 Airbyte + * + * Permission is hereby granted, free of charge, to any person obtaining a copy + * of this software and associated documentation files (the "Software"), to deal + * in the Software without restriction, including without limitation the rights + * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell + * copies of the Software, and to permit persons to whom the Software is + * furnished to do so, subject to the following conditions: + * + * The above copyright notice and this permission notice shall be included in all + * copies or substantial portions of the Software. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +package io.airbyte.migrate.migrations; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; + +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 io.airbyte.commons.functional.ListConsumer; +import io.airbyte.commons.json.Jsons; +import io.airbyte.commons.resources.MoreResources; +import io.airbyte.migrate.Migration; +import io.airbyte.migrate.MigrationTestUtils; +import io.airbyte.migrate.MigrationUtils; +import io.airbyte.migrate.Migrations; +import io.airbyte.migrate.ResourceId; +import io.airbyte.migrate.ResourceType; +import java.io.IOException; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.stream.Stream; +import org.junit.jupiter.api.Test; + +class MigrateV0_18_0Test { + + private static final ResourceId SYNC_RESOURCE_ID = ResourceId.fromConstantCase(ResourceType.CONFIG, "STANDARD_SYNC"); + + @Test + void testMigration() throws IOException { + final Migration migration = Migrations.MIGRATIONS + .stream() + .filter(m -> m instanceof MigrationV0_18_0) + .findAny() + .orElse(null); + assertNotNull(migration); + + // construct a sync object. in this migration we modify the catalog, so we will use this as + // a base. + final JsonNode syncWithoutCatalog = Jsons.jsonNode(ImmutableMap.builder() + .put("sourceId", UUID.randomUUID().toString()) + .put("destinationId", UUID.randomUUID().toString()) + .put("connectionId", UUID.randomUUID().toString()) + .put("name", "users_sync") + .put("status", "active") + .build()); + + // input Catalog + final JsonNode inputCatalog = Jsons.deserialize(MoreResources.readResource("migrations/migrationV0_18_0/example_input_catalog.json")); + final JsonNode syncInputCatalog = Jsons.clone(syncWithoutCatalog); + ((ObjectNode) syncInputCatalog).set("catalog", inputCatalog); + + // Output Catalog + final JsonNode outputCatalog = Jsons.deserialize(MoreResources.readResource("migrations/migrationV0_18_0/example_output_catalog.json")); + final JsonNode syncOutputCatalog = Jsons.clone(syncWithoutCatalog); + ((ObjectNode) syncOutputCatalog).set("catalog", outputCatalog); + + final Map> records = ImmutableMap.of(SYNC_RESOURCE_ID, Stream.of(syncInputCatalog)); + + final Map> outputConsumer = MigrationTestUtils.createOutputConsumer(migration.getOutputSchema().keySet()); + migration.migrate(records, MigrationUtils.mapRecordConsumerToConsumer(outputConsumer)); + + final Map> expectedOutputOverrides = ImmutableMap.of(SYNC_RESOURCE_ID, ImmutableList.of(syncOutputCatalog)); + final Map> expectedOutput = + MigrationTestUtils.createExpectedOutput(migration.getOutputSchema().keySet(), expectedOutputOverrides); + + final Map> outputAsList = MigrationTestUtils.collectConsumersToList(outputConsumer); + assertEquals(expectedOutput, outputAsList); + } + +} diff --git a/airbyte-migration/src/test/resources/migrations/migrationV0_18_0/example_input_catalog.json b/airbyte-migration/src/test/resources/migrations/migrationV0_18_0/example_input_catalog.json new file mode 100644 index 000000000000..136c49409bee --- /dev/null +++ b/airbyte-migration/src/test/resources/migrations/migrationV0_18_0/example_input_catalog.json @@ -0,0 +1,96 @@ +{ + "streams": [ + { + "stream": { + "name": "users", + "json_schema": { + "type": "object", + "properties": { + "last_name": { + "type": "string" + }, + "first_name": { + "type": "string" + } + } + }, + "supported_sync_modes": ["full_refresh", "incremental"], + "source_defined_cursor": false, + "default_cursor_field": ["last_name"], + "source_defined_primary_key": [["first_name"], ["last_name"]] + }, + "sync_mode": "incremental", + "cursor_field": ["first_name"] + }, + { + "stream": { + "name": "products", + "json_schema": { + "type": "object", + "properties": { + "name": { + "type": "string" + }, + "sku": { + "type": "string" + } + } + }, + "supported_sync_modes": ["full_refresh", "incremental"], + "source_defined_cursor": true, + "default_cursor_field": [] + }, + "sync_mode": "incremental", + "cursor_field": ["name"], + "destination_sync_mode": "append_dedup", + "primary_key": [["sku"]] + }, + { + "stream": { + "name": "product_lines", + "json_schema": { + "type": "object", + "properties": { + "name": { + "type": "string" + } + } + }, + "supported_sync_modes": ["full_refresh", "incremental"], + "source_defined_cursor": false, + "default_cursor_field": [] + }, + "sync_mode": "full_refresh" + }, + { + "stream": { + "name": "product_details", + "json_schema": { + "type": "object", + "properties": { + "number": { + "type": "number" + }, + "boolean": { + "type": "boolean" + }, + "string": { + "type": "string" + }, + "array": { + "type": "array" + }, + "object": { + "type": "object" + } + } + }, + "supported_sync_modes": ["full_refresh", "incremental"], + "source_defined_cursor": false, + "default_cursor_field": [] + }, + "sync_mode": "full_refresh", + "cursor_field": [] + } + ] +} diff --git a/airbyte-migration/src/test/resources/migrations/migrationV0_18_0/example_output_catalog.json b/airbyte-migration/src/test/resources/migrations/migrationV0_18_0/example_output_catalog.json new file mode 100644 index 000000000000..0f88c86b5ba4 --- /dev/null +++ b/airbyte-migration/src/test/resources/migrations/migrationV0_18_0/example_output_catalog.json @@ -0,0 +1,103 @@ +{ + "streams": [ + { + "stream": { + "name": "users", + "json_schema": { + "type": "object", + "properties": { + "last_name": { + "type": "string" + }, + "first_name": { + "type": "string" + } + } + }, + "supported_sync_modes": ["full_refresh", "incremental"], + "source_defined_cursor": false, + "default_cursor_field": ["last_name"], + "source_defined_primary_key": [["first_name"], ["last_name"]] + }, + "sync_mode": "incremental", + "cursor_field": ["first_name"], + "destination_sync_mode": "append", + "primary_key": [["first_name"], ["last_name"]] + }, + { + "stream": { + "name": "products", + "json_schema": { + "type": "object", + "properties": { + "name": { + "type": "string" + }, + "sku": { + "type": "string" + } + } + }, + "supported_sync_modes": ["full_refresh", "incremental"], + "source_defined_cursor": true, + "default_cursor_field": [] + }, + "sync_mode": "incremental", + "cursor_field": ["name"], + "destination_sync_mode": "append_dedup", + "primary_key": [["sku"]] + }, + { + "stream": { + "name": "product_lines", + "json_schema": { + "type": "object", + "properties": { + "name": { + "type": "string" + } + } + }, + "supported_sync_modes": ["full_refresh", "incremental"], + "source_defined_cursor": false, + "default_cursor_field": [] + }, + "sync_mode": "full_refresh", + "cursor_field": [], + "destination_sync_mode": "overwrite", + "primary_key": [] + }, + { + "stream": { + "name": "product_details", + "json_schema": { + "type": "object", + "properties": { + "number": { + "type": "number" + }, + "boolean": { + "type": "boolean" + }, + "string": { + "type": "string" + }, + "array": { + "type": "array" + }, + "object": { + "type": "object" + } + } + }, + "supported_sync_modes": ["full_refresh", "incremental"], + "source_defined_cursor": false, + "default_cursor_field": [] + }, + "sync_mode": "full_refresh", + "cursor_field": [], + "destination_sync_mode": "overwrite", + "primary_key": [] + } + ] +} diff --git a/airbyte-protocol/models/src/main/java/io/airbyte/protocol/models/CatalogHelpers.java b/airbyte-protocol/models/src/main/java/io/airbyte/protocol/models/CatalogHelpers.java index ee83d61ec346..70427ee8f917 100644 --- a/airbyte-protocol/models/src/main/java/io/airbyte/protocol/models/CatalogHelpers.java +++ b/airbyte-protocol/models/src/main/java/io/airbyte/protocol/models/CatalogHelpers.java @@ -64,21 +64,25 @@ public static ConfiguredAirbyteStream createConfiguredAirbyteStream(String strea public static ConfiguredAirbyteStream createConfiguredAirbyteStream(String streamName, List fields) { return new ConfiguredAirbyteStream().withStream(new AirbyteStream().withName(streamName).withJsonSchema(fieldsToJsonSchema(fields))) - .withSyncMode(SyncMode.FULL_REFRESH); + .withSyncMode(SyncMode.FULL_REFRESH).withDestinationSyncMode(DestinationSyncMode.OVERWRITE); } public static ConfiguredAirbyteStream createIncrementalConfiguredAirbyteStream( String streamName, SyncMode syncMode, + DestinationSyncMode destinationSyncMode, String cursorFieldName, + List primaryKeys, Field... fields) { - return createIncrementalConfiguredAirbyteStream(streamName, syncMode, cursorFieldName, Arrays.asList(fields)); + return createIncrementalConfiguredAirbyteStream(streamName, syncMode, destinationSyncMode, cursorFieldName, primaryKeys, Arrays.asList(fields)); } public static ConfiguredAirbyteStream createIncrementalConfiguredAirbyteStream( String streamName, SyncMode syncMode, + DestinationSyncMode destinationSyncMode, String cursorFieldName, + List primaryKeys, List fields) { return new ConfiguredAirbyteStream() .withStream(new AirbyteStream() @@ -86,7 +90,9 @@ public static ConfiguredAirbyteStream createIncrementalConfiguredAirbyteStream( .withSupportedSyncModes(Collections.singletonList(syncMode)) .withJsonSchema(fieldsToJsonSchema(fields))) .withSyncMode(syncMode) - .withCursorField(Collections.singletonList(cursorFieldName)); + .withCursorField(Collections.singletonList(cursorFieldName)) + .withDestinationSyncMode(destinationSyncMode) + .withPrimaryKey(primaryKeys.stream().map(Collections::singletonList).collect(Collectors.toList())); } /** @@ -109,7 +115,7 @@ public static ConfiguredAirbyteStream toDefaultConfiguredStream(AirbyteStream st .withStream(stream) .withSyncMode(SyncMode.FULL_REFRESH) .withCursorField(new ArrayList<>()) - .withDestinationSyncMode(DestinationSyncMode.APPEND) + .withDestinationSyncMode(DestinationSyncMode.OVERWRITE) .withPrimaryKey(new ArrayList<>()); } diff --git a/airbyte-protocol/models/src/main/resources/airbyte_protocol/airbyte_protocol.yaml b/airbyte-protocol/models/src/main/resources/airbyte_protocol/airbyte_protocol.yaml index 55ee791b4357..10e7d3fea358 100644 --- a/airbyte-protocol/models/src/main/resources/airbyte_protocol/airbyte_protocol.yaml +++ b/airbyte-protocol/models/src/main/resources/airbyte_protocol/airbyte_protocol.yaml @@ -167,6 +167,8 @@ definitions: additionalProperties: true required: - stream + - sync_mode + - destination_sync_mode properties: stream: "$ref": "#/definitions/AirbyteStream" @@ -199,7 +201,7 @@ definitions: - append - overwrite #- upsert_dedup # TODO chris: SCD Type 1 can be implemented later - - append_dedup # SCD Type 2 + - append_dedup # SCD Type 1 & 2 ConnectorSpecification: description: Specification of a connector (source/destination) type: object diff --git a/airbyte-scheduler/persistence/src/main/java/io/airbyte/scheduler/persistence/DefaultJobCreator.java b/airbyte-scheduler/persistence/src/main/java/io/airbyte/scheduler/persistence/DefaultJobCreator.java index 981e630c9f1b..1b2424ae211a 100644 --- a/airbyte-scheduler/persistence/src/main/java/io/airbyte/scheduler/persistence/DefaultJobCreator.java +++ b/airbyte-scheduler/persistence/src/main/java/io/airbyte/scheduler/persistence/DefaultJobCreator.java @@ -35,6 +35,7 @@ import io.airbyte.config.SourceConnection; import io.airbyte.config.StandardSync; import io.airbyte.protocol.models.ConfiguredAirbyteCatalog; +import io.airbyte.protocol.models.ConfiguredAirbyteStream.DestinationSyncMode; import io.airbyte.protocol.models.SyncMode; import java.io.IOException; import java.util.Optional; @@ -124,7 +125,7 @@ public Optional createSyncJob(SourceConnection source, } // Strategy: - // 1. Set all streams to full refresh. + // 1. Set all streams to full refresh - overwrite. // 2. Create a job where the source emits no records. // 3. Run a sync from the empty source to the destination. This will overwrite all data for each // stream in the destination. @@ -134,7 +135,10 @@ public Optional createSyncJob(SourceConnection source, public Optional createResetConnectionJob(DestinationConnection destination, StandardSync standardSync, String destinationDockerImage) throws IOException { final ConfiguredAirbyteCatalog configuredAirbyteCatalog = standardSync.getCatalog(); - configuredAirbyteCatalog.getStreams().forEach(configuredAirbyteStream -> configuredAirbyteStream.setSyncMode(SyncMode.FULL_REFRESH)); + configuredAirbyteCatalog.getStreams().forEach(configuredAirbyteStream -> { + configuredAirbyteStream.setSyncMode(SyncMode.FULL_REFRESH); + configuredAirbyteStream.setDestinationSyncMode(DestinationSyncMode.OVERWRITE); + }); final JobResetConnectionConfig resetConnectionConfig = new JobResetConnectionConfig() .withPrefix(standardSync.getPrefix()) diff --git a/airbyte-scheduler/persistence/src/test/java/io/airbyte/scheduler/persistence/DefaultJobCreatorTest.java b/airbyte-scheduler/persistence/src/test/java/io/airbyte/scheduler/persistence/DefaultJobCreatorTest.java index c82eff7970ba..3589365ff4c0 100644 --- a/airbyte-scheduler/persistence/src/test/java/io/airbyte/scheduler/persistence/DefaultJobCreatorTest.java +++ b/airbyte-scheduler/persistence/src/test/java/io/airbyte/scheduler/persistence/DefaultJobCreatorTest.java @@ -45,6 +45,7 @@ import io.airbyte.protocol.models.CatalogHelpers; import io.airbyte.protocol.models.ConfiguredAirbyteCatalog; import io.airbyte.protocol.models.ConfiguredAirbyteStream; +import io.airbyte.protocol.models.ConfiguredAirbyteStream.DestinationSyncMode; import io.airbyte.protocol.models.Field; import io.airbyte.protocol.models.Field.JsonSchemaPrimitive; import java.io.IOException; @@ -238,7 +239,10 @@ void testCreateSyncJobEnsureNoQueuing() throws IOException { void testCreateResetConnectionJob() throws IOException { final ConfiguredAirbyteCatalog expectedCatalog = STANDARD_SYNC.getCatalog(); expectedCatalog.getStreams() - .forEach(configuredAirbyteStream -> configuredAirbyteStream.setSyncMode(io.airbyte.protocol.models.SyncMode.FULL_REFRESH)); + .forEach(configuredAirbyteStream -> { + configuredAirbyteStream.setSyncMode(io.airbyte.protocol.models.SyncMode.FULL_REFRESH); + configuredAirbyteStream.setDestinationSyncMode(DestinationSyncMode.OVERWRITE); + }); final JobResetConnectionConfig JobResetConnectionConfig = new JobResetConnectionConfig() .withPrefix(STANDARD_SYNC.getPrefix()) @@ -264,7 +268,10 @@ void testCreateResetConnectionJob() throws IOException { void testCreateResetConnectionJobEnsureNoQueuing() throws IOException { final ConfiguredAirbyteCatalog expectedCatalog = STANDARD_SYNC.getCatalog(); expectedCatalog.getStreams() - .forEach(configuredAirbyteStream -> configuredAirbyteStream.setSyncMode(io.airbyte.protocol.models.SyncMode.FULL_REFRESH)); + .forEach(configuredAirbyteStream -> { + configuredAirbyteStream.setSyncMode(io.airbyte.protocol.models.SyncMode.FULL_REFRESH); + configuredAirbyteStream.setDestinationSyncMode(DestinationSyncMode.OVERWRITE); + }); final JobResetConnectionConfig JobResetConnectionConfig = new JobResetConnectionConfig() .withPrefix(STANDARD_SYNC.getPrefix()) diff --git a/airbyte-server/src/main/java/io/airbyte/server/converters/CatalogConverter.java b/airbyte-server/src/main/java/io/airbyte/server/converters/CatalogConverter.java index 22cb5b189c86..a3db566ad483 100644 --- a/airbyte-server/src/main/java/io/airbyte/server/converters/CatalogConverter.java +++ b/airbyte-server/src/main/java/io/airbyte/server/converters/CatalogConverter.java @@ -69,8 +69,8 @@ private static io.airbyte.api.model.AirbyteStreamConfiguration generateDefaultCo io.airbyte.api.model.AirbyteStreamConfiguration result = new io.airbyte.api.model.AirbyteStreamConfiguration() .aliasName(Names.toAlphanumericAndUnderscore(stream.getName())) .cursorField(stream.getDefaultCursorField()) - .primaryKey(stream.getSourceDefinedPrimaryKey()) .destinationSyncMode(io.airbyte.api.model.DestinationSyncMode.APPEND) + .primaryKey(stream.getSourceDefinedPrimaryKey()) .selected(true); if (stream.getSupportedSyncModes().size() > 0) result.setSyncMode(stream.getSupportedSyncModes().get(0)); @@ -86,8 +86,8 @@ public static io.airbyte.api.model.AirbyteCatalog toApi(final io.airbyte.protoco final io.airbyte.api.model.AirbyteStreamConfiguration configuration = new io.airbyte.api.model.AirbyteStreamConfiguration() .syncMode(Enums.convertTo(configuredStream.getSyncMode(), io.airbyte.api.model.SyncMode.class)) .cursorField(configuredStream.getCursorField()) - .primaryKey(configuredStream.getPrimaryKey()) .destinationSyncMode(Enums.convertTo(configuredStream.getDestinationSyncMode(), io.airbyte.api.model.DestinationSyncMode.class)) + .primaryKey(configuredStream.getPrimaryKey()) .aliasName(Names.toAlphanumericAndUnderscore(configuredStream.getStream().getName())) .selected(true); return new io.airbyte.api.model.AirbyteStreamAndConfiguration() @@ -106,9 +106,9 @@ public static io.airbyte.protocol.models.ConfiguredAirbyteCatalog toProtocol(fin .withStream(toProtocol(s.getStream())) .withSyncMode(Enums.convertTo(s.getConfig().getSyncMode(), io.airbyte.protocol.models.SyncMode.class)) .withCursorField(s.getConfig().getCursorField()) - .withPrimaryKey(s.getConfig().getPrimaryKey()) .withDestinationSyncMode(Enums.convertTo(s.getConfig().getDestinationSyncMode(), - io.airbyte.protocol.models.ConfiguredAirbyteStream.DestinationSyncMode.class))) + io.airbyte.protocol.models.ConfiguredAirbyteStream.DestinationSyncMode.class)) + .withPrimaryKey(s.getConfig().getPrimaryKey())) .collect(Collectors.toList()); return new io.airbyte.protocol.models.ConfiguredAirbyteCatalog() .withStreams(streams); diff --git a/airbyte-server/src/main/java/io/airbyte/server/handlers/WebBackendConnectionsHandler.java b/airbyte-server/src/main/java/io/airbyte/server/handlers/WebBackendConnectionsHandler.java index 35d61d5999f9..37e8ca0e7399 100644 --- a/airbyte-server/src/main/java/io/airbyte/server/handlers/WebBackendConnectionsHandler.java +++ b/airbyte-server/src/main/java/io/airbyte/server/handlers/WebBackendConnectionsHandler.java @@ -192,10 +192,18 @@ protected static AirbyteCatalog updateSchemaWithDiscovery(AirbyteCatalog origina else outputStreamConfig.setSyncMode(discoveredStreamConfig.getSyncMode()); - if (originalStreamConfig.getCursorField().size() > 0) + if (originalStreamConfig.getCursorField().size() > 0) { outputStreamConfig.setCursorField(originalStreamConfig.getCursorField()); - else + } else { outputStreamConfig.setCursorField(discoveredStreamConfig.getCursorField()); + } + + outputStreamConfig.setDestinationSyncMode(originalStreamConfig.getDestinationSyncMode()); + if (originalStreamConfig.getPrimaryKey().size() > 0) { + outputStreamConfig.setPrimaryKey(originalStreamConfig.getPrimaryKey()); + } else { + outputStreamConfig.setPrimaryKey(discoveredStreamConfig.getPrimaryKey()); + } outputStreamConfig.setAliasName(originalStreamConfig.getAliasName()); outputStreamConfig.setSelected(originalStreamConfig.getSelected()); diff --git a/airbyte-server/src/test/java/io/airbyte/server/handlers/WebBackendConnectionsHandlerTest.java b/airbyte-server/src/test/java/io/airbyte/server/handlers/WebBackendConnectionsHandlerTest.java index e79f2a75e02b..a74ec83ef6c6 100644 --- a/airbyte-server/src/test/java/io/airbyte/server/handlers/WebBackendConnectionsHandlerTest.java +++ b/airbyte-server/src/test/java/io/airbyte/server/handlers/WebBackendConnectionsHandlerTest.java @@ -45,6 +45,7 @@ import io.airbyte.api.model.ConnectionUpdate; import io.airbyte.api.model.DestinationIdRequestBody; import io.airbyte.api.model.DestinationRead; +import io.airbyte.api.model.DestinationSyncMode; import io.airbyte.api.model.JobConfigType; import io.airbyte.api.model.JobInfoRead; import io.airbyte.api.model.JobListRequestBody; @@ -353,6 +354,8 @@ public void testUpdateSchemaWithDiscoveryFromEmpty() { discovered.getStreams().get(0).getConfig() .syncMode(SyncMode.FULL_REFRESH) .cursorField(Collections.emptyList()) + .destinationSyncMode(DestinationSyncMode.OVERWRITE) + .primaryKey(Collections.emptyList()) .aliasName("stream1"); final AirbyteCatalog expected = ConnectionHelpers.generateBasicApiCatalog(); @@ -363,6 +366,8 @@ public void testUpdateSchemaWithDiscoveryFromEmpty() { expected.getStreams().get(0).getConfig() .syncMode(SyncMode.FULL_REFRESH) .cursorField(Collections.emptyList()) + .destinationSyncMode(DestinationSyncMode.OVERWRITE) + .primaryKey(Collections.emptyList()) .aliasName("stream1"); final AirbyteCatalog actual = WebBackendConnectionsHandler.updateSchemaWithDiscovery(original, discovered); @@ -384,6 +389,8 @@ public void testUpdateSchemaWithDiscoveryResetStream() { original.getStreams().get(0).getConfig() .syncMode(SyncMode.INCREMENTAL) .cursorField(List.of("field1")) + .destinationSyncMode(DestinationSyncMode.APPEND) + .primaryKey(Collections.emptyList()) .aliasName("random_stream"); final AirbyteCatalog discovered = ConnectionHelpers.generateBasicApiCatalog(); @@ -395,6 +402,8 @@ public void testUpdateSchemaWithDiscoveryResetStream() { discovered.getStreams().get(0).getConfig() .syncMode(SyncMode.FULL_REFRESH) .cursorField(Collections.emptyList()) + .destinationSyncMode(DestinationSyncMode.OVERWRITE) + .primaryKey(Collections.emptyList()) .aliasName("stream1"); final AirbyteCatalog expected = ConnectionHelpers.generateBasicApiCatalog(); @@ -406,6 +415,8 @@ public void testUpdateSchemaWithDiscoveryResetStream() { expected.getStreams().get(0).getConfig() .syncMode(SyncMode.FULL_REFRESH) .cursorField(Collections.emptyList()) + .destinationSyncMode(DestinationSyncMode.OVERWRITE) + .primaryKey(Collections.emptyList()) .aliasName("stream1"); final AirbyteCatalog actual = WebBackendConnectionsHandler.updateSchemaWithDiscovery(original, discovered); @@ -427,6 +438,8 @@ public void testUpdateSchemaWithDiscoveryMergeNewStream() { original.getStreams().get(0).getConfig() .syncMode(SyncMode.INCREMENTAL) .cursorField(List.of("field1")) + .destinationSyncMode(DestinationSyncMode.APPEND) + .primaryKey(Collections.emptyList()) .aliasName("renamed_stream"); final AirbyteCatalog discovered = ConnectionHelpers.generateBasicApiCatalog(); @@ -438,6 +451,8 @@ public void testUpdateSchemaWithDiscoveryMergeNewStream() { discovered.getStreams().get(0).getConfig() .syncMode(SyncMode.FULL_REFRESH) .cursorField(Collections.emptyList()) + .destinationSyncMode(DestinationSyncMode.OVERWRITE) + .primaryKey(Collections.emptyList()) .aliasName("stream1"); final AirbyteStreamAndConfiguration newStream = ConnectionHelpers.generateBasicApiCatalog().getStreams().get(0); newStream.getStream() @@ -448,6 +463,8 @@ public void testUpdateSchemaWithDiscoveryMergeNewStream() { newStream.getConfig() .syncMode(SyncMode.FULL_REFRESH) .cursorField(Collections.emptyList()) + .destinationSyncMode(DestinationSyncMode.OVERWRITE) + .primaryKey(Collections.emptyList()) .aliasName("stream2"); discovered.getStreams().add(newStream); @@ -460,6 +477,8 @@ public void testUpdateSchemaWithDiscoveryMergeNewStream() { expected.getStreams().get(0).getConfig() .syncMode(SyncMode.INCREMENTAL) .cursorField(List.of("field1")) + .destinationSyncMode(DestinationSyncMode.APPEND) + .primaryKey(Collections.emptyList()) .aliasName("renamed_stream"); final AirbyteStreamAndConfiguration expectedNewStream = ConnectionHelpers.generateBasicApiCatalog().getStreams().get(0); expectedNewStream.getStream() @@ -470,6 +489,8 @@ public void testUpdateSchemaWithDiscoveryMergeNewStream() { expectedNewStream.getConfig() .syncMode(SyncMode.FULL_REFRESH) .cursorField(Collections.emptyList()) + .destinationSyncMode(DestinationSyncMode.OVERWRITE) + .primaryKey(Collections.emptyList()) .aliasName("stream2"); expected.getStreams().add(expectedNewStream); diff --git a/airbyte-server/src/test/java/io/airbyte/server/helpers/ConnectionHelpers.java b/airbyte-server/src/test/java/io/airbyte/server/helpers/ConnectionHelpers.java index 6b8b9431196f..0bc800e5f6de 100644 --- a/airbyte-server/src/test/java/io/airbyte/server/helpers/ConnectionHelpers.java +++ b/airbyte-server/src/test/java/io/airbyte/server/helpers/ConnectionHelpers.java @@ -41,6 +41,7 @@ import io.airbyte.protocol.models.CatalogHelpers; import io.airbyte.protocol.models.ConfiguredAirbyteCatalog; import io.airbyte.protocol.models.ConfiguredAirbyteStream; +import io.airbyte.protocol.models.ConfiguredAirbyteStream.DestinationSyncMode; import io.airbyte.protocol.models.Field; import io.airbyte.protocol.models.Field.JsonSchemaPrimitive; import java.util.Collections; @@ -125,7 +126,8 @@ public static ConfiguredAirbyteCatalog generateBasicConfiguredAirbyteCatalog() { final ConfiguredAirbyteStream stream = new ConfiguredAirbyteStream() .withStream(generateBasicAirbyteStream()) .withCursorField(Lists.newArrayList(FIELD_NAME)) - .withSyncMode(io.airbyte.protocol.models.SyncMode.INCREMENTAL); + .withSyncMode(io.airbyte.protocol.models.SyncMode.INCREMENTAL) + .withDestinationSyncMode(DestinationSyncMode.APPEND); return new ConfiguredAirbyteCatalog().withStreams(Collections.singletonList(stream)); } @@ -146,6 +148,8 @@ private static AirbyteStreamConfiguration generateBasicApiStreamConfig() { return new AirbyteStreamConfiguration() .syncMode(SyncMode.INCREMENTAL) .cursorField(Lists.newArrayList(FIELD_NAME)) + .destinationSyncMode(io.airbyte.api.model.DestinationSyncMode.APPEND) + .primaryKey(Collections.emptyList()) .aliasName(Names.toAlphanumericAndUnderscore(STREAM_NAME)) .selected(true); } diff --git a/airbyte-tests/src/acceptanceTests/java/io/airbyte/test/acceptance/AcceptanceTests.java b/airbyte-tests/src/acceptanceTests/java/io/airbyte/test/acceptance/AcceptanceTests.java index d9ec050efc59..484a4706ce98 100644 --- a/airbyte-tests/src/acceptanceTests/java/io/airbyte/test/acceptance/AcceptanceTests.java +++ b/airbyte-tests/src/acceptanceTests/java/io/airbyte/test/acceptance/AcceptanceTests.java @@ -311,7 +311,8 @@ public void testCreateConnection() throws ApiException { final String name = "test-connection-" + UUID.randomUUID().toString(); final ConnectionSchedule schedule = new ConnectionSchedule().timeUnit(MINUTES).units(100L); final SyncMode syncMode = SyncMode.FULL_REFRESH; - catalog.getStreams().forEach(s -> s.getConfig().syncMode(syncMode)); + final DestinationSyncMode destinationSyncMode = DestinationSyncMode.OVERWRITE; + catalog.getStreams().forEach(s -> s.getConfig().syncMode(syncMode).destinationSyncMode(destinationSyncMode)); final ConnectionRead createdConnection = createConnection(name, sourceId, destinationId, catalog, schedule, syncMode); assertEquals(sourceId, createdConnection.getSourceId()); @@ -329,13 +330,13 @@ public void testManualSync() throws Exception { final UUID destinationId = createDestination().getDestinationId(); final AirbyteCatalog catalog = discoverSourceSchema(sourceId); final SyncMode syncMode = SyncMode.FULL_REFRESH; - catalog.getStreams().forEach(s -> s.getConfig().syncMode(syncMode)); - + final DestinationSyncMode destinationSyncMode = DestinationSyncMode.OVERWRITE; + catalog.getStreams().forEach(s -> s.getConfig().syncMode(syncMode).destinationSyncMode(destinationSyncMode)); final UUID connectionId = createConnection(connectionName, sourceId, destinationId, catalog, null, syncMode).getConnectionId(); final JobInfoRead connectionSyncRead = apiClient.getConnectionApi().syncConnection(new ConnectionIdRequestBody().connectionId(connectionId)); waitForSuccessfulJob(apiClient.getJobsApi(), connectionSyncRead.getJob()); - assertSourceAndTargetDbInSync(sourcePsql); + assertSourceAndTargetDbInSync(sourcePsql, false); } @Test @@ -352,17 +353,21 @@ public void testIncrementalSync() throws Exception { // instead of assertFalse to avoid NPE from unboxed. assertNull(stream.getSourceDefinedCursor()); assertTrue(stream.getDefaultCursorField().isEmpty()); + assertTrue(stream.getSourceDefinedPrimaryKey().isEmpty()); final SyncMode syncMode = SyncMode.INCREMENTAL; - catalog.getStreams().forEach(s -> s.getConfig().syncMode(syncMode).cursorField(List.of(COLUMN_ID))); - + final DestinationSyncMode destinationSyncMode = DestinationSyncMode.APPEND; + catalog.getStreams().forEach(s -> s.getConfig() + .syncMode(syncMode) + .cursorField(List.of(COLUMN_ID)) + .destinationSyncMode(destinationSyncMode)); final UUID connectionId = createConnection(connectionName, sourceId, destinationId, catalog, null, syncMode).getConnectionId(); final JobInfoRead connectionSyncRead1 = apiClient.getConnectionApi() .syncConnection(new ConnectionIdRequestBody().connectionId(connectionId)); waitForSuccessfulJob(apiClient.getJobsApi(), connectionSyncRead1.getJob()); - assertSourceAndTargetDbInSync(sourcePsql); + assertSourceAndTargetDbInSync(sourcePsql, false); // add new records and run again. final Database source = getDatabase(sourcePsql); @@ -380,18 +385,18 @@ public void testIncrementalSync() throws Exception { final JobInfoRead connectionSyncRead2 = apiClient.getConnectionApi() .syncConnection(new ConnectionIdRequestBody().connectionId(connectionId)); waitForSuccessfulJob(apiClient.getJobsApi(), connectionSyncRead2.getJob()); - assertDestinationContains(expectedRecords, STREAM_NAME); + assertRawDestinationContains(expectedRecords, STREAM_NAME); // reset back to no data. final JobInfoRead jobInfoRead = apiClient.getConnectionApi().resetConnection(new ConnectionIdRequestBody().connectionId(connectionId)); waitForSuccessfulJob(apiClient.getJobsApi(), jobInfoRead.getJob()); - assertDestinationContains(Collections.emptyList(), STREAM_NAME); + assertRawDestinationContains(Collections.emptyList(), STREAM_NAME); // sync one more time. verify it is the equivalent of a full refresh. final JobInfoRead connectionSyncRead3 = apiClient.getConnectionApi() .syncConnection(new ConnectionIdRequestBody().connectionId(connectionId)); waitForSuccessfulJob(apiClient.getJobsApi(), connectionSyncRead3.getJob()); - assertSourceAndTargetDbInSync(sourcePsql); + assertSourceAndTargetDbInSync(sourcePsql, false); } @Test @@ -407,18 +412,62 @@ public void testScheduledSync() throws Exception { final ConnectionSchedule connectionSchedule = new ConnectionSchedule().units(1L).timeUnit(MINUTES); final SyncMode syncMode = SyncMode.FULL_REFRESH; - catalog.getStreams().forEach(s -> s.getConfig().syncMode(syncMode)); + final DestinationSyncMode destinationSyncMode = DestinationSyncMode.OVERWRITE; + catalog.getStreams().forEach(s -> s.getConfig().syncMode(syncMode).destinationSyncMode(destinationSyncMode)); createConnection(connectionName, sourceId, destinationId, catalog, connectionSchedule, syncMode); // When a new connection is created, Airbyte might sync it immediately (before the sync interval). // Then it will wait the sync interval. Thread.sleep(Duration.of(30, SECONDS).toMillis()); - assertSourceAndTargetDbInSync(sourcePsql); + assertSourceAndTargetDbInSync(sourcePsql, false); } @Test @Order(10) + public void testIncrementalDedupSync() throws Exception { + final String connectionName = "test-connection"; + final UUID sourceId = createPostgresSource().getSourceId(); + final UUID destinationId = createDestination().getDestinationId(); + final AirbyteCatalog catalog = discoverSourceSchema(sourceId); + final SyncMode syncMode = SyncMode.INCREMENTAL; + final DestinationSyncMode destinationSyncMode = DestinationSyncMode.APPEND_DEDUP; + catalog.getStreams().forEach(s -> s.getConfig() + .syncMode(syncMode) + .cursorField(List.of(COLUMN_ID)) + .destinationSyncMode(destinationSyncMode) + .primaryKey(List.of(List.of(COLUMN_NAME)))); + final UUID connectionId = createConnection(connectionName, sourceId, destinationId, catalog, null, syncMode).getConnectionId(); + + // sync from start + final JobInfoRead connectionSyncRead1 = apiClient.getConnectionApi() + .syncConnection(new ConnectionIdRequestBody().connectionId(connectionId)); + waitForSuccessfulJob(apiClient.getJobsApi(), connectionSyncRead1.getJob()); + + assertSourceAndTargetDbInSync(sourcePsql, true); + + // add new records and run again. + final Database source = getDatabase(sourcePsql); + final List expectedRawRecords = retrieveSourceRecords(source, STREAM_NAME); + expectedRawRecords.add(Jsons.jsonNode(ImmutableMap.builder().put(COLUMN_ID, 6).put(COLUMN_NAME, "sherif").build())); + expectedRawRecords.add(Jsons.jsonNode(ImmutableMap.builder().put(COLUMN_ID, 7).put(COLUMN_NAME, "chris").build())); + source.query(ctx -> ctx.execute("UPDATE id_and_name SET id=6 WHERE name='sherif'")); + source.query(ctx -> ctx.execute("INSERT INTO id_and_name(id, name) VALUES(7, 'chris')")); + // retrieve latest snapshot of source records after modifications; the deduplicated table in + // destination should mirror this latest state of records + final List expectedNormalizedRecords = retrieveSourceRecords(source, STREAM_NAME); + source.close(); + + final JobInfoRead connectionSyncRead2 = apiClient.getConnectionApi() + .syncConnection(new ConnectionIdRequestBody().connectionId(connectionId)); + waitForSuccessfulJob(apiClient.getJobsApi(), connectionSyncRead2.getJob()); + + assertRawDestinationContains(expectedRawRecords, STREAM_NAME); + assertNormalizedDestinationContains(expectedNormalizedRecords); + } + + @Test + @Order(11) public void testRedactionOfSensitiveRequestBodies() throws Exception { // check that the source password is not present in the logs final List serverLogLines = Files.readLines( @@ -443,12 +492,23 @@ private AirbyteCatalog discoverSourceSchema(UUID sourceId) throws ApiException { return apiClient.getSourceApi().discoverSchemaForSource(new SourceIdRequestBody().sourceId(sourceId)).getCatalog(); } - private void assertSourceAndTargetDbInSync(PostgreSQLContainer sourceDb) throws Exception { + private void assertSourceAndTargetDbInSync(PostgreSQLContainer sourceDb, boolean withScdTable) throws Exception { final Database source = getDatabase(sourceDb); final Set sourceStreams = listStreams(source); - final Set sourceStreamsWithRawPrefix = - sourceStreams.stream().map(x -> String.format("_airbyte_raw_%s%s", OUTPUT_NAMESPACE, x.replace(".", "_"))).collect(Collectors.toSet()); + final Set sourceStreamsWithRawPrefix = sourceStreams.stream().flatMap(x -> { + final String cleanedNameStream = x.replace(".", "_"); + if (withScdTable) { + return List.of( + String.format("_airbyte_raw_%s%s", OUTPUT_NAMESPACE, cleanedNameStream), + String.format("%s%s_scd", OUTPUT_NAMESPACE, cleanedNameStream), + String.format("%s%s", OUTPUT_NAMESPACE, cleanedNameStream)).stream(); + } else { + return List.of( + String.format("_airbyte_raw_%s%s", OUTPUT_NAMESPACE, cleanedNameStream), + String.format("%s%s", OUTPUT_NAMESPACE, cleanedNameStream)).stream(); + } + }).collect(Collectors.toSet()); final Database destination = getDatabase(destinationPsql); final Set destinationStreams = listDestinationStreams(destination); assertEquals(sourceStreamsWithRawPrefix, destinationStreams, @@ -489,7 +549,7 @@ private Set listDestinationStreams(Database destination) throws SQLExcep .collect(Collectors.toSet()); } - private void assertDestinationContains(List sourceRecords, String streamName) throws Exception { + private void assertRawDestinationContains(List sourceRecords, String streamName) throws Exception { final Set destinationRecords = new HashSet<>(retrieveDestinationRecords(streamName)); assertEquals(sourceRecords.size(), destinationRecords.size(), @@ -501,9 +561,26 @@ private void assertDestinationContains(List sourceRecords, String stre } } + private void assertNormalizedDestinationContains(final List sourceRecords) throws Exception { + final Database destination = getDatabase(destinationPsql); + final String finalDestinationTable = String.format("%s%s", OUTPUT_NAMESPACE, STREAM_NAME.replace(".", "_")); + final List destinationRecords = retrieveSourceRecords(destination, finalDestinationTable); + + assertEquals(sourceRecords.size(), destinationRecords.size(), + String.format("destination contains: %s record. source contains: %s", sourceRecords.size(), destinationRecords.size())); + + for (JsonNode sourceStreamRecord : sourceRecords) { + assertTrue( + destinationRecords.stream() + .anyMatch(r -> r.get(COLUMN_NAME).asText().equals(sourceStreamRecord.get(COLUMN_NAME).asText()) + && r.get(COLUMN_ID).asInt() == sourceStreamRecord.get(COLUMN_ID).asInt()), + String.format("destination does not contain record:\n %s \n destination contains:\n %s\n", sourceStreamRecord, destinationRecords)); + } + } + private void assertStreamsEquivalent(Database source, String table) throws Exception { final List sourceRecords = retrieveSourceRecords(source, table); - assertDestinationContains(sourceRecords, table); + assertRawDestinationContains(sourceRecords, table); } private ConnectionRead createConnection(String name, @@ -589,18 +666,18 @@ private JsonNode getSourceDbConfig() { } private JsonNode getDestinationDbConfig() { - return getDbConfig(destinationPsql, false, true); + return getDbConfig(destinationPsql, false, true, true); } private JsonNode getDestinationDbConfigWithHiddenPassword() { - return getDbConfig(destinationPsql, true, true); + return getDbConfig(destinationPsql, true, true, true); } private JsonNode getDbConfig(PostgreSQLContainer psql) { - return getDbConfig(psql, false, false); + return getDbConfig(psql, false, false, false); } - private JsonNode getDbConfig(PostgreSQLContainer psql, boolean hiddenPassword, boolean withSchema) { + private JsonNode getDbConfig(PostgreSQLContainer psql, boolean hiddenPassword, boolean withSchema, boolean withNormalization) { try { final Map dbConfig = new HashMap<>(); @@ -621,6 +698,10 @@ private JsonNode getDbConfig(PostgreSQLContainer psql, boolean hiddenPassword, b dbConfig.put("schema", "public"); } + if (withNormalization) { + dbConfig.put("basic_normalization", true); + } + return Jsons.jsonNode(dbConfig); } catch (UnknownHostException e) { throw new RuntimeException(e); @@ -685,7 +766,7 @@ private void deleteDestination(UUID destinationId) throws ApiException { private static void waitForSuccessfulJob(JobsApi jobsApi, JobRead originalJob) throws InterruptedException, ApiException { JobRead job = originalJob; int count = 0; - while (count < 15 && (job.getStatus() == JobStatus.PENDING || job.getStatus() == JobStatus.RUNNING)) { + while (count < 60 && (job.getStatus() == JobStatus.PENDING || job.getStatus() == JobStatus.RUNNING)) { Thread.sleep(1000); count++; diff --git a/airbyte-workers/src/main/java/io/airbyte/workers/DefaultSyncWorker.java b/airbyte-workers/src/main/java/io/airbyte/workers/DefaultSyncWorker.java index bbc48c7d2631..bb2d3d1487ff 100644 --- a/airbyte-workers/src/main/java/io/airbyte/workers/DefaultSyncWorker.java +++ b/airbyte-workers/src/main/java/io/airbyte/workers/DefaultSyncWorker.java @@ -32,7 +32,6 @@ import io.airbyte.config.StandardTargetConfig; import io.airbyte.config.State; import io.airbyte.protocol.models.AirbyteMessage; -import io.airbyte.protocol.models.SyncMode; import io.airbyte.workers.normalization.NormalizationRunner; import io.airbyte.workers.protocols.Destination; import io.airbyte.workers.protocols.Mapper; @@ -85,8 +84,7 @@ public StandardSyncOutput run(StandardSyncInput syncInput, Path jobRoot) throws LOGGER.info("configured sync modes: {}", syncInput.getCatalog().getStreams() .stream() - .collect(Collectors.toMap(s -> s.getStream().getName(), s -> s.getSyncMode() != null ? s.getSyncMode() : SyncMode.FULL_REFRESH))); - + .collect(Collectors.toMap(s -> s.getStream().getName(), s -> String.format("%s - %s", s.getSyncMode(), s.getDestinationSyncMode())))); final StandardTapConfig tapConfig = WorkerUtils.syncToTapConfig(syncInput); final StandardTargetConfig targetConfig = WorkerUtils.syncToTargetConfig(syncInput); targetConfig.setCatalog(mapper.mapCatalog(targetConfig.getCatalog())); diff --git a/airbyte-workers/src/test/java/io/airbyte/workers/protocols/airbyte/DefaultAirbyteSourceTest.java b/airbyte-workers/src/test/java/io/airbyte/workers/protocols/airbyte/DefaultAirbyteSourceTest.java index 0b3d0b3848ef..50005998d17d 100644 --- a/airbyte-workers/src/test/java/io/airbyte/workers/protocols/airbyte/DefaultAirbyteSourceTest.java +++ b/airbyte-workers/src/test/java/io/airbyte/workers/protocols/airbyte/DefaultAirbyteSourceTest.java @@ -41,13 +41,10 @@ import io.airbyte.config.StandardTapConfig; import io.airbyte.config.State; import io.airbyte.protocol.models.AirbyteMessage; -import io.airbyte.protocol.models.AirbyteStream; import io.airbyte.protocol.models.CatalogHelpers; import io.airbyte.protocol.models.ConfiguredAirbyteCatalog; -import io.airbyte.protocol.models.ConfiguredAirbyteStream; import io.airbyte.protocol.models.Field; import io.airbyte.protocol.models.Field.JsonSchemaPrimitive; -import io.airbyte.protocol.models.SyncMode; import io.airbyte.workers.WorkerConstants; import io.airbyte.workers.WorkerException; import io.airbyte.workers.process.IntegrationLauncher; @@ -58,7 +55,6 @@ import java.nio.file.Files; import java.nio.file.Path; import java.time.Duration; -import java.util.Collections; import java.util.List; import java.util.Map; import org.junit.jupiter.api.Assertions; @@ -71,13 +67,9 @@ class DefaultAirbyteSourceTest { private static final String STREAM_NAME = "user_preferences"; private static final String FIELD_NAME = "favorite_color"; - private static final ConfiguredAirbyteCatalog CATALOG = new ConfiguredAirbyteCatalog() - .withStreams(Collections.singletonList( - new ConfiguredAirbyteStream() - .withSyncMode(SyncMode.FULL_REFRESH) - .withStream(new AirbyteStream() - .withName("hudi:latest") - .withJsonSchema(CatalogHelpers.fieldsToJsonSchema(new Field(FIELD_NAME, Field.JsonSchemaPrimitive.STRING)))))); + private static final ConfiguredAirbyteCatalog CATALOG = CatalogHelpers.createConfiguredAirbyteCatalog( + "hudi:latest", + Field.of(FIELD_NAME, JsonSchemaPrimitive.STRING)); private static final StandardTapConfig TAP_CONFIG = new StandardTapConfig() .withState(new State().withState(Jsons.jsonNode(ImmutableMap.of("checkpoint", "the future.")))) diff --git a/docs/api/generated-api-html/index.html b/docs/api/generated-api-html/index.html index 8543215ecd8c..91b75bb95a27 100644 --- a/docs/api/generated-api-html/index.html +++ b/docs/api/generated-api-html/index.html @@ -4344,9 +4344,9 @@

AirbyteStreamAndConfiguration<

AirbyteStreamConfiguration - Up

the mutable part of the stream to configure the destination
-
syncMode (optional)
SyncMode
+
syncMode
SyncMode
cursorField (optional)
array[String] Path to the field that will be used to determine if a record is new or modified since the last sync. This field is REQUIRED if sync_mode is incremental. Otherwise it is ignored.
-
destinationSyncMode (optional)
DestinationSyncMode
+
destinationSyncMode
DestinationSyncMode
primaryKey (optional)
array[array[String]] Paths to the fields that will be used as primary key. This field is REQUIRED if destination_sync_mode is *_dedup. Otherwise it is ignored.
aliasName (optional)
String Alias name to the stream to be used in the destination
selected (optional)
Boolean