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.index.sai.utils;
  
  import java.math.BigInteger;
  import java.net.InetAddress;
  import java.nio.ByteBuffer;
  import java.util.ArrayList;
  import java.util.Arrays;
  import java.util.Collection;
  import java.util.Collections;
  import java.util.Comparator;
  import java.util.EnumSet;
  import java.util.Iterator;
  import java.util.List;
  import java.util.Objects;
  import java.util.Set;
  import java.util.stream.Stream;
  import java.util.stream.StreamSupport;
  
  import com.google.common.base.MoreObjects;
  import com.google.common.collect.ImmutableSet;
  
  import com.googlecode.concurrenttrees.radix.ConcurrentRadixTree;
  import org.apache.cassandra.cql3.CQL3Type;
  import org.apache.cassandra.cql3.Operator;
  import org.apache.cassandra.cql3.statements.schema.IndexTarget;
  import org.apache.cassandra.db.DecoratedKey;
  import org.apache.cassandra.db.filter.RowFilter;
  import org.apache.cassandra.db.marshal.AbstractType;
  import org.apache.cassandra.db.marshal.AsciiType;
  import org.apache.cassandra.db.marshal.BooleanType;
  import org.apache.cassandra.db.marshal.ByteBufferAccessor;
  import org.apache.cassandra.db.marshal.CollectionType;
  import org.apache.cassandra.db.marshal.CompositeType;
  import org.apache.cassandra.db.marshal.DecimalType;
  import org.apache.cassandra.db.marshal.InetAddressType;
  import org.apache.cassandra.db.marshal.IntegerType;
  import org.apache.cassandra.db.marshal.LongType;
  import org.apache.cassandra.db.marshal.StringType;
  import org.apache.cassandra.db.marshal.UTF8Type;
  import org.apache.cassandra.db.marshal.UUIDType;
  import org.apache.cassandra.db.marshal.VectorType;
  import org.apache.cassandra.db.rows.Cell;
  import org.apache.cassandra.db.rows.ComplexColumnData;
  import org.apache.cassandra.db.rows.Row;
  import org.apache.cassandra.index.sai.plan.Expression;
  import org.apache.cassandra.schema.ColumnMetadata;
  import org.apache.cassandra.serializers.MarshalException;
  import org.apache.cassandra.utils.ByteBufferUtil;
  import org.apache.cassandra.utils.FastByteOperations;
  import org.apache.cassandra.utils.bytecomparable.ByteComparable;
  import org.apache.cassandra.utils.bytecomparable.ByteSource;
  import org.apache.cassandra.utils.bytecomparable.ByteSourceInverse;
  
  /**
   * This class is a representation of an {@link AbstractType} as an indexable type. It is responsible for determining the
   * capabilities of the type and provides helper methods for handling term values associated with the type.
   */
  public class IndexTermType
  {
      private static final Set<AbstractType<?>> EQ_ONLY_TYPES = ImmutableSet.of(UTF8Type.instance,
                                                                                AsciiType.instance,
                                                                                BooleanType.instance,
                                                                                UUIDType.instance);
  
      private static final byte[] IPV4_PREFIX = new byte[] { 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, -1, -1 };
  
      /**
       * DecimalType / BigDecimal values are indexed by truncating their asComparableBytes representation to this size,
       * padding on the right with zero-value-bytes until this size is reached (if necessary).  This causes
       * false-positives that must be filtered in a separate step after hitting the index and reading the associated
       * (full) values.
       */
      private static final int DECIMAL_APPROXIMATION_BYTES = 24;
      private static final int BIG_INTEGER_APPROXIMATION_BYTES = 20;
      private static final int INET_ADDRESS_SIZE = 16;
      private static final int DEFAULT_FIXED_LENGTH = 16;
  
      private enum Capability
      {
          STRING,
          VECTOR,
          INET_ADDRESS,
          BIG_INTEGER,
          BIG_DECIMAL,
          LONG,
          BOOLEAN,
          LITERAL,
          REVERSED,
          FROZEN,
          COLLECTION,
          NON_FROZEN_COLLECTION,
          COMPOSITE,
          COMPOSITE_PARTITION
      }
  
      private final ColumnMetadata columnMetadata;
      private final IndexTarget.Type indexTargetType;
      private final AbstractType<?> indexType;
      private final List<IndexTermType> subTypes;
      private final AbstractType<?> vectorElementType;
      private final int vectorDimension;
      private final EnumSet<Capability> capabilities;
  
      /**
       * Create an {@link IndexTermType} from a {@link ColumnMetadata} and {@link IndexTarget.Type}.
       *
       * @param columnMetadata the {@link ColumnMetadata} for the column being indexed
       * @param partitionColumns the partition columns for the table this column belongs to. This is used for identifying
       *                         if the {@code columnMetadata} is a partition column and if it belongs to a composite
       *                         partition
       * @param indexTargetType the {@link IndexTarget.Type} for the index
       *
       * @return the {@link IndexTermType}
       */
      public static IndexTermType create(ColumnMetadata columnMetadata, List<ColumnMetadata> partitionColumns, IndexTarget.Type indexTargetType)
      {
          return new IndexTermType(columnMetadata, partitionColumns, indexTargetType);
      }
  
      private IndexTermType(ColumnMetadata columnMetadata, List<ColumnMetadata> partitionColumns, IndexTarget.Type indexTargetType)
      {
          this.columnMetadata = columnMetadata;
          this.indexTargetType = indexTargetType;
          this.capabilities = calculateCapabilities(columnMetadata, partitionColumns, indexTargetType);
          this.indexType = calculateIndexType(columnMetadata.type, capabilities, indexTargetType);
  
          AbstractType<?> baseType = indexType.unwrap();
  
          // We only need to inspect subtypes when it is possible for them to be queried individually.
          if (baseType.subTypes().isEmpty() || indexTargetType == IndexTarget.Type.SIMPLE || indexTargetType == IndexTarget.Type.FULL)
          {
              this.subTypes = Collections.emptyList();
          }
          else
          {
              List<IndexTermType> subTypes = new ArrayList<>(baseType.subTypes().size());
              for (AbstractType<?> subType : baseType.subTypes())
                  subTypes.add(new IndexTermType(columnMetadata.withNewType(subType), partitionColumns, indexTargetType));
              this.subTypes = Collections.unmodifiableList(subTypes);
          }
  
          if (isVector())
          {
              VectorType<?> vectorType = (VectorType<?>) baseType;
              vectorElementType = vectorType.elementType;
              vectorDimension = vectorType.dimension;
          }
          else
          {
              vectorElementType = null;
              vectorDimension = -1;
          }
      }
  
      /**
       * Returns {@code true} if the index type is a literal type and will use a literal index. This applies to
       * string types, frozen types, composite types and boolean type.
       */
      public boolean isLiteral()
      {
          return capabilities.contains(Capability.LITERAL);
      }
  
      /**
       * Returns {@code true} if the index type is a string type. This is used to determine if the type supports
       * analysis.
       */
      public boolean isString()
      {
          return capabilities.contains(Capability.STRING);
      }
  
      /**
       * Returns {@code true} if the index type is a vector type. Note: being a vector type does not mean that the type
       * is valid for indexing in that we don't check the element type and dimension constraints here.
       */
      public boolean isVector()
      {
          return capabilities.contains(Capability.VECTOR);
      }
  
      /**
       * Returns {@code true} if the index type is reversed. This is only the case (currently) for clustering keys with
       * descending ordering.
       */
      public boolean isReversed()
      {
          return capabilities.contains(Capability.REVERSED);
      }
  
      /**
       * Returns {@code true} if the index type is frozen, e.g. the type is wrapped with {@code frozen<type>}.
       */
      public boolean isFrozen()
      {
          return capabilities.contains(Capability.FROZEN);
      }
  
      /**
       * Returns {@code true} if the index type is a non-frozen collection
       */
      public boolean isNonFrozenCollection()
      {
          return capabilities.contains(Capability.NON_FROZEN_COLLECTION);
      }
  
      /**
       * Returns {@code true} if the index type is a frozen collection. This is the inverse of a non-frozen collection
       * but this method is here for clarity.
       */
      public boolean isFrozenCollection()
      {
          return capabilities.contains(Capability.COLLECTION) && capabilities.contains(Capability.FROZEN);
      }
  
      /**
       * Returns {@code true} if the index type is a composite type, e.g. it has the form {@code Composite<typea, typeb>}
       */
      public boolean isComposite()
      {
          return capabilities.contains(Capability.COMPOSITE);
      }
  
      /**
       * Returns {@code true} if the {@link RowFilter.Expression} passed is backed by a non-frozen collection and the
       * {@code Operator} is one that cannot be merged together.
       */
      public boolean isMultiExpression(RowFilter.Expression expression)
      {
          boolean multiExpression = false;
          switch (expression.operator())
          {
              case EQ:
                  multiExpression = isNonFrozenCollection();
                  break;
              case CONTAINS:
              case CONTAINS_KEY:
                  multiExpression = true;
                  break;
          }
          return multiExpression;
      }
  
      /**
       * Returns <code>true</code> if given buffer would pass the {@link AbstractType#validate(ByteBuffer)}
       * check. False otherwise.
       */
      public boolean isValid(ByteBuffer term)
      {
          try
          {
              indexType.validate(term);
              return true;
          }
          catch (MarshalException e)
          {
              return false;
          }
      }
  
      /**
       * @return {@code true} if the empty values of the given type should be excluded from indexing
       */
      public boolean skipsEmptyValue()
      {
          return !indexType.allowsEmpty() || !isLiteral();
      }
  
      public AbstractType<?> indexType()
      {
          return indexType;
      }
  
      public Collection<IndexTermType> subTypes()
      {
          return subTypes;
      }
  
      public CQL3Type asCQL3Type()
      {
          return indexType.asCQL3Type();
      }
  
      public ColumnMetadata columnMetadata()
      {
          return columnMetadata;
      }
  
      public String columnName()
      {
          return columnMetadata.name.toString();
      }
  
      public AbstractType<?> vectorElementType()
      {
          assert isVector();
  
          return vectorElementType;
      }
  
      public int vectorDimension()
      {
          assert isVector();
  
          return vectorDimension;
      }
  
      public boolean dependsOn(ColumnMetadata columnMetadata)
      {
          return this.columnMetadata.compareTo(columnMetadata) == 0;
      }
  
      public static boolean isEqOnlyType(AbstractType<?> type)
      {
          return EQ_ONLY_TYPES.contains(type);
      }
  
      /**
       * Indicates if the type encoding supports rounding of the raw value.
       * <p>
       * This is significant in range searches where we have to make all range
       * queries inclusive when searching the indexes in order to avoid excluding
       * rounded values. Excluded values are removed by post-filtering.
       */
      public boolean supportsRounding()
      {
          return isBigInteger() || isBigDecimal();
      }
  
      /**
       * Returns the value length for the given {@link AbstractType}, selecting 16 for types
       * that officially use VARIABLE_LENGTH but are, in fact, of a fixed length.
       */
      public int fixedSizeOf()
      {
          if (indexType.isValueLengthFixed())
              return indexType.valueLengthIfFixed();
          else if (isInetAddress())
              return INET_ADDRESS_SIZE;
          else if (isBigInteger())
              return BIG_INTEGER_APPROXIMATION_BYTES;
          else if (isBigDecimal())
              return DECIMAL_APPROXIMATION_BYTES;
          return DEFAULT_FIXED_LENGTH;
      }
  
      /**
       * Allows overriding the default getString method for {@link CompositeType}. It is
       * a requirement of the {@link ConcurrentRadixTree} that the keys are strings but
       * the getString method of {@link CompositeType} does not return a string that compares
       * in the same order as the underlying {@link ByteBuffer}. To get round this we convert
       * the {@link CompositeType} bytes to a hex string.
       */
      public String asString(ByteBuffer value)
      {
          if (isComposite())
              return ByteBufferUtil.bytesToHex(value);
          return indexType.getString(value);
      }
  
      /**
       * The inverse of the above method. Overrides the fromString method on {@link CompositeType}
       * in order to convert the hex string to bytes.
       */
      public ByteBuffer fromString(String value)
      {
          if (isComposite())
              return ByteBufferUtil.hexToBytes(value);
          return indexType.fromString(value);
      }
  
      /**
       * Returns the cell value from the {@link DecoratedKey} or {@link Row} for the {@link IndexTermType} based on the
       * kind of column this {@link IndexTermType} is based on.
       *
       * @param key the {@link DecoratedKey} of the row
       * @param row the {@link Row} containing the non-partition column data
       * @param nowInSecs the time that the index write operation started
       *
       * @return a {@link ByteBuffer} containing the cell value
       */
      public ByteBuffer valueOf(DecoratedKey key, Row row, long nowInSecs)
      {
          if (row == null)
              return null;
  
          switch (columnMetadata.kind)
          {
              case PARTITION_KEY:
                  return isCompositePartition() ? CompositeType.extractComponent(key.getKey(), columnMetadata.position())
                                                : key.getKey();
              case CLUSTERING:
                  // skip indexing of static clustering when regular column is indexed
                  return row.isStatic() ? null : row.clustering().bufferAt(columnMetadata.position());
  
              // treat static cell retrieval the same was as regular
              // only if row kind is STATIC otherwise return null
              case STATIC:
                  if (!row.isStatic())
                      return null;
              case REGULAR:
                  Cell<?> cell = row.getCell(columnMetadata);
                  return cell == null || !cell.isLive(nowInSecs) ? null : cell.buffer();
  
              default:
                  return null;
          }
      }
  
      /**
       * Returns a value iterator for collection type {@link IndexTermType}s.
       *
       * @param row the {@link Row} containing the column data
       * @param nowInSecs the time that the index write operation started
       *
       * @return an {@link Iterator} of the collection values
       */
      public Iterator<ByteBuffer> valuesOf(Row row, long nowInSecs)
      {
          if (row == null)
              return null;
  
          switch (columnMetadata.kind)
          {
              // treat static cell retrieval the same was as regular
              // only if row kind is STATIC otherwise return null
              case STATIC:
                  if (!row.isStatic())
                      return null;
              case REGULAR:
                  return collectionIterator(row.getComplexColumnData(columnMetadata), nowInSecs);
  
              default:
                  return null;
          }
      }
  
      public Comparator<ByteBuffer> comparator()
      {
          // Override the comparator for BigInteger, frozen collections and composite types
          if (isBigInteger() || isBigDecimal() || isComposite() || isFrozen())
              return FastByteOperations::compareUnsigned;
  
          return indexType;
      }
  
      /**
       * Compare two terms based on their type. This is used in place of {@link AbstractType#compare(ByteBuffer, ByteBuffer)}
       * so that the default comparison can be overridden for specific types.
       * <p>
       * Note: This should be used for all term comparison
       */
      public int compare(ByteBuffer b1, ByteBuffer b2)
      {
          if (isInetAddress())
              return compareInet(b1, b2);
          else if (isLong())
              return indexType.unwrap().compare(b1, b2);
          // BigInteger values, frozen types and composite types (map entries) use compareUnsigned to maintain
          // a consistent order between the in-memory index and the on-disk index.
          else if (isBigInteger() || isBigDecimal() || isComposite() || isFrozen())
              return FastByteOperations.compareUnsigned(b1, b2);
  
          return indexType.compare(b1, b2);
      }
  
      /**
       * Returns the smaller of two {@code ByteBuffer} values, based on the result of {@link
       * #compare(ByteBuffer, ByteBuffer)} comparision.
       */
      public ByteBuffer min(ByteBuffer a, ByteBuffer b)
      {
          return a == null ? b : (b == null || compare(b, a) > 0) ? a : b;
      }
  
      /**
       * Returns the greater of two {@code ByteBuffer} values, based on the result of {@link
       * #compare(ByteBuffer, ByteBuffer)} comparision.
       */
      public ByteBuffer max(ByteBuffer a, ByteBuffer b)
      {
          return a == null ? b : (b == null || compare(b, a) < 0) ? a : b;
      }
  
      /**
       * This is used for value comparison in post-filtering - {@link Expression#isSatisfiedBy(ByteBuffer)}.
       * <p>
       * This allows types to decide whether they should be compared based on their encoded value or their
       * raw value. At present only {@link InetAddressType} values are compared by their encoded values to
       * allow for ipv4 -> ipv6 equivalency in searches.
       */
      public int comparePostFilter(Expression.Value requestedValue, Expression.Value columnValue)
      {
          if (isInetAddress())
              return compareInet(requestedValue.encoded, columnValue.encoded);
          // bigint, decimal, and varint are not indexed in reversed byte-comparable form or treated as reversed types by
          // Expression, so it is correct to compare with the base/unwrapped type
          else if (isLong() || isBigDecimal() || isBigInteger())
              return indexType.unwrap().compare(requestedValue.raw, columnValue.raw);
          // Override comparisons for frozen collections and composite types (map entries)
          else if (isComposite() || isFrozen())
              return FastByteOperations.compareUnsigned(requestedValue.raw, columnValue.raw);
  
          // Reversed types are treated as such by Expression here, so we cannot blindly compare with the unwrapped type.
          // In the future, we might consider simplifying things to have SAI ignore reversed types altogether, but this
          // will require a change to the on-disk formats.
          return indexType.compare(requestedValue.raw, columnValue.raw);
      }
  
      /**
       * Fills a byte array with the comparable bytes for a type.
       * <p>
       * This method expects a {@code value} parameter generated by calling {@link #asIndexBytes(ByteBuffer)}.
       * It is not generally safe to pass the output of other serialization methods to this method.  For instance, it is
       * not generally safe to pass the output of {@link AbstractType#decompose(Object)} as the {@code value} parameter
       * (there are certain types for which this is technically OK, but that doesn't hold for all types).
       *
       * @param value a value buffer returned by {@link #asIndexBytes(ByteBuffer)}
       * @param bytes this method's output
       */
      public void toComparableBytes(ByteBuffer value, byte[] bytes)
      {
          if (isInetAddress())
              ByteBufferUtil.copyBytes(value, value.hasArray() ? value.arrayOffset() + value.position() : value.position(), bytes, 0, INET_ADDRESS_SIZE);
          else if (isBigInteger())
              ByteBufferUtil.copyBytes(value, value.hasArray() ? value.arrayOffset() + value.position() : value.position(), bytes, 0, BIG_INTEGER_APPROXIMATION_BYTES);
          else if (isBigDecimal())
              ByteBufferUtil.copyBytes(value, value.hasArray() ? value.arrayOffset() + value.position() : value.position(), bytes, 0, DECIMAL_APPROXIMATION_BYTES);
          else
              ByteSourceInverse.copyBytes(asComparableBytes(value, ByteComparable.Version.OSS50), bytes);
      }
  
      public ByteSource asComparableBytes(ByteBuffer value, ByteComparable.Version version)
      {
          if (value.remaining() == 0)
              return ByteSource.EMPTY;
  
          if (isInetAddress() || isBigInteger() || isBigDecimal())
              return ByteSource.optionalFixedLength(ByteBufferAccessor.instance, value);
          else if (isLong())
              // The LongType.asComparableBytes uses variableLengthInteger which doesn't play well with
              // the balanced tree because it is expecting fixed length data. So for SAI we use a optionalSignedFixedLengthNumber
              // to keep all comparable values the same length
              return ByteSource.optionalSignedFixedLengthNumber(ByteBufferAccessor.instance, value);
          else if (isFrozen())
              // We need to override the default frozen implementation here because it will defer to the underlying
              // type's implementation which will be incorrect, for us, for the case of multi-cell types.
              return ByteSource.of(value, version);
          return indexType.asComparableBytes(value, version);
      }
  
      /**
       * Translates the external value of specific types into a format used by the index.
       */
      public ByteBuffer asIndexBytes(ByteBuffer value)
      {
          if (value == null || value.remaining() == 0)
              return value;
  
          if (isInetAddress())
              return encodeInetAddress(value);
          else if (isBigInteger())
              return encodeBigInteger(value);
          else if (isBigDecimal())
              return encodeDecimal(value);
          return value;
      }
  
      public float[] decomposeVector(ByteBuffer byteBuffer)
      {
          assert isVector();
          return ((VectorType<?>) indexType).composeAsFloat(byteBuffer);
      }
  
      public boolean supports(Operator operator)
      {
          if (operator == Operator.LIKE ||
              operator == Operator.LIKE_CONTAINS ||
              operator == Operator.LIKE_PREFIX ||
              operator == Operator.LIKE_MATCHES ||
              operator == Operator.LIKE_SUFFIX ||
              operator == Operator.IN) return false;
  
          // ANN is only supported against vectors, and vector indexes only support ANN
          if (operator == Operator.ANN)
              return isVector();
  
          Expression.IndexOperator indexOperator = Expression.IndexOperator.valueOf(operator);
  
          if (isNonFrozenCollection())
          {
              if (indexTargetType == IndexTarget.Type.KEYS) return indexOperator == Expression.IndexOperator.CONTAINS_KEY;
              if (indexTargetType == IndexTarget.Type.VALUES) return indexOperator == Expression.IndexOperator.CONTAINS_VALUE;
              return indexTargetType == IndexTarget.Type.KEYS_AND_VALUES && indexOperator == Expression.IndexOperator.EQ;
          }
  
          if (indexTargetType == IndexTarget.Type.FULL)
              return indexOperator == Expression.IndexOperator.EQ;
  
          if (indexOperator != Expression.IndexOperator.EQ && EQ_ONLY_TYPES.contains(indexType)) return false;
  
          // RANGE only applicable to non-literal indexes
          return (indexOperator != null) && !(isLiteral() && indexOperator == Expression.IndexOperator.RANGE);
      }
  
      @Override
      public String toString()
      {
          return MoreObjects.toStringHelper(this)
                            .add("column", columnMetadata)
                            .add("type", indexType)
                            .add("indexType", indexTargetType)
                            .toString();
      }
  
      @Override
      public boolean equals(Object obj)
      {
          if (obj == this)
              return true;
  
          if (!(obj instanceof IndexTermType))
              return false;
  
          IndexTermType other = (IndexTermType) obj;
  
          return Objects.equals(columnMetadata, other.columnMetadata) && (indexTargetType == other.indexTargetType);
      }
  
      @Override
      public int hashCode()
      {
          return Objects.hash(columnMetadata, indexTargetType);
      }
  
      private EnumSet<Capability> calculateCapabilities(ColumnMetadata columnMetadata, List<ColumnMetadata> partitionKeyColumns, IndexTarget.Type indexTargetType)
      {
          EnumSet<Capability> capabilities = EnumSet.noneOf(Capability.class);
  
          if (partitionKeyColumns.contains(columnMetadata) && partitionKeyColumns.size() > 1)
              capabilities.add(Capability.COMPOSITE_PARTITION);
  
          AbstractType<?> type = columnMetadata.type;
          boolean reversed = type.isReversed();
          AbstractType<?> baseType = type.unwrap();
  
          if (baseType.isCollection())
              capabilities.add(Capability.COLLECTION);
  
          if (baseType.isCollection() && baseType.isMultiCell())
              capabilities.add(Capability.NON_FROZEN_COLLECTION);
  
          if (!baseType.subTypes().isEmpty() && !baseType.isMultiCell())
              capabilities.add(Capability.FROZEN);
  
          AbstractType<?> indexType = calculateIndexType(baseType, capabilities, indexTargetType);
  
          if (indexType instanceof CompositeType)
              capabilities.add(Capability.COMPOSITE);
          else if (!indexType.subTypes().isEmpty() && !indexType.isMultiCell())
              capabilities.add(Capability.FROZEN);
  
          if (indexType instanceof StringType)
              capabilities.add(Capability.STRING);
  
          if (indexType instanceof BooleanType)
              capabilities.add(Capability.BOOLEAN);
  
          if (capabilities.contains(Capability.STRING) ||
              capabilities.contains(Capability.BOOLEAN) ||
              capabilities.contains(Capability.FROZEN) ||
              capabilities.contains(Capability.COMPOSITE))
              capabilities.add(Capability.LITERAL);
  
          if (indexType instanceof VectorType<?>)
              capabilities.add(Capability.VECTOR);
  
          if (indexType instanceof InetAddressType)
          {
              capabilities.add(Capability.INET_ADDRESS);
              reversed = false;
          }
  
          if (indexType instanceof IntegerType)
          {
              capabilities.add(Capability.BIG_INTEGER);
              reversed = false;
          }
  
          if (indexType instanceof DecimalType)
          {
              capabilities.add(Capability.BIG_DECIMAL);
              reversed = false;
          }
  
          if (indexType instanceof LongType)
          {
              capabilities.add(Capability.LONG);
              reversed = false;
          }
  
          if (reversed)
              capabilities.add(Capability.REVERSED);
  
          return capabilities;
      }
  
      private AbstractType<?> calculateIndexType(AbstractType<?> baseType, EnumSet<Capability> capabilities, IndexTarget.Type indexTargetType)
      {
          return capabilities.contains(Capability.NON_FROZEN_COLLECTION) ? collectionCellValueType(baseType, indexTargetType) : baseType;
      }
  
      private Iterator<ByteBuffer> collectionIterator(ComplexColumnData cellData, long nowInSecs)
      {
          if (cellData == null)
              return null;
  
          Stream<ByteBuffer> stream = StreamSupport.stream(cellData.spliterator(), false)
                                                   .filter(cell -> cell != null && cell.isLive(nowInSecs))
                                                   .map(this::cellValue);
  
          if (isInetAddress())
              stream = stream.sorted((c1, c2) -> compareInet(encodeInetAddress(c1), encodeInetAddress(c2)));
  
          return stream.iterator();
      }
  
      private ByteBuffer cellValue(Cell<?> cell)
      {
          if (isNonFrozenCollection())
          {
              switch (((CollectionType<?>) columnMetadata.type).kind)
              {
                  case LIST:
                      return cell.buffer();
                  case SET:
                      return cell.path().get(0);
                  case MAP:
                      switch (indexTargetType)
                      {
                          case KEYS:
                              return cell.path().get(0);
                          case VALUES:
                              return cell.buffer();
                          case KEYS_AND_VALUES:
                              return CompositeType.build(ByteBufferAccessor.instance, cell.path().get(0), cell.buffer());
                      }
              }
          }
          return cell.buffer();
      }
  
      private AbstractType<?> collectionCellValueType(AbstractType<?> type, IndexTarget.Type indexType)
      {
          CollectionType<?> collection = ((CollectionType<?>) type);
          switch (collection.kind)
          {
              case LIST:
                  return collection.valueComparator();
              case SET:
                  return collection.nameComparator();
              case MAP:
                  switch (indexType)
                  {
                      case KEYS:
                          return collection.nameComparator();
                      case VALUES:
                          return collection.valueComparator();
                      case KEYS_AND_VALUES:
                          return CompositeType.getInstance(collection.nameComparator(), collection.valueComparator());
                  }
              default:
                  throw new IllegalArgumentException("Unsupported collection type: " + collection.kind);
          }
      }
  
      private boolean isCompositePartition()
      {
          return capabilities.contains(Capability.COMPOSITE_PARTITION);
      }
  
      /**
       * Returns <code>true</code> if given {@link AbstractType} is {@link InetAddressType}
       */
      private boolean isInetAddress()
      {
          return capabilities.contains(Capability.INET_ADDRESS);
      }
  
      /**
       * Returns <code>true</code> if given {@link AbstractType} is {@link IntegerType}
       */
      private boolean isBigInteger()
      {
          return capabilities.contains(Capability.BIG_INTEGER);
      }
  
      /**
       * Returns <code>true</code> if given {@link AbstractType} is {@link DecimalType}
       */
      private boolean isBigDecimal()
      {
          return capabilities.contains(Capability.BIG_DECIMAL);
      }
  
      private boolean isLong()
      {
          return capabilities.contains(Capability.LONG);
      }
  
      /**
       * Compares 2 InetAddress terms by ensuring that both addresses are represented as
       * ipv6 addresses.
       */
      private static int compareInet(ByteBuffer b1, ByteBuffer b2)
      {
          assert isIPv6(b1) && isIPv6(b2);
  
          return FastByteOperations.compareUnsigned(b1, b2);
      }
  
      private static boolean isIPv6(ByteBuffer address)
      {
          return address.remaining() == INET_ADDRESS_SIZE;
      }
  
      /**
       * Encode a {@link InetAddress} into a fixed width 16 byte encoded value.
       * <p>
       * The encoded value is byte comparable and prefix compressible.
       * <p>
       * The encoding is done by converting ipv4 addresses to their ipv6 equivalent.
       */
      private static ByteBuffer encodeInetAddress(ByteBuffer value)
      {
          if (value.remaining() == 4)
          {
              int position = value.hasArray() ? value.arrayOffset() + value.position() : value.position();
              ByteBuffer mapped = ByteBuffer.allocate(INET_ADDRESS_SIZE);
              System.arraycopy(IPV4_PREFIX, 0, mapped.array(), 0, IPV4_PREFIX.length);
              ByteBufferUtil.copyBytes(value, position, mapped, IPV4_PREFIX.length, value.remaining());
              return mapped;
          }
          return value;
      }
  
      /**
       * Encode a {@link BigInteger} into a fixed width 20 byte encoded value. The encoded value is byte comparable
       * and prefix compressible.
       * <p>
       * The format of the encoding is:
       * <p>
       *  The first 4 bytes contain the integer length of the {@link BigInteger} byte array
       *  with the top bit flipped for positive values.
       * <p>
       *  The remaining 16 bytes contain the 16 most significant bytes of the
       *  {@link BigInteger} byte array.
       * <p>
       *  For {@link BigInteger} values whose underlying byte array is less than
       *  16 bytes, the encoded value is sign extended.
       */
      public static ByteBuffer encodeBigInteger(ByteBuffer value)
      {
          int size = value.remaining();
          int position = value.hasArray() ? value.arrayOffset() + value.position() : value.position();
          byte[] bytes = new byte[BIG_INTEGER_APPROXIMATION_BYTES];
          if (size < BIG_INTEGER_APPROXIMATION_BYTES - Integer.BYTES)
          {
              ByteBufferUtil.copyBytes(value, position, bytes, bytes.length - size, size);
              if ((bytes[bytes.length - size] & 0x80) != 0)
                  Arrays.fill(bytes, Integer.BYTES, bytes.length - size, (byte)0xff);
              else
                  Arrays.fill(bytes, Integer.BYTES, bytes.length - size, (byte)0x00);
          }
          else
          {
              ByteBufferUtil.copyBytes(value, position, bytes, Integer.BYTES, BIG_INTEGER_APPROXIMATION_BYTES - Integer.BYTES);
          }
          if ((bytes[4] & 0x80) != 0)
          {
              size = -size;
          }
          bytes[0] = (byte)(size >> 24 & 0xff);
          bytes[1] = (byte)(size >> 16 & 0xff);
          bytes[2] = (byte)(size >> 8 & 0xff);
          bytes[3] = (byte)(size & 0xff);
          bytes[0] ^= 0x80;
          return ByteBuffer.wrap(bytes);
      }
  
      public static ByteBuffer encodeDecimal(ByteBuffer value)
      {
          ByteSource bs = DecimalType.instance.asComparableBytes(value, ByteComparable.Version.OSS50);
          bs = ByteSource.cutOrRightPad(bs, DECIMAL_APPROXIMATION_BYTES, 0);
          return ByteBuffer.wrap(ByteSourceInverse.readBytes(bs, DECIMAL_APPROXIMATION_BYTES));
      }
  }
  