class BigQueryHelper {

  public void importFromGcs(
      String projectId,
      TableReference tableRef,
      @Nullable TableSchema schema,
      @Nullable TimePartitioning timePartitioning,
      @Nullable String kmsKeyName,
      BigQueryFileFormat sourceFormat,
      String createDisposition,
      String writeDisposition,
      List<String> gcsPaths,
      boolean awaitCompletion)
      throws IOException, InterruptedException {
    logger.atInfo().log(
        "Importing into table '%s' from %s paths; path[0] is '%s'; awaitCompletion: %s;"
            + " timePartitioning: %s",
        lazy(() -> BigQueryStrings.toString(tableRef)),
        gcsPaths.size(),
        gcsPaths.isEmpty() ? "(empty)" : gcsPaths.get(0),
        awaitCompletion,
        timePartitioning);

    // Create load conf with minimal requirements.
    JobConfigurationLoad loadConfig = new JobConfigurationLoad();
    loadConfig.setSchema(schema);
    loadConfig.setSourceFormat(sourceFormat.getFormatIdentifier());
    loadConfig.setSourceUris(gcsPaths);
    loadConfig.setDestinationTable(tableRef);
    loadConfig.setTimePartitioning(timePartitioning);
    loadConfig.setCreateDisposition(createDisposition);
    loadConfig.setWriteDisposition(writeDisposition);
    if (!Strings.isNullOrEmpty(kmsKeyName)) {
      loadConfig.setDestinationEncryptionConfiguration(
          new EncryptionConfiguration().setKmsKeyName(kmsKeyName));
    }
    // Auto detect the schema if we're not given one, otherwise use the passed schema.
    if (schema == null) {
      logger.atInfo().log("No import schema provided, auto detecting schema.");
      loadConfig.setAutodetect(true);
    } else {
      logger.atInfo().log("Using provided import schema '%s'.", schema);
    }

    JobConfiguration config = new JobConfiguration();
    config.setLoad(loadConfig);

    // Get the dataset to determine the location
    Dataset dataset;
    try {
      dataset = service.datasets().get(tableRef.getProjectId(), tableRef.getDatasetId()).execute();
    } catch (IOException ioe) {
      throw new IOException(
          String.format(
              "Failed to get dataset '%s' in project '%s' for table '%s'",
              tableRef.getDatasetId(), tableRef.getProjectId(), tableRef),
          ioe);
    }

    JobReference jobReference =
        createJobReference(projectId, "direct-bigqueryhelper-import", dataset.getLocation());
    Job job = new Job();
    job.setConfiguration(config);
    job.setJobReference(jobReference);

    // Insert and run job.
    insertJobOrFetchDuplicate(projectId, job);

    if (awaitCompletion) {
      // Poll until job is complete.
      BigQueryUtils.waitForJobCompletion(getRawBigquery(), projectId, jobReference, () -> {});
    }
  }

  public  BigQueryHelper(Bigquery service);

}