Author: slebresne
Date: Mon Aug 1 14:26:24 2011
New Revision: 1152795
URL: http://svn.apache.org/viewvc?rev=1152795&view=rev
Log:
don't throw exception on batch of counter super columns
patch by slebresne; reviewed by soverton for CASSANDRA-2949
Added:
cassandra/branches/cassandra-0.8/test/unit/org/apache/cassandra/db/CounterMutationTest.java
Modified:
cassandra/branches/cassandra-0.8/CHANGES.txt
cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/ColumnFamily.java
cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/CounterMutation.java
cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/RowMutation.java
cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/SuperColumn.java
Modified: cassandra/branches/cassandra-0.8/CHANGES.txt
URL:
http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/CHANGES.txt?rev=1152795&r1=1152794&r2=1152795&view=diff
==============================================================================
--- cassandra/branches/cassandra-0.8/CHANGES.txt (original)
+++ cassandra/branches/cassandra-0.8/CHANGES.txt Mon Aug 1 14:26:24 2011
@@ -19,6 +19,8 @@
* add counter support to Hadoop InputFormat (CASSANDRA-2981)
* fix bug where dirty commit logs were removed (and avoid keeping segment
with no post-flush activity permanently dirty) (CASSANDRA-2829)
+ * fix throwing exception with batch mutation of counter super columns
+ (CASSANDRA-2949)
0.8.2
Modified:
cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/ColumnFamily.java
URL:
http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/ColumnFamily.java?rev=1152795&r1=1152794&r2=1152795&view=diff
==============================================================================
---
cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/ColumnFamily.java
(original)
+++
cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/ColumnFamily.java
Mon Aug 1 14:26:24 2011
@@ -419,6 +419,43 @@ public class ColumnFamily implements ICo
return getColumnCount();
}
+ public void retainAll(ColumnFamily cf)
+ {
+ Iterator<IColumn> iter = iterator();
+ Iterator<IColumn> toRetain = cf.iterator();
+ IColumn current = iter.hasNext() ? iter.next() : null;
+ IColumn retain = toRetain.hasNext() ? toRetain.next() : null;
+ AbstractType comparator = getComparator();
+ while (current != null && retain != null)
+ {
+ int c = comparator.compare(current.name(), retain.name());
+ if (c == 0)
+ {
+ if (isSuper())
+ {
+ assert current instanceof SuperColumn && retain instanceof
SuperColumn;
+ ((SuperColumn)current).retainAll((SuperColumn)retain);
+ }
+ current = iter.hasNext() ? iter.next() : null;
+ retain = toRetain.hasNext() ? toRetain.next() : null;
+ }
+ else if (c < 0)
+ {
+ iter.remove();
+ current = iter.hasNext() ? iter.next() : null;
+ }
+ else // c > 0
+ {
+ retain = toRetain.hasNext() ? toRetain.next() : null;
+ }
+ }
+ while (current != null)
+ {
+ iter.remove();
+ current = iter.hasNext() ? iter.next() : null;
+ }
+ }
+
public Iterator<IColumn> iterator()
{
return columns.values().iterator();
Modified:
cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/CounterMutation.java
URL:
http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/CounterMutation.java?rev=1152795&r1=1152794&r2=1152795&view=diff
==============================================================================
---
cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/CounterMutation.java
(original)
+++
cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/CounterMutation.java
Mon Aug 1 14:26:24 2011
@@ -117,37 +117,18 @@ public class CounterMutation implements
continue;
row = mergeOldShards(readCommand.table, row);
- replicationMutation.add(row.cf);
+ ColumnFamily cf = row.cf;
+ if (cf.isSuper())
+ cf.retainAll(rowMutation.getColumnFamily(cf.metadata().cfId));
+ replicationMutation.add(cf);
}
return replicationMutation;
}
private void addReadCommandFromColumnFamily(String table, ByteBuffer key,
ColumnFamily columnFamily, List<ReadCommand> commands)
{
- // CF type: regular
- if (!columnFamily.isSuper())
- {
- QueryPath queryPath = new
QueryPath(columnFamily.metadata().cfName);
- commands.add(new SliceByNamesReadCommand(table, key, queryPath,
columnFamily.getColumnNames()));
- }
- else
- {
- // CF type: super
- for (IColumn superColumn : columnFamily.getSortedColumns())
- {
- QueryPath queryPath = new
QueryPath(columnFamily.metadata().cfName, superColumn.name());
-
- // construct set of sub-column names
- Collection<IColumn> subColumns = superColumn.getSubColumns();
- Collection<ByteBuffer> subColNames = new
HashSet<ByteBuffer>(subColumns.size());
- for (IColumn subCol : subColumns)
- {
- subColNames.add(subCol.name());
- }
-
- commands.add(new SliceByNamesReadCommand(table, key,
queryPath, subColNames));
- }
- }
+ QueryPath queryPath = new QueryPath(columnFamily.metadata().cfName);
+ commands.add(new SliceByNamesReadCommand(table, key, queryPath,
columnFamily.getColumnNames()));
}
private Row mergeOldShards(String table, Row row) throws IOException
Modified:
cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/RowMutation.java
URL:
http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/RowMutation.java?rev=1152795&r1=1152794&r2=1152795&view=diff
==============================================================================
---
cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/RowMutation.java
(original)
+++
cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/RowMutation.java
Mon Aug 1 14:26:24 2011
@@ -98,6 +98,11 @@ public class RowMutation implements IMut
return modifications_.values();
}
+ public ColumnFamily getColumnFamily(Integer cfId)
+ {
+ return modifications_.get(cfId);
+ }
+
void addHints(RowMutation rm) throws IOException
{
for (ColumnFamily cf : rm.getColumnFamilies())
Modified:
cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/SuperColumn.java
URL:
http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/SuperColumn.java?rev=1152795&r1=1152794&r2=1152795&view=diff
==============================================================================
---
cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/SuperColumn.java
(original)
+++
cassandra/branches/cassandra-0.8/src/java/org/apache/cassandra/db/SuperColumn.java
Mon Aug 1 14:26:24 2011
@@ -26,6 +26,7 @@ import java.security.MessageDigest;
import java.util.Collection;
import java.util.Comparator;
import java.util.IdentityHashMap;
+import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.atomic.AtomicInteger;
@@ -342,6 +343,38 @@ public class SuperColumn implements ICol
column.validateFields(metadata);
}
}
+
+ public void retainAll(SuperColumn sc)
+ {
+ Iterator<IColumn> iter = columns_.values().iterator();
+ Iterator<IColumn> toRetain = sc.columns_.values().iterator();
+ IColumn current = iter.hasNext() ? iter.next() : null;
+ IColumn retain = toRetain.hasNext() ? toRetain.next() : null;
+ AbstractType comparator = getComparator();
+ while (current != null && retain != null)
+ {
+ int c = comparator.compare(current.name(), retain.name());
+ if (c == 0)
+ {
+ current = iter.hasNext() ? iter.next() : null;
+ retain = toRetain.hasNext() ? toRetain.next() : null;
+ }
+ else if (c < 0)
+ {
+ iter.remove();
+ current = iter.hasNext() ? iter.next() : null;
+ }
+ else // c > 0
+ {
+ retain = toRetain.hasNext() ? toRetain.next() : null;
+ }
+ }
+ while (current != null)
+ {
+ iter.remove();
+ current = iter.hasNext() ? iter.next() : null;
+ }
+ }
}
class SuperColumnSerializer implements IColumnSerializer
Added:
cassandra/branches/cassandra-0.8/test/unit/org/apache/cassandra/db/CounterMutationTest.java
URL:
http://svn.apache.org/viewvc/cassandra/branches/cassandra-0.8/test/unit/org/apache/cassandra/db/CounterMutationTest.java?rev=1152795&view=auto
==============================================================================
---
cassandra/branches/cassandra-0.8/test/unit/org/apache/cassandra/db/CounterMutationTest.java
(added)
+++
cassandra/branches/cassandra-0.8/test/unit/org/apache/cassandra/db/CounterMutationTest.java
Mon Aug 1 14:26:24 2011
@@ -0,0 +1,66 @@
+/**
+ * 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.db;
+
+import java.io.IOException;
+
+import org.junit.Test;
+
+import org.apache.cassandra.db.filter.*;
+import org.apache.cassandra.CleanupHelper;
+import org.apache.cassandra.thrift.*;
+import org.apache.cassandra.utils.*;
+import static org.apache.cassandra.db.context.CounterContext.ContextState;
+import static org.apache.cassandra.utils.ByteBufferUtil.bytes;
+
+public class CounterMutationTest extends CleanupHelper
+{
+ @Test
+ public void testMutateSuperColumns() throws IOException
+ {
+ RowMutation rm;
+ CounterMutation cm;
+
+ rm = new RowMutation("Keyspace1", bytes("key1"));
+ rm.addCounter(new QueryPath("SuperCounter1", bytes("sc1"),
bytes("Column1")), 1);
+ rm.addCounter(new QueryPath("SuperCounter1", bytes("sc2"),
bytes("Column1")), 1);
+ cm = new CounterMutation(rm, ConsistencyLevel.ONE);
+ cm.apply();
+
+ rm = new RowMutation("Keyspace1", bytes("key1"));
+ rm.addCounter(new QueryPath("SuperCounter1", bytes("sc1"),
bytes("Column2")), 1);
+ rm.addCounter(new QueryPath("SuperCounter1", bytes("sc2"),
bytes("Column2")), 1);
+ cm = new CounterMutation(rm, ConsistencyLevel.ONE);
+ cm.apply();
+
+ RowMutation reprm = cm.makeReplicationMutation();
+ ColumnFamily cf = reprm.getColumnFamilies().iterator().next();
+
+ assert cf.getColumnCount() == 2;
+
+ IColumn sc1 = cf.getColumn(bytes("sc1"));
+ assert sc1 != null && sc1 instanceof SuperColumn;
+ assert sc1.getSubColumns().size() == 1;
+ assert sc1.getSubColumn(bytes("Column2")) != null;
+
+ IColumn sc2 = cf.getColumn(bytes("sc2"));
+ assert sc2 != null && sc2 instanceof SuperColumn;
+ assert sc2.getSubColumns().size() == 1;
+ assert sc2.getSubColumn(bytes("Column2")) != null;
+ }
+}
