public static GenericRow getIntermediateRow(final TableRow row) {
    final GenericKey key = row.key();
    final GenericRow value = row.value();

    final List<?> keyFields = key.values();

    value.ensureAdditionalCapacity(
        1 // ROWTIME
            + keyFields.size() //all the keys
            + row.window().map(w -> 2).orElse(0) //windows
    );

    value.append(row.rowTime());
    value.appendAll(keyFields);

    row.window().ifPresent(window -> {
      value.append(window.start().toEpochMilli());
      value.append(window.end().toEpochMilli());
    });

    return value;
  }
@Test
  public void shouldReturnIntermediateRowNonWindowed() {
    // Given:
    final GenericRow intermediateRow1 = aValue.append(aRowtime).append(aKey);
    final GenericRow intermediateRow2 = aValue2.append(aRowtime).append(aKey2);

    // When:
    final GenericRow genericRow1 = KsqlMaterialization.getIntermediateRow(row);
    final GenericRow genericRow2 = KsqlMaterialization.getIntermediateRow(row2);

    // Then:
    assertThat(genericRow1, is(intermediateRow1));
    assertThat(genericRow2, is(intermediateRow2));
  }