@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 shouldCreateCommandForCreateSourceTable() {
    // Given:
    final CreateTable statement = new CreateTable(SOME_NAME,
        TableElements.of(
            tableElement("COL1", new Type(SqlTypes.BIGINT)),
            tableElement("COL2", new Type(SqlTypes.STRING))),
        false, true, withProperties, true);

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

    // Then:
    assertThat(result, is(createTableCommand));
    verify(createSourceFactory).createTableCommand(statement, ksqlConfig);
  }