@SuppressWarnings("UnstableApiUsage")
  @Override
  public Stream<ColumnName> resolveSelectStar(
      final Optional<SourceName> sourceName
  ) {
    final Stream<ColumnName> names = Stream.of(left, right)
        .flatMap(JoinNode::getPreJoinProjectDataSources)
        .filter(s -> !sourceName.isPresent() || sourceName.equals(s.getSourceName()))
        .flatMap(s -> s.resolveSelectStar(sourceName));

    if (sourceName.isPresent() || !joinKey.isSynthetic() || !finalJoin) {
      return names;
    }

    // if we use a synthetic key, we know there's only a single key element
    final Column syntheticKey = getOnlyElement(getSchema().key());

    return Streams.concat(Stream.of(syntheticKey.name()), names);
  }
@Test
  public void shouldIncludeSyntheticJoinKeyInResolvedSelectStart() {
    // Given:
    when(joinKey.isSynthetic()).thenReturn(true);

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

    when(left.resolveSelectStar(any())).thenReturn(Stream.of(ColumnName.of("l")));
    when(right.resolveSelectStar(any())).thenReturn(Stream.of(ColumnName.of("r")));

    // When:
    final Stream<ColumnName> result = joinNode.resolveSelectStar(empty());

    // Then:
    final List<ColumnName> columns = result.collect(Collectors.toList());
    assertThat(columns, contains(SYNTH_KEY, ColumnName.of("l"), ColumnName.of("r")));
  }