Bug fix: NPE on mergeCells edge case

Author: Nitsan Wakart

src/java/org/apache/cassandra/db/compaction/CursorCompactor.java

@@ -619,6 +619,7 @@ private void skipRowsOnStrictLiveness(int rowMergeLimit, boolean isStatic) throw
 
     private DataOutputBuffer tempCellBuffer1 = new DataOutputBuffer();
     private DataOutputBuffer tempCellBuffer2 = new DataOutputBuffer();
+    private final byte[] copyColumnValueBuffer = new byte[4096]; // used to copy cell contents (maybe piecemeal if very large, since we don't have a direct read option)
 
     /**
      * {@link Row.Merger.ColumnDataReducer#getReduced()} <-- applied the delete before reconcile, should not make a difference?
@@ -670,7 +671,9 @@ else if (cellResolution == RIGHT) {
                         if (tempCellBuffer != null)
                             throw new IllegalStateException("tempCellBuffer should be null if cellSource has a value to be read.");
                         tempCellBuffer1.clear();
-                        ssTableCursorWriter.copyCellValue(cellSource, tempCellBuffer1);
+
+                        cellSource.copyCellValue(tempCellBuffer1, copyColumnValueBuffer);
+
                         tempCellBuffer = tempCellBuffer1; // assume cell1 is going to be bigger
                     }
                     else if (tempCellBuffer == null) {
@@ -681,7 +684,8 @@ else if (tempCellBuffer != tempCellBuffer1) {
                         throw new IllegalStateException("tempCellBuffer should be tempCellBuffer1 if cellSource has been read.");
                     }
                     tempCellBuffer2.clear();
-                    if (oCellSource.state() == CELL_VALUE_START) ssTableCursorWriter.copyCellValue(oCellSource, tempCellBuffer2);
+                    if (oCellSource.state() == CELL_VALUE_START)
+                        oCellSource.copyCellValue(tempCellBuffer2, copyColumnValueBuffer);
 
                     int compare = Arrays.compareUnsigned(tempCellBuffer1.getData(), 0, tempCellBuffer1.getLength(), tempCellBuffer2.getData(), 0, tempCellBuffer2.getLength());
                     if (compare >= 0) {
@@ -774,7 +778,7 @@ else if (tempCellBuffer != null) {
                 if (cellSource.state() == CELL_VALUE_START)
                 {
                     if (tempCellBuffer != null) throw new IllegalStateException("Either copied buffer or ready to copy reader, not both.");
-                    ssTableCursorWriter.writeCellValue(cellSource);
+                    ssTableCursorWriter.writeCellValue(cellSource, copyColumnValueBuffer);
                 }
                 else if (tempCellBuffer != null)
                 {

src/java/org/apache/cassandra/io/sstable/SSTableCursorWriter.java

@@ -79,7 +79,6 @@ public class SSTableCursorWriter implements AutoCloseable
     // ROW contents, needed because of the order of writing and the var int fields
     private int rowFlags; // discovered as we go along
     private int rowExtendedFlags;
-    private final byte[] copyColumnValueBuffer = new byte[4096]; // used to copy cell contents (maybe piecemiel if very large, since we don't have a direct read option)
     private final DataOutputBuffer rowHeaderBuffer = new DataOutputBuffer(); // holds the contents between FLAGS and SIZE
     private final DataOutputBuffer rowBuffer = new DataOutputBuffer();
     private final ReusableDeletionTime openMarker = ReusableDeletionTime.live();
@@ -415,14 +414,9 @@ private void writeCellHeader(int cellFlags, ReusableLivenessInfo cellLiveness, D
         metadataCollector.updateCellLiveness(cellLiveness);
     }
 
-    public int writeCellValue(SSTableCursorReader cursor) throws IOException
+    public int writeCellValue(SSTableCursorReader cursor, byte[] copyColumnValueBuffer) throws IOException
     {
-        return copyCellValue(cursor, rowBuffer);
-    }
-
-    public int copyCellValue(SSTableCursorReader cursor, DataOutputBuffer dataOutputBuffer) throws IOException
-    {
-        return cursor.copyCellValue(dataOutputBuffer, copyColumnValueBuffer);
+        return cursor.copyCellValue(rowBuffer, copyColumnValueBuffer);
     }
 
     public void writeCellValue(DataOutputBuffer tempCellBuffer) throws IOException

test/microbench/org/apache/cassandra/test/microbench/sstable/SSTableCursorPipeUtil.java

@@ -33,6 +33,7 @@
 
 public class SSTableCursorPipeUtil
 {
+
     public static void copySSTable(SSTableCursorReader reader, SSTableCursorWriter writer) throws Throwable
     {
         PartitionDescriptor pHeader = new PartitionDescriptor(reader.ssTableReader().getPartitioner().createReusableKey(0));
@@ -108,6 +109,8 @@ public static int copyRangeTombstone(SSTableCursorReader reader, SSTableCursorWr
         return readerState;
     }
 
+    private final static byte[] copyColumnValueBuffer = new byte[4096]; // used to copy cell contents (maybe piecemeal if very large, since we don't have a direct read option)
+
     public static int copyRowAfterDescriptor(SSTableCursorReader reader, SSTableCursorWriter writer, UnfilteredDescriptor unfilteredDescriptor, int readerState, boolean isStatic, boolean updateClusteringMetadata) throws IOException
     {
         writer.writeRowStart(unfilteredDescriptor.livenessInfo(), unfilteredDescriptor.deletionTime(), isStatic);
@@ -128,7 +131,7 @@ public static int copyRowAfterDescriptor(SSTableCursorReader reader, SSTableCurs
             writer.writeCellHeader(cellFlags, cellLiveness, cellCursor.cellColumn);
             if (readerState == CELL_VALUE_START)
             {
-                readerState = writer.writeCellValue(reader);
+                readerState = writer.writeCellValue(reader, copyColumnValueBuffer);
             }
             else if (Cell.Serializer.hasValue(cellFlags))
             {