You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by ty...@apache.org on 2015/04/30 20:16:37 UTC
[1/3] cassandra git commit: Handle corrupted compression chunks in
scrub
Repository: cassandra
Updated Branches:
refs/heads/trunk 3bfe4b6ae -> 49b089893
Handle corrupted compression chunks in scrub
Patch by Stefania Alborghetti; reviewed by Tyler Hobbs for
CASSANDRA-9140
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/593a7257
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/593a7257
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/593a7257
Branch: refs/heads/trunk
Commit: 593a7257b5d5243c4ceb75b5417fb05e8a98cc90
Parents: a85e731
Author: Stefania Alborghetti <st...@datastax.com>
Authored: Thu Apr 30 13:02:23 2015 -0500
Committer: Tyler Hobbs <ty...@gmail.com>
Committed: Thu Apr 30 13:02:23 2015 -0500
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../cassandra/db/compaction/Scrubber.java | 131 ++++++++++++----
.../unit/org/apache/cassandra/SchemaLoader.java | 17 ++-
.../unit/org/apache/cassandra/db/ScrubTest.java | 153 +++++++++++++++++--
4 files changed, 256 insertions(+), 46 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/593a7257/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 9f90cc2..4e7a5d0 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
2.0.15:
+ * Allow scrub to handle corrupted compressed chunks (CASSANDRA-9140)
* Fix assertion error when resetlocalschema is run during repair (CASSANDRA-9249)
* Disable single sstable tombstone compactions for DTCS by default (CASSANDRA-9234)
* Do more agressive ttl expiration checks to be able to
http://git-wip-us.apache.org/repos/asf/cassandra/blob/593a7257/src/java/org/apache/cassandra/db/compaction/Scrubber.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/compaction/Scrubber.java b/src/java/org/apache/cassandra/db/compaction/Scrubber.java
index 6a61e56..1752b21 100644
--- a/src/java/org/apache/cassandra/db/compaction/Scrubber.java
+++ b/src/java/org/apache/cassandra/db/compaction/Scrubber.java
@@ -21,6 +21,7 @@ import java.nio.ByteBuffer;
import java.io.*;
import java.util.*;
+import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Throwables;
import org.apache.cassandra.db.*;
@@ -53,6 +54,11 @@ public class Scrubber implements Closeable
private int badRows;
private int emptyRows;
+ private ByteBuffer currentIndexKey;
+ private ByteBuffer nextIndexKey;
+ long currentRowPositionFromIndex;
+ long nextRowPositionFromIndex;
+
private final OutputHandler outputHandler;
private static final Comparator<Row> rowComparator = new Comparator<Row>()
@@ -99,6 +105,9 @@ public class Scrubber implements Closeable
: sstable.openDataReader(CompactionManager.instance.getRateLimiter());
this.indexFile = RandomAccessReader.open(new File(sstable.descriptor.filenameFor(Component.PRIMARY_INDEX)));
this.scrubInfo = new ScrubInfo(dataFile, sstable);
+
+ this.currentRowPositionFromIndex = 0;
+ this.nextRowPositionFromIndex = 0;
}
public void scrub()
@@ -106,7 +115,7 @@ public class Scrubber implements Closeable
outputHandler.output(String.format("Scrubbing %s (%s bytes)", sstable, dataFile.length()));
try
{
- ByteBuffer nextIndexKey = ByteBufferUtil.readWithShortLength(indexFile);
+ nextIndexKey = ByteBufferUtil.readWithShortLength(indexFile);
{
// throw away variable so we don't have a side effect in the assert
long firstRowPositionFromIndex = RowIndexEntry.serializer.deserialize(indexFile, sstable.descriptor.version).position;
@@ -122,6 +131,7 @@ public class Scrubber implements Closeable
{
if (scrubInfo.isStopRequested())
throw new CompactionInterruptedException(scrubInfo.getCompactionInfo());
+
long rowStart = dataFile.getFilePointer();
outputHandler.debug("Reading row at " + rowStart);
@@ -142,29 +152,19 @@ public class Scrubber implements Closeable
// check for null key below
}
- ByteBuffer currentIndexKey = nextIndexKey;
- long nextRowPositionFromIndex;
- try
- {
- nextIndexKey = indexFile.isEOF() ? null : ByteBufferUtil.readWithShortLength(indexFile);
- nextRowPositionFromIndex = indexFile.isEOF()
- ? dataFile.length()
- : RowIndexEntry.serializer.deserialize(indexFile, sstable.descriptor.version).position;
- }
- catch (Throwable th)
- {
- outputHandler.warn("Error reading index file", th);
- nextIndexKey = null;
- nextRowPositionFromIndex = dataFile.length();
- }
+ updateIndexKey();
long dataStart = dataFile.getFilePointer();
- long dataStartFromIndex = currentIndexKey == null
- ? -1
- : rowStart + 2 + currentIndexKey.remaining();
- if (sstable.descriptor.version.hasRowSizeAndColumnCount)
- dataStartFromIndex += 8;
- long dataSizeFromIndex = nextRowPositionFromIndex - dataStartFromIndex;
+ long dataStartFromIndex = -1;
+ long dataSizeFromIndex = -1;
+ if (currentIndexKey != null)
+ {
+ dataStartFromIndex = currentRowPositionFromIndex + 2 + currentIndexKey.remaining();
+ if (sstable.descriptor.version.hasRowSizeAndColumnCount)
+ dataStartFromIndex += 8;
+
+ dataSizeFromIndex = nextRowPositionFromIndex - dataStartFromIndex;
+ }
if (!sstable.descriptor.version.hasRowSizeAndColumnCount)
{
@@ -186,8 +186,21 @@ public class Scrubber implements Closeable
{
if (key == null)
throw new IOError(new IOException("Unable to read row key from data file"));
+
+ if (!key.key.equals(currentIndexKey))
+ {
+ throw new IOError(new IOException(String.format("Key from data file (%s) does not match key from index file (%s)",
+ ByteBufferUtil.bytesToHex(key.key), ByteBufferUtil.bytesToHex(currentIndexKey))));
+ }
+
if (dataSize > dataFile.length())
- throw new IOError(new IOException("Impossible row size " + dataSize));
+ throw new IOError(new IOException("Impossible row size (greater than file length): " + dataSize));
+
+ if (dataStart != dataStartFromIndex)
+ outputHandler.warn(String.format("Data file row position %d differs from index file row position %d", dataStart, dataSizeFromIndex));
+
+ if (dataSize != dataSizeFromIndex)
+ outputHandler.warn(String.format("Data file row size %d differs from index file row size %d", dataSize, dataSizeFromIndex));
SSTableIdentityIterator atoms = new SSTableIdentityIterator(sstable, dataFile, key, dataSize, true);
if (prevKey != null && prevKey.compareTo(key) > 0)
@@ -201,9 +214,8 @@ public class Scrubber implements Closeable
emptyRows++;
else
goodRows++;
+
prevKey = key;
- if (!key.key.equals(currentIndexKey) || dataStart != dataStartFromIndex)
- outputHandler.warn("Index file contained a different key or row size; using key from data file");
}
catch (Throwable th)
{
@@ -219,6 +231,8 @@ public class Scrubber implements Closeable
key = sstable.partitioner.decorateKey(currentIndexKey);
try
{
+ dataFile.seek(dataStartFromIndex);
+
SSTableIdentityIterator atoms = new SSTableIdentityIterator(sstable, dataFile, key, dataSize, true);
if (prevKey != null && prevKey.compareTo(key) > 0)
{
@@ -231,6 +245,7 @@ public class Scrubber implements Closeable
emptyRows++;
else
goodRows++;
+
prevKey = key;
}
catch (Throwable th2)
@@ -240,8 +255,8 @@ public class Scrubber implements Closeable
outputHandler.warn("Retry failed too. Skipping to next row (retry's stacktrace follows)", th2);
writer.resetAndTruncate();
- dataFile.seek(nextRowPositionFromIndex);
badRows++;
+ seekToNextRow();
}
}
else
@@ -249,9 +264,9 @@ public class Scrubber implements Closeable
throwIfCommutative(key, th);
outputHandler.warn("Row starting at position " + dataStart + " is unreadable; skipping to next");
- if (currentIndexKey != null)
- dataFile.seek(nextRowPositionFromIndex);
badRows++;
+ if (currentIndexKey != null)
+ seekToNextRow();
}
}
}
@@ -294,6 +309,45 @@ public class Scrubber implements Closeable
}
}
+ private void updateIndexKey()
+ {
+ currentIndexKey = nextIndexKey;
+ currentRowPositionFromIndex = nextRowPositionFromIndex;
+ try
+ {
+ nextIndexKey = indexFile.isEOF() ? null : ByteBufferUtil.readWithShortLength(indexFile);
+ nextRowPositionFromIndex = indexFile.isEOF()
+ ? dataFile.length()
+ : RowIndexEntry.serializer.deserialize(indexFile, sstable.descriptor.version).position;
+ }
+ catch (Throwable th)
+ {
+ outputHandler.warn("Error reading index file", th);
+ nextIndexKey = null;
+ nextRowPositionFromIndex = dataFile.length();
+ }
+ }
+
+ private void seekToNextRow()
+ {
+ while(nextRowPositionFromIndex < dataFile.length())
+ {
+ try
+ {
+ dataFile.seek(nextRowPositionFromIndex);
+ return;
+ }
+ catch (Throwable th)
+ {
+ throwIfFatal(th);
+ outputHandler.warn(String.format("Failed to seek to next row position %d", nextRowPositionFromIndex), th);
+ badRows++;
+ }
+
+ updateIndexKey();
+ }
+ }
+
private void saveOutOfOrderRow(DecoratedKey prevKey, DecoratedKey key, SSTableIdentityIterator atoms)
{
// TODO bitch if the row is too large? if it is there's not much we can do ...
@@ -389,4 +443,25 @@ public class Scrubber implements Closeable
return false;
}
}
+
+ @VisibleForTesting
+ public ScrubResult scrubWithResult()
+ {
+ scrub();
+ return new ScrubResult(this);
+ }
+
+ public static final class ScrubResult
+ {
+ public final int goodRows;
+ public final int badRows;
+ public final int emptyRows;
+
+ public ScrubResult(Scrubber scrubber)
+ {
+ this.goodRows = scrubber.goodRows;
+ this.badRows = scrubber.badRows;
+ this.emptyRows = scrubber.emptyRows;
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/593a7257/test/unit/org/apache/cassandra/SchemaLoader.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/SchemaLoader.java b/test/unit/org/apache/cassandra/SchemaLoader.java
index 7dea52c..6b34b9a 100644
--- a/test/unit/org/apache/cassandra/SchemaLoader.java
+++ b/test/unit/org/apache/cassandra/SchemaLoader.java
@@ -60,6 +60,11 @@ public class SchemaLoader
public static void loadSchema(boolean withOldCfIds) throws IOException, ConfigurationException
{
+ loadSchema(withOldCfIds, null);
+ }
+
+ public static void loadSchema(boolean withOldCfIds, Integer compressionChunkLength) throws IOException, ConfigurationException
+ {
// Cleanup first
cleanupAndLeaveDirs();
@@ -78,7 +83,7 @@ public class SchemaLoader
startGossiper();
// if you're messing with low-level sstable stuff, it can be useful to inject the schema directly
// Schema.instance.load(schemaDefinition(withOldCfIds));
- for (KSMetaData ksm : schemaDefinition(withOldCfIds))
+ for (KSMetaData ksm : schemaDefinition(withOldCfIds, compressionChunkLength))
MigrationManager.announceNewKeyspace(ksm);
}
@@ -93,7 +98,7 @@ public class SchemaLoader
Gossiper.instance.stop();
}
- public static Collection<KSMetaData> schemaDefinition(boolean withOldCfIds) throws ConfigurationException
+ public static Collection<KSMetaData> schemaDefinition(boolean withOldCfIds, Integer compressionChunkLength) throws ConfigurationException
{
List<KSMetaData> schema = new ArrayList<KSMetaData>();
@@ -344,7 +349,7 @@ public class SchemaLoader
if (Boolean.parseBoolean(System.getProperty("cassandra.test.compression", "false")))
- useCompression(schema);
+ useCompression(schema, compressionChunkLength);
return schema;
}
@@ -368,13 +373,15 @@ public class SchemaLoader
}});
}
- private static void useCompression(List<KSMetaData> schema)
+ private static void useCompression(List<KSMetaData> schema, Integer chunkLength) throws ConfigurationException
{
for (KSMetaData ksm : schema)
{
for (CFMetaData cfm : ksm.cfMetaData().values())
{
- cfm.compressionParameters(new CompressionParameters(SnappyCompressor.instance));
+ cfm.compressionParameters(new CompressionParameters(SnappyCompressor.instance,
+ chunkLength,
+ Collections.<String, String>emptyMap()));
}
}
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/593a7257/test/unit/org/apache/cassandra/db/ScrubTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/db/ScrubTest.java b/test/unit/org/apache/cassandra/db/ScrubTest.java
index 632ce1c..08237a4 100644
--- a/test/unit/org/apache/cassandra/db/ScrubTest.java
+++ b/test/unit/org/apache/cassandra/db/ScrubTest.java
@@ -21,6 +21,7 @@ package org.apache.cassandra.db;
*/
import java.io.*;
+import java.nio.ByteBuffer;
import java.util.Collections;
import java.util.HashSet;
import java.util.Iterator;
@@ -31,13 +32,14 @@ import java.util.concurrent.ExecutionException;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.db.compaction.OperationType;
import org.apache.cassandra.exceptions.RequestExecutionException;
+import org.apache.cassandra.io.compress.CompressionMetadata;
import org.apache.cassandra.utils.UUIDGen;
import org.apache.commons.lang3.StringUtils;
+import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.apache.cassandra.OrderedJUnit4ClassRunner;
-import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.cql3.UntypedResultSet;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.Util;
@@ -50,8 +52,10 @@ import org.apache.cassandra.db.compaction.CompactionManager;
import org.apache.cassandra.io.sstable.*;
import org.apache.cassandra.utils.ByteBufferUtil;
+import static junit.framework.Assert.assertNotNull;
import static org.apache.cassandra.Util.column;
import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.junit.Assume.assumeTrue;
@@ -62,6 +66,13 @@ public class ScrubTest extends SchemaLoader
public String CF = "Standard1";
public String CF3 = "Standard2";
public String COUNTER_CF = "Counter1";
+ private static Integer COMPRESSION_CHUNK_LENGTH = 4096;
+
+ @BeforeClass
+ public static void loadSchema() throws IOException, ConfigurationException
+ {
+ loadSchema(false, COMPRESSION_CHUNK_LENGTH);
+ }
@Test
public void testScrubOneRow() throws IOException, ExecutionException, InterruptedException, ConfigurationException
@@ -88,7 +99,69 @@ public class ScrubTest extends SchemaLoader
@Test
public void testScrubCorruptedCounterRow() throws IOException, InterruptedException, ExecutionException
{
- // skip the test when compression is enabled until CASSANDRA-9140 is complete
+ // When compression is enabled, for testing corrupted chunks we need enough partitions to cover
+ // at least 3 chunks of size COMPRESSION_CHUNK_LENGTH
+ int numPartitions = 1000;
+
+ CompactionManager.instance.disableAutoCompaction();
+ Keyspace keyspace = Keyspace.open(KEYSPACE);
+ ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(COUNTER_CF);
+ cfs.clearUnsafe();
+
+ fillCounterCF(cfs, numPartitions);
+
+ List<Row> rows = cfs.getRangeSlice(Util.range("", ""), null, new IdentityQueryFilter(), numPartitions*10);
+ assertEquals(numPartitions, rows.size());
+ assertEquals(1, cfs.getSSTables().size());
+
+ SSTableReader sstable = cfs.getSSTables().iterator().next();
+
+ //make sure to override at most 1 chunk when compression is enabled
+ overrideWithGarbage(sstable, ByteBufferUtil.bytes("0"), ByteBufferUtil.bytes("1"));
+
+ // with skipCorrupted == false, the scrub is expected to fail
+ Scrubber scrubber = new Scrubber(cfs, sstable, false);
+ try
+ {
+ scrubber.scrub();
+ fail("Expected a CorruptSSTableException to be thrown");
+ }
+ catch (IOError err) {}
+
+ // with skipCorrupted == true, the corrupt row will be skipped
+ Scrubber.ScrubResult scrubResult;
+ scrubber = new Scrubber(cfs, sstable, true);
+ scrubResult = scrubber.scrubWithResult();
+ scrubber.close();
+ cfs.replaceCompactedSSTables(Collections.singletonList(sstable), Collections.singletonList(scrubber.getNewSSTable()), OperationType.SCRUB);
+
+ assertNotNull(scrubResult);
+
+ boolean compression = Boolean.parseBoolean(System.getProperty("cassandra.test.compression", "false"));
+ if (compression)
+ {
+ assertEquals(0, scrubResult.emptyRows);
+ assertEquals(numPartitions, scrubResult.badRows + scrubResult.goodRows);
+ //because we only corrupted 1 chunk and we chose enough partitions to cover at least 3 chunks
+ assertTrue(scrubResult.goodRows >= scrubResult.badRows * 2);
+ }
+ else
+ {
+ assertEquals(0, scrubResult.emptyRows);
+ assertEquals(1, scrubResult.badRows);
+ assertEquals(numPartitions-1, scrubResult.goodRows);
+ }
+ assertEquals(1, cfs.getSSTables().size());
+
+ rows = cfs.getRangeSlice(Util.range("", ""), null, new IdentityQueryFilter(), 1000);
+ assertEquals(scrubResult.goodRows, rows.size());
+ }
+
+
+ @Test
+ public void testScrubCorruptedRowInSmallFile() throws IOException, InterruptedException, ExecutionException
+ {
+ // cannot test this with compression
assumeTrue(!Boolean.parseBoolean(System.getProperty("cassandra.test.compression", "false")));
CompactionManager.instance.disableAutoCompaction();
@@ -102,17 +175,7 @@ public class ScrubTest extends SchemaLoader
assertEquals(2, rows.size());
SSTableReader sstable = cfs.getSSTables().iterator().next();
-
- // overwrite one row with garbage
- long row0Start = sstable.getPosition(RowPosition.forKey(ByteBufferUtil.bytes("0"), sstable.partitioner), SSTableReader.Operator.EQ).position;
- long row1Start = sstable.getPosition(RowPosition.forKey(ByteBufferUtil.bytes("1"), sstable.partitioner), SSTableReader.Operator.EQ).position;
- long startPosition = row0Start < row1Start ? row0Start : row1Start;
- long endPosition = row0Start < row1Start ? row1Start : row0Start;
-
- RandomAccessFile file = new RandomAccessFile(sstable.getFilename(), "rw");
- file.seek(startPosition);
- file.writeBytes(StringUtils.repeat('z', (int) (endPosition - startPosition)));
- file.close();
+ overrideWithGarbage(sstable, ByteBufferUtil.bytes("0"), ByteBufferUtil.bytes("1"));
// with skipCorrupted == false, the scrub is expected to fail
Scrubber scrubber = new Scrubber(cfs, sstable, false);
@@ -136,6 +199,34 @@ public class ScrubTest extends SchemaLoader
}
@Test
+ public void testScrubOneRowWithCorruptedKey() throws IOException, ExecutionException, InterruptedException, ConfigurationException
+ {
+ // cannot test this with compression
+ assumeTrue(!Boolean.parseBoolean(System.getProperty("cassandra.test.compression", "false")));
+
+ CompactionManager.instance.disableAutoCompaction();
+ Keyspace keyspace = Keyspace.open(KEYSPACE);
+ ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF);
+ cfs.clearUnsafe();
+
+ List<Row> rows;
+
+ // insert data and verify we get it back w/ range query
+ fillCF(cfs, 4);
+ rows = cfs.getRangeSlice(Util.range("", ""), null, new IdentityQueryFilter(), 1000);
+ assertEquals(4, rows.size());
+
+ SSTableReader sstable = cfs.getSSTables().iterator().next();
+ overrideWithGarbage(sstable, 0, 2);
+
+ CompactionManager.instance.performScrub(cfs, false);
+
+ // check data is still there
+ rows = cfs.getRangeSlice(Util.range("", ""), null, new IdentityQueryFilter(), 1000);
+ assertEquals(4, rows.size());
+ }
+
+ @Test
public void testScrubDeletedRow() throws IOException, ExecutionException, InterruptedException, ConfigurationException
{
CompactionManager.instance.disableAutoCompaction();
@@ -239,6 +330,42 @@ public class ScrubTest extends SchemaLoader
assert rows.size() == 6 : "Got " + rows.size();
}
+ private void overrideWithGarbage(SSTableReader sstable, ByteBuffer key1, ByteBuffer key2) throws IOException
+ {
+ boolean compression = Boolean.parseBoolean(System.getProperty("cassandra.test.compression", "false"));
+ long startPosition, endPosition;
+
+ if (compression)
+ { // overwrite with garbage the compression chunks from key1 to key2
+ CompressionMetadata compData = CompressionMetadata.create(sstable.getFilename());
+
+ CompressionMetadata.Chunk chunk1 = compData.chunkFor(
+ sstable.getPosition(RowPosition.forKey(key1, sstable.partitioner), SSTableReader.Operator.EQ).position);
+ CompressionMetadata.Chunk chunk2 = compData.chunkFor(
+ sstable.getPosition(RowPosition.forKey(key2, sstable.partitioner), SSTableReader.Operator.EQ).position);
+
+ startPosition = Math.min(chunk1.offset, chunk2.offset);
+ endPosition = Math.max(chunk1.offset + chunk1.length, chunk2.offset + chunk2.length);
+ }
+ else
+ { // overwrite with garbage from key1 to key2
+ long row0Start = sstable.getPosition(RowPosition.forKey(key1, sstable.partitioner), SSTableReader.Operator.EQ).position;
+ long row1Start = sstable.getPosition(RowPosition.forKey(key2, sstable.partitioner), SSTableReader.Operator.EQ).position;
+ startPosition = Math.min(row0Start, row1Start);
+ endPosition = Math.max(row0Start, row1Start);
+ }
+
+ overrideWithGarbage(sstable, startPosition, endPosition);
+ }
+
+ private void overrideWithGarbage(SSTable sstable, long startPosition, long endPosition) throws IOException
+ {
+ RandomAccessFile file = new RandomAccessFile(sstable.getFilename(), "rw");
+ file.seek(startPosition);
+ file.writeBytes(StringUtils.repeat('z', (int) (endPosition - startPosition)));
+ file.close();
+ }
+
private static boolean isRowOrdered(List<Row> rows)
{
DecoratedKey prev = null;
[2/3] cassandra git commit: Merge branch 'cassandra-2.0' into
cassandra-2.1
Posted by ty...@apache.org.
Merge branch 'cassandra-2.0' into cassandra-2.1
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/db127a10
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/db127a10
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/db127a10
Branch: refs/heads/trunk
Commit: db127a1073cfec426019c06d8f33e6c44292949f
Parents: 80c5191 593a725
Author: Tyler Hobbs <ty...@gmail.com>
Authored: Thu Apr 30 13:09:11 2015 -0500
Committer: Tyler Hobbs <ty...@gmail.com>
Committed: Thu Apr 30 13:09:11 2015 -0500
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../cassandra/db/compaction/Scrubber.java | 127 +++++++++++----
.../unit/org/apache/cassandra/SchemaLoader.java | 17 +-
.../unit/org/apache/cassandra/db/ScrubTest.java | 156 +++++++++++++++++--
4 files changed, 256 insertions(+), 45 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/db127a10/CHANGES.txt
----------------------------------------------------------------------
diff --cc CHANGES.txt
index cb235a4,4e7a5d0..c063368
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@@ -1,98 -1,11 +1,99 @@@
-2.0.15:
+2.1.6
+ * Fix PITR commitlog replay (CASSANDRA-9195)
+ * GCInspector logs very different times (CASSANDRA-9124)
+ * Fix deleting from an empty list (CASSANDRA-9198)
+ * Update tuple and collection types that use a user-defined type when that UDT
+ is modified (CASSANDRA-9148, CASSANDRA-9192)
+Merged from 2.0:
+ * Allow scrub to handle corrupted compressed chunks (CASSANDRA-9140)
* Fix assertion error when resetlocalschema is run during repair (CASSANDRA-9249)
* Disable single sstable tombstone compactions for DTCS by default (CASSANDRA-9234)
- * Do more agressive ttl expiration checks to be able to
- drop more sstables (CASSANDRA-8243)
* IncomingTcpConnection thread is not named (CASSANDRA-9262)
* Close incoming connections when MessagingService is stopped (CASSANDRA-9238)
+
+2.1.5
+ * Re-add deprecated cold_reads_to_omit param for backwards compat (CASSANDRA-9203)
+ * Make anticompaction visible in compactionstats (CASSANDRA-9098)
+ * Improve nodetool getendpoints documentation about the partition
+ key parameter (CASSANDRA-6458)
+ * Don't check other keyspaces for schema changes when an user-defined
+ type is altered (CASSANDRA-9187)
+ * Allow takeColumnFamilySnapshot to take a list of tables (CASSANDRA-8348)
+ * Limit major sstable operations to their canonical representation (CASSANDRA-8669)
+ * cqlsh: Add tests for INSERT and UPDATE tab completion (CASSANDRA-9125)
+ * cqlsh: quote column names when needed in COPY FROM inserts (CASSANDRA-9080)
+ * Add generate-idea-files target to build.xml (CASSANDRA-9123)
+ * Do not load read meter for offline operations (CASSANDRA-9082)
+ * cqlsh: Make CompositeType data readable (CASSANDRA-8919)
+ * cqlsh: Fix display of triggers (CASSANDRA-9081)
+ * Fix NullPointerException when deleting or setting an element by index on
+ a null list collection (CASSANDRA-9077)
+ * Buffer bloom filter serialization (CASSANDRA-9066)
+ * Fix anti-compaction target bloom filter size (CASSANDRA-9060)
+ * Make FROZEN and TUPLE unreserved keywords in CQL (CASSANDRA-9047)
+ * Prevent AssertionError from SizeEstimatesRecorder (CASSANDRA-9034)
+ * Avoid overwriting index summaries for sstables with an older format that
+ does not support downsampling; rebuild summaries on startup when this
+ is detected (CASSANDRA-8993)
+ * Fix potential data loss in CompressedSequentialWriter (CASSANDRA-8949)
+ * Make PasswordAuthenticator number of hashing rounds configurable (CASSANDRA-8085)
+ * Fix AssertionError when binding nested collections in DELETE (CASSANDRA-8900)
+ * Check for overlap with non-early sstables in LCS (CASSANDRA-8739)
+ * Only calculate max purgable timestamp if we have to (CASSANDRA-8914)
+ * (cqlsh) Greatly improve performance of COPY FROM (CASSANDRA-8225)
+ * IndexSummary effectiveIndexInterval is now a guideline, not a rule (CASSANDRA-8993)
+ * Use correct bounds for page cache eviction of compressed files (CASSANDRA-8746)
+ * SSTableScanner enforces its bounds (CASSANDRA-8946)
+ * Cleanup cell equality (CASSANDRA-8947)
+ * Introduce intra-cluster message coalescing (CASSANDRA-8692)
+ * DatabaseDescriptor throws NPE when rpc_interface is used (CASSANDRA-8839)
+ * Don't check if an sstable is live for offline compactions (CASSANDRA-8841)
+ * Don't set clientMode in SSTableLoader (CASSANDRA-8238)
+ * Fix SSTableRewriter with disabled early open (CASSANDRA-8535)
+ * Allow invalidating permissions and cache time (CASSANDRA-8722)
+ * Log warning when queries that will require ALLOW FILTERING in Cassandra 3.0
+ are executed (CASSANDRA-8418)
+ * Fix cassandra-stress so it respects the CL passed in user mode (CASSANDRA-8948)
+ * Fix rare NPE in ColumnDefinition#hasIndexOption() (CASSANDRA-8786)
+ * cassandra-stress reports per-operation statistics, plus misc (CASSANDRA-8769)
+ * Use long for key count in cfstats (CASSANDRA-8913)
+ * Make SSTableRewriter.abort() more robust to failure (CASSANDRA-8832)
+ * Remove cold_reads_to_omit from STCS (CASSANDRA-8860)
+ * Make EstimatedHistogram#percentile() use ceil instead of floor (CASSANDRA-8883)
+ * Fix top partitions reporting wrong cardinality (CASSANDRA-8834)
+ * Fix rare NPE in KeyCacheSerializer (CASSANDRA-8067)
+ * Pick sstables for validation as late as possible inc repairs (CASSANDRA-8366)
+ * Fix commitlog getPendingTasks to not increment (CASSANDRA-8856)
+ * Fix parallelism adjustment in range and secondary index queries
+ when the first fetch does not satisfy the limit (CASSANDRA-8856)
+ * Check if the filtered sstables is non-empty in STCS (CASSANDRA-8843)
+ * Upgrade java-driver used for cassandra-stress (CASSANDRA-8842)
+ * Fix CommitLog.forceRecycleAllSegments() memory access error (CASSANDRA-8812)
+ * Improve assertions in Memory (CASSANDRA-8792)
+ * Fix SSTableRewriter cleanup (CASSANDRA-8802)
+ * Introduce SafeMemory for CompressionMetadata.Writer (CASSANDRA-8758)
+ * 'nodetool info' prints exception against older node (CASSANDRA-8796)
+ * Ensure SSTableReader.last corresponds exactly with the file end (CASSANDRA-8750)
+ * Make SSTableWriter.openEarly more robust and obvious (CASSANDRA-8747)
+ * Enforce SSTableReader.first/last (CASSANDRA-8744)
+ * Cleanup SegmentedFile API (CASSANDRA-8749)
+ * Avoid overlap with early compaction replacement (CASSANDRA-8683)
+ * Safer Resource Management++ (CASSANDRA-8707)
+ * Write partition size estimates into a system table (CASSANDRA-7688)
+ * cqlsh: Fix keys() and full() collection indexes in DESCRIBE output
+ (CASSANDRA-8154)
+ * Show progress of streaming in nodetool netstats (CASSANDRA-8886)
+ * IndexSummaryBuilder utilises offheap memory, and shares data between
+ each IndexSummary opened from it (CASSANDRA-8757)
+ * markCompacting only succeeds if the exact SSTableReader instances being
+ marked are in the live set (CASSANDRA-8689)
+ * cassandra-stress support for varint (CASSANDRA-8882)
+ * Fix Adler32 digest for compressed sstables (CASSANDRA-8778)
+ * Add nodetool statushandoff/statusbackup (CASSANDRA-8912)
+ * Use stdout for progress and stats in sstableloader (CASSANDRA-8982)
+ * Correctly identify 2i datadir from older versions (CASSANDRA-9116)
+Merged from 2.0:
+ * Ignore gossip SYNs after shutdown (CASSANDRA-9238)
* Avoid overflow when calculating max sstable size in LCS (CASSANDRA-9235)
* Make sstable blacklisting work with compression (CASSANDRA-9138)
* Do not attempt to rebuild indexes if no index accepts any column (CASSANDRA-9196)
http://git-wip-us.apache.org/repos/asf/cassandra/blob/db127a10/src/java/org/apache/cassandra/db/compaction/Scrubber.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/db/compaction/Scrubber.java
index 2f53ab9,1752b21..1f5c7de
--- a/src/java/org/apache/cassandra/db/compaction/Scrubber.java
+++ b/src/java/org/apache/cassandra/db/compaction/Scrubber.java
@@@ -21,8 -21,8 +21,9 @@@ import java.nio.ByteBuffer
import java.io.*;
import java.util.*;
+ import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Throwables;
+import com.google.common.collect.Sets;
import org.apache.cassandra.db.*;
import org.apache.cassandra.io.sstable.*;
@@@ -109,14 -113,12 +118,14 @@@ public class Scrubber implements Closea
public void scrub()
{
outputHandler.output(String.format("Scrubbing %s (%s bytes)", sstable, dataFile.length()));
+ Set<SSTableReader> oldSSTable = Sets.newHashSet(sstable);
+ SSTableRewriter writer = new SSTableRewriter(cfs, oldSSTable, sstable.maxDataAge, isOffline);
try
{
- ByteBuffer nextIndexKey = ByteBufferUtil.readWithShortLength(indexFile);
+ nextIndexKey = ByteBufferUtil.readWithShortLength(indexFile);
{
// throw away variable so we don't have a side effect in the assert
- long firstRowPositionFromIndex = RowIndexEntry.serializer.deserialize(indexFile, sstable.descriptor.version).position;
+ long firstRowPositionFromIndex = sstable.metadata.comparator.rowIndexEntrySerializer().deserialize(indexFile, sstable.descriptor.version).position;
assert firstRowPositionFromIndex == 0 : firstRowPositionFromIndex;
}
@@@ -143,33 -152,32 +153,22 @@@
// check for null key below
}
- ByteBuffer currentIndexKey = nextIndexKey;
- long nextRowPositionFromIndex;
- try
- {
- nextIndexKey = indexFile.isEOF() ? null : ByteBufferUtil.readWithShortLength(indexFile);
- nextRowPositionFromIndex = indexFile.isEOF()
- ? dataFile.length()
- : sstable.metadata.comparator.rowIndexEntrySerializer().deserialize(indexFile, sstable.descriptor.version).position;
- }
- catch (Throwable th)
- {
- JVMStabilityInspector.inspectThrowable(th);
- outputHandler.warn("Error reading index file", th);
- nextIndexKey = null;
- nextRowPositionFromIndex = dataFile.length();
- }
+ updateIndexKey();
long dataStart = dataFile.getFilePointer();
- long dataStartFromIndex = currentIndexKey == null
- ? -1
- : rowStart + 2 + currentIndexKey.remaining();
- long dataSizeFromIndex = nextRowPositionFromIndex - dataStartFromIndex;
++
+ long dataStartFromIndex = -1;
+ long dataSizeFromIndex = -1;
+ if (currentIndexKey != null)
+ {
+ dataStartFromIndex = currentRowPositionFromIndex + 2 + currentIndexKey.remaining();
- if (sstable.descriptor.version.hasRowSizeAndColumnCount)
- dataStartFromIndex += 8;
-
+ dataSizeFromIndex = nextRowPositionFromIndex - dataStartFromIndex;
+ }
- if (!sstable.descriptor.version.hasRowSizeAndColumnCount)
- {
- dataSize = dataSizeFromIndex;
- // avoid an NPE if key is null
- String keyName = key == null ? "(unreadable key)" : ByteBufferUtil.bytesToHex(key.key);
- outputHandler.debug(String.format("row %s is %s bytes", keyName, dataSize));
- }
- else
- {
- if (currentIndexKey != null)
- outputHandler.debug(String.format("Index doublecheck: row %s is %s bytes", ByteBufferUtil.bytesToHex(currentIndexKey), dataSizeFromIndex));
- }
+ dataSize = dataSizeFromIndex;
+ // avoid an NPE if key is null
+ String keyName = key == null ? "(unreadable key)" : ByteBufferUtil.bytesToHex(key.getKey());
+ outputHandler.debug(String.format("row %s is %s bytes", keyName, dataSize));
assert currentIndexKey != null || indexFile.isEOF();
@@@ -177,9 -186,22 +176,22 @@@
{
if (key == null)
throw new IOError(new IOException("Unable to read row key from data file"));
+
- if (!key.key.equals(currentIndexKey))
++ if (!key.getKey().equals(currentIndexKey))
+ {
+ throw new IOError(new IOException(String.format("Key from data file (%s) does not match key from index file (%s)",
- ByteBufferUtil.bytesToHex(key.key), ByteBufferUtil.bytesToHex(currentIndexKey))));
++ ByteBufferUtil.bytesToHex(key.getKey()), ByteBufferUtil.bytesToHex(currentIndexKey))));
+ }
+
if (dataSize > dataFile.length())
- throw new IOError(new IOException("Impossible row size (greater than file length): " + dataSize));
+ throw new IOError(new IOException("Impossible row size " + dataSize));
+ if (dataStart != dataStartFromIndex)
- outputHandler.warn(String.format("Data file row position %d differs from index file row position %d", dataStart, dataSizeFromIndex));
++ outputHandler.warn(String.format("Data file row position %d different from index file row position %d", dataStart, dataSizeFromIndex));
+
+ if (dataSize != dataSizeFromIndex)
- outputHandler.warn(String.format("Data file row size %d differs from index file row size %d", dataSize, dataSizeFromIndex));
++ outputHandler.warn(String.format("Data file row size %d different from index file row size %d", dataSize, dataSizeFromIndex));
+
SSTableIdentityIterator atoms = new SSTableIdentityIterator(sstable, dataFile, key, dataSize, true);
if (prevKey != null && prevKey.compareTo(key) > 0)
{
@@@ -229,8 -254,9 +243,8 @@@
throwIfCommutative(key, th2);
outputHandler.warn("Retry failed too. Skipping to next row (retry's stacktrace follows)", th2);
- dataFile.seek(nextRowPositionFromIndex);
- writer.resetAndTruncate();
badRows++;
+ seekToNextRow();
}
}
else
@@@ -290,6 -309,45 +304,46 @@@
}
}
+ private void updateIndexKey()
+ {
+ currentIndexKey = nextIndexKey;
+ currentRowPositionFromIndex = nextRowPositionFromIndex;
+ try
+ {
+ nextIndexKey = indexFile.isEOF() ? null : ByteBufferUtil.readWithShortLength(indexFile);
+ nextRowPositionFromIndex = indexFile.isEOF()
+ ? dataFile.length()
- : RowIndexEntry.serializer.deserialize(indexFile, sstable.descriptor.version).position;
++ : sstable.metadata.comparator.rowIndexEntrySerializer().deserialize(indexFile, sstable.descriptor.version).position;
+ }
+ catch (Throwable th)
+ {
++ JVMStabilityInspector.inspectThrowable(th);
+ outputHandler.warn("Error reading index file", th);
+ nextIndexKey = null;
+ nextRowPositionFromIndex = dataFile.length();
+ }
+ }
+
+ private void seekToNextRow()
+ {
+ while(nextRowPositionFromIndex < dataFile.length())
+ {
+ try
+ {
+ dataFile.seek(nextRowPositionFromIndex);
+ return;
+ }
+ catch (Throwable th)
+ {
+ throwIfFatal(th);
+ outputHandler.warn(String.format("Failed to seek to next row position %d", nextRowPositionFromIndex), th);
+ badRows++;
+ }
+
+ updateIndexKey();
+ }
+ }
+
private void saveOutOfOrderRow(DecoratedKey prevKey, DecoratedKey key, SSTableIdentityIterator atoms)
{
// TODO bitch if the row is too large? if it is there's not much we can do ...
@@@ -380,9 -438,30 +434,30 @@@
}
@Override
- public boolean shouldPurge(DecoratedKey key, long delTimestamp)
+ public long maxPurgeableTimestamp(DecoratedKey key)
{
- return false;
+ return Long.MIN_VALUE;
}
}
+
+ @VisibleForTesting
+ public ScrubResult scrubWithResult()
+ {
+ scrub();
+ return new ScrubResult(this);
+ }
+
+ public static final class ScrubResult
+ {
+ public final int goodRows;
+ public final int badRows;
+ public final int emptyRows;
+
+ public ScrubResult(Scrubber scrubber)
+ {
+ this.goodRows = scrubber.goodRows;
+ this.badRows = scrubber.badRows;
+ this.emptyRows = scrubber.emptyRows;
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/cassandra/blob/db127a10/test/unit/org/apache/cassandra/SchemaLoader.java
----------------------------------------------------------------------
diff --cc test/unit/org/apache/cassandra/SchemaLoader.java
index db1758f,6b34b9a..c6a3855
--- a/test/unit/org/apache/cassandra/SchemaLoader.java
+++ b/test/unit/org/apache/cassandra/SchemaLoader.java
@@@ -51,30 -53,17 +51,35 @@@ public class SchemaLoade
private static Logger logger = LoggerFactory.getLogger(SchemaLoader.class);
@BeforeClass
- public static void loadSchema() throws IOException, ConfigurationException
+ public static void loadSchema() throws ConfigurationException
{
- loadSchema(false);
++ loadSchema(null);
++ }
++
++ public static void loadSchema(Integer compressionChunkLength) throws ConfigurationException
++ {
+ prepareServer();
+
+ // Migrations aren't happy if gossiper is not started. Even if we don't use migrations though,
+ // some tests now expect us to start gossip for them.
+ startGossiper();
+
+ // if you're messing with low-level sstable stuff, it can be useful to inject the schema directly
+ // Schema.instance.load(schemaDefinition());
- for (KSMetaData ksm : schemaDefinition())
++ for (KSMetaData ksm : schemaDefinition(compressionChunkLength))
+ MigrationManager.announceNewKeyspace(ksm);
}
- public static void loadSchema(boolean withOldCfIds) throws IOException, ConfigurationException
+ @After
+ public void leakDetect() throws InterruptedException
{
- loadSchema(withOldCfIds, null);
+ System.gc();
+ System.gc();
+ System.gc();
+ Thread.sleep(10);
}
- public static void loadSchema(boolean withOldCfIds, Integer compressionChunkLength) throws IOException, ConfigurationException
+ public static void prepareServer()
{
// Cleanup first
cleanupAndLeaveDirs();
@@@ -103,7 -98,7 +108,7 @@@
Gossiper.instance.stop();
}
- public static Collection<KSMetaData> schemaDefinition() throws ConfigurationException
- public static Collection<KSMetaData> schemaDefinition(boolean withOldCfIds, Integer compressionChunkLength) throws ConfigurationException
++ public static Collection<KSMetaData> schemaDefinition(Integer compressionChunkLength) throws ConfigurationException
{
List<KSMetaData> schema = new ArrayList<KSMetaData>();
@@@ -355,15 -359,21 +360,15 @@@
final Map<String, String> indexOptions = Collections.singletonMap(
SecondaryIndex.CUSTOM_INDEX_OPTION_NAME,
PerRowSecondaryIndexTest.TestIndex.class.getName());
- return standardCFMD(ksName, cfName)
- .keyValidator(AsciiType.instance)
- .columnMetadata(new HashMap<ByteBuffer, ColumnDefinition>()
- {{
- ByteBuffer cName = ByteBuffer.wrap("indexed".getBytes(StandardCharsets.UTF_8));
- put(cName, new ColumnDefinition(cName,
- AsciiType.instance,
- IndexType.CUSTOM,
- indexOptions,
- ByteBufferUtil.bytesToHex(cName),
- null, ColumnDefinition.Type.REGULAR));
- }});
+
+ CFMetaData cfm = CFMetaData.sparseCFMetaData(ksName, cfName, AsciiType.instance).keyValidator(AsciiType.instance);
+
+ ByteBuffer cName = ByteBufferUtil.bytes("indexed");
+ return cfm.addOrReplaceColumnDefinition(ColumnDefinition.regularDef(cfm, cName, AsciiType.instance, null)
+ .setIndex("indexe1", IndexType.CUSTOM, indexOptions));
}
- private static void useCompression(List<KSMetaData> schema)
+ private static void useCompression(List<KSMetaData> schema, Integer chunkLength) throws ConfigurationException
{
for (KSMetaData ksm : schema)
{
http://git-wip-us.apache.org/repos/asf/cassandra/blob/db127a10/test/unit/org/apache/cassandra/db/ScrubTest.java
----------------------------------------------------------------------
diff --cc test/unit/org/apache/cassandra/db/ScrubTest.java
index 383b1d0,08237a4..a19c76d
--- a/test/unit/org/apache/cassandra/db/ScrubTest.java
+++ b/test/unit/org/apache/cassandra/db/ScrubTest.java
@@@ -31,9 -30,12 +30,12 @@@ import java.util.Set
import java.util.concurrent.ExecutionException;
import org.apache.cassandra.cql3.QueryProcessor;
-import org.apache.cassandra.db.compaction.OperationType;
++import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.exceptions.RequestExecutionException;
+ import org.apache.cassandra.io.compress.CompressionMetadata;
import org.apache.cassandra.utils.UUIDGen;
import org.apache.commons.lang3.StringUtils;
+ import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.runner.RunWith;
@@@ -43,18 -45,17 +45,21 @@@ import org.apache.cassandra.SchemaLoade
import org.apache.cassandra.Util;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.Schema;
-import org.apache.cassandra.db.compaction.Scrubber;
-import org.apache.cassandra.exceptions.ConfigurationException;
import org.apache.cassandra.db.columniterator.IdentityQueryFilter;
import org.apache.cassandra.db.compaction.CompactionManager;
-import org.apache.cassandra.io.sstable.*;
+import org.apache.cassandra.db.compaction.Scrubber;
+import org.apache.cassandra.exceptions.WriteTimeoutException;
+import org.apache.cassandra.io.sstable.Component;
+import org.apache.cassandra.io.sstable.Descriptor;
+import org.apache.cassandra.io.sstable.SSTableReader;
import org.apache.cassandra.utils.ByteBufferUtil;
-import static junit.framework.Assert.assertNotNull;
+import static org.apache.cassandra.Util.cellname;
import static org.apache.cassandra.Util.column;
++
++import static junit.framework.Assert.assertNotNull;
import static org.junit.Assert.assertEquals;
+ import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
import static org.junit.Assume.assumeTrue;
@@@ -65,9 -66,16 +70,16 @@@ public class ScrubTest extends SchemaLo
public String CF = "Standard1";
public String CF3 = "Standard2";
public String COUNTER_CF = "Counter1";
+ private static Integer COMPRESSION_CHUNK_LENGTH = 4096;
+
+ @BeforeClass
- public static void loadSchema() throws IOException, ConfigurationException
++ public static void loadSchema() throws ConfigurationException
+ {
- loadSchema(false, COMPRESSION_CHUNK_LENGTH);
++ loadSchema(COMPRESSION_CHUNK_LENGTH);
+ }
@Test
- public void testScrubOneRow() throws IOException, ExecutionException, InterruptedException, ConfigurationException
+ public void testScrubOneRow() throws ExecutionException, InterruptedException
{
CompactionManager.instance.disableAutoCompaction();
Keyspace keyspace = Keyspace.open(KEYSPACE);
@@@ -89,9 -97,71 +101,69 @@@
}
@Test
- public void testScrubCorruptedCounterRow() throws IOException, InterruptedException, ExecutionException
+ public void testScrubCorruptedCounterRow() throws IOException, WriteTimeoutException
{
- // skip the test when compression is enabled until CASSANDRA-9140 is complete
+ // When compression is enabled, for testing corrupted chunks we need enough partitions to cover
+ // at least 3 chunks of size COMPRESSION_CHUNK_LENGTH
+ int numPartitions = 1000;
+
+ CompactionManager.instance.disableAutoCompaction();
+ Keyspace keyspace = Keyspace.open(KEYSPACE);
+ ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(COUNTER_CF);
+ cfs.clearUnsafe();
+
+ fillCounterCF(cfs, numPartitions);
+
+ List<Row> rows = cfs.getRangeSlice(Util.range("", ""), null, new IdentityQueryFilter(), numPartitions*10);
+ assertEquals(numPartitions, rows.size());
+ assertEquals(1, cfs.getSSTables().size());
+
+ SSTableReader sstable = cfs.getSSTables().iterator().next();
+
+ //make sure to override at most 1 chunk when compression is enabled
+ overrideWithGarbage(sstable, ByteBufferUtil.bytes("0"), ByteBufferUtil.bytes("1"));
+
+ // with skipCorrupted == false, the scrub is expected to fail
- Scrubber scrubber = new Scrubber(cfs, sstable, false);
++ Scrubber scrubber = new Scrubber(cfs, sstable, false, false);
+ try
+ {
+ scrubber.scrub();
+ fail("Expected a CorruptSSTableException to be thrown");
+ }
+ catch (IOError err) {}
+
+ // with skipCorrupted == true, the corrupt row will be skipped
+ Scrubber.ScrubResult scrubResult;
- scrubber = new Scrubber(cfs, sstable, true);
++ scrubber = new Scrubber(cfs, sstable, true, false);
+ scrubResult = scrubber.scrubWithResult();
+ scrubber.close();
- cfs.replaceCompactedSSTables(Collections.singletonList(sstable), Collections.singletonList(scrubber.getNewSSTable()), OperationType.SCRUB);
+
+ assertNotNull(scrubResult);
+
+ boolean compression = Boolean.parseBoolean(System.getProperty("cassandra.test.compression", "false"));
+ if (compression)
+ {
+ assertEquals(0, scrubResult.emptyRows);
+ assertEquals(numPartitions, scrubResult.badRows + scrubResult.goodRows);
+ //because we only corrupted 1 chunk and we chose enough partitions to cover at least 3 chunks
+ assertTrue(scrubResult.goodRows >= scrubResult.badRows * 2);
+ }
+ else
+ {
+ assertEquals(0, scrubResult.emptyRows);
+ assertEquals(1, scrubResult.badRows);
+ assertEquals(numPartitions-1, scrubResult.goodRows);
+ }
+ assertEquals(1, cfs.getSSTables().size());
+
+ rows = cfs.getRangeSlice(Util.range("", ""), null, new IdentityQueryFilter(), 1000);
+ assertEquals(scrubResult.goodRows, rows.size());
+ }
+
-
+ @Test
- public void testScrubCorruptedRowInSmallFile() throws IOException, InterruptedException, ExecutionException
++ public void testScrubCorruptedRowInSmallFile() throws IOException, WriteTimeoutException
+ {
+ // cannot test this with compression
assumeTrue(!Boolean.parseBoolean(System.getProperty("cassandra.test.compression", "false")));
CompactionManager.instance.disableAutoCompaction();
@@@ -105,20 -175,10 +177,12 @@@
assertEquals(2, rows.size());
SSTableReader sstable = cfs.getSSTables().iterator().next();
+
+ // overwrite one row with garbage
- long row0Start = sstable.getPosition(RowPosition.ForKey.get(ByteBufferUtil.bytes("0"), sstable.partitioner), SSTableReader.Operator.EQ).position;
- long row1Start = sstable.getPosition(RowPosition.ForKey.get(ByteBufferUtil.bytes("1"), sstable.partitioner), SSTableReader.Operator.EQ).position;
- long startPosition = row0Start < row1Start ? row0Start : row1Start;
- long endPosition = row0Start < row1Start ? row1Start : row0Start;
-
- RandomAccessFile file = new RandomAccessFile(sstable.getFilename(), "rw");
- file.seek(startPosition);
- file.writeBytes(StringUtils.repeat('z', (int) (endPosition - startPosition)));
- file.close();
+ overrideWithGarbage(sstable, ByteBufferUtil.bytes("0"), ByteBufferUtil.bytes("1"));
// with skipCorrupted == false, the scrub is expected to fail
- Scrubber scrubber = new Scrubber(cfs, sstable, false);
+ Scrubber scrubber = new Scrubber(cfs, sstable, false, false);
try
{
scrubber.scrub();
@@@ -138,7 -199,35 +202,35 @@@
}
@Test
+ public void testScrubOneRowWithCorruptedKey() throws IOException, ExecutionException, InterruptedException, ConfigurationException
+ {
+ // cannot test this with compression
+ assumeTrue(!Boolean.parseBoolean(System.getProperty("cassandra.test.compression", "false")));
+
+ CompactionManager.instance.disableAutoCompaction();
+ Keyspace keyspace = Keyspace.open(KEYSPACE);
+ ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF);
+ cfs.clearUnsafe();
+
+ List<Row> rows;
+
+ // insert data and verify we get it back w/ range query
+ fillCF(cfs, 4);
+ rows = cfs.getRangeSlice(Util.range("", ""), null, new IdentityQueryFilter(), 1000);
+ assertEquals(4, rows.size());
+
+ SSTableReader sstable = cfs.getSSTables().iterator().next();
+ overrideWithGarbage(sstable, 0, 2);
+
+ CompactionManager.instance.performScrub(cfs, false);
+
+ // check data is still there
+ rows = cfs.getRangeSlice(Util.range("", ""), null, new IdentityQueryFilter(), 1000);
+ assertEquals(4, rows.size());
+ }
+
+ @Test
- public void testScrubDeletedRow() throws IOException, ExecutionException, InterruptedException, ConfigurationException
+ public void testScrubDeletedRow() throws ExecutionException, InterruptedException
{
CompactionManager.instance.disableAutoCompaction();
Keyspace keyspace = Keyspace.open(KEYSPACE);
@@@ -241,6 -330,42 +333,42 @@@
assert rows.size() == 6 : "Got " + rows.size();
}
+ private void overrideWithGarbage(SSTableReader sstable, ByteBuffer key1, ByteBuffer key2) throws IOException
+ {
+ boolean compression = Boolean.parseBoolean(System.getProperty("cassandra.test.compression", "false"));
+ long startPosition, endPosition;
+
+ if (compression)
+ { // overwrite with garbage the compression chunks from key1 to key2
+ CompressionMetadata compData = CompressionMetadata.create(sstable.getFilename());
+
+ CompressionMetadata.Chunk chunk1 = compData.chunkFor(
- sstable.getPosition(RowPosition.forKey(key1, sstable.partitioner), SSTableReader.Operator.EQ).position);
++ sstable.getPosition(RowPosition.ForKey.get(key1, sstable.partitioner), SSTableReader.Operator.EQ).position);
+ CompressionMetadata.Chunk chunk2 = compData.chunkFor(
- sstable.getPosition(RowPosition.forKey(key2, sstable.partitioner), SSTableReader.Operator.EQ).position);
++ sstable.getPosition(RowPosition.ForKey.get(key2, sstable.partitioner), SSTableReader.Operator.EQ).position);
+
+ startPosition = Math.min(chunk1.offset, chunk2.offset);
+ endPosition = Math.max(chunk1.offset + chunk1.length, chunk2.offset + chunk2.length);
+ }
+ else
+ { // overwrite with garbage from key1 to key2
- long row0Start = sstable.getPosition(RowPosition.forKey(key1, sstable.partitioner), SSTableReader.Operator.EQ).position;
- long row1Start = sstable.getPosition(RowPosition.forKey(key2, sstable.partitioner), SSTableReader.Operator.EQ).position;
++ long row0Start = sstable.getPosition(RowPosition.ForKey.get(key1, sstable.partitioner), SSTableReader.Operator.EQ).position;
++ long row1Start = sstable.getPosition(RowPosition.ForKey.get(key2, sstable.partitioner), SSTableReader.Operator.EQ).position;
+ startPosition = Math.min(row0Start, row1Start);
+ endPosition = Math.max(row0Start, row1Start);
+ }
+
+ overrideWithGarbage(sstable, startPosition, endPosition);
+ }
+
- private void overrideWithGarbage(SSTable sstable, long startPosition, long endPosition) throws IOException
++ private void overrideWithGarbage(SSTableReader sstable, long startPosition, long endPosition) throws IOException
+ {
+ RandomAccessFile file = new RandomAccessFile(sstable.getFilename(), "rw");
+ file.seek(startPosition);
+ file.writeBytes(StringUtils.repeat('z', (int) (endPosition - startPosition)));
+ file.close();
+ }
+
private static boolean isRowOrdered(List<Row> rows)
{
DecoratedKey prev = null;
[3/3] cassandra git commit: Merge branch 'cassandra-2.1' into trunk
Posted by ty...@apache.org.
Merge branch 'cassandra-2.1' into trunk
Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/49b08989
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/49b08989
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/49b08989
Branch: refs/heads/trunk
Commit: 49b089893b5407ba42dad389804ff21d535a8537
Parents: 3bfe4b6 db127a1
Author: Tyler Hobbs <ty...@gmail.com>
Authored: Thu Apr 30 13:15:57 2015 -0500
Committer: Tyler Hobbs <ty...@gmail.com>
Committed: Thu Apr 30 13:15:57 2015 -0500
----------------------------------------------------------------------
CHANGES.txt | 1 +
.../cassandra/db/compaction/Scrubber.java | 128 ++++++++++++---
.../compress/CompressedRandomAccessReader.java | 2 +
.../unit/org/apache/cassandra/SchemaLoader.java | 7 +-
.../unit/org/apache/cassandra/db/ScrubTest.java | 157 ++++++++++++++++---
5 files changed, 242 insertions(+), 53 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/49b08989/CHANGES.txt
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/cassandra/blob/49b08989/src/java/org/apache/cassandra/db/compaction/Scrubber.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/db/compaction/Scrubber.java
index 675b604,1f5c7de..1e50da6
--- a/src/java/org/apache/cassandra/db/compaction/Scrubber.java
+++ b/src/java/org/apache/cassandra/db/compaction/Scrubber.java
@@@ -121,10 -122,10 +130,10 @@@ public class Scrubber implements Closea
SSTableRewriter writer = new SSTableRewriter(cfs, oldSSTable, sstable.maxDataAge, isOffline);
try
{
- ByteBuffer nextIndexKey = ByteBufferUtil.readWithShortLength(indexFile);
+ nextIndexKey = ByteBufferUtil.readWithShortLength(indexFile);
{
// throw away variable so we don't have a side effect in the assert
- long firstRowPositionFromIndex = sstable.metadata.comparator.rowIndexEntrySerializer().deserialize(indexFile, sstable.descriptor.version).position;
+ long firstRowPositionFromIndex = rowIndexEntrySerializer.deserialize(indexFile, sstable.descriptor.version).position;
assert firstRowPositionFromIndex == 0 : firstRowPositionFromIndex;
}
@@@ -150,30 -153,19 +160,19 @@@
// check for null key below
}
- ByteBuffer currentIndexKey = nextIndexKey;
- long nextRowPositionFromIndex;
- try
- {
- nextIndexKey = indexFile.isEOF() ? null : ByteBufferUtil.readWithShortLength(indexFile);
- nextRowPositionFromIndex = indexFile.isEOF()
- ? dataFile.length()
- : rowIndexEntrySerializer.deserialize(indexFile, sstable.descriptor.version).position;
- }
- catch (Throwable th)
- {
- JVMStabilityInspector.inspectThrowable(th);
- outputHandler.warn("Error reading index file", th);
- nextIndexKey = null;
- nextRowPositionFromIndex = dataFile.length();
- }
+ updateIndexKey();
long dataStart = dataFile.getFilePointer();
- long dataStartFromIndex = currentIndexKey == null
- ? -1
- : rowStart + 2 + currentIndexKey.remaining();
- long dataSizeFromIndex = nextRowPositionFromIndex - dataStartFromIndex;
+
+ long dataStartFromIndex = -1;
+ long dataSizeFromIndex = -1;
+ if (currentIndexKey != null)
+ {
+ dataStartFromIndex = currentRowPositionFromIndex + 2 + currentIndexKey.remaining();
+ dataSizeFromIndex = nextRowPositionFromIndex - dataStartFromIndex;
+ }
- dataSize = dataSizeFromIndex;
+ long dataSize = dataSizeFromIndex;
// avoid an NPE if key is null
String keyName = key == null ? "(unreadable key)" : ByteBufferUtil.bytesToHex(key.getKey());
outputHandler.debug(String.format("row %s is %s bytes", keyName, dataSize));
@@@ -187,7 -186,13 +193,14 @@@
if (dataSize > dataFile.length())
throw new IOError(new IOException("Impossible row size " + dataSize));
+ if (dataStart != dataStartFromIndex)
+ outputHandler.warn(String.format("Data file row position %d different from index file row position %d", dataStart, dataSizeFromIndex));
+
+ if (dataSize != dataSizeFromIndex)
+ outputHandler.warn(String.format("Data file row size %d different from index file row size %d", dataSize, dataSizeFromIndex));
+
- SSTableIdentityIterator atoms = new SSTableIdentityIterator(sstable, dataFile, key, dataSize, true);
+ SSTableIdentityIterator atoms = new SSTableIdentityIterator(sstable, dataFile, key, checkData);
++
if (prevKey != null && prevKey.compareTo(key) > 0)
{
saveOutOfOrderRow(prevKey, key, atoms);
@@@ -216,7 -220,9 +228,9 @@@
key = sstable.partitioner.decorateKey(currentIndexKey);
try
{
+ dataFile.seek(dataStartFromIndex);
+
- SSTableIdentityIterator atoms = new SSTableIdentityIterator(sstable, dataFile, key, dataSize, true);
+ SSTableIdentityIterator atoms = new SSTableIdentityIterator(sstable, dataFile, key, checkData);
if (prevKey != null && prevKey.compareTo(key) > 0)
{
saveOutOfOrderRow(prevKey, key, atoms);
@@@ -233,21 -240,21 +248,21 @@@
catch (Throwable th2)
{
throwIfFatal(th2);
- throwIfCommutative(key, th2);
+ throwIfCannotContinue(key, th2);
outputHandler.warn("Retry failed too. Skipping to next row (retry's stacktrace follows)", th2);
- dataFile.seek(nextRowPositionFromIndex);
badRows++;
+ seekToNextRow();
}
}
else
{
- throwIfCommutative(key, th);
+ throwIfCannotContinue(key, th);
outputHandler.warn("Row starting at position " + dataStart + " is unreadable; skipping to next");
- if (currentIndexKey != null)
- dataFile.seek(nextRowPositionFromIndex);
badRows++;
+ if (currentIndexKey != null)
+ seekToNextRow();
}
}
}
@@@ -297,6 -304,46 +312,46 @@@
}
}
+ private void updateIndexKey()
+ {
+ currentIndexKey = nextIndexKey;
+ currentRowPositionFromIndex = nextRowPositionFromIndex;
+ try
+ {
+ nextIndexKey = indexFile.isEOF() ? null : ByteBufferUtil.readWithShortLength(indexFile);
+ nextRowPositionFromIndex = indexFile.isEOF()
+ ? dataFile.length()
- : sstable.metadata.comparator.rowIndexEntrySerializer().deserialize(indexFile, sstable.descriptor.version).position;
++ : rowIndexEntrySerializer.deserialize(indexFile, sstable.descriptor.version).position;
+ }
+ catch (Throwable th)
+ {
+ JVMStabilityInspector.inspectThrowable(th);
+ outputHandler.warn("Error reading index file", th);
+ nextIndexKey = null;
+ nextRowPositionFromIndex = dataFile.length();
+ }
+ }
+
+ private void seekToNextRow()
+ {
+ while(nextRowPositionFromIndex < dataFile.length())
+ {
+ try
+ {
+ dataFile.seek(nextRowPositionFromIndex);
+ return;
+ }
+ catch (Throwable th)
+ {
+ throwIfFatal(th);
+ outputHandler.warn(String.format("Failed to seek to next row position %d", nextRowPositionFromIndex), th);
+ badRows++;
+ }
+
+ updateIndexKey();
+ }
+ }
+
private void saveOutOfOrderRow(DecoratedKey prevKey, DecoratedKey key, SSTableIdentityIterator atoms)
{
// TODO bitch if the row is too large? if it is there's not much we can do ...
http://git-wip-us.apache.org/repos/asf/cassandra/blob/49b08989/src/java/org/apache/cassandra/io/compress/CompressedRandomAccessReader.java
----------------------------------------------------------------------
diff --cc src/java/org/apache/cassandra/io/compress/CompressedRandomAccessReader.java
index 1b3cd06,184db9c..edf8c68
--- a/src/java/org/apache/cassandra/io/compress/CompressedRandomAccessReader.java
+++ b/src/java/org/apache/cassandra/io/compress/CompressedRandomAccessReader.java
@@@ -70,226 -60,87 +70,228 @@@ public class CompressedRandomAccessRead
private ByteBuffer compressed;
// re-use single crc object
- private final Checksum checksum;
+ private final Adler32 checksum;
// raw checksum bytes
- private final ByteBuffer checksumBytes = ByteBuffer.wrap(new byte[4]);
+ private ByteBuffer checksumBytes;
- protected CompressedRandomAccessReader(String dataFilePath, CompressionMetadata metadata, PoolingSegmentedFile owner) throws FileNotFoundException
+ protected CompressedRandomAccessReader(ChannelProxy channel, CompressionMetadata metadata, PoolingSegmentedFile owner) throws FileNotFoundException
{
- super(new File(dataFilePath), metadata.chunkLength(), metadata.compressedFileLength, owner);
+ super(channel, metadata.chunkLength(), metadata.compressedFileLength, metadata.compressor().useDirectOutputByteBuffers(), owner);
this.metadata = metadata;
- checksum = metadata.hasPostCompressionAdlerChecksums ? new Adler32() : new CRC32();
- compressed = ByteBuffer.wrap(new byte[metadata.compressor().initialCompressedBufferLength(metadata.chunkLength())]);
- }
+ checksum = new Adler32();
- @Override
- protected void reBuffer()
- {
- try
+ if (!useMmap)
{
- decompressChunk(metadata.chunkFor(current));
+ compressed = ByteBuffer.wrap(new byte[metadata.compressor().initialCompressedBufferLength(metadata.chunkLength())]);
+ checksumBytes = ByteBuffer.wrap(new byte[4]);
}
- catch (CorruptBlockException e)
+ else
{
- throw new CorruptSSTableException(e, getPath());
+ try
+ {
+ createMappedSegments();
+ }
+ catch (IOException e)
+ {
+ throw new IOError(e);
+ }
}
- catch (IOException e)
+ }
+
+ private void createMappedSegments() throws IOException
+ {
+ chunkSegments = new TreeMap<>();
+ long offset = 0;
+ long lastSegmentOffset = 0;
+ long segmentSize = 0;
+
+ while (offset < metadata.dataLength)
{
- throw new FSReadError(e, getPath());
+ CompressionMetadata.Chunk chunk = metadata.chunkFor(offset);
+
+ //Reached a new mmap boundary
+ if (segmentSize + chunk.length + 4 > MAX_SEGMENT_SIZE)
+ {
+ chunkSegments.put(lastSegmentOffset, channel.map(FileChannel.MapMode.READ_ONLY, lastSegmentOffset, segmentSize));
+ lastSegmentOffset += segmentSize;
+ segmentSize = 0;
+ }
+
+ segmentSize += chunk.length + 4; //checksum
+ offset += metadata.chunkLength();
}
+
+ if (segmentSize > 0)
+ chunkSegments.put(lastSegmentOffset, channel.map(FileChannel.MapMode.READ_ONLY, lastSegmentOffset, segmentSize));
}
- private void decompressChunk(CompressionMetadata.Chunk chunk) throws IOException
+ protected ByteBuffer allocateBuffer(int bufferSize, boolean useDirect)
{
- if (channel.position() != chunk.offset)
- channel.position(chunk.offset);
+ assert Integer.bitCount(bufferSize) == 1;
+ return useMmap && useDirect
+ ? ByteBuffer.allocateDirect(bufferSize)
+ : ByteBuffer.allocate(bufferSize);
+ }
- if (compressed.capacity() < chunk.length)
- compressed = ByteBuffer.wrap(new byte[chunk.length]);
- else
- compressed.clear();
- compressed.limit(chunk.length);
+ @Override
+ public void deallocate()
+ {
+ super.deallocate();
- if (channel.read(compressed) != chunk.length)
- throw new CorruptBlockException(getPath(), chunk);
+ if (chunkSegments != null)
+ {
+ for (Map.Entry<Long, MappedByteBuffer> entry : chunkSegments.entrySet())
+ {
+ FileUtils.clean(entry.getValue());
+ }
+ }
- // technically flip() is unnecessary since all the remaining work uses the raw array, but if that changes
- // in the future this will save a lot of hair-pulling
- compressed.flip();
+ chunkSegments = null;
+ }
+
+ private void reBufferStandard()
+ {
try
{
- validBufferBytes = metadata.compressor().uncompress(compressed.array(), 0, chunk.length, buffer, 0);
+ long position = current();
+ assert position < metadata.dataLength;
+
+ CompressionMetadata.Chunk chunk = metadata.chunkFor(position);
+
+ if (compressed.capacity() < chunk.length)
+ compressed = ByteBuffer.wrap(new byte[chunk.length]);
+ else
+ compressed.clear();
+ compressed.limit(chunk.length);
+
+ if (channel.read(compressed, chunk.offset) != chunk.length)
+ throw new CorruptBlockException(getPath(), chunk);
+
+ // technically flip() is unnecessary since all the remaining work uses the raw array, but if that changes
+ // in the future this will save a lot of hair-pulling
+ compressed.flip();
+ buffer.clear();
+ int decompressedBytes;
+ try
+ {
+ decompressedBytes = metadata.compressor().uncompress(compressed.array(), 0, chunk.length, buffer.array(), 0);
+ buffer.limit(decompressedBytes);
+ }
+ catch (IOException e)
+ {
++ buffer.limit(0);
+ throw new CorruptBlockException(getPath(), chunk);
+ }
+
+ if (metadata.parameters.getCrcCheckChance() > ThreadLocalRandom.current().nextDouble())
+ {
+
+ checksum.update(compressed.array(), 0, chunk.length);
+
+ if (checksum(chunk) != (int) checksum.getValue())
+ throw new CorruptBlockException(getPath(), chunk);
+
+ // reset checksum object back to the original (blank) state
+ checksum.reset();
+ }
+
+ // buffer offset is always aligned
+ bufferOffset = position & ~(buffer.capacity() - 1);
+ buffer.position((int) (position - bufferOffset));
+ // the length() can be provided at construction time, to override the true (uncompressed) length of the file;
+ // this is permitted to occur within a compressed segment, so we truncate validBufferBytes if we cross the imposed length
+ if (bufferOffset + buffer.limit() > length())
+ buffer.limit((int)(length() - bufferOffset));
+ }
+ catch (CorruptBlockException e)
+ {
+ throw new CorruptSSTableException(e, getPath());
}
catch (IOException e)
{
- throw new CorruptBlockException(getPath(), chunk, e);
+ throw new FSReadError(e, getPath());
}
+ }
- if (metadata.parameters.getCrcCheckChance() > ThreadLocalRandom.current().nextDouble())
+ private void reBufferMmap()
+ {
+ try
{
+ long position = current();
+ assert position < metadata.dataLength;
- if (metadata.hasPostCompressionAdlerChecksums)
+ CompressionMetadata.Chunk chunk = metadata.chunkFor(position);
+
+ Map.Entry<Long, MappedByteBuffer> entry = chunkSegments.floorEntry(chunk.offset);
+ long segmentOffset = entry.getKey();
+ int chunkOffset = Ints.checkedCast(chunk.offset - segmentOffset);
+ MappedByteBuffer compressedChunk = entry.getValue();
+
+ compressedChunk.position(chunkOffset);
+ compressedChunk.limit(chunkOffset + chunk.length);
+ compressedChunk.mark();
+
+ buffer.clear();
+ int decompressedBytes;
+ try
{
- checksum.update(compressed.array(), 0, chunk.length);
+ decompressedBytes = metadata.compressor().uncompress(compressedChunk, buffer);
+ buffer.limit(decompressedBytes);
}
- else
+ catch (IOException e)
+ {
++ buffer.limit(0);
+ throw new CorruptBlockException(getPath(), chunk);
+ }
+ finally
{
- checksum.update(buffer, 0, validBufferBytes);
+ compressedChunk.limit(compressedChunk.capacity());
}
- if (checksum(chunk) != (int) checksum.getValue())
- throw new CorruptBlockException(getPath(), chunk);
+ if (metadata.parameters.getCrcCheckChance() > ThreadLocalRandom.current().nextDouble())
+ {
+ compressedChunk.reset();
+ compressedChunk.limit(chunkOffset + chunk.length);
+
+ FBUtilities.directCheckSum(checksum, compressedChunk);
+
+ compressedChunk.limit(compressedChunk.capacity());
- // reset checksum object back to the original (blank) state
- checksum.reset();
+
+ if (compressedChunk.getInt() != (int) checksum.getValue())
+ throw new CorruptBlockException(getPath(), chunk);
+
+ // reset checksum object back to the original (blank) state
+ checksum.reset();
+ }
+
+ // buffer offset is always aligned
+ bufferOffset = position & ~(buffer.capacity() - 1);
+ buffer.position((int) (position - bufferOffset));
+ // the length() can be provided at construction time, to override the true (uncompressed) length of the file;
+ // this is permitted to occur within a compressed segment, so we truncate validBufferBytes if we cross the imposed length
+ if (bufferOffset + buffer.limit() > length())
+ buffer.limit((int)(length() - bufferOffset));
+ }
+ catch (CorruptBlockException e)
+ {
+ throw new CorruptSSTableException(e, getPath());
}
- // buffer offset is always aligned
- bufferOffset = current & ~(buffer.length - 1);
- // the length() can be provided at construction time, to override the true (uncompressed) length of the file;
- // this is permitted to occur within a compressed segment, so we truncate validBufferBytes if we cross the imposed length
- if (bufferOffset + validBufferBytes > length())
- validBufferBytes = (int)(length() - bufferOffset);
+ }
+
+ @Override
+ protected void reBuffer()
+ {
+ if (useMmap)
+ {
+ reBufferMmap();
+ }
+ else
+ {
+ reBufferStandard();
+ }
}
private int checksum(CompressionMetadata.Chunk chunk) throws IOException
http://git-wip-us.apache.org/repos/asf/cassandra/blob/49b08989/test/unit/org/apache/cassandra/SchemaLoader.java
----------------------------------------------------------------------
diff --cc test/unit/org/apache/cassandra/SchemaLoader.java
index 8f9df8f,c6a3855..46f4a9a
--- a/test/unit/org/apache/cassandra/SchemaLoader.java
+++ b/test/unit/org/apache/cassandra/SchemaLoader.java
@@@ -435,26 -415,16 +435,31 @@@ public class SchemaLoade
private static CFMetaData jdbcCFMD(String ksName, String cfName, AbstractType comp)
{
- return CFMetaData.denseCFMetaData(ksName, cfName, comp).defaultValidator(comp);
+ return CFMetaData.denseCFMetaData(ksName, cfName, comp).defaultValidator(comp).compressionParameters(getCompressionParameters());
}
- private static CFMetaData jdbcSparseCFMD(String ksName, String cfName, AbstractType comp)
+ public static CFMetaData jdbcSparseCFMD(String ksName, String cfName, AbstractType comp)
{
- return CFMetaData.sparseCFMetaData(ksName, cfName, comp).defaultValidator(comp);
+ return CFMetaData.sparseCFMetaData(ksName, cfName, comp).defaultValidator(comp).compressionParameters(getCompressionParameters());
}
- public static void cleanupAndLeaveDirs()
+ public static CompressionParameters getCompressionParameters()
{
++ return getCompressionParameters(null);
++ }
++
++ public static CompressionParameters getCompressionParameters(Integer chunkSize)
++ {
+ if (Boolean.parseBoolean(System.getProperty("cassandra.test.compression", "false")))
- return new CompressionParameters(SnappyCompressor.instance);
++ return new CompressionParameters(SnappyCompressor.instance, chunkSize, Collections.<String, String>emptyMap());
+ else
+ return new CompressionParameters(null);
+ }
+
+ public static void cleanupAndLeaveDirs() throws IOException
+ {
+ // We need to stop and unmap all CLS instances prior to cleanup() or we'll get failures on Windows.
+ CommitLog.instance.stopUnsafe(true);
mkdirs();
cleanup();
mkdirs();
http://git-wip-us.apache.org/repos/asf/cassandra/blob/49b08989/test/unit/org/apache/cassandra/db/ScrubTest.java
----------------------------------------------------------------------
diff --cc test/unit/org/apache/cassandra/db/ScrubTest.java
index dce8d14,a19c76d..128d1b0
--- a/test/unit/org/apache/cassandra/db/ScrubTest.java
+++ b/test/unit/org/apache/cassandra/db/ScrubTest.java
@@@ -21,18 -21,19 +21,13 @@@ package org.apache.cassandra.db
*/
import java.io.*;
+import java.lang.reflect.Field;
+import java.lang.reflect.Modifier;
import java.nio.ByteBuffer;
- import java.util.Arrays;
-import java.util.Collections;
--import java.util.HashSet;
--import java.util.Iterator;
--import java.util.List;
--import java.util.Set;
++import java.util.*;
import java.util.concurrent.ExecutionException;
- import org.apache.cassandra.io.compress.CompressionParameters;
- import org.apache.cassandra.io.compress.SnappyCompressor;
-import org.apache.cassandra.cql3.QueryProcessor;
-import org.apache.cassandra.exceptions.ConfigurationException;
-import org.apache.cassandra.exceptions.RequestExecutionException;
+ import org.apache.cassandra.io.compress.CompressionMetadata;
-import org.apache.cassandra.utils.UUIDGen;
import org.apache.commons.lang3.StringUtils;
import org.junit.BeforeClass;
import org.junit.Test;
@@@ -54,57 -51,31 +49,57 @@@ import org.apache.cassandra.exceptions.
import org.apache.cassandra.exceptions.WriteTimeoutException;
import org.apache.cassandra.io.sstable.Component;
import org.apache.cassandra.io.sstable.Descriptor;
-import org.apache.cassandra.io.sstable.SSTableReader;
+import org.apache.cassandra.io.sstable.SSTableRewriter;
+import org.apache.cassandra.io.sstable.format.SSTableFormat;
+import org.apache.cassandra.io.sstable.format.SSTableReader;
+import org.apache.cassandra.locator.SimpleStrategy;
+import org.apache.cassandra.OrderedJUnit4ClassRunner;
+import org.apache.cassandra.SchemaLoader;
+import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.Util;
import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.UUIDGen;
+
+import static org.junit.Assert.*;
++import static org.junit.Assume.assumeTrue;
import static org.apache.cassandra.Util.cellname;
import static org.apache.cassandra.Util.column;
- import static org.junit.Assert.assertEquals;
- import static org.junit.Assert.fail;
- import static org.junit.Assume.assumeTrue;
-import static junit.framework.Assert.assertNotNull;
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
-import static org.junit.Assume.assumeTrue;
-
@RunWith(OrderedJUnit4ClassRunner.class)
-public class ScrubTest extends SchemaLoader
+public class ScrubTest
{
- public String KEYSPACE = "Keyspace1";
- public String CF = "Standard1";
- public String CF3 = "Standard2";
- public String COUNTER_CF = "Counter1";
- private static Integer COMPRESSION_CHUNK_LENGTH = 4096;
+ public static final String KEYSPACE = "Keyspace1";
+ public static final String CF = "Standard1";
+ public static final String CF2 = "Standard2";
+ public static final String CF3 = "Standard3";
+ public static final String COUNTER_CF = "Counter1";
+ public static final String CF_UUID = "UUIDKeys";
+ public static final String CF_INDEX1 = "Indexed1";
+ public static final String CF_INDEX2 = "Indexed2";
+
+ public static final String COL_KEYS_INDEX = "birthdate";
+ public static final String COL_COMPOSITES_INDEX = "col1";
+ public static final String COL_NON_INDEX = "notanindexcol";
+
++ public static final Integer COMPRESSION_CHUNK_LENGTH = 4096;
+
@BeforeClass
- public static void loadSchema() throws ConfigurationException
+ public static void defineSchema() throws ConfigurationException
{
- loadSchema(COMPRESSION_CHUNK_LENGTH);
+ SchemaLoader.loadSchema();
+ SchemaLoader.createKeyspace(KEYSPACE,
+ SimpleStrategy.class,
+ KSMetaData.optsWithRF(1),
+ SchemaLoader.standardCFMD(KEYSPACE, CF),
+ SchemaLoader.standardCFMD(KEYSPACE, CF2),
+ SchemaLoader.standardCFMD(KEYSPACE, CF3),
+ SchemaLoader.standardCFMD(KEYSPACE, COUNTER_CF)
+ .defaultValidator(CounterColumnType.instance)
- .compressionParameters(SchemaLoader.getCompressionParameters()),
++ .compressionParameters(SchemaLoader.getCompressionParameters(COMPRESSION_CHUNK_LENGTH)),
+ SchemaLoader.standardCFMD(KEYSPACE, CF_UUID).keyValidator(UUIDType.instance),
+ SchemaLoader.indexCFMD(KEYSPACE, CF_INDEX1, true),
+ SchemaLoader.compositeIndexCFMD(KEYSPACE, CF_INDEX2, true));
}
@Test
@@@ -140,28 -112,88 +136,89 @@@
ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(COUNTER_CF);
cfs.clearUnsafe();
- fillCounterCF(cfs, 2);
+ fillCounterCF(cfs, numPartitions);
- List<Row> rows = cfs.getRangeSlice(Util.range("", ""), null, new IdentityQueryFilter(), 1000);
- assertEquals(2, rows.size());
+ List<Row> rows = cfs.getRangeSlice(Util.range("", ""), null, new IdentityQueryFilter(), numPartitions*10);
+ assertEquals(numPartitions, rows.size());
+
- overrdeWithGarbage(cfs, ByteBufferUtil.bytes("0"), ByteBufferUtil.bytes("1"));
+ assertEquals(1, cfs.getSSTables().size());
SSTableReader sstable = cfs.getSSTables().iterator().next();
+ //make sure to override at most 1 chunk when compression is enabled
+ overrideWithGarbage(sstable, ByteBufferUtil.bytes("0"), ByteBufferUtil.bytes("1"));
+
// with skipCorrupted == false, the scrub is expected to fail
- Scrubber scrubber = new Scrubber(cfs, sstable, false, false);
- try
+ try(Scrubber scrubber = new Scrubber(cfs, sstable, false, false))
{
scrubber.scrub();
fail("Expected a CorruptSSTableException to be thrown");
}
catch (IOError err) {}
-- // with skipCorrupted == true, the corrupt row will be skipped
++ // with skipCorrupted == true, the corrupt rows will be skipped
+ Scrubber.ScrubResult scrubResult;
- scrubber = new Scrubber(cfs, sstable, true, false);
- scrubResult = scrubber.scrubWithResult();
- scrubber.close();
+ try(Scrubber scrubber = new Scrubber(cfs, sstable, true, false))
+ {
++ scrubResult = scrubber.scrubWithResult();
++ }
+
+ assertNotNull(scrubResult);
+
+ boolean compression = Boolean.parseBoolean(System.getProperty("cassandra.test.compression", "false"));
+ if (compression)
+ {
+ assertEquals(0, scrubResult.emptyRows);
+ assertEquals(numPartitions, scrubResult.badRows + scrubResult.goodRows);
+ //because we only corrupted 1 chunk and we chose enough partitions to cover at least 3 chunks
+ assertTrue(scrubResult.goodRows >= scrubResult.badRows * 2);
+ }
+ else
+ {
+ assertEquals(0, scrubResult.emptyRows);
+ assertEquals(1, scrubResult.badRows);
+ assertEquals(numPartitions-1, scrubResult.goodRows);
+ }
+ assertEquals(1, cfs.getSSTables().size());
+
+ rows = cfs.getRangeSlice(Util.range("", ""), null, new IdentityQueryFilter(), 1000);
+ assertEquals(scrubResult.goodRows, rows.size());
+ }
+
+ @Test
+ public void testScrubCorruptedRowInSmallFile() throws IOException, WriteTimeoutException
+ {
+ // cannot test this with compression
+ assumeTrue(!Boolean.parseBoolean(System.getProperty("cassandra.test.compression", "false")));
+
+ CompactionManager.instance.disableAutoCompaction();
+ Keyspace keyspace = Keyspace.open(KEYSPACE);
+ ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(COUNTER_CF);
+ cfs.clearUnsafe();
+
+ fillCounterCF(cfs, 2);
+
+ List<Row> rows = cfs.getRangeSlice(Util.range("", ""), null, new IdentityQueryFilter(), 1000);
+ assertEquals(2, rows.size());
+
+ SSTableReader sstable = cfs.getSSTables().iterator().next();
+
+ // overwrite one row with garbage
+ overrideWithGarbage(sstable, ByteBufferUtil.bytes("0"), ByteBufferUtil.bytes("1"));
+
+ // with skipCorrupted == false, the scrub is expected to fail
+ Scrubber scrubber = new Scrubber(cfs, sstable, false, false);
+ try
+ {
scrubber.scrub();
+ fail("Expected a CorruptSSTableException to be thrown");
}
+ catch (IOError err) {}
+
+ // with skipCorrupted == true, the corrupt row will be skipped
+ scrubber = new Scrubber(cfs, sstable, true, false);
+ scrubber.scrub();
+ scrubber.close();
assertEquals(1, cfs.getSSTables().size());
// verify that we can read all of the rows, and there is now one less row
@@@ -170,21 -202,34 +227,49 @@@
}
@Test
+ public void testScrubOneRowWithCorruptedKey() throws IOException, ExecutionException, InterruptedException, ConfigurationException
+ {
+ // cannot test this with compression
+ assumeTrue(!Boolean.parseBoolean(System.getProperty("cassandra.test.compression", "false")));
+
+ CompactionManager.instance.disableAutoCompaction();
+ Keyspace keyspace = Keyspace.open(KEYSPACE);
+ ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(CF);
+ cfs.clearUnsafe();
+
+ List<Row> rows;
+
+ // insert data and verify we get it back w/ range query
+ fillCF(cfs, 4);
+ rows = cfs.getRangeSlice(Util.range("", ""), null, new IdentityQueryFilter(), 1000);
+ assertEquals(4, rows.size());
+
+ SSTableReader sstable = cfs.getSSTables().iterator().next();
+ overrideWithGarbage(sstable, 0, 2);
+
+ CompactionManager.instance.performScrub(cfs, false);
+
+ // check data is still there
+ rows = cfs.getRangeSlice(Util.range("", ""), null, new IdentityQueryFilter(), 1000);
+ assertEquals(4, rows.size());
+ }
+
+ @Test
+ public void testScrubCorruptedCounterRowNoEarlyOpen() throws IOException, WriteTimeoutException
+ {
+ long oldOpenVal = SSTableRewriter.getOpenInterval();
+ try
+ {
+ SSTableRewriter.overrideOpenInterval(Long.MAX_VALUE);
+ testScrubCorruptedCounterRow();
+ }
+ finally
+ {
+ SSTableRewriter.overrideOpenInterval(oldOpenVal);
+ }
+ }
+
+ @Test
public void testScrubDeletedRow() throws ExecutionException, InterruptedException
{
CompactionManager.instance.disableAutoCompaction();
@@@ -432,129 -471,4 +537,129 @@@
assertEquals("bar", iter.next().getString("c"));
assertEquals("boo", iter.next().getString("c"));
}
+
+ @Test /* CASSANDRA-5174 */
+ public void testScrubKeysIndex_preserveOrder() throws IOException, ExecutionException, InterruptedException
+ {
+ //If the partitioner preserves the order then SecondaryIndex uses BytesType comparator,
+ // otherwise it uses LocalByPartitionerType
+ setKeyComparator(BytesType.instance);
+ testScrubIndex(CF_INDEX1, COL_KEYS_INDEX, false, true);
+ }
+
+ @Test /* CASSANDRA-5174 */
+ public void testScrubCompositeIndex_preserveOrder() throws IOException, ExecutionException, InterruptedException
+ {
+ setKeyComparator(BytesType.instance);
+ testScrubIndex(CF_INDEX2, COL_COMPOSITES_INDEX, true, true);
+ }
+
+ @Test /* CASSANDRA-5174 */
+ public void testScrubKeysIndex() throws IOException, ExecutionException, InterruptedException
+ {
+ setKeyComparator(new LocalByPartionerType(StorageService.getPartitioner()));
+ testScrubIndex(CF_INDEX1, COL_KEYS_INDEX, false, true);
+ }
+
+ @Test /* CASSANDRA-5174 */
+ public void testScrubCompositeIndex() throws IOException, ExecutionException, InterruptedException
+ {
+ setKeyComparator(new LocalByPartionerType(StorageService.getPartitioner()));
+ testScrubIndex(CF_INDEX2, COL_COMPOSITES_INDEX, true, true);
+ }
+
+ @Test /* CASSANDRA-5174 */
+ public void testFailScrubKeysIndex() throws IOException, ExecutionException, InterruptedException
+ {
+ testScrubIndex(CF_INDEX1, COL_KEYS_INDEX, false, false);
+ }
+
+ @Test /* CASSANDRA-5174 */
+ public void testFailScrubCompositeIndex() throws IOException, ExecutionException, InterruptedException
+ {
+ testScrubIndex(CF_INDEX2, COL_COMPOSITES_INDEX, true, false);
+ }
+
+ @Test /* CASSANDRA-5174 */
+ public void testScrubTwice() throws IOException, ExecutionException, InterruptedException
+ {
+ testScrubIndex(CF_INDEX1, COL_KEYS_INDEX, false, true, true);
+ }
+
+ /** The SecondaryIndex class is used for custom indexes so to avoid
+ * making a public final field into a private field with getters
+ * and setters, we resort to this hack in order to test it properly
+ * since it can have two values which influence the scrubbing behavior.
+ * @param comparator - the key comparator we want to test
+ */
+ private void setKeyComparator(AbstractType<?> comparator)
+ {
+ try
+ {
+ Field keyComparator = SecondaryIndex.class.getDeclaredField("keyComparator");
+ keyComparator.setAccessible(true);
+ int modifiers = keyComparator.getModifiers();
+ Field modifierField = keyComparator.getClass().getDeclaredField("modifiers");
+ modifiers = modifiers & ~Modifier.FINAL;
+ modifierField.setAccessible(true);
+ modifierField.setInt(keyComparator, modifiers);
+
+ keyComparator.set(null, comparator);
+ }
+ catch (Exception ex)
+ {
+ fail("Failed to change key comparator in secondary index : " + ex.getMessage());
+ ex.printStackTrace();
+ }
+ }
+
+ private void testScrubIndex(String cfName, String colName, boolean composite, boolean ... scrubs)
+ throws IOException, ExecutionException, InterruptedException
+ {
+ CompactionManager.instance.disableAutoCompaction();
+ Keyspace keyspace = Keyspace.open(KEYSPACE);
+ ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(cfName);
+ cfs.clearUnsafe();
+
+ int numRows = 1000;
+ long[] colValues = new long [numRows * 2]; // each row has two columns
+ for (int i = 0; i < colValues.length; i+=2)
+ {
+ colValues[i] = (i % 4 == 0 ? 1L : 2L); // index column
+ colValues[i+1] = 3L; //other column
+ }
+ fillIndexCF(cfs, composite, colValues);
+
+ // check index
+ IndexExpression expr = new IndexExpression(ByteBufferUtil.bytes(colName), Operator.EQ, ByteBufferUtil.bytes(1L));
+ List<Row> rows = cfs.search(Util.range("", ""), Arrays.asList(expr), new IdentityQueryFilter(), numRows);
+ assertNotNull(rows);
+ assertEquals(numRows / 2, rows.size());
+
+ // scrub index
+ Set<ColumnFamilyStore> indexCfss = cfs.indexManager.getIndexesBackedByCfs();
+ assertTrue(indexCfss.size() == 1);
+ for(ColumnFamilyStore indexCfs : indexCfss)
+ {
+ for (int i = 0; i < scrubs.length; i++)
+ {
+ boolean failure = !scrubs[i];
+ if (failure)
+ { //make sure the next scrub fails
- overrdeWithGarbage(indexCfs, ByteBufferUtil.bytes(1L), ByteBufferUtil.bytes(2L));
++ overrideWithGarbage(indexCfs.getSSTables().iterator().next(), ByteBufferUtil.bytes(1L), ByteBufferUtil.bytes(2L));
+ }
+ CompactionManager.AllSSTableOpStatus result = indexCfs.scrub(false, false, true);
+ assertEquals(failure ?
+ CompactionManager.AllSSTableOpStatus.ABORTED :
+ CompactionManager.AllSSTableOpStatus.SUCCESSFUL,
+ result);
+ }
+ }
+
+
+ // check index is still working
+ rows = cfs.search(Util.range("", ""), Arrays.asList(expr), new IdentityQueryFilter(), numRows);
+ assertNotNull(rows);
+ assertEquals(numRows / 2, rows.size());
+ }
}