🎉 Destination Redshift: Add "Loading Method" option to Redshift Desti…

…nation spec and UI (#13415) * airbyte-12709 Add "Loading Method" option to Redshift Destination spec. * airbyte-12709: Fixed unit tests. * airbyte-12709: Updated README.md. * airbyte-12709: change the number of PR in redshift.md. * airbyte-12709: Added backward compatibility for old json schema. * airbyte-12709: Fix PR comments. * airbyte-12709: Removed throwing an exception. Fixed PR comments. * airbyte-12709: Bump the airbyte version. * auto-bump connector version Co-authored-by: Octavia Squidington III <octavia-squidington-iii@users.noreply.github.com>
airbytehq · Jun 10, 2022 · b45014d · b45014d
1 parent 20bd923
commit b45014d
Show file tree

Hide file tree

Showing 14 changed files with 353 additions and 245 deletions.
diff --git a/airbyte-config/init/src/main/resources/seed/destination_definitions.yaml b/airbyte-config/init/src/main/resources/seed/destination_definitions.yaml
@@ -225,7 +225,7 @@
 - name: Redshift
   destinationDefinitionId: f7a7d195-377f-cf5b-70a5-be6b819019dc
   dockerRepository: airbyte/destination-redshift
-  dockerImageTag: 0.3.37
+  dockerImageTag: 0.3.39
   documentationUrl: https://docs.airbyte.io/integrations/destinations/redshift
   icon: redshift.svg
   resourceRequirements:

diff --git a/airbyte-config/init/src/main/resources/seed/destination_specs.yaml b/airbyte-config/init/src/main/resources/seed/destination_specs.yaml
@@ -3678,7 +3678,7 @@
     supported_destination_sync_modes:
     - "overwrite"
     - "append"
-- dockerImage: "airbyte/destination-redshift:0.3.37"
+- dockerImage: "airbyte/destination-redshift:0.3.39"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/destinations/redshift"
     connectionSpecification:
@@ -3730,94 +3730,121 @@
           - "public"
           default: "public"
           title: "Default Schema"
-        s3_bucket_name:
-          title: "S3 Bucket Name (Optional)"
-          type: "string"
-          description: "The name of the staging S3 bucket to use if utilising a COPY\
-            \ strategy. COPY is recommended for production workloads for better speed\
-            \ and scalability. See <a href=\"https://docs.aws.amazon.com/AmazonS3/latest/userguide/creating-bucket.html\"\
-            >AWS docs</a> for more details."
-          examples:
-          - "airbyte.staging"
-        s3_bucket_path:
-          title: "S3 Bucket Path (Optional)"
-          type: "string"
-          description: "The directory under the S3 bucket where data will be written.\
-            \ If not provided, then defaults to the root directory. See <a href=\"\
-            https://docs.aws.amazon.com/prescriptive-guidance/latest/defining-bucket-names-data-lakes/faq.html#:~:text=be%20globally%20unique.-,For%20S3%20bucket%20paths,-%2C%20you%20can%20use\"\
-            >path's name recommendations</a> for more details."
-          examples:
-          - "data_sync/test"
-        s3_bucket_region:
-          title: "S3 Bucket Region (Optional)"
-          type: "string"
-          default: ""
-          description: "The region of the S3 staging bucket to use if utilising a\
-            \ COPY strategy. See <a href=\"https://docs.aws.amazon.com/AmazonS3/latest/userguide/creating-bucket.html#:~:text=In-,Region,-%2C%20choose%20the%20AWS\"\
-            >AWS docs</a> for details."
-          enum:
-          - ""
-          - "us-east-1"
-          - "us-east-2"
-          - "us-west-1"
-          - "us-west-2"
-          - "af-south-1"
-          - "ap-east-1"
-          - "ap-south-1"
-          - "ap-northeast-1"
-          - "ap-northeast-2"
-          - "ap-northeast-3"
-          - "ap-southeast-1"
-          - "ap-southeast-2"
-          - "ca-central-1"
-          - "cn-north-1"
-          - "cn-northwest-1"
-          - "eu-central-1"
-          - "eu-north-1"
-          - "eu-south-1"
-          - "eu-west-1"
-          - "eu-west-2"
-          - "eu-west-3"
-          - "sa-east-1"
-          - "me-south-1"
-        access_key_id:
-          type: "string"
-          description: "This ID grants access to the above S3 staging bucket. Airbyte\
-            \ requires Read and Write permissions to the given bucket. See <a href=\"\
-            https://docs.aws.amazon.com/general/latest/gr/aws-sec-cred-types.html#access-keys-and-secret-access-keys\"\
-            >AWS docs</a> on how to generate an access key ID and secret access key."
-          title: "S3 Key Id (Optional)"
-          airbyte_secret: true
-        secret_access_key:
-          type: "string"
-          description: "The corresponding secret to the above access key id. See <a\
-            \ href=\"https://docs.aws.amazon.com/general/latest/gr/aws-sec-cred-types.html#access-keys-and-secret-access-keys\"\
-            >AWS docs</a> on how to generate an access key ID and secret access key."
-          title: "S3 Access Key (Optional)"
-          airbyte_secret: true
-        part_size:
-          type: "integer"
-          minimum: 10
-          maximum: 100
-          examples:
-          - "10"
-          description: "Increase this if syncing tables larger than 100GB. Only relevant\
-            \ for COPY. Files are streamed to S3 in parts. This determines the size\
-            \ of each part, in MBs. As S3 has a limit of 10,000 parts per file, part\
-            \ size affects the table size. This is 10MB by default, resulting in a\
-            \ default limit of 100GB tables. Note: a larger part size will result\
-            \ in larger memory requirements. A rule of thumb is to multiply the part\
-            \ size by 10 to get the memory requirement. Modify this with care. See\
-            \ <a href=\"https://docs.airbyte.com/integrations/destinations/redshift/#:~:text=above%20key%20id.-,Part%20Size,-Affects%20the%20size\"\
-            ,> docs</a> for details."
-          title: "Stream Part Size (Optional)"
-        purge_staging_data:
-          title: "Purge Staging Files and Tables (Optional)"
-          type: "boolean"
-          description: "Whether to delete the staging files from S3 after completing\
-            \ the sync. See <a href=\"https://docs.airbyte.com/integrations/destinations/redshift/#:~:text=the%20root%20directory.-,Purge%20Staging%20Data,-Whether%20to%20delete\"\
-            > docs</a> for details."
-          default: true
+        uploading_method:
+          title: "Uploading Method"
+          type: "object"
+          description: "The method how the data will be uploaded to the database."
+          oneOf:
+          - title: "Standard"
+            additionalProperties: false
+            required:
+            - "method"
+            properties:
+              method:
+                type: "string"
+                const: "Standard"
+          - title: "S3 Staging"
+            additionalProperties: false
+            required:
+            - "method"
+            - "s3_bucket_name"
+            - "s3_bucket_region"
+            - "access_key_id"
+            - "secret_access_key"
+            properties:
+              method:
+                type: "string"
+                const: "S3 Staging"
+              s3_bucket_name:
+                title: "S3 Bucket Name"
+                type: "string"
+                description: "The name of the staging S3 bucket to use if utilising\
+                  \ a COPY strategy. COPY is recommended for production workloads\
+                  \ for better speed and scalability. See <a href=\"https://docs.aws.amazon.com/AmazonS3/latest/userguide/creating-bucket.html\"\
+                  >AWS docs</a> for more details."
+                examples:
+                - "airbyte.staging"
+              s3_bucket_path:
+                title: "S3 Bucket Path (Optional)"
+                type: "string"
+                description: "The directory under the S3 bucket where data will be\
+                  \ written. If not provided, then defaults to the root directory.\
+                  \ See <a href=\"https://docs.aws.amazon.com/prescriptive-guidance/latest/defining-bucket-names-data-lakes/faq.html#:~:text=be%20globally%20unique.-,For%20S3%20bucket%20paths,-%2C%20you%20can%20use\"\
+                  >path's name recommendations</a> for more details."
+                examples:
+                - "data_sync/test"
+              s3_bucket_region:
+                title: "S3 Bucket Region"
+                type: "string"
+                default: ""
+                description: "The region of the S3 staging bucket to use if utilising\
+                  \ a COPY strategy. See <a href=\"https://docs.aws.amazon.com/AmazonS3/latest/userguide/creating-bucket.html#:~:text=In-,Region,-%2C%20choose%20the%20AWS\"\
+                  >AWS docs</a> for details."
+                enum:
+                - ""
+                - "us-east-1"
+                - "us-east-2"
+                - "us-west-1"
+                - "us-west-2"
+                - "af-south-1"
+                - "ap-east-1"
+                - "ap-south-1"
+                - "ap-northeast-1"
+                - "ap-northeast-2"
+                - "ap-northeast-3"
+                - "ap-southeast-1"
+                - "ap-southeast-2"
+                - "ca-central-1"
+                - "cn-north-1"
+                - "cn-northwest-1"
+                - "eu-central-1"
+                - "eu-north-1"
+                - "eu-south-1"
+                - "eu-west-1"
+                - "eu-west-2"
+                - "eu-west-3"
+                - "sa-east-1"
+                - "me-south-1"
+              access_key_id:
+                type: "string"
+                description: "This ID grants access to the above S3 staging bucket.\
+                  \ Airbyte requires Read and Write permissions to the given bucket.\
+                  \ See <a href=\"https://docs.aws.amazon.com/general/latest/gr/aws-sec-cred-types.html#access-keys-and-secret-access-keys\"\
+                  >AWS docs</a> on how to generate an access key ID and secret access\
+                  \ key."
+                title: "S3 Key Id"
+                airbyte_secret: true
+              secret_access_key:
+                type: "string"
+                description: "The corresponding secret to the above access key id.\
+                  \ See <a href=\"https://docs.aws.amazon.com/general/latest/gr/aws-sec-cred-types.html#access-keys-and-secret-access-keys\"\
+                  >AWS docs</a> on how to generate an access key ID and secret access\
+                  \ key."
+                title: "S3 Access Key"
+                airbyte_secret: true
+              part_size:
+                type: "integer"
+                minimum: 10
+                maximum: 100
+                examples:
+                - "10"
+                description: "Increase this if syncing tables larger than 100GB. Only\
+                  \ relevant for COPY. Files are streamed to S3 in parts. This determines\
+                  \ the size of each part, in MBs. As S3 has a limit of 10,000 parts\
+                  \ per file, part size affects the table size. This is 10MB by default,\
+                  \ resulting in a default limit of 100GB tables. Note: a larger part\
+                  \ size will result in larger memory requirements. A rule of thumb\
+                  \ is to multiply the part size by 10 to get the memory requirement.\
+                  \ Modify this with care. See <a href=\"https://docs.airbyte.com/integrations/destinations/redshift/#:~:text=above%20key%20id.-,Part%20Size,-Affects%20the%20size\"\
+                  ,> docs</a> for details."
+                title: "Stream Part Size (Optional)"
+              purge_staging_data:
+                title: "Purge Staging Files and Tables (Optional)"
+                type: "boolean"
+                description: "Whether to delete the staging files from S3 after completing\
+                  \ the sync. See <a href=\"https://docs.airbyte.com/integrations/destinations/redshift/#:~:text=the%20root%20directory.-,Purge%20Staging%20Data,-Whether%20to%20delete\"\
+                  > docs</a> for details."
+                default: true
     supportsIncremental: true
     supportsNormalization: true
     supportsDBT: true

diff --git a/airbyte-integrations/connectors/destination-redshift/Dockerfile b/airbyte-integrations/connectors/destination-redshift/Dockerfile
@@ -16,5 +16,5 @@ ENV APPLICATION destination-redshift
 
 COPY --from=build /airbyte /airbyte
 
-LABEL io.airbyte.version=0.3.37
+LABEL io.airbyte.version=0.3.39
 LABEL io.airbyte.name=airbyte/destination-redshift
diff --git a/airbyte-integrations/connectors/destination-redshift/README.md b/airbyte-integrations/connectors/destination-redshift/README.md
@@ -17,4 +17,8 @@ redshift.connString=<conn_string>
 redshift.user=<user>
 redshift.pass=<pass>
 ```
+## Actual secrets
+The actual secrets for integration tests could be found in Google Secrets Manager. It could be found by next labels:
+- SECRET_DESTINATION-REDSHIFT__CREDS - used for Standard tests. (__config.json__)
+- SECRET_DESTINATION-REDSHIFT_STAGING__CREDS - used for S3 Staging tests. (__config_staging.json__)
 
diff --git a/...shift/src/main/java/io/airbyte/integrations/destination/redshift/RedshiftDestination.java b/...shift/src/main/java/io/airbyte/integrations/destination/redshift/RedshiftDestination.java
@@ -4,6 +4,9 @@
 
 package io.airbyte.integrations.destination.redshift;
 
+import static io.airbyte.integrations.destination.redshift.validator.RedshiftUtil.findS3Options;
+import static io.airbyte.integrations.destination.redshift.validator.RedshiftUtil.anyOfS3FieldsAreNullOrEmpty;
+
 import com.fasterxml.jackson.databind.JsonNode;
 import io.airbyte.integrations.base.Destination;
 import io.airbyte.integrations.base.IntegrationRunner;
@@ -13,54 +16,43 @@
 import org.slf4j.LoggerFactory;
 
 /**
- * The Redshift Destination offers two replication strategies. The first inserts via a typical SQL
- * Insert statement. Although less efficient, this requires less user set up. See
- * {@link RedshiftInsertDestination} for more detail. The second inserts via streaming the data to
- * an S3 bucket, and Cop-ing the date into Redshift. This is more efficient, and recommended for
- * production workloads, but does require users to set up an S3 bucket and pass in additional
- * credentials. See {@link RedshiftStagingS3Destination} for more detail. This class inspect the
- * given arguments to determine which strategy to use.
+ * The Redshift Destination offers two replication strategies. The first inserts via a typical SQL Insert statement. Although less efficient, this requires less user set up. See {@link
+ * RedshiftInsertDestination} for more detail. The second inserts via streaming the data to an S3 bucket, and Cop-ing the date into Redshift. This is more efficient, and recommended for production
+ * workloads, but does require users to set up an S3 bucket and pass in additional credentials. See {@link RedshiftStagingS3Destination} for more detail. This class inspect the given arguments to
+ * determine which strategy to use.
  */
 public class RedshiftDestination extends SwitchingDestination<RedshiftDestination.DestinationType> {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(RedshiftDestination.class);
+  private static final String METHOD = "method";
+
+  private static final Map<DestinationType, Destination> destinationMap = Map.of(
+      DestinationType.STANDARD, new RedshiftInsertDestination(),
+      DestinationType.COPY_S3, new RedshiftStagingS3Destination()
+  );
 
   enum DestinationType {
     STANDARD,
     COPY_S3
   }
 
   public RedshiftDestination() {
-    super(DestinationType.class, RedshiftDestination::getTypeFromConfig, getTypeToDestination());
+    super(DestinationType.class, RedshiftDestination::getTypeFromConfig, destinationMap);
   }
 
-  public static DestinationType getTypeFromConfig(final JsonNode config) {
+  private static DestinationType getTypeFromConfig(final JsonNode config) {
     return determineUploadMode(config);
   }
 
-  public static Map<DestinationType, Destination> getTypeToDestination() {
-    return Map.of(
-        DestinationType.STANDARD, new RedshiftInsertDestination(),
-        DestinationType.COPY_S3, new RedshiftStagingS3Destination());
-  }
-
   public static DestinationType determineUploadMode(final JsonNode config) {
-    final var bucketNode = config.get("s3_bucket_name");
-    final var regionNode = config.get("s3_bucket_region");
-    final var accessKeyIdNode = config.get("access_key_id");
-    final var secretAccessKeyNode = config.get("secret_access_key");
 
-    if (isNullOrEmpty(bucketNode) && isNullOrEmpty(regionNode) && isNullOrEmpty(accessKeyIdNode)
-        && isNullOrEmpty(secretAccessKeyNode)) {
+    final JsonNode jsonNode = findS3Options(config);
+
+    if (anyOfS3FieldsAreNullOrEmpty(jsonNode)) {
       LOGGER.warn("The \"standard\" upload mode is not performant, and is not recommended for production. " +
           "Please use the Amazon S3 upload mode if you are syncing a large amount of data.");
       return DestinationType.STANDARD;
     }
-
-    if (isNullOrEmpty(bucketNode) && isNullOrEmpty(regionNode) && isNullOrEmpty(accessKeyIdNode)
-        && isNullOrEmpty(secretAccessKeyNode)) {
-      throw new RuntimeException("Error: Partially missing S3 Configuration.");
-    }
     return DestinationType.COPY_S3;
   }
 
@@ -70,9 +62,4 @@ public static void main(final String[] args) throws Exception {
     new IntegrationRunner(destination).run(args);
     LOGGER.info("completed destination: {}", RedshiftDestination.class);
   }
-
-  private static boolean isNullOrEmpty(JsonNode jsonNode) {
-    return jsonNode == null || jsonNode.asText().equals("");
-  }
-
 }
diff --git a/.../main/java/io/airbyte/integrations/destination/redshift/RedshiftStagingS3Destination.java b/.../main/java/io/airbyte/integrations/destination/redshift/RedshiftStagingS3Destination.java
@@ -9,6 +9,7 @@
 import static io.airbyte.integrations.destination.redshift.RedshiftInsertDestination.SSL_JDBC_PARAMETERS;
 import static io.airbyte.integrations.destination.redshift.RedshiftInsertDestination.USERNAME;
 import static io.airbyte.integrations.destination.redshift.RedshiftInsertDestination.getJdbcConfig;
+import static io.airbyte.integrations.destination.redshift.validator.RedshiftUtil.findS3Options;
 import static io.airbyte.integrations.destination.s3.S3DestinationConfig.getS3DestinationConfig;
 
 import com.fasterxml.jackson.databind.JsonNode;
@@ -48,7 +49,7 @@ public RedshiftStagingS3Destination() {
 
   @Override
   public AirbyteConnectionStatus check(final JsonNode config) {
-    final S3DestinationConfig s3Config = getS3DestinationConfig(config);
+    final S3DestinationConfig s3Config = getS3DestinationConfig(findS3Options(config));
     S3Destination.attemptS3WriteAndDelete(new S3StorageOperations(new RedshiftSQLNameTransformer(), s3Config.getS3Client(), s3Config), s3Config, "");
 
     final NamingConventionTransformer nameTransformer = getNamingResolver();
@@ -104,9 +105,9 @@ public JsonNode toJdbcConfig(final JsonNode config) {
 
   @Override
   public AirbyteMessageConsumer getConsumer(final JsonNode config,
-                                            final ConfiguredAirbyteCatalog catalog,
-                                            final Consumer<AirbyteMessage> outputRecordCollector) {
-    final S3DestinationConfig s3Config = getS3DestinationConfig(config);
+      final ConfiguredAirbyteCatalog catalog,
+      final Consumer<AirbyteMessage> outputRecordCollector) {
+    final S3DestinationConfig s3Config = getS3DestinationConfig(findS3Options(config));
     return new StagingConsumerFactory().create(
         outputRecordCollector,
         getDatabase(getDataSource(config)),

diff --git a/.../io/airbyte/integrations/destination/redshift/constants/RedshiftDestinationConstants.java b/.../io/airbyte/integrations/destination/redshift/constants/RedshiftDestinationConstants.java
@@ -0,0 +1,12 @@
+package io.airbyte.integrations.destination.redshift.constants;
+
+/**
+ * Constant holder for Redshift Destination
+ */
+public class RedshiftDestinationConstants {
+
+  private RedshiftDestinationConstants() {
+  }
+
+  public static final String UPLOADING_METHOD = "uploading_method";
+}