diff --git a/src/java/org/apache/cassandra/cql3/ResultSet.java b/src/java/org/apache/cassandra/cql3/ResultSet.java
index 82feba697ea3..d93ae6cc7234 100644
--- a/src/java/org/apache/cassandra/cql3/ResultSet.java
+++ b/src/java/org/apache/cassandra/cql3/ResultSet.java
@@ -46,19 +46,27 @@ public class ResultSet
     public static final Codec codec = new Codec();
 
     public final ResultMetadata metadata;
-    public final List<List<ByteBuffer>> rows;
+    public final List<List<byte[]>> rows;
 
     public ResultSet(ResultMetadata resultMetadata)
     {
-        this(resultMetadata, new ArrayList<List<ByteBuffer>>());
+        this(resultMetadata, new ArrayList<List<byte[]>>());
     }
 
-    public ResultSet(ResultMetadata resultMetadata, List<List<ByteBuffer>> rows)
+    public ResultSet(ResultMetadata resultMetadata, List<List<byte[]>> rows)
     {
         this.metadata = resultMetadata;
         this.rows = rows;
     }
 
+    public static ResultSet fromByteBufferRows(ResultMetadata resultMetadata, List<List<ByteBuffer>> bbRows)
+    {
+        List<List<byte[]>> converted = new ArrayList<>(bbRows.size());
+        for (List<ByteBuffer> bbRow : bbRows)
+            converted.add(convertByteBufferList(bbRow));
+        return new ResultSet(resultMetadata, converted);
+    }
+
     public int size()
     {
         return rows.size();
@@ -69,21 +77,40 @@ public boolean isEmpty()
         return size() == 0;
     }
 
-    public void addRow(List<ByteBuffer> row)
+    public void addRow(List<byte[]> row)
     {
         assert row.size() == metadata.valueCount();
         rows.add(row);
     }
 
-    public void addColumnValue(ByteBuffer value)
+    public void addByteBufferRow(List<ByteBuffer> row)
+    {
+        assert row.size() == metadata.valueCount();
+        rows.add(convertByteBufferList(row));
+    }
+
+    private static List<byte[]> convertByteBufferList(List<ByteBuffer> row)
+    {
+        List<byte[]> converted = new ArrayList<>(row.size());
+        for (ByteBuffer bb : row)
+            converted.add(ByteBufferUtil.getArrayUnsafeNullable(bb));
+        return converted;
+    }
+
+    public void addColumnValue(byte[] value)
     {
         if (rows.isEmpty() || lastRow().size() == metadata.valueCount())
-            rows.add(new ArrayList<ByteBuffer>(metadata.valueCount()));
+            rows.add(new ArrayList<byte[]>(metadata.valueCount()));
 
         lastRow().add(value);
     }
 
-    private List<ByteBuffer> lastRow()
+    public void addColumnValue(ByteBuffer value)
+    {
+        addColumnValue(ByteBufferUtil.getArrayUnsafeNullable(value));
+    }
+
+    private List<byte[]> lastRow()
     {
         return rows.get(rows.size() - 1);
     }
@@ -110,11 +137,11 @@ public String toString()
         {
             StringBuilder sb = new StringBuilder();
             sb.append(metadata).append('\n');
-            for (List<ByteBuffer> row : rows)
+            for (List<byte[]> row : rows)
             {
                 for (int i = 0; i < row.size(); i++)
                 {
-                    ByteBuffer v = row.get(i);
+                    byte[] v = row.get(i);
                     if (v == null)
                     {
                         sb.append(" | null");
@@ -123,9 +150,9 @@ public String toString()
                     {
                         sb.append(" | ");
                         if (metadata.flags.contains(Flag.NO_METADATA))
-                            sb.append("0x").append(ByteBufferUtil.bytesToHex(v));
+                            sb.append("0x").append(ByteBufferUtil.bytesToHex(ByteBuffer.wrap(v)));
                         else
-                            sb.append(metadata.names.get(i).type.getString(v));
+                            sb.append(metadata.names.get(i).type.getString(ByteBuffer.wrap(v)));
                     }
                 }
                 sb.append('\n');
@@ -151,12 +178,12 @@ public ResultSet decode(ByteBuf body, ProtocolVersion version)
         {
             ResultMetadata m = ResultMetadata.codec.decode(body, version);
             int rowCount = body.readInt();
-            ResultSet rs = new ResultSet(m, new ArrayList<List<ByteBuffer>>(rowCount));
+            ResultSet rs = new ResultSet(m, new ArrayList<List<byte[]>>(rowCount));
 
             // rows
             int totalValues = rowCount * m.columnCount;
             for (int i = 0; i < totalValues; i++)
-                rs.addColumnValue(CBUtil.readValue(body));
+                rs.addColumnValue(CBUtil.readValueAsBytes(body));
 
             return rs;
         }
@@ -165,7 +192,7 @@ public void encode(ResultSet rs, ByteBuf dest, ProtocolVersion version)
         {
             ResultMetadata.codec.encode(rs.metadata, dest, version);
             dest.writeInt(rs.rows.size());
-            for (List<ByteBuffer> row : rs.rows)
+            for (List<byte[]> row : rs.rows)
             {
                 // Note that we do only want to serialize only the first columnCount values, even if the row
                 // as more: see comment on ResultMetadata.names field.
@@ -177,7 +204,7 @@ public void encode(ResultSet rs, ByteBuf dest, ProtocolVersion version)
         public int encodedSize(ResultSet rs, ProtocolVersion version)
         {
             int size = ResultMetadata.codec.encodedSize(rs.metadata, version) + 4;
-            for (List<ByteBuffer> row : rs.rows)
+            for (List<byte[]> row : rs.rows)
             {
                 for (int i = 0; i < rs.metadata.columnCount; i++)
                     size += CBUtil.sizeOfValue(row.get(i));
diff --git a/src/java/org/apache/cassandra/cql3/UntypedResultSet.java b/src/java/org/apache/cassandra/cql3/UntypedResultSet.java
index e2349062c2d2..c20b1909738f 100644
--- a/src/java/org/apache/cassandra/cql3/UntypedResultSet.java
+++ b/src/java/org/apache/cassandra/cql3/UntypedResultSet.java
@@ -130,7 +130,7 @@ public Iterator<Row> iterator()
         {
             return new AbstractIterator<Row>()
             {
-                final Iterator<List<ByteBuffer>> iter = cqlRows.rows.iterator();
+                final Iterator<List<byte[]>> iter = cqlRows.rows.iterator();
 
                 protected Row computeNext()
                 {
@@ -176,7 +176,7 @@ public Iterator<Row> iterator()
         {
             return new AbstractIterator<Row>()
             {
-                private Iterator<List<ByteBuffer>> currentPage;
+                private Iterator<List<byte[]>> currentPage;
 
                 protected Row computeNext()
                 {
@@ -242,7 +242,7 @@ public Iterator<Row> iterator()
         {
             return new AbstractIterator<Row>()
             {
-                private Iterator<List<ByteBuffer>> currentPage;
+                private Iterator<List<byte[]>> currentPage;
 
                 protected Row computeNext()
                 {
@@ -275,11 +275,27 @@ public static class Row
         @Nonnull
         private final List<ColumnSpecification> columns;
 
-        public Row(@Nonnull List<ColumnSpecification> names, @Nonnull List<ByteBuffer> columns)
+        public Row(@Nonnull List<ColumnSpecification> names, @Nonnull List<byte[]> columns)
         {
             this.columns = ImmutableList.copyOf(names);
             for (int i = 0; i < names.size(); i++)
-                data.put(names.get(i).name.toString(), columns.get(i));
+            {
+                byte[] v = columns.get(i);
+                data.put(names.get(i).name.toString(), v == null ? null : ByteBuffer.wrap(v));
+            }
+        }
+
+        public static Row fromByteBuffers(@Nonnull List<ColumnSpecification> names, @Nonnull List<ByteBuffer> columns)
+        {
+            Row row = new Row(names);
+            for (int i = 0; i < names.size(); i++)
+                row.data.put(names.get(i).name.toString(), columns.get(i));
+            return row;
+        }
+
+        private Row(@Nonnull List<ColumnSpecification> names)
+        {
+            this.columns = ImmutableList.copyOf(names);
         }
 
         public boolean has(String column)
diff --git a/src/java/org/apache/cassandra/cql3/selection/ResultSetBuilder.java b/src/java/org/apache/cassandra/cql3/selection/ResultSetBuilder.java
index 9ab5ca0370cd..6920df373f2b 100644
--- a/src/java/org/apache/cassandra/cql3/selection/ResultSetBuilder.java
+++ b/src/java/org/apache/cassandra/cql3/selection/ResultSetBuilder.java
@@ -73,12 +73,12 @@ public ResultSetBuilder(ResultMetadata metadata, Selectors selectors, boolean un
         this.unmask = unmask;
     }
 
-    private void addSize(List<ByteBuffer> row)
+    private void addSize(List<byte[]> row)
     {
         for (int i=0, isize=row.size(); i<isize; i++)
         {
-            ByteBuffer value = row.get(i);
-            size += value != null ? value.remaining() : 0;
+            byte[] value = row.get(i);
+            size += value != null ? value.length : 0;
         }
     }
 
@@ -107,6 +107,11 @@ public void add(ByteBuffer v)
         inputRow.add(v);
     }
 
+    public void add(byte[] v)
+    {
+        inputRow.add(v);
+    }
+
     public void add(Cell<?> c, long nowInSec)
     {
         inputRow.add(c, nowInSec);
@@ -170,9 +175,9 @@ public ResultSet build()
         return resultSet;
     }
 
-    private List<ByteBuffer> getOutputRow()
+    private List<byte[]> getOutputRow()
     {
-        List<ByteBuffer> row = selectors.getOutputRow();
+        List<byte[]> row = selectors.getOutputRow();
         addSize(row);
         return row;
     }
diff --git a/src/java/org/apache/cassandra/cql3/selection/Selection.java b/src/java/org/apache/cassandra/cql3/selection/Selection.java
index 9d081667204a..2e8510ab2dc1 100644
--- a/src/java/org/apache/cassandra/cql3/selection/Selection.java
+++ b/src/java/org/apache/cassandra/cql3/selection/Selection.java
@@ -17,7 +17,6 @@
  */
 package org.apache.cassandra.cql3.selection;
 
-import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -39,11 +38,13 @@
 import org.apache.cassandra.cql3.functions.Function;
 import org.apache.cassandra.cql3.selection.Selector.InputRow;
 import org.apache.cassandra.db.filter.ColumnFilter;
+import org.apache.cassandra.db.marshal.ByteArrayAccessor;
 import org.apache.cassandra.db.marshal.UTF8Type;
 import org.apache.cassandra.exceptions.InvalidRequestException;
 import org.apache.cassandra.schema.ColumnMetadata;
 import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.transport.ProtocolVersion;
+import org.apache.cassandra.utils.ByteBufferUtil;
 import org.apache.cassandra.utils.JsonUtils;
 
 import static org.apache.cassandra.utils.LocalizeString.toLowerCaseLocalized;
@@ -318,22 +319,22 @@ public String toString()
                           .toString();
     }
 
-    private static List<ByteBuffer> rowToJson(List<ByteBuffer> row,
-                                              ProtocolVersion protocolVersion,
-                                              ResultSet.ResultMetadata metadata,
-                                              List<ColumnMetadata> orderingColumns)
+    private static List<byte[]> rowToJson(List<byte[]> row,
+                                          ProtocolVersion protocolVersion,
+                                          ResultSet.ResultMetadata metadata,
+                                          List<ColumnMetadata> orderingColumns)
     {
-        ByteBuffer[] jsonRow = new ByteBuffer[orderingColumns.size() + 1];
+        byte[][] jsonRow = new byte[orderingColumns.size() + 1][];
         StringBuilder sb = new StringBuilder("{");
         for (int i = 0; i < metadata.names.size(); i++)
         {
             ColumnSpecification spec = metadata.names.get(i);
-            ByteBuffer buffer = row.get(i);
+            byte[] value = row.get(i);
 
             // If it is an ordering column we need to keep it in case we need it for post ordering
             int index = orderingColumns.indexOf(spec);
             if (index >= 0)
-                jsonRow[index + 1] = buffer;
+                jsonRow[index + 1] = value;
 
             // If the column is only used for ordering we can stop here.
             if (i >= metadata.getColumnCount())
@@ -349,14 +350,14 @@ private static List<ByteBuffer> rowToJson(List<ByteBuffer> row,
             sb.append('"');
             sb.append(JsonUtils.quoteAsJsonString(columnName));
             sb.append("\": ");
-            if (buffer == null)
+            if (value == null)
                 sb.append("null");
             else
-                sb.append(spec.type.toJSONString(buffer, protocolVersion));
+                sb.append(spec.type.toJSONString(value, ByteArrayAccessor.instance, protocolVersion));
         }
         sb.append("}");
 
-        jsonRow[0] = UTF8Type.instance.getSerializer().serialize(sb.toString());
+        jsonRow[0] = ByteBufferUtil.getArrayUnsafeNullable(UTF8Type.instance.getSerializer().serialize(sb.toString()));
         return Arrays.asList(jsonRow);
     }
 
@@ -400,14 +401,14 @@ public interface Selectors
          */
         void addInputRow(InputRow input);
 
-        List<ByteBuffer> getOutputRow();
+        List<byte[]> getOutputRow();
 
         void reset();
     }
 
     public static class SimpleSelectors implements Selectors
     {
-        protected List<ByteBuffer> current;
+        protected List<byte[]> current;
 
         @Override
         public void addInputRow(InputRow input)
@@ -416,7 +417,7 @@ public void addInputRow(InputRow input)
         }
 
         @Override
-        public List<ByteBuffer> getOutputRow()
+        public List<byte[]> getOutputRow()
         {
             return current;
         }
@@ -500,7 +501,7 @@ public Selectors newSelectors(QueryOptions options)
             return new SimpleSelectors()
             {
                 @Override
-                public List<ByteBuffer> getOutputRow()
+                public List<byte[]> getOutputRow()
                 {
                     if (isJson)
                         return rowToJson(current, options.getProtocolVersion(), metadata, orderingColumns);
@@ -591,12 +592,12 @@ public boolean hasProcessing()
                     return true;
                 }
 
-                public List<ByteBuffer> getOutputRow()
+                public List<byte[]> getOutputRow()
                 {
-                    List<ByteBuffer> outputRow = new ArrayList<>(selectors.size());
+                    List<byte[]> outputRow = new ArrayList<>(selectors.size());
 
                     for (Selector selector: selectors)
-                        outputRow.add(selector.getOutput(options.getProtocolVersion()));
+                        outputRow.add(selector.getOutputAsBytes(options.getProtocolVersion()));
 
                     return isJson ? rowToJson(outputRow, options.getProtocolVersion(), metadata, orderingColumns) : outputRow;
                 }
diff --git a/src/java/org/apache/cassandra/cql3/selection/Selector.java b/src/java/org/apache/cassandra/cql3/selection/Selector.java
index fa22ea0bb5e2..6c02abc48d16 100644
--- a/src/java/org/apache/cassandra/cql3/selection/Selector.java
+++ b/src/java/org/apache/cassandra/cql3/selection/Selector.java
@@ -45,6 +45,7 @@
 import org.apache.cassandra.schema.Schema;
 import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.transport.ProtocolVersion;
+import org.apache.cassandra.utils.ByteArrayUtil;
 import org.apache.cassandra.utils.ByteBufferUtil;
 
 import static org.apache.cassandra.cql3.statements.RequestValidations.invalidRequest;
@@ -311,7 +312,7 @@ public static final class InputRow
         private final boolean collectWritetimes;
         private final boolean collectTTLs;
 
-        private ByteBuffer[] values;
+        private byte[][] values;
         private RowTimestamps writetimes;
         private RowTimestamps ttls;
         private int index;
@@ -333,7 +334,7 @@ public InputRow(ProtocolVersion protocolVersion,
             this.collectWritetimes = collectWritetimes;
             this.collectTTLs = collectTTLs;
 
-            values = new ByteBuffer[columns.size()];
+            values = new byte[columns.size()][];
             writetimes = initTimestamps(TimestampsType.WRITETIMES, collectWritetimes, columns);
             ttls = initTimestamps(TimestampsType.TTLS, collectTTLs, columns);
         }
@@ -357,6 +358,18 @@ public boolean unmask()
         }
 
         public void add(ByteBuffer v)
+        {
+            values[index] = ByteBufferUtil.getArrayUnsafeNullable(v);
+
+            if (v != null)
+            {
+                writetimes.addNoTimestamp(index);
+                ttls.addNoTimestamp(index);
+            }
+            index++;
+        }
+
+        public void add(byte[] v)
         {
             values[index] = v;
 
@@ -373,7 +386,7 @@ public void add(ColumnData columnData, long nowInSec)
             ColumnMetadata column = columns.get(index);
             if (columnData == null)
             {
-                add(null);
+                add((byte[]) null);
             }
             else
             {
@@ -390,7 +403,7 @@ public void add(ColumnData columnData, long nowInSec)
 
         private void add(Cell<?> c, long nowInSec)
         {
-            values[index] = value(c);
+            values[index] = valueAsArray(c);
             writetimes.addTimestamp(index, c, nowInSec);
             ttls.addTimestamp(index, c, nowInSec);
             index++;
@@ -401,7 +414,7 @@ private void add(ComplexColumnData ccd, long nowInSec)
             AbstractType<?> type = columns.get(index).type;
             if (type.isCollection())
             {
-                values[index] = ((CollectionType<?>) type).serializeForNativeProtocol(ccd.iterator());
+                values[index] = ((CollectionType<?>) type).serializeForNativeProtocolAsByteArrays(ccd.iterator());
 
                 for (Cell<?> cell : ccd)
                 {
@@ -414,7 +427,7 @@ private void add(ComplexColumnData ccd, long nowInSec)
                 UserType udt = (UserType) type;
                 int size = udt.size();
 
-                values[index] = udt.serializeForNativeProtocol(ccd.iterator());
+                values[index] = udt.serializeForNativeProtocolAsByteArrays(ccd.iterator());
 
                 short fieldPosition = 0;
                 for (Cell<?> cell : ccd)
@@ -444,11 +457,11 @@ private void add(ComplexColumnData ccd, long nowInSec)
             index++;
         }
 
-        private <V> ByteBuffer value(Cell<V> c)
+        private <V> byte[] valueAsArray(Cell<V> c)
         {
             return c.isCounterCell()
-                 ? ByteBufferUtil.bytes(CounterContext.instance().total(c.value(), c.accessor()))
-                 : c.buffer();
+                 ? ByteArrayUtil.bytes(CounterContext.instance().total(c.value(), c.accessor()))
+                 : c.accessor().toArray(c.value());
         }
 
         /**
@@ -458,6 +471,12 @@ private <V> ByteBuffer value(Cell<V> c)
          * @return the value of the column with the specified index
          */
         public ByteBuffer getValue(int index)
+        {
+            byte[] v = values[index];
+            return v == null ? null : ByteBuffer.wrap(v);
+        }
+
+        public byte[] getValueAsBytes(int index)
         {
             return values[index];
         }
@@ -477,7 +496,7 @@ public void reset(boolean deep)
             this.ttls = initTimestamps(TimestampsType.TTLS, collectTTLs, columns);
 
             if (deep)
-                values = new ByteBuffer[values.length];
+                values = new byte[values.length][];
         }
 
         /**
@@ -507,7 +526,7 @@ ColumnTimestamps getTtls(int columnIndex)
          *
          * @return the column values as list.
          */
-        public List<ByteBuffer> getValues()
+        public List<byte[]> getValues()
         {
             return Arrays.asList(values);
         }
@@ -530,6 +549,19 @@ public List<ByteBuffer> getValues()
      */
     public abstract ByteBuffer getOutput(ProtocolVersion protocolVersion) throws InvalidRequestException;
 
+    /**
+     * Returns the selector output as a byte array.
+     *
+     * @param protocolVersion protocol version used for serialization
+     * @return the selector output as byte[], or null
+     * @throws InvalidRequestException if a problem occurs while computing the output value
+     */
+    public byte[] getOutputAsBytes(ProtocolVersion protocolVersion) throws InvalidRequestException
+    {
+        ByteBuffer output = getOutput(protocolVersion);
+        return ByteBufferUtil.getArrayUnsafeNullable(output);
+    }
+
     protected ColumnTimestamps getWritetimes(ProtocolVersion protocolVersion)
     {
         throw new UnsupportedOperationException();
diff --git a/src/java/org/apache/cassandra/cql3/selection/SimpleSelector.java b/src/java/org/apache/cassandra/cql3/selection/SimpleSelector.java
index 965f3a49f94d..2a2308b6a890 100644
--- a/src/java/org/apache/cassandra/cql3/selection/SimpleSelector.java
+++ b/src/java/org/apache/cassandra/cql3/selection/SimpleSelector.java
@@ -121,7 +121,7 @@ public ColumnMetadata getColumn()
     public final ColumnMetadata column;
     private final int idx;
     private final ColumnMask.Masker masker;
-    private ByteBuffer current;
+    private byte[] currentBytes;
     private ColumnTimestamps writetimes;
     private ColumnTimestamps ttls;
     private boolean isSet;
@@ -152,16 +152,23 @@ public void addInput(InputRow input) throws InvalidRequestException
             - The input row is for a user with UNMASK permission, indicated by input.unmask()
             - Dynamic data masking is globally disabled
              */
-            ByteBuffer value = input.getValue(idx);
-            current = masker == null || input.unmask() || !DatabaseDescriptor.getDynamicDataMaskingEnabled()
-                      ? value : masker.mask(value);
+            if (masker == null || input.unmask() || !DatabaseDescriptor.getDynamicDataMaskingEnabled())
+                currentBytes = input.getValueAsBytes(idx);
+            else
+                currentBytes = ByteBufferUtil.getArrayUnsafeNullable(masker.mask(input.getValue(idx)));
         }
     }
 
     @Override
     public ByteBuffer getOutput(ProtocolVersion protocolVersion)
     {
-        return current;
+        return currentBytes == null ? null : ByteBuffer.wrap(currentBytes);
+    }
+
+    @Override
+    public byte[] getOutputAsBytes(ProtocolVersion protocolVersion)
+    {
+        return currentBytes;
     }
 
     @Override
@@ -180,7 +187,7 @@ protected ColumnTimestamps getTTLs(ProtocolVersion protocolVersion)
     public void reset()
     {
         isSet = false;
-        current = null;
+        currentBytes = null;
         writetimes = null;
         ttls = null;
     }
diff --git a/src/java/org/apache/cassandra/cql3/statements/DescribeStatement.java b/src/java/org/apache/cassandra/cql3/statements/DescribeStatement.java
index e67f211f1bb2..5f47063809c3 100644
--- a/src/java/org/apache/cassandra/cql3/statements/DescribeStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/DescribeStatement.java
@@ -182,13 +182,13 @@ public ResultMessage executeLocally(QueryState state, QueryOptions options)
         if (pageSize > 0)
             stream = stream.limit(pageSize);
 
-        List<List<ByteBuffer>> rows = stream.map(e -> toRow(e, includeInternalDetails))
-                                            .collect(Collectors.toList());
+        List<List<ByteBuffer>> bbRows = stream.map(e -> toRow(e, includeInternalDetails))
+                                              .collect(Collectors.toList());
 
         ResultSet.ResultMetadata resultMetadata = new ResultSet.ResultMetadata(metadata(state.getClientState()));
-        ResultSet result = new ResultSet(resultMetadata, rows);
+        ResultSet result = ResultSet.fromByteBufferRows(resultMetadata, bbRows);
 
-        if (pageSize > 0 && rows.size() == pageSize)
+        if (pageSize > 0 && bbRows.size() == pageSize)
             result.metadata.setHasMorePages(getPagingState(offset + pageSize, schemaVersion));
 
         return new ResultMessage.Rows(result);
diff --git a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
index 3eab979f3a2b..ed5ecc078761 100644
--- a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
@@ -775,7 +775,11 @@ static ResultSet buildCasResultSet(String ksName,
         boolean success = partition == null;
 
         ResultSet.ResultMetadata metadata = buildCASSuccessMetadata(ksName, tableName);
-        List<List<ByteBuffer>> rows = Collections.singletonList(Collections.singletonList(BooleanType.instance.decompose(success)));
+        List<List<byte[]>> rows = Collections.singletonList(
+            Collections.singletonList(
+                ByteBufferUtil.getArrayUnsafeNullable(BooleanType.instance.decompose(success))
+            )
+        );
 
         ResultSet rs = new ResultSet(metadata, rows);
         return success ? rs : merge(rs, buildCasFailureResultSet(partition, columnsWithConditions, isBatch, options, options.getNowInSeconds(state)));
@@ -793,10 +797,10 @@ else if (right.size() == 0)
         List<ColumnSpecification> specs = new ArrayList<ColumnSpecification>(size);
         specs.addAll(left.metadata.names);
         specs.addAll(right.metadata.names);
-        List<List<ByteBuffer>> rows = new ArrayList<>(right.size());
+        List<List<byte[]>> rows = new ArrayList<>(right.size());
         for (int i = 0; i < right.size(); i++)
         {
-            List<ByteBuffer> row = new ArrayList<ByteBuffer>(size);
+            List<byte[]> row = new ArrayList<byte[]>(size);
             row.addAll(left.rows.get(0));
             row.addAll(right.rows.get(i));
             rows.add(row);
diff --git a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
index 921dda011373..af82a5d23bf0 100644
--- a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
@@ -99,6 +99,8 @@
 import org.apache.cassandra.db.filter.IndexHints;
 import org.apache.cassandra.db.filter.RowFilter;
 import org.apache.cassandra.db.guardrails.Guardrails;
+import org.apache.cassandra.db.marshal.AbstractType;
+import org.apache.cassandra.db.marshal.ByteArrayAccessor;
 import org.apache.cassandra.db.marshal.CompositeType;
 import org.apache.cassandra.db.marshal.Int32Type;
 import org.apache.cassandra.db.partitions.PartitionIterator;
@@ -191,7 +193,7 @@ public class SelectStatement implements CQLStatement.SingleKeyspaceCqlStatement,
     /**
      * The comparator used to orders results when multiple keys are selected (using IN).
      */
-    private final ColumnComparator<List<ByteBuffer>> orderingComparator;
+    private final ColumnComparator<List<byte[]>> orderingComparator;
 
     private final List<Function> functions;
 
@@ -211,7 +213,7 @@ public SelectStatement(TableMetadata table,
                            StatementRestrictions restrictions,
                            boolean isReversed,
                            AggregationSpecification.Factory aggregationSpecFactory,
-                           ColumnComparator<List<ByteBuffer>> orderingComparator,
+                           ColumnComparator<List<byte[]>> orderingComparator,
                            Term limit,
                            Term perPartitionLimit,
                            StatementSource source,
@@ -1226,7 +1228,7 @@ public void processPartition(RowIterator partition, QueryOptions options, Result
                         result.add(keyComponents[def.position()]);
                         break;
                     case CLUSTERING:
-                        result.add(row.clustering().bufferAt(def.position()));
+                        result.add(row.clustering().arrayAt(def.position()));
                         break;
                     case REGULAR:
                         result.add(row.getColumnData(def), nowInSec);
@@ -1260,7 +1262,7 @@ private void orderResults(ResultSet cqlRows, QueryOptions options, ClientState s
         if (cqlRows.size() == 0 || !needsPostQueryOrdering())
             return;
 
-        Comparator<List<ByteBuffer>> comparator = orderingComparator.prepareFor(table, getRowFilter(options, state), options);
+        Comparator<List<byte[]>> comparator = orderingComparator.prepareFor(table, getRowFilter(options, state), options);
         if (comparator != null)
             cqlRows.rows.sort(comparator);
     }
@@ -1366,7 +1368,7 @@ public SelectStatement prepare(ClientState state, boolean forView, VariableSpeci
                        && perPartitionLimit != null,
                        "PER PARTITION LIMIT is not allowed with aggregate queries.");
 
-            ColumnComparator<List<ByteBuffer>> orderingComparator = null;
+            ColumnComparator<List<byte[]>> orderingComparator = null;
             boolean isReversed = false;
 
             if (!orderingColumns.isEmpty())
@@ -1653,9 +1655,9 @@ private void validateGroupByFunction(WithFunction withFunction)
             checkFalse(f.isAggregate(), "Aggregate functions are not supported within the GROUP BY clause, got: %s", f.name());
         }
 
-        private ColumnComparator<List<ByteBuffer>> getOrderingComparator(Selection selection,
-                                                                         StatementRestrictions restrictions,
-                                                                         Map<ColumnMetadata, Ordering> orderingColumns) throws InvalidRequestException
+        private ColumnComparator<List<byte[]>> getOrderingComparator(Selection selection,
+                                                                     StatementRestrictions restrictions,
+                                                                     Map<ColumnMetadata, Ordering> orderingColumns) throws InvalidRequestException
         {
             for (Map.Entry<ColumnMetadata, Ordering> e : orderingColumns.entrySet())
             {
@@ -1670,7 +1672,7 @@ private ColumnComparator<List<ByteBuffer>> getOrderingComparator(Selection selec
                 return null;
 
             List<Integer> idToSort = new ArrayList<>(orderingColumns.size());
-            List<Comparator<ByteBuffer>> sorters = new ArrayList<>(orderingColumns.size());
+            List<AbstractType<?>> sorters = new ArrayList<>(orderingColumns.size());
 
             for (ColumnMetadata orderingColumn : orderingColumns.keySet())
             {
@@ -1796,12 +1798,12 @@ public Parameters(List<Ordering.Raw> orderings,
 
     private static abstract class ColumnComparator<T> implements Comparator<T>
     {
-        protected final int compare(Comparator<ByteBuffer> comparator, ByteBuffer aValue, ByteBuffer bValue)
+        protected final int compare(AbstractType<?> type, byte[] aValue, byte[] bValue)
         {
             if (aValue == null)
                 return bValue == null ? 0 : -1;
 
-            return bValue == null ? 1 : comparator.compare(aValue, bValue);
+            return bValue == null ? 1 : type.compare(aValue, ByteArrayAccessor.instance, bValue, ByteArrayAccessor.instance);
         }
 
         public ColumnComparator<T> reverse()
@@ -1845,24 +1847,24 @@ public int compare(T o1, T o2)
     /**
      * Used in orderResults(...) method when single 'ORDER BY' condition where given
      */
-    private static class SingleColumnComparator extends ColumnComparator<List<ByteBuffer>>
+    private static class SingleColumnComparator extends ColumnComparator<List<byte[]>>
     {
         private final int index;
-        private final Comparator<ByteBuffer> comparator;
+        private final AbstractType<?> comparator;
 
-        public SingleColumnComparator(int columnIndex, Comparator<ByteBuffer> orderer)
+        public SingleColumnComparator(int columnIndex, AbstractType<?> orderer)
         {
             index = columnIndex;
             comparator = orderer;
         }
 
-        public int compare(List<ByteBuffer> a, List<ByteBuffer> b)
+        public int compare(List<byte[]> a, List<byte[]> b)
         {
             return compare(comparator, a.get(index), b.get(index));
         }
     }
 
-    private static class IndexColumnComparator extends ColumnComparator<List<ByteBuffer>>
+    private static class IndexColumnComparator extends ColumnComparator<List<byte[]>>
     {
         private final SingleRestriction restriction;
         private final int columnIndex;
@@ -1880,7 +1882,7 @@ public boolean indexOrdering()
         }
 
         @Override
-        public Comparator<List<ByteBuffer>> prepareFor(TableMetadata table, RowFilter rowFilter, QueryOptions options)
+        public Comparator<List<byte[]>> prepareFor(TableMetadata table, RowFilter rowFilter, QueryOptions options)
         {
             if (table.indexes.isEmpty() || rowFilter.isEmpty())
                 return this;
@@ -1891,11 +1893,19 @@ public Comparator<List<ByteBuffer>> prepareFor(TableMetadata table, RowFilter ro
             Index index = restriction.findSupportingIndex(indexQueryPlan.getIndexes(), IndexHints.NONE);
             assert index != null;
             Comparator<ByteBuffer> comparator = index.getPostQueryOrdering(restriction, options);
-            return (a, b) -> compare(comparator, a.get(columnIndex), b.get(columnIndex));
+            return (a, b) -> {
+                byte[] aVal = a.get(columnIndex);
+                byte[] bVal = b.get(columnIndex);
+                if (aVal == null)
+                    return bVal == null ? 0 : -1;
+                if (bVal == null)
+                    return 1;
+                return comparator.compare(ByteBuffer.wrap(aVal), ByteBuffer.wrap(bVal));
+            };
         }
 
         @Override
-        public int compare(List<ByteBuffer> o1, List<ByteBuffer> o2)
+        public int compare(List<byte[]> o1, List<byte[]> o2)
         {
             throw new UnsupportedOperationException();
         }
@@ -1904,22 +1914,22 @@ public int compare(List<ByteBuffer> o1, List<ByteBuffer> o2)
     /**
      * Used in orderResults(...) method when multiple 'ORDER BY' conditions where given
      */
-    private static class CompositeComparator extends ColumnComparator<List<ByteBuffer>>
+    private static class CompositeComparator extends ColumnComparator<List<byte[]>>
     {
-        private final List<Comparator<ByteBuffer>> orderTypes;
+        private final List<AbstractType<?>> orderTypes;
         private final List<Integer> positions;
 
-        private CompositeComparator(List<Comparator<ByteBuffer>> orderTypes, List<Integer> positions)
+        private CompositeComparator(List<AbstractType<?>> orderTypes, List<Integer> positions)
         {
             this.orderTypes = orderTypes;
             this.positions = positions;
         }
 
-        public int compare(List<ByteBuffer> a, List<ByteBuffer> b)
+        public int compare(List<byte[]> a, List<byte[]> b)
         {
             for (int i = 0; i < positions.size(); i++)
             {
-                Comparator<ByteBuffer> type = orderTypes.get(i);
+                AbstractType<?> type = orderTypes.get(i);
                 int columnPos = positions.get(i);
 
                 int comparison = compare(type, a.get(columnPos), b.get(columnPos));
diff --git a/src/java/org/apache/cassandra/db/ClusteringPrefix.java b/src/java/org/apache/cassandra/db/ClusteringPrefix.java
index 8f304b0052e2..44ccf7be3ab3 100644
--- a/src/java/org/apache/cassandra/db/ClusteringPrefix.java
+++ b/src/java/org/apache/cassandra/db/ClusteringPrefix.java
@@ -313,6 +313,11 @@ default ByteBuffer bufferAt(int i)
         return accessor().toBuffer(get(i));
     }
 
+    default byte[] arrayAt(int i)
+    {
+        return accessor().toArray(get(i));
+    }
+
     default String stringAt(int i, ClusteringComparator comparator)
     {
         return comparator.subtype(i).getString(get(i), accessor());
diff --git a/src/java/org/apache/cassandra/db/marshal/CollectionType.java b/src/java/org/apache/cassandra/db/marshal/CollectionType.java
index 9a4c298d075f..254545ceb254 100644
--- a/src/java/org/apache/cassandra/db/marshal/CollectionType.java
+++ b/src/java/org/apache/cassandra/db/marshal/CollectionType.java
@@ -103,6 +103,8 @@ protected CollectionType(ComparisonType comparisonType, Kind kind)
 
     protected abstract List<ByteBuffer> serializedValues(Iterator<Cell<?>> cells);
 
+    protected abstract List<byte[]> serializedValuesAsByteArrays(Iterator<Cell<?>> cells);
+
     @Override
     public abstract CollectionSerializer<T> getSerializer();
 
@@ -184,6 +186,13 @@ public ByteBuffer serializeForNativeProtocol(Iterator<Cell<?>> cells)
         return getSerializer().pack(values);
     }
 
+    public byte[] serializeForNativeProtocolAsByteArrays(Iterator<Cell<?>> cells)
+    {
+        assert isMultiCell();
+        List<byte[]> values = serializedValuesAsByteArrays(cells);
+        return getSerializer().pack(values, ByteArrayAccessor.instance);
+    }
+
     @Override
     public boolean isCompatibleWith(AbstractType<?> previous)
     {
diff --git a/src/java/org/apache/cassandra/db/marshal/ListType.java b/src/java/org/apache/cassandra/db/marshal/ListType.java
index 7f578fecbd58..83a0023996c7 100644
--- a/src/java/org/apache/cassandra/db/marshal/ListType.java
+++ b/src/java/org/apache/cassandra/db/marshal/ListType.java
@@ -222,6 +222,15 @@ public List<ByteBuffer> serializedValues(Iterator<Cell<?>> cells)
         return bbs;
     }
 
+    public List<byte[]> serializedValuesAsByteArrays(Iterator<Cell<?>> cells)
+    {
+        assert isMultiCell;
+        List<byte[]> bbs = new ArrayList<>();
+        while (cells.hasNext())
+            bbs.add(cells.next().valueAsArray());
+        return bbs;
+    }
+
     @Override
     public Term fromJSONObject(Object parsed) throws MarshalException
     {
diff --git a/src/java/org/apache/cassandra/db/marshal/MapType.java b/src/java/org/apache/cassandra/db/marshal/MapType.java
index 8d206c15723f..2e2b16162a61 100644
--- a/src/java/org/apache/cassandra/db/marshal/MapType.java
+++ b/src/java/org/apache/cassandra/db/marshal/MapType.java
@@ -44,6 +44,7 @@
 import org.apache.cassandra.serializers.MapSerializer;
 import org.apache.cassandra.serializers.MarshalException;
 import org.apache.cassandra.transport.ProtocolVersion;
+import org.apache.cassandra.utils.ByteBufferUtil;
 import org.apache.cassandra.utils.JsonUtils;
 import org.apache.cassandra.utils.Pair;
 import org.apache.cassandra.utils.bytecomparable.ByteComparable.Version;
@@ -315,6 +316,19 @@ public List<ByteBuffer> serializedValues(Iterator<Cell<?>> cells)
         return bbs;
     }
 
+    public List<byte[]> serializedValuesAsByteArrays(Iterator<Cell<?>> cells)
+    {
+        assert isMultiCell;
+        List<byte[]> bbs = new ArrayList<>();
+        while (cells.hasNext())
+        {
+            Cell<?> c = cells.next();
+            bbs.add(ByteBufferUtil.getArrayUnsafeNullable(c.path().get(0)));
+            bbs.add(c.valueAsArray());
+        }
+        return bbs;
+    }
+
     @Override
     public Term fromJSONObject(Object parsed) throws MarshalException
     {
diff --git a/src/java/org/apache/cassandra/db/marshal/SetType.java b/src/java/org/apache/cassandra/db/marshal/SetType.java
index 8a8ae45e0401..107619fa80d5 100644
--- a/src/java/org/apache/cassandra/db/marshal/SetType.java
+++ b/src/java/org/apache/cassandra/db/marshal/SetType.java
@@ -39,6 +39,7 @@
 import org.apache.cassandra.serializers.MarshalException;
 import org.apache.cassandra.serializers.SetSerializer;
 import org.apache.cassandra.transport.ProtocolVersion;
+import org.apache.cassandra.utils.ByteBufferUtil;
 import org.apache.cassandra.utils.JsonUtils;
 import org.apache.cassandra.utils.bytecomparable.ByteComparable;
 import org.apache.cassandra.utils.bytecomparable.ByteSource;
@@ -213,6 +214,14 @@ public List<ByteBuffer> serializedValues(Iterator<Cell<?>> cells)
         return bbs;
     }
 
+    public List<byte[]> serializedValuesAsByteArrays(Iterator<Cell<?>> cells)
+    {
+        List<byte[]> bbs = new ArrayList<>();
+        while (cells.hasNext())
+            bbs.add(ByteBufferUtil.getArrayUnsafeNullable(cells.next().path().get(0)));
+        return bbs;
+    }
+
     @Override
     public Term fromJSONObject(Object parsed) throws MarshalException
     {
diff --git a/src/java/org/apache/cassandra/db/marshal/UserType.java b/src/java/org/apache/cassandra/db/marshal/UserType.java
index df89e14de631..06e41e6abf20 100644
--- a/src/java/org/apache/cassandra/db/marshal/UserType.java
+++ b/src/java/org/apache/cassandra/db/marshal/UserType.java
@@ -236,10 +236,20 @@ public ShortType nameComparator()
     }
 
     public ByteBuffer serializeForNativeProtocol(Iterator<Cell<?>> cells)
+    {
+        return serializeForNativeProtocol(cells, ByteBufferAccessor.instance);
+    }
+
+    public byte[] serializeForNativeProtocolAsByteArrays(Iterator<Cell<?>> cells)
+    {
+        return serializeForNativeProtocol(cells, ByteArrayAccessor.instance);
+    }
+
+    public <V> V serializeForNativeProtocol(Iterator<Cell<?>> cells, ValueAccessor<V> accessor)
     {
         assert isMultiCell;
 
-        List<ByteBuffer> components = new ArrayList<>(size());
+        List<V> components = new ArrayList<>(size());
         while (cells.hasNext())
         {
             Cell<?> cell = cells.next();
@@ -249,14 +259,19 @@ public ByteBuffer serializeForNativeProtocol(Iterator<Cell<?>> cells)
             while (components.size() < fieldPositionOfCell)
                 components.add(null);
 
-            components.add(cell.buffer());
+            components.add(getValue(cell, accessor));
         }
 
         // append trailing nulls for missing cells
         while (components.size() < size())
             components.add(null);
 
-        return pack(components);
+        return pack(components, accessor);
+    }
+
+    private static <V1, V2> V2 getValue(Cell<V1> cell, ValueAccessor<V2> targetAccessor)
+    {
+        return targetAccessor.convert(cell.value(), cell.accessor());
     }
 
     public <V> void validateCell(Cell<V> cell) throws MarshalException
diff --git a/src/java/org/apache/cassandra/transport/CBUtil.java b/src/java/org/apache/cassandra/transport/CBUtil.java
index c66d5416ce0d..a456b6e2a389 100644
--- a/src/java/org/apache/cassandra/transport/CBUtil.java
+++ b/src/java/org/apache/cassandra/transport/CBUtil.java
@@ -458,6 +458,15 @@ public static ByteBuffer readValue(ByteBuf cb)
         return ByteBuffer.wrap(readRawBytes(cb, length));
     }
 
+    public static byte[] readValueAsBytes(ByteBuf cb)
+    {
+        int length = cb.readInt();
+        if (length < 0)
+            return null;
+
+        return readRawBytes(cb, length);
+    }
+
     public static ByteBuffer readValueNoCopy(ByteBuf cb)
     {
         int length = cb.readInt();
diff --git a/src/java/org/apache/cassandra/utils/ByteBufferUtil.java b/src/java/org/apache/cassandra/utils/ByteBufferUtil.java
index 5ba341f4d233..a985362e635f 100644
--- a/src/java/org/apache/cassandra/utils/ByteBufferUtil.java
+++ b/src/java/org/apache/cassandra/utils/ByteBufferUtil.java
@@ -45,6 +45,8 @@
 import java.util.UUID;
 import java.util.stream.Collectors;
 
+import javax.annotation.Nullable;
+
 import net.nicoulaj.compilecommand.annotations.DontInline;
 import net.nicoulaj.compilecommand.annotations.Inline;
 
@@ -227,6 +229,14 @@ public static byte[] getArrayUnsafe(ByteBuffer buffer)
         return getArrayUnsafe(buffer, buffer.position(), buffer.remaining());
     }
 
+    @Nullable
+    public static byte[] getArrayUnsafeNullable(ByteBuffer buffer)
+    {
+        if (buffer == null)
+            return null;
+        return getArrayUnsafe(buffer, buffer.position(), buffer.remaining());
+    }
+
     /**
      * You should almost never use this.  Instead, use the write* methods to avoid copies.
      */
diff --git a/test/burn/org/apache/cassandra/transport/BurnTestUtil.java b/test/burn/org/apache/cassandra/transport/BurnTestUtil.java
index 75ccf4679afe..39fcd71acbba 100644
--- a/test/burn/org/apache/cassandra/transport/BurnTestUtil.java
+++ b/test/burn/org/apache/cassandra/transport/BurnTestUtil.java
@@ -109,7 +109,7 @@ public static ResultMessage.Rows generateRows(int idx, SizeCaps sizeCaps)
             rows.add(row);
         }
 
-        ResultSet resultSet = new ResultSet(new ResultSet.ResultMetadata(columns), rows);
+        ResultSet resultSet = ResultSet.fromByteBufferRows(new ResultSet.ResultMetadata(columns), rows);
         return new ResultMessage.Rows(resultSet);
     }
 
diff --git a/test/burn/org/apache/cassandra/transport/DriverBurnTest.java b/test/burn/org/apache/cassandra/transport/DriverBurnTest.java
index 4f279d33c323..236d9b60dc2c 100644
--- a/test/burn/org/apache/cassandra/transport/DriverBurnTest.java
+++ b/test/burn/org/apache/cassandra/transport/DriverBurnTest.java
@@ -170,11 +170,11 @@ public int encodedSize(QueryMessage queryMessage, ProtocolVersion version)
 
                             for (int i = 0; i < actualRS.size(); i++)
                             {
-                                List<ByteBuffer> expected = expectedRS.result.rows.get(i);
+                                List<byte[]> expected = expectedRS.result.rows.get(i);
                                 Row actual = actualRS.get(i);
 
                                 for (int col = 0; col < expected.size(); col++)
-                                    Assert.assertEquals(actual.getBytes(col), expected.get(col));
+                                    Assert.assertEquals(actual.getBytes(col), ByteBuffer.wrap(expected.get(col)));
                             }
                         }
                         counter++;
diff --git a/test/burn/org/apache/cassandra/transport/SimpleClientBurnTest.java b/test/burn/org/apache/cassandra/transport/SimpleClientBurnTest.java
index c58e6f3ba617..ce72a79f64aa 100644
--- a/test/burn/org/apache/cassandra/transport/SimpleClientBurnTest.java
+++ b/test/burn/org/apache/cassandra/transport/SimpleClientBurnTest.java
@@ -20,7 +20,6 @@
 
 import java.net.InetAddress;
 import java.net.ServerSocket;
-import java.nio.ByteBuffer;
 import java.security.SecureRandom;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -190,11 +189,11 @@ public int encodedSize(QueryMessage queryMessage, ProtocolVersion version)
                             Assert.assertEquals(expected.result.rows.size(), actual.result.rows.size());
                             for (int i = 0; i < expected.result.rows.size(); i++)
                             {
-                                List<ByteBuffer> expectedRow = expected.result.rows.get(i);
-                                List<ByteBuffer> actualRow = actual.result.rows.get(i);
+                                List<byte[]> expectedRow = expected.result.rows.get(i);
+                                List<byte[]> actualRow = actual.result.rows.get(i);
                                 Assert.assertEquals(expectedRow.size(), actualRow.size());
                                 for (int col = 0; col < expectedRow.size(); col++)
-                                    Assert.assertEquals(expectedRow.get(col), actualRow.get(col));
+                                    Assert.assertArrayEquals(expectedRow.get(col), actualRow.get(col));
                             }
                         }
                         counter++;
diff --git a/test/distributed/org/apache/cassandra/distributed/impl/RowUtil.java b/test/distributed/org/apache/cassandra/distributed/impl/RowUtil.java
index 25ef17218790..fec988450334 100644
--- a/test/distributed/org/apache/cassandra/distributed/impl/RowUtil.java
+++ b/test/distributed/org/apache/cassandra/distributed/impl/RowUtil.java
@@ -31,9 +31,11 @@
 
 import org.apache.cassandra.cql3.ColumnSpecification;
 import org.apache.cassandra.cql3.UntypedResultSet;
+import org.apache.cassandra.db.marshal.ByteArrayAccessor;
 import org.apache.cassandra.distributed.api.QueryResults;
 import org.apache.cassandra.distributed.api.SimpleQueryResult;
 import org.apache.cassandra.transport.messages.ResultMessage;
+import org.apache.cassandra.utils.ByteBufferUtil;
 
 public class RowUtil
 {
@@ -79,24 +81,26 @@ public static Object[][] toObjects(ResultMessage.Rows rows, boolean deserialize)
         return toObjects(rows.result.metadata.requestNames(), rows.result.rows, deserialize);
     }
 
-    public static Object[][] toObjects(List<ColumnSpecification> specs, List<List<ByteBuffer>> rows)
+    public static Object[][] toObjects(List<ColumnSpecification> specs, List<List<byte[]>> rows)
     {
         return toObjects(specs, rows, true);
     }
 
-    public static Object[][] toObjects(List<ColumnSpecification> specs, List<List<ByteBuffer>> rows, boolean deserialize)
+    public static Object[][] toObjects(List<ColumnSpecification> specs, List<List<byte[]>> rows, boolean deserialize)
     {
         Object[][] result = new Object[rows.size()][];
         for (int i = 0; i < rows.size(); i++)
         {
-            List<ByteBuffer> row = rows.get(i);
+            List<byte[]> row = rows.get(i);
             result[i] = new Object[specs.size()];
             for (int j = 0; j < specs.size(); j++)
             {
-                ByteBuffer bb = row.get(j);
+                byte[] bytes = row.get(j);
 
-                if (bb != null)
-                    result[i][j] = deserialize ? specs.get(j).type.getSerializer().deserialize(bb) : bb;
+                if (bytes != null)
+                {
+                    result[i][j] = deserialize ? specs.get(j).type.getSerializer().deserialize(bytes, ByteArrayAccessor.instance) : ByteBuffer.wrap(bytes);
+                }
             }
         }
         return result;
@@ -149,20 +153,21 @@ public static Iterator<Object[]> toIter(ResultMessage.Rows rows)
 
     public static Iterator<Object[]> toIter(List<ColumnSpecification> columnSpecs, Iterator<UntypedResultSet.Row> rs)
     {
-        Iterator<List<ByteBuffer>> iter = Iterators.transform(rs,
-                                                              (row) -> {
-                                                                  List<ByteBuffer> bbs = new ArrayList<>(columnSpecs.size());
-                                                                  for (int i = 0; i < columnSpecs.size(); i++)
-                                                                  {
-                                                                      ColumnSpecification columnSpec = columnSpecs.get(i);
-                                                                      bbs.add(row.getBytes(columnSpec.name.toString()));
-                                                                  }
-                                                                  return bbs;
-                                                              });
+        Iterator<List<byte[]>> iter = Iterators.transform(rs,
+                                                          (row) -> {
+                                                              List<byte[]> bbs = new ArrayList<>(columnSpecs.size());
+                                                              for (int i = 0; i < columnSpecs.size(); i++)
+                                                              {
+                                                                  ColumnSpecification columnSpec = columnSpecs.get(i);
+                                                                  ByteBuffer bb = row.getBytes(columnSpec.name.toString());
+                                                                  bbs.add(bb == null ? null : ByteBufferUtil.getArrayUnsafe(bb));
+                                                              }
+                                                              return bbs;
+                                                          });
         return toIterInternal(columnSpecs, Lists.newArrayList(iter));
     }
 
-    private static Iterator<Object[]> toIterInternal(List<ColumnSpecification> columnSpecs, List<List<ByteBuffer>> rs)
+    private static Iterator<Object[]> toIterInternal(List<ColumnSpecification> columnSpecs, List<List<byte[]>> rs)
     {
         return Iterators.transform(rs.iterator(),
                                    (row) -> {
@@ -170,10 +175,10 @@ private static Iterator<Object[]> toIterInternal(List<ColumnSpecification> colum
                                        for (int i = 0; i < columnSpecs.size(); i++)
                                        {
                                            ColumnSpecification columnSpec = columnSpecs.get(i);
-                                           ByteBuffer bb = row.get(i);
+                                           byte[] bytes = row.get(i);
 
-                                           if (bb != null)
-                                               objectRow[i] = columnSpec.type.getSerializer().deserialize(bb);
+                                           if (bytes != null)
+                                               objectRow[i] = columnSpec.type.getSerializer().deserialize(bytes, ByteArrayAccessor.instance);
 
                                        }
                                        return objectRow;
diff --git a/test/unit/org/apache/cassandra/cql3/PreparedStatementsTest.java b/test/unit/org/apache/cassandra/cql3/PreparedStatementsTest.java
index 95b4f830b124..d92e75e775cd 100644
--- a/test/unit/org/apache/cassandra/cql3/PreparedStatementsTest.java
+++ b/test/unit/org/apache/cassandra/cql3/PreparedStatementsTest.java
@@ -617,8 +617,8 @@ private ResultMessage.Prepared verifyMetadataFlagsWithLWTsSelect(SimpleClient si
                      rows.result.metadata.names.stream().map(cs -> cs.name.toString()).collect(Collectors.toList()));
         assertEquals(1,
                      rows.result.size());
-        assertEquals(expectedRow,
-                     rows.result.rows.get(0));
+        assertEqualRows(expectedRow,
+                        rows.result.rows.get(0));
 
         if (resultFlags.contains(org.apache.cassandra.cql3.ResultSet.Flag.METADATA_CHANGED))
             prepSelect = prepSelect.withResultMetadata(rows.result.metadata);
@@ -644,8 +644,8 @@ private void verifyMetadataFlagsWithLWTsUpdate(SimpleClient simpleClient,
                      rows.result.metadata.names.stream().map(cs -> cs.name.toString()).collect(Collectors.toList()));
         assertEquals(1,
                      rows.result.size());
-        assertEquals(expectedRow,
-                     rows.result.rows.get(0));
+        assertEqualRows(expectedRow,
+                        rows.result.rows.get(0));
     }
 
     @Test
@@ -785,6 +785,15 @@ public void testPrepareWithAccordCurrent()
         testPrepareWithAccord(ProtocolVersion.CURRENT);
     }
 
+    private static void assertEqualRows(List<ByteBuffer> expectedRow, List<byte[]> actualRow)
+    {
+        assertEquals(expectedRow.size(), actualRow.size());
+
+        for (int i = 0; i < expectedRow.size(); i++) {
+            assertEquals(expectedRow.get(i), actualRow.get(i) != null ? ByteBuffer.wrap(actualRow.get(i)) : null);
+        }
+    }
+
     private void testPrepareWithAccord(ProtocolVersion version)
     {
         int maxAttempts = 3;
diff --git a/test/unit/org/apache/cassandra/cql3/UntypedResultSetTest.java b/test/unit/org/apache/cassandra/cql3/UntypedResultSetTest.java
index a99ef45ea81b..87aa929cbe5b 100644
--- a/test/unit/org/apache/cassandra/cql3/UntypedResultSetTest.java
+++ b/test/unit/org/apache/cassandra/cql3/UntypedResultSetTest.java
@@ -85,7 +85,7 @@ private static Gen<UntypedResultSet.Row> row()
                 AbstractTypeGenerators.TypeSupport<?> support = AbstractTypeGenerators.getTypeSupport(columns.get(i).type);
                 data.add(fromQT(support.bytesGen()).next(rs));
             }
-            return new UntypedResultSet.Row(columns, data);
+            return UntypedResultSet.Row.fromByteBuffers(columns, data);
         });
     }
 
@@ -99,7 +99,7 @@ private static Gen<ResultSet> resultSet()
             for (int i = 0; i < numRows; i++)
             {
                 List<ByteBuffer> row = dataGens.stream().map(g -> g.next(rs)).collect(Collectors.toList());
-                result.addRow(row);
+                result.addByteBufferRow(row);
             }
             return result;
         };
diff --git a/test/unit/org/apache/cassandra/db/guardrails/AbstractGenerationalTest.java b/test/unit/org/apache/cassandra/db/guardrails/AbstractGenerationalTest.java
index e138601ddfa9..35cb214d8bc6 100644
--- a/test/unit/org/apache/cassandra/db/guardrails/AbstractGenerationalTest.java
+++ b/test/unit/org/apache/cassandra/db/guardrails/AbstractGenerationalTest.java
@@ -18,13 +18,13 @@
 
 package org.apache.cassandra.db.guardrails;
 
-import java.nio.ByteBuffer;
 import java.util.List;
 import java.util.Map;
 
 import org.junit.After;
 import org.junit.Ignore;
 
+import org.apache.cassandra.db.marshal.ByteArrayAccessor;
 import org.apache.cassandra.db.marshal.UTF8Type;
 import org.apache.cassandra.transport.Message;
 import org.apache.cassandra.transport.messages.ResultMessage;
@@ -92,10 +92,10 @@ private String extractOne(ResultMessage resultMessage, String columnName)
         assertEquals(1, rows.result.metadata.names.size());
         assertEquals(UTF8Type.instance.asCQL3Type(), rows.result.metadata.names.get(0).type.asCQL3Type());
         assertEquals(columnName, rows.result.metadata.names.get(0).name.toString());
-        List<ByteBuffer> byteBuffer = rows.result.rows.get(0);
-        assertNotNull(byteBuffer);
-        assertEquals(1, byteBuffer.size());
-        return UTF8Type.instance.getSerializer().deserialize(byteBuffer.get(0));
+        List<byte[]> byteArrayRow = rows.result.rows.get(0);
+        assertNotNull(byteArrayRow);
+        assertEquals(1, byteArrayRow.size());
+        return UTF8Type.instance.getSerializer().deserialize(byteArrayRow.get(0), ByteArrayAccessor.instance);
     }
 
     protected Pair<String, String> extractPasswordAndRoleName(ResultMessage resultMessage)
@@ -112,11 +112,11 @@ protected Pair<String, String> extractPasswordAndRoleName(ResultMessage resultMe
         assertEquals(UTF8Type.instance.asCQL3Type(), rows.result.metadata.names.get(1).type.asCQL3Type());
         assertEquals("generated_password", rows.result.metadata.names.get(0).name.toString());
         assertEquals("generated_role_name", rows.result.metadata.names.get(1).name.toString());
-        List<ByteBuffer> row = rows.result.rows.get(0);
+        List<byte[]> row = rows.result.rows.get(0);
         assertNotNull(row);
         assertEquals(2, row.size());
-        String password = UTF8Type.instance.getSerializer().deserialize(row.get(0));
-        String roleName = UTF8Type.instance.getSerializer().deserialize(row.get(1));
+        String password = UTF8Type.instance.getSerializer().deserialize(row.get(0), ByteArrayAccessor.instance);
+        String roleName = UTF8Type.instance.getSerializer().deserialize(row.get(1), ByteArrayAccessor.instance);
         return Pair.create(password, roleName);
     }
 }
diff --git a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlOperation.java b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlOperation.java
index 1fe49bdf9c9a..4006170c9f62 100644
--- a/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlOperation.java
+++ b/tools/stress/src/org/apache/cassandra/stress/operations/predefined/CqlOperation.java
@@ -434,10 +434,13 @@ public ByteBuffer[][] apply(ResultMessage result)
                     ByteBuffer[][] r = new ByteBuffer[rows.result.size()][];
                     for (int i = 0 ; i < r.length ; i++)
                     {
-                        List<ByteBuffer> row = rows.result.rows.get(i);
+                        List<byte[]> row = rows.result.rows.get(i);
                         r[i] = new ByteBuffer[row.size()];
                         for (int j = 0 ; j < row.size() ; j++)
-                            r[i][j] = row.get(j);
+                        {
+                            byte[] v = row.get(j);
+                            r[i][j] = v == null ? null : ByteBuffer.wrap(v);
+                        }
                     }
                     return r;
                 }
@@ -484,7 +487,7 @@ public byte[][] apply(ResultMessage result)
                         ResultMessage.Rows rows = ((ResultMessage.Rows) result);
                         byte[][] r = new byte[rows.result.size()][];
                         for (int i = 0 ; i < r.length ; i++)
-                            r[i] = rows.result.rows.get(i).get(0).array();
+                            r[i] = rows.result.rows.get(i).get(0);
                         return r;
                     }
                     return null;