### Question:
AbstractJdbcDataSetProvider implements DataSetProvider<T> { @Nonnull @VisibleForTesting T filterByDateTime(@Nonnull final T dataSet, @Nonnull final String fieldName, @Nullable final Long value, @Nullable final Long overlap) { long startTime = 0; long endTime = DateTimeUtils.currentTimeMillis(); if (value != null) { if (value < endTime) { startTime = value; } else { log.warn("Value for high water mark is the future: {}", value); startTime = endTime; } } if (overlap != null) { startTime = Math.max(startTime - overlap, 0); endTime -= overlap; } final Column dateColumn = new Column(fieldName); final Column startFilter = (startTime > 0) ? dateColumn.gt(functions.lit(new Timestamp(startTime))) : null; final Column endFilter = dateColumn.lt(functions.lit(new Timestamp(endTime))); return filter(dataSet, (startFilter != null) ? startFilter.and(endFilter) : endFilter); }  @Override boolean supportsFormat(@Nonnull final String source); @Nonnull @Override final T read(@Nonnull final KyloCatalogClient<T> client, @Nonnull final DataSetOptions options); @Override final void write(@Nonnull final KyloCatalogClient<T> client, @Nonnull final DataSetOptions options, @Nonnull final T dataSet);  }### Answer:
@Test public void filterByDateTime() { DateTimeUtils.setCurrentMillisFixed(1524960000000L); final DataFrame dataSet = Mockito.mock(DataFrame.class); final DataFrame filterDataSet = Mockito.mock(DataFrame.class); Mockito.when(dataSet.filter(Mockito.any(Column.class))).thenReturn(filterDataSet); final MockJdbcDataSetProvider provider = new MockJdbcDataSetProvider(); final DataFrame newDataSet = provider.filterByDateTime(dataSet, "mockField", null, null); Assert.assertEquals(filterDataSet, newDataSet); final ArgumentCaptor<Column> conditionCaptor = ArgumentCaptor.forClass(Column.class); Mockito.verify(dataSet).filter(conditionCaptor.capture()); Assert.assertEquals("(mockField < 1524960000000000)", conditionCaptor.getValue().toString()); }