cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From s...@apache.org
Subject [2/3] cassandra git commit: Support for custom index expressions in SELECT
Date Fri, 18 Sep 2015 10:35:22 GMT
Support for custom index expressions in SELECT

Patch by Sam Tunnicliffe; reviewed by Sylvain Lebresne for
CASSANDRA-10217


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/64e2f5dd
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/64e2f5dd
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/64e2f5dd

Branch: refs/heads/trunk
Commit: 64e2f5ddaa5a659c5b4109017a06d481290ec27d
Parents: 9f335fe
Author: Sam Tunnicliffe <sam@beobal.com>
Authored: Mon Sep 7 19:43:14 2015 +0100
Committer: Sam Tunnicliffe <sam@beobal.com>
Committed: Fri Sep 18 11:30:32 2015 +0100

----------------------------------------------------------------------
 CHANGES.txt                                     |   1 +
 src/java/org/apache/cassandra/cql3/Cql.g        |  26 +++-
 .../org/apache/cassandra/cql3/WhereClause.java  |  74 ++++++++++
 .../restrictions/CustomIndexExpression.java     |  58 ++++++++
 .../cql3/restrictions/IndexRestrictions.java    |  83 +++++++++++
 .../restrictions/StatementRestrictions.java     |  76 +++++++---
 .../cql3/statements/DeleteStatement.java        |   4 +-
 .../cql3/statements/ModificationStatement.java  |  14 +-
 .../cql3/statements/SelectStatement.java        |  11 +-
 .../cql3/statements/UpdateStatement.java        |  21 ++-
 .../apache/cassandra/db/filter/RowFilter.java   | 141 ++++++++++++++++---
 src/java/org/apache/cassandra/index/Index.java  |  22 ++-
 .../cassandra/index/SecondaryIndexManager.java  |  46 ++++--
 .../index/internal/CassandraIndex.java          |   5 +
 .../apache/cassandra/net/MessagingService.java  |  32 +++--
 .../cassandra/cql3/PreparedStatementsTest.java  |  38 ++++-
 .../apache/cassandra/index/CustomIndexTest.java | 106 ++++++++++++++
 .../org/apache/cassandra/index/StubIndex.java   |  44 +++++-
 .../index/internal/CustomCassandraIndex.java    |   6 +
 19 files changed, 717 insertions(+), 91 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/64e2f5dd/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index d6554e9..cc57ba6 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 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)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/64e2f5dd/src/java/org/apache/cassandra/cql3/Cql.g
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/Cql.g b/src/java/org/apache/cassandra/cql3/Cql.g
index afef224..cd52c1c 100644
--- a/src/java/org/apache/cassandra/cql3/Cql.g
+++ b/src/java/org/apache/cassandra/cql3/Cql.g
@@ -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)); }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/64e2f5dd/src/java/org/apache/cassandra/cql3/WhereClause.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/WhereClause.java b/src/java/org/apache/cassandra/cql3/WhereClause.java
new file mode 100644
index 0000000..9d4e51a
--- /dev/null
+++ b/src/java/org/apache/cassandra/cql3/WhereClause.java
@@ -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);
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/64e2f5dd/src/java/org/apache/cassandra/cql3/restrictions/CustomIndexExpression.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/restrictions/CustomIndexExpression.java b/src/java/org/apache/cassandra/cql3/restrictions/CustomIndexExpression.java
new file mode 100644
index 0000000..65c1bb3
--- /dev/null
+++ b/src/java/org/apache/cassandra/cql3/restrictions/CustomIndexExpression.java
@@ -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));
+    }
+}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/64e2f5dd/src/java/org/apache/cassandra/cql3/restrictions/IndexRestrictions.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/restrictions/IndexRestrictions.java b/src/java/org/apache/cassandra/cql3/restrictions/IndexRestrictions.java
new file mode 100644
index 0000000..c7f6b5f
--- /dev/null
+++ b/src/java/org/apache/cassandra/cql3/restrictions/IndexRestrictions.java
@@ -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()));
+    }
+}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/64e2f5dd/src/java/org/apache/cassandra/cql3/restrictions/StatementRestrictions.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/restrictions/StatementRestrictions.java b/src/java/org/apache/cassandra/cql3/restrictions/StatementRestrictions.java
index 3cf6bfb..b1c7aff 100644
--- a/src/java/org/apache/cassandra/cql3/restrictions/StatementRestrictions.java
+++ b/src/java/org/apache/cassandra/cql3/restrictions/StatementRestrictions.java
@@ -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());
     }
+
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/64e2f5dd/src/java/org/apache/cassandra/cql3/statements/DeleteStatement.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/statements/DeleteStatement.java b/src/java/org/apache/cassandra/cql3/statements/DeleteStatement.java
index cd6ce77..da188a9 100644
--- a/src/java/org/apache/cassandra/cql3/statements/DeleteStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/DeleteStatement.java
@@ -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)
         {

http://git-wip-us.apache.org/repos/asf/cassandra/blob/64e2f5dd/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
index a04af4c..3f3f3f5 100644
--- a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
@@ -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);
         }
 
         /**

http://git-wip-us.apache.org/repos/asf/cassandra/blob/64e2f5dd/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
index 18e402b..cb6de2b 100644
--- a/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/SelectStatement.java
@@ -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;
         }
 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/64e2f5dd/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java b/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
index 8fa16e1..f8435eb 100644
--- a/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/UpdateStatement.java
@@ -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)
         {

http://git-wip-us.apache.org/repos/asf/cassandra/blob/64e2f5dd/src/java/org/apache/cassandra/db/filter/RowFilter.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/filter/RowFilter.java b/src/java/org/apache/cassandra/db/filter/RowFilter.java
index bf92efb..b5968d5 100644
--- a/src/java/org/apache/cassandra/db/filter/RowFilter.java
+++ b/src/java/org/apache/cassandra/db/filter/RowFilter.java
@@ -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);

http://git-wip-us.apache.org/repos/asf/cassandra/blob/64e2f5dd/src/java/org/apache/cassandra/index/Index.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/index/Index.java b/src/java/org/apache/cassandra/index/Index.java
index f07baad..3ceec13 100644
--- a/src/java/org/apache/cassandra/index/Index.java
+++ b/src/java/org/apache/cassandra/index/Index.java
@@ -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
@@ -227,6 +227,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.
      * Used in ReadCommand#executeLocal to reduce the amount of filtering performed on the
@@ -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.

http://git-wip-us.apache.org/repos/asf/cassandra/blob/64e2f5dd/src/java/org/apache/cassandra/index/SecondaryIndexManager.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/index/SecondaryIndexManager.java b/src/java/org/apache/cassandra/index/SecondaryIndexManager.java
index 1af2f6e..47364f6 100644
--- a/src/java/org/apache/cassandra/index/SecondaryIndexManager.java
+++ b/src/java/org/apache/cassandra/index/SecondaryIndexManager.java
@@ -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())

http://git-wip-us.apache.org/repos/asf/cassandra/blob/64e2f5dd/src/java/org/apache/cassandra/index/internal/CassandraIndex.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/index/internal/CassandraIndex.java b/src/java/org/apache/cassandra/index/internal/CassandraIndex.java
index d10af1f..f6a10e5 100644
--- a/src/java/org/apache/cassandra/index/internal/CassandraIndex.java
+++ b/src/java/org/apache/cassandra/index/internal/CassandraIndex.java
@@ -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();

http://git-wip-us.apache.org/repos/asf/cassandra/blob/64e2f5dd/src/java/org/apache/cassandra/net/MessagingService.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/net/MessagingService.java b/src/java/org/apache/cassandra/net/MessagingService.java
index 23da27a..4fb67ec 100644
--- a/src/java/org/apache/cassandra/net/MessagingService.java
+++ b/src/java/org/apache/cassandra/net/MessagingService.java
@@ -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)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/64e2f5dd/test/unit/org/apache/cassandra/cql3/PreparedStatementsTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/cql3/PreparedStatementsTest.java b/test/unit/org/apache/cassandra/cql3/PreparedStatementsTest.java
index b5a28df..e01b812 100644
--- a/test/unit/org/apache/cassandra/cql3/PreparedStatementsTest.java
+++ b/test/unit/org/apache/cassandra/cql3/PreparedStatementsTest.java
@@ -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());
+        }
+    }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/64e2f5dd/test/unit/org/apache/cassandra/index/CustomIndexTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/index/CustomIndexTest.java b/test/unit/org/apache/cassandra/index/CustomIndexTest.java
index 4497364..40fb526 100644
--- a/test/unit/org/apache/cassandra/index/CustomIndexTest.java
+++ b/test/unit/org/apache/cassandra/index/CustomIndexTest.java
@@ -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");
+        }
+    }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/64e2f5dd/test/unit/org/apache/cassandra/index/StubIndex.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/index/StubIndex.java b/test/unit/org/apache/cassandra/index/StubIndex.java
index 544d482..c8a3241 100644
--- a/test/unit/org/apache/cassandra/index/StubIndex.java
+++ b/test/unit/org/apache/cassandra/index/StubIndex.java
@@ -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);
+        }
     }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/64e2f5dd/test/unit/org/apache/cassandra/index/internal/CustomCassandraIndex.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/index/internal/CustomCassandraIndex.java b/test/unit/org/apache/cassandra/index/internal/CustomCassandraIndex.java
index 8695018..cbcf069 100644
--- a/test/unit/org/apache/cassandra/index/internal/CustomCassandraIndex.java
+++ b/test/unit/org/apache/cassandra/index/internal/CustomCassandraIndex.java
@@ -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());


Mime
View raw message