🎉 Destination S3 & GCS: support additional properties (#7288)

* Log json schema * Use patched json avro converter * Rename schema * Update unit test cases * Fix ab ap field schema conversion * Rename files * Add unit test cases * Fix dependency for databricks * Bump versions * Update documentations * Update gcs doc * Set additional properties field name * Revert s3 and gcs version * Specify extra props fields * Refactor json avro conversion doc * Update connector doc * Fix databricks spec typo * Bump connector versions in seed
airbytehq · Nov 3, 2021 · c9c41dc · c9c41dc
1 parent 9a07136
commit c9c41dc
Show file tree

Hide file tree

Showing 39 changed files with 812 additions and 474 deletions.
diff --git a/...esources/config/STANDARD_DESTINATION_DEFINITION/4816b78f-1489-44c1-9060-4b19d5fa9362.json b/...esources/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.12",
+  "dockerImageTag": "0.1.13",
   "documentationUrl": "https://docs.airbyte.io/integrations/destinations/s3"
 }
diff --git a/...esources/config/STANDARD_DESTINATION_DEFINITION/ca8f6566-e555-4b40-943a-545bf123117a.json b/...esources/config/STANDARD_DESTINATION_DEFINITION/ca8f6566-e555-4b40-943a-545bf123117a.json
@@ -2,6 +2,6 @@
   "destinationDefinitionId": "ca8f6566-e555-4b40-943a-545bf123117a",
   "name": "Google Cloud Storage (GCS)",
   "dockerRepository": "airbyte/destination-gcs",
-  "dockerImageTag": "0.1.2",
+  "dockerImageTag": "0.1.3",
   "documentationUrl": "https://docs.airbyte.io/integrations/destinations/gcs"
 }
diff --git a/airbyte-config/init/src/main/resources/seed/destination_definitions.yaml b/airbyte-config/init/src/main/resources/seed/destination_definitions.yaml
@@ -31,7 +31,7 @@
 - name: Google Cloud Storage (GCS)
   destinationDefinitionId: ca8f6566-e555-4b40-943a-545bf123117a
   dockerRepository: airbyte/destination-gcs
-  dockerImageTag: 0.1.2
+  dockerImageTag: 0.1.3
   documentationUrl: https://docs.airbyte.io/integrations/destinations/gcs
 - name: Google PubSub
   destinationDefinitionId: 356668e2-7e34-47f3-a3b0-67a8a481b692
@@ -93,7 +93,7 @@
 - name: S3
   destinationDefinitionId: 4816b78f-1489-44c1-9060-4b19d5fa9362
   dockerRepository: airbyte/destination-s3
-  dockerImageTag: 0.1.12
+  dockerImageTag: 0.1.13
   documentationUrl: https://docs.airbyte.io/integrations/destinations/s3
 - name: Snowflake
   destinationDefinitionId: 424892c4-daac-4491-b35d-c6688ba547ba

diff --git a/airbyte-integrations/connectors/destination-databricks/Dockerfile b/airbyte-integrations/connectors/destination-databricks/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.1
+LABEL io.airbyte.version=0.1.2
 LABEL io.airbyte.name=airbyte/destination-databricks
diff --git a/airbyte-integrations/connectors/destination-databricks/build.gradle b/airbyte-integrations/connectors/destination-databricks/build.gradle
@@ -24,7 +24,11 @@ dependencies {
     implementation group: 'org.apache.hadoop', name: 'hadoop-aws', version: '3.3.0'
     implementation group: 'org.apache.hadoop', name: 'hadoop-mapreduce-client-core', version: '3.3.0'
     implementation group: 'org.apache.parquet', name: 'parquet-avro', version: '1.12.0'
-    implementation group: 'tech.allegro.schema.json2avro', name: 'converter', version: '0.2.10'
+    implementation('tech.allegro.schema.json2avro:converter') {
+        version {
+            branch = 'master'
+        }
+    }
 
     integrationTestJavaImplementation project(':airbyte-integrations:bases:standard-destination-test')
     integrationTestJavaImplementation project(':airbyte-integrations:connectors:destination-databricks')

diff --git a/.../src/main/java/io/airbyte/integrations/destination/databricks/DatabricksStreamCopier.java b/.../src/main/java/io/airbyte/integrations/destination/databricks/DatabricksStreamCopier.java
@@ -90,7 +90,7 @@ public DatabricksStreamCopier(final String stagingFolder,
         s3Config.getBucketName(), s3Config.getBucketPath(), databricksConfig.getDatabaseSchema(), streamName);
 
     LOGGER.info("[Stream {}] Database schema: {}", streamName, schemaName);
-    LOGGER.info("[Stream {}] Parquet schema: {}", streamName, parquetWriter.getParquetSchema());
+    LOGGER.info("[Stream {}] Parquet schema: {}", streamName, parquetWriter.getSchema());
     LOGGER.info("[Stream {}] Tmp table {} location: {}", streamName, tmpTableName, tmpTableLocation);
     LOGGER.info("[Stream {}] Data table {} location: {}", streamName, destTableName, destTableLocation);
 

diff --git a/airbyte-integrations/connectors/destination-databricks/src/main/resources/spec.json b/airbyte-integrations/connectors/destination-databricks/src/main/resources/spec.json
@@ -84,7 +84,7 @@
                 "examples": ["airbyte.staging"]
               },
               "s3_bucket_path": {
-                "Title": "S3 Bucket Path",
+                "title": "S3 Bucket Path",
                 "type": "string",
                 "description": "The directory under the S3 bucket where data will be written.",
                 "examples": ["data_sync/test"]

diff --git a/airbyte-integrations/connectors/destination-gcs/Dockerfile b/airbyte-integrations/connectors/destination-gcs/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.2
+LABEL io.airbyte.version=0.1.3
 LABEL io.airbyte.name=airbyte/destination-gcs
diff --git a/airbyte-integrations/connectors/destination-gcs/build.gradle b/airbyte-integrations/connectors/destination-gcs/build.gradle
@@ -30,7 +30,11 @@ dependencies {
     implementation group: 'org.apache.hadoop', name: 'hadoop-aws', version: '3.3.0'
     implementation group: 'org.apache.hadoop', name: 'hadoop-mapreduce-client-core', version: '3.3.0'
     implementation group: 'org.apache.parquet', name: 'parquet-avro', version: '1.12.0'
-    implementation group: 'tech.allegro.schema.json2avro', name: 'converter', version: '0.2.10'
+    implementation('tech.allegro.schema.json2avro:converter') {
+        version {
+            branch = 'master'
+        }
+    }
 
     testImplementation 'org.apache.commons:commons-lang3:3.11'
 

diff --git a/...ination-gcs/src/main/java/io/airbyte/integrations/destination/gcs/avro/GcsAvroWriter.java b/...ination-gcs/src/main/java/io/airbyte/integrations/destination/gcs/avro/GcsAvroWriter.java
@@ -11,7 +11,6 @@
 import io.airbyte.integrations.destination.gcs.writer.BaseGcsWriter;
 import io.airbyte.integrations.destination.s3.S3Format;
 import io.airbyte.integrations.destination.s3.avro.AvroRecordFactory;
-import io.airbyte.integrations.destination.s3.avro.JsonFieldNameUpdater;
 import io.airbyte.integrations.destination.s3.avro.S3AvroFormatConfig;
 import io.airbyte.integrations.destination.s3.util.S3StreamTransferManagerHelper;
 import io.airbyte.integrations.destination.s3.writer.S3Writer;
@@ -27,6 +26,7 @@
 import org.apache.avro.generic.GenericDatumWriter;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import tech.allegro.schema.json2avro.converter.JsonAvroConverter;
 
 public class GcsAvroWriter extends BaseGcsWriter implements S3Writer {
 
@@ -42,7 +42,7 @@ public GcsAvroWriter(final GcsDestinationConfig config,
                        final ConfiguredAirbyteStream configuredStream,
                        final Timestamp uploadTimestamp,
                        final Schema schema,
-                       final JsonFieldNameUpdater nameUpdater)
+                       final JsonAvroConverter converter)
       throws IOException {
     super(config, s3Client, configuredStream);
 
@@ -52,7 +52,7 @@ public GcsAvroWriter(final GcsDestinationConfig config,
     LOGGER.info("Full GCS path for stream '{}': {}/{}", stream.getName(), config.getBucketName(),
         objectKey);
 
-    this.avroRecordFactory = new AvroRecordFactory(schema, nameUpdater);
+    this.avroRecordFactory = new AvroRecordFactory(schema, converter);
     this.uploadManager = S3StreamTransferManagerHelper.getDefault(
         config.getBucketName(), objectKey, s3Client, config.getFormatConfig().getPartSize());
     // We only need one output stream as we only have one input stream. This is reasonably performant.

diff --git a/...n-gcs/src/main/java/io/airbyte/integrations/destination/gcs/parquet/GcsParquetWriter.java b/...n-gcs/src/main/java/io/airbyte/integrations/destination/gcs/parquet/GcsParquetWriter.java
@@ -5,16 +5,13 @@
 package io.airbyte.integrations.destination.gcs.parquet;
 
 import com.amazonaws.services.s3.AmazonS3;
-import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.ObjectWriter;
-import com.fasterxml.jackson.databind.node.ObjectNode;
-import io.airbyte.integrations.base.JavaBaseConstants;
 import io.airbyte.integrations.destination.gcs.GcsDestinationConfig;
 import io.airbyte.integrations.destination.gcs.credential.GcsHmacKeyCredentialConfig;
 import io.airbyte.integrations.destination.gcs.writer.BaseGcsWriter;
 import io.airbyte.integrations.destination.s3.S3Format;
-import io.airbyte.integrations.destination.s3.avro.JsonFieldNameUpdater;
+import io.airbyte.integrations.destination.s3.avro.AvroRecordFactory;
 import io.airbyte.integrations.destination.s3.parquet.S3ParquetFormatConfig;
 import io.airbyte.integrations.destination.s3.writer.S3Writer;
 import io.airbyte.protocol.models.AirbyteRecordMessage;
@@ -42,21 +39,17 @@ public class GcsParquetWriter extends BaseGcsWriter implements S3Writer {
   private static final ObjectMapper MAPPER = new ObjectMapper();
   private static final ObjectWriter WRITER = MAPPER.writer();
 
-  private final Schema schema;
-  private final JsonFieldNameUpdater nameUpdater;
   private final ParquetWriter<Record> parquetWriter;
-  private final JsonAvroConverter converter = new JsonAvroConverter();
+  private final AvroRecordFactory avroRecordFactory;
 
   public GcsParquetWriter(final GcsDestinationConfig config,
                           final AmazonS3 s3Client,
                           final ConfiguredAirbyteStream configuredStream,
                           final Timestamp uploadTimestamp,
                           final Schema schema,
-                          final JsonFieldNameUpdater nameUpdater)
+                          final JsonAvroConverter converter)
       throws URISyntaxException, IOException {
     super(config, s3Client, configuredStream);
-    this.schema = schema;
-    this.nameUpdater = nameUpdater;
 
     final String outputFilename = BaseGcsWriter.getOutputFilename(uploadTimestamp, S3Format.PARQUET);
     final String objectKey = String.join("/", outputPrefix, outputFilename);
@@ -78,6 +71,7 @@ public GcsParquetWriter(final GcsDestinationConfig config,
         .withDictionaryPageSize(formatConfig.getDictionaryPageSize())
         .withDictionaryEncoding(formatConfig.isDictionaryEncoding())
         .build();
+    this.avroRecordFactory = new AvroRecordFactory(schema, converter);
   }
 
   public static Configuration getHadoopConfig(final GcsDestinationConfig config) {
@@ -99,16 +93,7 @@ public static Configuration getHadoopConfig(final GcsDestinationConfig config) {
 
   @Override
   public void write(final UUID id, final AirbyteRecordMessage recordMessage) throws IOException {
-    JsonNode inputData = recordMessage.getData();
-    inputData = nameUpdater.getJsonWithStandardizedFieldNames(inputData);
-
-    final ObjectNode jsonRecord = MAPPER.createObjectNode();
-    jsonRecord.put(JavaBaseConstants.COLUMN_NAME_AB_ID, UUID.randomUUID().toString());
-    jsonRecord.put(JavaBaseConstants.COLUMN_NAME_EMITTED_AT, recordMessage.getEmittedAt());
-    jsonRecord.setAll((ObjectNode) inputData);
-
-    final GenericData.Record avroRecord = converter.convertToGenericDataRecord(WRITER.writeValueAsBytes(jsonRecord), schema);
-    parquetWriter.write(avroRecord);
+    parquetWriter.write(avroRecordFactory.getAvroRecord(id, recordMessage));
   }
 
   @Override

diff --git a/...src/main/java/io/airbyte/integrations/destination/gcs/writer/ProductionWriterFactory.java b/...src/main/java/io/airbyte/integrations/destination/gcs/writer/ProductionWriterFactory.java
@@ -11,7 +11,7 @@
 import io.airbyte.integrations.destination.gcs.jsonl.GcsJsonlWriter;
 import io.airbyte.integrations.destination.gcs.parquet.GcsParquetWriter;
 import io.airbyte.integrations.destination.s3.S3Format;
-import io.airbyte.integrations.destination.s3.avro.JsonFieldNameUpdater;
+import io.airbyte.integrations.destination.s3.avro.AvroConstants;
 import io.airbyte.integrations.destination.s3.avro.JsonToAvroSchemaConverter;
 import io.airbyte.integrations.destination.s3.writer.S3Writer;
 import io.airbyte.protocol.models.AirbyteStream;
@@ -35,20 +35,17 @@ public S3Writer create(final GcsDestinationConfig config,
 
     if (format == S3Format.AVRO || format == S3Format.PARQUET) {
       final AirbyteStream stream = configuredStream.getStream();
+      LOGGER.info("Json schema for stream {}: {}", stream.getName(), stream.getJsonSchema());
 
       final JsonToAvroSchemaConverter schemaConverter = new JsonToAvroSchemaConverter();
       final Schema avroSchema = schemaConverter.getAvroSchema(stream.getJsonSchema(), stream.getName(), stream.getNamespace(), true);
-      final JsonFieldNameUpdater nameUpdater = new JsonFieldNameUpdater(schemaConverter.getStandardizedNames());
 
-      LOGGER.info("Paquet schema for stream {}: {}", stream.getName(), avroSchema.toString(false));
-      if (nameUpdater.hasNameUpdate()) {
-        LOGGER.info("The following field names will be standardized: {}", nameUpdater);
-      }
+      LOGGER.info("Avro schema for stream {}: {}", stream.getName(), avroSchema.toString(false));
 
       if (format == S3Format.AVRO) {
-        return new GcsAvroWriter(config, s3Client, configuredStream, uploadTimestamp, avroSchema, nameUpdater);
+        return new GcsAvroWriter(config, s3Client, configuredStream, uploadTimestamp, avroSchema, AvroConstants.JSON_CONVERTER);
       } else {
-        return new GcsParquetWriter(config, s3Client, configuredStream, uploadTimestamp, avroSchema, nameUpdater);
+        return new GcsParquetWriter(config, s3Client, configuredStream, uploadTimestamp, avroSchema, AvroConstants.JSON_CONVERTER);
       }
     }
 

diff --git a/...ration/java/io/airbyte/integrations/destination/gcs/GcsAvroDestinationAcceptanceTest.java b/...ration/java/io/airbyte/integrations/destination/gcs/GcsAvroDestinationAcceptanceTest.java
@@ -10,6 +10,7 @@
 import com.fasterxml.jackson.databind.ObjectReader;
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.integrations.destination.s3.S3Format;
+import io.airbyte.integrations.destination.s3.avro.AvroConstants;
 import io.airbyte.integrations.destination.s3.avro.JsonFieldNameUpdater;
 import io.airbyte.integrations.destination.s3.util.AvroRecordHelper;
 import java.util.LinkedList;
@@ -19,12 +20,9 @@
 import org.apache.avro.generic.GenericData;
 import org.apache.avro.generic.GenericData.Record;
 import org.apache.avro.generic.GenericDatumReader;
-import tech.allegro.schema.json2avro.converter.JsonAvroConverter;
 
 public class GcsAvroDestinationAcceptanceTest extends GcsDestinationAcceptanceTest {
 
-  private final JsonAvroConverter converter = new JsonAvroConverter();
-
   protected GcsAvroDestinationAcceptanceTest() {
     super(S3Format.AVRO);
   }
@@ -56,7 +54,7 @@ protected List<JsonNode> retrieveRecords(final TestDestinationEnv testEnv,
         final ObjectReader jsonReader = MAPPER.reader();
         while (dataFileReader.hasNext()) {
           final GenericData.Record record = dataFileReader.next();
-          final byte[] jsonBytes = converter.convertToJson(record);
+          final byte[] jsonBytes = AvroConstants.JSON_CONVERTER.convertToJson(record);
           JsonNode jsonRecord = jsonReader.readTree(jsonBytes);
           jsonRecord = nameUpdater.getJsonWithOriginalFieldNames(jsonRecord);
           jsonRecords.add(AvroRecordHelper.pruneAirbyteJson(jsonRecord));

diff --git a/...ntegration/java/io/airbyte/integrations/destination/gcs/GcsDestinationAcceptanceTest.java b/...ntegration/java/io/airbyte/integrations/destination/gcs/GcsDestinationAcceptanceTest.java
@@ -4,8 +4,6 @@
 
 package io.airbyte.integrations.destination.gcs;
 
-import static io.airbyte.integrations.destination.s3.S3DestinationConstants.NAME_TRANSFORMER;
-
 import com.amazonaws.services.s3.AmazonS3;
 import com.amazonaws.services.s3.model.DeleteObjectsRequest.KeyVersion;
 import com.amazonaws.services.s3.model.S3ObjectSummary;
@@ -15,6 +13,7 @@
 import io.airbyte.commons.io.IOs;
 import io.airbyte.commons.jackson.MoreMappers;
 import io.airbyte.commons.json.Jsons;
+import io.airbyte.integrations.destination.s3.S3DestinationConstants;
 import io.airbyte.integrations.destination.s3.S3Format;
 import io.airbyte.integrations.destination.s3.S3FormatConfig;
 import io.airbyte.integrations.destination.s3.util.S3OutputPathHelper;
@@ -89,7 +88,7 @@ protected List<S3ObjectSummary> getAllSyncedObjects(final String streamName, fin
         .listObjects(config.getBucketName(), outputPrefix)
         .getObjectSummaries()
         .stream()
-        .filter(o -> o.getKey().contains(NAME_TRANSFORMER.convertStreamName(streamName) + "/"))
+        .filter(o -> o.getKey().contains(S3DestinationConstants.NAME_TRANSFORMER.convertStreamName(streamName) + "/"))
         .sorted(Comparator.comparingLong(o -> o.getLastModified().getTime()))
         .collect(Collectors.toList());
     LOGGER.info(

diff --git a/...ion/java/io/airbyte/integrations/destination/gcs/GcsParquetDestinationAcceptanceTest.java b/...ion/java/io/airbyte/integrations/destination/gcs/GcsParquetDestinationAcceptanceTest.java
@@ -11,6 +11,7 @@
 import io.airbyte.commons.json.Jsons;
 import io.airbyte.integrations.destination.gcs.parquet.GcsParquetWriter;
 import io.airbyte.integrations.destination.s3.S3Format;
+import io.airbyte.integrations.destination.s3.avro.AvroConstants;
 import io.airbyte.integrations.destination.s3.avro.JsonFieldNameUpdater;
 import io.airbyte.integrations.destination.s3.util.AvroRecordHelper;
 import java.io.IOException;
@@ -22,12 +23,9 @@
 import org.apache.hadoop.conf.Configuration;
 import org.apache.parquet.avro.AvroReadSupport;
 import org.apache.parquet.hadoop.ParquetReader;
-import tech.allegro.schema.json2avro.converter.JsonAvroConverter;
 
 public class GcsParquetDestinationAcceptanceTest extends GcsDestinationAcceptanceTest {
 
-  private final JsonAvroConverter converter = new JsonAvroConverter();
-
   protected GcsParquetDestinationAcceptanceTest() {
     super(S3Format.PARQUET);
   }
@@ -63,7 +61,7 @@ protected List<JsonNode> retrieveRecords(final TestDestinationEnv testEnv,
         final ObjectReader jsonReader = MAPPER.reader();
         GenericData.Record record;
         while ((record = parquetReader.read()) != null) {
-          final byte[] jsonBytes = converter.convertToJson(record);
+          final byte[] jsonBytes = AvroConstants.JSON_CONVERTER.convertToJson(record);
           JsonNode jsonRecord = jsonReader.readTree(jsonBytes);
           jsonRecord = nameUpdater.getJsonWithOriginalFieldNames(jsonRecord);
           jsonRecords.add(AvroRecordHelper.pruneAirbyteJson(jsonRecord));

diff --git a/airbyte-integrations/connectors/destination-s3/Dockerfile b/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.12
+LABEL io.airbyte.version=0.1.13
 LABEL io.airbyte.name=airbyte/destination-s3
diff --git a/airbyte-integrations/connectors/destination-s3/build.gradle b/airbyte-integrations/connectors/destination-s3/build.gradle
@@ -26,7 +26,11 @@ dependencies {
     implementation group: 'org.apache.hadoop', name: 'hadoop-aws', version: '3.3.0'
     implementation group: 'org.apache.hadoop', name: 'hadoop-mapreduce-client-core', version: '3.3.0'
     implementation group: 'org.apache.parquet', name: 'parquet-avro', version: '1.12.0'
-    implementation group: 'tech.allegro.schema.json2avro', name: 'converter', version: '0.2.10'
+    implementation('tech.allegro.schema.json2avro:converter') {
+        version {
+            branch = 'master'
+        }
+    }
 
     testImplementation 'org.apache.commons:commons-lang3:3.11'
 

diff --git a/...stination-s3/src/main/java/io/airbyte/integrations/destination/s3/avro/AvroConstants.java b/...stination-s3/src/main/java/io/airbyte/integrations/destination/s3/avro/AvroConstants.java
@@ -0,0 +1,26 @@
+/*
+ * Copyright (c) 2021 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.integrations.destination.s3.avro;
+
+import java.util.Set;
+import tech.allegro.schema.json2avro.converter.JsonAvroConverter;
+
+public class AvroConstants {
+
+  // Field name with special character
+  public static final String DOC_KEY_VALUE_DELIMITER = ":";
+  public static final String DOC_KEY_ORIGINAL_NAME = "_airbyte_original_name";
+
+  public static final String AVRO_EXTRA_PROPS_FIELD = "_airbyte_additional_properties";
+  // This set must include _ab_additional_col in source_s3/source_files_abstract/stream.py
+  public static final Set<String> JSON_EXTRA_PROPS_FIELDS = Set.of("_ab_additional_properties", AVRO_EXTRA_PROPS_FIELD);
+  public static final AvroNameTransformer NAME_TRANSFORMER = new AvroNameTransformer();
+  public static final JsonAvroConverter JSON_CONVERTER = JsonAvroConverter.builder()
+      .setNameTransformer(NAME_TRANSFORMER::getIdentifier)
+      .setJsonAdditionalPropsFieldNames(JSON_EXTRA_PROPS_FIELDS)
+      .setAvroAdditionalPropsFieldName(AVRO_EXTRA_PROPS_FIELD)
+      .build();
+
+}
diff --git a/...ons/destination/s3/S3NameTransformer.java → ...tination/s3/avro/AvroNameTransformer.java b/...ons/destination/s3/S3NameTransformer.java → ...tination/s3/avro/AvroNameTransformer.java
@@ -2,11 +2,11 @@
  * Copyright (c) 2021 Airbyte, Inc., all rights reserved.
  */
 
-package io.airbyte.integrations.destination.s3;
+package io.airbyte.integrations.destination.s3.avro;
 
 import io.airbyte.integrations.destination.ExtendedNameTransformer;
 
-public class S3NameTransformer extends ExtendedNameTransformer {
+public class AvroNameTransformer extends ExtendedNameTransformer {
 
   @Override
   protected String applyDefaultCase(final String input) {