code: /*
   * Copyright (c) 2022 Airbyte, Inc., all rights reserved.
   */
  
  package io.airbyte.integrations.destination.bigquery.formatter;
  
  import com.fasterxml.jackson.databind.JsonNode;
  import com.google.cloud.bigquery.Schema;
  import io.airbyte.integrations.destination.StandardNameTransformer;
  import io.airbyte.protocol.models.AirbyteRecordMessage;
  import java.util.HashMap;
  import java.util.HashSet;
  import java.util.Map;
  import java.util.Set;
  import org.slf4j.Logger;
  import org.slf4j.LoggerFactory;
  
  /**
   * The class formats incoming JsonSchema and AirbyteRecord in order to be inline with a
   * corresponding uploader.
   */
  public abstract class BigQueryRecordFormatter {
  
    private static final Logger LOGGER = LoggerFactory.getLogger(BigQueryRecordFormatter.class);
  
    private Schema bigQuerySchema;
    private final Map<String, Set<String>> mapOfFailedFields = new HashMap<>();
    protected final StandardNameTransformer namingResolver;
    protected final JsonNode jsonSchema;
  
    /**
     * These parameters are required for the correct operation of denormalize version of the connector.
     */
    protected final Set<String> invalidKeys = new HashSet<>();
    protected final Set<String> fieldsContainRefDefinitionValue = new HashSet<>();
  
    public BigQueryRecordFormatter(final JsonNode jsonSchema, final StandardNameTransformer namingResolver) {
      this.namingResolver = namingResolver;
      this.jsonSchema = formatJsonSchema(jsonSchema.deepCopy());
    }
  
    protected JsonNode formatJsonSchema(final JsonNode jsonSchema) {
      // Do nothing by default
      return jsonSchema;
    };
  
    public abstract JsonNode formatRecord(AirbyteRecordMessage recordMessage);
  
    public Schema getBigQuerySchema() {
      if (bigQuerySchema == null) {
        bigQuerySchema = getBigQuerySchema(jsonSchema);
      }
      return bigQuerySchema;
    }
  
    public JsonNode getJsonSchema() {
      return jsonSchema;
    }
  
    protected abstract Schema getBigQuerySchema(JsonNode jsonSchema);
  
    protected void logFieldFail(final String error, final String fieldName) {
      mapOfFailedFields.putIfAbsent(error, new HashSet<>());
      mapOfFailedFields.get(error).add(fieldName);
    }
  
    public void printAndCleanFieldFails() {
      if (!mapOfFailedFields.isEmpty()) {
        mapOfFailedFields.forEach(
            (error, fieldNames) -> LOGGER.warn(
                "Field(s) fail with error {}. Fields : {} ",
                error,
                String.join(", ", fieldNames)));
        mapOfFailedFields.clear();
      }
    }
  
  }
  