@Override
  void validateKeyPresent(final SourceName sinkName, final Projection projection) {

    if (joinKey.isForeignKey()) {
      final DataSourceNode leftInputTable = getLeftmostSourceNode();
      final SourceName leftInputTableName = leftInputTable.getAlias();
      final List<Column> leftInputTableKeys = leftInputTable.getDataSource().getSchema().key();
      final List<Column> missingKeys =
          leftInputTableKeys.stream().filter(
              k -> !projection.containsExpression(
                    new QualifiedColumnReferenceExp(leftInputTableName, k.name()))
                  && !projection.containsExpression(new UnqualifiedColumnReferenceExp(
                      ColumnNames.generatedJoinColumnAlias(leftInputTableName, k.name())
                  ))
          ).collect(Collectors.toList());

      if (!missingKeys.isEmpty()) {
        throwMissingKeyColumnForFkJoinException(missingKeys, leftInputTableName);
      }
    } else {
      final boolean atLeastOneKey = joinKey.getAllViableKeys(schema).stream()
          .anyMatch(projection::containsExpression);

      if (!atLeastOneKey) {
        final boolean synthetic = joinKey.isSynthetic();
        final List<? extends Expression> viable = joinKey.getOriginalViableKeys(schema);

        throwKeysNotIncludedError(sinkName, "join expression", viable, false, synthetic);
      }
    }
  }
@Test
  public void shouldTestExpressionForBothOriginalAndRewrittenJoinKeys() {
    // Given:
    when(projection.containsExpression(any())).thenReturn(false, true);

    final JoinNode joinNode = new JoinNode(nodeId, LEFT, joinKey,      true, left,
        right, empty(),"KAFKA");

    when(joinKey.getAllViableKeys(any()))
        .thenReturn((List) ImmutableList.of(expression1, expression2));

    // When:
    joinNode.validateKeyPresent(SINK, projection);

    // Then:
    verify(projection).containsExpression(expression2);
    verify(projection).containsExpression(expression1);
  }