static QueryId buildId(
      final Statement statement,
      final EngineContext engineContext,
      final QueryIdGenerator idGenerator,
      final OutputNode outputNode,
      final boolean createOrReplaceEnabled,
      final Optional<String> withQueryId) {
    if (withQueryId.isPresent()) {
      final String queryId = withQueryId.get().toUpperCase();
      validateWithQueryId(queryId);
      return new QueryId(queryId);
    }
    if (statement instanceof CreateTable && ((CreateTable) statement).isSource()) {
      // Use the CST name as part of the QueryID
      final String suffix = ((CreateTable) statement).getName().text().toUpperCase()
          + "_" + idGenerator.getNext().toUpperCase();
      return new QueryId(ReservedQueryIdsPrefixes.CST + suffix);
    }

    if (!outputNode.getSinkName().isPresent()) {
      final String prefix =
          "transient_" + outputNode.getSource().getLeftmostSourceNode().getAlias().text() + "_";
      return new QueryId(prefix + Math.abs(ThreadLocalRandom.current().nextLong()));
    }

    final KsqlStructuredDataOutputNode structured = (KsqlStructuredDataOutputNode) outputNode;
    if (!structured.createInto()) {
      return new QueryId(ReservedQueryIdsPrefixes.INSERT + idGenerator.getNext());
    }

    final SourceName sink = outputNode.getSinkName().get();
    final Set<QueryId> queriesForSink = engineContext.getQueryRegistry().getQueriesWithSink(sink);
    if (queriesForSink.size() > 1) {
      throw new KsqlException("REPLACE for sink " + sink + " is not supported because there are "
          + "multiple queries writing into it: " + queriesForSink);
    } else if (!queriesForSink.isEmpty()) {
      if (!createOrReplaceEnabled) {
        final String type = outputNode.getNodeOutputType().getKsqlType().toLowerCase();
        throw new UnsupportedOperationException(
            String.format(
                "Cannot add %s '%s': A %s with the same name already exists",
                type,
                sink.text(),
                type));
      }
      return Iterables.getOnlyElement(queriesForSink);
    }

    final String suffix = outputNode.getId().toString().toUpperCase()
        + "_" + idGenerator.getNext().toUpperCase();
    return new QueryId(
        outputNode.getNodeOutputType() == DataSourceType.KTABLE
            ? ReservedQueryIdsPrefixes.CTAS + suffix
            : ReservedQueryIdsPrefixes.CSAS + suffix
    );
  }
@Test
  public void shouldThrowIfMultipleQueriesExist() {
    // Given:
    when(plan.getSinkName()).thenReturn(Optional.of(SINK));
    when(plan.createInto()).thenReturn(true);
    when(queryRegistry.getQueriesWithSink(SINK))
        .thenReturn(ImmutableSet.of(new QueryId("CTAS_FOO_1"), new QueryId("INSERTQUERY_1")));

    // When:
    final KsqlException e = assertThrows(KsqlException.class, () ->
        QueryIdUtil.buildId(statement, engineContext, idGenerator, plan, false,
            Optional.empty()));

    // Then:
    assertThat(e.getMessage(), containsString("there are multiple queries writing"));
  }