Read only

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

@@ -208,8 +208,8 @@ public <V> String getString(V input, ValueAccessor<V> accessor)
     public ByteBuffer fromString(String source)
     {
         List<String> parts = split(source);
-        List<ByteBuffer> components = new ArrayList<ByteBuffer>(parts.size());
-        List<ParsedComparator> comparators = new ArrayList<ParsedComparator>(parts.size());
+        List<ByteBuffer> components = new ArrayList<>(parts.size());
+        List<ParsedComparator> comparators = new ArrayList<>(parts.size());
         int totalLength = 0, i = 0;
         boolean lastByteIsOne = false;
         boolean lastByteIsMinusOne = false;
@@ -244,7 +244,7 @@ else if (part.equals("_"))
         {
             comparators.get(i).serializeComparator(bb);
             ByteBufferUtil.writeShortLength(bb, component.remaining());
-            bb.put(component); // it's ok to consume component as we won't use it anymore
+            bb.put(component.duplicate()); // it's not ok to consume component as we did not create it (CASSANDRA-14752)
             bb.put((byte)0);
             ++i;
         }

src/java/org/apache/cassandra/io/compress/BufferType.java

@@ -42,4 +42,20 @@ public static BufferType typeOf(ByteBuffer buffer)
     {
         return buffer.isDirect() ? OFF_HEAP : ON_HEAP;
     }
+
+    /**
+     * Allocate a buffer of the current type, write the content to it and return it.
+     * The buffer is repositioned at the beginning, and it's ready for reading.
+     *
+     * @param content - the content to write
+     *
+     * @return a byte buffer with the specified content written in it
+     */
+    public ByteBuffer withContent(byte[] content)
+    {
+        ByteBuffer ret = allocate(content.length);
+        ret.put(content);
+        ret.rewind();
+        return ret;
+    }
 }

src/java/org/apache/cassandra/serializers/BooleanSerializer.java

@@ -20,12 +20,21 @@
 import java.nio.ByteBuffer;
 
 import org.apache.cassandra.db.marshal.ValueAccessor;
+import org.apache.cassandra.io.compress.BufferType;
 import org.apache.cassandra.utils.ByteBufferUtil;
 
 public class BooleanSerializer extends TypeSerializer<Boolean>
 {
-    private static final ByteBuffer TRUE = ByteBuffer.wrap(new byte[] {1});
-    private static final ByteBuffer FALSE = ByteBuffer.wrap(new byte[] {0});
+    // Note: if using read-only byte buffers, it's important that they are off-heap because on-heap read only
+    // byte buffers will return false when hasArray()  is called and we have code that incorrectly assumes
+    // !hasArray() == isDirect()
+    // Direct read-only BBs should work fine: isDirect returns true and MemoryUtil will grant access
+    // to their address.
+    // Read-only off-heap byte buffers are safer than writable on-heap buffers, but not as safe as read-only on-heap
+    // buffers since people can always write directly to their address and use them as source in a put() - which would
+    // increment the position. However, they at least protect against people calling put() directly on the buffer.
+    private static final ByteBuffer TRUE = BufferType.OFF_HEAP.withContent(new byte[] {1}).asReadOnlyBuffer();
+    private static final ByteBuffer FALSE = BufferType.OFF_HEAP.withContent(new byte[] {0}).asReadOnlyBuffer();
 
     public static final BooleanSerializer instance = new BooleanSerializer();
 

src/java/org/apache/cassandra/utils/memory/MemoryUtil.java

@@ -21,6 +21,7 @@
 import java.nio.Buffer;
 import java.nio.ByteBuffer;
 import java.nio.ByteOrder;
+import java.nio.ReadOnlyBufferException;
 
 import com.sun.jna.Native;
 
@@ -215,8 +216,8 @@ public static void setAttachment(ByteBuffer instance, Object next)
 
     public static ByteBuffer duplicateDirectByteBuffer(ByteBuffer source, ByteBuffer hollowBuffer)
     {
-        assert source.getClass() == DIRECT_BYTE_BUFFER_CLASS || source.getClass() == RO_DIRECT_BYTE_BUFFER_CLASS;
-        unsafe.putLong(hollowBuffer, DIRECT_BYTE_BUFFER_ADDRESS_OFFSET, unsafe.getLong(source, DIRECT_BYTE_BUFFER_ADDRESS_OFFSET));
+        assert source.getClass() == RO_DIRECT_BYTE_BUFFER_CLASS;
+        setAddress(hollowBuffer, unsafe.getLong(source, DIRECT_BYTE_BUFFER_ADDRESS_OFFSET));
         unsafe.putInt(hollowBuffer, DIRECT_BYTE_BUFFER_POSITION_OFFSET, unsafe.getInt(source, DIRECT_BYTE_BUFFER_POSITION_OFFSET));
         unsafe.putInt(hollowBuffer, DIRECT_BYTE_BUFFER_LIMIT_OFFSET, unsafe.getInt(source, DIRECT_BYTE_BUFFER_LIMIT_OFFSET));
         unsafe.putInt(hollowBuffer, DIRECT_BYTE_BUFFER_CAPACITY_OFFSET, unsafe.getInt(source, DIRECT_BYTE_BUFFER_CAPACITY_OFFSET));
@@ -232,7 +233,7 @@ public static ByteBuffer sliceDirectByteBuffer(ByteBuffer source, ByteBuffer hol
 
     public static void setDirectByteBuffer(ByteBuffer instance, long address, int length)
     {
-        unsafe.putLong(instance, DIRECT_BYTE_BUFFER_ADDRESS_OFFSET, address);
+        setAddress(instance, address);
         unsafe.putInt(instance, DIRECT_BYTE_BUFFER_POSITION_OFFSET, 0);
         unsafe.putInt(instance, DIRECT_BYTE_BUFFER_CAPACITY_OFFSET, length);
         unsafe.putInt(instance, DIRECT_BYTE_BUFFER_LIMIT_OFFSET, length);
@@ -243,6 +244,14 @@ public static void setByteBufferCapacity(ByteBuffer instance, int capacity)
         unsafe.putInt(instance, DIRECT_BYTE_BUFFER_CAPACITY_OFFSET, capacity);
     }
 
+    private static void setAddress(ByteBuffer instance, long address)
+    {
+        if (instance.isReadOnly())
+            throw new ReadOnlyBufferException();
+
+        unsafe.putLong(instance, DIRECT_BYTE_BUFFER_ADDRESS_OFFSET, address);
+    }
+
     public static long getLongByByte(long address)
     {
         if (BIG_ENDIAN)

test/unit/org/apache/cassandra/db/SchemaCQLHelperTest.java

@@ -476,6 +476,6 @@ public void testBooleanCompositeKey() throws Throwable
         cfs.getSSTablesForKey("false:true");
 
         execute("insert into %s (t_id, id, ck, nk) VALUES (true, true, false, true)");
-        assertRows(execute("select t_id, id, ck, nk from %s"), row(true, false, false, true), row(true, true, false, true));
+        assertRows(execute("select t_id, id, ck, nk from %s"), row(true, true, false, true), row(true, false, false, true));
     }
 }