@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 shouldNotThrowIfProjectionContainsKeyColumns() {
    // Given:
    when(projection.containsExpression(any())).thenReturn(true);

    // When:
    node.validateKeyPresent(SOURCE_NAME, projection);

    // Then: did not throw.
  }