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.concurrent.ConcurrentHashMap;
  import java.util.function.Consumer;
  
  import javax.annotation.Nullable;
  
  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.ColumnData;
  import org.apache.cassandra.db.rows.ComplexColumnData;
  import org.apache.cassandra.exceptions.ConfigurationException;
  import org.apache.cassandra.exceptions.SyntaxException;
  import org.apache.cassandra.serializers.ListSerializer;
  import org.apache.cassandra.serializers.MarshalException;
  import org.apache.cassandra.utils.ByteBufferUtil;
  import org.apache.cassandra.transport.ProtocolVersion;
  import org.apache.cassandra.utils.JsonUtils;
  import org.apache.cassandra.utils.TimeUUID;
  import org.apache.cassandra.utils.bytecomparable.ByteComparable.Version;
  import org.apache.cassandra.utils.bytecomparable.ByteSource;
  
  import static org.apache.cassandra.cql3.statements.RequestValidations.checkFalse;
  
  public class ListType<T> extends CollectionType<List<T>>
  {
      // interning instances
      private static final ConcurrentHashMap<AbstractType<?>, ListType> instances = new ConcurrentHashMap<>();
      private static final ConcurrentHashMap<AbstractType<?>, ListType> frozenInstances = new ConcurrentHashMap<>();
  
      private final AbstractType<T> elements;
      private final ListSerializer<T> serializer;
      private final boolean isMultiCell;
  
      public static ListType<?> getInstance(TypeParser parser) throws ConfigurationException, SyntaxException
      {
          List<AbstractType<?>> l = parser.getTypeParameters();
          if (l.size() != 1)
              throw new ConfigurationException("ListType takes exactly 1 type parameter");
  
          return getInstance(l.get(0).freeze(), true);
      }
  
      public static <T> ListType<T> getInstance(AbstractType<T> elements, boolean isMultiCell)
      {
          ConcurrentHashMap<AbstractType<?>, ListType> internMap = isMultiCell ? instances : frozenInstances;
          ListType<T> t = internMap.get(elements);
          return null == t
               ? internMap.computeIfAbsent(elements, k -> new ListType<>(k, isMultiCell))
               : t;
      }
  
      private ListType(AbstractType<T> elements, boolean isMultiCell)
      {
          super(ComparisonType.CUSTOM, Kind.LIST);
          this.elements = elements;
          this.serializer = ListSerializer.getInstance(elements.getSerializer());
          this.isMultiCell = isMultiCell;
      }
  
      @Override
      public <V> boolean referencesUserType(V name, ValueAccessor<V> accessor)
      {
          return elements.referencesUserType(name, accessor);
      }
  
      @Override
      public ListType<?> 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);
      }
  
      @Override
      public boolean referencesDuration()
      {
          return getElementsType().referencesDuration();
      }
  
      public AbstractType<T> getElementsType()
      {
          return elements;
      }
  
      public AbstractType<TimeUUID> nameComparator()
      {
          return TimeUUIDType.instance;
      }
  
      public AbstractType<T> valueComparator()
      {
          return elements;
      }
  
      public ListSerializer<T> getSerializer()
      {
          return serializer;
      }
  
      @Override
      public ListType<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 AbstractType<?> freezeNestedMulticellTypes()
      {
          if (!isMultiCell())
              return this;
  
          if (elements.isFreezable() && elements.isMultiCell())
              return getInstance(elements.freeze(), isMultiCell);
  
          return getInstance(elements.freezeNestedMulticellTypes(), isMultiCell);
      }
  
      @Override
      public List<AbstractType<?>> subTypes()
      {
          return Collections.singletonList(elements);
      }
  
      @Override
      public boolean isMultiCell()
      {
          return isMultiCell;
      }
  
      @Override
      public boolean isCompatibleWithFrozen(CollectionType<?> previous)
      {
          assert !isMultiCell;
          return this.elements.isCompatibleWith(((ListType<?>) previous).elements);
      }
  
      @Override
      public boolean isValueCompatibleWithFrozen(CollectionType<?> previous)
      {
          assert !isMultiCell;
          return this.elements.isValueCompatibleWithInternal(((ListType<?>) previous).elements);
      }
  
      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, Version version)
      {
          return asComparableBytesListOrSet(getElementsType(), accessor, data, version);
      }
  
      @Override
      public <V> V fromComparableBytes(ValueAccessor<V> accessor, ByteSource.Peekable comparableBytes, Version version)
      {
          return fromComparableBytesListOrSet(accessor, comparableBytes, version, getElementsType());
      }
  
      @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)
      {
          assert isMultiCell;
          List<ByteBuffer> bbs = new ArrayList<ByteBuffer>();
          while (cells.hasNext())
              bbs.add(cells.next().buffer());
          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, 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 list");
              terms.add(elements.fromJSONObject(element));
          }
  
          return new MultiElements.DelayedValue(this, terms);
      }
  
      public ByteBuffer getSliceFromSerialized(ByteBuffer collection, ByteBuffer from, ByteBuffer to)
      {
          // We don't support slicing on lists so we don't need that function
          throw new UnsupportedOperationException();
      }
  
      @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.emptyList());
      }
  
      @Override
      public List<ByteBuffer> filterSortAndValidateElements(List<ByteBuffer> buffers)
      {
          for (ByteBuffer buffer: buffers)
          {
              if (buffer == null)
                  throw new MarshalException("null is not supported inside collections");
              elements.validate(buffer);
          }
          return buffers;
      }
  
      @Override
      protected int compareNextCell(Iterator<Cell<?>> cellIterator, Iterator<ByteBuffer> elementIter)
      {
          return getElementsType().compare(cellIterator.next().buffer(), elementIter.next());
      }
  
      @Override
      public boolean contains(ComplexColumnData columnData, ByteBuffer value)
      {
          Iterator<Cell<?>> iter = columnData.iterator();
          while(iter.hasNext())
          {
              ByteBuffer cellValue = iter.next().buffer();
              if(valueComparator().compare(cellValue, value) == 0)
                  return true;
          }
          return false;
      }
  
      @Override
      public AbstractType<?> elementType(ByteBuffer keyOrIndex)
      {
          return getElementsType();
      }
  
      @Override
      public ByteBuffer getElement(@Nullable ColumnData columnData, ByteBuffer keyOrIndex)
      {
          if (columnData == null)
              return null;
  
          int idx = listIndex(keyOrIndex);
  
          if (isMultiCell())
          {
              ComplexColumnData complexColumnData = (ComplexColumnData) columnData;
  
              if (idx >= complexColumnData.cellsCount())
                  return null;
  
              Cell<?> cell = complexColumnData.getCellByIndex(idx);
              return cell == null ? null : cell.buffer();
          }
  
          List<ByteBuffer> cells = unpack(((Cell<?>) columnData).buffer());
          return idx >= cells.size() ? null : cells.get(idx);
      }
  
      private int listIndex(ByteBuffer index)
      {
          int idx = ByteBufferUtil.toInt(index);
          checkFalse(idx < 0, "Invalid negative list index %d", idx);
          return idx;
      }
  }
  