BigQueryTasks { @VisibleForTesting static Task<TableId> lookupLatestDaily(F0<FloBigQueryClient> bigQuerySupplier, String project, String dataset, String tableName, Date start, final int lookBackDays) { Objects.requireNonNull(project, "project"); Objects.requireNonNull(dataset, "dataset"); Objects.requireNonNull(tableName, "tableName"); Objects.requireNonNull(start, "start"); return Task.named("bigquery.lookupLatestDaily", project, dataset, tableName, start, lookBackDays) .ofType(TableId.class) .operator(BigQueryLookupOperator.of(bigQuerySupplier)) .process(bq -> bq.lookupLatestDaily(project, dataset, tableName, start, lookBackDays)); } 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 lookupLatestDailyShouldBeSerializable() throws SerializationException { final ByteArrayOutputStream baos = new ByteArrayOutputStream(); final Task<TableId> task = BigQueryTasks.lookupLatestDaily("foo", "bar", "baz", Date.parse("2018-01-01"), 7); 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 lookupLatestDailyShouldBeRunnable() throws Exception { final Future<TableId> future = FloRunner.runTask(BigQueryTasks.lookupLatestDaily( "foo", "bar", "baz", Date.parse("2018-01-01"), 7)).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 lookupLatestDailyShouldThrowNotReadyForNonExistentOrTooOldTable() throws Exception { final Task<TableId> lookup = BigQueryTasks.lookupLatestDaily(() -> { FloBigQueryClient bq = mock(FloBigQueryClient.class); when(bq.tableExists(any())).thenReturn(false); return bq; }, "foo", "bar", "baz", Date.parse("2018-01-01"), 7); exception.expectCause(instanceOf(NotReady.class)); FloRunner.runTask(lookup) .future().get(30, TimeUnit.SECONDS); }
@Test public void lookupLatestDailyShouldReturnTableIdForExistingTableFromStartDate() throws Exception { final TableId expected = TableId.of("foo", "bar", "baz_20180101"); final Task<TableId> lookup = BigQueryTasks.lookupLatestDaily(() -> { FloBigQueryClient bq = mock(FloBigQueryClient.class); when(bq.tableExists(expected)).thenReturn(true); return bq; }, "foo", "bar", "baz", Date.parse("2018-01-01"), 7); final TableId tableId = FloRunner.runTask(lookup) .future().get(30, TimeUnit.SECONDS); assertThat(tableId, is(expected)); }
@Test public void lookupLatestDailyShouldReturnTableIdForExistingTableInLookBackRange() throws Exception { final TableId expected = TableId.of("foo", "bar", "baz_20180105"); final Task<TableId> lookup = BigQueryTasks.lookupLatestDaily(() -> { FloBigQueryClient bq = mock(FloBigQueryClient.class); when(bq.tableExists(expected)).thenReturn(true); return bq; }, "foo", "bar", "baz", Date.parse("2018-01-10"), 7); final TableId tableId = FloRunner.runTask(lookup) .future().get(30, TimeUnit.SECONDS); assertThat(tableId, is(expected)); }
@Test public void lookupLatestDailyShouldReturnTableIdForExistingTableFromLastLookBackDate() throws Exception { final TableId expected = TableId.of("foo", "bar", "baz_20180103"); final Task<TableId> lookup = BigQueryTasks.lookupLatestDaily(() -> { FloBigQueryClient bq = mock(FloBigQueryClient.class); when(bq.tableExists(expected)).thenReturn(true); return bq; }, "foo", "bar", "baz", Date.parse("2018-01-10"), 7); final TableId tableId = FloRunner.runTask(lookup) .future().get(30, TimeUnit.SECONDS); assertThat(tableId, is(expected)); }
@Test public void lookupLatestDailyShouldHaveNameAndId() { final TaskId id = BigQueryTasks.lookupLatestDaily("foo", "bar", "baz", Date.parse("2018-01-01"), 7).id(); assertThat(id.name(), is("bigquery.lookupLatestDaily")); assertThat(id.toString(), startsWith("bigquery.lookupLatestDaily(foo,bar,baz,2018-01-01,7)#")); }