Map number(integer) into an integer rather than a float

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

@@ -60,10 +60,6 @@
   dockerImageTag: 1.2.9
   documentationUrl: https://docs.airbyte.com/integrations/destinations/bigquery
   icon: bigquery.svg
-  normalizationConfig:
-    normalizationRepository: airbyte/normalization
-    normalizationTag: 0.2.25
-    normalizationIntegrationType: bigquery
   resourceRequirements:
     jobSpecific:
       - jobType: sync

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

@@ -40,6 +40,7 @@
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 import java.util.stream.Collectors;
+import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -68,15 +69,15 @@ private ArrayFormatter getArrayFormatter() {
     return arrayFormatter;
   }
 
-  public void setArrayFormatter(ArrayFormatter arrayFormatter) {
+  public void setArrayFormatter(final ArrayFormatter arrayFormatter) {
     this.arrayFormatter = arrayFormatter;
     this.jsonSchema = formatJsonSchema(this.originalJsonSchema.deepCopy());
     this.bigQuerySchema = getBigQuerySchema(jsonSchema);
   }
 
   @Override
   protected JsonNode formatJsonSchema(final JsonNode jsonSchema) {
-    var modifiedJsonSchema = jsonSchema.deepCopy(); // Issue #5912 is reopened (PR #11166) formatAllOfAndAnyOfFields(namingResolver, jsonSchema);
+    final var modifiedJsonSchema = jsonSchema.deepCopy(); // Issue #5912 is reopened (PR #11166) formatAllOfAndAnyOfFields(namingResolver, jsonSchema);
     getArrayFormatter().populateEmptyArrays(modifiedJsonSchema);
     getArrayFormatter().surroundArraysByObjects(modifiedJsonSchema);
     return modifiedJsonSchema;
@@ -117,7 +118,7 @@ private JsonNode formatData(final FieldList fields, final JsonNode root) {
     if (fields == null) {
       return root;
     }
-    JsonNode formattedData;
+    final JsonNode formattedData;
     if (root.isObject()) {
       formattedData = getObjectNode(fields, root);
     } else if (root.isArray()) {
@@ -151,7 +152,7 @@ private JsonNode getArrayNode(final FieldList fields, final JsonNode root) {
     } else {
       subFields = arrayField.getSubFields();
     }
-    List<JsonNode> arrayItems = MoreIterators.toList(root.elements()).stream()
+    final List<JsonNode> arrayItems = MoreIterators.toList(root.elements()).stream()
         .map(p -> formatData(subFields, p))
         .toList();
 
@@ -245,15 +246,15 @@ private static JsonNode getFileDefinition(final JsonNode fieldDefinition) {
   }
 
   private static JsonNode allOfAndAnyOfFieldProcessing(final String fieldName, final JsonNode fieldDefinition) {
-    ObjectReader reader = mapper.readerFor(new TypeReference<List<JsonNode>>() {});
-    List<JsonNode> list;
+    final ObjectReader reader = mapper.readerFor(new TypeReference<List<JsonNode>>() {});
+    final List<JsonNode> list;
     try {
       list = reader.readValue(fieldDefinition.get(fieldName));
-    } catch (IOException e) {
+    } catch (final IOException e) {
       throw new IllegalStateException(
           String.format("Failed to read and process the following field - %s", fieldDefinition));
     }
-    ObjectNode objectNode = mapper.createObjectNode();
+    final ObjectNode objectNode = mapper.createObjectNode();
     list.forEach(field -> {
       objectNode.set("big_query_" + field.get("type").asText(), field);
     });
@@ -268,8 +269,8 @@ private static JsonNode allOfAndAnyOfFieldProcessing(final String fieldName, fin
   private static Builder getField(final StandardNameTransformer namingResolver, final String key, final JsonNode fieldDefinition) {
     final String fieldName = namingResolver.getIdentifier(key);
     final Builder builder = Field.newBuilder(fieldName, StandardSQLTypeName.STRING);
-    JsonNode updatedFileDefinition = getFileDefinition(fieldDefinition);
-    JsonNode type = updatedFileDefinition.get(TYPE_FIELD);
+    final JsonNode updatedFileDefinition = getFileDefinition(fieldDefinition);
+    final JsonNode type = updatedFileDefinition.get(TYPE_FIELD);
     final JsonNode airbyteType = updatedFileDefinition.get(AIRBYTE_TYPE);
     final List<JsonSchemaType> fieldTypes = getTypes(fieldName, type);
     for (int i = 0; i < fieldTypes.size(); i++) {
@@ -288,9 +289,11 @@ private static Builder getField(final StandardNameTransformer namingResolver, fi
             builder.setType(primaryType.getBigQueryType());
           }
           case NUMBER -> {
-            if (airbyteType != null && airbyteType.asText().equals("big_integer")) {
+            if (airbyteType != null
+                && StringUtils.equalsAnyIgnoreCase(airbyteType.asText(),
+                "big_integer", "integer")) {
               builder.setType(StandardSQLTypeName.INT64);
-            } else {
+            }  else {
               builder.setType(primaryType.getBigQueryType());
             }
           }

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

@@ -2,7 +2,7 @@
   "documentationUrl": "https://docs.airbyte.com/integrations/destinations/bigquery",
   "supportsIncremental": true,
   "supportsNormalization": false,
-  "supportsDBT": true,
+  "supportsDBT": false,
   "supported_destination_sync_modes": ["overwrite", "append"],
   "connectionSpecification": {
     "$schema": "http://json-schema.org/draft-07/schema#",

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

@@ -81,11 +81,6 @@ protected JsonNode getFailCheckConfig() {
     return config;
   }
 
-  @Override
-  protected boolean supportsDBT() {
-    return true;
-  }
-
   @Override
   protected boolean implementsNamespaces() {
     return true;