@Override
  public MaterializedTable nonWindowed() {
    return new KsqlMaterializedTable(inner.nonWindowed());
  }
@Test
  public void shouldPipeTransforms_fullTableScan() {
    // Given:
    final MaterializedTable table = materialization.nonWindowed();
    givenNoopProject();
    when(filter.apply(any(), any(), any())).thenReturn(Optional.of(transformed));

    // When:
    final Iterator<Row> result =
        table.get(partition);
    result.next();
    result.next();

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