natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Merge branch 'cassandra-3.0' into cassandra-3.11

This commit is contained in:
Blake Eggleston 2018-11-06 11:17:47 -08:00
parent 69f8cc7d25 d60c78358b
commit a6a9dce157
16 changed files with 166 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,5 +1,6 @@
3.11.4
Merged from 3.0:
* Sstable min/max metadata can cause data loss (CASSANDRA-14861)
* Dropped columns can cause reverse sstable iteration to return prematurely (CASSANDRA-14838)
* Legacy sstables with multi block range tombstones create invalid bound sequences (CASSANDRA-14823)
* Expand range tombstone validation checks to multiple interim request stages (CASSANDRA-14824)

25
src/java/org/apache/cassandra/db/Slice.java
View File

@ -235,29 +235,8 @@ public class Slice
*/
public boolean intersects(ClusteringComparator comparator, List<ByteBuffer> minClusteringValues, List<ByteBuffer> maxClusteringValues)
{
// If this slice start after max or end before min, it can't intersect
if (start.compareTo(comparator, maxClusteringValues) > 0 || end.compareTo(comparator, minClusteringValues) < 0)
return false;
// We could safely return true here, but there's a minor optimization: if the first component
// of the slice is restricted to a single value (typically the slice is [4:5, 4:7]), we can
// check that the second component falls within the min/max for that component (and repeat for
// all components).
for (int j = 0; j < minClusteringValues.size() && j < maxClusteringValues.size(); j++)
{
ByteBuffer s = j < start.size() ? start.get(j) : null;
ByteBuffer f = j < end.size() ? end.get(j) : null;
// we already know the first component falls within its min/max range (otherwise we wouldn't get here)
if (j > 0 && (j < end.size() && comparator.compareComponent(j, f, minClusteringValues.get(j)) < 0 ||
j < start.size() && comparator.compareComponent(j, s, maxClusteringValues.get(j)) > 0))
return false;
// if this component isn't equal in the start and finish, we don't need to check any more
if (j >= start.size() || j >= end.size() || comparator.compareComponent(j, s, f) != 0)
break;
}
return true;
// If this slice starts after max clustering or ends before min clustering, it can't intersect
return start.compareTo(comparator, maxClusteringValues) <= 0 && end.compareTo(comparator, minClusteringValues) >= 0;
}
public String toString(CFMetaData metadata)

2
src/java/org/apache/cassandra/io/sstable/format/Version.java
View File

@ -74,6 +74,8 @@ public abstract class Version
public abstract boolean hasCommitLogIntervals();
public abstract boolean hasAccurateMinMax();
public String getVersion()
{
return version;

9
src/java/org/apache/cassandra/io/sstable/format/big/BigFormat.java
View File

@ -125,6 +125,7 @@ public class BigFormat implements SSTableFormat
// store rows natively
// mb (3.0.7, 3.7): commit log lower bound included
// mc (3.0.8, 3.9): commit log intervals included
// md (3.0.18, 3.11.4): corrected sstable min/max clustering
//
// NOTE: when adding a new version, please add that to LegacySSTableTest, too.
@ -146,6 +147,7 @@ public class BigFormat implements SSTableFormat
private final boolean hasOldBfHashOrder;
private final boolean hasCommitLogLowerBound;
private final boolean hasCommitLogIntervals;
private final boolean hasAccurateMinMax;
/**
* CASSANDRA-7066: compaction ancerstors are no longer used and have been removed.
@ -188,6 +190,7 @@ public class BigFormat implements SSTableFormat
hasCommitLogLowerBound = (version.compareTo("lb") >= 0 && version.compareTo("ma") < 0)
|| version.compareTo("mb") >= 0;
hasCommitLogIntervals = version.compareTo("mc") >= 0;
hasAccurateMinMax = version.compareTo("md") >= 0;
}
@Override
@ -262,6 +265,12 @@ public class BigFormat implements SSTableFormat
return hasCommitLogIntervals;
}
@Override
public boolean hasAccurateMinMax()
{
return hasAccurateMinMax;
}
@Override
public boolean storeRows()
{

25
src/java/org/apache/cassandra/io/sstable/metadata/MetadataCollector.java
View File

@ -19,11 +19,15 @@ package org.apache.cassandra.io.sstable.metadata;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.EnumMap;
import java.util.List;
import java.util.Map;
import com.google.common.base.Preconditions;
import com.google.common.collect.Maps;
import com.clearspring.analytics.stream.cardinality.HyperLogLogPlus;
import com.clearspring.analytics.stream.cardinality.ICardinality;
import org.apache.cassandra.db.*;
@ -93,8 +97,8 @@ public class MetadataCollector implements PartitionStatisticsCollector
protected double compressionRatio = NO_COMPRESSION_RATIO;
protected StreamingHistogram.StreamingHistogramBuilder estimatedTombstoneDropTime = defaultTombstoneDropTimeHistogramBuilder();
protected int sstableLevel;
protected ByteBuffer[] minClusteringValues;
protected ByteBuffer[] maxClusteringValues;
private ClusteringPrefix minClustering = ClusteringBound.TOP;
private ClusteringPrefix maxClustering = ClusteringBound.BOTTOM;
protected boolean hasLegacyCounterShards = false;
protected long totalColumnsSet;
protected long totalRows;
@ -112,8 +116,6 @@ public class MetadataCollector implements PartitionStatisticsCollector
{
this.comparator = comparator;
this.minClusteringValues = new ByteBuffer[comparator.size()];
this.maxClusteringValues = new ByteBuffer[comparator.size()];
}
public MetadataCollector(Iterable<SSTableReader> sstables, ClusteringComparator comparator, int level)
@ -228,14 +230,8 @@ public class MetadataCollector implements PartitionStatisticsCollector
public MetadataCollector updateClusteringValues(ClusteringPrefix clustering)
{
int size = clustering.size();
for (int i = 0; i < size; i++)
{
AbstractType<?> type = comparator.subtype(i);
ByteBuffer newValue = clustering.get(i);
minClusteringValues[i] = maybeMinimize(min(minClusteringValues[i], newValue, type));
maxClusteringValues[i] = maybeMinimize(max(maxClusteringValues[i], newValue, type));
}
minClustering = comparator.compare(clustering, minClustering) < 0 ? clustering : minClustering;
maxClustering = comparator.compare(clustering, maxClustering) > 0 ? clustering : maxClustering;
return this;
}
@ -277,6 +273,7 @@ public class MetadataCollector implements PartitionStatisticsCollector
public Map<MetadataType, MetadataComponent> finalizeMetadata(String partitioner, double bloomFilterFPChance, long repairedAt, SerializationHeader header)
{
Preconditions.checkState(comparator.compare(maxClustering, minClustering) >= 0);
Map<MetadataType, MetadataComponent> components = new EnumMap<>(MetadataType.class);
components.put(MetadataType.VALIDATION, new ValidationMetadata(partitioner, bloomFilterFPChance));
components.put(MetadataType.STATS, new StatsMetadata(estimatedPartitionSize,
@ -291,8 +288,8 @@ public class MetadataCollector implements PartitionStatisticsCollector
compressionRatio,
estimatedTombstoneDropTime.build(),
sstableLevel,
makeList(minClusteringValues),
makeList(maxClusteringValues),
makeList(minClustering.getRawValues()),
makeList(maxClustering.getRawValues()),
hasLegacyCounterShards,
repairedAt,
totalColumnsSet,

14
src/java/org/apache/cassandra/io/sstable/metadata/StatsMetadata.java
View File

@ -319,15 +319,25 @@ public class StatsMetadata extends MetadataComponent
if (version.hasRepairedAt())
repairedAt = in.readLong();
// for legacy sstables, we skip deserializing the min and max clustering value
// to prevent erroneously excluding sstables from reads (see CASSANDRA-14861)
int colCount = in.readInt();
List<ByteBuffer> minClusteringValues = new ArrayList<>(colCount);
for (int i = 0; i < colCount; i++)
minClusteringValues.add(ByteBufferUtil.readWithShortLength(in));
{
ByteBuffer val = ByteBufferUtil.readWithShortLength(in);
if (version.hasAccurateMinMax())
minClusteringValues.add(val);
}
colCount = in.readInt();
List<ByteBuffer> maxClusteringValues = new ArrayList<>(colCount);
for (int i = 0; i < colCount; i++)
maxClusteringValues.add(ByteBufferUtil.readWithShortLength(in));
{
ByteBuffer val = ByteBufferUtil.readWithShortLength(in);
if (version.hasAccurateMinMax())
maxClusteringValues.add(val);
}
boolean hasLegacyCounterShards = true;
if (version.tracksLegacyCounterShards())

BIN
test/data/legacy-sstables/mc/legacy_tables/legacy_mc_inaccurate_min_max/mc-1-big-CompressionInfo.db Normal file
View File

Binary file not shown.

BIN
test/data/legacy-sstables/mc/legacy_tables/legacy_mc_inaccurate_min_max/mc-1-big-Data.db Normal file
View File

Binary file not shown.

1
test/data/legacy-sstables/mc/legacy_tables/legacy_mc_inaccurate_min_max/mc-1-big-Digest.crc32 Normal file
View File

@ -0,0 +1 @@
4091794686

BIN
test/data/legacy-sstables/mc/legacy_tables/legacy_mc_inaccurate_min_max/mc-1-big-Filter.db Normal file
View File

Binary file not shown.

BIN
test/data/legacy-sstables/mc/legacy_tables/legacy_mc_inaccurate_min_max/mc-1-big-Index.db Normal file
View File

Binary file not shown.

BIN
test/data/legacy-sstables/mc/legacy_tables/legacy_mc_inaccurate_min_max/mc-1-big-Statistics.db Normal file
View File

Binary file not shown.

BIN
test/data/legacy-sstables/mc/legacy_tables/legacy_mc_inaccurate_min_max/mc-1-big-Summary.db Normal file
View File

Binary file not shown.

8
test/data/legacy-sstables/mc/legacy_tables/legacy_mc_inaccurate_min_max/mc-1-big-TOC.txt Normal file
View File

@ -0,0 +1,8 @@
Digest.crc32
CompressionInfo.db
TOC.txt
Summary.db
Statistics.db
Index.db
Data.db
Filter.db

87
test/unit/org/apache/cassandra/db/SinglePartitionSliceCommandTest.java
View File

@ -24,8 +24,14 @@ import static org.junit.Assert.*;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.TimeUnit;
import com.google.common.collect.Iterables;
import com.google.common.primitives.Ints;
import org.junit.Assert;
import org.junit.Before;
import org.junit.BeforeClass;
@ -40,8 +46,10 @@ import org.apache.cassandra.config.ColumnDefinition;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.cql3.ColumnIdentifier;
import org.apache.cassandra.cql3.QueryOptions;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.cql3.UntypedResultSet;
import org.apache.cassandra.cql3.statements.SelectStatement;
import org.apache.cassandra.db.filter.AbstractClusteringIndexFilter;
import org.apache.cassandra.db.filter.ClusteringIndexNamesFilter;
import org.apache.cassandra.db.filter.ClusteringIndexSliceFilter;
@ -51,6 +59,7 @@ import org.apache.cassandra.db.filter.RowFilter;
import org.apache.cassandra.db.marshal.Int32Type;
import org.apache.cassandra.db.marshal.IntegerType;
import org.apache.cassandra.db.marshal.UTF8Type;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
import org.apache.cassandra.db.rows.Cell;
import org.apache.cassandra.db.rows.RangeTombstoneMarker;
@ -63,6 +72,7 @@ import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputBuffer;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.schema.KeyspaceParams;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.btree.BTreeSet;
@ -387,4 +397,81 @@ public class SinglePartitionSliceCommandTest
Assert.assertNotNull(ret);
Assert.assertFalse(ret.isEmpty());
}
public static List<Unfiltered> getUnfilteredsFromSinglePartition(String q)
{
SelectStatement stmt = (SelectStatement) QueryProcessor.parseStatement(q).prepare(ClientState.forInternalCalls()).statement;
List<Unfiltered> unfiltereds = new ArrayList<>();
SinglePartitionReadCommand.Group query = (SinglePartitionReadCommand.Group) stmt.getQuery(QueryOptions.DEFAULT, FBUtilities.nowInSeconds());
Assert.assertEquals(1, query.commands.size());
SinglePartitionReadCommand command = Iterables.getOnlyElement(query.commands);
try (ReadExecutionController controller = ReadExecutionController.forCommand(command);
UnfilteredPartitionIterator partitions = command.executeLocally(controller))
{
assert partitions.hasNext();
try (UnfilteredRowIterator partition = partitions.next())
{
while (partition.hasNext())
{
Unfiltered next = partition.next();
unfiltereds.add(next);
}
}
assert !partitions.hasNext();
}
return unfiltereds;
}
private static void assertQueryReturnsSingleRT(String query)
{
List<Unfiltered> unfiltereds = getUnfilteredsFromSinglePartition(query);
Assert.assertEquals(2, unfiltereds.size());
Assert.assertTrue(unfiltereds.get(0).isRangeTombstoneMarker());
Assert.assertTrue(((RangeTombstoneMarker) unfiltereds.get(0)).isOpen(false));
Assert.assertTrue(unfiltereds.get(1).isRangeTombstoneMarker());
Assert.assertTrue(((RangeTombstoneMarker) unfiltereds.get(1)).isClose(false));
}
private static ByteBuffer bb(int v)
{
return Int32Type.instance.decompose(v);
}
/**
* tests the bug raised in CASSANDRA-14861, where the sstable min/max can
* exclude range tombstones for clustering ranges not also covered by rows
*/
@Test
public void sstableFiltering()
{
QueryProcessor.executeOnceInternal("CREATE TABLE ks.legacy_mc_inaccurate_min_max (k int, c1 int, c2 int, c3 int, v int, primary key (k, c1, c2, c3))");
CFMetaData metadata = Schema.instance.getCFMetaData("ks", "legacy_mc_inaccurate_min_max");
ColumnFamilyStore cfs = Schema.instance.getColumnFamilyStoreInstance(metadata.cfId);
QueryProcessor.executeOnceInternal("INSERT INTO ks.legacy_mc_inaccurate_min_max (k, c1, c2, c3, v) VALUES (100, 2, 2, 2, 2)");
QueryProcessor.executeOnceInternal("DELETE FROM ks.legacy_mc_inaccurate_min_max WHERE k=100 AND c1=1");
assertQueryReturnsSingleRT("SELECT * FROM ks.legacy_mc_inaccurate_min_max WHERE k=100 AND c1=1 AND c2=1");
cfs.forceBlockingFlush();
assertQueryReturnsSingleRT("SELECT * FROM ks.legacy_mc_inaccurate_min_max WHERE k=100 AND c1=1 AND c2=1");
assertQueryReturnsSingleRT("SELECT * FROM ks.legacy_mc_inaccurate_min_max WHERE k=100 AND c1=1 AND c2=1 AND c3=1"); // clustering names
cfs.truncateBlocking();
long nowMillis = System.currentTimeMillis();
Slice slice = Slice.make(Clustering.make(bb(2), bb(3)), Clustering.make(bb(10), bb(10)));
RangeTombstone rt = new RangeTombstone(slice, new DeletionTime(TimeUnit.MILLISECONDS.toMicros(nowMillis),
Ints.checkedCast(TimeUnit.MILLISECONDS.toSeconds(nowMillis))));
PartitionUpdate update = new PartitionUpdate(cfs.metadata, bb(100), cfs.metadata.partitionColumns(), 1);
update.add(rt);
new Mutation(update).apply();
assertQueryReturnsSingleRT("SELECT * FROM ks.legacy_mc_inaccurate_min_max WHERE k=100 AND c1=3 AND c2=2");
cfs.forceBlockingFlush();
assertQueryReturnsSingleRT("SELECT * FROM ks.legacy_mc_inaccurate_min_max WHERE k=100 AND c1=3 AND c2=2");
assertQueryReturnsSingleRT("SELECT * FROM ks.legacy_mc_inaccurate_min_max WHERE k=100 AND c1=3 AND c2=2 AND c3=2"); // clustering names
}
}

34
test/unit/org/apache/cassandra/io/sstable/LegacySSTableTest.java
View File

@ -25,6 +25,7 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import com.google.common.collect.Iterables;
import org.junit.After;
import org.junit.Assert;
import org.junit.BeforeClass;
@ -36,11 +37,20 @@ import org.slf4j.LoggerFactory;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.QueryOptions;
import org.apache.cassandra.cql3.QueryProcessor;
import org.apache.cassandra.cql3.UntypedResultSet;
import org.apache.cassandra.cql3.statements.SelectStatement;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.Keyspace;
import org.apache.cassandra.db.ReadExecutionController;
import org.apache.cassandra.db.SinglePartitionReadCommand;
import org.apache.cassandra.db.SinglePartitionSliceCommandTest;
import org.apache.cassandra.db.compaction.Verifier;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
import org.apache.cassandra.db.rows.RangeTombstoneMarker;
import org.apache.cassandra.db.rows.Unfiltered;
import org.apache.cassandra.db.rows.UnfilteredRowIterator;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
@ -50,6 +60,7 @@ import org.apache.cassandra.io.sstable.format.SSTableReader;
import org.apache.cassandra.io.sstable.format.Version;
import org.apache.cassandra.io.sstable.format.big.BigFormat;
import org.apache.cassandra.service.CacheService;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.streaming.StreamPlan;
import org.apache.cassandra.streaming.StreamSession;
@ -278,6 +289,27 @@ public class LegacySSTableTest
}
}
@Test
public void testInaccurateSSTableMinMax() throws Exception
{
QueryProcessor.executeInternal("CREATE TABLE legacy_tables.legacy_mc_inaccurate_min_max (k int, c1 int, c2 int, c3 int, v int, primary key (k, c1, c2, c3))");
loadLegacyTable("legacy_%s_inaccurate_min_max%s", "mc", "");
/*
sstable has the following mutations:
INSERT INTO legacy_tables.legacy_mc_inaccurate_min_max (k, c1, c2, c3, v) VALUES (100, 4, 4, 4, 4)
DELETE FROM legacy_tables.legacy_mc_inaccurate_min_max WHERE k=100 AND c1<3
*/
String query = "SELECT * FROM legacy_tables.legacy_mc_inaccurate_min_max WHERE k=100 AND c1=1 AND c2=1";
List<Unfiltered> unfiltereds = SinglePartitionSliceCommandTest.getUnfilteredsFromSinglePartition(query);
Assert.assertEquals(2, unfiltereds.size());
Assert.assertTrue(unfiltereds.get(0).isRangeTombstoneMarker());
Assert.assertTrue(((RangeTombstoneMarker) unfiltereds.get(0)).isOpen(false));
Assert.assertTrue(unfiltereds.get(1).isRangeTombstoneMarker());
Assert.assertTrue(((RangeTombstoneMarker) unfiltereds.get(1)).isClose(false));
}
@Test
public void testVerifyOldSSTables() throws Exception
{
@ -569,7 +601,7 @@ public class LegacySSTableTest
}
}
private void copySstablesFromTestData(String table, File ksDir) throws IOException
public static void copySstablesFromTestData(String table, File ksDir) throws IOException
{
File cfDir = new File(ksDir, table);
cfDir.mkdir();