class QueryIdUtil {

  static QueryId buildId(
      final MetaStore metaStore,
      final QueryIdGenerator idGenerator,
      final OutputNode outputNode,
      final boolean createOrReplaceEnabled) {
    if (!outputNode.getSinkName().isPresent()) {
      return new QueryId(String.valueOf(Math.abs(ThreadLocalRandom.current().nextLong())));
    }

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

    final SourceName sink = outputNode.getSinkName().get();
    final Set<String> queriesForSink = metaStore.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 new QueryId(Iterables.getOnlyElement(queriesForSink));
    }

    final String suffix = outputNode.getId().toString().toUpperCase()
        + "_" + idGenerator.getNext().toUpperCase();
    return new QueryId(
        outputNode.getNodeOutputType() == DataSourceType.KTABLE
            ? "CTAS_" + suffix
            : "CSAS_" + suffix
    );
  }

  private  QueryIdUtil();

}