S3 destination: Updating processing data types for Avro/Parquet formats #13483
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,99 @@ | ||
| package io.airbyte.integrations.destination.gcs; | ||
|
|
||
| import static org.junit.jupiter.api.Assertions.assertEquals; | ||
| import static org.junit.jupiter.api.Assertions.assertTrue; | ||
|
|
||
| import com.fasterxml.jackson.databind.JsonNode; | ||
| import io.airbyte.commons.json.Jsons; | ||
| import io.airbyte.commons.resources.MoreResources; | ||
| import io.airbyte.integrations.destination.s3.S3Format; | ||
| import io.airbyte.integrations.destination.s3.avro.JsonSchemaType; | ||
| import io.airbyte.protocol.models.AirbyteCatalog; | ||
| 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 java.io.IOException; | ||
| import java.util.Arrays; | ||
| import java.util.List; | ||
| import java.util.Optional; | ||
| import java.util.Set; | ||
| import java.util.stream.Collectors; | ||
| import org.apache.avro.Schema; | ||
| import org.apache.avro.Schema.Type; | ||
| import org.apache.avro.generic.GenericData.Record; | ||
| import org.junit.jupiter.api.Test; | ||
|
|
||
| public abstract class GcsAvroParquetDestinationAcceptanceTest extends GcsDestinationAcceptanceTest { | ||
|
|
||
| protected GcsAvroParquetDestinationAcceptanceTest(S3Format s3Format) { | ||
| super(s3Format); | ||
| } | ||
|
|
||
| @Test | ||
| public void testNumberDataType() throws Exception { | ||
| final AirbyteCatalog catalog = readCatalogFromFile("number_data_type_test_catalog.json"); | ||
| final ConfiguredAirbyteCatalog configuredCatalog = CatalogHelpers.toDefaultConfiguredCatalog(catalog); | ||
| final List<AirbyteMessage> messages = readMessagesFromFile("number_data_type_test_messages.txt"); | ||
|
|
||
| final JsonNode config = getConfig(); | ||
| final String defaultSchema = getDefaultSchema(config); | ||
| runSyncAndVerifyStateOutput(config, messages, configuredCatalog, false); | ||
|
|
||
| for (final AirbyteStream stream : catalog.getStreams()) { | ||
| final String streamName = stream.getName(); | ||
| final String schema = stream.getNamespace() != null ? stream.getNamespace() : defaultSchema; | ||
|
|
||
| Set<Type> actualSchemaTypes = retrieveDataTypesFromPersistedFiles(streamName, schema); | ||
| Optional<Type> actualSchemaTypesWithoutNull = actualSchemaTypes.stream().filter(type -> !type.equals(Type.NULL)).findAny(); | ||
|
|
||
| JsonNode fieldDefinition = stream.getJsonSchema().get("properties").get("data"); | ||
| List<Type> expectedTypeList = getExpectedSchemaType(fieldDefinition); | ||
| assertEquals(1, expectedTypeList.size(), "Several not null data types are not supported for single stream"); | ||
| assertTrue(actualSchemaTypesWithoutNull.isPresent()); | ||
| assertEquals(expectedTypeList.get(0), actualSchemaTypesWithoutNull.get()); | ||
| } | ||
| } | ||
|
|
||
| private List<Type> getExpectedSchemaType(JsonNode fieldDefinition) { | ||
| final JsonNode typeProperty = fieldDefinition.get("type"); | ||
| final JsonNode airbyteTypeProperty = fieldDefinition.get("airbyte_type"); | ||
| final String airbyteTypePropertyText = airbyteTypeProperty == null ? null : airbyteTypeProperty.asText(); | ||
| return Arrays.stream(JsonSchemaType.values()) | ||
| .filter( | ||
| value -> value.getJsonSchemaType().equals(typeProperty.asText()) && compareAirbyteTypes(airbyteTypePropertyText, value)) | ||
| .map(JsonSchemaType::getAvroType) | ||
| .toList(); | ||
| } | ||
|
|
||
| private boolean compareAirbyteTypes(String airbyteTypePropertyText, JsonSchemaType value) { | ||
| if (airbyteTypePropertyText == null){ | ||
| return value.getJsonSchemaAirbyteType() == null; | ||
| } | ||
| return airbyteTypePropertyText.equals(value.getJsonSchemaAirbyteType()); | ||
| } | ||
|
|
||
| private AirbyteCatalog readCatalogFromFile(final String catalogFilename) throws IOException { | ||
| return Jsons.deserialize(MoreResources.readResource(catalogFilename), AirbyteCatalog.class); | ||
| } | ||
|
|
||
| private List<AirbyteMessage> readMessagesFromFile(final String messagesFilename) throws IOException { | ||
| return MoreResources.readResource(messagesFilename).lines() | ||
| .map(record -> Jsons.deserialize(record, AirbyteMessage.class)).collect(Collectors.toList()); | ||
| } | ||
|
|
||
| protected abstract Set<Type> retrieveDataTypesFromPersistedFiles(final String streamName, final String namespace) throws Exception; | ||
|
|
||
| protected Set<Type> getTypes(Record record) { | ||
| List<Schema> listAvroTypes = record | ||
| .getSchema() | ||
| .getField("data") | ||
| .schema() | ||
| .getTypes(); | ||
|
|
||
| return listAvroTypes | ||
| .stream() | ||
| .map(Schema::getType) | ||
| .collect(Collectors.toSet()); | ||
| } | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,47 @@ | ||
| { | ||
| "streams": [ | ||
| { | ||
| "name": "int_test", | ||
| "json_schema": { | ||
| "properties": { | ||
| "data": { | ||
| "type": "number", | ||
| "airbyte_type": "integer" | ||
| } | ||
| } | ||
| } | ||
| }, | ||
| { | ||
| "name": "big_integer_test", | ||
| "json_schema": { | ||
| "properties": { | ||
| "data": { | ||
| "type": "number", | ||
| "airbyte_type": "big_integer" | ||
| } | ||
| } | ||
| } | ||
| }, | ||
| { | ||
| "name": "float_test", | ||
| "json_schema": { | ||
| "properties": { | ||
| "data": { | ||
| "type": "number", | ||
| "airbyte_type": "float" | ||
| } | ||
| } | ||
| } | ||
| }, | ||
| { | ||
| "name": "default_number_test", | ||
| "json_schema": { | ||
| "properties": { | ||
| "data": { | ||
| "type": "number" | ||
| } | ||
| } | ||
| } | ||
| } | ||
| ] | ||
| } |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,13 @@ | ||
| {"type": "RECORD", "record": {"stream": "int_test", "emitted_at": 1602637589100, "data": { "data" : 42 }}} | ||
| {"type": "RECORD", "record": {"stream": "int_test", "emitted_at": 1602637589200, "data": { "data" : 0 }}} | ||
| {"type": "RECORD", "record": {"stream": "int_test", "emitted_at": 1602637589300, "data": { "data" : -12345 }}} | ||
| {"type": "RECORD", "record": {"stream": "big_integer_test", "emitted_at": 1602637589100, "data": { "data" : 1231123412412314 }}} | ||
| {"type": "RECORD", "record": {"stream": "big_integer_test", "emitted_at": 1602637589200, "data": { "data" : 0 }}} | ||
| {"type": "RECORD", "record": {"stream": "big_integer_test", "emitted_at": 1602637589300, "data": { "data" : -1234 }}} | ||
| {"type": "RECORD", "record": {"stream": "float_test", "emitted_at": 1602637589100, "data": { "data" : 56.78 }}} | ||
| {"type": "RECORD", "record": {"stream": "float_test", "emitted_at": 1602637589200, "data": { "data" : 0 }}} | ||
| {"type": "RECORD", "record": {"stream": "float_test", "emitted_at": 1602637589300, "data": { "data" : -12345.678 }}} | ||
| {"type": "RECORD", "record": {"stream": "default_number_test", "emitted_at": 1602637589100, "data": { "data" : 10000000000000000000000.1234 }}} | ||
| {"type": "RECORD", "record": {"stream": "default_number_test", "emitted_at": 1602637589200, "data": { "data" : 0 }}} | ||
| {"type": "RECORD", "record": {"stream": "default_number_test", "emitted_at": 1602637589300, "data": { "data" : -12345.678 }}} | ||
| {"type": "STATE", "state": { "data": {"start_date": "2022-02-14"}}} | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -4,39 +4,70 @@ | |
|
|
||
| package io.airbyte.integrations.destination.s3.avro; | ||
|
|
||
| import java.util.Arrays; | ||
| import java.util.List; | ||
| import javax.annotation.Nonnull; | ||
| import javax.annotation.Nullable; | ||
| import org.apache.avro.Schema; | ||
|
|
||
| /** | ||
| * Mapping of JsonSchema types to Avro types. | ||
| */ | ||
| public enum JsonSchemaType { | ||
|
|
||
| STRING("string", true, null, Schema.Type.STRING), | ||
| NUMBER_INT("number", true, "integer", Schema.Type.INT), | ||
| NUMBER_LONG("number", true, "big_integer", Schema.Type.LONG), | ||
|
There was a problem hiding this comment. this doesn't fully align with https://docs.airbyte.com/understanding-airbyte/supported-data-types#the-types, is that intentional? I.e. |
||
| NUMBER_FLOAT("number", true, "float", Schema.Type.FLOAT), | ||
| NUMBER("number", true, null, Schema.Type.DOUBLE), | ||
| INTEGER("integer", true, null, Schema.Type.INT), | ||
| BOOLEAN("boolean", true, null, Schema.Type.BOOLEAN), | ||
| NULL("null", true, null, Schema.Type.NULL), | ||
| OBJECT("object", false, null, Schema.Type.RECORD), | ||
| ARRAY("array", false, null, Schema.Type.ARRAY), | ||
| COMBINED("combined", false, null, Schema.Type.UNION); | ||
|
|
||
| private final String jsonSchemaType; | ||
| private final boolean isPrimitive; | ||
| private final Schema.Type avroType; | ||
| private final String jsonSchemaAirbyteType; | ||
|
|
||
| JsonSchemaType(final String jsonSchemaType, final boolean isPrimitive, final String jsonSchemaAirbyteType, final Schema.Type avroType) { | ||
| this.jsonSchemaType = jsonSchemaType; | ||
| this.jsonSchemaAirbyteType = jsonSchemaAirbyteType; | ||
| this.isPrimitive = isPrimitive; | ||
| this.avroType = avroType; | ||
| } | ||
|
|
||
| public static JsonSchemaType fromJsonSchemaType(final String jsonSchemaType) { | ||
| return fromJsonSchemaType(jsonSchemaType, null); | ||
| } | ||
|
|
||
| public static JsonSchemaType fromJsonSchemaType(final @Nonnull String jsonSchemaType, final @Nullable String jsonSchemaAirbyteType) { | ||
sashaNeshcheret marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| List<JsonSchemaType> matchSchemaType = null; | ||
| // Match by Type + airbyteType | ||
| if (jsonSchemaAirbyteType != null) { | ||
| matchSchemaType = Arrays.stream(values()) | ||
| .filter(type -> jsonSchemaType.equals(type.jsonSchemaType)) | ||
| .filter(type -> jsonSchemaAirbyteType.equals(type.jsonSchemaAirbyteType)) | ||
| .toList(); | ||
| } | ||
|
|
||
| // Match by Type are no results already | ||
| if (matchSchemaType == null || matchSchemaType.isEmpty()) { | ||
| matchSchemaType = | ||
| Arrays.stream(values()).filter(format -> jsonSchemaType.equals(format.jsonSchemaType) && format.jsonSchemaAirbyteType == null).toList(); | ||
| } | ||
|
|
||
| if (matchSchemaType.isEmpty()) { | ||
| throw new IllegalArgumentException("Unexpected json schema type: " + jsonSchemaType); | ||
sashaNeshcheret marked this conversation as resolved.
Show resolved
Hide resolved
|
||
| } else if (matchSchemaType.size() > 1) { | ||
| throw new RuntimeException( | ||
| "Match with more than one json format! Matched formats : " + matchSchemaType + ", Inputs jsonSchemaFormat : " + jsonSchemaType | ||
| + ", jsonSchemaAirbyteType : " + jsonSchemaAirbyteType); | ||
| } else { | ||
| return matchSchemaType.get(0); | ||
| } | ||
| } | ||
|
|
||
| public String getJsonSchemaType() { | ||
|
|
@@ -56,4 +87,7 @@ public String toString() { | |
| return jsonSchemaType; | ||
| } | ||
|
|
||
| public String getJsonSchemaAirbyteType() { | ||
| return jsonSchemaAirbyteType; | ||
| } | ||
| } | ||
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Uh oh!
There was an error while loading. Please reload this page.