Repository: cassandra Updated Branches: refs/heads/cassandra-2.0 dc7d5a0a4 -> 93c27549a refs/heads/cassandra-2.1 4b2e946f6 -> 367a3696b refs/heads/cassandra-2.1.0 6b988fddd -> 545e2e1cb refs/heads/trunk 287bea327 -> e69fee52b
Fix row size miscalculation in LazilyCompactedRow when RangeTombstone is involved. patch by yukim; reviewed by jbellis for CASSANDRA-7543 Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/192596ad Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/192596ad Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/192596ad Branch: refs/heads/cassandra-2.1.0 Commit: 192596ad75f2f45098ca10a5983b9fef0585fe87 Parents: 0d90b03 Author: Yuki Morishita <[email protected]> Authored: Wed Jul 16 15:57:31 2014 -0500 Committer: Yuki Morishita <[email protected]> Committed: Wed Jul 16 15:57:31 2014 -0500 ---------------------------------------------------------------------- CHANGES.txt | 1 + .../org/apache/cassandra/db/ColumnIndex.java | 12 +++- .../db/compaction/LazilyCompactedRow.java | 2 +- .../cassandra/io/LazilyCompactedRowTest.java | 60 ++++++++++++++++++-- 4 files changed, 68 insertions(+), 7 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/cassandra/blob/192596ad/CHANGES.txt ---------------------------------------------------------------------- diff --git a/CHANGES.txt b/CHANGES.txt index 2951fed..676c4e5 100644 --- a/CHANGES.txt +++ b/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) http://git-wip-us.apache.org/repos/asf/cassandra/blob/192596ad/src/java/org/apache/cassandra/db/ColumnIndex.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/ColumnIndex.java b/src/java/org/apache/cassandra/db/ColumnIndex.java index 23e3c00..b152f30 100644 --- a/src/java/org/apache/cassandra/db/ColumnIndex.java +++ b/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; + } } } http://git-wip-us.apache.org/repos/asf/cassandra/blob/192596ad/src/java/org/apache/cassandra/db/compaction/LazilyCompactedRow.java ---------------------------------------------------------------------- diff --git a/src/java/org/apache/cassandra/db/compaction/LazilyCompactedRow.java b/src/java/org/apache/cassandra/db/compaction/LazilyCompactedRow.java index 9a03598..d9f753c 100644 --- a/src/java/org/apache/cassandra/db/compaction/LazilyCompactedRow.java +++ b/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; http://git-wip-us.apache.org/repos/asf/cassandra/blob/192596ad/test/unit/org/apache/cassandra/io/LazilyCompactedRowTest.java ---------------------------------------------------------------------- diff --git a/test/unit/org/apache/cassandra/io/LazilyCompactedRowTest.java b/test/unit/org/apache/cassandra/io/LazilyCompactedRowTest.java index 7f9ca18..eeaa3f2 100644 --- a/test/unit/org/apache/cassandra/io/LazilyCompactedRowTest.java +++ b/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 { @@ -323,6 +361,12 @@ public class LazilyCompactedRowTest extends SchemaLoader } @Override + public boolean shouldPurge(DecoratedKey key, long maxDeletionTimestamp) + { + return false; + } + + @Override public AbstractCompactedRow getCompactedRow(List<SSTableIdentityIterator> rows) { return new LazilyCompactedRow(this, rows); @@ -337,6 +381,12 @@ public class LazilyCompactedRowTest extends SchemaLoader } @Override + public boolean shouldPurge(DecoratedKey key, long maxDeletionTimestamp) + { + return false; + } + + @Override public AbstractCompactedRow getCompactedRow(List<SSTableIdentityIterator> rows) { return new PrecompactedRow(this, rows);
