@Override
  public KsMaterializedQueryResult<Row> get(
      final GenericKey key,
      final int partition,
      final Optional<Position> position
  ) {
    try {
      final ReadOnlyKeyValueStore<GenericKey, ValueAndTimestamp<GenericRow>> store = stateStore
          .store(QueryableStoreTypes.timestampedKeyValueStore(), partition);

      final ValueAndTimestamp<GenericRow> row = store.get(key);
      if (row == null) {
        return KsMaterializedQueryResult.rowIterator(Collections.emptyIterator());
      } else {
        return KsMaterializedQueryResult.rowIterator(ImmutableList.of(Row.of(
            stateStore.schema(), key, row.value(), row.timestamp())).iterator());
      }
    } catch (final Exception e) {
      throw new MaterializationException("Failed to get value from materialized table", e);
    }
  }
@Test
  public void shouldReturnValuesFullTableScan() {
    // Given:
    when(tableStore.all()).thenReturn(keyValueIterator);
    when(keyValueIterator.hasNext()).thenReturn(true, true, false);
    when(keyValueIterator.next())
        .thenReturn(KEY_VALUE1)
        .thenReturn(KEY_VALUE2);

    // When:
    final Iterator<Row> rowIterator = table.get(PARTITION).rowIterator;

    // Then:
    assertThat(rowIterator.hasNext(), is(true));
    assertThat(rowIterator.next(), is(Row.of(SCHEMA, A_KEY, ROW1, TIME1)));
    assertThat(rowIterator.next(), is(Row.of(SCHEMA, A_KEY2, ROW2, TIME2)));
    assertThat(rowIterator.hasNext(), is(false));
  }