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 shouldThrowIfTableIsMissingPrimaryKey() {
    // Given:
    ksqlConfig = new KsqlConfig(ImmutableMap.of(
        KsqlConfig.KSQL_WRAP_SINGLE_VALUES, false
    ));

    final TableElements noKey = TableElements.of(ELEMENT1);

    final CreateTable statement =
        new CreateTable(SOME_NAME, noKey,
            false, true, withProperties, false);

    // When:
    final Exception e = assertThrows(
        KsqlException.class,
        () -> createSourceFactory.createTableCommand(statement, ksqlConfig)
    );

    // Then:
    assertThat(e.getMessage(),
        containsString("Tables require a PRIMARY KEY. Please define the PRIMARY KEY."));
  }