public GenericRow createRow(final KeyValue<List<?>, GenericRow> row) {
    if (row.value() != null) {
      throw new IllegalArgumentException("Not a tombstone: " + row);
    }

    final List<?> key = row.key();
    if (key.size() < keyIndexes.size()) {
      throw new IllegalArgumentException("Not enough key columns. "
          + "expected at least" + keyIndexes.size() + ", got: " + key);
    }

    final GenericRow values = new GenericRow(numColumns);

    for (int columnIdx = 0; columnIdx < numColumns; columnIdx++) {
      final Integer keyIdx = keyIndexes.get(columnIdx);
      if (keyIdx == null) {
        values.append(null);
      } else {
        values.append(key.get(keyIdx));
      }
    }

    return values;
  }
@Test
  public void shouldHandleSomeKeyColumnsNotInProjection() {
    // Given:
    givenSchema(LogicalSchema.builder()
        .keyColumn(ColumnName.of("K1"), SqlTypes.INTEGER)
        .keyColumn(ColumnName.of("K2"), SqlTypes.INTEGER)
        .valueColumn(ColumnName.of("V0"), SqlTypes.INTEGER)
        .valueColumn(ColumnName.of("K2"), SqlTypes.INTEGER)
        .valueColumn(ColumnName.of("V1"), SqlTypes.INTEGER)
        .valueColumn(ColumnName.of("V2"), SqlTypes.INTEGER)
        .build());

    final KeyValue<List<?>, GenericRow> kv = KeyValue.keyValue(
        ImmutableList.of(10, 2),
        null
    );

    // When:
    final GenericRow result = factory.createRow(kv);

    // Then:
    assertThat(result, is(genericRow(null, 2, null, null)));
  }