cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From slebre...@apache.org
Subject [4/4] git commit: Support list index operation with conditions
Date Mon, 18 Aug 2014 08:22:51 GMT
Support list index operation with conditions

patch by slebresne; reviewed by thobbs for CASSANDRA-7499


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

Branch: refs/heads/cassandra-2.1
Commit: 5db108c314fa5064669eefef8e5f6a52a1500b96
Parents: 2ea11c1
Author: Sylvain Lebresne <sylvain@datastax.com>
Authored: Mon Aug 18 10:21:44 2014 +0200
Committer: Sylvain Lebresne <sylvain@datastax.com>
Committed: Mon Aug 18 10:22:40 2014 +0200

----------------------------------------------------------------------
 CHANGES.txt                                     |   1 +
 build.xml                                       |   2 +-
 .../cql3/statements/BatchStatement.java         |  23 +-
 .../cql3/statements/CQL3CasConditions.java      | 203 --------------
 .../cql3/statements/CQL3CasRequest.java         | 268 +++++++++++++++++++
 .../cql3/statements/ModificationStatement.java  |  57 ++--
 .../apache/cassandra/service/CASConditions.java |  39 ---
 .../apache/cassandra/service/CASRequest.java    |  45 ++++
 .../apache/cassandra/service/StorageProxy.java  |  13 +-
 .../cassandra/thrift/CassandraServer.java       |  16 +-
 10 files changed, 353 insertions(+), 314 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/5db108c3/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 4fa537b..cecf153 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 2.1.1
+ * Support list index operations with conditions (CASSANDRA-7499)
  * Add max live/tombstoned cells to nodetool cfstats output (CASSANDRA-7731)
  * Validate IPv6 wildcard addresses properly (CASSANDRA-7680)
  * (cqlsh) Error when tracing query (CASSANDRA-7613)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/5db108c3/build.xml
----------------------------------------------------------------------
diff --git a/build.xml b/build.xml
index 54f5f3d..d747bbc 100644
--- a/build.xml
+++ b/build.xml
@@ -25,7 +25,7 @@
     <property name="debuglevel" value="source,lines,vars"/>
 
     <!-- default version and SCM information -->
-    <property name="base.version" value="2.1.0-rc6"/>
+    <property name="base.version" value="2.1.1"/>
     <property name="scm.connection" value="scm:git://git.apache.org/cassandra.git"/>
     <property name="scm.developerConnection" value="scm:git://git.apache.org/cassandra.git"/>
     <property name="scm.url" value="http://git-wip-us.apache.org/repos/asf?p=cassandra.git;a=tree"/>

http://git-wip-us.apache.org/repos/asf/cassandra/blob/5db108c3/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java b/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java
index 90be914..17d1771 100644
--- a/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/BatchStatement.java
@@ -149,9 +149,6 @@ public class BatchStatement implements CQLStatement, MeasurableForPreparedCache
                     throw new InvalidRequestException("Batch with conditions cannot span
multiple tables");
                 ksName = stmt.keyspace();
                 cfName = stmt.columnFamily();
-
-                if (stmt.requiresRead())
-                    throw new InvalidRequestException("Operations using list indexes are
not allowed with IF conditions");
             }
         }
     }
@@ -240,7 +237,7 @@ public class BatchStatement implements CQLStatement, MeasurableForPreparedCache
      * Checks batch size to ensure threshold is met. If not, a warning is logged.
      * @param cfs ColumnFamilies that will store the batch's mutations.
      */
-    private void verifyBatchSize(Iterable<ColumnFamily> cfs)
+    public static void verifyBatchSize(Iterable<ColumnFamily> cfs)
     {
         long size = 0;
         long warnThreshold = DatabaseDescriptor.getBatchSizeWarnThreshold();
@@ -306,8 +303,7 @@ public class BatchStatement implements CQLStatement, MeasurableForPreparedCache
         ByteBuffer key = null;
         String ksName = null;
         String cfName = null;
-        ColumnFamily updates = null;
-        CQL3CasConditions conditions = null;
+        CQL3CasRequest casRequest = null;
         Set<ColumnDefinition> columnsWithConditions = new LinkedHashSet<>();
 
         for (int i = 0; i < statements.size(); i++)
@@ -323,8 +319,7 @@ public class BatchStatement implements CQLStatement, MeasurableForPreparedCache
                 key = pks.get(0);
                 ksName = statement.cfm.ksName;
                 cfName = statement.cfm.cfName;
-                conditions = new CQL3CasConditions(statement.cfm, now);
-                updates = ArrayBackedSortedColumns.factory.create(statement.cfm);
+                casRequest = new CQL3CasRequest(statement.cfm, key, true);
             }
             else if (!key.equals(pks.get(0)))
             {
@@ -334,22 +329,18 @@ public class BatchStatement implements CQLStatement, MeasurableForPreparedCache
             Composite clusteringPrefix = statement.createClusteringPrefix(statementOptions);
             if (statement.hasConditions())
             {
-                statement.addUpdatesAndConditions(key, clusteringPrefix, updates, conditions,
statementOptions, timestamp);
+                statement.addConditions(clusteringPrefix, casRequest, statementOptions);
                 // As soon as we have a ifNotExists, we set columnsWithConditions to null
so that everything is in the resultSet
                 if (statement.hasIfNotExistCondition() || statement.hasIfExistCondition())
                     columnsWithConditions = null;
                 else if (columnsWithConditions != null)
                     Iterables.addAll(columnsWithConditions, statement.getColumnsWithConditions());
             }
-            else
-            {
-                UpdateParameters params = statement.makeUpdateParameters(Collections.singleton(key),
clusteringPrefix, statementOptions, false, now);
-                statement.addUpdateForKey(updates, key, clusteringPrefix, params);
-            }
+            casRequest.addRowUpdate(clusteringPrefix, statement, statementOptions, timestamp);
         }
 
-        verifyBatchSize(Collections.singleton(updates));
-        ColumnFamily result = StorageProxy.cas(ksName, cfName, key, conditions, updates,
options.getSerialConsistency(), options.getConsistency());
+        ColumnFamily result = StorageProxy.cas(ksName, cfName, key, casRequest, options.getSerialConsistency(),
options.getConsistency());
+
         return new ResultMessage.Rows(ModificationStatement.buildCasResultSet(ksName, key,
cfName, result, columnsWithConditions, true, options.forStatement(0)));
     }
 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/5db108c3/src/java/org/apache/cassandra/cql3/statements/CQL3CasConditions.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/statements/CQL3CasConditions.java b/src/java/org/apache/cassandra/cql3/statements/CQL3CasConditions.java
deleted file mode 100644
index 8b5a403..0000000
--- a/src/java/org/apache/cassandra/cql3/statements/CQL3CasConditions.java
+++ /dev/null
@@ -1,203 +0,0 @@
-/*
- * 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.statements;
-
-import java.nio.ByteBuffer;
-import java.util.*;
-
-import org.apache.cassandra.cql3.*;
-import org.apache.cassandra.config.CFMetaData;
-import org.apache.cassandra.db.*;
-import org.apache.cassandra.db.composites.Composite;
-import org.apache.cassandra.db.filter.*;
-import org.apache.cassandra.exceptions.InvalidRequestException;
-import org.apache.cassandra.service.CASConditions;
-import org.apache.cassandra.utils.Pair;
-
-/**
- * Processed CAS conditions on potentially multiple rows of the same partition.
- */
-public class CQL3CasConditions implements CASConditions
-{
-    private final CFMetaData cfm;
-    private final long now;
-
-    // We index RowCondition by the prefix of the row they applied to for 2 reasons:
-    //   1) this allows to keep things sorted to build the ColumnSlice array below
-    //   2) this allows to detect when contradictory conditions are set (not exists with
some other conditions on the same row)
-    private final SortedMap<Composite, RowCondition> conditions;
-
-    public CQL3CasConditions(CFMetaData cfm, long now)
-    {
-        this.cfm = cfm;
-        // We will use now for Cell.isLive() which expects milliseconds but the argument
is in microseconds.
-        this.now = now / 1000;
-        this.conditions = new TreeMap<>(cfm.comparator);
-    }
-
-    public void addNotExist(Composite prefix) throws InvalidRequestException
-    {
-        RowCondition previous = conditions.put(prefix, new NotExistCondition(prefix, now));
-        if (previous != null && !(previous instanceof NotExistCondition))
-        {
-            // these should be prevented by the parser, but it doesn't hurt to check
-            if (previous instanceof ExistCondition)
-                throw new InvalidRequestException("Cannot mix IF EXISTS and IF NOT EXISTS
conditions for the same row");
-            else
-                throw new InvalidRequestException("Cannot mix IF conditions and IF NOT EXISTS
for the same row");
-        }
-    }
-
-    public void addExist(Composite prefix) throws InvalidRequestException
-    {
-        RowCondition previous = conditions.put(prefix, new ExistCondition(prefix, now));
-        // this should be prevented by the parser, but it doesn't hurt to check
-        if (previous != null && previous instanceof NotExistCondition)
-            throw new InvalidRequestException("Cannot mix IF EXISTS and IF NOT EXISTS conditions
for the same row");
-    }
-
-    public void addConditions(Composite prefix, Collection<ColumnCondition> conds,
QueryOptions options) throws InvalidRequestException
-    {
-        RowCondition condition = conditions.get(prefix);
-        if (condition == null)
-        {
-            condition = new ColumnsConditions(prefix, now);
-            conditions.put(prefix, condition);
-        }
-        else if (!(condition instanceof ColumnsConditions))
-        {
-            throw new InvalidRequestException("Cannot mix IF conditions and IF NOT EXISTS
for the same row");
-        }
-        ((ColumnsConditions)condition).addConditions(conds, options);
-    }
-
-    public IDiskAtomFilter readFilter()
-    {
-        assert !conditions.isEmpty();
-        ColumnSlice[] slices = new ColumnSlice[conditions.size()];
-        int i = 0;
-        // We always read CQL rows entirely as on CAS failure we want to be able to distinguish
between "row exists
-        // but all values for which there were conditions are null" and "row doesn't exists",
and we can't rely on the
-        // row marker for that (see #6623)
-        for (Composite prefix : conditions.keySet())
-            slices[i++] = prefix.slice();
-
-        int toGroup = cfm.comparator.isDense() ? -1 : cfm.clusteringColumns().size();
-        assert ColumnSlice.validateSlices(slices, cfm.comparator, false);
-        return new SliceQueryFilter(slices, false, slices.length, toGroup);
-    }
-
-    public boolean appliesTo(ColumnFamily current) throws InvalidRequestException
-    {
-        for (RowCondition condition : conditions.values())
-        {
-            if (!condition.appliesTo(current))
-                return false;
-        }
-        return true;
-    }
-
-    private static abstract class RowCondition
-    {
-        public final Composite rowPrefix;
-        protected final long now;
-
-        protected RowCondition(Composite rowPrefix, long now)
-        {
-            this.rowPrefix = rowPrefix;
-            this.now = now;
-        }
-
-        public abstract boolean appliesTo(ColumnFamily current) throws InvalidRequestException;
-    }
-
-    private static class NotExistCondition extends RowCondition
-    {
-        private NotExistCondition(Composite rowPrefix, long now)
-        {
-            super(rowPrefix, now);
-        }
-
-        public boolean appliesTo(ColumnFamily current)
-        {
-            if (current == null)
-                return true;
-
-            Iterator<Cell> iter = current.iterator(new ColumnSlice[]{ rowPrefix.slice()
});
-            while (iter.hasNext())
-                if (iter.next().isLive(now))
-                    return false;
-            return true;
-        }
-    }
-
-    private static class ExistCondition extends RowCondition
-    {
-        private ExistCondition(Composite rowPrefix, long now)
-        {
-            super (rowPrefix, now);
-        }
-
-        public boolean appliesTo(ColumnFamily current)
-        {
-            if (current == null)
-                return false;
-
-            Iterator<Cell> iter = current.iterator(new ColumnSlice[]{ rowPrefix.slice()
});
-            while (iter.hasNext())
-                if (iter.next().isLive(now))
-                    return true;
-            return false;
-        }
-    }
-
-    private static class ColumnsConditions extends RowCondition
-    {
-        private final Map<Pair<ColumnIdentifier, ByteBuffer>, ColumnCondition.Bound>
conditions = new HashMap<>();
-
-        private ColumnsConditions(Composite rowPrefix, long now)
-        {
-            super(rowPrefix, now);
-        }
-
-        public void addConditions(Collection<ColumnCondition> conds, QueryOptions options)
throws InvalidRequestException
-        {
-            for (ColumnCondition condition : conds)
-            {
-                // We will need the variables in appliesTo but with protocol batches, each
condition in this object can have a
-                // different list of variables.
-                ColumnCondition.Bound current = condition.bind(options);
-                ColumnCondition.Bound previous = conditions.put(Pair.create(condition.column.name,
current.getCollectionElementValue()), current);
-                // If 2 conditions are actually equal, let it slide
-                if (previous != null && !previous.equals(current))
-                    throw new InvalidRequestException("Duplicate and incompatible conditions
for column " + condition.column.name);
-            }
-        }
-
-        public boolean appliesTo(ColumnFamily current) throws InvalidRequestException
-        {
-            if (current == null)
-                return conditions.isEmpty();
-
-            for (ColumnCondition.Bound condition : conditions.values())
-                if (!condition.appliesTo(rowPrefix, current, now))
-                    return false;
-            return true;
-        }
-    }
-}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/5db108c3/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java b/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java
new file mode 100644
index 0000000..a85c1e5
--- /dev/null
+++ b/src/java/org/apache/cassandra/cql3/statements/CQL3CasRequest.java
@@ -0,0 +1,268 @@
+/*
+ * 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.statements;
+
+import java.nio.ByteBuffer;
+import java.util.*;
+
+import org.apache.cassandra.cql3.*;
+import org.apache.cassandra.config.CFMetaData;
+import org.apache.cassandra.db.*;
+import org.apache.cassandra.db.composites.Composite;
+import org.apache.cassandra.db.filter.*;
+import org.apache.cassandra.db.marshal.CompositeType;
+import org.apache.cassandra.exceptions.InvalidRequestException;
+import org.apache.cassandra.service.CASRequest;
+import org.apache.cassandra.utils.Pair;
+
+/**
+ * Processed CAS conditions and update on potentially multiple rows of the same partition.
+ */
+public class CQL3CasRequest implements CASRequest
+{
+    private final CFMetaData cfm;
+    private final ByteBuffer key;
+    private final long now;
+    private final boolean isBatch;
+
+    // We index RowCondition by the prefix of the row they applied to for 2 reasons:
+    //   1) this allows to keep things sorted to build the ColumnSlice array below
+    //   2) this allows to detect when contradictory conditions are set (not exists with
some other conditions on the same row)
+    private final SortedMap<Composite, RowCondition> conditions;
+
+    private final List<RowUpdate> updates = new ArrayList<>();
+
+    public CQL3CasRequest(CFMetaData cfm, ByteBuffer key, boolean isBatch)
+    {
+        this.cfm = cfm;
+        // When checking if conditions apply, we want to use a fixed reference time for a
whole request to check
+        // for expired cells. Note that this is unrelated to the cell timestamp.
+        this.now = System.currentTimeMillis();
+        this.key = key;
+        this.conditions = new TreeMap<>(cfm.comparator);
+        this.isBatch = isBatch;
+    }
+
+    public void addRowUpdate(Composite prefix, ModificationStatement stmt, QueryOptions options,
long timestamp)
+    {
+        updates.add(new RowUpdate(prefix, stmt, options, timestamp));
+    }
+
+    public void addNotExist(Composite prefix) throws InvalidRequestException
+    {
+        RowCondition previous = conditions.put(prefix, new NotExistCondition(prefix, now));
+        if (previous != null && !(previous instanceof NotExistCondition))
+        {
+            // these should be prevented by the parser, but it doesn't hurt to check
+            if (previous instanceof ExistCondition)
+                throw new InvalidRequestException("Cannot mix IF EXISTS and IF NOT EXISTS
conditions for the same row");
+            else
+                throw new InvalidRequestException("Cannot mix IF conditions and IF NOT EXISTS
for the same row");
+        }
+    }
+
+    public void addExist(Composite prefix) throws InvalidRequestException
+    {
+        RowCondition previous = conditions.put(prefix, new ExistCondition(prefix, now));
+        // this should be prevented by the parser, but it doesn't hurt to check
+        if (previous != null && previous instanceof NotExistCondition)
+            throw new InvalidRequestException("Cannot mix IF EXISTS and IF NOT EXISTS conditions
for the same row");
+    }
+
+    public void addConditions(Composite prefix, Collection<ColumnCondition> conds,
QueryOptions options) throws InvalidRequestException
+    {
+        RowCondition condition = conditions.get(prefix);
+        if (condition == null)
+        {
+            condition = new ColumnsConditions(prefix, now);
+            conditions.put(prefix, condition);
+        }
+        else if (!(condition instanceof ColumnsConditions))
+        {
+            throw new InvalidRequestException("Cannot mix IF conditions and IF NOT EXISTS
for the same row");
+        }
+        ((ColumnsConditions)condition).addConditions(conds, options);
+    }
+
+    public IDiskAtomFilter readFilter()
+    {
+        assert !conditions.isEmpty();
+        ColumnSlice[] slices = new ColumnSlice[conditions.size()];
+        int i = 0;
+        // We always read CQL rows entirely as on CAS failure we want to be able to distinguish
between "row exists
+        // but all values for which there were conditions are null" and "row doesn't exists",
and we can't rely on the
+        // row marker for that (see #6623)
+        for (Composite prefix : conditions.keySet())
+            slices[i++] = prefix.slice();
+
+        int toGroup = cfm.comparator.isDense() ? -1 : cfm.clusteringColumns().size();
+        assert ColumnSlice.validateSlices(slices, cfm.comparator, false);
+        return new SliceQueryFilter(slices, false, slices.length, toGroup);
+    }
+
+    public boolean appliesTo(ColumnFamily current) throws InvalidRequestException
+    {
+        for (RowCondition condition : conditions.values())
+        {
+            if (!condition.appliesTo(current))
+                return false;
+        }
+        return true;
+    }
+
+    public ColumnFamily makeUpdates(ColumnFamily current) throws InvalidRequestException
+    {
+        ColumnFamily cf = ArrayBackedSortedColumns.factory.create(cfm);
+        for (RowUpdate upd : updates)
+            upd.applyUpdates(current, cf);
+
+        if (isBatch)
+            BatchStatement.verifyBatchSize(Collections.singleton(cf));
+
+        return cf;
+    }
+
+    /**
+     * Due to some operation on lists, we can't generate the update that a given Modification
statement does before
+     * we get the values read by the initial read of Paxos. A RowUpdate thus just store the
relevant information
+     * (include the statement iself) to generate those updates. We'll have multiple RowUpdate
for a Batch, otherwise
+     * we'll have only one.
+     */
+    private class RowUpdate
+    {
+        private final Composite rowPrefix;
+        private final ModificationStatement stmt;
+        private final QueryOptions options;
+        private final long timestamp;
+
+        private RowUpdate(Composite rowPrefix, ModificationStatement stmt, QueryOptions options,
long timestamp)
+        {
+            this.rowPrefix = rowPrefix;
+            this.stmt = stmt;
+            this.options = options;
+            this.timestamp = timestamp;
+        }
+
+        public void applyUpdates(ColumnFamily current, ColumnFamily updates) throws InvalidRequestException
+        {
+            Map<ByteBuffer, CQL3Row> map = null;
+            if (stmt.requiresRead())
+            {
+                // Uses the "current" values read by Paxos for lists operation that requires
a read
+                Iterator<CQL3Row> iter = cfm.comparator.CQL3RowBuilder(cfm, now).group(current.iterator(new
ColumnSlice[]{ rowPrefix.slice() }));
+                if (iter.hasNext())
+                {
+                    map = Collections.singletonMap(key, iter.next());
+                    assert !iter.hasNext() : "We shoudn't be updating more than one CQL row
per-ModificationStatement";
+                }
+            }
+
+            UpdateParameters params = new UpdateParameters(cfm, options, timestamp, stmt.getTimeToLive(options),
map);
+            stmt.addUpdateForKey(updates, key, rowPrefix, params);
+        }
+    }
+
+    private static abstract class RowCondition
+    {
+        public final Composite rowPrefix;
+        protected final long now;
+
+        protected RowCondition(Composite rowPrefix, long now)
+        {
+            this.rowPrefix = rowPrefix;
+            this.now = now;
+        }
+
+        public abstract boolean appliesTo(ColumnFamily current) throws InvalidRequestException;
+    }
+
+    private static class NotExistCondition extends RowCondition
+    {
+        private NotExistCondition(Composite rowPrefix, long now)
+        {
+            super(rowPrefix, now);
+        }
+
+        public boolean appliesTo(ColumnFamily current)
+        {
+            if (current == null)
+                return true;
+
+            Iterator<Cell> iter = current.iterator(new ColumnSlice[]{ rowPrefix.slice()
});
+            while (iter.hasNext())
+                if (iter.next().isLive(now))
+                    return false;
+            return true;
+        }
+    }
+
+    private static class ExistCondition extends RowCondition
+    {
+        private ExistCondition(Composite rowPrefix, long now)
+        {
+            super (rowPrefix, now);
+        }
+
+        public boolean appliesTo(ColumnFamily current)
+        {
+            if (current == null)
+                return false;
+
+            Iterator<Cell> iter = current.iterator(new ColumnSlice[]{ rowPrefix.slice()
});
+            while (iter.hasNext())
+                if (iter.next().isLive(now))
+                    return true;
+            return false;
+        }
+    }
+
+    private static class ColumnsConditions extends RowCondition
+    {
+        private final Map<Pair<ColumnIdentifier, ByteBuffer>, ColumnCondition.Bound>
conditions = new HashMap<>();
+
+        private ColumnsConditions(Composite rowPrefix, long now)
+        {
+            super(rowPrefix, now);
+        }
+
+        public void addConditions(Collection<ColumnCondition> conds, QueryOptions options)
throws InvalidRequestException
+        {
+            for (ColumnCondition condition : conds)
+            {
+                // We will need the variables in appliesTo but with protocol batches, each
condition in this object can have a
+                // different list of variables.
+                ColumnCondition.Bound current = condition.bind(options);
+                ColumnCondition.Bound previous = conditions.put(Pair.create(condition.column.name,
current.getCollectionElementValue()), current);
+                // If 2 conditions are actually equal, let it slide
+                if (previous != null && !previous.equals(current))
+                    throw new InvalidRequestException("Duplicate and incompatible conditions
for column " + condition.column.name);
+            }
+        }
+
+        public boolean appliesTo(ColumnFamily current) throws InvalidRequestException
+        {
+            if (current == null)
+                return conditions.isEmpty();
+
+            for (ColumnCondition.Bound condition : conditions.values())
+                if (!condition.appliesTo(rowPrefix, current, now))
+                    return false;
+            return true;
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/5db108c3/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 fef0e94..774883d 100644
--- a/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
+++ b/src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java
@@ -147,14 +147,8 @@ public abstract class ModificationStatement implements CQLStatement,
MeasurableF
 
     public void validate(ClientState state) throws InvalidRequestException
     {
-        if (hasConditions())
-        {
-            if (attrs.isTimestampSet())
-                throw new InvalidRequestException("Cannot provide custom timestamp for conditional
updates");
-
-            if (requiresRead())
-                throw new InvalidRequestException("Operations using list indexes are not
allowed with IF conditions");
-        }
+        if (hasConditions() && attrs.isTimestampSet())
+            throw new InvalidRequestException("Cannot provide custom timestamp for conditional
updates");
 
         if (isCounter() && attrs.isTimestampSet())
             throw new InvalidRequestException("Cannot provide custom timestamp for counter
updates");
@@ -414,32 +408,20 @@ public abstract class ModificationStatement implements CQLStatement,
MeasurableF
 
     public boolean requiresRead()
     {
+        // Lists SET operation incurs a read.
         for (Operation op : columnOperations)
             if (op.requiresRead())
                 return true;
+
         return false;
     }
 
     protected Map<ByteBuffer, CQL3Row> readRequiredRows(Collection<ByteBuffer>
partitionKeys, Composite clusteringPrefix, boolean local, ConsistencyLevel cl)
     throws RequestExecutionException, RequestValidationException
     {
-        // Lists SET operation incurs a read.
-        boolean requiresRead = false;
-        for (Operation op : columnOperations)
-        {
-            if (op.requiresRead())
-            {
-                requiresRead = true;
-                break;
-            }
-        }
-
-        return requiresRead ? readRows(partitionKeys, clusteringPrefix, cfm, local, cl) :
null;
-    }
+        if (!requiresRead())
+            return null;
 
-    protected Map<ByteBuffer, CQL3Row> readRows(Collection<ByteBuffer> partitionKeys,
Composite rowPrefix, CFMetaData cfm, boolean local, ConsistencyLevel cl)
-    throws RequestExecutionException, RequestValidationException
-    {
         try
         {
             cl.validateForRead(keyspace());
@@ -449,7 +431,7 @@ public abstract class ModificationStatement implements CQLStatement, MeasurableF
             throw new InvalidRequestException(String.format("Write operation require a read
but consistency %s is not supported on reads", cl));
         }
 
-        ColumnSlice[] slices = new ColumnSlice[]{ rowPrefix.slice() };
+        ColumnSlice[] slices = new ColumnSlice[]{ clusteringPrefix.slice() };
         List<ReadCommand> commands = new ArrayList<ReadCommand>(partitionKeys.size());
         long now = System.currentTimeMillis();
         for (ByteBuffer key : partitionKeys)
@@ -527,46 +509,41 @@ public abstract class ModificationStatement implements CQLStatement,
MeasurableF
             throw new InvalidRequestException("IN on the partition key is not supported with
conditional updates");
 
         ByteBuffer key = keys.get(0);
-
         long now = options.getTimestamp(queryState);
-        CQL3CasConditions conditions = new CQL3CasConditions(cfm, now);
         Composite prefix = createClusteringPrefix(options);
-        ColumnFamily updates = ArrayBackedSortedColumns.factory.create(cfm);
-        addUpdatesAndConditions(key, prefix, updates, conditions, options, getTimestamp(now,
options));
+
+        CQL3CasRequest request = new CQL3CasRequest(cfm, key, false);
+        addConditions(prefix, request, options);
+        request.addRowUpdate(prefix, this, options, now);
 
         ColumnFamily result = StorageProxy.cas(keyspace(),
                                                columnFamily(),
                                                key,
-                                               conditions,
-                                               updates,
+                                               request,
                                                options.getSerialConsistency(),
                                                options.getConsistency());
         return new ResultMessage.Rows(buildCasResultSet(key, result, options));
     }
 
-    public void addUpdatesAndConditions(ByteBuffer key, Composite clusteringPrefix, ColumnFamily
updates, CQL3CasConditions conditions, QueryOptions options, long now)
-    throws InvalidRequestException
+    public void addConditions(Composite clusteringPrefix, CQL3CasRequest request, QueryOptions
options) throws InvalidRequestException
     {
-        UpdateParameters updParams = new UpdateParameters(cfm, options, now, getTimeToLive(options),
null);
-        addUpdateForKey(updates, key, clusteringPrefix, updParams);
-
         if (ifNotExists)
         {
             // If we use ifNotExists, if the statement applies to any non static columns,
then the condition is on the row of the non-static
             // columns and the prefix should be the clusteringPrefix. But if only static
columns are set, then the ifNotExists apply to the existence
             // of any static columns and we should use the prefix for the "static part" of
the partition.
-            conditions.addNotExist(clusteringPrefix);
+            request.addNotExist(clusteringPrefix);
         }
         else if (ifExists)
         {
-            conditions.addExist(clusteringPrefix);
+            request.addExist(clusteringPrefix);
         }
         else
         {
             if (columnConditions != null)
-                conditions.addConditions(clusteringPrefix, columnConditions, options);
+                request.addConditions(clusteringPrefix, columnConditions, options);
             if (staticConditions != null)
-                conditions.addConditions(cfm.comparator.staticPrefix(), staticConditions,
options);
+                request.addConditions(cfm.comparator.staticPrefix(), staticConditions, options);
         }
     }
 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/5db108c3/src/java/org/apache/cassandra/service/CASConditions.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/CASConditions.java b/src/java/org/apache/cassandra/service/CASConditions.java
deleted file mode 100644
index c0a2111..0000000
--- a/src/java/org/apache/cassandra/service/CASConditions.java
+++ /dev/null
@@ -1,39 +0,0 @@
-/*
- * 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.service;
-
-import org.apache.cassandra.db.ColumnFamily;
-import org.apache.cassandra.db.filter.IDiskAtomFilter;
-import org.apache.cassandra.exceptions.InvalidRequestException;
-
-/**
- * Abstract the conditions to be fulfilled by a CAS operation.
- */
-public interface CASConditions
-{
-    /**
-     * The filter to use to fetch the value to compare for the CAS.
-     */
-    public IDiskAtomFilter readFilter();
-
-    /**
-     * Returns whether the provided CF, that represents the values fetched using the
-     * readFilter(), match the CAS conditions this object stands for.
-     */
-    public boolean appliesTo(ColumnFamily current) throws InvalidRequestException;
-}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/5db108c3/src/java/org/apache/cassandra/service/CASRequest.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/CASRequest.java b/src/java/org/apache/cassandra/service/CASRequest.java
new file mode 100644
index 0000000..3d86637
--- /dev/null
+++ b/src/java/org/apache/cassandra/service/CASRequest.java
@@ -0,0 +1,45 @@
+/*
+ * 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.service;
+
+import org.apache.cassandra.db.ColumnFamily;
+import org.apache.cassandra.db.filter.IDiskAtomFilter;
+import org.apache.cassandra.exceptions.InvalidRequestException;
+
+/**
+ * Abstract the conditions and updates for a CAS operation.
+ */
+public interface CASRequest
+{
+    /**
+     * The filter to use to fetch the value to compare for the CAS.
+     */
+    public IDiskAtomFilter readFilter();
+
+    /**
+     * Returns whether the provided CF, that represents the values fetched using the
+     * readFilter(), match the CAS conditions this object stands for.
+     */
+    public boolean appliesTo(ColumnFamily current) throws InvalidRequestException;
+
+    /**
+     * The updates to perform of a CAS success. The values fetched using the readFilter()
+     * are passed as argument.
+     */
+    public ColumnFamily makeUpdates(ColumnFamily current) throws InvalidRequestException;
+}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/5db108c3/src/java/org/apache/cassandra/service/StorageProxy.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java
index 62fc0d0..1c0c482 100644
--- a/src/java/org/apache/cassandra/service/StorageProxy.java
+++ b/src/java/org/apache/cassandra/service/StorageProxy.java
@@ -190,8 +190,7 @@ public class StorageProxy implements StorageProxyMBean
      * @param keyspaceName the keyspace for the CAS
      * @param cfName the column family for the CAS
      * @param key the row key for the row to CAS
-     * @param conditions the conditions for the CAS to apply.
-     * @param updates the value to insert if {@code condtions} matches the current values.
+     * @param request the conditions for the CAS to apply as well as the update to perform
if the conditions hold.
      * @param consistencyForPaxos the consistency for the paxos prepare and propose round.
This can only be either SERIAL or LOCAL_SERIAL.
      * @param consistencyForCommit the consistency for write done during the commit phase.
This can be anything, except SERIAL or LOCAL_SERIAL.
      *
@@ -201,8 +200,7 @@ public class StorageProxy implements StorageProxyMBean
     public static ColumnFamily cas(String keyspaceName,
                                    String cfName,
                                    ByteBuffer key,
-                                   CASConditions conditions,
-                                   ColumnFamily updates,
+                                   CASRequest request,
                                    ConsistencyLevel consistencyForPaxos,
                                    ConsistencyLevel consistencyForCommit)
     throws UnavailableException, IsBootstrappingException, ReadTimeoutException, WriteTimeoutException,
InvalidRequestException
@@ -226,18 +224,19 @@ public class StorageProxy implements StorageProxyMBean
             // read the current values and check they validate the conditions
             Tracing.trace("Reading existing values for CAS precondition");
             long timestamp = System.currentTimeMillis();
-            ReadCommand readCommand = ReadCommand.create(keyspaceName, key, cfName, timestamp,
conditions.readFilter());
+            ReadCommand readCommand = ReadCommand.create(keyspaceName, key, cfName, timestamp,
request.readFilter());
             List<Row> rows = read(Arrays.asList(readCommand), consistencyForPaxos ==
ConsistencyLevel.LOCAL_SERIAL? ConsistencyLevel.LOCAL_QUORUM : ConsistencyLevel.QUORUM);
             ColumnFamily current = rows.get(0).cf;
-            if (!conditions.appliesTo(current))
+            if (!request.appliesTo(current))
             {
-                Tracing.trace("CAS precondition {} does not match current values {}", conditions,
current);
+                Tracing.trace("CAS precondition does not match current values {}", current);
                 // We should not return null as this means success
                 return current == null ? ArrayBackedSortedColumns.factory.create(metadata)
: current;
             }
 
             // finish the paxos round w/ the desired updates
             // TODO turn null updates into delete?
+            ColumnFamily updates = request.makeUpdates(current);
 
             // Apply triggers to cas updates. A consideration here is that
             // triggers emit Mutations, and so a given trigger implementation

http://git-wip-us.apache.org/repos/asf/cassandra/blob/5db108c3/src/java/org/apache/cassandra/thrift/CassandraServer.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/thrift/CassandraServer.java b/src/java/org/apache/cassandra/thrift/CassandraServer.java
index 33cd012..de0b125 100644
--- a/src/java/org/apache/cassandra/thrift/CassandraServer.java
+++ b/src/java/org/apache/cassandra/thrift/CassandraServer.java
@@ -63,7 +63,7 @@ import org.apache.cassandra.locator.DynamicEndpointSnitch;
 import org.apache.cassandra.metrics.ClientMetrics;
 import org.apache.cassandra.scheduler.IRequestScheduler;
 import org.apache.cassandra.serializers.MarshalException;
-import org.apache.cassandra.service.CASConditions;
+import org.apache.cassandra.service.CASRequest;
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.service.MigrationManager;
 import org.apache.cassandra.service.StorageProxy;
@@ -784,8 +784,7 @@ public class CassandraServer implements Cassandra.Iface
             ColumnFamily result = StorageProxy.cas(cState.getKeyspace(),
                                                    column_family,
                                                    key,
-                                                   new ThriftCASConditions(cfExpected),
-                                                   cfUpdates,
+                                                   new ThriftCASRequest(cfExpected, cfUpdates),
                                                    ThriftConversion.fromThrift(serial_consistency_level),
                                                    ThriftConversion.fromThrift(commit_consistency_level));
             return result == null
@@ -2249,13 +2248,15 @@ public class CassandraServer implements Cassandra.Iface
         });
     }
 
-    private static class ThriftCASConditions implements CASConditions
+    private static class ThriftCASRequest implements CASRequest
     {
         private final ColumnFamily expected;
+        private final ColumnFamily updates;
 
-        private ThriftCASConditions(ColumnFamily expected)
+        private ThriftCASRequest(ColumnFamily expected, ColumnFamily updates)
         {
             this.expected = expected;
+            this.updates = updates;
         }
 
         public IDiskAtomFilter readFilter()
@@ -2300,10 +2301,9 @@ public class CassandraServer implements Cassandra.Iface
             return cf != null && !cf.hasOnlyTombstones(now);
         }
 
-        @Override
-        public String toString()
+        public ColumnFamily makeUpdates(ColumnFamily current)
         {
-            return expected.toString();
+            return updates;
         }
     }
 }


Mime
View raw message