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.db.marshal;
  
  import java.nio.ByteBuffer;
  import java.util.ArrayList;
  import java.util.Collections;
  import java.util.Iterator;
  import java.util.List;
  import java.util.Set;
  import java.util.SortedSet;
  import java.util.TreeSet;
  import java.util.concurrent.ConcurrentHashMap;
  import java.util.function.Consumer;
  
  import org.apache.cassandra.cql3.terms.MultiElements;
  import org.apache.cassandra.cql3.terms.Term;
  import org.apache.cassandra.db.rows.Cell;
  import org.apache.cassandra.db.rows.CellPath;
  import org.apache.cassandra.db.rows.ComplexColumnData;
  import org.apache.cassandra.exceptions.ConfigurationException;
  import org.apache.cassandra.exceptions.SyntaxException;
  import org.apache.cassandra.serializers.MarshalException;
  import org.apache.cassandra.serializers.SetSerializer;
  import org.apache.cassandra.transport.ProtocolVersion;
  import org.apache.cassandra.utils.JsonUtils;
  import org.apache.cassandra.utils.bytecomparable.ByteComparable;
  import org.apache.cassandra.utils.bytecomparable.ByteSource;
  
  public class SetType<T> extends CollectionType<Set<T>>
  {
      // interning instances
      private static final ConcurrentHashMap<AbstractType<?>, SetType> instances = new ConcurrentHashMap<>();
      private static final ConcurrentHashMap<AbstractType<?>, SetType> frozenInstances = new ConcurrentHashMap<>();
  
      private final AbstractType<T> elements;
      private final SetSerializer<T> serializer;
      private final boolean isMultiCell;
  
      public static SetType<?> getInstance(TypeParser parser) throws ConfigurationException, SyntaxException
      {
          List<AbstractType<?>> l = parser.getTypeParameters();
          if (l.size() != 1)
              throw new ConfigurationException("SetType takes exactly 1 type parameter");
  
          return getInstance(l.get(0).freeze(), true);
      }
  
      public static <T> SetType<T> getInstance(AbstractType<T> elements, boolean isMultiCell)
      {
          ConcurrentHashMap<AbstractType<?>, SetType> internMap = isMultiCell ? instances : frozenInstances;
          SetType<T> t = internMap.get(elements);
          return null == t
               ? internMap.computeIfAbsent(elements, k -> new SetType<>(k, isMultiCell))
               : t;
      }
  
      public SetType(AbstractType<T> elements, boolean isMultiCell)
      {
          super(ComparisonType.CUSTOM, Kind.SET);
          this.elements = elements;
          this.serializer = SetSerializer.getInstance(elements.getSerializer(), elements.comparatorSet);
          this.isMultiCell = isMultiCell;
      }
  
      @Override
      public <V> boolean referencesUserType(V name, ValueAccessor<V> accessor)
      {
          return elements.referencesUserType(name, accessor);
      }
  
      @Override
      public SetType<?> withUpdatedUserType(UserType udt)
      {
          if (!referencesUserType(udt.name))
              return this;
  
          (isMultiCell ? instances : frozenInstances).remove(elements);
  
          return getInstance(elements.withUpdatedUserType(udt), isMultiCell);
      }
  
      @Override
      public AbstractType<?> expandUserTypes()
      {
          return getInstance(elements.expandUserTypes(), isMultiCell);
      }
  
      public AbstractType<T> getElementsType()
      {
          return elements;
      }
  
      public AbstractType<T> nameComparator()
      {
          return elements;
      }
  
      public AbstractType<?> valueComparator()
      {
          return EmptyType.instance;
      }
  
      @Override
      public boolean isMultiCell()
      {
          return isMultiCell;
      }
  
      @Override
      public SetType<T> freeze()
      {
          // freeze elements to match org.apache.cassandra.cql3.CQL3Type.Raw.RawCollection.freeze
          return isMultiCell ? getInstance(this.elements.freeze(), false) : this;
      }
  
      @Override
      public AbstractType<?> unfreeze()
      {
          return isMultiCell ? this : getInstance(this.elements, true);
      }
  
      @Override
      public List<AbstractType<?>> subTypes()
      {
          return Collections.singletonList(elements);
      }
  
      @Override
      public AbstractType<?> freezeNestedMulticellTypes()
      {
          if (!isMultiCell())
              return this;
  
          if (elements.isFreezable() && elements.isMultiCell())
              return getInstance(elements.freeze(), isMultiCell);
  
          return getInstance(elements.freezeNestedMulticellTypes(), isMultiCell);
      }
  
      @Override
      public boolean isCompatibleWithFrozen(CollectionType<?> previous)
      {
          assert !isMultiCell;
          return this.elements.isCompatibleWith(((SetType<?>) previous).elements);
      }
  
      @Override
      public boolean isValueCompatibleWithFrozen(CollectionType<?> previous)
      {
          // because sets are ordered, any changes to the type must maintain the ordering
          return isCompatibleWithFrozen(previous);
      }
  
      public <VL, VR> int compareCustom(VL left, ValueAccessor<VL> accessorL, VR right, ValueAccessor<VR> accessorR)
      {
          return compareListOrSet(elements, left, accessorL, right, accessorR);
      }
  
      @Override
      public <V> ByteSource asComparableBytes(ValueAccessor<V> accessor, V data, ByteComparable.Version version)
      {
          return asComparableBytesListOrSet(getElementsType(), accessor, data, version);
      }
  
      @Override
      public <V> V fromComparableBytes(ValueAccessor<V> accessor, ByteSource.Peekable comparableBytes, ByteComparable.Version version)
      {
          return fromComparableBytesListOrSet(accessor, comparableBytes, version, getElementsType());
      }
  
      public SetSerializer<T> getSerializer()
      {
          return serializer;
      }
  
      @Override
      public String toString(boolean ignoreFreezing)
      {
          boolean includeFrozenType = !ignoreFreezing && !isMultiCell();
  
          StringBuilder sb = new StringBuilder();
          if (includeFrozenType)
              sb.append(FrozenType.class.getName()).append("(");
          sb.append(getClass().getName());
          sb.append(TypeParser.stringifyTypeParameters(Collections.<AbstractType<?>>singletonList(elements), ignoreFreezing || !isMultiCell));
          if (includeFrozenType)
              sb.append(")");
          return sb.toString();
      }
  
      public List<ByteBuffer> serializedValues(Iterator<Cell<?>> cells)
      {
          List<ByteBuffer> bbs = new ArrayList<>();
          while (cells.hasNext())
              bbs.add(cells.next().path().get(0));
          return bbs;
      }
  
      @Override
      public Term fromJSONObject(Object parsed) throws MarshalException
      {
          if (parsed instanceof String)
              parsed = JsonUtils.decodeJson((String) parsed);
  
          if (!(parsed instanceof List))
              throw new MarshalException(String.format(
                      "Expected a list (representing a set), but got a %s: %s", parsed.getClass().getSimpleName(), parsed));
  
          List<?> list = (List<?>) parsed;
          List<Term> terms = new ArrayList<>(list.size());
          for (Object element : list)
          {
              if (element == null)
                  throw new MarshalException("Invalid null element in set");
              terms.add(elements.fromJSONObject(element));
          }
  
          return new MultiElements.DelayedValue(this, terms);
      }
  
      @Override
      public String toJSONString(ByteBuffer buffer, ProtocolVersion protocolVersion)
      {
          return setOrListToJsonString(buffer, elements, protocolVersion);
      }
  
      @Override
      public void forEach(ByteBuffer input, Consumer<ByteBuffer> action)
      {
          serializer.forEach(input, action);
      }
  
      @Override
      public ByteBuffer getMaskedValue()
      {
          return decompose(Collections.emptySet());
      }
  
      @Override
      public List<ByteBuffer> filterSortAndValidateElements(List<ByteBuffer> buffers)
      {
          SortedSet<ByteBuffer> sorted = new TreeSet<>(elements);
          for (ByteBuffer buffer: buffers)
          {
              if (buffer == null)
                  throw new MarshalException("null is not supported inside collections");
              elements.validate(buffer);
              sorted.add(buffer);
          }
          return new ArrayList<>(sorted);
      }
  
      @Override
      protected int compareNextCell(Iterator<Cell<?>> cellIterator, Iterator<ByteBuffer> elementIter)
      {
          return getElementsType().compare(cellIterator.next().path().get(0), elementIter.next());
      }
  
      @Override
      public boolean contains(ComplexColumnData columnData, ByteBuffer value)
      {
          return columnData.getCell(CellPath.create(value)) != null;
      }
  }
  