🐛 Destination S3: fixed s3 destination field naming for Parquet and Avro formats

airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/4816b78f-1489-44c1-9060-4b19d5fa9362.json

@@ -2,6 +2,6 @@
   "destinationDefinitionId": "4816b78f-1489-44c1-9060-4b19d5fa9362",
   "name": "S3",
   "dockerRepository": "airbyte/destination-s3",
-  "dockerImageTag": "0.1.10",
+  "dockerImageTag": "0.1.11",
   "documentationUrl": "https://docs.airbyte.io/integrations/destinations/s3"
 }

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

@@ -47,7 +47,7 @@
 - destinationDefinitionId: 4816b78f-1489-44c1-9060-4b19d5fa9362
   name: S3
   dockerRepository: airbyte/destination-s3
-  dockerImageTag: 0.1.10
+  dockerImageTag: 0.1.11
   documentationUrl: https://docs.airbyte.io/integrations/destinations/s3
 - destinationDefinitionId: f7a7d195-377f-cf5b-70a5-be6b819019dc
   name: Redshift

airbyte-db/lib/src/main/java/io/airbyte/db/mongodb/MongoUtils.java

@@ -110,8 +110,8 @@ private static void readBson(final Document document, final ObjectNode o, final
   }
 
   /**
-   * Gets 10.000 documents from collection, gathers all unique fields and its type. In case when one field has different types in 2 and more
-   * documents, the type is set to String.
+   * Gets 10.000 documents from collection, gathers all unique fields and its type. In case when one
+   * field has different types in 2 and more documents, the type is set to String.
    *
    * @param collection mongo collection
    * @return map of unique fields and its type

airbyte-integrations/connectors/destination-s3/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.1.10
+LABEL io.airbyte.version=0.1.11
 LABEL io.airbyte.name=airbyte/destination-s3

airbyte-integrations/connectors/destination-s3/src/main/java/io/airbyte/integrations/destination/s3/S3NameTransformer.java

@@ -0,0 +1,51 @@
+/*
+ * 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.integrations.destination.s3;
+
+import com.google.common.annotations.VisibleForTesting;
+import io.airbyte.integrations.destination.ExtendedNameTransformer;
+
+@VisibleForTesting
+public class S3NameTransformer extends ExtendedNameTransformer {
+
+  @Override
+  protected String applyDefaultCase(String input) {
+    return super.convertStreamName(input).toLowerCase();
+  }
+
+  @Override
+  public String getIdentifier(String name) {
+    return checkFirsCharInStreamName(convertStreamName(name));
+  }
+
+  private String checkFirsCharInStreamName(String name) {
+    if (name.substring(0, 1).matches("[A-Za-z_]")) {
+      return name;
+    } else {
+      return "_" + name;
+    }
+  }
+
+}

airbyte-integrations/connectors/destination-s3/src/main/java/io/airbyte/integrations/destination/s3/avro/JsonToAvroSchemaConverter.java

@@ -29,7 +29,7 @@
 import com.google.common.base.Preconditions;
 import io.airbyte.commons.util.MoreIterators;
 import io.airbyte.integrations.base.JavaBaseConstants;
-import io.airbyte.integrations.destination.StandardNameTransformer;
+import io.airbyte.integrations.destination.s3.S3NameTransformer;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
@@ -62,7 +62,7 @@ public class JsonToAvroSchemaConverter {
   private static final Logger LOGGER = LoggerFactory.getLogger(JsonToAvroSchemaConverter.class);
   private static final Schema TIMESTAMP_MILLIS_SCHEMA = LogicalTypes.timestampMillis()
       .addToSchema(Schema.create(Type.LONG));
-  private static final StandardNameTransformer NAME_TRANSFORMER = new StandardNameTransformer();
+  private static final S3NameTransformer NAME_TRANSFORMER = new S3NameTransformer();
 
   private final Map<String, String> standardizedNames = new HashMap<>();
 

airbyte-integrations/connectors/source-mongodb-v2/src/test-integration/java/io/airbyte/integrations/io/airbyte/integration_tests/sources/MongoDbSourceAcceptanceTest.java

@@ -115,10 +115,10 @@ protected ConfiguredAirbyteCatalog getConfiguredCatalog() throws Exception {
             .withDestinationSyncMode(DestinationSyncMode.APPEND)
             .withCursorField(List.of("_id"))
             .withStream(CatalogHelpers.createAirbyteStream(
-                    "test.acceptance_test",
-                    Field.of("_id", JsonSchemaPrimitive.STRING),
-                    Field.of("id", JsonSchemaPrimitive.STRING),
-                    Field.of("name", JsonSchemaPrimitive.STRING))
+                "test.acceptance_test",
+                Field.of("_id", JsonSchemaPrimitive.STRING),
+                Field.of("id", JsonSchemaPrimitive.STRING),
+                Field.of("name", JsonSchemaPrimitive.STRING))
                 .withSupportedSyncModes(Lists.newArrayList(SyncMode.INCREMENTAL))
                 .withDefaultCursorField(List.of("_id")))));
   }

docs/integrations/destinations/s3.md

@@ -374,6 +374,7 @@ Under the hood, an Airbyte data stream in Json schema is first converted to an A
 
 | Version | Date | Pull Request | Subject |
 | :--- | :---  | :--- | :--- |
+| 0.1.11 | 2021-08-30 | [#5729](https://github.com/airbytehq/airbyte/pull/5729) | Updated the field names to start with `_` for formats `Parquet` and `Avro`. | 
 | 0.1.10 | 2021-09-13 | [#4699](https://github.com/airbytehq/airbyte/pull/4699) | Added json config validator | 
 | 0.1.9 | 2021-07-12 | [#4666](https://github.com/airbytehq/airbyte/pull/4666) | Fix MinIO output for Parquet format. |
 | 0.1.8 | 2021-07-07 | [#4613](https://github.com/airbytehq/airbyte/pull/4613) | Patched schema converter to support combined restrictions. |