public static StatementExecutorResponse execute(
      final ConfiguredStatement<Explain> statement,
      final SessionProperties sessionProperties,
      final KsqlExecutionContext executionContext,
      final ServiceContext serviceContext
  ) {
    return StatementExecutorResponse.handled(Optional
        .of(ExplainExecutor.explain(
            serviceContext,
            statement,
            executionContext,
            sessionProperties)));
  }
@Test
  public void shouldExplainPersistentStatement() {
    // Given:
    engine.givenSource(DataSourceType.KSTREAM, "Y");
    final String statementText = "CREATE STREAM X AS SELECT * FROM Y;";
    final ConfiguredStatement<?> explain = engine.configure("EXPLAIN " + statementText);

    // When:
    final QueryDescriptionEntity query = (QueryDescriptionEntity) CustomExecutors.EXPLAIN.execute(
        explain,
        sessionProperties,
        engine.getEngine(),
        engine.getServiceContext()
    ).getEntity().orElseThrow(IllegalStateException::new);

    // Then:
    assertThat(query.getQueryDescription().getStatementText(), equalTo(statementText));
    assertThat(query.getQueryDescription().getSources(), containsInAnyOrder("Y"));
    assertThat("No side effects should happen", engine.getEngine().getPersistentQueries(), is(empty()));
    assertThat(query.getQueryDescription().getKsqlHostQueryStatus(), equalTo(Collections.emptyMap()));
  }