BigQueryOperator implements TaskOperator<BigQueryOperation.Provider<T>, BigQueryOperation<T>, T> { public static <T> BigQueryOperator<T> create() { return new BigQueryOperator<>(); } private  BigQueryOperator(); @Override T perform(BigQueryOperation<T> spec, Listener listener); static BigQueryOperator<T> create(); @Override Provider<T> provide(EvalContext evalContext);  }
@Test public void shouldRunExtractJobInTestMode() throws Exception { final TableId srcTable = TableId.of("foo", "bar", "baz"); final String destinationUri = "gs: final Task<String> task = Task.named("task") .ofType(String.class) .operator(BigQueryOperator.create()) .process(bq -> bq.job( JobInfo.of(ExtractJobConfiguration.of(srcTable, destinationUri))) .success(response -> destinationUri)); try (TestScope scope = FloTesting.scope()) { final String result = FloRunner.runTask(task).future() .get(30, SECONDS); assertThat(result, is(destinationUri)); } }
@Test public void shouldRunQueryTestMode() throws Exception { final Task<String> task = Task.named("task") .ofType(String.class) .operator(BigQueryOperator.create()) .process(bq -> bq.query("SELECT foo FROM input") .success(result -> "success!")); try (TestScope scope = FloTesting.scope()) { final String result = FloRunner.runTask(task).future() .get(30, SECONDS); assertThat(result, is("success!")); } }
@Test public void shouldRunQueryJobInTestMode() throws Exception { final TableId table = TableId.of("foo", "bar", "baz"); final Task<TableId> task = Task.named("task") .ofType(TableId.class) .output(BigQueryOutput.create(table)) .operator(BigQueryOperator.create()) .process((stagingTable, bq) -> bq.job( JobInfo.of(QueryJobConfiguration.newBuilder("SELECT foo FROM input") .setDestinationTable(stagingTable.tableId()) .build())) .success(response -> stagingTable.publish())); try (TestScope scope = FloTesting.scope()) { final TableId result = FloRunner.runTask(task).future() .get(30, SECONDS); assertThat(result, is(table)); assertThat(BigQueryMocking.mock().tablePublished(table), is(true)); assertThat(BigQueryMocking.mock().tableExists(table), is(true)); } }
@Test public void shouldRunCopyJobInTestMode() throws Exception { final TableId srcTable = TableId.of("foo", "bar", "src"); final TableId dstTable = TableId.of("foo", "bar", "dst"); final Task<TableId> task = Task.named("task") .ofType(TableId.class) .operator(BigQueryOperator.create()) .process(bq -> bq.job( JobInfo.of(CopyJobConfiguration.of(dstTable, srcTable))) .success(response -> dstTable)); try (TestScope scope = FloTesting.scope()) { final TableId result = FloRunner.runTask(task).future() .get(30, SECONDS); assertThat(result, is(dstTable)); } }
@Test public void shouldRunLoadJobInTestMode() throws Exception { final TableId dstTable = TableId.of("foo", "bar", "baz"); final String srcUri = "gs: final Task<TableId> task = Task.named("task") .ofType(TableId.class) .operator(BigQueryOperator.create()) .process(bq -> bq.job( JobInfo.of(LoadJobConfiguration.of(dstTable, srcUri))) .success(response -> dstTable)); try (TestScope scope = FloTesting.scope()) { final TableId result = FloRunner.runTask(task).future() .get(30, SECONDS); assertThat(result, is(dstTable)); } }