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

    final Instant now = Instant.now();
    final TimeWindow window1 =
        new TimeWindow(now.plusMillis(10).toEpochMilli(), now.plusMillis(11).toEpochMilli());

    final SessionWindow window2 =
        new SessionWindow(now.toEpochMilli(), now.plusMillis(2).toEpochMilli());

    final TimeWindow window3 =
        new TimeWindow(now.toEpochMilli(), now.plusMillis(3).toEpochMilli());

    final ImmutableList<WindowedRow> rows = ImmutableList.of(
        WindowedRow.of(schema, new Windowed<>(aKey, window1), aValue, aRowtime),
        WindowedRow.of(schema, new Windowed<>(aKey, window2), aValue, aRowtime),
        WindowedRow.of(schema, new Windowed<>(aKey, window3), aValue, aRowtime)
    );

    when(innerWindowed.get(any(), anyInt(), any(), any(), any())).thenReturn(
        KsMaterializedQueryResult.rowIteratorWithPosition(rows.iterator(), position));

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

    // Then:
    assertThat(result.hasNext(), is(true));
    final List<WindowedRow> resultList = Lists.newArrayList(result);
    assertThat(resultList, hasSize(rows.size()));
    assertThat(resultList.get(0).windowedKey().window(), is(window1));
    assertThat(resultList.get(1).windowedKey().window(), is(window2));
    assertThat(resultList.get(2).windowedKey().window(), is(window3));
  }