🎉Redshift, Databricks, Snowflake, S3 Destinations: Make S3 output filename configurable (airbytehq#14494)

airbyte-4184: Make S3 output filename configurable

Author: alexandertsukanov

airbyte-config/init/src/main/resources/seed/destination_definitions.yaml

@@ -73,7 +73,7 @@
 - name: Databricks Lakehouse
   destinationDefinitionId: 072d5540-f236-4294-ba7c-ade8fd918496
   dockerRepository: airbyte/destination-databricks
-  dockerImageTag: 0.2.4
+  dockerImageTag: 0.2.5
   documentationUrl: https://docs.airbyte.io/integrations/destinations/databricks
   icon: databricks.svg
   releaseStage: alpha
@@ -231,7 +231,7 @@
 - name: Redshift
   destinationDefinitionId: f7a7d195-377f-cf5b-70a5-be6b819019dc
   dockerRepository: airbyte/destination-redshift
-  dockerImageTag: 0.3.46
+  dockerImageTag: 0.3.47
   documentationUrl: https://docs.airbyte.io/integrations/destinations/redshift
   icon: redshift.svg
   resourceRequirements:
@@ -250,7 +250,7 @@
 - name: S3
   destinationDefinitionId: 4816b78f-1489-44c1-9060-4b19d5fa9362
   dockerRepository: airbyte/destination-s3
-  dockerImageTag: 0.3.10
+  dockerImageTag: 0.3.11
   documentationUrl: https://docs.airbyte.io/integrations/destinations/s3
   icon: s3.svg
   resourceRequirements:
@@ -270,7 +270,7 @@
 - name: Snowflake
   destinationDefinitionId: 424892c4-daac-4491-b35d-c6688ba547ba
   dockerRepository: airbyte/destination-snowflake
-  dockerImageTag: 0.4.32
+  dockerImageTag: 0.4.33
   documentationUrl: https://docs.airbyte.io/integrations/destinations/snowflake
   icon: snowflake.svg
   resourceRequirements:

airbyte-config/init/src/main/resources/seed/destination_specs.yaml

@@ -966,7 +966,7 @@
     - "overwrite"
     - "append"
     - "append_dedup"
-- dockerImage: "airbyte/destination-databricks:0.2.4"
+- dockerImage: "airbyte/destination-databricks:0.2.5"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/destinations/databricks"
     connectionSpecification:
@@ -1116,6 +1116,18 @@
                 - "a012345678910ABCDEFGH/AbCdEfGhEXAMPLEKEY"
                 airbyte_secret: true
                 order: 6
+              file_name_pattern:
+                type: "string"
+                description: "The pattern allows you to set the file-name format for\
+                  \ the S3 staging file(s)"
+                title: "S3 Filename pattern (Optional)"
+                examples:
+                - "{date}"
+                - "{date:yyyy_MM}"
+                - "{timestamp}"
+                - "{part_number}"
+                - "{sync_id}"
+                order: 7
           order: 7
         purge_staging_data:
           title: "Purge Staging Files and Tables"
@@ -3708,7 +3720,7 @@
     supported_destination_sync_modes:
     - "overwrite"
     - "append"
-- dockerImage: "airbyte/destination-redshift:0.3.46"
+- dockerImage: "airbyte/destination-redshift:0.3.47"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/destinations/redshift"
     connectionSpecification:
@@ -3833,6 +3845,18 @@
                 - "eu-west-3"
                 - "sa-east-1"
                 - "me-south-1"
+              file_name_pattern:
+                type: "string"
+                description: "The pattern allows you to set the file-name format for\
+                  \ the S3 staging file(s)"
+                title: "S3 Filename pattern (Optional)"
+                examples:
+                - "{date}"
+                - "{date:yyyy_MM}"
+                - "{timestamp}"
+                - "{part_number}"
+                - "{sync_id}"
+                order: 8
               access_key_id:
                 type: "string"
                 description: "This ID grants access to the above S3 staging bucket.\
@@ -3946,7 +3970,7 @@
     supported_destination_sync_modes:
     - "append"
     - "overwrite"
-- dockerImage: "airbyte/destination-s3:0.3.10"
+- dockerImage: "airbyte/destination-s3:0.3.11"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/destinations/s3"
     connectionSpecification:
@@ -4305,6 +4329,18 @@
           examples:
           - "${NAMESPACE}/${STREAM_NAME}/${YEAR}_${MONTH}_${DAY}_${EPOCH}_"
           order: 7
+        file_name_pattern:
+          type: "string"
+          description: "The pattern allows you to set the file-name format for the\
+            \ S3 staging file(s)"
+          title: "S3 Filename pattern (Optional)"
+          examples:
+          - "{date}"
+          - "{date:yyyy_MM}"
+          - "{timestamp}"
+          - "{part_number}"
+          - "{sync_id}"
+          order: 8
     supportsIncremental: true
     supportsNormalization: false
     supportsDBT: false
@@ -4364,7 +4400,7 @@
     supported_destination_sync_modes:
     - "overwrite"
     - "append"
-- dockerImage: "airbyte/destination-snowflake:0.4.32"
+- dockerImage: "airbyte/destination-snowflake:0.4.33"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/destinations/snowflake"
     connectionSpecification:
@@ -4635,6 +4671,18 @@
                         \ or 256 bits. Leave blank to have Airbyte generate an ephemeral\
                         \ key for each sync."
                       airbyte_secret: true
+              file_name_pattern:
+                type: "string"
+                description: "The pattern allows you to set the file-name format for\
+                  \ the S3 staging file(s)"
+                title: "S3 Filename pattern (Optional)"
+                examples:
+                - "{date}"
+                - "{date:yyyy_MM}"
+                - "{timestamp}"
+                - "{part_number}"
+                - "{sync_id}"
+                order: 7
           - title: "Google Cloud Storage Staging"
             description: "Recommended for large production workloads for better speed\
               \ and scalability."

airbyte-integrations/connectors/destination-databricks/Dockerfile

@@ -16,5 +16,5 @@ ENV APPLICATION destination-databricks
 
 COPY --from=build /airbyte /airbyte
 
-LABEL io.airbyte.version=0.2.4
+LABEL io.airbyte.version=0.2.5
 LABEL io.airbyte.name=airbyte/destination-databricks

airbyte-integrations/connectors/destination-databricks/src/main/java/io/airbyte/integrations/destination/databricks/DatabricksDestinationConfig.java

@@ -8,6 +8,7 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.google.common.base.Preconditions;
 import io.airbyte.integrations.destination.s3.S3DestinationConfig;
+import io.airbyte.integrations.destination.s3.constant.S3Constants;
 import io.airbyte.integrations.destination.s3.parquet.S3ParquetFormatConfig;
 
 /**
@@ -59,15 +60,18 @@ public static DatabricksDestinationConfig get(final JsonNode config) {
   }
 
   public static S3DestinationConfig getDataSource(final JsonNode dataSource) {
-    return S3DestinationConfig.create(
-        dataSource.get("s3_bucket_name").asText(),
-        dataSource.get("s3_bucket_path").asText(),
-        dataSource.get("s3_bucket_region").asText())
+    final S3DestinationConfig.Builder builder = S3DestinationConfig.create(
+        dataSource.get(S3Constants.S_3_BUCKET_NAME).asText(),
+        dataSource.get(S3Constants.S_3_BUCKET_PATH).asText(),
+        dataSource.get(S3Constants.S_3_BUCKET_REGION).asText())
         .withAccessKeyCredential(
-            dataSource.get("s3_access_key_id").asText(),
-            dataSource.get("s3_secret_access_key").asText())
-        .withFormatConfig(new S3ParquetFormatConfig(new ObjectMapper().createObjectNode()))
-        .get();
+            dataSource.get(S3Constants.S_3_ACCESS_KEY_ID).asText(),
+            dataSource.get(S3Constants.S_3_SECRET_ACCESS_KEY).asText())
+        .withFormatConfig(new S3ParquetFormatConfig(new ObjectMapper().createObjectNode()));
+    if (dataSource.has(S3Constants.FILE_NAME_PATTERN)) {
+      builder.withFileNamePattern(dataSource.get(S3Constants.FILE_NAME_PATTERN).asText());
+    }
+    return builder.get();
   }
 
   public String getDatabricksServerHostname() {

airbyte-integrations/connectors/destination-databricks/src/main/java/io/airbyte/integrations/destination/databricks/DatabricksStreamCopier.java

@@ -4,6 +4,8 @@
 
 package io.airbyte.integrations.destination.databricks;
 
+import static org.apache.logging.log4j.util.Strings.EMPTY;
+
 import com.amazonaws.services.s3.AmazonS3;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import io.airbyte.db.jdbc.JdbcDatabase;
@@ -18,6 +20,7 @@
 import io.airbyte.protocol.models.ConfiguredAirbyteStream;
 import io.airbyte.protocol.models.DestinationSyncMode;
 import java.sql.Timestamp;
+import java.util.Optional;
 import java.util.UUID;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -209,6 +212,7 @@ static S3DestinationConfig getStagingS3DestinationConfig(final S3DestinationConf
     return S3DestinationConfig.create(config)
         .withBucketPath(String.join("/", config.getBucketPath(), stagingFolder))
         .withFormatConfig(new S3ParquetFormatConfig(MAPPER.createObjectNode()))
+        .withFileNamePattern(Optional.ofNullable(config.getFileNamePattern()).orElse(EMPTY))
         .get();
   }
 

airbyte-integrations/connectors/destination-databricks/src/main/resources/spec.json

@@ -147,6 +147,15 @@
                 "examples": ["a012345678910ABCDEFGH/AbCdEfGhEXAMPLEKEY"],
                 "airbyte_secret": true,
                 "order": 6
+              },
+              "file_name_pattern": {
+                "type": "string",
+                "description": "The pattern allows you to set the file-name format for the S3 staging file(s)",
+                "title": "S3 Filename pattern (Optional)",
+                "examples": [
+                  "{date}", "{date:yyyy_MM}", "{timestamp}", "{part_number}", "{sync_id}"
+                ],
+                "order": 7
               }
             }
           }

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.46
+LABEL io.airbyte.version=0.3.47
 LABEL io.airbyte.name=airbyte/destination-redshift

airbyte-integrations/connectors/destination-redshift/src/main/resources/spec.json

@@ -121,6 +121,15 @@
                   "me-south-1"
                 ]
               },
+              "file_name_pattern": {
+                "type": "string",
+                "description": "The pattern allows you to set the file-name format for the S3 staging file(s)",
+                "title": "S3 Filename pattern (Optional)",
+                "examples": [
+                  "{date}", "{date:yyyy_MM}", "{timestamp}", "{part_number}", "{sync_id}"
+                ],
+                "order": 8
+              },
               "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.",

airbyte-integrations/connectors/destination-s3/Dockerfile

@@ -16,5 +16,5 @@ ENV APPLICATION destination-s3
 
 COPY --from=build /airbyte /airbyte
 
-LABEL io.airbyte.version=0.3.10
+LABEL io.airbyte.version=0.3.11
 LABEL io.airbyte.name=airbyte/destination-s3