class MetaStoreImpl implements MutableMetaStore {

  @Override
  public void updateForPersistentQuery(
      final String queryId,
      final Set<SourceName> sourceNames,
      final Set<SourceName> sinkNames
  ) {
    synchronized (referentialIntegrityLock) {
      final String sourceAlreadyRegistered = streamSources(sourceNames)
          .filter(source -> source.referentialIntegrity.getSourceForQueries().contains(queryId))
          .map(source -> source.source.getName())
          .map(Object::toString)
          .collect(Collectors.joining(","));

      final String sinkAlreadyRegistered = streamSources(sinkNames)
          .filter(source -> source.referentialIntegrity.getSinkForQueries().contains(queryId))
          .map(source -> source.source.getName())
          .map(Object::toString)
          .collect(Collectors.joining(","));

      if (!sourceAlreadyRegistered.isEmpty() || !sinkAlreadyRegistered.isEmpty()) {
        throw new KsqlException("query already registered."
            + " queryId: " + queryId
            + ", registeredAgainstSource: " + sourceAlreadyRegistered
            + ", registeredAgainstSink: " + sinkAlreadyRegistered);
      }

      streamSources(sourceNames)
          .forEach(source -> source.referentialIntegrity.addSourceForQueries(queryId));
      streamSources(sinkNames)
          .forEach(source -> source.referentialIntegrity.addSinkForQueries(queryId));
    }
  }

  public  MetaStoreImpl(final FunctionRegistry functionRegistry);
  private  MetaStoreImpl(
      final Map<SourceName, SourceInfo> dataSources,
      final FunctionRegistry functionRegistry,
      final TypeRegistry typeRegistry
  );

}