🐛 Destination BigQueryDenormalized : fixed stackoverflow and array type parsing when source forget to send "items" details for it

airbyte-config/init/src/main/resources/config/STANDARD_DESTINATION_DEFINITION/079d5540-f236-4294-ba7c-ade8fd918496.json

@@ -2,6 +2,6 @@
   "destinationDefinitionId": "079d5540-f236-4294-ba7c-ade8fd918496",
   "name": "BigQuery (denormalized typed struct)",
   "dockerRepository": "airbyte/destination-bigquery-denormalized",
-  "dockerImageTag": "0.1.2",
+  "dockerImageTag": "0.1.4",
   "documentationUrl": "https://docs.airbyte.io/integrations/destinations/bigquery"
 }

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

@@ -27,7 +27,7 @@
 - destinationDefinitionId: 079d5540-f236-4294-ba7c-ade8fd918496
   name: BigQuery (denormalized typed struct)
   dockerRepository: airbyte/destination-bigquery-denormalized
-  dockerImageTag: 0.1.2
+  dockerImageTag: 0.1.4
   documentationUrl: https://docs.airbyte.io/integrations/destinations/bigquery
 - destinationDefinitionId: ca8f6566-e555-4b40-943a-545bf123117a
   name: Google Cloud Storage (GCS)

airbyte-integrations/connectors/destination-bigquery-denormalized/Dockerfile

@@ -8,5 +8,5 @@ COPY build/distributions/${APPLICATION}*.tar ${APPLICATION}.tar
 
 RUN tar xf ${APPLICATION}.tar --strip-components=1
 
-LABEL io.airbyte.version=0.1.3
+LABEL io.airbyte.version=0.1.4
 LABEL io.airbyte.name=airbyte/destination-bigquery-denormalized

airbyte-integrations/connectors/destination-bigquery-denormalized/src/main/java/io/airbyte/integrations/destination/bigquery/BigQueryDenormalizedDestination.java

@@ -117,7 +117,10 @@ private static Builder getField(BigQuerySQLNameTransformer namingResolver, Strin
             if (fieldDefinition.has("items")) {
               items = fieldDefinition.get("items");
             } else {
-              items = fieldDefinition;
+              LOGGER.warn("Source connector provided schema for ARRAY with missed \"items\", will assume that it's a String type");
+              // this is handler for case when we get "array" without "items"
+              // (https://github.com/airbytehq/airbyte/issues/5486)
+              items = getTypeStringSchema();
             }
             final Builder subField = getField(namingResolver, fieldName, items).setMode(Mode.REPEATED);
             // "Array of Array of" (nested arrays) are not permitted by BigQuery ("Array of Record of Array of"
@@ -152,6 +155,14 @@ private static Builder getField(BigQuerySQLNameTransformer namingResolver, Strin
     return builder;
   }
 
+  private static JsonNode getTypeStringSchema() {
+    return Jsons.deserialize("{\n"
+        + "    \"type\": [\n"
+        + "      \"string\"\n"
+        + "    ]\n"
+        + "  }");
+  }
+
   private static List<JsonSchemaType> getTypes(String fieldName, JsonNode type) {
     if (type == null) {
       LOGGER.warn("Field {} has no type defined, defaulting to STRING", fieldName);

airbyte-integrations/connectors/destination-bigquery-denormalized/src/test-integration/java/io/airbyte/integrations/destination/bigquery/BigQueryDenormalizedDestinationTest.java

@@ -82,6 +82,7 @@ class BigQueryDenormalizedDestinationTest {
   private BigQuery bigquery;
   private Dataset dataset;
   private ConfiguredAirbyteCatalog catalog;
+  private String datasetId;
 
   private boolean tornDown = true;
 
@@ -106,14 +107,10 @@ void setup(TestInfo info) throws IOException {
         .build()
         .getService();
 
-    final String datasetId = Strings.addRandomSuffix("airbyte_tests", "_", 8);
+    datasetId = Strings.addRandomSuffix("airbyte_tests", "_", 8);
     final String datasetLocation = "EU";
     MESSAGE_USERS1.getRecord().setNamespace(datasetId);
 
-    catalog = new ConfiguredAirbyteCatalog().withStreams(Lists.newArrayList(new ConfiguredAirbyteStream()
-        .withStream(new AirbyteStream().withName(USERS_STREAM_NAME).withNamespace(datasetId).withJsonSchema(getSchema()))
-        .withSyncMode(SyncMode.FULL_REFRESH).withDestinationSyncMode(DestinationSyncMode.OVERWRITE)));
-
     final DatasetInfo datasetInfo = DatasetInfo.newBuilder(datasetId).setLocation(datasetLocation).build();
     dataset = bigquery.create(datasetInfo);
 
@@ -162,6 +159,32 @@ private void tearDownBigQuery() {
 
   @Test
   void testNestedWrite() throws Exception {
+    catalog = new ConfiguredAirbyteCatalog().withStreams(Lists.newArrayList(new ConfiguredAirbyteStream()
+        .withStream(new AirbyteStream().withName(USERS_STREAM_NAME).withNamespace(datasetId).withJsonSchema(getSchema()))
+        .withSyncMode(SyncMode.FULL_REFRESH).withDestinationSyncMode(DestinationSyncMode.OVERWRITE)));
+
+    final BigQueryDestination destination = new BigQueryDenormalizedDestination();
+    final AirbyteMessageConsumer consumer = destination.getConsumer(config, catalog, Destination::defaultOutputRecordCollector);
+
+    consumer.accept(MESSAGE_USERS1);
+    consumer.close();
+
+    final List<JsonNode> usersActual = retrieveRecordsAsJson(USERS_STREAM_NAME);
+    final JsonNode expectedUsersJson = MESSAGE_USERS1.getRecord().getData();
+    assertEquals(usersActual.size(), 1);
+    final JsonNode resultJson = usersActual.get(0);
+    assertEquals(extractJsonValues(resultJson, "name"), extractJsonValues(expectedUsersJson, "name"));
+    assertEquals(extractJsonValues(resultJson, "grants"), extractJsonValues(expectedUsersJson, "grants"));
+    assertEquals(extractJsonValues(resultJson, "domain"), extractJsonValues(expectedUsersJson, "domain"));
+
+  }
+
+  @Test
+  void testNestedWriteHandleMissedItemsForArrayType() throws Exception {
+    catalog = new ConfiguredAirbyteCatalog().withStreams(Lists.newArrayList(new ConfiguredAirbyteStream()
+        .withStream(new AirbyteStream().withName(USERS_STREAM_NAME).withNamespace(datasetId).withJsonSchema(getSchemaWithInvalidArrayType()))
+        .withSyncMode(SyncMode.FULL_REFRESH).withDestinationSyncMode(DestinationSyncMode.OVERWRITE)));
+
     final BigQueryDestination destination = new BigQueryDenormalizedDestination();
     final AirbyteMessageConsumer consumer = destination.getConsumer(config, catalog, Destination::defaultOutputRecordCollector);
 
@@ -254,6 +277,45 @@ private JsonNode getSchema() {
 
   }
 
+  private JsonNode getSchemaWithInvalidArrayType() {
+    return Jsons.deserialize(
+        "{\n"
+            + "  \"type\": [\n"
+            + "    \"object\"\n"
+            + "  ],\n"
+            + "  \"properties\": {\n"
+            + "    \"name\": {\n"
+            + "      \"type\": [\n"
+            + "        \"string\"\n"
+            + "      ]\n"
+            + "    },\n"
+            + "    \"permissions\": {\n"
+            + "      \"type\": [\n"
+            + "        \"array\"\n"
+            + "      ],\n"
+            + "      \"items\": {\n"
+            + "        \"type\": [\n"
+            + "          \"object\"\n"
+            + "        ],\n"
+            + "        \"properties\": {\n"
+            + "          \"domain\": {\n"
+            + "            \"type\": [\n"
+            + "              \"string\"\n"
+            + "            ]\n"
+            + "          },\n"
+            + "          \"grants\": {\n"
+            + "            \"type\": [\n"
+            + "              \"array\"\n" // missed "items" element
+            + "            ]\n"
+            + "          }\n"
+            + "        }\n"
+            + "      }\n"
+            + "    }\n"
+            + "  }\n"
+            + "}");
+
+  }
+
   private static JsonNode getData() {
     return Jsons.deserialize(
         "{\n"

docs/integrations/destinations/bigquery.md

@@ -124,6 +124,7 @@ Therefore, Airbyte BigQuery destination will convert any invalid characters into
 
 | Version | Date | Pull Request | Subject |
 | :--- | :---  | :--- | :--- |
+| 0.1.4 | 2021-09-04 | [#5813](https://github.com/airbytehq/airbyte/pull/5813) | fix Stackoverflow error when receive a schema from source where "Array" type doesn't contain a required "items" element |
 | 0.1.3 | 2021-08-07 | [#5261](https://github.com/airbytehq/airbyte/pull/5261) | 🐛 Destination BigQuery(Denormalized): Fix processing arrays of records |
 | 0.1.2 | 2021-07-30 | [#5125](https://github.com/airbytehq/airbyte/pull/5125) | Enable `additionalPropertities` in spec.json |
 | 0.1.1 | 2021-06-21 | [#3555](https://github.com/airbytehq/airbyte/pull/3555) | Partial Success in BufferedStreamConsumer |