@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 shouldResolveNestedAliasedSelectStarByCallingOnlyCorrectParentWithMultiJoins() {
    // Given:
    final JoinNode inner =
        new JoinNode(new PlanNodeId("foo"), LEFT, joinKey,          true, right,
            right2, empty(), "KAFKA");

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

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

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

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

    verify(left, never()).resolveSelectStar(any());
    verify(right2, never()).resolveSelectStar(any());
    verify(right).resolveSelectStar(Optional.of(RIGHT_ALIAS));
  }