class MetaStoreImpl implements MutableMetaStore {

  @Override
  public void putSource(final DataSource dataSource, final boolean allowReplace) {
    final SourceInfo existing = dataSources.get(dataSource.getName());
    if (existing != null && !allowReplace) {
      final SourceName name = dataSource.getName();
      final String newType = dataSource.getDataSourceType().getKsqlType().toLowerCase();
      final String existingType = existing.source.getDataSourceType().getKsqlType().toLowerCase();

      throw new KsqlException(String.format(
          "Cannot add %s '%s': A %s with the same name already exists",
          newType, name.text(), existingType));
    } else if (existing != null) {
      existing.source.canUpgradeTo(dataSource).ifPresent(msg -> {
        throw new KsqlException("Cannot REPLACE data source: " + msg);
      });
    }

    dataSources.put(dataSource.getName(), new SourceInfo(dataSource));
  }

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

}