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

Validate indexed values rather than the base column 

Patch by Sam Tunnicliffe; reviewed by Carl Yeksigian for CASSANDRA-9564
This commit is contained in:
Sam Tunnicliffe 2015-06-08 14:25:02 +01:00
parent a8ceb2c650
commit 3ddd17b77c
12 changed files with 147 additions and 90 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -1,4 +1,5 @@
2.0.16:
* Backport indexed value validation fix from CASSANDRA-9057 (CASSANDRA-9564)
* Don't accumulate more range than necessary in RangeTombstone.Tracker (CASSANDRA-9486)
* Add broadcast and rpc addresses to system.local (CASSANDRA-9436)
* Always mark sstable suspect when corrupted (CASSANDRA-9478)

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

@ -23,6 +23,7 @@ import java.util.*;
import org.apache.cassandra.cql3.*;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.index.SecondaryIndex;
import org.apache.cassandra.db.index.SecondaryIndexManager;
import org.apache.cassandra.exceptions.*;
import org.apache.cassandra.utils.ByteBufferUtil;
@ -118,7 +119,7 @@ public class UpdateStatement extends ModificationStatement
// validateIndexedColumns trigger a call to Keyspace.open() which we want to be able to avoid in some case
//(e.g. when using CQLSSTableWriter)
if (validateIndexedColumns)
validateIndexedColumns(cf);
validateIndexedColumns(key, cf);
}
/**
@ -127,17 +128,18 @@ public class UpdateStatement extends ModificationStatement
* @param cf the column family
* @throws InvalidRequestException if one of the values is invalid
*/
private void validateIndexedColumns(ColumnFamily cf) throws InvalidRequestException
private void validateIndexedColumns(ByteBuffer key, ColumnFamily cf) throws InvalidRequestException
{
SecondaryIndexManager indexManager = Keyspace.open(cfm.ksName).getColumnFamilyStore(cfm.cfId).indexManager;
if (indexManager.hasIndexes())
{
for (Column column : cf)
{
if (!indexManager.validate(column))
SecondaryIndex failedIndex = indexManager.validate(key, column);
if (failedIndex != null)
throw new InvalidRequestException(String.format("Can't index column value of size %d for index %s on %s.%s",
column.value().remaining(),
cfm.getColumnDefinitionFromColumnName(column.name()).getIndexName(),
failedIndex.getIndexName(),
cfm.ksName,
cfm.cfName));
}

7
src/java/org/apache/cassandra/db/index/AbstractSimplePerColumnSecondaryIndex.java
View File

@ -26,6 +26,7 @@ import org.apache.cassandra.db.marshal.*;
import org.apache.cassandra.dht.*;
import org.apache.cassandra.thrift.IndexExpression;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
/**
* Implements a secondary index for a column family using a second column family
@ -163,4 +164,10 @@ public abstract class AbstractSimplePerColumnSecondaryIndex extends PerColumnSec
indexCfs.metadata.reloadSecondaryIndexMetadata(baseCfs.metadata);
indexCfs.reload();
}
public boolean validate(ByteBuffer rowKey, Column column)
{
return getIndexedValue(rowKey, column).remaining() < FBUtilities.MAX_UNSIGNED_SHORT
&& makeIndexColumnName(rowKey, column).remaining() < FBUtilities.MAX_UNSIGNED_SHORT;
}
}

5
src/java/org/apache/cassandra/db/index/PerColumnSecondaryIndex.java
View File

@ -63,6 +63,11 @@ public abstract class PerColumnSecondaryIndex extends SecondaryIndex
}
@Override
public boolean validate(ByteBuffer rowKey, Column column)
{
return validate(column);
}
public boolean validate(Column column)
{
return column.value().remaining() < FBUtilities.MAX_UNSIGNED_SHORT;

5
src/java/org/apache/cassandra/db/index/PerRowSecondaryIndex.java
View File

@ -59,6 +59,11 @@ public abstract class PerRowSecondaryIndex extends SecondaryIndex
}
@Override
public boolean validate(ByteBuffer rowKey, Column column)
{
return validate(column);
}
public boolean validate(Column column)
{
return true;

2
src/java/org/apache/cassandra/db/index/SecondaryIndex.java
View File

@ -361,7 +361,7 @@ public abstract class SecondaryIndex
return index;
}
public abstract boolean validate(Column column);
public abstract boolean validate(ByteBuffer key, Column column);
/**
* Returns the index comparator for index backed by CFS, or null.

8
src/java/org/apache/cassandra/db/index/SecondaryIndexManager.java
View File

@ -573,12 +573,12 @@ public class SecondaryIndexManager
index.setIndexRemoved();
}
public boolean validate(Column column)
public SecondaryIndex validate(ByteBuffer rowKey, Column column)
{
for (SecondaryIndex index : indexFor(column.name()))
if (!index.validate(column))
return false;
return true;
if (!index.validate(rowKey, column))
return index;
return null;
}
private Set<String> filterByColumn(Set<String> idxNames)

6
src/java/org/apache/cassandra/thrift/CassandraServer.java
View File

@ -654,7 +654,7 @@ public class CassandraServer implements Cassandra.Iface
throw new org.apache.cassandra.exceptions.InvalidRequestException("missing mandatory super column name for super CF " + column_parent.column_family);
}
ThriftValidation.validateColumnNames(metadata, column_parent, Arrays.asList(column.name));
ThriftValidation.validateColumnData(metadata, column, column_parent.super_column != null);
ThriftValidation.validateColumnData(metadata, key, column, column_parent.super_column != null);
RowMutation rm;
try
@ -747,7 +747,7 @@ public class CassandraServer implements Cassandra.Iface
});
ThriftValidation.validateColumnNames(metadata, new ColumnParent(column_family), names);
for (Column column : updates)
ThriftValidation.validateColumnData(metadata, column, false);
ThriftValidation.validateColumnData(metadata, key, column, false);
CFMetaData cfm = Schema.instance.getCFMetaData(cState.getKeyspace(), column_family);
UnsortedColumns cfUpdates = UnsortedColumns.factory.create(cfm);
@ -839,7 +839,7 @@ public class CassandraServer implements Cassandra.Iface
for (Mutation mutation : columnFamilyMutations.getValue())
{
ThriftValidation.validateMutation(metadata, mutation);
ThriftValidation.validateMutation(metadata, key, mutation);
if (mutation.deletion != null)
{

23
src/java/org/apache/cassandra/thrift/ThriftValidation.java
View File

@ -20,6 +20,7 @@ package org.apache.cassandra.thrift;
import java.nio.ByteBuffer;
import java.util.*;
import org.apache.cassandra.db.index.SecondaryIndex;
import org.apache.cassandra.serializers.MarshalException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -298,7 +299,7 @@ public class ThriftValidation
}
}
public static void validateColumnOrSuperColumn(CFMetaData metadata, ColumnOrSuperColumn cosc)
public static void validateColumnOrSuperColumn(CFMetaData metadata, ByteBuffer key, ColumnOrSuperColumn cosc)
throws org.apache.cassandra.exceptions.InvalidRequestException
{
boolean isCommutative = metadata.getDefaultValidator().isCommutative();
@ -319,7 +320,7 @@ public class ThriftValidation
validateTtl(cosc.column);
validateColumnPath(metadata, new ColumnPath(metadata.cfName).setSuper_column((ByteBuffer)null).setColumn(cosc.column.name));
validateColumnData(metadata, cosc.column, false);
validateColumnData(metadata, key, cosc.column, false);
}
if (cosc.super_column != null)
@ -330,7 +331,7 @@ public class ThriftValidation
for (Column c : cosc.super_column.columns)
{
validateColumnPath(metadata, new ColumnPath(metadata.cfName).setSuper_column(cosc.super_column.name).setColumn(c.name));
validateColumnData(metadata, c, true);
validateColumnData(metadata, key, c, true);
}
}
@ -369,7 +370,7 @@ public class ThriftValidation
}
}
public static void validateMutation(CFMetaData metadata, Mutation mut)
public static void validateMutation(CFMetaData metadata, ByteBuffer key, Mutation mut)
throws org.apache.cassandra.exceptions.InvalidRequestException
{
ColumnOrSuperColumn cosc = mut.column_or_supercolumn;
@ -386,7 +387,7 @@ public class ThriftValidation
if (cosc != null)
{
validateColumnOrSuperColumn(metadata, cosc);
validateColumnOrSuperColumn(metadata, key, cosc);
}
else
{
@ -435,7 +436,7 @@ public class ThriftValidation
/**
* Validates the data part of the column (everything in the Column object but the name, which is assumed to be valid)
*/
public static void validateColumnData(CFMetaData metadata, Column column, boolean isSubColumn) throws org.apache.cassandra.exceptions.InvalidRequestException
public static void validateColumnData(CFMetaData metadata, ByteBuffer key, Column column, boolean isSubColumn) throws org.apache.cassandra.exceptions.InvalidRequestException
{
validateTtl(column);
if (!column.isSetValue())
@ -462,10 +463,14 @@ public class ThriftValidation
}
// Indexed column values cannot be larger than 64K. See CASSANDRA-3057/4240 for more details
if (!Keyspace.open(metadata.ksName).getColumnFamilyStore(metadata.cfName).indexManager.validate(asDBColumn(column)))
throw new org.apache.cassandra.exceptions.InvalidRequestException(String.format("Can't index column value of size %d for index %s in CF %s of KS %s",
SecondaryIndex failedIndex = Keyspace.open(metadata.ksName)
.getColumnFamilyStore(metadata.cfName)
.indexManager
.validate(key, asDBColumn(column));
if (failedIndex != null)
throw new org.apache.cassandra.exceptions.InvalidRequestException(String.format("Can't index column value of size %d for index %s in CF %s of KS %s",
column.value.remaining(),
metadata.getColumnDefinitionFromColumnName(column.name).getIndexName(),
failedIndex.getIndexName(),
metadata.cfName,
metadata.ksName));
}

151
test/unit/org/apache/cassandra/cql3/IndexedValuesValidationTest.java
View File

@ -17,20 +17,28 @@
* * limitations under the License.
*
*/
package org.apache.cassandra.cql3;
import java.nio.ByteBuffer;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import com.google.common.collect.*;
import org.junit.BeforeClass;
import org.junit.Test;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.db.ConsistencyLevel;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.db.marshal.BytesType;
import org.apache.cassandra.db.marshal.Int32Type;
import org.apache.cassandra.db.marshal.MapType;
import org.apache.cassandra.exceptions.RequestExecutionException;
import org.apache.cassandra.exceptions.RequestValidationException;
import org.apache.cassandra.service.ClientState;
import org.apache.cassandra.service.QueryState;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.MD5Digest;
import static org.junit.Assert.fail;
@ -38,6 +46,8 @@ import static org.apache.cassandra.cql3.QueryProcessor.process;
public class IndexedValuesValidationTest
{
private static final int TOO_BIG = 1024 * 65;
static ClientState clientState;
static String keyspace = "indexed_value_validation_test";
@ -49,66 +59,6 @@ public class IndexedValuesValidationTest
clientState = ClientState.forInternalCalls();
}
// CASSANDRA-8280/8081
// reject updates with indexed values where value > 64k
@Test
public void testIndexOnCompositeValueOver64k() throws Throwable
{
executeSchemaChange("CREATE TABLE %s.composite_index_table (a int, b int, c blob, PRIMARY KEY (a))");
executeSchemaChange("CREATE INDEX ON %s.composite_index_table(c)");
performInsertWithIndexedValueOver64k("INSERT INTO %s.composite_index_table (a, b, c) VALUES (0, 0, ?)");
}
@Test
public void testIndexOnClusteringValueOver64k() throws Throwable
{
executeSchemaChange("CREATE TABLE %s.ck_index_table (a int, b blob, c int, PRIMARY KEY (a, b))");
executeSchemaChange("CREATE INDEX ON %s.ck_index_table(b)");
performInsertWithIndexedValueOver64k("INSERT INTO %s.ck_index_table (a, b, c) VALUES (0, ?, 0)");
}
@Test
public void testIndexOnPartitionKeyOver64k() throws Throwable
{
executeSchemaChange("CREATE TABLE %s.pk_index_table (a blob, b int, c int, PRIMARY KEY ((a, b)))");
executeSchemaChange("CREATE INDEX ON %s.pk_index_table(a)");
performInsertWithIndexedValueOver64k("INSERT INTO %s.pk_index_table (a, b, c) VALUES (?, 0, 0)");
}
@Test
public void testCompactTableWithValueOver64k() throws Throwable
{
executeSchemaChange("CREATE TABLE %s.compact_table (a int, b blob, PRIMARY KEY (a)) WITH COMPACT STORAGE");
executeSchemaChange("CREATE INDEX ON %s.compact_table(b)");
performInsertWithIndexedValueOver64k("INSERT INTO %s.compact_table (a, b) VALUES (0, ?)");
}
private static void performInsertWithIndexedValueOver64k(String insertCQL) throws Exception
{
ByteBuffer buf = ByteBuffer.allocate(1024 * 65);
buf.clear();
for (int i=0; i<1024 + 1; i++)
buf.put((byte)0);
try
{
execute(String.format(insertCQL, keyspace), buf);
fail("Expected statement to fail validation");
}
catch (InvalidRequestException e)
{
// as expected
}
}
private static void execute(String query, ByteBuffer value) throws RequestValidationException, RequestExecutionException
{
MD5Digest statementId = QueryProcessor.prepare(String.format(query, keyspace), clientState, false).statementId;
CQLStatement statement = QueryProcessor.instance.getPrepared(statementId);
statement.executeInternal(QueryState.forInternalCalls(),
new QueryOptions(ConsistencyLevel.ONE, Collections.singletonList(value)));
}
private static void executeSchemaChange(String query) throws Throwable
{
try
@ -120,5 +70,82 @@ public class IndexedValuesValidationTest
throw exc.getCause();
}
}
}
// CASSANDRA-8280/8081
// reject updates with indexed values where value > 64k
@Test
public void testIndexOnCompositeValueOver64k() throws Throwable
{
executeSchemaChange("CREATE TABLE %s.t1(a int, b int, c blob, PRIMARY KEY (a))");
executeSchemaChange("CREATE INDEX ON %s.t1(c)");
failInsert("INSERT INTO %s.t1 (a, b, c) VALUES (0, 0, ?)", ByteBuffer.allocate(TOO_BIG));
}
@Test
public void testIndexOnClusteringColumnInsertPartitionKeyAndClusteringsOver64k() throws Throwable
{
executeSchemaChange("CREATE TABLE %s.t2(a blob, b blob, c blob, d int, PRIMARY KEY (a, b, c))");
executeSchemaChange("CREATE INDEX ON %s.t2(b)");
// CompositeIndexOnClusteringKey creates index entries composed of the
// PK plus all of the non-indexed clustering columns from the primary row
// so we should reject where len(a) + len(c) > 65560 as this will form the
// total clustering in the index table
ByteBuffer a = ByteBuffer.allocate(100);
ByteBuffer b = ByteBuffer.allocate(10);
ByteBuffer c = ByteBuffer.allocate(FBUtilities.MAX_UNSIGNED_SHORT - 99);
failInsert("INSERT INTO %s.t2 (a, b, c, d) VALUES (?, ?, ?, 0)", a, b, c);
}
@Test
public void testCompactTableWithValueOver64k() throws Throwable
{
executeSchemaChange("CREATE TABLE %s.t3(a int, b blob, PRIMARY KEY (a)) WITH COMPACT STORAGE");
executeSchemaChange("CREATE INDEX ON %s.t3(b)");
failInsert("INSERT INTO %s.t3 (a, b) VALUES (0, ?)", ByteBuffer.allocate(TOO_BIG));
}
@Test
public void testIndexOnPartitionKeyInsertValueOver64k() throws Throwable
{
executeSchemaChange("CREATE TABLE %s.t6(a int, b int, c blob, PRIMARY KEY ((a, b)))");
executeSchemaChange("CREATE INDEX ON %s.t6(a)");
succeedInsert("INSERT INTO %s.t6 (a, b, c) VALUES (0, 0, ?)", ByteBuffer.allocate(TOO_BIG));
}
@Test
public void testIndexOnClusteringColumnInsertValueOver64k() throws Throwable
{
executeSchemaChange("CREATE TABLE %s.t7(a int, b int, c blob, PRIMARY KEY (a, b))");
executeSchemaChange("CREATE INDEX ON %s.t7(b)");
succeedInsert("INSERT INTO %s.t7 (a, b, c) VALUES (0, 0, ?)", ByteBuffer.allocate(TOO_BIG));
}
public void failInsert(String insertCQL, ByteBuffer...args) throws Throwable
{
try
{
execute(String.format(insertCQL, keyspace), args);
fail("Expected statement to fail validation");
}
catch (Exception e)
{
// as expected
}
}
private static void execute(String query, ByteBuffer...value) throws RequestValidationException, RequestExecutionException
{
MD5Digest statementId = QueryProcessor.prepare(String.format(query, keyspace), clientState, false).statementId;
CQLStatement statement = QueryProcessor.instance.getPrepared(statementId);
statement.executeInternal(QueryState.forInternalCalls(),
new QueryOptions(ConsistencyLevel.ONE, com.google.common.collect.Lists.newArrayList(
value)));
}
public void succeedInsert(String insertCQL, ByteBuffer...args) throws Throwable
{
execute(String.format(insertCQL,keyspace), args);
}
}

9
test/unit/org/apache/cassandra/db/SecondaryIndexColumnSizeTest.java
View File

@ -48,12 +48,13 @@ public class SecondaryIndexColumnSizeTest
// for read
buffer.flip();
Column column = new Column(ByteBufferUtil.bytes("test"), buffer, 0);
ByteBuffer key = ByteBufferUtil.bytes("key");
SecondaryIndexColumnSizeTest.MockRowIndex mockRowIndex = new SecondaryIndexColumnSizeTest.MockRowIndex();
SecondaryIndexColumnSizeTest.MockColumnIndex mockColumnIndex = new SecondaryIndexColumnSizeTest.MockColumnIndex();
assertTrue(mockRowIndex.validate(column));
assertFalse(mockColumnIndex.validate(column));
assertTrue(mockRowIndex.validate(key, column));
assertFalse(mockColumnIndex.validate(key, column));
// test less than 64k value
buffer.flip();
@ -61,8 +62,8 @@ public class SecondaryIndexColumnSizeTest
buffer.putInt(20);
buffer.flip();
assertTrue(mockRowIndex.validate(column));
assertTrue(mockColumnIndex.validate(column));
assertTrue(mockRowIndex.validate(key, column));
assertTrue(mockColumnIndex.validate(key, column));
}
private class MockRowIndex extends PerRowSecondaryIndex

10
test/unit/org/apache/cassandra/thrift/ThriftValidationTest.java
View File

@ -31,6 +31,7 @@ import org.apache.cassandra.locator.LocalStrategy;
import org.apache.cassandra.locator.NetworkTopologyStrategy;
import org.apache.cassandra.utils.ByteBufferUtil;
import java.nio.ByteBuffer;
import java.util.Arrays;
import static org.junit.Assert.assertNotNull;
@ -92,7 +93,8 @@ public class ThriftValidationTest extends SchemaLoader
Column column = new Column(ByteBufferUtil.bytes("id"));
column.setValue(ByteBufferUtil.bytes("not a long"));
column.setTimestamp(1234);
ThriftValidation.validateColumnData(newMetadata, column, false);
ByteBuffer key = ByteBufferUtil.bytes("key");
ThriftValidation.validateColumnData(newMetadata, key, column, false);
}
@Test
@ -107,7 +109,8 @@ public class ThriftValidationTest extends SchemaLoader
Column column = new Column(ByteBufferUtil.bytes(CFMetaData.DEFAULT_KEY_ALIAS));
column.setValue(ByteBufferUtil.bytes("not a uuid"));
column.setTimestamp(1234);
ThriftValidation.validateColumnData(metaData, column, false);
ByteBuffer key = ByteBufferUtil.bytes("key");
ThriftValidation.validateColumnData(metaData, key, column, false);
IndexExpression expression = new IndexExpression(ByteBufferUtil.bytes(CFMetaData.DEFAULT_KEY_ALIAS), IndexOperator.EQ, ByteBufferUtil.bytes("a"));
ThriftValidation.validateFilterClauses(metaData, Arrays.asList(expression));
@ -124,7 +127,8 @@ public class ThriftValidationTest extends SchemaLoader
Column column = new Column(ByteBufferUtil.bytes(CFMetaData.DEFAULT_COLUMN_ALIAS + 1));
column.setValue(ByteBufferUtil.bytes("not a long"));
column.setTimestamp(1234);
ThriftValidation.validateColumnData(metaData, column, false);
ByteBuffer key = ByteBufferUtil.bytes("key");
ThriftValidation.validateColumnData(metaData, key, column, false);
}
@Test