@Override
  void validateKeyPresent(final SourceName sinkName, final Projection projection) {
    if (getSchema().key().isEmpty()) {
      // No key column.
      return;
    }

    final List<Column> keys = getSchema().key();

    for (final Column keyCol : keys) {
      final ColumnName keyName = keyCol.name();
      if (!projection.containsExpression(new QualifiedColumnReferenceExp(getAlias(), keyName))
          && !projection.containsExpression(new UnqualifiedColumnReferenceExp(keyName))
      ) {
        throwKeysNotIncludedError(
            sinkName,
            "key column",
            keys.stream()
                .map(Column::name)
                .map(UnqualifiedColumnReferenceExp::new)
                .collect(Collectors.toList())
        );
      }
    }
  }
@Test
  public void shouldThrowIfProjectionDoesNotContainKeyColumns() {
    // Given:
    when(projection.containsExpression(any())).thenReturn(false);

    // When:
    final Exception e = assertThrows(
        KsqlException.class,
        () -> node.validateKeyPresent(SOURCE_NAME, projection)
    );

    // Then:
    assertThat(e.getMessage(), containsString("The query used to build `datasource` must include "
        + "the key columns k0 and k1 in its projection (eg, SELECT k0, k1...)."));
  }