@Override
  public MaterializedWindowedTable windowed() {
    return new KsqlMaterializedWindowedTable(inner.windowed());
  }
@Test
  public void shouldPipeTransformsWindowed_fullTableScan() {
    // Given:
    final MaterializedWindowedTable table = materialization.windowed();
    givenNoopProject();
    when(filter.apply(any(), any(), any())).thenReturn(Optional.of(transformed));

    // When:
    final Iterator<WindowedRow> result =
        table.get(partition, windowStartBounds, windowEndBounds);
    result.next();
    result.next();

    // Then:
    verify(project).apply(
        new Windowed<>(aKey, streamWindow),
            transformed,
        new PullProcessingContext(aRowtime)
    );
    verify(project).apply(
        new Windowed<>(aKey2, streamWindow),
            transformed,
        new PullProcessingContext(aRowtime)
    );
  }