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

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

    // When:
    final CreateTableCommand cmd = createSourceFactory
        .createTableCommand(statement, ksqlConfig);

    // Then:
    assertThat(cmd.getFormats().getValueFeatures(),
        is(SerdeFeatures.of(SerdeFeature.UNWRAP_SINGLES)));
  }