public static ParsedCommand parse(
          // CHECKSTYLE_RULES.ON: CyclomaticComplexity
          final String sql, final Map<String, String> variables) {

    validateSupportedStatementType(sql);
    final String substituted;
    try {
      substituted = VariableSubstitutor.substitute(KSQL_PARSER.parse(sql).get(0), variables);
    } catch (ParseFailedException e) {
      throw new MigrationException(String.format(
              "Failed to parse the statement. Statement: %s. Reason: %s",
              sql, e.getMessage()));
    }
    final SqlBaseParser.SingleStatementContext statementContext = KSQL_PARSER.parse(substituted)
            .get(0).getStatement();
    final boolean isStatement = StatementType.get(statementContext.statement().getClass())
            == StatementType.STATEMENT;
    return new ParsedCommand(substituted,
            isStatement ? Optional.empty() : Optional.of(new AstBuilder(TypeRegistry.EMPTY)
            .buildStatement(statementContext)));
  }
@Test
  public void shouldParseCreateStatement() {
    // When:
    List<CommandParser.ParsedCommand> commands = parse("CREATE STREAM FOO (A STRING) WITH (KAFKA_TOPIC='FOO', PARTITIONS=1, VALUE_FORMAT='DELIMITED');");

    // Then:
    assertThat(commands.size(), is(1));
    assertThat(commands.get(0).getStatement().isPresent(), is (false));
    assertThat(commands.get(0).getCommand(), is("CREATE STREAM FOO (A STRING) WITH (KAFKA_TOPIC='FOO', PARTITIONS=1, VALUE_FORMAT='DELIMITED');"));
  }