public static QueryDescription forQueryMetadata(
      final QueryMetadata queryMetadata,
      final Map<KsqlHostInfoEntity, KsqlQueryStatus> ksqlHostQueryStatus
  ) {
    if (queryMetadata instanceof PersistentQueryMetadata) {
      final PersistentQueryMetadata persistentQuery = (PersistentQueryMetadata) queryMetadata;
      return create(
          persistentQuery,
          persistentQuery.getResultTopic().map(t -> t.getKeyFormat().getWindowType())
              .orElse(Optional.empty()),
          persistentQuery.getSinkName(),
          ksqlHostQueryStatus
      );
    }

    return create(
        queryMetadata,
        Optional.empty(),
        Optional.empty(),
        ksqlHostQueryStatus
    );
  }
@Test
  public void shouldHandleRowKeyInValueSchemaForTransientQuery() {
    // Given:
    final LogicalSchema schema = LogicalSchema.builder()
        .valueColumn(ColumnName.of("field1"), SqlTypes.INTEGER)
        .valueColumn(ColumnName.of("ROWKEY"), SqlTypes.STRING)
        .valueColumn(ColumnName.of("field2"), SqlTypes.STRING)
        .build();

    transientQuery = new TransientQueryMetadata(
        SQL_TEXT,
        schema,
        SOURCE_NAMES,
        "execution plan",
        queryQueue,
        QUERY_ID,
        "app id",
        topology,
        kafkaStreamsBuilder,
        STREAMS_PROPS,
        PROP_OVERRIDES,
        closeTimeout,
        10,
        ResultType.STREAM,
        0L,
        0L,
        listener,
        processingLoggerFactory
    );
    transientQuery.initialize();

    // When:
    transientQueryDescription = QueryDescriptionFactory.forQueryMetadata(transientQuery, Collections.emptyMap());

    // Then:
    assertThat(transientQueryDescription.getFields(), contains(
        new FieldInfo("field1", new SchemaInfo(SqlBaseType.INTEGER, null, null), Optional.empty()),
        new FieldInfo("ROWKEY", new SchemaInfo(SqlBaseType.STRING, null, null), Optional.empty()),
        new FieldInfo("field2", new SchemaInfo(SqlBaseType.STRING, null, null), Optional.empty())));
  }