Fix row size miscalculation in LazilyCompactedRow

when RangeTombstone is involved.
patch by yukim; reviewed by jbellis for CASSANDRA-7543

CHANGES.txt

@@ -1,6 +1,7 @@
 1.2.19
  * Set correct stream ID on responses when non-Exception Throwables
    are thrown while handling native protocol messages (CASSANDRA-7470)
+ * Fix row size miscalculation in LazilyCompactedRow (CASSANDRA-7543)
 
 1.2.18
  * Support Thrift tables clustering columns on CqlPagingInputFormat (CASSANDRA-7445)

src/java/org/apache/cassandra/db/ColumnIndex.java

@@ -64,6 +64,7 @@ public class ColumnIndex
         private final RangeTombstone.Tracker tombstoneTracker;
         private final OnDiskAtom.Serializer atomSerializer;
         private int atomCount;
+        private long openedMarkerSize = 0;
 
         public Builder(ColumnFamily cf,
                        ByteBuffer key,
@@ -159,7 +160,11 @@ public class ColumnIndex
                 startPosition = endPosition;
                 // TODO: have that use the firstColumn as min + make sure we optimize that on read
                 if (tombstoneTracker != null)
-                    endPosition += tombstoneTracker.writeOpenedMarker(firstColumn, output, atomSerializer);
+                {
+                    long tombstoneSize = tombstoneTracker.writeOpenedMarker(firstColumn, output, atomSerializer);
+                    endPosition += tombstoneSize;
+                    openedMarkerSize += tombstoneSize;
+                }
                 blockSize = 0; // We don't count repeated tombstone marker in the block size, to avoid a situation
                                // where we wouldn't make any progress because a block is filled by said marker
             }
@@ -204,5 +209,10 @@ public class ColumnIndex
             assert result.columnsIndex.size() > 0;
             return result;
         }
+
+        public long getOpenedMarkerSize()
+        {
+            return openedMarkerSize;
+        }
     }
 }

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

@@ -118,7 +118,7 @@ public class LazilyCompactedRow extends AbstractCompactedRow implements Iterable
         DataOutputBuffer clockOut = new DataOutputBuffer();
         DeletionTime.serializer.serialize(emptyColumnFamily.deletionInfo().getTopLevelDeletion(), clockOut);
 
-        long dataSize = clockOut.getLength() + columnSerializedSize;
+        long dataSize = clockOut.getLength() + columnSerializedSize + this.indexBuilder.getOpenedMarkerSize();
         if (logger.isDebugEnabled())
             logger.debug(String.format("clock / column sizes are %s / %s", clockOut.getLength(), columnSerializedSize));
         assert dataSize > 0;

test/unit/org/apache/cassandra/io/LazilyCompactedRowTest.java

@@ -29,6 +29,7 @@ import java.util.Collection;
 import java.util.List;
 import java.util.concurrent.ExecutionException;
 
+import org.junit.Before;
 import org.junit.Test;
 
 import org.apache.cassandra.SchemaLoader;
@@ -93,8 +94,12 @@ public class LazilyCompactedRowTest extends SchemaLoader
             AbstractCompactedRow row2 = iter2.next();
             DataOutputBuffer out1 = new DataOutputBuffer();
             DataOutputBuffer out2 = new DataOutputBuffer();
-            row1.write(out1);
-            row2.write(out2);
+            long size1 = row1.write(out1);
+            long size2 = row2.write(out2);
+
+            // check if written size is the same as reported row size
+            assert size1 == out1.getLength() - 8;
+            assert size2 == out2.getLength() - 8;
 
             File tmpFile1 = File.createTempFile("lcrt1", null);
             File tmpFile2 = File.createTempFile("lcrt2", null);
@@ -127,9 +132,9 @@ public class LazilyCompactedRowTest extends SchemaLoader
             assert columns == in2.readInt();
             for (int i = 0; i < columns; i++)
             {
-                IColumn c1 = (IColumn)cf1.getOnDiskSerializer().deserializeFromSSTable(in1, Descriptor.Version.CURRENT);
-                IColumn c2 = (IColumn)cf2.getOnDiskSerializer().deserializeFromSSTable(in2, Descriptor.Version.CURRENT);
-                assert c1.equals(c2) : c1.getString(cfs.metadata.comparator) + " != " + c2.getString(cfs.metadata.comparator);
+                OnDiskAtom c1 = cf1.getOnDiskSerializer().deserializeFromSSTable(in1, Descriptor.Version.CURRENT);
+                OnDiskAtom c2 = cf2.getOnDiskSerializer().deserializeFromSSTable(in2, Descriptor.Version.CURRENT);
+                assert c1.equals(c2) : "column mismatch";
             }
             // that should be everything
             assert in1.available() == 0;
@@ -166,6 +171,14 @@ public class LazilyCompactedRowTest extends SchemaLoader
         }
     }
 
+    @Before
+    public void setUp()
+    {
+        Table table = Table.open("Keyspace1");
+        ColumnFamilyStore cfs = table.getColumnFamilyStore("Standard1");
+        cfs.clearUnsafe();
+    }
+
     @Test
     public void testOneRow() throws IOException, ExecutionException, InterruptedException, NoSuchAlgorithmException
     {
@@ -314,6 +327,31 @@ public class LazilyCompactedRowTest extends SchemaLoader
         assertBytes(cfs, Integer.MAX_VALUE);
     }
 
+    @Test
+    public void testOneRowWithRangeTombstone() throws Exception
+    {
+        CompactionManager.instance.disableAutoCompaction();
+
+        Table table = Table.open("Keyspace1");
+        ColumnFamilyStore cfs = table.getColumnFamilyStore("Standard1");
+
+        ByteBuffer key = ByteBufferUtil.bytes("k");
+        RowMutation rm = new RowMutation("Keyspace1", key);
+        ColumnFamily cf = rm.addOrGet(cfs.metadata);
+        cf.addColumn(new QueryPath("Standard1", null, ByteBufferUtil.bytes("a")), ByteBuffer.allocate(DatabaseDescriptor.getColumnIndexSize()), 1);
+        cf.addColumn(new QueryPath("Standard1", null, ByteBufferUtil.bytes("c")), ByteBuffer.allocate(DatabaseDescriptor.getColumnIndexSize()), 1);
+        cf.addColumn(new QueryPath("Standard1", null, ByteBufferUtil.bytes("d")), ByteBuffer.allocate(DatabaseDescriptor.getColumnIndexSize()), 1);
+        rm.apply();
+        cfs.forceBlockingFlush();
+
+        rm = new RowMutation("Keyspace1", key);
+        cf = rm.addOrGet(cfs.metadata);
+        cf.addAtom(new RangeTombstone(ByteBufferUtil.bytes("b"), ByteBufferUtil.bytes("d"), 0, (int)(System.currentTimeMillis()/1000)));
+        rm.apply();
+        cfs.forceBlockingFlush();
+
+        assertBytes(cfs, 0);
+    }
 
     private static class LazilyCompactingController extends CompactionController
     {
@@ -322,6 +360,12 @@ public class LazilyCompactedRowTest extends SchemaLoader
             super(cfs, sstables, gcBefore);
         }
 
+        @Override
+        public boolean shouldPurge(DecoratedKey key, long maxDeletionTimestamp)
+        {
+            return false;
+        }
+
         @Override
         public AbstractCompactedRow getCompactedRow(List<SSTableIdentityIterator> rows)
         {
@@ -336,6 +380,12 @@ public class LazilyCompactedRowTest extends SchemaLoader
             super(cfs, sstables, gcBefore);
         }
 
+        @Override
+        public boolean shouldPurge(DecoratedKey key, long maxDeletionTimestamp)
+        {
+            return false;
+        }
+
         @Override
         public AbstractCompactedRow getCompactedRow(List<SSTableIdentityIterator> rows)
         {