@Override
  public DdlCommand create(
      final String sqlExpression,
      final DdlStatement ddlStatement,
      final SessionConfig config
  ) {
    return FACTORIES
        .getOrDefault(ddlStatement.getClass(), (statement, cf, ci) -> {
          throw new KsqlException(
              "Unable to find ddl command factory for statement:"
                  + statement.getClass()
                  + " valid statements:"
                  + FACTORIES.keySet()
          );
        })
        .handle(
            this,
            new CallInfo(sqlExpression, config),
            ddlStatement);
  }
@Test
  public void shouldCreateTableCommandWithSingleValueWrappingFromOverridesNotConfig() {
    // Given:
    givenCommandFactories();
    ksqlConfig = new KsqlConfig(ImmutableMap.of(
        KsqlConfig.KSQL_WRAP_SINGLE_VALUES, true
    ));

    final ImmutableMap<String, Object> overrides = ImmutableMap.of(
        KsqlConfig.KSQL_WRAP_SINGLE_VALUES, false
    );

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

    // When:
    final DdlCommand cmd = commandFactories
        .create(sqlExpression, statement, SessionConfig.of(ksqlConfig, overrides));

    // Then:
    assertThat(cmd, is(instanceOf(CreateTableCommand.class)));
    assertThat(((CreateTableCommand) cmd).getFormats().getValueFeatures().all(),
        contains(SerdeFeature.UNWRAP_SINGLES));
  }