import com.google.api.services.bigquery.model.TableDataInsertAllResponse; //导入依赖的package包/类
/**
 * Tests that {@link DatasetServiceImpl#insertAll} retries quota rate limited attempts.
 */
@Test
public void testInsertRetry() throws Exception {
  TableReference ref =
      new TableReference().setProjectId("project").setDatasetId("dataset").setTableId("table");
  List<ValueInSingleWindow<TableRow>> rows = new ArrayList<>();
  rows.add(wrapTableRow(new TableRow()));

  // 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(new TableDataInsertAllResponse()));

  DatasetServiceImpl dataService =
      new DatasetServiceImpl(bigquery, PipelineOptionsFactory.create());
  dataService.insertAll(ref, rows, null,
      BackOffAdapter.toGcpBackOff(TEST_BACKOFF.backoff()), new MockSleeper(),
      InsertRetryPolicy.alwaysRetry(), null);
  verify(response, times(2)).getStatusCode();
  verify(response, times(2)).getContent();
  verify(response, times(2)).getContentType();
  expectedLogs.verifyInfo("BigQuery insertAll exceeded rate limit, retrying");
}
 

import com.google.api.services.bigquery.model.TableDataInsertAllResponse; //导入依赖的package包/类
Map<TableRow, List<TableDataInsertAllResponse.InsertErrors>> getInsertErrors() {
  Map<TableRow, List<TableDataInsertAllResponse.InsertErrors>> parsedInsertErrors =
      Maps.newHashMap();
  synchronized (tables) {
    for (Map.Entry<String, List<String>> entry : this.insertErrors.entrySet()) {
      TableRow tableRow = BigQueryHelpers.fromJsonString(entry.getKey(), TableRow.class);
      List<TableDataInsertAllResponse.InsertErrors> allErrors = Lists.newArrayList();
      for (String errorsString : entry.getValue()) {
        allErrors.add(BigQueryHelpers.fromJsonString(
            errorsString, TableDataInsertAllResponse.InsertErrors.class));
      }
      parsedInsertErrors.put(tableRow, allErrors);
    }
  }
  return parsedInsertErrors;
}
 

import com.google.api.services.bigquery.model.TableDataInsertAllResponse; //导入依赖的package包/类
@Before
public void before() throws Exception {
  createTld("tld");

  action = new VerifyEntityIntegrityAction();
  action.mrRunner = new MapreduceRunner(Optional.of(2), Optional.of(2));
  action.response = new FakeResponse();
  BatchComponent component = mock(BatchComponent.class);
  inject.setStaticField(VerifyEntityIntegrityAction.class, "component", component);
  integrity =
      new VerifyEntityIntegrityStreamer(
          "project-id",
          bigqueryFactory,
          new Retrier(new FakeSleeper(new FakeClock()), 1),
          Suppliers.ofInstance("rowid"),
          now);
  when(bigqueryFactory.create(anyString(), anyString(), anyString())).thenReturn(bigquery);
  when(component.verifyEntityIntegrityStreamerFactory()).thenReturn(streamerFactory);
  when(streamerFactory.create(any(DateTime.class))).thenReturn(integrity);
  when(bigquery.tabledata()).thenReturn(bigqueryTableData);
  rowsCaptor = ArgumentCaptor.forClass(TableDataInsertAllRequest.class);
  when(bigqueryTableData.insertAll(anyString(), anyString(), anyString(), rowsCaptor.capture()))
          .thenReturn(bigqueryInsertAll);
  when(bigqueryInsertAll.execute()).thenReturn(new TableDataInsertAllResponse());

}
 

import com.google.api.services.bigquery.model.TableDataInsertAllResponse; //导入依赖的package包/类
@Test
public void test() throws IOException {
	String projectId = "";
	String datasetId = "ontologies";
	String tableId = "";
	String timestamp = Long.toString((new Date()).getTime());
	
	Bigquery bigquery = null;
	TableRow row = new TableRow();
	row.set("column_name", 7.7);
	TableDataInsertAllRequest.Rows rows = new TableDataInsertAllRequest.Rows();
	rows.setInsertId(timestamp);
	
	rows.setJson(row);
	List  rowList =
	    new ArrayList();
	rowList.add(rows);
	TableDataInsertAllRequest content = 
	    new TableDataInsertAllRequest().setRows(rowList);
	TableDataInsertAllResponse response =
	    bigquery.tabledata().insertAll(
	        projectId, datasetId, tableId, content).execute();
}
 

import com.google.api.services.bigquery.model.TableDataInsertAllResponse; //导入依赖的package包/类
public static void main(String[] args) throws IOException{
  final Scanner scanner = new Scanner(System.in);
  System.out.println("Enter your project id: ");
  String projectId = scanner.nextLine();
  System.out.println("Enter your dataset id: ");
  String datasetId = scanner.nextLine();
  System.out.println("Enter your table id: ");
  String tableId = scanner.nextLine();
  scanner.close();
  
  System.out.println("Enter JSON to stream to BigQuery: \n"
      + "Press End-of-stream (CTRL-D) to stop");
  
  JsonReader fromCLI = new JsonReader(new InputStreamReader(System.in));
  
  Iterator<TableDataInsertAllResponse> responses = run(projectId,
      datasetId,
      tableId,
      fromCLI);
  
  while(responses.hasNext()){
    System.out.println(responses.next());
  }
  
  fromCLI.close();
}
 

import com.google.api.services.bigquery.model.TableDataInsertAllResponse; //导入依赖的package包/类
@Test
public void testStream() throws IOException{
  JsonReader json = new JsonReader(
      new FileReader(
          new File(RESOURCE_PATH.resolve("streamrows.json").toString())));
  Iterator<TableDataInsertAllResponse> response = StreamingSample.run(
      CONSTANTS.getProjectId(),
      CONSTANTS.getDatasetId(),
      CONSTANTS.getCurrentTableId(),
      json);
  
  while(response.hasNext()){
    assertTrue(!response.next().isEmpty());
  }
  
}
 

import com.google.api.services.bigquery.model.TableDataInsertAllResponse; //导入依赖的package包/类
public void run() {
    try {
        // Prepare target table
        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
        Date now = new Date();
        String date = sdf.format(now);
        String targetTable = filterId + "_results_" + date + "_v" + parent.TABLE_STRUCTURE_VERSION;
        targetTable = targetTable.replace('-', '_');
        parent.prepareTable(targetTable);

        // Execute
        TableDataInsertAllRequest ir = new TableDataInsertAllRequest().setRows(rows);
        TableDataInsertAllResponse response = parent.bigquery.tabledata().insertAll(parent.projectId, parent.datasetId, targetTable, ir).execute();
        List<TableDataInsertAllResponse.InsertErrors> errors = response.getInsertErrors();
        if (errors != null) {
            LOG.error(errors.size() + " error(s) while writing " + filterId + " to BigQuery");
        } else {
            // Log lines for debug
            LOG.info(rows.size() + " lines written for " + filterId);
        }
    } catch (Exception e) {
        LOG.error("Failed to write to BigQuery", e);
    }
}
 

import com.google.api.services.bigquery.model.TableDataInsertAllResponse; //导入依赖的package包/类
/**
 * Tests that {@link DatasetServiceImpl#insertAll} retries selected rows on failure.
 */
@Test
public void testInsertRetrySelectRows() throws Exception {
  TableReference ref =
      new TableReference().setProjectId("project").setDatasetId("dataset").setTableId("table");
  List<ValueInSingleWindow<TableRow>> rows = ImmutableList.of(
      wrapTableRow(new TableRow().set("row", "a")),
      wrapTableRow(new TableRow().set("row", "b")));
  List<String> insertIds = ImmutableList.of("a", "b");

  final TableDataInsertAllResponse bFailed = new TableDataInsertAllResponse()
      .setInsertErrors(ImmutableList.of(
          new InsertErrors().setIndex(1L).setErrors(ImmutableList.of(new ErrorProto()))));

  final TableDataInsertAllResponse allRowsSucceeded = new TableDataInsertAllResponse();

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

  DatasetServiceImpl dataService =
      new DatasetServiceImpl(bigquery, PipelineOptionsFactory.create());
  dataService.insertAll(ref, rows, insertIds,
      BackOffAdapter.toGcpBackOff(TEST_BACKOFF.backoff()), new MockSleeper(),
      InsertRetryPolicy.alwaysRetry(), null);
  verify(response, times(2)).getStatusCode();
  verify(response, times(2)).getContent();
  verify(response, times(2)).getContentType();
}
 

import com.google.api.services.bigquery.model.TableDataInsertAllResponse; //导入依赖的package包/类
/**
 * Tests that {@link DatasetServiceImpl#insertAll} does not retry non-rate-limited attempts.
 */
@Test
public void testInsertDoesNotRetry() throws Throwable {
  TableReference ref =
      new TableReference().setProjectId("project").setDatasetId("dataset").setTableId("table");
  List<ValueInSingleWindow<TableRow>> rows = new ArrayList<>();
  rows.add(wrapTableRow(new TableRow()));

  // First response is 403 not-rate-limited, second response has valid payload but should not
  // be invoked.
  when(response.getContentType()).thenReturn(Json.MEDIA_TYPE);
  when(response.getStatusCode()).thenReturn(403).thenReturn(200);
  when(response.getContent())
      .thenReturn(toStream(errorWithReasonAndStatus("actually forbidden", 403)))
      .thenReturn(toStream(new TableDataInsertAllResponse()));

  thrown.expect(GoogleJsonResponseException.class);
  thrown.expectMessage("actually forbidden");

  DatasetServiceImpl dataService =
      new DatasetServiceImpl(bigquery, PipelineOptionsFactory.create());

  try {
    dataService.insertAll(ref, rows, null,
        BackOffAdapter.toGcpBackOff(TEST_BACKOFF.backoff()), new MockSleeper(),
        InsertRetryPolicy.alwaysRetry(), null);
    fail();
  } catch (RuntimeException e) {
    verify(response, times(1)).getStatusCode();
    verify(response, times(1)).getContent();
    verify(response, times(1)).getContentType();
    throw e.getCause();
  }
}
 

import com.google.api.services.bigquery.model.TableDataInsertAllResponse; //导入依赖的package包/类
private TableDataInsertAllResponse.InsertErrors generateErrorAmongMany(
    int numErrors, String baseReason, String exceptionalReason) {
  // The retry policies are expected to search through the entire list of ErrorProtos to determine
  // whether to retry. Stick the exceptionalReason in a random position to exercise this.
  List<ErrorProto> errorProtos = Lists.newArrayListWithExpectedSize(numErrors);
  int exceptionalPosition = ThreadLocalRandom.current().nextInt(numErrors);
  for (int i = 0; i < numErrors; ++i) {
    ErrorProto error = new ErrorProto();
    error.setReason((i == exceptionalPosition) ? exceptionalReason : baseReason);
    errorProtos.add(error);
  }
  TableDataInsertAllResponse.InsertErrors errors = new TableDataInsertAllResponse.InsertErrors();
  errors.setErrors(errorProtos);
  return errors;
}
 

import com.google.api.services.bigquery.model.TableDataInsertAllResponse; //导入依赖的package包/类
private void onInsertAll(List<List<Long>> errorIndicesSequence) throws Exception {
  when(mockClient.tabledata())
      .thenReturn(mockTabledata);

  final List<TableDataInsertAllResponse> responses = new ArrayList<>();
  for (List<Long> errorIndices : errorIndicesSequence) {
    List<TableDataInsertAllResponse.InsertErrors> errors = new ArrayList<>();
    for (long i : errorIndices) {
      TableDataInsertAllResponse.InsertErrors error =
          new TableDataInsertAllResponse.InsertErrors();
      error.setIndex(i);
    }
    TableDataInsertAllResponse response = new TableDataInsertAllResponse();
    response.setInsertErrors(errors);
    responses.add(response);
  }

  doAnswer(
      new Answer<Bigquery.Tabledata.InsertAll>() {
        @Override
        public Bigquery.Tabledata.InsertAll answer(InvocationOnMock invocation) throws Throwable {
          Bigquery.Tabledata.InsertAll mockInsertAll = mock(Bigquery.Tabledata.InsertAll.class);
          when(mockInsertAll.execute())
              .thenReturn(responses.get(0),
                  responses.subList(1, responses.size()).toArray(
                      new TableDataInsertAllResponse[responses.size() - 1]));
          return mockInsertAll;
        }
      })
      .when(mockTabledata)
      .insertAll(anyString(), anyString(), anyString(), any(TableDataInsertAllRequest.class));
}
 

import com.google.api.services.bigquery.model.TableDataInsertAllResponse; //导入依赖的package包/类
/**
 * Cause a given {@link TableRow} object to fail when it's inserted. The errors link the list
 * will be returned on subsequent retries, and the insert will succeed when the errors run out.
 */
public void failOnInsert(
    Map<TableRow, List<TableDataInsertAllResponse.InsertErrors>> insertErrors) {
  synchronized (tables) {
    for (Map.Entry<TableRow, List<TableDataInsertAllResponse.InsertErrors>> entry
        : insertErrors.entrySet()) {
      List<String> errorStrings = Lists.newArrayList();
      for (TableDataInsertAllResponse.InsertErrors errors : entry.getValue()) {
        errorStrings.add(BigQueryHelpers.toJsonString(errors));
      }
      this.insertErrors.put(BigQueryHelpers.toJsonString(entry.getKey()), errorStrings);
    }
  }
}
 

import com.google.api.services.bigquery.model.TableDataInsertAllResponse; //导入依赖的package包/类
@Test
public void testFailuresNoRetryPolicy() throws Exception {
  TableRow row1 = new TableRow().set("name", "a").set("number", "1");
  TableRow row2 = new TableRow().set("name", "b").set("number", "2");
  TableRow row3 = new TableRow().set("name", "c").set("number", "3");

  TableDataInsertAllResponse.InsertErrors ephemeralError =
      new TableDataInsertAllResponse.InsertErrors().setErrors(
          ImmutableList.of(new ErrorProto().setReason("timeout")));

  fakeDatasetService.failOnInsert(
      ImmutableMap.<TableRow, List<TableDataInsertAllResponse.InsertErrors>>of(
          row1, ImmutableList.of(ephemeralError, ephemeralError),
          row2, ImmutableList.of(ephemeralError, ephemeralError)));

  p.apply(Create.of(row1, row2, row3))
      .apply(
          BigQueryIO.writeTableRows()
              .to("project-id:dataset-id.table-id")
              .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
              .withMethod(BigQueryIO.Write.Method.STREAMING_INSERTS)
              .withSchema(
                  new TableSchema()
                      .setFields(
                          ImmutableList.of(
                              new TableFieldSchema().setName("name").setType("STRING"),
                              new TableFieldSchema().setName("number").setType("INTEGER"))))
              .withTestServices(fakeBqServices)
              .withoutValidation());
  p.run();

  assertThat(
      fakeDatasetService.getAllRows("project-id", "dataset-id", "table-id"),
      containsInAnyOrder(row1, row2, row3));
}
 

import com.google.api.services.bigquery.model.TableDataInsertAllResponse; //导入依赖的package包/类
@Test
public void testRetryPolicy() throws Exception {
  TableRow row1 = new TableRow().set("name", "a").set("number", "1");
  TableRow row2 = new TableRow().set("name", "b").set("number", "2");
  TableRow row3 = new TableRow().set("name", "c").set("number", "3");

  TableDataInsertAllResponse.InsertErrors ephemeralError =
      new TableDataInsertAllResponse.InsertErrors().setErrors(
          ImmutableList.of(new ErrorProto().setReason("timeout")));
  TableDataInsertAllResponse.InsertErrors persistentError =
      new TableDataInsertAllResponse.InsertErrors().setErrors(
          ImmutableList.of(new ErrorProto().setReason("invalidQuery")));

  fakeDatasetService.failOnInsert(
      ImmutableMap.<TableRow, List<TableDataInsertAllResponse.InsertErrors>>of(
          row1, ImmutableList.of(ephemeralError, ephemeralError),
          row2, ImmutableList.of(ephemeralError, ephemeralError, persistentError)));

  PCollection<TableRow> failedRows =
      p.apply(Create.of(row1, row2, row3))
          .apply(BigQueryIO.writeTableRows().to("project-id:dataset-id.table-id")
              .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
              .withMethod(BigQueryIO.Write.Method.STREAMING_INSERTS)
              .withSchema(new TableSchema().setFields(
                  ImmutableList.of(
                      new TableFieldSchema().setName("name").setType("STRING"),
                      new TableFieldSchema().setName("number").setType("INTEGER"))))
              .withFailedInsertRetryPolicy(InsertRetryPolicy.retryTransientErrors())
              .withTestServices(fakeBqServices)
              .withoutValidation()).getFailedInserts();
  // row2 finally fails with a non-retryable error, so we expect to see it in the collection of
  // failed rows.
  PAssert.that(failedRows).containsInAnyOrder(row2);
  p.run();

  // Only row1 and row3 were successfully inserted.
  assertThat(
      fakeDatasetService.getAllRows("project-id", "dataset-id", "table-id"),
      containsInAnyOrder(row1, row3));
}
 

import com.google.api.services.bigquery.model.TableDataInsertAllResponse; //导入依赖的package包/类
public static TableDataInsertAllResponse streamRow(Bigquery bigquery,
    String projectId,
    String datasetId,
    String tableId,
    TableDataInsertAllRequest.Rows row) throws IOException{
  
  return bigquery.tabledata().insertAll(
      projectId, 
      datasetId, 
      tableId, 
      new TableDataInsertAllRequest().setRows(Collections.singletonList(row))).execute();
  
}
 

import com.google.api.services.bigquery.model.TableDataInsertAllResponse; //导入依赖的package包/类
@Before
public void setUp() throws Exception {
    configuration = new GoogleBigQueryConfiguration();

    bigquery = Mockito.mock(Bigquery.class);
    endpoint = Mockito.mock(GoogleBigQueryEndpoint.class);
    tabledata = Mockito.mock(Bigquery.Tabledata.class);
    mockInsertall = Mockito.mock(Bigquery.Tabledata.InsertAll.class);

    Mockito.when(bigquery.tabledata()).thenReturn(tabledata);
    Mockito.when(tabledata.insertAll(Mockito.anyString(), Mockito.anyString(), Mockito.anyString(), Mockito.any())).thenReturn(mockInsertall);

    TableDataInsertAllResponse mockResponse = new TableDataInsertAllResponse();
    Mockito.when(mockInsertall.execute()).thenReturn(mockResponse);
}
 

import com.google.api.services.bigquery.model.TableDataInsertAllResponse; //导入依赖的package包/类
public Context(TableDataInsertAllResponse.InsertErrors errors) {
  this.errors = errors;
}
 

import com.google.api.services.bigquery.model.TableDataInsertAllResponse; //导入依赖的package包/类
/**
 * Tests that {@link DatasetServiceImpl#insertAll} fails gracefully when persistent issues.
 */
@Test
public void testInsertFailsGracefully() throws Exception {
  TableReference ref =
      new TableReference().setProjectId("project").setDatasetId("dataset").setTableId("table");
  List<ValueInSingleWindow<TableRow>> rows = ImmutableList.of(
      wrapTableRow(new TableRow()), wrapTableRow(new TableRow()));

  final TableDataInsertAllResponse row1Failed = new TableDataInsertAllResponse()
      .setInsertErrors(ImmutableList.of(new InsertErrors().setIndex(1L)));

  final TableDataInsertAllResponse row0Failed = new TableDataInsertAllResponse()
      .setInsertErrors(ImmutableList.of(new InsertErrors().setIndex(0L)));

  when(response.getContentType()).thenReturn(Json.MEDIA_TYPE);
  // Always return 200.
  when(response.getStatusCode()).thenReturn(200);
  // Return row 1 failing, then we retry row 1 as row 0, and row 0 persistently fails.
  when(response.getContent())
      .thenReturn(toStream(row1Failed))
      .thenAnswer(new Answer<InputStream>() {
        @Override
        public InputStream answer(InvocationOnMock invocation) throws Throwable {
          return toStream(row0Failed);
        }
      });


  DatasetServiceImpl dataService =
      new DatasetServiceImpl(bigquery, PipelineOptionsFactory.create());

  // Expect it to fail.
  try {
    dataService.insertAll(ref, rows, null,
        BackOffAdapter.toGcpBackOff(TEST_BACKOFF.backoff()), new MockSleeper(),
        InsertRetryPolicy.alwaysRetry(), null);
    fail();
  } catch (IOException e) {
    assertThat(e, instanceOf(IOException.class));
    assertThat(e.getMessage(), containsString("Insert failed:"));
    assertThat(e.getMessage(), containsString("[{\"index\":0}]"));
  }

  // Verify the exact number of retries as well as log messages.
  verify(response, times(4)).getStatusCode();
  verify(response, times(4)).getContent();
  verify(response, times(4)).getContentType();
  expectedLogs.verifyInfo("Retrying 1 failed inserts to BigQuery");
}
 

import com.google.api.services.bigquery.model.TableDataInsertAllResponse; //导入依赖的package包/类
/**
 * Tests that {@link DatasetServiceImpl#insertAll} uses the supplied {@link InsertRetryPolicy},
 * and returns the list of rows not retried.
 */
@Test
public void testInsertRetryPolicy() throws InterruptedException, IOException {
  TableReference ref =
      new TableReference().setProjectId("project").setDatasetId("dataset").setTableId("table");
  List<ValueInSingleWindow<TableRow>> rows = ImmutableList.of(
      wrapTableRow(new TableRow()), wrapTableRow(new TableRow()));

  // First time row0 fails with a retryable error, and row1 fails with a persistent error.
  final TableDataInsertAllResponse firstFailure = new TableDataInsertAllResponse()
      .setInsertErrors(ImmutableList.of(
          new InsertErrors().setIndex(0L).setErrors(
              ImmutableList.of(new ErrorProto().setReason("timeout"))),
          new InsertErrors().setIndex(1L).setErrors(
          ImmutableList.of(new ErrorProto().setReason("invalid")))));

  // Second time there is only one row, which fails with a retryable error.
  final TableDataInsertAllResponse secondFialure = new TableDataInsertAllResponse()
      .setInsertErrors(ImmutableList.of(new InsertErrors().setIndex(0L).setErrors(
          ImmutableList.of(new ErrorProto().setReason("timeout")))));

  // On the final attempt, no failures are returned.
  final TableDataInsertAllResponse allRowsSucceeded = new TableDataInsertAllResponse();

  when(response.getContentType()).thenReturn(Json.MEDIA_TYPE);
  // Always return 200.
  when(response.getStatusCode()).thenReturn(200);
  when(response.getContentType()).thenReturn(Json.MEDIA_TYPE);
  when(response.getStatusCode()).thenReturn(200).thenReturn(200);

  // First fail
  when(response.getContent())
      .thenReturn(toStream(firstFailure))
      .thenReturn(toStream(secondFialure))
      .thenReturn(toStream(allRowsSucceeded));

  DatasetServiceImpl dataService =
      new DatasetServiceImpl(bigquery, PipelineOptionsFactory.create());

  List<ValueInSingleWindow<TableRow>> failedInserts = Lists.newArrayList();
  dataService.insertAll(ref, rows, null,
      BackOffAdapter.toGcpBackOff(TEST_BACKOFF.backoff()), new MockSleeper(),
      InsertRetryPolicy.retryTransientErrors(), failedInserts);
  assertEquals(1, failedInserts.size());
  expectedLogs.verifyInfo("Retrying 1 failed inserts to BigQuery");
}
 

import com.google.api.services.bigquery.model.TableDataInsertAllResponse; //导入依赖的package包/类
@Test
public void testNeverRetry() {
  assertFalse(InsertRetryPolicy.neverRetry().shouldRetry(
      new Context(new TableDataInsertAllResponse.InsertErrors())));
}