content: /*
   * Licensed to the Apache Software Foundation (ASF) under one
   * or more contributor license agreements.  See the NOTICE file
   * distributed with this work for additional information
   * regarding copyright ownership.  The ASF licenses this file
   * to you under the Apache License, Version 2.0 (the
   * "License"); you may not use this file except in compliance
   * with the License.  You may obtain a copy of the License at
   *
   *     http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.cassandra.schema;
  
  import java.io.IOException;
  import java.nio.ByteBuffer;
  import java.util.*;
  import java.util.function.Predicate;
  import java.util.stream.Stream;
  import java.util.stream.StreamSupport;
  
  import javax.annotation.Nullable;
  
  import com.google.common.collect.*;
  
  import org.apache.cassandra.cql3.FieldIdentifier;
  import org.apache.cassandra.cql3.CQL3Type;
  import org.apache.cassandra.db.marshal.AbstractType;
  import org.apache.cassandra.db.marshal.UserType;
  import org.apache.cassandra.exceptions.ConfigurationException;
  import org.apache.cassandra.io.util.DataInputPlus;
  import org.apache.cassandra.io.util.DataOutputPlus;
  import org.apache.cassandra.tcm.serialization.Version;
  
  import static java.lang.String.format;
  import static java.util.stream.Collectors.toList;
  
  import static com.google.common.collect.Iterables.any;
  import static com.google.common.collect.Iterables.transform;
  
  import static org.apache.cassandra.db.TypeSizes.sizeof;
  import static org.apache.cassandra.utils.ByteBufferUtil.bytes;
  
  /**
   * An immutable container for a keyspace's UDTs.
   */
  public final class Types implements Iterable<UserType>
  {
      public static final Serializer serializer = new Serializer();
  
      private static final Types NONE = new Types(ImmutableMap.of());
  
      private final Map<ByteBuffer, UserType> types;
  
      private Types(Builder builder)
      {
          types = builder.types.build();
      }
  
      /*
       * For use in RawBuilder::build only.
       */
      private Types(Map<ByteBuffer, UserType> types)
      {
          this.types = types;
      }
  
      public static Builder builder()
      {
          return new Builder();
      }
  
      public static RawBuilder rawBuilder(String keyspace)
      {
          return new RawBuilder(keyspace);
      }
  
      public static Types none()
      {
          return NONE;
      }
  
      public static Types of(UserType... types)
      {
          return builder().add(types).build();
      }
  
      public Iterator<UserType> iterator()
      {
          return types.values().iterator();
      }
  
      public Stream<UserType> stream()
      {
          return StreamSupport.stream(spliterator(), false);
      }
  
      /**
       * Returns a stream of user types sorted by dependencies
       * @return a stream of user types sorted by dependencies
       */
      public Stream<UserType> sortedStream()
      {
          Set<ByteBuffer> sorted = new LinkedHashSet<>();
          types.values().forEach(t -> addUserTypes(t, sorted));
          return sorted.stream().map(n -> types.get(n));
      }
  
      public Iterable<UserType> referencingUserType(ByteBuffer name)
      {
          return Iterables.filter(types.values(), t -> t.referencesUserType(name) && !t.name.equals(name));
      }
  
      public boolean isEmpty()
      {
          return types.isEmpty();
      }
  
      /**
       * Get the type with the specified name
       *
       * @param name a non-qualified type name
       * @return an empty {@link Optional} if the type name is not found; a non-empty optional of {@link UserType} otherwise
       */
      public Optional<UserType> get(ByteBuffer name)
      {
          return Optional.ofNullable(types.get(name));
      }
  
      /**
       * Get the type with the specified name
       *
       * @param name a non-qualified type name
       * @return null if the type name is not found; the found {@link UserType} otherwise
       */
      @Nullable
      public UserType getNullable(ByteBuffer name)
      {
          return types.get(name);
      }
  
      boolean containsType(ByteBuffer name)
      {
          return types.containsKey(name);
      }
  
      Types filter(Predicate<UserType> predicate)
      {
          Builder builder = builder();
          types.values().stream().filter(predicate).forEach(builder::add);
          return builder.build();
      }
  
      /**
       * Create a Types instance with the provided type added
       */
      public Types with(UserType type)
      {
          if (get(type.name).isPresent())
              throw new IllegalStateException(format("Type %s already exists", type.name));
  
          return builder().add(this).add(type).build();
      }
  
      /**
       * Creates a Types instance with the type with the provided name removed
       */
      public Types without(ByteBuffer name)
      {
          UserType type =
              get(name).orElseThrow(() -> new IllegalStateException(format("Type %s doesn't exists", name)));
  
          return without(type);
      }
  
      public Types without(UserType type)
      {
          return filter(t -> t != type);
      }
  
      public Types withUpdatedUserType(UserType udt)
      {
          return any(this, t -> t.referencesUserType(udt.name))
               ? builder().add(transform(this, t -> t.withUpdatedUserType(udt))).build()
               : this;
      }
  
      @Override
      public boolean equals(Object o)
      {
          if (this == o)
              return true;
  
          if (!(o instanceof Types))
              return false;
  
          Types other = (Types) o;
  
          if (types.size() != other.types.size())
              return false;
  
          Iterator<Map.Entry<ByteBuffer, UserType>> thisIter = this.types.entrySet().iterator();
          Iterator<Map.Entry<ByteBuffer, UserType>> otherIter = other.types.entrySet().iterator();
          while (thisIter.hasNext())
          {
              Map.Entry<ByteBuffer, UserType> thisNext = thisIter.next();
              Map.Entry<ByteBuffer, UserType> otherNext = otherIter.next();
              if (!thisNext.getKey().equals(otherNext.getKey()))
                  return false;
  
              if (!thisNext.getValue().equals(otherNext.getValue()))
                  return false;
          }
          return true;
      }
  
      @Override
      public int hashCode()
      {
          return types.hashCode();
      }
  
      @Override
      public String toString()
      {
          return types.values().toString();
      }
  
      /**
       * Sorts the types by dependencies.
       *
       * @param types the types to sort
       * @return the types sorted by dependencies and names
       */
      private static Set<ByteBuffer> sortByDependencies(Collection<UserType> types)
      {
          Set<ByteBuffer> sorted = new LinkedHashSet<>();
          types.stream().forEach(t -> addUserTypes(t, sorted));
          return sorted;
      }
  
      /**
       * Find all user types used by the specified type and add them to the set.
       *
       * @param type the type to check for user types.
       * @param types the set of UDT names to which to add new user types found in {@code type}. Note that the
       * insertion ordering is important and ensures that if a user type A uses another user type B, then B will appear
       * before A in iteration order.
       */
      private static void addUserTypes(AbstractType<?> type, Set<ByteBuffer> types)
      {
          // Reach into subtypes first, so that if the type is a UDT, it's dependencies are recreated first.
          type.subTypes().forEach(t -> addUserTypes(t, types));
  
          if (type.isUDT())
              types.add(((UserType) type).name);
      }
  
      public static final class Builder
      {
          final ImmutableSortedMap.Builder<ByteBuffer, UserType> types = ImmutableSortedMap.naturalOrder();
  
          private Builder()
          {
          }
  
          public Types build()
          {
              return new Types(this);
          }
  
          public Builder add(UserType type)
          {
              assert type.isMultiCell();
              types.put(type.name, type);
              return this;
          }
  
          public Builder add(UserType... types)
          {
              for (UserType type : types)
                  add(type);
              return this;
          }
  
          public Builder add(Iterable<UserType> types)
          {
              types.forEach(this::add);
              return this;
          }
      }
  
      public static final class RawBuilder
      {
          final String keyspace;
          final List<RawUDT> definitions;
  
          private RawBuilder(String keyspace)
          {
              this.keyspace = keyspace;
              this.definitions = new ArrayList<>();
          }
  
          /**
           * Build a Types instance from Raw definitions.
           *
           * Constructs a DAG of graph dependencies and resolves them 1 by 1 in topological order.
           */
          public Types build()
          {
              if (definitions.isEmpty())
                  return Types.none();
  
              /*
               * build a DAG of UDT dependencies
               */
              Map<RawUDT, Integer> vertices = Maps.newHashMapWithExpectedSize(definitions.size()); // map values are numbers of referenced types
              for (RawUDT udt : definitions)
                  vertices.put(udt, 0);
  
              Multimap<RawUDT, RawUDT> adjacencyList = HashMultimap.create();
              for (RawUDT udt1 : definitions)
                  for (RawUDT udt2 : definitions)
                      if (udt1 != udt2 && udt1.referencesUserType(udt2))
                          adjacencyList.put(udt2, udt1);
  
              /*
               * resolve dependencies in topological order, using Kahn's algorithm
               */
              adjacencyList.values().forEach(vertex -> vertices.put(vertex, vertices.get(vertex) + 1));
  
              Queue<RawUDT> resolvableTypes = new LinkedList<>(); // UDTs with 0 dependencies
              for (Map.Entry<RawUDT, Integer> entry : vertices.entrySet())
                  if (entry.getValue() == 0)
                      resolvableTypes.add(entry.getKey());
  
              Types types = new Types(new HashMap<>());
              while (!resolvableTypes.isEmpty())
              {
                  RawUDT vertex = resolvableTypes.remove();
  
                  for (RawUDT dependentType : adjacencyList.get(vertex))
                      if (vertices.replace(dependentType, vertices.get(dependentType) - 1) == 1)
                          resolvableTypes.add(dependentType);
  
                  UserType udt = vertex.prepare(keyspace, types);
                  types.types.put(udt.name, udt);
              }
  
              if (types.types.size() != definitions.size())
                  throw new ConfigurationException(format("Cannot resolve UDTs for keyspace %s: some types are missing", keyspace));
  
              /*
               * return an immutable copy
               */
              return Types.builder().add(types).build();
          }
  
          public void add(String name, List<String> fieldNames, List<String> fieldTypes)
          {
              List<CQL3Type.Raw> rawFieldTypes =
                  fieldTypes.stream()
                            .map(CQLTypeParser::parseRaw)
                            .collect(toList());
  
              definitions.add(new RawUDT(name, fieldNames, rawFieldTypes));
          }
  
          private static final class RawUDT
          {
              final String name;
              final List<String> fieldNames;
              final List<CQL3Type.Raw> fieldTypes;
  
              RawUDT(String name, List<String> fieldNames, List<CQL3Type.Raw> fieldTypes)
              {
                  this.name = name;
                  this.fieldNames = fieldNames;
                  this.fieldTypes = fieldTypes;
              }
  
              boolean referencesUserType(RawUDT other)
              {
                  return fieldTypes.stream().anyMatch(t -> t.referencesUserType(other.name));
              }
  
              UserType prepare(String keyspace, Types types)
              {
                  List<FieldIdentifier> preparedFieldNames =
                      fieldNames.stream()
                                .map(FieldIdentifier::forInternalString)
                                .collect(toList());
  
                  List<AbstractType<?>> preparedFieldTypes =
                      fieldTypes.stream()
                                .map(t -> t.prepareInternal(keyspace, types).getType())
                                .collect(toList());
  
                  return new UserType(keyspace, bytes(name), preparedFieldNames, preparedFieldTypes, true);
              }
  
              @Override
              public int hashCode()
              {
                  return name.hashCode();
              }
  
              @Override
              public boolean equals(Object other)
              {
                  return name.equals(((RawUDT) other).name);
              }
          }
      }
  
      static TypesDiff diff(Types before, Types after)
      {
          return TypesDiff.diff(before, after);
      }
  
      static final class TypesDiff extends Diff<Types, UserType>
      {
          private static final TypesDiff NONE = new TypesDiff(Types.none(), Types.none(), ImmutableList.of());
  
          private TypesDiff(Types created, Types dropped, ImmutableCollection<Altered<UserType>> altered)
          {
              super(created, dropped, altered);
          }
  
          private static TypesDiff diff(Types before, Types after)
          {
              if (before == after)
                  return NONE;
  
              Types created = after.filter(t -> !before.containsType(t.name));
              Types dropped = before.filter(t -> !after.containsType(t.name));
  
              ImmutableList.Builder<Altered<UserType>> altered = ImmutableList.builder();
              before.forEach(typeBefore ->
              {
                  UserType typeAfter = after.getNullable(typeBefore.name);
                  if (null != typeAfter)
                      typeBefore.compare(typeAfter).ifPresent(kind -> altered.add(new Altered<>(typeBefore, typeAfter, kind)));
              });
  
              return new TypesDiff(created, dropped, altered.build());
          }
      }
  
      // Not quite a MetadataSerializer as it needs the keyspace name during deserialization.
      public static class Serializer
      {
          public void serialize(Types t, DataOutputPlus out, Version version) throws IOException
          {
              out.writeInt(t.types.size());
              for (UserType type : t.types.values())
              {
                  out.writeUTF(type.getNameAsString());
                  List<String> fieldNames = type.fieldNames().stream().map(FieldIdentifier::toString).collect(toList());
                  List<String> fieldTypes = type.fieldTypes().stream().map(AbstractType::asCQL3Type).map(CQL3Type::toString).collect(toList());
                  out.writeInt(fieldNames.size());
                  for (String s : fieldNames)
                      out.writeUTF(s);
                  out.writeInt(fieldTypes.size());
                  for (String s : fieldTypes)
                      out.writeUTF(s);
              }
          }
  
          public Types deserialize(String keyspace, DataInputPlus in, Version version) throws IOException
          {
              int count = in.readInt();
              Types.RawBuilder builder = Types.rawBuilder(keyspace);
              for (int i = 0; i < count; i++)
              {
                  String name = in.readUTF();
                  int fieldNamesSize = in.readInt();
                  List<String> fieldNames = new ArrayList<>(fieldNamesSize);
                  for (int x = 0; x < fieldNamesSize; x++)
                      fieldNames.add(in.readUTF());
                  int fieldTypeSize = in.readInt();
                  List<String> fieldTypes = new ArrayList<>(fieldTypeSize);
                  for (int x = 0; x < fieldTypeSize; x++)
                      fieldTypes.add(in.readUTF());
                  builder.add(name, fieldNames, fieldTypes);
              }
              return builder.build();
          }
  
          public long serializedSize(Types t, Version version)
          {
              long size = sizeof(t.types.size());
              for (UserType type : t.types.values())
              {
                  size += sizeof(type.getNameAsString());
                  List<String> fieldNames = type.fieldNames().stream().map(FieldIdentifier::toString).collect(toList());
                  List<String> fieldTypes = type.fieldTypes().stream().map(AbstractType::asCQL3Type).map(CQL3Type::toString).collect(toList());
                  size += sizeof(fieldNames.size());
                  for (String s : fieldNames)
                      size += sizeof(s);
                  size += sizeof(fieldTypes.size());
                  for (String s : fieldTypes)
                      size += sizeof(s);
              }
              return size;
          }
      }
  }
  