Fix detected issues

src/java/org/apache/cassandra/db/marshal/AbstractCompositeType.java

@@ -59,7 +59,7 @@ public abstract class AbstractCompositeType extends AbstractType<ByteBuffer>
         int offsetL = startingOffset(isStaticL);
         int offsetR = startingOffset(isStaticR);
 
-        while (!accessorL.isEmptyFromOffset(left, offsetL) && !accessorR.isEmptyFromOffset(right, offsetL))
+        while (!accessorL.isEmptyFromOffset(left, offsetL) && !accessorR.isEmptyFromOffset(right, offsetR))
         {
             AbstractType<?> comparator = getComparator(i, left, accessorL, right, accessorR, offsetL, offsetR);
             offsetL += getComparatorSize(i, left, accessorL, offsetL);

src/java/org/apache/cassandra/db/marshal/CompositeType.java

@@ -21,6 +21,7 @@ import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 
@@ -29,7 +30,9 @@ import com.google.common.collect.Lists;
 
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.exceptions.SyntaxException;
+import org.apache.cassandra.serializers.BytesSerializer;
 import org.apache.cassandra.serializers.MarshalException;
+import org.apache.cassandra.serializers.TypeSerializer;
 import org.apache.cassandra.utils.ByteBufferUtil;
 
 import static com.google.common.collect.Iterables.any;
@@ -62,9 +65,37 @@ import static com.google.common.collect.Iterables.transform;
  */
 public class CompositeType extends AbstractCompositeType
 {
+    private static class Serializer extends BytesSerializer
+    {
+        // types are held to make sure the serializer is unique for each collection of types, this is to make sure it's
+        // safe to cache in all cases
+        public final List<AbstractType<?>> types;
+
+        public Serializer(List<AbstractType<?>> types)
+        {
+            this.types = types;
+        }
+
+        @Override
+        public boolean equals(Object o)
+        {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+            Serializer that = (Serializer) o;
+            return types.equals(that.types);
+        }
+
+        @Override
+        public int hashCode()
+        {
+            return Objects.hash(types);
+        }
+    }
+
     private static final int STATIC_MARKER = 0xFFFF;
 
     public final List<AbstractType<?>> types;
+    private final Serializer serializer;
 
     // interning instances
     private static final ConcurrentMap<List<AbstractType<?>>, CompositeType> instances = new ConcurrentHashMap<>();
@@ -136,8 +167,16 @@ public class CompositeType extends AbstractCompositeType
     protected CompositeType(List<AbstractType<?>> types)
     {
         this.types = ImmutableList.copyOf(types);
+        this.serializer = new Serializer(this.types);
     }
 
+    @Override
+    public TypeSerializer<ByteBuffer> getSerializer()
+    {
+        return serializer;
+    }
+
+
     protected <V> AbstractType<?> getComparator(int i, V value, ValueAccessor<V> accessor, int offset)
     {
         try

src/java/org/apache/cassandra/db/marshal/DynamicCompositeType.java

@@ -20,9 +20,11 @@ package org.apache.cassandra.db.marshal;
 import java.nio.ByteBuffer;
 import java.nio.charset.CharacterCodingException;
 import java.util.Map;
+import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
 
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.ImmutableMap;
 import com.google.common.collect.Maps;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -31,6 +33,7 @@ import org.apache.cassandra.cql3.Term;
 import org.apache.cassandra.db.TypeSizes;
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.exceptions.SyntaxException;
+import org.apache.cassandra.serializers.BytesSerializer;
 import org.apache.cassandra.serializers.MarshalException;
 import org.apache.cassandra.serializers.TypeSerializer;
 import org.apache.cassandra.transport.ProtocolVersion;
@@ -61,8 +64,36 @@ public class DynamicCompositeType extends AbstractCompositeType
 {
     private static final Logger logger = LoggerFactory.getLogger(DynamicCompositeType.class);
 
+    public static class Serializer extends BytesSerializer
+    {
+        // aliases are held to make sure the serializer is unique for each collection of types, this is to make sure it's
+        // safe to cache in all cases
+        private final Map<Byte, AbstractType<?>> aliases;
+
+        public Serializer(Map<Byte, AbstractType<?>> aliases)
+        {
+            this.aliases = aliases;
+        }
+
+        @Override
+        public boolean equals(Object o)
+        {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+            Serializer that = (Serializer) o;
+            return aliases.equals(that.aliases);
+        }
+
+        @Override
+        public int hashCode()
+        {
+            return Objects.hash(aliases);
+        }
+    }
+
     @VisibleForTesting
     public final Map<Byte, AbstractType<?>> aliases;
+    private final Serializer serializer;
 
     // interning instances
     private static final ConcurrentHashMap<Map<Byte, AbstractType<?>>, DynamicCompositeType> instances = new ConcurrentHashMap<>();
@@ -82,7 +113,14 @@ public class DynamicCompositeType extends AbstractCompositeType
 
     private DynamicCompositeType(Map<Byte, AbstractType<?>> aliases)
     {
-        this.aliases = aliases;
+        this.aliases = ImmutableMap.copyOf(aliases);
+        this.serializer = new Serializer(aliases);
+    }
+
+    @Override
+    public TypeSerializer<ByteBuffer> getSerializer()
+    {
+        return serializer;
     }
 
     protected <V> boolean readIsStatic(V value, ValueAccessor<V> accessor)

src/java/org/apache/cassandra/db/marshal/LexicalUUIDType.java

@@ -31,6 +31,12 @@ public class LexicalUUIDType extends AbstractType<UUID>
 {
     public static final LexicalUUIDType instance = new LexicalUUIDType();
 
+    // we just need it to be something different to UUIDSerializer
+    private static class Serializer extends UUIDSerializer
+    {
+        public static final Serializer instance = new Serializer();
+    }
+
     LexicalUUIDType()
     {
         super(ComparisonType.CUSTOM);
@@ -80,7 +86,7 @@ public class LexicalUUIDType extends AbstractType<UUID>
 
     public TypeSerializer<UUID> getSerializer()
     {
-        return UUIDSerializer.instance;
+        return Serializer.instance;
     }
 
     @Override

src/java/org/apache/cassandra/db/marshal/TupleType.java

@@ -435,7 +435,7 @@ public class TupleType extends AbstractType<ByteBuffer>
     @Override
     public boolean equals(Object o)
     {
-        if(!(o instanceof TupleType))
+        if (o.getClass() != TupleType.class)
             return false;
 
         TupleType that = (TupleType)o;

src/java/org/apache/cassandra/db/marshal/UserType.java

@@ -342,7 +342,7 @@ public class UserType extends TupleType implements SchemaElement
     @Override
     public boolean equals(Object o)
     {
-        if(!(o instanceof UserType))
+        if (o.getClass() != UserType.class)
             return false;
 
         UserType that = (UserType)o;