@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 shouldCreateDropType() {
    // Given:
    final DropType dropType = new DropType(Optional.empty(), SOME_TYPE_NAME, false);

    // When:
    final DropTypeCommand cmd = (DropTypeCommand) commandFactories.create(
        "sqlExpression",
        dropType,
        SessionConfig.of(ksqlConfig, emptyMap())
    );

    // Then:
    assertThat(cmd, is(dropTypeCommand));
    verify(dropTypeFactory).create(dropType);
  }