public CreateTableCommand createTableCommand(
      final KsqlStructuredDataOutputNode outputNode,
      final Optional<RefinementInfo> emitStrategy
  ) {
    Optional<WindowInfo> windowInfo =
        outputNode.getKsqlTopic().getKeyFormat().getWindowInfo();

    if (windowInfo.isPresent() && emitStrategy.isPresent()) {
      final WindowInfo info = windowInfo.get();
      windowInfo = Optional.of(WindowInfo.of(
          info.getType(),
          info.getSize(),
          Optional.of(emitStrategy.get().getOutputRefinement())
      ));
    }

    return new CreateTableCommand(
        outputNode.getSinkName().get(),
        outputNode.getSchema(),
        outputNode.getTimestampColumn(),
        outputNode.getKsqlTopic().getKafkaTopicName(),
        Formats.from(outputNode.getKsqlTopic()),
        windowInfo,
        Optional.of(outputNode.getOrReplace()),
        Optional.of(false)
    );
  }
@Test
  public void shouldCreateCommandForCreateTable() {
    // Given:
    final CreateTable ddlStatement = new CreateTable(SOME_NAME,
        TableElements.of(
            tableElement("COL1", new Type(BIGINT), PRIMARY_KEY_CONSTRAINT),
            tableElement("COL2", new Type(SqlTypes.STRING))),
        false, true, withProperties, false);

    // When:
    final CreateTableCommand result = createSourceFactory
        .createTableCommand(ddlStatement, ksqlConfig);

    // Then:
    assertThat(result.getSourceName(), is(SOME_NAME));
    assertThat(result.getTopicName(), is(TOPIC_NAME));
    assertThat(result.getIsSource(), is(false));
  }