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

Ensure indexed values are only validated once per partition 

Patch by Sam Tunnicliffe; reviewed by Aleksey Yeschenko for
CASSANDRA-10536
This commit is contained in:
Sam Tunnicliffe 2015-10-16 13:53:52 +01:00
parent 89293efc7d
commit aad3ae2cbe
11 changed files with 164 additions and 51 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
3.0-rc2
* Ensure validation of indexed values only occurs once per-partition (CASSANDRA-10536)
* Fix handling of static columns for range tombstones in thrift (CASSANDRA-10174)
* Support empty ColumnFilter for backward compatility on empty IN (CASSANDRA-10471)
* Remove Pig support (CASSANDRA-10542)

27
src/java/org/apache/cassandra/cql3/UpdateParameters.java
View File

@ -26,10 +26,8 @@ import org.apache.cassandra.db.*;
import org.apache.cassandra.db.context.CounterContext;
import org.apache.cassandra.db.filter.ColumnFilter;
import org.apache.cassandra.db.partitions.Partition;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.db.rows.*;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.index.SecondaryIndexManager;
import org.apache.cassandra.utils.FBUtilities;
/**
@ -47,8 +45,6 @@ public class UpdateParameters
private final DeletionTime deletionTime;
private final SecondaryIndexManager indexManager;
// For lists operation that require a read-before-write. Will be null otherwise.
private final Map<DecoratedKey, Partition> prefetchedRows;
@ -63,8 +59,7 @@ public class UpdateParameters
QueryOptions options,
long timestamp,
int ttl,
Map<DecoratedKey, Partition> prefetchedRows,
boolean validateIndexedColumns)
Map<DecoratedKey, Partition> prefetchedRows)
throws InvalidRequestException
{
this.metadata = metadata;
@ -79,32 +74,12 @@ public class UpdateParameters
this.prefetchedRows = prefetchedRows;
// Index column validation triggers a call to Keyspace.open() which we want
// to be able to avoid in some case (e.g. when using CQLSSTableWriter)
if (validateIndexedColumns)
{
SecondaryIndexManager manager = Keyspace.openAndGetStore(metadata).indexManager;
indexManager = manager.hasIndexes() ? manager : null;
}
else
{
indexManager = null;
}
// We use MIN_VALUE internally to mean the absence of of timestamp (in Selection, in sstable stats, ...), so exclude
// it to avoid potential confusion.
if (timestamp == Long.MIN_VALUE)
throw new InvalidRequestException(String.format("Out of bound timestamp, must be in [%d, %d]", Long.MIN_VALUE + 1, Long.MAX_VALUE));
}
public void validateIndexedColumns(PartitionUpdate update)
{
if (indexManager == null)
return;
indexManager.validate(update);
}
public void newRow(Clustering clustering) throws InvalidRequestException
{
if (metadata.isDense() && !metadata.isCompound())

10
src/java/org/apache/cassandra/cql3/statements/BatchStatement.java
View File

@ -23,21 +23,18 @@ import java.util.concurrent.TimeUnit;
import com.google.common.base.Function;
import com.google.common.collect.Iterables;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.helpers.MessageFormatter;
import org.apache.cassandra.config.ColumnDefinition;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.*;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.db.rows.RowIterator;
import org.apache.cassandra.db.partitions.*;
import org.apache.cassandra.exceptions.*;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.service.ClientWarn;
import org.apache.cassandra.service.QueryState;
import org.apache.cassandra.service.StorageProxy;
import org.apache.cassandra.service.*;
import org.apache.cassandra.tracing.Tracing;
import org.apache.cassandra.transport.messages.ResultMessage;
import org.apache.cassandra.utils.NoSpamLogger;
@ -242,6 +239,7 @@ public class BatchStatement implements CQLStatement
.getMessage());
}
collector.validateIndexedColumns();
return collector.toMutations();
}

14
src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java
View File

@ -22,11 +22,15 @@ import java.util.*;
import com.google.common.collect.HashMultimap;
import com.google.common.collect.Multimap;
import org.apache.cassandra.cql3.*;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.cql3.*;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.filter.*;
import org.apache.cassandra.db.partitions.*;
import org.apache.cassandra.db.filter.ClusteringIndexSliceFilter;
import org.apache.cassandra.db.filter.ColumnFilter;
import org.apache.cassandra.db.partitions.FilteredPartition;
import org.apache.cassandra.db.partitions.Partition;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.service.CASRequest;
import org.apache.cassandra.utils.Pair;
@ -170,6 +174,8 @@ public class CQL3CasRequest implements CASRequest
for (RowUpdate upd : updates)
upd.applyUpdates(current, update);
Keyspace.openAndGetStore(cfm).indexManager.validate(update);
if (isBatch)
BatchStatement.verifyBatchSize(Collections.singleton(update));
@ -200,7 +206,7 @@ public class CQL3CasRequest implements CASRequest
public void applyUpdates(FilteredPartition current, PartitionUpdate updates) throws InvalidRequestException
{
Map<DecoratedKey, Partition> map = stmt.requiresRead() ? Collections.<DecoratedKey, Partition>singletonMap(key, current) : null;
UpdateParameters params = new UpdateParameters(cfm, updates.columns(), options, timestamp, stmt.getTimeToLive(options), map, true);
UpdateParameters params = new UpdateParameters(cfm, updates.columns(), options, timestamp, stmt.getTimeToLive(options), map);
stmt.addUpdateForKey(updates, clustering, params);
}
}

2
src/java/org/apache/cassandra/cql3/statements/DeleteStatement.java
View File

@ -93,8 +93,6 @@ public class DeleteStatement extends ModificationStatement
update.add(params.buildRow());
}
}
params.validateIndexedColumns(update);
}
@Override

4
src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
View File

@ -604,6 +604,8 @@ public abstract class ModificationStatement implements CQLStatement
{
UpdatesCollector collector = new UpdatesCollector(updatedColumns, 1);
addUpdates(collector, options, local, now);
collector.validateIndexedColumns();
return collector.toMutations();
}
@ -706,7 +708,7 @@ public abstract class ModificationStatement implements CQLStatement
{
// Some lists operation requires reading
Map<DecoratedKey, Partition> lists = readRequiredLists(keys, filter, limits, local, options.getConsistency());
return new UpdateParameters(cfm, updatedColumns(), options, getTimestamp(now, options), getTimeToLive(options), lists, true);
return new UpdateParameters(cfm, updatedColumns(), options, getTimestamp(now, options), getTimeToLive(options), lists);
}
private Slices toSlices(SortedSet<Slice.Bound> startBounds, SortedSet<Slice.Bound> endBounds)

2
src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
View File

@ -103,8 +103,6 @@ public class UpdateStatement extends ModificationStatement
op.execute(update.partitionKey(), params);
update.add(params.buildRow());
}
params.validateIndexedColumns(update);
}
@Override

20
src/java/org/apache/cassandra/cql3/statements/UpdatesCollector.java
View File

@ -20,13 +20,9 @@ package org.apache.cassandra.cql3.statements;
import java.nio.ByteBuffer;
import java.util.*;
import org.apache.cassandra.db.CounterMutation;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.db.PartitionColumns;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.partitions.PartitionUpdate;
/**
* Utility class to collect updates.
@ -80,6 +76,18 @@ final class UpdatesCollector
return upd;
}
/**
* Check all partition updates contain only valid values for any
* indexed columns.
*/
public void validateIndexedColumns()
{
for (Map<ByteBuffer, IMutation> perKsMutations : mutations.values())
for (IMutation mutation : perKsMutations.values())
for (PartitionUpdate update : mutation.getPartitionUpdates())
Keyspace.openAndGetStore(update.metadata()).indexManager.validate(update);
}
private Mutation getMutation(CFMetaData cfm, DecoratedKey dk, ConsistencyLevel consistency)
{
String ksName = cfm.ksName;
@ -127,4 +135,4 @@ final class UpdatesCollector
}
return ksMap;
}
}
}

8
src/java/org/apache/cassandra/io/sstable/CQLSSTableWriter.java
View File

@ -28,9 +28,10 @@ import org.apache.cassandra.cql3.*;
import org.apache.cassandra.cql3.statements.CreateTableStatement;
import org.apache.cassandra.cql3.statements.ParsedStatement;
import org.apache.cassandra.cql3.statements.UpdateStatement;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.partitions.*;
import org.apache.cassandra.db.Clustering;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.partitions.Partition;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.exceptions.InvalidRequestException;
@ -216,8 +217,7 @@ public class CQLSSTableWriter implements Closeable
options,
insert.getTimestamp(now, options),
insert.getTimeToLive(options),
Collections.<DecoratedKey, Partition>emptyMap(),
false);
Collections.<DecoratedKey, Partition>emptyMap());
try
{

72
test/unit/org/apache/cassandra/cql3/validation/entities/SecondaryIndexTest.java
View File

@ -25,6 +25,7 @@ import org.junit.Test;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.ColumnDefinition;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.cql3.ColumnIdentifier;
import org.apache.cassandra.cql3.statements.IndexTarget;
@ -555,12 +556,23 @@ public class SecondaryIndexTest extends CQLTester
// CASSANDRA-8280/8081
// reject updates with indexed values where value > 64k
// make sure we check conditional and unconditional statements,
// both singly and in batches (CASSANDRA-10536)
@Test
public void testIndexOnCompositeValueOver64k() throws Throwable
{
createTable("CREATE TABLE %s(a int, b int, c blob, PRIMARY KEY (a))");
createIndex("CREATE INDEX ON %s(c)");
failInsert("INSERT INTO %s (a, b, c) VALUES (0, 0, ?)", ByteBuffer.allocate(TOO_BIG));
failInsert("INSERT INTO %s (a, b, c) VALUES (0, 0, ?) IF NOT EXISTS", ByteBuffer.allocate(TOO_BIG));
failInsert("BEGIN BATCH\n" +
"INSERT INTO %s (a, b, c) VALUES (0, 0, ?);\n" +
"APPLY BATCH",
ByteBuffer.allocate(TOO_BIG));
failInsert("BEGIN BATCH\n" +
"INSERT INTO %s (a, b, c) VALUES (0, 0, ?) IF NOT EXISTS;\n" +
"APPLY BATCH",
ByteBuffer.allocate(TOO_BIG));
}
@Test
@ -569,6 +581,15 @@ public class SecondaryIndexTest extends CQLTester
createTable("CREATE TABLE %s(a int, b blob, PRIMARY KEY (a)) WITH COMPACT STORAGE");
createIndex("CREATE INDEX ON %s(b)");
failInsert("INSERT INTO %s (a, b) VALUES (0, ?)", ByteBuffer.allocate(TOO_BIG));
failInsert("INSERT INTO %s (a, b) VALUES (0, ?) IF NOT EXISTS", ByteBuffer.allocate(TOO_BIG));
failInsert("BEGIN BATCH\n" +
"INSERT INTO %s (a, b) VALUES (0, ?);\n" +
"APPLY BATCH",
ByteBuffer.allocate(TOO_BIG));
failInsert("BEGIN BATCH\n" +
"INSERT INTO %s (a, b) VALUES (0, ?) IF NOT EXISTS;\n" +
"APPLY BATCH",
ByteBuffer.allocate(TOO_BIG));
}
@Test
@ -576,7 +597,29 @@ public class SecondaryIndexTest extends CQLTester
{
createTable("CREATE TABLE %s(a int, b int, c blob, PRIMARY KEY ((a, b)))");
createIndex("CREATE INDEX ON %s(a)");
succeedInsert("INSERT INTO %s (a, b, c) VALUES (0, 0, ?) IF NOT EXISTS", ByteBuffer.allocate(TOO_BIG));
succeedInsert("INSERT INTO %s (a, b, c) VALUES (0, 0, ?)", ByteBuffer.allocate(TOO_BIG));
succeedInsert("BEGIN BATCH\n" +
"INSERT INTO %s (a, b, c) VALUES (0, 0, ?);\n" +
"APPLY BATCH", ByteBuffer.allocate(TOO_BIG));
// the indexed value passes validation, but the batch size will
// exceed the default failure threshold, so temporarily raise it
// (the non-conditional batch doesn't hit this because
// BatchStatement::executeInternal skips the size check but CAS
// path does not)
long batchSizeThreshold = DatabaseDescriptor.getBatchSizeFailThreshold();
try
{
DatabaseDescriptor.setBatchSizeFailThresholdInKB( (TOO_BIG / 1024) * 2);
succeedInsert("BEGIN BATCH\n" +
"INSERT INTO %s (a, b, c) VALUES (1, 1, ?) IF NOT EXISTS;\n" +
"APPLY BATCH", ByteBuffer.allocate(TOO_BIG));
}
finally
{
DatabaseDescriptor.setBatchSizeFailThresholdInKB((int) (batchSizeThreshold / 1024));
}
}
@Test
@ -584,7 +627,29 @@ public class SecondaryIndexTest extends CQLTester
{
createTable("CREATE TABLE %s(a int, b int, c blob, PRIMARY KEY (a, b))");
createIndex("CREATE INDEX ON %s(b)");
succeedInsert("INSERT INTO %s (a, b, c) VALUES (0, 0, ?) IF NOT EXISTS", ByteBuffer.allocate(TOO_BIG));
succeedInsert("INSERT INTO %s (a, b, c) VALUES (0, 0, ?)", ByteBuffer.allocate(TOO_BIG));
succeedInsert("BEGIN BATCH\n" +
"INSERT INTO %s (a, b, c) VALUES (0, 0, ?);\n" +
"APPLY BATCH", ByteBuffer.allocate(TOO_BIG));
// the indexed value passes validation, but the batch size will
// exceed the default failure threshold, so temporarily raise it
// (the non-conditional batch doesn't hit this because
// BatchStatement::executeInternal skips the size check but CAS
// path does not)
long batchSizeThreshold = DatabaseDescriptor.getBatchSizeFailThreshold();
try
{
DatabaseDescriptor.setBatchSizeFailThresholdInKB( (TOO_BIG / 1024) * 2);
succeedInsert("BEGIN BATCH\n" +
"INSERT INTO %s (a, b, c) VALUES (1, 1, ?) IF NOT EXISTS;\n" +
"APPLY BATCH", ByteBuffer.allocate(TOO_BIG));
}
finally
{
DatabaseDescriptor.setBatchSizeFailThresholdInKB((int)(batchSizeThreshold / 1024));
}
}
@Test
@ -595,6 +660,13 @@ public class SecondaryIndexTest extends CQLTester
Map<Integer, ByteBuffer> map = new HashMap();
map.put(0, ByteBuffer.allocate(1024 * 65));
failInsert("INSERT INTO %s (a, b) VALUES (0, ?)", map);
failInsert("INSERT INTO %s (a, b) VALUES (0, ?) IF NOT EXISTS", map);
failInsert("BEGIN BATCH\n" +
"INSERT INTO %s (a, b) VALUES (0, ?);\n" +
"APPLY BATCH", map);
failInsert("BEGIN BATCH\n" +
"INSERT INTO %s (a, b) VALUES (0, ?) IF NOT EXISTS;\n" +
"APPLY BATCH", map);
}
public void failInsert(String insertCQL, Object...args) throws Throwable

55
test/unit/org/apache/cassandra/index/internal/CassandraIndexTest.java
View File

@ -413,6 +413,61 @@ public class CassandraIndexTest extends CQLTester
assertIndexRowTtl(indexCfs, indexedVal, updatedTtl);
}
@Test
public void indexBatchStatements() throws Throwable
{
// see CASSANDRA-10536
createTable("CREATE TABLE %s (a int, b int, c int, PRIMARY KEY (a, b))");
createIndex("CREATE INDEX ON %s(c)");
// Multi partition batch
execute("BEGIN BATCH\n" +
"UPDATE %1$s SET c = 0 WHERE a = 0 AND b = 0;\n" +
"UPDATE %1$s SET c = 1 WHERE a = 1 AND b = 1;\n" +
"APPLY BATCH");
assertRows(execute("SELECT * FROM %s WHERE c = 0"), row(0, 0, 0));
assertRows(execute("SELECT * FROM %s WHERE c = 1"), row(1, 1, 1));
// Single Partition batch
execute("BEGIN BATCH\n" +
"UPDATE %1$s SET c = 2 WHERE a = 2 AND b = 0;\n" +
"UPDATE %1$s SET c = 3 WHERE a = 2 AND b = 1;\n" +
"APPLY BATCH");
assertRows(execute("SELECT * FROM %s WHERE c = 2"), row(2, 0, 2));
assertRows(execute("SELECT * FROM %s WHERE c = 3"), row(2, 1, 3));
}
@Test
public void indexStatementsWithConditions() throws Throwable
{
// see CASSANDRA-10536
createTable("CREATE TABLE %s (a int, b int, c int, PRIMARY KEY (a, b))");
createIndex("CREATE INDEX ON %s(c)");
execute("INSERT INTO %s (a, b, c) VALUES (0, 0, 0) IF NOT EXISTS");
assertRows(execute("SELECT * FROM %s WHERE c = 0"), row(0, 0, 0));
execute("INSERT INTO %s (a, b, c) VALUES (0, 0, 1) IF NOT EXISTS");
assertEmpty(execute("SELECT * FROM %s WHERE c = 1"));
execute("UPDATE %s SET c = 1 WHERE a = 0 AND b =0 IF c = 0");
assertRows(execute("SELECT * FROM %s WHERE c = 1"), row(0, 0, 1));
assertEmpty(execute("SELECT * FROM %s WHERE c = 0"));
execute("DELETE FROM %s WHERE a = 0 AND b = 0 IF c = 0");
assertRows(execute("SELECT * FROM %s WHERE c = 1"), row(0, 0, 1));
execute("DELETE FROM %s WHERE a = 0 AND b = 0 IF c = 1");
assertEmpty(execute("SELECT * FROM %s WHERE c = 1"));
execute("BEGIN BATCH\n" +
"INSERT INTO %1$s (a, b, c) VALUES (2, 2, 2) IF NOT EXISTS;\n" +
"INSERT INTO %1$s (a, b, c) VALUES (2, 3, 3)\n" +
"APPLY BATCH");
assertRows(execute("SELECT * FROM %s WHERE c = 2"), row(2, 2, 2));
assertRows(execute("SELECT * FROM %s WHERE c = 3"), row(2, 3, 3));
}
// this is slightly annoying, but we cannot read rows from the methods in Util as
// ReadCommand#executeInternal uses metadata retrieved via the cfId, which the index
// CFS inherits from the base CFS. This has the 'wrong' partitioner (the index table