BigQueryTasks { @VisibleForTesting static Task<TableId> lookup(F0<FloBigQueryClient> bigQuerySupplier, TableId tableId) { return Task.named("bigquery.lookup", tableId.getProject(), tableId.getDataset(), tableId.getTable()) .ofType(TableId.class) .operator(BigQueryLookupOperator.of(bigQuerySupplier)) .process(bq -> bq.lookup(tableId)); } private  BigQueryTasks(); static Task<TableId> lookup(TableId tableId); static Task<TableId> lookup(String project, String dataset, String table); static Task<TableId> lookupLatestDaily(String project, String dataset, String tableName, Date start, final int lookBackDays); static String formatTableDate(Date date); static String formatTableDateHour(DateHour dateHour); static String legacyTableRef(TableId tableId); static String tableRef(TableId tableId);  }
@Test public void lookupShouldBeSerializable() throws SerializationException { final ByteArrayOutputStream baos = new ByteArrayOutputStream(); final Task<TableId> task = BigQueryTasks.lookup("foo", "bar", "baz"); Serialization.serialize(task, baos); final ByteArrayInputStream bais = new ByteArrayInputStream(baos.toByteArray()); final Task<TableId> deserializedTask = Serialization.deserialize(bais); assertThat(deserializedTask, is(notNullValue())); }
@Test public void lookupShouldBeRunnable() throws Exception { final Future<TableId> future = FloRunner.runTask(BigQueryTasks.lookup( "non-existent-project", "non-existent-dataset", "non-existent-table")).future(); try { future.get(); fail("Did not expect to find a non-existent table"); } catch (ExecutionException e) { final Throwable rootCause = Throwables.getRootCause(e); if (rootCause instanceof NotReady) { } else if (rootCause instanceof GoogleJsonResponseException) { } else if (rootCause instanceof IllegalArgumentException && rootCause.getMessage().startsWith("A project ID is required")) { } else { throw e; } } }
@Test public void lookupShouldThrowNotReadyForNonExistentTable() throws Exception { final Task<TableId> lookup = BigQueryTasks.lookup(() -> { FloBigQueryClient bq = mock(FloBigQueryClient.class); when(bq.tableExists(any())).thenReturn(false); return bq; }, TableId.of("foo", "bar", "baz")); exception.expectCause(instanceOf(NotReady.class)); FloRunner.runTask(lookup) .future().get(30, TimeUnit.SECONDS); }
@Test public void lookupShouldReturnTableIdForExistingTable() throws Exception { final TableId expected = TableId.of("foo", "bar", "baz"); final Task<TableId> lookup = BigQueryTasks.lookup(() -> { FloBigQueryClient bq = mock(FloBigQueryClient.class); when(bq.tableExists(expected)).thenReturn(true); return bq; }, expected); final TableId tableId = FloRunner.runTask(lookup) .future().get(30, TimeUnit.SECONDS); assertThat(tableId, is(expected)); }
@Test public void lookupShouldHaveNameAndId() { final TaskId id = BigQueryTasks.lookup("foo", "bar", "baz").id(); assertThat(id.name(), is("bigquery.lookup")); assertThat(id.toString(), startsWith("bigquery.lookup(foo,bar,baz)#")); }
@Test public void lookupOfTableIdShouldHaveNameAndId() { final TaskId id = BigQueryTasks.lookup(TableId.of("foo", "bar", "baz")).id(); assertThat(id.name(), is("bigquery.lookup")); assertThat(id.toString(), startsWith("bigquery.lookup(foo,bar,baz)#")); }