import com.google.api.services.bigquery.model.Job; //导入依赖的package包/类
@Override
void cleanup(PassThroughThenCleanup.ContextContainer c) throws Exception {
  PipelineOptions options = c.getPipelineOptions();
  BigQueryOptions bqOptions = options.as(BigQueryOptions.class);
  String jobUuid = c.getJobId();
  final String extractDestinationDir =
      resolveTempLocation(bqOptions.getTempLocation(), "BigQueryExtractTemp", jobUuid);
  final String executingProject = bqOptions.getProject();
  JobReference jobRef =
      new JobReference()
          .setProjectId(executingProject)
          .setJobId(getExtractJobId(createJobIdToken(bqOptions.getJobName(), jobUuid)));

  Job extractJob = getBigQueryServices().getJobService(bqOptions).getJob(jobRef);

  if (extractJob != null) {
    List<ResourceId> extractFiles =
        getExtractFilePaths(extractDestinationDir, extractJob);
    if (extractFiles != null && !extractFiles.isEmpty()) {
      FileSystems.delete(
          extractFiles, MoveOptions.StandardMoveOptions.IGNORE_MISSING_FILES);
    }
  }
}
 

import com.google.api.services.bigquery.model.Job; //导入依赖的package包/类
static List<ResourceId> getExtractFilePaths(String extractDestinationDir, Job extractJob)
    throws IOException {
  JobStatistics jobStats = extractJob.getStatistics();
  List<Long> counts = jobStats.getExtract().getDestinationUriFileCounts();
  if (counts.size() != 1) {
    String errorMessage = (counts.size() == 0
            ? "No destination uri file count received."
            : String.format(
                "More than one destination uri file count received. First two are %s, %s",
                counts.get(0), counts.get(1)));
    throw new RuntimeException(errorMessage);
  }
  long filesCount = counts.get(0);

  ImmutableList.Builder<ResourceId> paths = ImmutableList.builder();
  ResourceId extractDestinationDirResourceId =
      FileSystems.matchNewResource(extractDestinationDir, true /* isDirectory */);
  for (long i = 0; i < filesCount; ++i) {
    ResourceId filePath = extractDestinationDirResourceId.resolve(
        String.format("%012d%s", i, ".avro"),
        ResolveOptions.StandardResolveOptions.RESOLVE_FILE);
    paths.add(filePath);
  }
  return paths.build();
}
 

import com.google.api.services.bigquery.model.Job; //导入依赖的package包/类
@Override
public JobStatistics dryRunQuery(String projectId, JobConfigurationQuery queryConfig)
    throws InterruptedException, IOException {
  Job job = new Job()
      .setConfiguration(new JobConfiguration()
          .setQuery(queryConfig)
          .setDryRun(true));
  return executeWithRetries(
      client.jobs().insert(projectId, job),
      String.format(
          "Unable to dry run query: %s, aborting after %d retries.",
          queryConfig, MAX_RPC_RETRIES),
      Sleeper.DEFAULT,
      createDefaultBackoff(),
      ALWAYS_RETRY).getStatistics();
}
 

import com.google.api.services.bigquery.model.Job; //导入依赖的package包/类
private void executeQuery(
    String executingProject,
    String jobId,
    TableReference destinationTable,
    JobService jobService) throws IOException, InterruptedException {
  JobReference jobRef = new JobReference()
      .setProjectId(executingProject)
      .setJobId(jobId);

  JobConfigurationQuery queryConfig = createBasicQueryConfig()
      .setAllowLargeResults(true)
      .setCreateDisposition("CREATE_IF_NEEDED")
      .setDestinationTable(destinationTable)
      .setPriority("BATCH")
      .setWriteDisposition("WRITE_EMPTY");

  jobService.startQueryJob(jobRef, queryConfig);
  Job job = jobService.pollJob(jobRef, JOB_POLL_MAX_RETRIES);
  if (BigQueryHelpers.parseStatus(job) != Status.SUCCEEDED) {
    throw new IOException(String.format(
        "Query job %s failed, status: %s.", jobId,
        BigQueryHelpers.statusToPrettyString(job.getStatus())));
  }
}
 

import com.google.api.services.bigquery.model.Job; //导入依赖的package包/类
/**
 * Tests that {@link BigQueryServicesImpl.JobServiceImpl#startLoadJob} succeeds.
 */
@Test
public void testStartLoadJobSucceeds() throws IOException, InterruptedException {
  Job testJob = new Job();
  JobReference jobRef = new JobReference();
  jobRef.setJobId("jobId");
  jobRef.setProjectId("projectId");
  testJob.setJobReference(jobRef);

  when(response.getContentType()).thenReturn(Json.MEDIA_TYPE);
  when(response.getStatusCode()).thenReturn(200);
  when(response.getContent()).thenReturn(toStream(testJob));

  Sleeper sleeper = new FastNanoClockAndSleeper();
  JobServiceImpl.startJob(
      testJob, new ApiErrorExtractor(), bigquery, sleeper,
      BackOffAdapter.toGcpBackOff(FluentBackoff.DEFAULT.backoff()));

  verify(response, times(1)).getStatusCode();
  verify(response, times(1)).getContent();
  verify(response, times(1)).getContentType();
  expectedLogs.verifyInfo(String.format("Started BigQuery job: %s", jobRef));
}
 

import com.google.api.services.bigquery.model.Job; //导入依赖的package包/类
/**
 * Tests that {@link BigQueryServicesImpl.JobServiceImpl#startLoadJob} succeeds
 * with an already exist job.
 */
@Test
public void testStartLoadJobSucceedsAlreadyExists() throws IOException, InterruptedException {
  Job testJob = new Job();
  JobReference jobRef = new JobReference();
  jobRef.setJobId("jobId");
  jobRef.setProjectId("projectId");
  testJob.setJobReference(jobRef);

  when(response.getStatusCode()).thenReturn(409); // 409 means already exists

  Sleeper sleeper = new FastNanoClockAndSleeper();
  JobServiceImpl.startJob(
      testJob, new ApiErrorExtractor(), bigquery, sleeper,
      BackOffAdapter.toGcpBackOff(FluentBackoff.DEFAULT.backoff()));

  verify(response, times(1)).getStatusCode();
  verify(response, times(1)).getContent();
  verify(response, times(1)).getContentType();
  expectedLogs.verifyNotLogged("Started BigQuery job");
}
 

import com.google.api.services.bigquery.model.Job; //导入依赖的package包/类
/**
 * Tests that {@link BigQueryServicesImpl.JobServiceImpl#startLoadJob} succeeds with a retry.
 */
@Test
public void testStartLoadJobRetry() throws IOException, InterruptedException {
  Job testJob = new Job();
  JobReference jobRef = new JobReference();
  jobRef.setJobId("jobId");
  jobRef.setProjectId("projectId");
  testJob.setJobReference(jobRef);

  // First response is 403 rate limited, second response has valid payload.
  when(response.getContentType()).thenReturn(Json.MEDIA_TYPE);
  when(response.getStatusCode()).thenReturn(403).thenReturn(200);
  when(response.getContent())
      .thenReturn(toStream(errorWithReasonAndStatus("rateLimitExceeded", 403)))
      .thenReturn(toStream(testJob));

  Sleeper sleeper = new FastNanoClockAndSleeper();
  JobServiceImpl.startJob(
      testJob, new ApiErrorExtractor(), bigquery, sleeper,
      BackOffAdapter.toGcpBackOff(FluentBackoff.DEFAULT.backoff()));

  verify(response, times(2)).getStatusCode();
  verify(response, times(2)).getContent();
  verify(response, times(2)).getContentType();
}
 

import com.google.api.services.bigquery.model.Job; //导入依赖的package包/类
/**
 * Tests that {@link BigQueryServicesImpl.JobServiceImpl#pollJob} succeeds.
 */
@Test
public void testPollJobSucceeds() throws IOException, InterruptedException {
  Job testJob = new Job();
  testJob.setStatus(new JobStatus().setState("DONE"));

  when(response.getContentType()).thenReturn(Json.MEDIA_TYPE);
  when(response.getStatusCode()).thenReturn(200);
  when(response.getContent()).thenReturn(toStream(testJob));

  BigQueryServicesImpl.JobServiceImpl jobService =
      new BigQueryServicesImpl.JobServiceImpl(bigquery);
  JobReference jobRef = new JobReference()
      .setProjectId("projectId")
      .setJobId("jobId");
  Job job = jobService.pollJob(jobRef, Sleeper.DEFAULT, BackOff.ZERO_BACKOFF);

  assertEquals(testJob, job);
  verify(response, times(1)).getStatusCode();
  verify(response, times(1)).getContent();
  verify(response, times(1)).getContentType();
}
 

import com.google.api.services.bigquery.model.Job; //导入依赖的package包/类
/**
 * Tests that {@link BigQueryServicesImpl.JobServiceImpl#pollJob} fails.
 */
@Test
public void testPollJobFailed() throws IOException, InterruptedException {
  Job testJob = new Job();
  testJob.setStatus(new JobStatus().setState("DONE").setErrorResult(new ErrorProto()));

  when(response.getContentType()).thenReturn(Json.MEDIA_TYPE);
  when(response.getStatusCode()).thenReturn(200);
  when(response.getContent()).thenReturn(toStream(testJob));

  BigQueryServicesImpl.JobServiceImpl jobService =
      new BigQueryServicesImpl.JobServiceImpl(bigquery);
  JobReference jobRef = new JobReference()
      .setProjectId("projectId")
      .setJobId("jobId");
  Job job = jobService.pollJob(jobRef, Sleeper.DEFAULT, BackOff.ZERO_BACKOFF);

  assertEquals(testJob, job);
  verify(response, times(1)).getStatusCode();
  verify(response, times(1)).getContent();
  verify(response, times(1)).getContentType();
}
 

import com.google.api.services.bigquery.model.Job; //导入依赖的package包/类
/**
 * Tests that {@link BigQueryServicesImpl.JobServiceImpl#pollJob} returns UNKNOWN.
 */
@Test
public void testPollJobUnknown() throws IOException, InterruptedException {
  Job testJob = new Job();
  testJob.setStatus(new JobStatus());

  when(response.getContentType()).thenReturn(Json.MEDIA_TYPE);
  when(response.getStatusCode()).thenReturn(200);
  when(response.getContent()).thenReturn(toStream(testJob));

  BigQueryServicesImpl.JobServiceImpl jobService =
      new BigQueryServicesImpl.JobServiceImpl(bigquery);
  JobReference jobRef = new JobReference()
      .setProjectId("projectId")
      .setJobId("jobId");
  Job job = jobService.pollJob(jobRef, Sleeper.DEFAULT, BackOff.STOP_BACKOFF);

  assertEquals(null, job);
  verify(response, times(1)).getStatusCode();
  verify(response, times(1)).getContent();
  verify(response, times(1)).getContentType();
}
 

import com.google.api.services.bigquery.model.Job; //导入依赖的package包/类
@Test
public void testGetJobSucceeds() throws Exception {
  Job testJob = new Job();
  testJob.setStatus(new JobStatus());

  when(response.getContentType()).thenReturn(Json.MEDIA_TYPE);
  when(response.getStatusCode()).thenReturn(200);
  when(response.getContent()).thenReturn(toStream(testJob));

  BigQueryServicesImpl.JobServiceImpl jobService =
      new BigQueryServicesImpl.JobServiceImpl(bigquery);
  JobReference jobRef = new JobReference()
      .setProjectId("projectId")
      .setJobId("jobId");
  Job job = jobService.getJob(jobRef, Sleeper.DEFAULT, BackOff.ZERO_BACKOFF);

  assertEquals(testJob, job);
  verify(response, times(1)).getStatusCode();
  verify(response, times(1)).getContent();
  verify(response, times(1)).getContentType();
}
 

import com.google.api.services.bigquery.model.Job; //导入依赖的package包/类
@Test
public void testGetJobNotFound() throws Exception {
  when(response.getContentType()).thenReturn(Json.MEDIA_TYPE);
  when(response.getStatusCode()).thenReturn(404);

  BigQueryServicesImpl.JobServiceImpl jobService =
      new BigQueryServicesImpl.JobServiceImpl(bigquery);
  JobReference jobRef = new JobReference()
      .setProjectId("projectId")
      .setJobId("jobId");
  Job job = jobService.getJob(jobRef, Sleeper.DEFAULT, BackOff.ZERO_BACKOFF);

  assertEquals(null, job);
  verify(response, times(1)).getStatusCode();
  verify(response, times(1)).getContent();
  verify(response, times(1)).getContentType();
}
 

import com.google.api.services.bigquery.model.Job; //导入依赖的package包/类
@Override
public void startExtractJob(JobReference jobRef, JobConfigurationExtract extractConfig)
    throws InterruptedException, IOException {
  checkArgument(extractConfig.getDestinationFormat().equals("AVRO"),
      "Only extract to AVRO is supported");
  synchronized (allJobs) {
    verifyUniqueJobId(jobRef.getJobId());
    ++numExtractJobCalls;

    Job job = new Job();
    job.setJobReference(jobRef);
    job.setConfiguration(new JobConfiguration().setExtract(extractConfig));
    job.setKind(" bigquery#job");
    job.setStatus(new JobStatus().setState("PENDING"));
    allJobs.put(jobRef.getProjectId(), jobRef.getJobId(), new JobInfo(job));
  }
}
 

import com.google.api.services.bigquery.model.Job; //导入依赖的package包/类
/**
 * Starts an asynchronous query job to populate the specified destination table with the results
 * of the specified query, or if the table is a view, to update the view to reflect that query.
 * Returns a ListenableFuture that holds the same destination table object on success.
 */
public ListenableFuture<DestinationTable> query(
    String querySql,
    DestinationTable dest) {
  if (dest.type == TableType.VIEW) {
    // Use Futures.transform() rather than calling apply() directly so that any exceptions thrown
    // by calling updateTable will be propagated on the get() call, not from here.
    return transform(
        Futures.immediateFuture(dest.withQuery(querySql)), this::updateTable, directExecutor());
  } else {
    Job job = new Job()
        .setConfiguration(new JobConfiguration()
            .setQuery(new JobConfigurationQuery()
                .setQuery(querySql)
                .setDefaultDataset(getDataset())
                .setWriteDisposition(dest.getWriteDisposition().toString())
                .setDestinationTable(dest.getTableReference())));
    return transform(runJobToCompletion(job, dest), this::updateTable, directExecutor());
  }
}
 

import com.google.api.services.bigquery.model.Job; //导入依赖的package包/类
/**
 * Starts an asynchronous job to extract the specified source table and output it to the
 * given GCS filepath in the specified destination format, optionally printing headers.
 * Returns a ListenableFuture that holds the destination GCS URI on success.
 */
private ListenableFuture<String> extractTable(
    DestinationTable sourceTable,
    String destinationUri,
    DestinationFormat destinationFormat,
    boolean printHeader) {
  checkArgument(sourceTable.type == TableType.TABLE);
  Job job = new Job()
      .setConfiguration(new JobConfiguration()
          .setExtract(new JobConfigurationExtract()
              .setSourceTable(sourceTable.getTableReference())
              .setDestinationFormat(destinationFormat.toString())
              .setDestinationUris(ImmutableList.of(destinationUri))
              .setPrintHeader(printHeader)));
  return runJobToCompletion(job, destinationUri);
}
 

import com.google.api.services.bigquery.model.Job; //导入依赖的package包/类
/**
 * Checks completed job for errors.
 *
 * @throws BigqueryJobFailureException
 */
private static Job checkJob(Job job) {
  verify(job.getStatus() != null);
  JobStatus jobStatus = job.getStatus();
  if (jobStatus.getErrorResult() != null) {
    throw BigqueryJobFailureException.create(jobStatus);
  } else {
    logger.info(summarizeCompletedJob(job));
    if (jobStatus.getErrors() != null) {
      for (ErrorProto error : jobStatus.getErrors()) {
        logger.warning(String.format("%s: %s", error.getReason(), error.getMessage()));
      }
    }
    return job;
  }
}
 

import com.google.api.services.bigquery.model.Job; //导入依赖的package包/类
/** Create a table from a SQL query if it doesn't already exist. */
public TableReference ensureTable(TableReference table, String sqlQuery) {
  try {
    runJob(new Job()
        .setConfiguration(new JobConfiguration()
            .setQuery(new JobConfigurationQuery()
                .setQuery(sqlQuery)
                .setDefaultDataset(getDataset())
                .setDestinationTable(table))));
  } catch (BigqueryJobFailureException e) {
    if (e.getReason().equals("duplicate")) {
      // Table already exists.
    } else {
      throw e;
    }
  }
  return table;
}
 

import com.google.api.services.bigquery.model.Job; //导入依赖的package包/类
/**
 * Returns true if the provided job succeeded, false if it failed, and throws an exception if it
 * is still pending.
 */
private boolean checkJobOutcome() {
  Job job = null;
  String jobRefString =
      toJobReferenceString(new JobReference().setProjectId(projectId).setJobId(jobId));

  try {
    job = bigquery.jobs().get(projectId, jobId).execute();
  } catch (IOException e) {
    // We will throw a new exception because done==false, but first log this exception.
    logger.warning(e, e.getMessage());
  }
  // If job is not yet done, then throw an exception so that we'll return a failing HTTP status
  // code and the task will be retried.
  if (job == null || !job.getStatus().getState().equals("DONE")) {
    throw new NotModifiedException(jobRefString);
  }

  // Check if the job ended with an error.
  if (job.getStatus().getErrorResult() != null) {
    logger.severefmt("Bigquery job failed - %s - %s", jobRefString, job);
    return false;
  }
  logger.infofmt("Bigquery job succeeded - %s", jobRefString);
  return true;
}
 

import com.google.api.services.bigquery.model.Job; //导入依赖的package包/类
private Job makeLoadJob(JobReference jobRef, String sourceUri, String tableId) {
  TableReference tableReference = new TableReference()
      .setProjectId(jobRef.getProjectId())
      .setDatasetId(SNAPSHOTS_DATASET)
      .setTableId(tableId);
  return new Job()
      .setJobReference(jobRef)
      .setConfiguration(new JobConfiguration()
          .setLoad(new JobConfigurationLoad()
              .setWriteDisposition(WriteDisposition.WRITE_EMPTY.toString())
              .setSourceFormat(SourceFormat.DATASTORE_BACKUP.toString())
              .setSourceUris(ImmutableList.of(sourceUri))
              .setDestinationTable(tableReference)));
}
 

import com.google.api.services.bigquery.model.Job; //导入依赖的package包/类
@Before
public void before() throws Exception {
  when(bigqueryFactory.create("Project-Id", "snapshots")).thenReturn(bigquery);
  when(bigquery.jobs()).thenReturn(bigqueryJobs);
  when(bigqueryJobs.insert(eq("Project-Id"), any(Job.class))).thenReturn(bigqueryJobsInsert);
  when(bigquery.datasets()).thenReturn(bigqueryDatasets);
  when(bigqueryDatasets.insert(eq("Project-Id"), any(Dataset.class)))
      .thenReturn(bigqueryDatasetsInsert);
  action = new LoadSnapshotAction();
  action.bigqueryFactory = bigqueryFactory;
  action.bigqueryPollEnqueuer = bigqueryPollEnqueuer;
  action.clock = clock;
  action.projectId = "Project-Id";
  action.snapshotFile = "gs://bucket/snapshot.backup_info";
  action.snapshotId = "id12345";
  action.snapshotKinds = "one,two,three";
}