mirror of https://github.com/apache/cassandra
Merge branch 'cassandra-3.9' into trunk
* cassandra-3.9: Fix problem with undeleteable rows on upgrade to new sstable format.
This commit is contained in:
commit
d314b60576
|
|
@ -14,7 +14,8 @@
|
|||
|
||||
|
||||
3.9
|
||||
|
||||
Merged from 3.0:
|
||||
* Fix problem with undeleteable rows on upgrade to new sstable format (CASSANDRA-12144)
|
||||
|
||||
3.8
|
||||
* Fix hdr logging for single operation workloads (CASSANDRA-12145)
|
||||
|
|
|
|||
|
|
@ -1207,9 +1207,26 @@ public abstract class LegacyLayout
|
|||
{
|
||||
if (tombstone.isRowDeletion(metadata))
|
||||
{
|
||||
// If we're already within a row, it can't be the same one
|
||||
if (clustering != null)
|
||||
{
|
||||
// If we're already in the row, there might be a chance that there were two range tombstones
|
||||
// written, as 2.x storage format does not guarantee just one range tombstone, unlike 3.x.
|
||||
// We have to make sure that clustering matches, which would mean that tombstone is for the
|
||||
// same row.
|
||||
if (rowDeletion != null && clustering.equals(tombstone.start.getAsClustering(metadata)))
|
||||
{
|
||||
// If the tombstone superceeds the previous delete, we discard the previous one
|
||||
if (tombstone.deletionTime.supersedes(rowDeletion.deletionTime))
|
||||
{
|
||||
builder.addRowDeletion(Row.Deletion.regular(tombstone.deletionTime));
|
||||
rowDeletion = tombstone;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
// If we're already within a row and there was no delete written before that one, it can't be the same one
|
||||
return false;
|
||||
}
|
||||
|
||||
clustering = tombstone.start.getAsClustering(metadata);
|
||||
builder.newRow(clustering);
|
||||
|
|
|
|||
|
|
@ -34,11 +34,7 @@ import org.apache.cassandra.io.sstable.format.SSTableWriter;
|
|||
import org.apache.cassandra.io.util.FileUtils;
|
||||
import org.apache.cassandra.io.util.RandomAccessReader;
|
||||
import org.apache.cassandra.service.ActiveRepairService;
|
||||
import org.apache.cassandra.utils.ByteBufferUtil;
|
||||
import org.apache.cassandra.utils.FBUtilities;
|
||||
import org.apache.cassandra.utils.JVMStabilityInspector;
|
||||
import org.apache.cassandra.utils.OutputHandler;
|
||||
import org.apache.cassandra.utils.UUIDGen;
|
||||
import org.apache.cassandra.utils.*;
|
||||
|
||||
public class Scrubber implements Closeable
|
||||
{
|
||||
|
|
@ -214,7 +210,7 @@ public class Scrubber implements Closeable
|
|||
if (indexFile != null && dataStart != dataStartFromIndex)
|
||||
outputHandler.warn(String.format("Data file row position %d differs from index file row position %d", dataStart, dataStartFromIndex));
|
||||
|
||||
try (UnfilteredRowIterator iterator = withValidation(new SSTableIdentityIterator(sstable, dataFile, key), dataFile.getPath()))
|
||||
try (UnfilteredRowIterator iterator = withValidation(new RowMergingSSTableIterator(sstable, dataFile, key), dataFile.getPath()))
|
||||
{
|
||||
if (prevKey != null && prevKey.compareTo(key) > 0)
|
||||
{
|
||||
|
|
@ -468,4 +464,46 @@ public class Scrubber implements Closeable
|
|||
this.emptyRows = scrubber.emptyRows;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* During 2.x migration, under some circumstances rows might have gotten duplicated.
|
||||
* Merging iterator merges rows with same clustering.
|
||||
*
|
||||
* For more details, refer to CASSANDRA-12144.
|
||||
*/
|
||||
private static class RowMergingSSTableIterator extends SSTableIdentityIterator
|
||||
{
|
||||
RowMergingSSTableIterator(SSTableReader sstable, RandomAccessReader file, DecoratedKey key)
|
||||
{
|
||||
super(sstable, file, key);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected Unfiltered doCompute()
|
||||
{
|
||||
if (!iterator.hasNext())
|
||||
return endOfData();
|
||||
|
||||
Unfiltered next = iterator.next();
|
||||
if (!next.isRow())
|
||||
return next;
|
||||
|
||||
while (iterator.hasNext())
|
||||
{
|
||||
Unfiltered peek = iterator.peek();
|
||||
// If there was a duplicate row, merge it.
|
||||
if (next.clustering().equals(peek.clustering()) && peek.isRow())
|
||||
{
|
||||
iterator.next(); // Make sure that the peeked item was consumed.
|
||||
next = Rows.merge((Row) next, (Row) peek, FBUtilities.nowInSeconds());
|
||||
}
|
||||
else
|
||||
{
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
return next;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -34,7 +34,7 @@ public class SSTableIdentityIterator extends AbstractIterator<Unfiltered> implem
|
|||
private final DeletionTime partitionLevelDeletion;
|
||||
private final String filename;
|
||||
|
||||
private final SSTableSimpleIterator iterator;
|
||||
protected final SSTableSimpleIterator iterator;
|
||||
private final Row staticRow;
|
||||
|
||||
/**
|
||||
|
|
@ -97,7 +97,7 @@ public class SSTableIdentityIterator extends AbstractIterator<Unfiltered> implem
|
|||
{
|
||||
try
|
||||
{
|
||||
return iterator.hasNext() ? iterator.next() : endOfData();
|
||||
return doCompute();
|
||||
}
|
||||
catch (IndexOutOfBoundsException e)
|
||||
{
|
||||
|
|
@ -118,6 +118,11 @@ public class SSTableIdentityIterator extends AbstractIterator<Unfiltered> implem
|
|||
}
|
||||
}
|
||||
|
||||
protected Unfiltered doCompute()
|
||||
{
|
||||
return iterator.hasNext() ? iterator.next() : endOfData();
|
||||
}
|
||||
|
||||
public void close()
|
||||
{
|
||||
// creator is responsible for closing file when finished
|
||||
|
|
|
|||
Binary file not shown.
Binary file not shown.
|
|
@ -0,0 +1 @@
|
|||
408097082
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
|
|
@ -0,0 +1,8 @@
|
|||
CompressionInfo.db
|
||||
Digest.adler32
|
||||
TOC.txt
|
||||
Filter.db
|
||||
Data.db
|
||||
Index.db
|
||||
Statistics.db
|
||||
Summary.db
|
||||
Binary file not shown.
Binary file not shown.
|
|
@ -0,0 +1 @@
|
|||
3332428483
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
Binary file not shown.
|
|
@ -0,0 +1,8 @@
|
|||
Statistics.db
|
||||
Digest.crc32
|
||||
Summary.db
|
||||
Index.db
|
||||
TOC.txt
|
||||
CompressionInfo.db
|
||||
Filter.db
|
||||
Data.db
|
||||
|
|
@ -20,6 +20,9 @@ package org.apache.cassandra.db;
|
|||
|
||||
import java.io.*;
|
||||
import java.nio.ByteBuffer;
|
||||
import java.nio.file.Files;
|
||||
import java.nio.file.Path;
|
||||
import java.nio.file.Paths;
|
||||
import java.util.*;
|
||||
import java.util.concurrent.ExecutionException;
|
||||
|
||||
|
|
@ -44,8 +47,7 @@ import org.apache.cassandra.db.marshal.UUIDType;
|
|||
import org.apache.cassandra.db.partitions.Partition;
|
||||
import org.apache.cassandra.db.partitions.PartitionUpdate;
|
||||
import org.apache.cassandra.db.rows.EncodingStats;
|
||||
import org.apache.cassandra.dht.ByteOrderedPartitioner;
|
||||
import org.apache.cassandra.dht.IPartitioner;
|
||||
import org.apache.cassandra.dht.*;
|
||||
import org.apache.cassandra.exceptions.ConfigurationException;
|
||||
import org.apache.cassandra.exceptions.RequestExecutionException;
|
||||
import org.apache.cassandra.exceptions.WriteTimeoutException;
|
||||
|
|
@ -68,6 +70,8 @@ import static org.junit.Assume.assumeTrue;
|
|||
@RunWith(OrderedJUnit4ClassRunner.class)
|
||||
public class ScrubTest
|
||||
{
|
||||
public static final String INVALID_LEGACY_SSTABLE_ROOT_PROP = "invalid-legacy-sstable-root";
|
||||
|
||||
public static final String KEYSPACE = "Keyspace1";
|
||||
public static final String CF = "Standard1";
|
||||
public static final String CF2 = "Standard2";
|
||||
|
|
@ -663,4 +667,83 @@ public class ScrubTest
|
|||
return dataFile.position();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Tests with invalid sstables (containing duplicate entries in 2.0 and 3.0 storage format),
|
||||
* that were caused by upgrading from 2.x with duplicate range tombstones.
|
||||
*
|
||||
* See CASSANDRA-12144 for details.
|
||||
*/
|
||||
@Test
|
||||
public void testFilterOutDuplicates() throws Exception
|
||||
{
|
||||
DatabaseDescriptor.setPartitionerUnsafe(Murmur3Partitioner.instance);
|
||||
QueryProcessor.process(String.format("CREATE TABLE \"%s\".cf_with_duplicates_3_0 (a int, b int, c int, PRIMARY KEY (a, b))", KEYSPACE), ConsistencyLevel.ONE);
|
||||
|
||||
Keyspace keyspace = Keyspace.open(KEYSPACE);
|
||||
ColumnFamilyStore cfs = keyspace.getColumnFamilyStore("cf_with_duplicates_3_0");
|
||||
|
||||
Path legacySSTableRoot = Paths.get(System.getProperty(INVALID_LEGACY_SSTABLE_ROOT_PROP),
|
||||
"Keyspace1",
|
||||
"cf_with_duplicates_3_0");
|
||||
|
||||
for (String filename : new String[]{ "mb-3-big-CompressionInfo.db",
|
||||
"mb-3-big-Digest.crc32",
|
||||
"mb-3-big-Index.db",
|
||||
"mb-3-big-Summary.db",
|
||||
"mb-3-big-Data.db",
|
||||
"mb-3-big-Filter.db",
|
||||
"mb-3-big-Statistics.db",
|
||||
"mb-3-big-TOC.txt" })
|
||||
{
|
||||
Files.copy(Paths.get(legacySSTableRoot.toString(), filename), cfs.getDirectories().getDirectoryForNewSSTables().toPath().resolve(filename));
|
||||
}
|
||||
|
||||
cfs.loadNewSSTables();
|
||||
|
||||
cfs.scrub(true, true, true, 1);
|
||||
|
||||
UntypedResultSet rs = QueryProcessor.executeInternal(String.format("SELECT * FROM \"%s\".cf_with_duplicates_3_0", KEYSPACE));
|
||||
assertEquals(1, rs.size());
|
||||
QueryProcessor.executeInternal(String.format("DELETE FROM \"%s\".cf_with_duplicates_3_0 WHERE a=1 AND b =2", KEYSPACE));
|
||||
rs = QueryProcessor.executeInternal(String.format("SELECT * FROM \"%s\".cf_with_duplicates_3_0", KEYSPACE));
|
||||
assertEquals(0, rs.size());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void testUpgradeSstablesWithDuplicates() throws Exception
|
||||
{
|
||||
DatabaseDescriptor.setPartitionerUnsafe(Murmur3Partitioner.instance);
|
||||
String cf = "cf_with_duplicates_2_0";
|
||||
QueryProcessor.process(String.format("CREATE TABLE \"%s\".%s (a int, b int, c int, PRIMARY KEY (a, b))", KEYSPACE, cf), ConsistencyLevel.ONE);
|
||||
|
||||
Keyspace keyspace = Keyspace.open(KEYSPACE);
|
||||
ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(cf);
|
||||
|
||||
Path legacySSTableRoot = Paths.get(System.getProperty(INVALID_LEGACY_SSTABLE_ROOT_PROP),
|
||||
"Keyspace1",
|
||||
cf);
|
||||
|
||||
for (String filename : new String[]{ "lb-1-big-CompressionInfo.db",
|
||||
"lb-1-big-Data.db",
|
||||
"lb-1-big-Digest.adler32",
|
||||
"lb-1-big-Filter.db",
|
||||
"lb-1-big-Index.db",
|
||||
"lb-1-big-Statistics.db",
|
||||
"lb-1-big-Summary.db",
|
||||
"lb-1-big-TOC.txt" })
|
||||
{
|
||||
Files.copy(Paths.get(legacySSTableRoot.toString(), filename), cfs.getDirectories().getDirectoryForNewSSTables().toPath().resolve(filename));
|
||||
}
|
||||
|
||||
cfs.loadNewSSTables();
|
||||
|
||||
cfs.sstablesRewrite(true, 1);
|
||||
|
||||
UntypedResultSet rs = QueryProcessor.executeInternal(String.format("SELECT * FROM \"%s\".%s", KEYSPACE, cf));
|
||||
assertEquals(1, rs.size());
|
||||
QueryProcessor.executeInternal(String.format("DELETE FROM \"%s\".%s WHERE a=1 AND b =2", KEYSPACE, cf));
|
||||
rs = QueryProcessor.executeInternal(String.format("SELECT * FROM \"%s\".%s", KEYSPACE, cf));
|
||||
assertEquals(0, rs.size());
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in New Issue