@Override
  public SchemaKStream<?> buildStream(final PlanBuildContext buildContext) {
    final SchemaKStream<?> stream = getSource().buildStream(buildContext);

    final List<ColumnName> keyColumnNames = getSchema().key().stream()
        .map(Column::name)
        .collect(Collectors.toList());

    return stream.select(
        keyColumnNames,
        getSelectExpressions(),
        buildContext.buildNodeContext(getId().toString()),
        buildContext,
        getFormatInfo()
    );
  }
@Test
  public void shouldBuildSourceOnceWhenBeingBuilt() {
    // When:
    projectNode.buildStream(planBuildContext);

    // Then:
    verify(source, times(1)).buildStream(planBuildContext);
  }