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 trunk

This commit is contained in:
Sam Tunnicliffe 2015-09-18 11:32:35 +01:00
parent 5400b93bbc 64e2f5ddaa
commit 0c3575fabd
19 changed files with 717 additions and 91 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGES.txt
View File

@ -3,6 +3,7 @@
3.0.0-rc1
* Add custom query expressions to SELECT (CASSANDRA-10217)
* Fix minor bugs in MV handling (CASSANDRA-10362)
* Allow custom indexes with 0,1 or multiple target columns (CASSANDRA-10124)
* Improve MV schema representation (CASSANDRA-9921)

26
src/java/org/apache/cassandra/cql3/Cql.g
View File

@ -39,6 +39,7 @@ options {
import org.apache.cassandra.auth.*;
import org.apache.cassandra.cql3.*;
import org.apache.cassandra.cql3.restrictions.CustomIndexExpression;
import org.apache.cassandra.cql3.statements.*;
import org.apache.cassandra.cql3.selection.*;
import org.apache.cassandra.cql3.functions.*;
@ -309,7 +310,8 @@ selectStatement returns [SelectStatement.RawStatement expr]
isDistinct,
allowFiltering,
isJson);
$expr = new SelectStatement.RawStatement(cf, params, sclause, wclause, limit);
WhereClause where = wclause == null ? WhereClause.empty() : wclause.build();
$expr = new SelectStatement.RawStatement(cf, params, sclause, where, limit);
}
;
@ -345,9 +347,19 @@ countArgument
| i=INTEGER { if (!i.getText().equals("1")) addRecognitionError("Only COUNT(1) is supported, got COUNT(" + i.getText() + ")");}
;
whereClause returns [List<Relation> clause]
@init{ $clause = new ArrayList<Relation>(); }
: relation[$clause] (K_AND relation[$clause])*
whereClause returns [WhereClause.Builder clause]
@init{ $clause = new WhereClause.Builder(); }
: relationOrExpression[$clause] (K_AND relationOrExpression[$clause])*
;
relationOrExpression [WhereClause.Builder clause]
: relation[$clause]
| customIndexExpression[$clause]
;
customIndexExpression [WhereClause.Builder clause]
@init{IndexName name = new IndexName();}
: 'expr(' idxName[name] ',' t=term ')' { clause.add(new CustomIndexExpression(name, t));}
;
orderByClause[Map<ColumnIdentifier.Raw, Boolean> orderings]
@ -437,7 +449,7 @@ updateStatement returns [UpdateStatement.ParsedUpdate expr]
return new UpdateStatement.ParsedUpdate(cf,
attrs,
operations,
wclause,
wclause.build(),
conditions == null ? Collections.<Pair<ColumnIdentifier.Raw, ColumnCondition.Raw>>emptyList() : conditions,
ifExists);
}
@ -471,7 +483,7 @@ deleteStatement returns [DeleteStatement.Parsed expr]
return new DeleteStatement.Parsed(cf,
attrs,
columnDeletions,
wclause,
wclause.build(),
conditions == null ? Collections.<Pair<ColumnIdentifier.Raw, ColumnCondition.Raw>>emptyList() : conditions,
ifExists);
}
@ -1409,7 +1421,7 @@ relationType returns [Operator op]
| '!=' { $op = Operator.NEQ; }
;
relation[List<Relation> clauses]
relation[WhereClause.Builder clauses]
: name=cident type=relationType t=term { $clauses.add(new SingleColumnRelation(name, type, t)); }
| K_TOKEN l=tupleOfIdentifiers type=relationType t=term
{ $clauses.add(new TokenRelation(l, type, t)); }

74
src/java/org/apache/cassandra/cql3/WhereClause.java Normal file
View File

@ -0,0 +1,74 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.cql3;
import java.util.List;
import com.google.common.collect.ImmutableList;
import org.apache.cassandra.cql3.restrictions.CustomIndexExpression;
public final class WhereClause
{
private static final WhereClause EMPTY = new WhereClause(new Builder());
public final List<Relation> relations;
public final List<CustomIndexExpression> expressions;
private WhereClause(Builder builder)
{
this.relations = builder.relations.build();
this.expressions = builder.expressions.build();
}
public static WhereClause empty()
{
return EMPTY;
}
public boolean containsCustomExpressions()
{
return !expressions.isEmpty();
}
public static final class Builder
{
ImmutableList.Builder<Relation> relations = new ImmutableList.Builder<>();
ImmutableList.Builder<CustomIndexExpression> expressions = new ImmutableList.Builder<>();
public Builder add(Relation relation)
{
relations.add(relation);
return this;
}
public Builder add(CustomIndexExpression expression)
{
expressions.add(expression);
return this;
}
public WhereClause build()
{
return new WhereClause(this);
}
}
}

58
src/java/org/apache/cassandra/cql3/restrictions/CustomIndexExpression.java Normal file
View File

@ -0,0 +1,58 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.cql3.restrictions;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.cql3.*;
import org.apache.cassandra.db.filter.RowFilter;
import org.apache.cassandra.db.marshal.UTF8Type;
public class CustomIndexExpression
{
private final ColumnIdentifier valueColId = new ColumnIdentifier("custom index expression", false);
public final IndexName targetIndex;
public final Term.Raw valueRaw;
private Term value;
public CustomIndexExpression(IndexName targetIndex, Term.Raw value)
{
this.targetIndex = targetIndex;
this.valueRaw = value;
}
public void prepareValue(CFMetaData cfm, VariableSpecifications boundNames)
{
ColumnSpecification spec = new ColumnSpecification(cfm.ksName, cfm.ksName, valueColId, UTF8Type.instance);
value = valueRaw.prepare(cfm.ksName, spec);
value.collectMarkerSpecification(boundNames);
}
public void addToRowFilter(RowFilter filter,
CFMetaData cfm,
QueryOptions options)
{
filter.addCustomIndexExpression(cfm,
cfm.getIndexes()
.get(targetIndex.getIdx())
.orElseThrow(() -> IndexRestrictions.indexNotFound(targetIndex, cfm)),
value.bindAndGet(options));
}
}

83
src/java/org/apache/cassandra/cql3/restrictions/IndexRestrictions.java Normal file
View File

@ -0,0 +1,83 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership. The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.cassandra.cql3.restrictions;
import java.util.ArrayList;
import java.util.List;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.cql3.IndexName;
import org.apache.cassandra.exceptions.InvalidRequestException;
public class IndexRestrictions
{
public static final String INDEX_NOT_FOUND = "Invalid index expression, index %s not found for %s.%s";
public static final String INVALID_INDEX = "Target index %s cannot be used to query %s.%s";
public static final String CUSTOM_EXPRESSION_NOT_SUPPORTED = "Index %s does not support custom expressions";
public static final String NON_CUSTOM_INDEX_IN_EXPRESSION = "Only CUSTOM indexes may be used in custom index expressions, %s is not valid";
public static final String MULTIPLE_EXPRESSIONS = "Multiple custom index expressions in a single query are not supported";
private final List<Restrictions> regularRestrictions = new ArrayList<>();
private final List<CustomIndexExpression> customExpressions = new ArrayList<>();
public void add(Restrictions restrictions)
{
regularRestrictions.add(restrictions);
}
public void add(CustomIndexExpression expression)
{
customExpressions.add(expression);
}
public boolean isEmpty()
{
return regularRestrictions.isEmpty() && customExpressions.isEmpty();
}
public List<Restrictions> getRestrictions()
{
return regularRestrictions;
}
public List<CustomIndexExpression> getCustomIndexExpressions()
{
return customExpressions;
}
static InvalidRequestException invalidIndex(IndexName indexName, CFMetaData cfm)
{
return new InvalidRequestException(String.format(INVALID_INDEX, indexName.getIdx(), cfm.ksName, cfm.cfName));
}
static InvalidRequestException indexNotFound(IndexName indexName, CFMetaData cfm)
{
return new InvalidRequestException(String.format(INDEX_NOT_FOUND,indexName.getIdx(), cfm.ksName, cfm.cfName));
}
static InvalidRequestException nonCustomIndexInExpression(IndexName indexName)
{
return new InvalidRequestException(String.format(NON_CUSTOM_INDEX_IN_EXPRESSION, indexName.getIdx()));
}
static InvalidRequestException customExpressionNotSupported(IndexName indexName)
{
return new InvalidRequestException(String.format(CUSTOM_EXPRESSION_NOT_SUPPORTED, indexName.getIdx()));
}
}

76
src/java/org/apache/cassandra/cql3/restrictions/StatementRestrictions.java
View File

@ -25,17 +25,17 @@ import com.google.common.collect.Iterables;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.ColumnDefinition;
import org.apache.cassandra.cql3.ColumnIdentifier;
import org.apache.cassandra.cql3.QueryOptions;
import org.apache.cassandra.cql3.Relation;
import org.apache.cassandra.cql3.VariableSpecifications;
import org.apache.cassandra.cql3.*;
import org.apache.cassandra.cql3.functions.Function;
import org.apache.cassandra.cql3.statements.Bound;
import org.apache.cassandra.cql3.statements.StatementType;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.filter.RowFilter;
import org.apache.cassandra.dht.*;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.index.Index;
import org.apache.cassandra.index.SecondaryIndexManager;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.btree.BTreeSet;
@ -51,6 +51,7 @@ public final class StatementRestrictions
{
public static final String NO_INDEX_FOUND_MESSAGE =
"No supported secondary index found for the non primary key columns restrictions";
/**
* The type of statement
*/
@ -79,7 +80,7 @@ public final class StatementRestrictions
/**
* The restrictions used to build the row filter
*/
private final List<Restrictions> indexRestrictions = new ArrayList<>();
private final IndexRestrictions indexRestrictions = new IndexRestrictions();
/**
* <code>true</code> if the secondary index need to be queried, <code>false</code> otherwise
@ -114,7 +115,7 @@ public final class StatementRestrictions
public StatementRestrictions(StatementType type,
CFMetaData cfm,
List<Relation> whereClause,
WhereClause whereClause,
VariableSpecifications boundNames,
boolean selectsOnlyStaticColumns,
boolean selectACollection,
@ -131,7 +132,7 @@ public final class StatementRestrictions
* - The value_alias cannot be restricted in any way (we don't support wide rows with indexed value
* in CQL so far)
*/
for (Relation relation : whereClause)
for (Relation relation : whereClause.relations)
addRestriction(relation.toRestriction(cfm, boundNames));
boolean hasQueriableClusteringColumnIndex = false;
@ -142,8 +143,12 @@ public final class StatementRestrictions
ColumnFamilyStore cfs = Keyspace.open(cfm.ksName).getColumnFamilyStore(cfm.cfName);
SecondaryIndexManager secondaryIndexManager = cfs.indexManager;
if (whereClause.containsCustomExpressions())
processCustomIndexExpressions(whereClause.expressions, boundNames, secondaryIndexManager);
hasQueriableClusteringColumnIndex = clusteringColumnsRestrictions.hasSupportingIndex(secondaryIndexManager);
hasQueriableIndex = hasQueriableClusteringColumnIndex
hasQueriableIndex = !indexRestrictions.getCustomIndexExpressions().isEmpty()
|| hasQueriableClusteringColumnIndex
|| partitionKeyRestrictions.hasSupportingIndex(secondaryIndexManager)
|| nonPrimaryKeyRestrictions.hasSupportingIndex(secondaryIndexManager);
}
@ -244,7 +249,7 @@ public final class StatementRestrictions
public Set<ColumnDefinition> nonPKRestrictedColumns()
{
Set<ColumnDefinition> columns = new HashSet<>();
for (Restrictions r : indexRestrictions)
for (Restrictions r : indexRestrictions.getRestrictions())
for (ColumnDefinition def : r.getColumnDefs())
if (!def.isPrimaryKeyColumn())
columns.add(def);
@ -432,15 +437,53 @@ public final class StatementRestrictions
return cfm.clusteringColumns().size() != clusteringColumnsRestrictions.size();
}
private void processCustomIndexExpressions(List<CustomIndexExpression> expressions,
VariableSpecifications boundNames,
SecondaryIndexManager indexManager)
{
if (!MessagingService.instance().areAllNodesAtLeast30())
throw new InvalidRequestException("Please upgrade all nodes to at least 3.0 before using custom index expressions");
if (expressions.size() > 1)
throw new InvalidRequestException(IndexRestrictions.MULTIPLE_EXPRESSIONS);
CustomIndexExpression expression = expressions.get(0);
expression.prepareValue(cfm, boundNames);
CFName cfName = expression.targetIndex.getCfName();
if (cfName.hasKeyspace()
&& !expression.targetIndex.getKeyspace().equals(cfm.ksName))
throw IndexRestrictions.invalidIndex(expression.targetIndex, cfm);
if (cfName.getColumnFamily() != null && !cfName.getColumnFamily().equals(cfm.cfName))
throw IndexRestrictions.invalidIndex(expression.targetIndex, cfm);
if (!cfm.getIndexes().has(expression.targetIndex.getIdx()))
throw IndexRestrictions.indexNotFound(expression.targetIndex, cfm);
Index index = indexManager.getIndex(cfm.getIndexes().get(expression.targetIndex.getIdx()).get());
if (!index.getIndexMetadata().isCustom())
throw IndexRestrictions.nonCustomIndexInExpression(expression.targetIndex);
if (index.customExpressionValueType() == null)
throw IndexRestrictions.customExpressionNotSupported(expression.targetIndex);
indexRestrictions.add(expression);
}
public RowFilter getRowFilter(SecondaryIndexManager indexManager, QueryOptions options)
{
if (indexRestrictions.isEmpty())
return RowFilter.NONE;
RowFilter filter = RowFilter.create();
for (Restrictions restrictions : indexRestrictions)
for (Restrictions restrictions : indexRestrictions.getRestrictions())
restrictions.addRowFilterTo(filter, indexManager, options);
for (CustomIndexExpression expression : indexRestrictions.getCustomIndexExpressions())
expression.addToRowFilter(filter, cfm, options);
return filter;
}
@ -629,13 +672,13 @@ public final class StatementRestrictions
*/
public boolean needFiltering()
{
int numberOfRestrictedColumns = 0;
for (Restrictions restrictions : indexRestrictions)
numberOfRestrictedColumns += restrictions.size();
int numberOfRestrictions = indexRestrictions.getCustomIndexExpressions().size();
for (Restrictions restrictions : indexRestrictions.getRestrictions())
numberOfRestrictions += restrictions.size();
return numberOfRestrictedColumns > 1
|| (numberOfRestrictedColumns == 0 && !clusteringColumnsRestrictions.isEmpty())
|| (numberOfRestrictedColumns != 0
return numberOfRestrictions > 1
|| (numberOfRestrictions == 0 && !clusteringColumnsRestrictions.isEmpty())
|| (numberOfRestrictions != 0
&& nonPrimaryKeyRestrictions.hasMultipleContains());
}
@ -664,4 +707,5 @@ public final class StatementRestrictions
&& !hasUnrestrictedClusteringColumns()
&& (clusteringColumnsRestrictions.isEQ() || clusteringColumnsRestrictions.isIN());
}
}

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

@ -112,12 +112,12 @@ public class DeleteStatement extends ModificationStatement
public static class Parsed extends ModificationStatement.Parsed
{
private final List<Operation.RawDeletion> deletions;
private final List<Relation> whereClause;
private final WhereClause whereClause;
public Parsed(CFName name,
Attributes.Raw attrs,
List<Operation.RawDeletion> deletions,
List<Relation> whereClause,
WhereClause whereClause,
List<Pair<ColumnIdentifier.Raw, ColumnCondition.Raw>> conditions,
boolean ifExists)
{

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

@ -52,8 +52,8 @@ import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.UUIDGen;
import static org.apache.cassandra.cql3.statements.RequestValidations.checkFalse;
import static org.apache.cassandra.cql3.statements.RequestValidations.checkNull;
import static org.apache.cassandra.cql3.statements.RequestValidations.checkNotNull;
import static org.apache.cassandra.cql3.statements.RequestValidations.checkNull;
/*
* Abstract parent class of individual modifications, i.e. INSERT, UPDATE and DELETE.
@ -62,6 +62,9 @@ public abstract class ModificationStatement implements CQLStatement
{
protected static final Logger logger = LoggerFactory.getLogger(ModificationStatement.class);
public static final String CUSTOM_EXPRESSIONS_NOT_ALLOWED =
"Custom index expressions cannot be used in WHERE clauses for UPDATE or DELETE statements";
private static final ColumnIdentifier CAS_RESULT_COLUMN = new ColumnIdentifier("[applied]", false);
protected final StatementType type;
@ -833,7 +836,7 @@ public abstract class ModificationStatement implements CQLStatement
* @param cfm the column family meta data
* @param boundNames the bound names
* @param operations the column operations
* @param relations the where relations
* @param where the where clause
* @param conditions the conditions
* @return the restrictions
*/
@ -841,11 +844,14 @@ public abstract class ModificationStatement implements CQLStatement
CFMetaData cfm,
VariableSpecifications boundNames,
Operations operations,
List<Relation> relations,
WhereClause where,
Conditions conditions)
{
if (where.containsCustomExpressions())
throw new InvalidRequestException(CUSTOM_EXPRESSIONS_NOT_ALLOWED);
boolean applyOnlyToStaticColumns = appliesOnlyToStaticColumns(operations, conditions);
return new StatementRestrictions(type, cfm, relations, boundNames, applyOnlyToStaticColumns, false, false);
return new StatementRestrictions(type, cfm, where, boundNames, applyOnlyToStaticColumns, false, false);
}
/**

11
src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
View File

@ -23,10 +23,9 @@ import java.util.*;
import com.google.common.base.Objects;
import com.google.common.base.Predicate;
import com.google.common.collect.Iterables;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.slf4j.Logger;
import org.apache.cassandra.auth.Permission;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.ColumnDefinition;
@ -47,7 +46,6 @@ import org.apache.cassandra.db.rows.RowIterator;
import org.apache.cassandra.db.view.View;
import org.apache.cassandra.dht.AbstractBounds;
import org.apache.cassandra.exceptions.*;
import org.apache.cassandra.index.Index;
import org.apache.cassandra.index.SecondaryIndexManager;
import org.apache.cassandra.serializers.MarshalException;
import org.apache.cassandra.service.ClientState;
@ -59,6 +57,7 @@ import org.apache.cassandra.thrift.ThriftValidation;
import org.apache.cassandra.transport.messages.ResultMessage;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
import static org.apache.cassandra.cql3.statements.RequestValidations.checkFalse;
import static org.apache.cassandra.cql3.statements.RequestValidations.checkNotNull;
import static org.apache.cassandra.cql3.statements.RequestValidations.checkTrue;
@ -725,15 +724,15 @@ public class SelectStatement implements CQLStatement
{
public final Parameters parameters;
public final List<RawSelector> selectClause;
public final List<Relation> whereClause;
public final WhereClause whereClause;
public final Term.Raw limit;
public RawStatement(CFName cfName, Parameters parameters, List<RawSelector> selectClause, List<Relation> whereClause, Term.Raw limit)
public RawStatement(CFName cfName, Parameters parameters, List<RawSelector> selectClause, WhereClause whereClause, Term.Raw limit)
{
super(cfName);
this.parameters = parameters;
this.selectClause = selectClause;
this.whereClause = whereClause == null ? Collections.<Relation>emptyList() : whereClause;
this.whereClause = whereClause;
this.limit = limit;
}

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

@ -17,7 +17,6 @@
*/
package org.apache.cassandra.cql3.statements;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.List;
@ -154,7 +153,7 @@ public class UpdateStatement extends ModificationStatement
checkFalse(columnNames.size() != columnValues.size(), "Unmatched column names/values");
checkContainsNoDuplicates(columnNames, "The column names contains duplicates");
List<Relation> relations = new ArrayList<>();
WhereClause.Builder whereClause = new WhereClause.Builder();
Operations operations = new Operations();
boolean hasClusteringColumnsSet = false;
@ -169,7 +168,7 @@ public class UpdateStatement extends ModificationStatement
if (def.isPrimaryKeyColumn())
{
relations.add(new SingleColumnRelation(columnNames.get(i), Operator.EQ, value));
whereClause.add(new SingleColumnRelation(columnNames.get(i), Operator.EQ, value));
}
else
{
@ -183,7 +182,7 @@ public class UpdateStatement extends ModificationStatement
StatementRestrictions restrictions = new StatementRestrictions(StatementType.INSERT,
cfm,
relations,
whereClause.build(),
boundNames,
applyOnlyToStaticColumns,
false,
@ -223,7 +222,7 @@ public class UpdateStatement extends ModificationStatement
Collection<ColumnDefinition> defs = cfm.allColumns();
Json.Prepared prepared = jsonValue.prepareAndCollectMarkers(cfm, defs, boundNames);
List<Relation> relations = new ArrayList<>();
WhereClause.Builder whereClause = new WhereClause.Builder();
Operations operations = new Operations();
boolean hasClusteringColumnsSet = false;
@ -235,9 +234,9 @@ public class UpdateStatement extends ModificationStatement
Term.Raw raw = prepared.getRawTermForColumn(def);
if (def.isPrimaryKeyColumn())
{
relations.add(new SingleColumnRelation(new ColumnIdentifier.ColumnIdentifierValue(def.name),
Operator.EQ,
raw));
whereClause.add(new SingleColumnRelation(new ColumnIdentifier.ColumnIdentifierValue(def.name),
Operator.EQ,
raw));
}
else
{
@ -251,7 +250,7 @@ public class UpdateStatement extends ModificationStatement
StatementRestrictions restrictions = new StatementRestrictions(StatementType.INSERT,
cfm,
relations,
whereClause.build(),
boundNames,
applyOnlyToStaticColumns,
false,
@ -271,7 +270,7 @@ public class UpdateStatement extends ModificationStatement
{
// Provided for an UPDATE
private final List<Pair<ColumnIdentifier.Raw, Operation.RawUpdate>> updates;
private final List<Relation> whereClause;
private final WhereClause whereClause;
/**
* Creates a new UpdateStatement from a column family name, columns map, consistency
@ -286,7 +285,7 @@ public class UpdateStatement extends ModificationStatement
public ParsedUpdate(CFName name,
Attributes.Raw attrs,
List<Pair<ColumnIdentifier.Raw, Operation.RawUpdate>> updates,
List<Relation> whereClause,
WhereClause whereClause,
List<Pair<ColumnIdentifier.Raw, ColumnCondition.Raw>> conditions,
boolean ifExists)
{

141
src/java/org/apache/cassandra/db/filter/RowFilter.java
View File

@ -22,6 +22,7 @@ import java.nio.ByteBuffer;
import java.util.*;
import com.google.common.base.Objects;
import org.apache.commons.lang3.builder.ToStringBuilder;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.ColumnDefinition;
@ -34,6 +35,7 @@ import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.io.util.DataInputPlus;
import org.apache.cassandra.io.util.DataOutputPlus;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.schema.IndexMetadata;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.FBUtilities;
@ -52,7 +54,7 @@ import static org.apache.cassandra.cql3.statements.RequestValidations.checkNotNu
public abstract class RowFilter implements Iterable<RowFilter.Expression>
{
public static final Serializer serializer = new Serializer();
public static final RowFilter NONE = new CQLFilter(Collections.<Expression>emptyList());
public static final RowFilter NONE = new CQLFilter(Collections.emptyList());
protected final List<Expression> expressions;
@ -63,17 +65,17 @@ public abstract class RowFilter implements Iterable<RowFilter.Expression>
public static RowFilter create()
{
return new CQLFilter(new ArrayList<Expression>());
return new CQLFilter(new ArrayList<>());
}
public static RowFilter create(int capacity)
{
return new CQLFilter(new ArrayList<Expression>(capacity));
return new CQLFilter(new ArrayList<>(capacity));
}
public static RowFilter forThrift(int capacity)
{
return new ThriftFilter(new ArrayList<Expression>(capacity));
return new ThriftFilter(new ArrayList<>(capacity));
}
public void add(ColumnDefinition def, Operator op, ByteBuffer value)
@ -92,6 +94,11 @@ public abstract class RowFilter implements Iterable<RowFilter.Expression>
expressions.add(new ThriftExpression(metadata, name, op, value));
}
public void addCustomIndexExpression(CFMetaData cfm, IndexMetadata targetIndex, ByteBuffer value)
{
expressions.add(new CustomExpression(cfm, targetIndex, value));
}
public List<Expression> getExpressions()
{
return expressions;
@ -254,7 +261,7 @@ public abstract class RowFilter implements Iterable<RowFilter.Expression>
private static final Serializer serializer = new Serializer();
// Note: the order of this enum matter, it's used for serialization
protected enum Kind { SIMPLE, MAP_EQUALITY, THRIFT_DYN_EXPR }
protected enum Kind { SIMPLE, MAP_EQUALITY, THRIFT_DYN_EXPR, CUSTOM }
abstract Kind kind();
protected final ColumnDefinition column;
@ -268,6 +275,11 @@ public abstract class RowFilter implements Iterable<RowFilter.Expression>
this.value = value;
}
public boolean isCustom()
{
return kind() == Kind.CUSTOM;
}
public ColumnDefinition column()
{
return column;
@ -369,12 +381,24 @@ public abstract class RowFilter implements Iterable<RowFilter.Expression>
{
public void serialize(Expression expression, DataOutputPlus out, int version) throws IOException
{
ByteBufferUtil.writeWithShortLength(expression.column.name.bytes, out);
expression.operator.writeTo(out);
if (version >= MessagingService.VERSION_30)
out.writeByte(expression.kind().ordinal());
// Custom expressions include neither a column or operator, but all
// other expressions do. Also, custom expressions are 3.0+ only, so
// the column & operator will always be the first things written for
// any pre-3.0 version
if (expression.kind() == Kind.CUSTOM)
{
assert version >= MessagingService.VERSION_30;
IndexMetadata.serializer.serialize(((CustomExpression)expression).targetIndex, out, version);
ByteBufferUtil.writeWithShortLength(expression.value, out);
return;
}
ByteBufferUtil.writeWithShortLength(expression.column.name.bytes, out);
expression.operator.writeTo(out);
switch (expression.kind())
{
case SIMPLE:
@ -400,19 +424,30 @@ public abstract class RowFilter implements Iterable<RowFilter.Expression>
public Expression deserialize(DataInputPlus in, int version, CFMetaData metadata) throws IOException
{
ByteBuffer name = ByteBufferUtil.readWithShortLength(in);
Operator operator = Operator.readFrom(in);
Kind kind = null;
ByteBuffer name;
Operator operator;
ColumnDefinition column;
ColumnDefinition column = metadata.getColumnDefinition(name);
if (!metadata.isCompactTable() && column == null)
throw new RuntimeException("Unknown (or dropped) column " + UTF8Type.instance.getString(name) + " during deserialization");
Kind kind;
if (version >= MessagingService.VERSION_30)
{
kind = Kind.values()[in.readByte()];
// custom expressions (3.0+ only) do not contain a column or operator, only a value
if (kind == Kind.CUSTOM)
{
return new CustomExpression(metadata,
IndexMetadata.serializer.deserialize(in, version, metadata),
ByteBufferUtil.readWithShortLength(in));
}
}
else
name = ByteBufferUtil.readWithShortLength(in);
operator = Operator.readFrom(in);
column = metadata.getColumnDefinition(name);
if (!metadata.isCompactTable() && column == null)
throw new RuntimeException("Unknown (or dropped) column " + UTF8Type.instance.getString(name) + " during deserialization");
if (version < MessagingService.VERSION_30)
{
if (column == null)
kind = Kind.THRIFT_DYN_EXPR;
@ -422,6 +457,7 @@ public abstract class RowFilter implements Iterable<RowFilter.Expression>
kind = Kind.SIMPLE;
}
assert kind != null;
switch (kind)
{
case SIMPLE:
@ -446,10 +482,16 @@ public abstract class RowFilter implements Iterable<RowFilter.Expression>
throw new AssertionError();
}
public long serializedSize(Expression expression, int version)
{
long size = ByteBufferUtil.serializedSizeWithShortLength(expression.column().name.bytes)
+ expression.operator.serializedSize();
// version 3.0+ includes a byte for Kind
long size = version >= MessagingService.VERSION_30 ? 1 : 0;
// custom expressions don't include a column or operator, all other expressions do
if (expression.kind() != Kind.CUSTOM)
size += ByteBufferUtil.serializedSizeWithShortLength(expression.column().name.bytes)
+ expression.operator.serializedSize();
switch (expression.kind())
{
@ -467,6 +509,11 @@ public abstract class RowFilter implements Iterable<RowFilter.Expression>
case THRIFT_DYN_EXPR:
size += ByteBufferUtil.serializedSizeWithShortLength(((ThriftExpression)expression).value);
break;
case CUSTOM:
if (version >= MessagingService.VERSION_30)
size += IndexMetadata.serializer.serializedSize(((CustomExpression)expression).targetIndex, version)
+ ByteBufferUtil.serializedSizeWithShortLength(expression.value);
break;
}
return size;
}
@ -743,6 +790,62 @@ public abstract class RowFilter implements Iterable<RowFilter.Expression>
}
}
/**
* A custom index expression for use with 2i implementations which support custom syntax and which are not
* necessarily linked to a single column in the base table.
*/
public static final class CustomExpression extends Expression
{
private final IndexMetadata targetIndex;
private final CFMetaData cfm;
public CustomExpression(CFMetaData cfm, IndexMetadata targetIndex, ByteBuffer value)
{
// The operator is not relevant, but Expression requires it so for now we just hardcode EQ
super(makeDefinition(cfm, targetIndex), Operator.EQ, value);
this.targetIndex = targetIndex;
this.cfm = cfm;
}
private static ColumnDefinition makeDefinition(CFMetaData cfm, IndexMetadata index)
{
// Similarly to how we handle non-defined columns in thift, we create a fake column definition to
// represent the target index. This is definitely something that can be improved though.
return ColumnDefinition.regularDef(cfm, ByteBuffer.wrap(index.name.getBytes()), BytesType.instance);
}
public IndexMetadata getTargetIndex()
{
return targetIndex;
}
public ByteBuffer getValue()
{
return value;
}
public String toString()
{
return String.format("expr(%s, %s)",
targetIndex.name,
Keyspace.openAndGetStore(cfm)
.indexManager
.getIndex(targetIndex)
.customExpressionValueType());
}
Kind kind()
{
return Kind.CUSTOM;
}
// Filtering by custom expressions isn't supported yet, so just accept any row
public boolean isSatisfiedBy(DecoratedKey partitionKey, Row row)
{
return true;
}
}
public static class Serializer
{
public void serialize(RowFilter filter, DataOutputPlus out, int version) throws IOException
@ -751,6 +854,7 @@ public abstract class RowFilter implements Iterable<RowFilter.Expression>
out.writeUnsignedVInt(filter.expressions.size());
for (Expression expr : filter.expressions)
Expression.serializer.serialize(expr, out, version);
}
public RowFilter deserialize(DataInputPlus in, int version, CFMetaData metadata) throws IOException
@ -760,6 +864,7 @@ public abstract class RowFilter implements Iterable<RowFilter.Expression>
List<Expression> expressions = new ArrayList<>(size);
for (int i = 0; i < size; i++)
expressions.add(Expression.serializer.deserialize(in, version, metadata));
return forThrift
? new ThriftFilter(expressions)
: new CQLFilter(expressions);

22
src/java/org/apache/cassandra/index/Index.java
View File

@ -8,6 +8,7 @@ import org.apache.cassandra.config.ColumnDefinition;
import org.apache.cassandra.cql3.Operator;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.filter.RowFilter;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.partitions.PartitionIterator;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
@ -215,7 +216,6 @@ public interface Index
*/
public boolean dependsOn(ColumnDefinition column);
// TODO : this will change when we decouple indexes from specific columns for real per-row indexes
/**
* Called to determine whether this index can provide a searcher to execute a query on the
* supplied column using the specified operator. This forms part of the query validation done
@ -226,6 +226,19 @@ public interface Index
*/
public boolean supportsExpression(ColumnDefinition column, Operator operator);
/**
* If the index supports custom search expressions using the
* {@code}SELECT * FROM table WHERE expr(index_name, expression){@code} syntax, this
* method should return the expected type of the expression argument.
* For example, if the index supports custom expressions as Strings, calls to this
* method should return {@code}UTF8Type.instance{@code}.
* If the index implementation does not support custom expressions, then it should
* return null.
* @return an the type of custom index expressions supported by this index, or an
* null if custom expressions are not supported.
*/
public AbstractType<?> customExpressionValueType();
/**
* Transform an initial RowFilter into the filter that will still need to applied
* to a set of Rows after the index has performed it's initial scan.
@ -393,10 +406,15 @@ public interface Index
/**
* Factory method for query time search helper.
* Custom index implementations should perform any validation of query expressions here and throw a meaningful
* InvalidRequestException when any expression is invalid.
*
* @param command the read command being executed
* @return an Searcher with which to perform the supplied command
* @throws InvalidRequestException if the command's expressions are invalid according to the
* specific syntax supported by the index implementation.
*/
public Searcher searcherFor(ReadCommand command);
public Searcher searcherFor(ReadCommand command) throws InvalidRequestException;
/**
* Performs the actual index lookup during execution of a ReadCommand.

46
src/java/org/apache/cassandra/index/SecondaryIndexManager.java
View File

@ -550,11 +550,21 @@ public class SecondaryIndexManager implements IndexRegistry
}
/**
* Called at query time to find the most selective of the registered index implementation
* (i.e. the one likely to return the fewest results) from those registered.
* Implementation specific validation of the target expression by the most selective
* index should be performed in the searcherFor method to ensure that we pick the right
* index regardless of the validity of the expression.
* Called at query time to choose which (if any) of the registered index implementations to use for a given query.
*
* This is a two step processes, firstly compiling the set of searchable indexes then choosing the one which reduces
* the search space the most.
*
* In the first phase, if the command's RowFilter contains any custom index expressions, the indexes that they
* specify are automatically included. Following that, the registered indexes are filtered to include only those
* which support the standard expressions in the RowFilter.
*
* The filtered set then sorted by selectivity, as reported by the Index implementations' getEstimatedResultRows
* method.
*
* Implementation specific validation of the target expression, either custom or standard, by the selected
* index should be performed in the searcherFor method to ensure that we pick the right index regardless of
* the validity of the expression.
*
* This method is only called once during the lifecycle of a ReadCommand and the result is
* cached for future use when obtaining a Searcher, getting the index's underlying CFS for
@ -569,12 +579,20 @@ public class SecondaryIndexManager implements IndexRegistry
if (indexes.isEmpty() || command.rowFilter().isEmpty())
return null;
Set<Index> searchableIndexes = new HashSet<>();
List<Index> searchableIndexes = new ArrayList<>();
for (RowFilter.Expression expression : command.rowFilter())
{
indexes.values().stream()
.filter(index -> index.supportsExpression(expression.column(), expression.operator()))
.forEach(searchableIndexes::add);
if (expression.isCustom())
{
RowFilter.CustomExpression customExpression = (RowFilter.CustomExpression)expression;
searchableIndexes.add(indexes.get(customExpression.getTargetIndex().name));
}
else
{
indexes.values().stream()
.filter(index -> index.supportsExpression(expression.column(), expression.operator()))
.forEach(searchableIndexes::add);
}
}
if (searchableIndexes.isEmpty())
@ -584,10 +602,12 @@ public class SecondaryIndexManager implements IndexRegistry
return null;
}
Index selected = searchableIndexes.stream()
.max((a, b) -> Longs.compare(a.getEstimatedResultRows(),
b.getEstimatedResultRows()))
.orElseThrow(() -> new AssertionError("Could not select most selective index"));
Index selected = searchableIndexes.size() == 1
? searchableIndexes.get(0)
: searchableIndexes.stream()
.max((a, b) -> Longs.compare(a.getEstimatedResultRows(),
b.getEstimatedResultRows()))
.orElseThrow(() -> new AssertionError("Could not select most selective index"));
// pay for an additional threadlocal get() rather than build the strings unnecessarily
if (Tracing.isTracing())

5
src/java/org/apache/cassandra/index/internal/CassandraIndex.java
View File

@ -246,6 +246,11 @@ public abstract class CassandraIndex implements Index
return supportsExpression(expression.column(), expression.operator());
}
public AbstractType<?> customExpressionValueType()
{
return null;
}
public long getEstimatedResultRows()
{
return indexCfs.getMeanColumns();

32
src/java/org/apache/cassandra/net/MessagingService.java
View File

@ -97,6 +97,7 @@ public final class MessagingService implements MessagingServiceMBean
public static final int PROTOCOL_MAGIC = 0xCA552DFA;
private boolean allNodesAtLeast22 = true;
private boolean allNodesAtLeast30 = true;
/* All verb handler identifiers */
public enum Verb
@ -845,6 +846,11 @@ public final class MessagingService implements MessagingServiceMBean
return allNodesAtLeast22;
}
public boolean areAllNodesAtLeast30()
{
return allNodesAtLeast30;
}
/**
* @return the last version associated with address, or @param version if this is the first such version
*/
@ -854,12 +860,14 @@ public final class MessagingService implements MessagingServiceMBean
if (version < VERSION_22)
allNodesAtLeast22 = false;
if (version < VERSION_30)
allNodesAtLeast30 = false;
Integer v = versions.put(endpoint, version);
// if the version was increased to 2.2 or later, see if all nodes are >= 2.2 now
if (v != null && v < VERSION_22 && version >= VERSION_22)
refreshAllNodesAtLeast22();
// if the version was increased to 2.2 or later see if the min version across the cluster has changed
if (v != null && (v < VERSION_30 && version >= VERSION_22))
refreshAllNodeMinVersions();
return v == null ? version : v;
}
@ -868,21 +876,29 @@ public final class MessagingService implements MessagingServiceMBean
{
logger.debug("Resetting version for {}", endpoint);
Integer removed = versions.remove(endpoint);
if (removed != null && removed <= VERSION_22)
refreshAllNodesAtLeast22();
if (removed != null && removed <= VERSION_30)
refreshAllNodeMinVersions();
}
private void refreshAllNodesAtLeast22()
private void refreshAllNodeMinVersions()
{
for (Integer version: versions.values())
boolean anyNodeLowerThan30 = false;
for (Integer version : versions.values())
{
if (version < VERSION_22)
if (version < MessagingService.VERSION_30)
{
anyNodeLowerThan30 = true;
allNodesAtLeast30 = false;
}
if (version < MessagingService.VERSION_22)
{
allNodesAtLeast22 = false;
return;
}
}
allNodesAtLeast22 = true;
allNodesAtLeast30 = !anyNodeLowerThan30;
}
public int getVersion(InetAddress endpoint)

38
test/unit/org/apache/cassandra/cql3/PreparedStatementsTest.java
View File

@ -24,12 +24,15 @@ import org.junit.Test;
import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.exceptions.SyntaxError;
import org.apache.cassandra.SchemaLoader;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.Schema;
import org.apache.cassandra.index.StubIndex;
import org.apache.cassandra.service.EmbeddedCassandraService;
import static junit.framework.Assert.assertEquals;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.fail;
public class PreparedStatementsTest extends SchemaLoader
{
@ -127,4 +130,37 @@ public class PreparedStatementsTest extends SchemaLoader
assertEquals(1, session.execute(preparedSelect.bind(1)).all().size());
}
@Test
public void prepareAndExecuteWithCustomExpressions() throws Throwable
{
session.execute(dropKsStatement);
session.execute(createKsStatement);
String table = "custom_expr_test";
String index = "custom_index";
session.execute(String.format("CREATE TABLE IF NOT EXISTS %s.%s (id int PRIMARY KEY, cid int, val text);",
KEYSPACE, table));
session.execute(String.format("CREATE CUSTOM INDEX %s ON %s.%s(val) USING '%s'",
index, KEYSPACE, table, StubIndex.class.getName()));
session.execute(String.format("INSERT INTO %s.%s(id, cid, val) VALUES (0, 0, 'test')", KEYSPACE, table));
PreparedStatement prepared1 = session.prepare(String.format("SELECT * FROM %s.%s WHERE expr(%s, 'foo')",
KEYSPACE, table, index));
assertEquals(1, session.execute(prepared1.bind()).all().size());
PreparedStatement prepared2 = session.prepare(String.format("SELECT * FROM %s.%s WHERE expr(%s, ?)",
KEYSPACE, table, index));
assertEquals(1, session.execute(prepared2.bind("foo bar baz")).all().size());
try
{
session.prepare(String.format("SELECT * FROM %s.%s WHERE expr(?, 'foo bar baz')", KEYSPACE, table));
fail("Expected syntax exception, but none was thrown");
}
catch(SyntaxError e)
{
assertEquals("Bind variables cannot be used for index names", e.getMessage());
}
}
}

106
test/unit/org/apache/cassandra/index/CustomIndexTest.java
View File

@ -10,8 +10,14 @@ import org.junit.Test;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.cql3.CQLTester;
import org.apache.cassandra.cql3.ColumnIdentifier;
import org.apache.cassandra.cql3.restrictions.IndexRestrictions;
import org.apache.cassandra.cql3.statements.IndexTarget;
import org.apache.cassandra.cql3.statements.ModificationStatement;
import org.apache.cassandra.cql3.statements.SelectStatement;
import org.apache.cassandra.db.ColumnFamilyStore;
import org.apache.cassandra.db.ReadCommand;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.schema.IndexMetadata;
import org.apache.cassandra.schema.Indexes;
@ -294,6 +300,80 @@ public class CustomIndexTest extends CQLTester
assertIndexCreated("no_targets", new HashMap<>());
}
@Test
public void testCustomIndexExpressionSyntax() throws Throwable
{
Object[] row = row(0, 0, 0, 0);
createTable("CREATE TABLE %s (a int, b int, c int, d int, PRIMARY KEY (a, b))");
execute("INSERT INTO %s (a, b, c, d) VALUES (?, ?, ?, ?)", row);
assertInvalidMessage(String.format(IndexRestrictions.INDEX_NOT_FOUND, "custom_index", keyspace(), currentTable()),
"SELECT * FROM %s WHERE expr(custom_index, 'foo bar baz')");
createIndex(String.format("CREATE CUSTOM INDEX custom_index ON %%s(c) USING '%s'", StubIndex.class.getName()));
assertInvalidMessage(String.format(IndexRestrictions.INDEX_NOT_FOUND, "no_such_index", keyspace(), currentTable()),
"SELECT * FROM %s WHERE expr(no_such_index, 'foo bar baz ')");
// simple case
assertRows(execute("SELECT * FROM %s WHERE expr(custom_index, 'foo bar baz')"), row);
assertRows(execute("SELECT * FROM %s WHERE expr(\"custom_index\", 'foo bar baz')"), row);
assertRows(execute("SELECT * FROM %s WHERE expr(custom_index, $$foo \" ~~~ bar Baz$$)"), row);
// multiple expressions on the same index
assertInvalidMessage(IndexRestrictions.MULTIPLE_EXPRESSIONS,
"SELECT * FROM %s WHERE expr(custom_index, 'foo') AND expr(custom_index, 'bar')");
// multiple expressions on different indexes
createIndex(String.format("CREATE CUSTOM INDEX other_custom_index ON %%s(d) USING '%s'", StubIndex.class.getName()));
assertInvalidMessage(IndexRestrictions.MULTIPLE_EXPRESSIONS,
"SELECT * FROM %s WHERE expr(custom_index, 'foo') AND expr(other_custom_index, 'bar')");
assertInvalidMessage(SelectStatement.REQUIRES_ALLOW_FILTERING_MESSAGE,
"SELECT * FROM %s WHERE expr(custom_index, 'foo') AND d=0");
assertRows(execute("SELECT * FROM %s WHERE expr(custom_index, 'foo') AND d=0 ALLOW FILTERING"), row);
}
@Test
public void customIndexDoesntSupportCustomExpressions() throws Throwable
{
createTable("CREATE TABLE %s (a int, b int, c int, d int, PRIMARY KEY (a, b))");
createIndex(String.format("CREATE CUSTOM INDEX custom_index ON %%s(c) USING '%s'",
NoCustomExpressionsIndex.class.getName()));
assertInvalidMessage(String.format( IndexRestrictions.CUSTOM_EXPRESSION_NOT_SUPPORTED, "custom_index"),
"SELECT * FROM %s WHERE expr(custom_index, 'foo bar baz')");
}
@Test
public void customIndexRejectsExpressionSyntax() throws Throwable
{
createTable("CREATE TABLE %s (a int, b int, c int, d int, PRIMARY KEY (a, b))");
createIndex(String.format("CREATE CUSTOM INDEX custom_index ON %%s(c) USING '%s'",
ExpressionRejectingIndex.class.getName()));
assertInvalidMessage("None shall pass", "SELECT * FROM %s WHERE expr(custom_index, 'foo bar baz')");
}
@Test
public void customExpressionsMustTargetCustomIndex() throws Throwable
{
createTable("CREATE TABLE %s (a int, b int, c int, d int, PRIMARY KEY (a, b))");
createIndex("CREATE INDEX non_custom_index ON %s(c)");
assertInvalidMessage(String.format(IndexRestrictions.NON_CUSTOM_INDEX_IN_EXPRESSION, "non_custom_index"),
"SELECT * FROM %s WHERE expr(non_custom_index, 'c=0')");
}
@Test
public void customExpressionsDisallowedInModifications() throws Throwable
{
createTable("CREATE TABLE %s (a int, b int, c int, d int, PRIMARY KEY (a, b))");
createIndex(String.format("CREATE CUSTOM INDEX custom_index ON %%s(c) USING '%s'", StubIndex.class.getName()));
assertInvalidMessage(ModificationStatement.CUSTOM_EXPRESSIONS_NOT_ALLOWED,
"DELETE FROM %s WHERE expr(custom_index, 'foo bar baz ')");
assertInvalidMessage(ModificationStatement.CUSTOM_EXPRESSIONS_NOT_ALLOWED,
"UPDATE %s SET d=0 WHERE expr(custom_index, 'foo bar baz ')");
}
private void testCreateIndex(String indexName, String... targetColumnNames) throws Throwable
{
createIndex(String.format("CREATE CUSTOM INDEX %s ON %%s(%s) USING '%s'",
@ -362,4 +442,30 @@ public class CustomIndexTest extends CQLTester
return false;
}
}
public static final class NoCustomExpressionsIndex extends StubIndex
{
public NoCustomExpressionsIndex(ColumnFamilyStore baseCfs, IndexMetadata metadata)
{
super(baseCfs, metadata);
}
public AbstractType<?> customExpressionValueType()
{
return null;
}
}
public static final class ExpressionRejectingIndex extends StubIndex
{
public ExpressionRejectingIndex(ColumnFamilyStore baseCfs, IndexMetadata metadata)
{
super(baseCfs, metadata);
}
public Searcher searcherFor(ReadCommand command) throws InvalidRequestException
{
throw new InvalidRequestException("None shall pass");
}
}
}

44
test/unit/org/apache/cassandra/index/StubIndex.java
View File

@ -26,8 +26,11 @@ import org.apache.cassandra.config.ColumnDefinition;
import org.apache.cassandra.cql3.Operator;
import org.apache.cassandra.db.*;
import org.apache.cassandra.db.filter.RowFilter;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.UTF8Type;
import org.apache.cassandra.db.partitions.PartitionIterator;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
import org.apache.cassandra.db.rows.Row;
import org.apache.cassandra.exceptions.InvalidRequestException;
import org.apache.cassandra.index.transactions.IndexTransaction;
@ -35,6 +38,12 @@ import org.apache.cassandra.schema.IndexMetadata;
import org.apache.cassandra.utils.Pair;
import org.apache.cassandra.utils.concurrent.OpOrder;
/**
* Basic custom index implementation for testing.
* During indexing by default it just records the updates for later inspection.
* At query time, the Searcher implementation simply performs a local scan of the entire target table
* with no further filtering applied.
*/
public class StubIndex implements Index
{
public List<DeletionTime> partitionDeletions = new ArrayList<>();
@ -79,6 +88,11 @@ public class StubIndex implements Index
return operator == Operator.EQ;
}
public AbstractType<?> customExpressionValueType()
{
return UTF8Type.instance;
}
public RowFilter getPostIndexQueryFilter(RowFilter filter)
{
return filter;
@ -185,13 +199,37 @@ public class StubIndex implements Index
}
public Searcher searcherFor(ReadCommand command)
public Searcher searcherFor(final ReadCommand command)
{
return null;
return orderGroup -> new InternalPartitionRangeReadCommand((PartitionRangeReadCommand)command)
.queryStorageInternal(baseCfs, orderGroup);
}
public BiFunction<PartitionIterator, ReadCommand, PartitionIterator> postProcessorFor(ReadCommand readCommand)
{
return null;
return (iter, command) -> iter;
}
private static final class InternalPartitionRangeReadCommand extends PartitionRangeReadCommand
{
private InternalPartitionRangeReadCommand(PartitionRangeReadCommand original)
{
super(original.isDigestQuery(),
original.digestVersion(),
original.isForThrift(),
original.metadata(),
original.nowInSec(),
original.columnFilter(),
original.rowFilter(),
original.limits(),
original.dataRange(),
Optional.empty());
}
private UnfilteredPartitionIterator queryStorageInternal(ColumnFamilyStore cfs, ReadOrderGroup orderGroup)
{
return queryStorage(cfs, orderGroup);
}
}
}

6
test/unit/org/apache/cassandra/index/internal/CustomCassandraIndex.java
View File

@ -20,6 +20,7 @@ import org.apache.cassandra.db.compaction.CompactionManager;
import org.apache.cassandra.db.filter.RowFilter;
import org.apache.cassandra.db.lifecycle.SSTableSet;
import org.apache.cassandra.db.lifecycle.View;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.partitions.PartitionIterator;
import org.apache.cassandra.db.partitions.PartitionUpdate;
import org.apache.cassandra.db.rows.*;
@ -185,6 +186,11 @@ public class CustomCassandraIndex implements Index
&& supportsOperator(indexedColumn, operator);
}
public AbstractType<?> customExpressionValueType()
{
return null;
}
private boolean supportsExpression(RowFilter.Expression expression)
{
return supportsExpression(expression.column(), expression.operator());