@Override
  public Optional<KsqlConstants.PersistentQueryType> getPersistentQueryType() {
    if (!queryPlan.isPresent()) {
      return Optional.empty();
    }

    // CREATE_AS and CREATE_SOURCE commands contain a DDL command and a Query plan.
    if (ddlCommand.isPresent()) {
      if (ddlCommand.get() instanceof CreateTableCommand
          && ((CreateTableCommand) ddlCommand.get()).getIsSource()) {
        return Optional.of(KsqlConstants.PersistentQueryType.CREATE_SOURCE);
      } else {
        return Optional.of(KsqlConstants.PersistentQueryType.CREATE_AS);
      }
    } else {
      // INSERT INTO persistent queries are the only queries types that exist without a
      // DDL command linked to the plan.
      return Optional.of(KsqlConstants.PersistentQueryType.INSERT);
    }
  }
@Test
  public void shouldReturnCreateAsPersistentQueryTypeOnCreateStream() {
    // Given:
    final CreateStreamCommand ddlCommand = Mockito.mock(CreateStreamCommand.class);
    final KsqlPlanV1 plan = new KsqlPlanV1(
        "stmt",
        Optional.of(ddlCommand),
        Optional.of(queryPlan1));

    // When/Then:
    assertThat(plan.getPersistentQueryType(),
        is(Optional.of(KsqlConstants.PersistentQueryType.CREATE_AS)));
  }