@Override
  void validateKeyPresent(final SourceName sinkName, final Projection projection) {
    if (!PartitionByParamsFactory.isPartitionByNull(partitionBys)
        && !containsExpressions(projection, partitionBys)) {
      throwKeysNotIncludedError(sinkName, "partitioning expression", originalPartitionBys);
    }
  }
@Test
  public void shouldThrowIfProjectionMissingPartitionBy() {
    // When:
    final Exception e = assertThrows(
        KsqlException.class,
        () -> repartitionNode.validateKeyPresent(SOURCE_NAME, projection)
    );

    // Then:
    assertThat(e.getMessage(), containsString("The query used to build `S1` "
        + "must include the partitioning expression T.ID in its projection (eg, SELECT T.ID..."));
  }