public CreateStreamCommand createStreamCommand(final KsqlStructuredDataOutputNode outputNode) {
    return new CreateStreamCommand(
        outputNode.getSinkName().get(),
        outputNode.getSchema(),
        outputNode.getTimestampColumn(),
        outputNode.getKsqlTopic().getKafkaTopicName(),
        Formats.from(outputNode.getKsqlTopic()),
        outputNode.getKsqlTopic().getKeyFormat().getWindowInfo(),
        Optional.of(outputNode.getOrReplace()),
        Optional.of(false)
    );
  }
@Test
  public void shouldBuildSchemaWithExplicitKeyFieldForStream() {
    // Given:
    final CreateStream statement = new CreateStream(
        SOME_NAME,
        TableElements.of(
            tableElement("k", new Type(SqlTypes.STRING), KEY_CONSTRAINT),
            ELEMENT1,
            ELEMENT2
        ),
        false,
        true,
        withProperties,
        false
    );

    // When:
    final CreateStreamCommand result = createSourceFactory.createStreamCommand(
        statement,
        ksqlConfig
    );

    // Then:
    assertThat(result.getSchema(), is(LogicalSchema.builder()
        .keyColumn(ColumnName.of("k"), SqlTypes.STRING)
        .valueColumn(ColumnName.of("bob"), SqlTypes.STRING)
        .valueColumn(ColumnName.of("hojjat"), BIGINT)
        .build()
    ));
  }