You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@cassandra.apache.org by sl...@apache.org on 2011/08/09 18:34:50 UTC
svn commit: r1155426 - in /cassandra/trunk: ./
src/java/org/apache/cassandra/db/
src/java/org/apache/cassandra/db/columniterator/
src/java/org/apache/cassandra/db/filter/
src/java/org/apache/cassandra/service/
src/java/org/apache/cassandra/thrift/ test...
Author: slebresne
Date: Tue Aug 9 16:34:49 2011
New Revision: 1155426
URL: http://svn.apache.org/viewvc?rev=1155426&view=rev
Log:
Make ColumnFamily backing column map pluggable and introduce unsynchronized ArrayList backed map for reads
patch by slebresne; reviewed by jbellis for CASSANDRA-2843
Added:
cassandra/trunk/src/java/org/apache/cassandra/db/ArrayBackedSortedColumns.java
cassandra/trunk/src/java/org/apache/cassandra/db/ISortedColumns.java
cassandra/trunk/src/java/org/apache/cassandra/db/ThreadSafeSortedColumns.java
cassandra/trunk/test/unit/org/apache/cassandra/db/ArrayBackedSortedColumnsTest.java
Modified:
cassandra/trunk/CHANGES.txt
cassandra/trunk/src/java/org/apache/cassandra/db/AbstractColumnContainer.java
cassandra/trunk/src/java/org/apache/cassandra/db/ColumnFamily.java
cassandra/trunk/src/java/org/apache/cassandra/db/ColumnFamilySerializer.java
cassandra/trunk/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
cassandra/trunk/src/java/org/apache/cassandra/db/CounterColumn.java
cassandra/trunk/src/java/org/apache/cassandra/db/CounterMutation.java
cassandra/trunk/src/java/org/apache/cassandra/db/IColumnContainer.java
cassandra/trunk/src/java/org/apache/cassandra/db/ReadResponse.java
cassandra/trunk/src/java/org/apache/cassandra/db/Row.java
cassandra/trunk/src/java/org/apache/cassandra/db/RowMutation.java
cassandra/trunk/src/java/org/apache/cassandra/db/SuperColumn.java
cassandra/trunk/src/java/org/apache/cassandra/db/Table.java
cassandra/trunk/src/java/org/apache/cassandra/db/columniterator/SSTableNamesIterator.java
cassandra/trunk/src/java/org/apache/cassandra/db/filter/IFilter.java
cassandra/trunk/src/java/org/apache/cassandra/db/filter/NamesQueryFilter.java
cassandra/trunk/src/java/org/apache/cassandra/db/filter/QueryFilter.java
cassandra/trunk/src/java/org/apache/cassandra/db/filter/SliceQueryFilter.java
cassandra/trunk/src/java/org/apache/cassandra/service/RowRepairResolver.java
cassandra/trunk/src/java/org/apache/cassandra/thrift/CassandraServer.java
cassandra/trunk/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java
cassandra/trunk/test/unit/org/apache/cassandra/db/RowTest.java
cassandra/trunk/test/unit/org/apache/cassandra/service/AntiEntropyServiceTestAbstract.java
cassandra/trunk/test/unit/org/apache/cassandra/streaming/StreamingTransferTest.java
Modified: cassandra/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/cassandra/trunk/CHANGES.txt?rev=1155426&r1=1155425&r2=1155426&view=diff
==============================================================================
--- cassandra/trunk/CHANGES.txt (original)
+++ cassandra/trunk/CHANGES.txt Tue Aug 9 16:34:49 2011
@@ -27,6 +27,9 @@
* add optional compression for sstables (CASSANDRA-47)
* add scheduler JMX metrics (CASSANDRA-2962)
* add block level checksum for compressed data (CASSANDRA-1717)
+ * make column family backed column map pluggable and introduce unsynchronized
+ ArrayList backed one to speedup reads (CASSANDRA-2843)
+
0.8.4
* include files-to-be-streamed in StreamInSession.getSources (CASSANDRA-2972)
Modified: cassandra/trunk/src/java/org/apache/cassandra/db/AbstractColumnContainer.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/db/AbstractColumnContainer.java?rev=1155426&r1=1155425&r2=1155426&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/db/AbstractColumnContainer.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/db/AbstractColumnContainer.java Tue Aug 9 16:34:49 2011
@@ -41,10 +41,10 @@ public abstract class AbstractColumnCont
{
private static Logger logger = LoggerFactory.getLogger(AbstractColumnContainer.class);
- protected final ConcurrentSkipListMap<ByteBuffer, IColumn> columns;
protected final AtomicReference<DeletionInfo> deletionInfo = new AtomicReference<DeletionInfo>(new DeletionInfo());
+ protected final ISortedColumns columns;
- protected AbstractColumnContainer(ConcurrentSkipListMap<ByteBuffer, IColumn> columns)
+ protected AbstractColumnContainer(ISortedColumns columns)
{
this.columns = columns;
}
@@ -85,7 +85,7 @@ public abstract class AbstractColumnCont
public AbstractType getComparator()
{
- return (AbstractType)columns.comparator();
+ return columns.getComparator();
}
public void maybeResetDeletionTimes(int gcBefore)
@@ -107,70 +107,48 @@ public abstract class AbstractColumnCont
*/
public void addAll(AbstractColumnContainer cc)
{
- for (IColumn column : cc.getSortedColumns())
- addColumn(column);
+ columns.addAll(cc.columns);
delete(cc);
}
- /*
- * If we find an old column that has the same name
- * the ask it to resolve itself else add the new column
- */
public void addColumn(IColumn column)
{
- ByteBuffer name = column.name();
- IColumn oldColumn;
- while ((oldColumn = columns.putIfAbsent(name, column)) != null)
- {
- if (oldColumn instanceof SuperColumn)
- {
- assert column instanceof SuperColumn;
- ((SuperColumn) oldColumn).putColumn((SuperColumn)column);
- break; // Delegated to SuperColumn
- }
- else
- {
- // calculate reconciled col from old (existing) col and new col
- IColumn reconciledColumn = column.reconcile(oldColumn);
- if (columns.replace(name, oldColumn, reconciledColumn))
- break;
-
- // We failed to replace column due to a concurrent update or a concurrent removal. Keep trying.
- // (Currently, concurrent removal should not happen (only updates), but let us support that anyway.)
- }
- }
+ columns.addColumn(column);
}
- abstract protected void putColumn(SuperColumn sc);
-
public IColumn getColumn(ByteBuffer name)
{
- return columns.get(name);
+ return columns.getColumn(name);
}
- public SortedSet<ByteBuffer> getColumnNames()
+ public boolean replace(IColumn oldColumn, IColumn newColumn)
{
- return columns.keySet();
+ return columns.replace(oldColumn, newColumn);
}
- public Collection<IColumn> getSortedColumns()
+ /*
+ * Note that for some of the implementation backing the container, the
+ * return set may not have implementation for tailSet, headSet and subSet.
+ * See ColumnNamesSet in ArrayBackedSortedColumns for more details.
+ */
+ public SortedSet<ByteBuffer> getColumnNames()
{
- return columns.values();
+ return columns.getColumnNames();
}
- public Collection<IColumn> getReverseSortedColumns()
+ public Collection<IColumn> getSortedColumns()
{
- return columns.descendingMap().values();
+ return columns.getSortedColumns();
}
- public Map<ByteBuffer, IColumn> getColumnsMap()
+ public Collection<IColumn> getReverseSortedColumns()
{
- return columns;
+ return columns.getReverseSortedColumns();
}
public void remove(ByteBuffer columnName)
{
- columns.remove(columnName);
+ columns.removeColumn(columnName);
}
public void retainAll(AbstractColumnContainer container)
@@ -229,7 +207,7 @@ public abstract class AbstractColumnCont
{
int count = 0;
- for (IColumn column : columns.values())
+ for (IColumn column : columns)
{
if (column.isLive())
count++;
@@ -240,7 +218,7 @@ public abstract class AbstractColumnCont
public Iterator<IColumn> iterator()
{
- return columns.values().iterator();
+ return columns.iterator();
}
protected static class DeletionInfo
Added: cassandra/trunk/src/java/org/apache/cassandra/db/ArrayBackedSortedColumns.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/db/ArrayBackedSortedColumns.java?rev=1155426&view=auto
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/db/ArrayBackedSortedColumns.java (added)
+++ cassandra/trunk/src/java/org/apache/cassandra/db/ArrayBackedSortedColumns.java Tue Aug 9 16:34:49 2011
@@ -0,0 +1,429 @@
+/**
+ * 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.nio.ByteBuffer;
+import java.util.*;
+
+import org.apache.cassandra.db.marshal.AbstractType;
+
+/**
+ * A ISortedColumns backed by an ArrayList.
+ * This implementation is not synchronized and should only be used when
+ * thread-safety is not required. This implementation makes sense when the
+ * main operations performed are iterating over the map and adding columns
+ * (especially if insertion is in sorted order).
+ */
+public class ArrayBackedSortedColumns extends ArrayList<IColumn> implements ISortedColumns
+{
+ private final AbstractType<?> comparator;
+ private final boolean reversed;
+
+ public static final ISortedColumns.Factory FACTORY = new Factory()
+ {
+ public ISortedColumns create(AbstractType<?> comparator, boolean insertReversed)
+ {
+ return new ArrayBackedSortedColumns(comparator, insertReversed);
+ }
+
+ public ISortedColumns fromSorted(SortedMap<ByteBuffer, IColumn> sortedMap, boolean insertReversed)
+ {
+ return new ArrayBackedSortedColumns(sortedMap.values(), (AbstractType)sortedMap.comparator(), insertReversed);
+ }
+ };
+
+ public static ISortedColumns.Factory factory()
+ {
+ return FACTORY;
+ }
+
+ private ArrayBackedSortedColumns(AbstractType<?> comparator, boolean reversed)
+ {
+ super();
+ this.comparator = comparator;
+ this.reversed = reversed;
+ }
+
+ private ArrayBackedSortedColumns(Collection<IColumn> columns, AbstractType<?> comparator, boolean reversed)
+ {
+ super(columns);
+ this.comparator = comparator;
+ this.reversed = reversed;
+ }
+
+ public AbstractType<?> getComparator()
+ {
+ return comparator;
+ }
+
+ public ISortedColumns cloneMe()
+ {
+ return new ArrayBackedSortedColumns(this, comparator, reversed);
+ }
+
+ private int compare(ByteBuffer name1, ByteBuffer name2)
+ {
+ if (reversed)
+ return comparator.reverseComparator.compare(name1, name2);
+ else
+ return comparator.compare(name1, name2);
+ }
+
+ public IColumn getColumn(ByteBuffer name)
+ {
+ int pos = binarySearch(name);
+ return pos >= 0 ? get(pos) : null;
+ }
+
+ /**
+ * AddColumn throws an exception if the column added does not sort after
+ * the last column in the map.
+ * The reasoning is that this implementation can get slower if too much
+ * insertions are done in unsorted order and right now we only use it when
+ * *all* insertion (with this method) are done in sorted order. The
+ * assertion throwing is thus a protection against performance regression
+ * without knowing about (we can revisit that decision later if we have
+ * use cases where most insert are in sorted order but a few are not).
+ */
+ public void addColumn(IColumn column)
+ {
+ if (isEmpty())
+ {
+ add(column);
+ return;
+ }
+
+ // Fast path if inserting at the tail
+ int c = compare(get(size() - 1).name(), column.name());
+ // note that we want an assertion here (see addColumn javadoc), but we also want that if
+ // assertion are disabled, addColumn works correctly with unsorted input
+ assert c <= 0 : "Added column does not sort as the " + (reversed ? "first" : "last") + " column";
+
+ if (c < 0)
+ {
+ // Insert as last
+ add(column);
+ }
+ else if (c == 0)
+ {
+ // Resolve against last
+ resolveAgainst(size() - 1, column);
+ }
+ else
+ {
+ int pos = binarySearch(column.name());
+ if (pos >= 0)
+ resolveAgainst(pos, column);
+ else
+ add(-pos-1, column);
+ }
+ }
+
+ /**
+ * Resolve against element at position i.
+ * Assume that i is a valid position.
+ */
+ private void resolveAgainst(int i, IColumn column)
+ {
+ IColumn oldColumn = get(i);
+ if (oldColumn instanceof SuperColumn)
+ {
+ // Delegated to SuperColumn
+ assert column instanceof SuperColumn;
+ ((SuperColumn) oldColumn).putColumn((SuperColumn)column);
+ }
+ else
+ {
+ // calculate reconciled col from old (existing) col and new col
+ IColumn reconciledColumn = column.reconcile(oldColumn);
+ set(i, reconciledColumn);
+ }
+ }
+
+ /**
+ * Simple binary search for a given column name.
+ * The return value has the exact same meaning that the one of Collections.binarySearch().
+ * (We don't use Collections.binarySearch() directly because it would require us to create
+ * a fake IColumn (as well an IColumn comparator) to do the search, which is ugly.
+ */
+ private int binarySearch(ByteBuffer name)
+ {
+ int low = 0;
+ int mid = size();
+ int high = mid - 1;
+ int result = -1;
+ while (low <= high)
+ {
+ mid = (low + high) >> 1;
+ if ((result = -compare(get(mid).name(), name)) > 0)
+ {
+ low = mid + 1;
+ }
+ else if (result == 0)
+ {
+ return mid;
+ }
+ else
+ {
+ high = mid - 1;
+ }
+ }
+ return -mid - (result < 0 ? 1 : 2);
+ }
+
+ public void addAll(ISortedColumns cm)
+ {
+ if (cm.isEmpty())
+ return;
+
+ IColumn[] copy = toArray(new IColumn[size()]);
+ int idx = 0;
+ Iterator<IColumn> other = reversed ? cm.reverseIterator() : cm.iterator();
+ IColumn otherColumn = other.next();
+
+ clear();
+
+ while (idx < copy.length && otherColumn != null)
+ {
+ int c = compare(copy[idx].name(), otherColumn.name());
+ if (c < 0)
+ {
+ add(copy[idx]);
+ idx++;
+ }
+ else if (c > 0)
+ {
+ add(otherColumn);
+ otherColumn = other.hasNext() ? other.next() : null;
+ }
+ else // c == 0
+ {
+ add(copy[idx]);
+ resolveAgainst(size() - 1, otherColumn);
+ idx++;
+ otherColumn = other.hasNext() ? other.next() : null;
+ }
+ }
+ while (idx < copy.length)
+ {
+ add(copy[idx++]);
+ }
+ while (otherColumn != null)
+ {
+ add(otherColumn);
+ otherColumn = other.hasNext() ? other.next() : null;
+ }
+ }
+
+ public boolean replace(IColumn oldColumn, IColumn newColumn)
+ {
+ if (!oldColumn.name().equals(newColumn.name()))
+ throw new IllegalArgumentException();
+
+ int pos = binarySearch(oldColumn.name());
+ if (pos >= 0)
+ {
+ set(pos, newColumn);
+ }
+
+ return pos >= 0;
+ }
+
+ public Collection<IColumn> getSortedColumns()
+ {
+ return reversed ? new ReverseSortedCollection() : this;
+ }
+
+ public Collection<IColumn> getReverseSortedColumns()
+ {
+ // If reversed, the element are sorted reversely, so we could expect
+ // to return *this*, but *this* redefine the iterator to be in sorted
+ // order, so we need a collection that uses the super constructor
+ return reversed ? new ForwardSortedCollection() : new ReverseSortedCollection();
+ }
+
+ public void removeColumn(ByteBuffer name)
+ {
+ int pos = binarySearch(name);
+ if (pos >= 0)
+ remove(pos);
+ }
+
+ public SortedSet<ByteBuffer> getColumnNames()
+ {
+ // we could memoize the created set but it's unlikely we'll call this method a lot on the same object anyway
+ return new ColumnNamesSet();
+ }
+
+ public int getEstimatedColumnCount()
+ {
+ return size();
+ }
+
+ @Override
+ public Iterator<IColumn> iterator()
+ {
+ return reversed ? reverseInternalIterator() : super.iterator();
+ }
+
+ public Iterator<IColumn> reverseIterator()
+ {
+ return reversed ? super.iterator() : reverseInternalIterator();
+ }
+
+ private Iterator<IColumn> reverseInternalIterator()
+ {
+ final ListIterator<IColumn> iter = listIterator(size());
+ return new Iterator<IColumn>()
+ {
+ public boolean hasNext()
+ {
+ return iter.hasPrevious();
+ }
+
+ public IColumn next()
+ {
+ return iter.previous();
+ }
+
+ public void remove()
+ {
+ iter.remove();
+ }
+ };
+ }
+
+ private class ReverseSortedCollection extends AbstractCollection<IColumn>
+ {
+ public int size()
+ {
+ return ArrayBackedSortedColumns.this.size();
+ }
+
+ public Iterator<IColumn> iterator()
+ {
+ return new Iterator<IColumn>()
+ {
+ int idx = size() - 1;
+
+ public boolean hasNext()
+ {
+ return idx >= 0;
+ }
+
+ public IColumn next()
+ {
+ return ArrayBackedSortedColumns.this.get(idx--);
+ }
+
+ public void remove()
+ {
+ ArrayBackedSortedColumns.this.remove(idx--);
+ }
+ };
+ }
+ }
+
+ private class ForwardSortedCollection extends AbstractCollection<IColumn>
+ {
+ public int size()
+ {
+ return ArrayBackedSortedColumns.this.size();
+ }
+
+ public Iterator<IColumn> iterator()
+ {
+ return ArrayBackedSortedColumns.super.iterator();
+ }
+ }
+
+ private class ColumnNamesSet extends AbstractSet<ByteBuffer> implements SortedSet<ByteBuffer>
+ {
+ public int size()
+ {
+ return ArrayBackedSortedColumns.this.size();
+ }
+
+ public Iterator<ByteBuffer> iterator()
+ {
+ final Iterator<IColumn> outerIterator = ArrayBackedSortedColumns.this.iterator(); // handles reversed
+ return new Iterator<ByteBuffer>()
+ {
+ public boolean hasNext()
+ {
+ return outerIterator.hasNext();
+ }
+
+ public ByteBuffer next()
+ {
+ return outerIterator.next().name();
+ }
+
+ public void remove()
+ {
+ outerIterator.remove();
+ }
+ };
+ }
+
+ public Comparator<ByteBuffer> comparator()
+ {
+ return ArrayBackedSortedColumns.this.getComparator();
+ }
+
+ public ByteBuffer first()
+ {
+ final ArrayBackedSortedColumns outerList = ArrayBackedSortedColumns.this;
+ if (outerList.isEmpty())
+ throw new NoSuchElementException();
+ return outerList.get(outerList.reversed ? size() - 1 : 0).name();
+ }
+
+ public ByteBuffer last()
+ {
+ final ArrayBackedSortedColumns outerList = ArrayBackedSortedColumns.this;
+ if (outerList.isEmpty())
+ throw new NoSuchElementException();
+ return outerList.get(outerList.reversed ? 0 : size() - 1).name();
+ }
+
+ /*
+ * It is fairly hard to implement headSet, tailSet and subSet so that they respect their specification.
+ * Namely, the part "The returned set is backed by this set, so changes in the returned set are reflected
+ * in this set, and vice-versa". Simply keeping a lower and upper index in the backing arrayList wouldn't
+ * ensure those property. Since we do not use those function so far, we prefer returning UnsupportedOperationException
+ * for now and revisit this when and if the need arise.
+ */
+ public SortedSet<ByteBuffer> headSet(ByteBuffer fromElement)
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ // see headSet
+ public SortedSet<ByteBuffer> tailSet(ByteBuffer toElement)
+ {
+ throw new UnsupportedOperationException();
+ }
+
+ // see headSet
+ public SortedSet<ByteBuffer> subSet(ByteBuffer fromElement, ByteBuffer toElement)
+ {
+ throw new UnsupportedOperationException();
+ }
+ }
+}
Modified: cassandra/trunk/src/java/org/apache/cassandra/db/ColumnFamily.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/db/ColumnFamily.java?rev=1155426&r1=1155425&r2=1155426&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/db/ColumnFamily.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/db/ColumnFamily.java Tue Aug 9 16:34:49 2011
@@ -23,10 +23,10 @@ import static org.apache.cassandra.db.DB
import java.nio.ByteBuffer;
import java.security.MessageDigest;
import java.util.Map;
-import java.util.concurrent.ConcurrentSkipListMap;
import org.apache.cassandra.config.CFMetaData;
import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.filter.IFilter;
import org.apache.cassandra.db.filter.QueryPath;
import org.apache.cassandra.db.marshal.AbstractType;
import org.apache.cassandra.db.marshal.MarshalException;
@@ -49,6 +49,11 @@ public class ColumnFamily extends Abstra
return create(DatabaseDescriptor.getCFMetaData(cfId));
}
+ public static ColumnFamily create(Integer cfId, ISortedColumns.Factory factory)
+ {
+ return create(DatabaseDescriptor.getCFMetaData(cfId), factory);
+ }
+
public static ColumnFamily create(String tableName, String cfName)
{
return create(DatabaseDescriptor.getCFMetaData(tableName, cfName));
@@ -56,15 +61,20 @@ public class ColumnFamily extends Abstra
public static ColumnFamily create(CFMetaData cfm)
{
- return new ColumnFamily(cfm);
+ return create(cfm, ThreadSafeSortedColumns.factory());
}
- public ColumnFamily(CFMetaData cfm)
+ public static ColumnFamily create(CFMetaData cfm, ISortedColumns.Factory factory)
{
- this(cfm, new ConcurrentSkipListMap<ByteBuffer, IColumn>(cfm.comparator));
+ return create(cfm, factory, false);
}
- private ColumnFamily(CFMetaData cfm, ConcurrentSkipListMap<ByteBuffer, IColumn> map)
+ public static ColumnFamily create(CFMetaData cfm, ISortedColumns.Factory factory, boolean reversedInsertOrder)
+ {
+ return new ColumnFamily(cfm, factory.create(cfm.comparator, reversedInsertOrder));
+ }
+
+ private ColumnFamily(CFMetaData cfm, ISortedColumns map)
{
super(map);
assert cfm != null;
@@ -73,7 +83,7 @@ public class ColumnFamily extends Abstra
public ColumnFamily cloneMeShallow()
{
- ColumnFamily cf = new ColumnFamily(cfm);
+ ColumnFamily cf = ColumnFamily.create(cfm);
// since deletion info is immutable, aliasing it is fine
cf.deletionInfo.set(deletionInfo.get());
return cf;
@@ -92,7 +102,7 @@ public class ColumnFamily extends Abstra
public ColumnFamily cloneMe()
{
- ColumnFamily cf = new ColumnFamily(cfm, columns.clone());
+ ColumnFamily cf = new ColumnFamily(cfm, columns.cloneMe());
// since deletion info is immutable, aliasing it is fine
cf.deletionInfo.set(deletionInfo.get());
return cf;
@@ -189,7 +199,7 @@ public class ColumnFamily extends Abstra
public ColumnFamily diff(ColumnFamily cfComposite)
{
assert cfComposite.id().equals(id());
- ColumnFamily cfDiff = new ColumnFamily(cfm);
+ ColumnFamily cfDiff = ColumnFamily.create(cfm);
if (cfComposite.getMarkedForDeleteAt() > getMarkedForDeleteAt())
{
cfDiff.delete(cfComposite.getLocalDeletionTime(), cfComposite.getMarkedForDeleteAt());
@@ -198,12 +208,10 @@ public class ColumnFamily extends Abstra
// (don't need to worry about cfNew containing IColumns that are shadowed by
// the delete tombstone, since cfNew was generated by CF.resolve, which
// takes care of those for us.)
- Map<ByteBuffer, IColumn> columns = cfComposite.getColumnsMap();
- for (Map.Entry<ByteBuffer, IColumn> entry : columns.entrySet())
+ for (IColumn columnExternal : cfComposite)
{
- ByteBuffer cName = entry.getKey();
- IColumn columnInternal = this.columns.get(cName);
- IColumn columnExternal = entry.getValue();
+ ByteBuffer cName = columnExternal.name();
+ IColumn columnInternal = this.columns.getColumn(cName);
if (columnInternal == null)
{
cfDiff.addColumn(columnExternal);
@@ -218,7 +226,7 @@ public class ColumnFamily extends Abstra
}
}
- if (!cfDiff.getColumnsMap().isEmpty() || cfDiff.isMarkedForDelete())
+ if (!cfDiff.isEmpty() || cfDiff.isMarkedForDelete())
return cfDiff;
return null;
}
@@ -226,7 +234,7 @@ public class ColumnFamily extends Abstra
int size()
{
int size = 0;
- for (IColumn column : columns.values())
+ for (IColumn column : columns)
{
size += column.size();
}
@@ -236,7 +244,7 @@ public class ColumnFamily extends Abstra
public long maxTimestamp()
{
long maxTimestamp = Long.MIN_VALUE;
- for (IColumn column : columns.values())
+ for (IColumn column : columns)
maxTimestamp = Math.max(maxTimestamp, column.maxTimestamp());
return maxTimestamp;
}
@@ -274,7 +282,7 @@ public class ColumnFamily extends Abstra
public void updateDigest(MessageDigest digest)
{
- for (IColumn column : columns.values())
+ for (IColumn column : columns)
column.updateDigest(digest);
}
@@ -312,7 +320,7 @@ public class ColumnFamily extends Abstra
int size = intSize // local deletion time
+ longSize // client deletion time
+ intSize; // column count
- for (IColumn column : columns.values())
+ for (IColumn column : columns)
size += column.serializedSize();
return size;
}
@@ -325,14 +333,9 @@ public class ColumnFamily extends Abstra
public void validateColumnFields() throws MarshalException
{
CFMetaData metadata = metadata();
- for (IColumn column : getSortedColumns())
+ for (IColumn column : this)
{
column.validateFields(metadata);
}
}
-
- protected void putColumn(SuperColumn sc)
- {
- throw new UnsupportedOperationException("Unsupported operation for a column family");
- }
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/db/ColumnFamilySerializer.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/db/ColumnFamilySerializer.java?rev=1155426&r1=1155425&r2=1155426&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/db/ColumnFamilySerializer.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/db/ColumnFamilySerializer.java Tue Aug 9 16:34:49 2011
@@ -110,10 +110,10 @@ public class ColumnFamilySerializer impl
public ColumnFamily deserialize(DataInput dis) throws IOException
{
- return deserialize(dis, false, false);
+ return deserialize(dis, false, false, ThreadSafeSortedColumns.FACTORY);
}
- public ColumnFamily deserialize(DataInput dis, boolean intern, boolean fromRemote) throws IOException
+ public ColumnFamily deserialize(DataInput dis, boolean intern, boolean fromRemote, ISortedColumns.Factory factory) throws IOException
{
if (!dis.readBoolean())
return null;
@@ -122,7 +122,7 @@ public class ColumnFamilySerializer impl
int cfId = dis.readInt();
if (CFMetaData.getCF(cfId) == null)
throw new UnserializableColumnFamilyException("Couldn't find cfId=" + cfId, cfId);
- ColumnFamily cf = ColumnFamily.create(cfId);
+ ColumnFamily cf = ColumnFamily.create(cfId, factory);
deserializeFromSSTableNoColumns(cf, dis);
deserializeColumns(dis, cf, intern, fromRemote);
return cf;
Modified: cassandra/trunk/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/db/ColumnFamilyStore.java?rev=1155426&r1=1155425&r2=1155426&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/db/ColumnFamilyStore.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/db/ColumnFamilyStore.java Tue Aug 9 16:34:49 2011
@@ -854,10 +854,9 @@ public class ColumnFamilyStore implement
private static void removeDeletedStandard(ColumnFamily cf, int gcBefore)
{
- for (Map.Entry<ByteBuffer, IColumn> entry : cf.getColumnsMap().entrySet())
+ for (IColumn c : cf)
{
- ByteBuffer cname = entry.getKey();
- IColumn c = entry.getValue();
+ ByteBuffer cname = c.name();
// remove columns if
// (a) the column itself is tombstoned or
// (b) the CF is tombstoned and the column is not newer than it
@@ -874,9 +873,10 @@ public class ColumnFamilyStore implement
// TODO assume deletion means "most are deleted?" and add to clone, instead of remove from original?
// this could be improved by having compaction, or possibly even removeDeleted, r/m the tombstone
// once gcBefore has passed, so if new stuff is added in it doesn't used the wrong algorithm forever
- for (Map.Entry<ByteBuffer, IColumn> entry : cf.getColumnsMap().entrySet())
+ Iterator<IColumn> iter = cf.iterator();
+ while (iter.hasNext())
{
- SuperColumn c = (SuperColumn) entry.getValue();
+ SuperColumn c = (SuperColumn)iter.next();
long minTimestamp = Math.max(c.getMarkedForDeleteAt(), cf.getMarkedForDeleteAt());
for (IColumn subColumn : c.getSubColumns())
{
@@ -891,7 +891,7 @@ public class ColumnFamilyStore implement
}
if (c.getSubColumns().isEmpty() && c.getLocalDeletionTime() <= gcBefore)
{
- cf.remove(c.name());
+ iter.remove();
}
else
{
@@ -1144,7 +1144,12 @@ public class ColumnFamilyStore implement
*/
public ColumnFamily getColumnFamily(QueryFilter filter)
{
- return getColumnFamily(filter, gcBefore());
+ return getColumnFamily(filter, gcBefore(), ThreadSafeSortedColumns.factory());
+ }
+
+ public ColumnFamily getColumnFamily(QueryFilter filter, ISortedColumns.Factory factory)
+ {
+ return getColumnFamily(filter, gcBefore(), factory);
}
public int gcBefore()
@@ -1157,7 +1162,8 @@ public class ColumnFamilyStore implement
ColumnFamily cached;
if ((cached = rowCache.get(key)) == null)
{
- cached = getTopLevelColumns(QueryFilter.getIdentityFilter(key, new QueryPath(columnFamily)), Integer.MIN_VALUE);
+ // We force ThreadSafeSortedColumns because cached row will be accessed concurrently
+ cached = getTopLevelColumns(QueryFilter.getIdentityFilter(key, new QueryPath(columnFamily)), Integer.MIN_VALUE, ThreadSafeSortedColumns.factory());
if (cached == null)
return null;
@@ -1165,7 +1171,7 @@ public class ColumnFamilyStore implement
{
// make a deep copy of column data so we don't keep references to direct buffers, which
// would prevent munmap post-compaction.
- for (IColumn column : cached.getSortedColumns())
+ for (IColumn column : cached)
{
cached.remove(column.name());
cached.addColumn(column.localCopy(this));
@@ -1178,7 +1184,7 @@ public class ColumnFamilyStore implement
return cached;
}
- private ColumnFamily getColumnFamily(QueryFilter filter, int gcBefore)
+ private ColumnFamily getColumnFamily(QueryFilter filter, int gcBefore, ISortedColumns.Factory factory)
{
assert columnFamily.equals(filter.getColumnFamilyName()) : filter.getColumnFamilyName();
@@ -1187,7 +1193,7 @@ public class ColumnFamilyStore implement
{
if (rowCache.getCapacity() == 0)
{
- ColumnFamily cf = getTopLevelColumns(filter, gcBefore);
+ ColumnFamily cf = getTopLevelColumns(filter, gcBefore, factory);
if (cf == null)
return null;
@@ -1297,11 +1303,11 @@ public class ColumnFamilyStore implement
return markCurrentViewReferenced().sstables;
}
- private ColumnFamily getTopLevelColumns(QueryFilter filter, int gcBefore)
+ private ColumnFamily getTopLevelColumns(QueryFilter filter, int gcBefore, ISortedColumns.Factory factory)
{
// we are querying top-level columns, do a merging fetch with indexes.
List<IColumnIterator> iterators = new ArrayList<IColumnIterator>();
- final ColumnFamily returnCF = ColumnFamily.create(metadata);
+ final ColumnFamily returnCF = ColumnFamily.create(metadata, factory, filter.filter.isReversed());
DataTracker.View currentView = markCurrentViewReferenced();
try
{
@@ -1552,7 +1558,7 @@ public class ColumnFamilyStore implement
ByteBuffer dataKey = null;
int n = 0;
- for (IColumn column : indexRow.getSortedColumns())
+ for (IColumn column : indexRow)
{
if (column.isMarkedForDelete())
continue;
Modified: cassandra/trunk/src/java/org/apache/cassandra/db/CounterColumn.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/db/CounterColumn.java?rev=1155426&r1=1155425&r2=1155426&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/db/CounterColumn.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/db/CounterColumn.java Tue Aug 9 16:34:49 2011
@@ -251,35 +251,27 @@ public class CounterColumn extends Colum
{
if (!cf.isSuper())
{
- for (Map.Entry<ByteBuffer, IColumn> entry : cf.getColumnsMap().entrySet())
+ for (IColumn c : cf)
{
- ByteBuffer cname = entry.getKey();
- IColumn c = entry.getValue();
if (!(c instanceof CounterColumn))
continue;
CounterColumn cleaned = ((CounterColumn) c).removeOldShards(gcBefore);
if (cleaned != c)
- {
- cf.remove(cname);
- cf.addColumn(cleaned);
- }
+ cf.replace(c, cleaned);
}
}
else
{
- for (Map.Entry<ByteBuffer, IColumn> entry : cf.getColumnsMap().entrySet())
+ for (IColumn col : cf)
{
- SuperColumn c = (SuperColumn) entry.getValue();
+ SuperColumn c = (SuperColumn)col;
for (IColumn subColumn : c.getSubColumns())
{
if (!(subColumn instanceof CounterColumn))
continue;
CounterColumn cleaned = ((CounterColumn) subColumn).removeOldShards(gcBefore);
if (cleaned != subColumn)
- {
- c.remove(subColumn.name());
- c.addColumn(cleaned);
- }
+ c.replace(subColumn, cleaned);
}
}
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/db/CounterMutation.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/db/CounterMutation.java?rev=1155426&r1=1155425&r2=1155426&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/db/CounterMutation.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/db/CounterMutation.java Tue Aug 9 16:34:49 2011
@@ -157,7 +157,7 @@ public class CounterMutation implements
// CF type: regular
if (!cf.isSuper())
{
- for (IColumn column : cf.getSortedColumns())
+ for (IColumn column : cf)
{
if (!(column instanceof CounterColumn))
continue;
@@ -172,7 +172,7 @@ public class CounterMutation implements
}
else // CF type: super
{
- for (IColumn superColumn : cf.getSortedColumns())
+ for (IColumn superColumn : cf)
{
IColumn mergerSuper = null;
for (IColumn column : superColumn.getSubColumns())
@@ -225,7 +225,7 @@ public class CounterMutation implements
{
ColumnFamily cf = cf_.cloneMeShallow();
ColumnFamilyStore cfs = table.getColumnFamilyStore(cf.id());
- for (IColumn column : cf_.getColumnsMap().values())
+ for (IColumn column : cf_)
{
cf.addColumn(column.localCopy(cfs));
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/db/IColumnContainer.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/db/IColumnContainer.java?rev=1155426&r1=1155425&r2=1155426&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/db/IColumnContainer.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/db/IColumnContainer.java Tue Aug 9 16:34:49 2011
@@ -30,6 +30,16 @@ public interface IColumnContainer
public void addColumn(IColumn column);
public void remove(ByteBuffer columnName);
+ /**
+ * Replace oldColumn if represent by newColumn.
+ * Returns true if oldColumn was present (and thus replaced)
+ * oldColumn and newColumn should have the same name.
+ * !NOTE! This should only be used if you know this is what you need. To
+ * add a column such that it use the usual column resolution rules in a
+ * thread safe manner, use addColumn.
+ */
+ public boolean replace(IColumn oldColumn, IColumn newColumn);
+
public boolean isMarkedForDelete();
public long getMarkedForDeleteAt();
public int getLocalDeletionTime();
Added: cassandra/trunk/src/java/org/apache/cassandra/db/ISortedColumns.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/db/ISortedColumns.java?rev=1155426&view=auto
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/db/ISortedColumns.java (added)
+++ cassandra/trunk/src/java/org/apache/cassandra/db/ISortedColumns.java Tue Aug 9 16:34:49 2011
@@ -0,0 +1,141 @@
+/**
+ * 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.nio.ByteBuffer;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.SortedMap;
+import java.util.SortedSet;
+
+import org.apache.cassandra.db.marshal.AbstractType;
+import org.apache.cassandra.io.util.IIterableColumns;
+
+/**
+ * A sorted map of columns.
+ * This represents the backing map of a colum family.
+ *
+ * Whether the implementation is thread safe or not is left to the
+ * implementing classes.
+ */
+public interface ISortedColumns extends IIterableColumns
+{
+ /**
+ * Shallow cloning of the column map.
+ */
+ public ISortedColumns cloneMe();
+
+ /**
+ * Adds a column to this column map.
+ * If a column with the same name is already present in the map, it will
+ * be replaced by the newly added column.
+ */
+ public void addColumn(IColumn column);
+
+ /**
+ * Adds all the columns of a given column map to this column map.
+ * This is equivalent to:
+ * <code>
+ * for (Column c : cm)
+ * add(c);
+ * </code>
+ * but is potentially faster.
+ */
+ public void addAll(ISortedColumns cm);
+
+ /**
+ * Replace oldColumn if present by newColumn.
+ * Returns true if oldColumn was present and thus replaced.
+ * oldColumn and newColumn should have the same name.
+ */
+ public boolean replace(IColumn oldColumn, IColumn newColumn);
+
+ /**
+ * Remove if present a column by name.
+ */
+ public void removeColumn(ByteBuffer name);
+
+ /**
+ * Clear this column map, removing all columns.
+ */
+ public void clear();
+
+ /**
+ * Get a column given its name, returning null if the column is not
+ * present.
+ */
+ public IColumn getColumn(ByteBuffer name);
+
+ /**
+ * Returns a set with the names of columns in this column map.
+ * The resulting set is sorted and the order is the one of the columns in
+ * this column map.
+ */
+ public SortedSet<ByteBuffer> getColumnNames();
+
+ /**
+ * Returns the columns of this column map as a collection.
+ * The columns in the returned collection should be sorted as the columns
+ * in this map.
+ */
+ public Collection<IColumn> getSortedColumns();
+
+ /**
+ * Returns the columns of this column map as a collection.
+ * The columns in the returned collection should be sorted in reverse
+ * order of the columns in this map.
+ */
+ public Collection<IColumn> getReverseSortedColumns();
+
+ /**
+ * Returns the number of columns in this map.
+ */
+ public int size();
+
+ /**
+ * Returns true if this map is empty, false otherwise.
+ */
+ public boolean isEmpty();
+
+ /**
+ * Returns an iterator that iterates over the columns of this map in
+ * reverse order.
+ */
+ public Iterator<IColumn> reverseIterator();
+
+ public interface Factory
+ {
+ /**
+ * Returns a (initially empty) column map whose columns are sorted
+ * according to the provided comparator.
+ * The {@code insertReversed} flag is an hint on how we expect insertion to be perfomed,
+ * either in sorted or reverse sorted order. This is used by ArrayBackedSortedColumns to
+ * allow optimizing for both forward and reversed slices. This does not matter for ThreadSafeSortedColumns.
+ * Note that this is only an hint on how we expect to do insertion, this does not change the map sorting.
+ */
+ public ISortedColumns create(AbstractType<?> comparator, boolean insertReversed);
+
+ /**
+ * Returns a column map whose columns are sorted according to the comparator of the provided sorted
+ * map (which thus, is assumed to _not_ be sorted by natural order) and that initially contains the
+ * columns in the provided sorted map.
+ * See {@code create} for the description of {@code insertReversed}
+ */
+ public ISortedColumns fromSorted(SortedMap<ByteBuffer, IColumn> sm, boolean insertReversed);
+ }
+}
Modified: cassandra/trunk/src/java/org/apache/cassandra/db/ReadResponse.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/db/ReadResponse.java?rev=1155426&r1=1155425&r2=1155426&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/db/ReadResponse.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/db/ReadResponse.java Tue Aug 9 16:34:49 2011
@@ -110,7 +110,7 @@ class ReadResponseSerializer implements
if (!isDigest)
{
// This is coming from a remote host
- row = Row.serializer().deserialize(dis, version, true);
+ row = Row.serializer().deserialize(dis, version, true, ArrayBackedSortedColumns.FACTORY);
}
return isDigest ? new ReadResponse(ByteBuffer.wrap(digest)) : new ReadResponse(row);
Modified: cassandra/trunk/src/java/org/apache/cassandra/db/Row.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/db/Row.java?rev=1155426&r1=1155425&r2=1155426&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/db/Row.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/db/Row.java Tue Aug 9 16:34:49 2011
@@ -63,15 +63,15 @@ public class Row
ColumnFamily.serializer().serialize(row.cf, dos);
}
- public Row deserialize(DataInputStream dis, int version, boolean fromRemote) throws IOException
+ public Row deserialize(DataInputStream dis, int version, boolean fromRemote, ISortedColumns.Factory factory) throws IOException
{
return new Row(StorageService.getPartitioner().decorateKey(ByteBufferUtil.readWithShortLength(dis)),
- ColumnFamily.serializer().deserialize(dis, false, fromRemote));
+ ColumnFamily.serializer().deserialize(dis, false, fromRemote, factory));
}
public Row deserialize(DataInputStream dis, int version) throws IOException
{
- return deserialize(dis, version, false);
+ return deserialize(dis, version, false, ThreadSafeSortedColumns.FACTORY);
}
}
-}
\ No newline at end of file
+}
Modified: cassandra/trunk/src/java/org/apache/cassandra/db/RowMutation.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/db/RowMutation.java?rev=1155426&r1=1155425&r2=1155426&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/db/RowMutation.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/db/RowMutation.java Tue Aug 9 16:34:49 2011
@@ -368,8 +368,8 @@ public class RowMutation implements IMut
{
ColumnFamily cf = entry.getValue().cloneMeShallow();
ColumnFamilyStore cfs = table.getColumnFamilyStore(cf.id());
- for (Map.Entry<ByteBuffer, IColumn> ce : entry.getValue().getColumnsMap().entrySet())
- cf.addColumn(ce.getValue().localCopy(cfs));
+ for (IColumn col : entry.getValue())
+ cf.addColumn(col.localCopy(cfs));
rm.modifications_.put(entry.getKey(), cf);
}
@@ -405,7 +405,7 @@ public class RowMutation implements IMut
for (int i = 0; i < size; ++i)
{
Integer cfid = Integer.valueOf(dis.readInt());
- ColumnFamily cf = ColumnFamily.serializer().deserialize(dis, true, fromRemote);
+ ColumnFamily cf = ColumnFamily.serializer().deserialize(dis, true, fromRemote, ThreadSafeSortedColumns.FACTORY);
modifications.put(cfid, cf);
}
return new RowMutation(table, key, modifications);
Modified: cassandra/trunk/src/java/org/apache/cassandra/db/SuperColumn.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/db/SuperColumn.java?rev=1155426&r1=1155425&r2=1155426&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/db/SuperColumn.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/db/SuperColumn.java Tue Aug 9 16:34:49 2011
@@ -57,10 +57,10 @@ public class SuperColumn extends Abstrac
public SuperColumn(ByteBuffer name, AbstractType comparator)
{
- this(name, new ConcurrentSkipListMap<ByteBuffer, IColumn>(comparator));
+ this(name, ThreadSafeSortedColumns.factory().create(comparator, false));
}
- SuperColumn(ByteBuffer name, ConcurrentSkipListMap<ByteBuffer, IColumn> columns)
+ SuperColumn(ByteBuffer name, ISortedColumns columns)
{
super(columns);
assert name != null;
@@ -78,7 +78,7 @@ public class SuperColumn extends Abstrac
public IColumn cloneMe()
{
- SuperColumn sc = new SuperColumn(name, new ConcurrentSkipListMap<ByteBuffer, IColumn>(columns));
+ SuperColumn sc = new SuperColumn(name, columns.cloneMe());
// since deletion info is immutable, aliasing it is fine
sc.deletionInfo.set(deletionInfo.get());
return sc;
@@ -96,7 +96,7 @@ public class SuperColumn extends Abstrac
public IColumn getSubColumn(ByteBuffer columnName)
{
- IColumn column = columns.get(columnName);
+ IColumn column = columns.getColumn(columnName);
assert column == null || column instanceof Column;
return column;
}
@@ -170,7 +170,7 @@ public class SuperColumn extends Abstrac
* Go through each sub column if it exists then as it to resolve itself
* if the column does not exist then create it.
*/
- protected void putColumn(SuperColumn column)
+ void putColumn(SuperColumn column)
{
for (IColumn subColumn : column.getSubColumns())
{
@@ -192,7 +192,7 @@ public class SuperColumn extends Abstrac
// takes care of those for us.)
for (IColumn subColumn : columnNew.getSubColumns())
{
- IColumn columnInternal = columns.get(subColumn.name());
+ IColumn columnInternal = columns.getColumn(subColumn.name());
if(columnInternal == null )
{
columnDiff.addColumn(subColumn);
@@ -269,9 +269,9 @@ public class SuperColumn extends Abstrac
// certainly just going to pollute our interning map with unique, dynamic values
SuperColumn sc = (SuperColumn)shallowCopy();
- for(Map.Entry<ByteBuffer, IColumn> c : columns.entrySet())
+ for(IColumn c : columns)
{
- sc.addColumn(c.getValue().localCopy(cfs));
+ sc.addColumn(c.localCopy(cfs));
}
return sc;
@@ -362,7 +362,7 @@ class SuperColumnSerializer implements I
int size = dis.readInt();
ColumnSerializer serializer = Column.serializer();
ColumnSortedMap preSortedMap = new ColumnSortedMap(comparator, serializer, dis, interner, size, fromRemote, expireBefore);
- SuperColumn superColumn = new SuperColumn(name, new ConcurrentSkipListMap<ByteBuffer,IColumn>(preSortedMap));
+ SuperColumn superColumn = new SuperColumn(name, ThreadSafeSortedColumns.factory().fromSorted(preSortedMap, false));
if (localDeleteTime != Integer.MIN_VALUE && localDeleteTime <= 0)
{
throw new IOException("Invalid localDeleteTime read: " + localDeleteTime);
Modified: cassandra/trunk/src/java/org/apache/cassandra/db/Table.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/db/Table.java?rev=1155426&r1=1155425&r2=1155426&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/db/Table.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/db/Table.java Tue Aug 9 16:34:49 2011
@@ -374,7 +374,7 @@ public class Table
public Row getRow(QueryFilter filter) throws IOException
{
ColumnFamilyStore cfStore = getColumnFamilyStore(filter.getColumnFamilyName());
- ColumnFamily columnFamily = cfStore.getColumnFamily(filter);
+ ColumnFamily columnFamily = cfStore.getColumnFamily(filter, ArrayBackedSortedColumns.factory());
return new Row(filter.key, columnFamily);
}
@@ -560,10 +560,9 @@ public class Table
if (oldIndexedColumns != null)
{
int localDeletionTime = (int) (System.currentTimeMillis() / 1000);
- for (Map.Entry<ByteBuffer, IColumn> entry : oldIndexedColumns.getColumnsMap().entrySet())
+ for (IColumn column : oldIndexedColumns)
{
- ByteBuffer columnName = entry.getKey();
- IColumn column = entry.getValue();
+ ByteBuffer columnName = column.name();
if (column.isMarkedForDelete())
continue;
DecoratedKey<LocalToken> valueKey = cfs.getIndexKeyFor(columnName, column.value());
Added: cassandra/trunk/src/java/org/apache/cassandra/db/ThreadSafeSortedColumns.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/db/ThreadSafeSortedColumns.java?rev=1155426&view=auto
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/db/ThreadSafeSortedColumns.java (added)
+++ cassandra/trunk/src/java/org/apache/cassandra/db/ThreadSafeSortedColumns.java Tue Aug 9 16:34:49 2011
@@ -0,0 +1,156 @@
+/**
+ * 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.nio.ByteBuffer;
+import java.util.Comparator;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.SortedMap;
+import java.util.SortedSet;
+import java.util.concurrent.ConcurrentSkipListMap;
+
+import org.apache.cassandra.db.marshal.AbstractType;
+
+public class ThreadSafeSortedColumns extends ConcurrentSkipListMap<ByteBuffer, IColumn> implements ISortedColumns
+{
+ public static final ISortedColumns.Factory FACTORY = new Factory()
+ {
+ public ISortedColumns create(AbstractType<?> comparator, boolean insertReversed)
+ {
+ return new ThreadSafeSortedColumns(comparator);
+ }
+
+ public ISortedColumns fromSorted(SortedMap<ByteBuffer, IColumn> sortedMap, boolean insertReversed)
+ {
+ return new ThreadSafeSortedColumns(sortedMap);
+ }
+ };
+
+ public static ISortedColumns.Factory factory()
+ {
+ return FACTORY;
+ }
+
+ public AbstractType<?> getComparator()
+ {
+ return (AbstractType)comparator();
+ }
+
+ private ThreadSafeSortedColumns(AbstractType<?> comparator)
+ {
+ super(comparator);
+ }
+
+ private ThreadSafeSortedColumns(SortedMap<ByteBuffer, IColumn> columns)
+ {
+ super(columns);
+ }
+
+ public ISortedColumns cloneMe()
+ {
+ return new ThreadSafeSortedColumns(this);
+ }
+
+ /*
+ * If we find an old column that has the same name
+ * the ask it to resolve itself else add the new column
+ */
+ public void addColumn(IColumn column)
+ {
+ ByteBuffer name = column.name();
+ IColumn oldColumn;
+ while ((oldColumn = putIfAbsent(name, column)) != null)
+ {
+ if (oldColumn instanceof SuperColumn)
+ {
+ assert column instanceof SuperColumn;
+ ((SuperColumn) oldColumn).putColumn((SuperColumn)column);
+ break; // Delegated to SuperColumn
+ }
+ else
+ {
+ // calculate reconciled col from old (existing) col and new col
+ IColumn reconciledColumn = column.reconcile(oldColumn);
+ if (replace(name, oldColumn, reconciledColumn))
+ break;
+
+ // We failed to replace column due to a concurrent update or a concurrent removal. Keep trying.
+ // (Currently, concurrent removal should not happen (only updates), but let us support that anyway.)
+ }
+ }
+ }
+
+ /**
+ * We need to go through each column in the column container and resolve it before adding
+ */
+ public void addAll(ISortedColumns cm)
+ {
+ for (IColumn column : cm.getSortedColumns())
+ addColumn(column);
+ }
+
+ public boolean replace(IColumn oldColumn, IColumn newColumn)
+ {
+ if (!oldColumn.name().equals(newColumn.name()))
+ throw new IllegalArgumentException();
+
+ return replace(oldColumn.name(), oldColumn, newColumn);
+ }
+
+ public IColumn getColumn(ByteBuffer name)
+ {
+ return get(name);
+ }
+
+ public void removeColumn(ByteBuffer name)
+ {
+ remove(name);
+ }
+
+ public Collection<IColumn> getSortedColumns()
+ {
+ return values();
+ }
+
+ public Collection<IColumn> getReverseSortedColumns()
+ {
+ return descendingMap().values();
+ }
+
+ public SortedSet<ByteBuffer> getColumnNames()
+ {
+ return keySet();
+ }
+
+ public int getEstimatedColumnCount()
+ {
+ return size();
+ }
+
+ public Iterator<IColumn> iterator()
+ {
+ return values().iterator();
+ }
+
+ public Iterator<IColumn> reverseIterator()
+ {
+ return getReverseSortedColumns().iterator();
+ }
+}
Modified: cassandra/trunk/src/java/org/apache/cassandra/db/columniterator/SSTableNamesIterator.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/db/columniterator/SSTableNamesIterator.java?rev=1155426&r1=1155425&r2=1155426&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/db/columniterator/SSTableNamesIterator.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/db/columniterator/SSTableNamesIterator.java Tue Aug 9 16:34:49 2011
@@ -130,7 +130,7 @@ public class SSTableNamesIterator extend
readIndexedColumns(sstable.metadata, file, columns, filteredColumnNames, indexList);
// create an iterator view of the columns we read
- iter = cf.getSortedColumns().iterator();
+ iter = cf.iterator();
}
private void readSimpleColumns(FileDataInput file, SortedSet<ByteBuffer> columnNames, List<ByteBuffer> filteredColumnNames) throws IOException
Modified: cassandra/trunk/src/java/org/apache/cassandra/db/filter/IFilter.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/db/filter/IFilter.java?rev=1155426&r1=1155425&r2=1155426&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/db/filter/IFilter.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/db/filter/IFilter.java Tue Aug 9 16:34:49 2011
@@ -74,4 +74,6 @@ public interface IFilter
public abstract SuperColumn filterSuperColumn(SuperColumn superColumn, int gcBefore);
public Comparator<IColumn> getColumnComparator(AbstractType comparator);
+
+ public boolean isReversed();
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/db/filter/NamesQueryFilter.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/db/filter/NamesQueryFilter.java?rev=1155426&r1=1155425&r2=1155426&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/db/filter/NamesQueryFilter.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/db/filter/NamesQueryFilter.java Tue Aug 9 16:34:49 2011
@@ -99,4 +99,9 @@ public class NamesQueryFilter implements
"columns=" + StringUtils.join(columns, ",") +
')';
}
+
+ public boolean isReversed()
+ {
+ return false;
+ }
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/db/filter/QueryFilter.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/db/filter/QueryFilter.java?rev=1155426&r1=1155425&r2=1155426&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/db/filter/QueryFilter.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/db/filter/QueryFilter.java Tue Aug 9 16:34:49 2011
@@ -121,7 +121,7 @@ public class QueryFilter
protected IColumn getReduced()
{
- IColumn c = curCF.getSortedColumns().iterator().next();
+ IColumn c = curCF.iterator().next();
if (superFilter != null)
{
// filterSuperColumn only looks at immediate parent (the supercolumn) when determining if a subcolumn
Modified: cassandra/trunk/src/java/org/apache/cassandra/db/filter/SliceQueryFilter.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/db/filter/SliceQueryFilter.java?rev=1155426&r1=1155425&r2=1155426&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/db/filter/SliceQueryFilter.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/db/filter/SliceQueryFilter.java Tue Aug 9 16:34:49 2011
@@ -150,4 +150,9 @@ public class SliceQueryFilter implements
", reversed=" + reversed +
", count=" + count + "]";
}
+
+ public boolean isReversed()
+ {
+ return reversed;
+ }
}
Modified: cassandra/trunk/src/java/org/apache/cassandra/service/RowRepairResolver.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/service/RowRepairResolver.java?rev=1155426&r1=1155425&r2=1155426&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/service/RowRepairResolver.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/service/RowRepairResolver.java Tue Aug 9 16:34:49 2011
@@ -157,7 +157,7 @@ public class RowRepairResolver extends A
{
if (version == null)
continue;
- iters.add(FBUtilities.closeableIterator(version.getColumnsMap().values().iterator()));
+ iters.add(FBUtilities.closeableIterator(version.iterator()));
}
filter.collateColumns(resolved, iters, resolved.metadata().comparator, Integer.MIN_VALUE);
return ColumnFamilyStore.removeDeleted(resolved, Integer.MIN_VALUE);
Modified: cassandra/trunk/src/java/org/apache/cassandra/thrift/CassandraServer.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/src/java/org/apache/cassandra/thrift/CassandraServer.java?rev=1155426&r1=1155425&r2=1155426&view=diff
==============================================================================
--- cassandra/trunk/src/java/org/apache/cassandra/thrift/CassandraServer.java (original)
+++ cassandra/trunk/src/java/org/apache/cassandra/thrift/CassandraServer.java Tue Aug 9 16:34:49 2011
@@ -294,11 +294,11 @@ public class CassandraServer implements
private List<ColumnOrSuperColumn> thriftifyColumnFamily(ColumnFamily cf, boolean subcolumnsOnly, boolean reverseOrder)
{
- if (cf == null || cf.getColumnsMap().size() == 0)
+ if (cf == null || cf.isEmpty())
return EMPTY_COLUMNS;
if (subcolumnsOnly)
{
- IColumn column = cf.getColumnsMap().values().iterator().next();
+ IColumn column = cf.iterator().next();
Collection<IColumn> subcolumns = column.getSubColumns();
if (subcolumns == null || subcolumns.isEmpty())
return EMPTY_COLUMNS;
Added: cassandra/trunk/test/unit/org/apache/cassandra/db/ArrayBackedSortedColumnsTest.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/test/unit/org/apache/cassandra/db/ArrayBackedSortedColumnsTest.java?rev=1155426&view=auto
==============================================================================
--- cassandra/trunk/test/unit/org/apache/cassandra/db/ArrayBackedSortedColumnsTest.java (added)
+++ cassandra/trunk/test/unit/org/apache/cassandra/db/ArrayBackedSortedColumnsTest.java Tue Aug 9 16:34:49 2011
@@ -0,0 +1,130 @@
+package org.apache.cassandra.db;
+
+import java.io.ByteArrayInputStream;
+import java.io.DataInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.*;
+
+import org.apache.cassandra.SchemaLoader;
+import org.junit.Test;
+
+import org.apache.cassandra.io.util.DataOutputBuffer;
+import org.apache.cassandra.db.filter.QueryPath;
+import static org.apache.cassandra.Util.column;
+import static org.junit.Assert.*;
+
+import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.db.marshal.BytesType;
+
+public class ArrayBackedSortedColumnsTest
+{
+ @Test
+ public void testAdd()
+ {
+ testAddInternal(false);
+ testAddInternal(true);
+ }
+
+ private void testAddInternal(boolean reversed)
+ {
+ ISortedColumns map = ArrayBackedSortedColumns.FACTORY.create(BytesType.instance, reversed);
+ int[] values = new int[]{ 1, 2, 2, 3 };
+
+ for (int i = 0; i < values.length; ++i)
+ map.addColumn(new Column(ByteBufferUtil.bytes(values[reversed ? values.length - 1 - i : i])));
+
+ Iterator<IColumn> iter = map.iterator();
+ assertEquals("1st column", 1, iter.next().name().getInt(0));
+ assertEquals("2nd column", 2, iter.next().name().getInt(0));
+ assertEquals("3rd column", 3, iter.next().name().getInt(0));
+ }
+
+ @Test
+ public void testAddAll()
+ {
+ testAddAllInternal(false);
+ testAddAllInternal(true);
+ }
+
+ private void testAddAllInternal(boolean reversed)
+ {
+ ISortedColumns map = ArrayBackedSortedColumns.FACTORY.create(BytesType.instance, reversed);
+ ISortedColumns map2 = ArrayBackedSortedColumns.FACTORY.create(BytesType.instance, reversed);
+
+ int[] values1 = new int[]{ 1, 3, 5, 6 };
+ int[] values2 = new int[]{ 2, 4, 5, 6 };
+
+ for (int i = 0; i < values1.length; ++i)
+ map.addColumn(new Column(ByteBufferUtil.bytes(values1[reversed ? values1.length - 1 - i : i])));
+
+ for (int i = 0; i < values2.length; ++i)
+ map2.addColumn(new Column(ByteBufferUtil.bytes(values2[reversed ? values2.length - 1 - i : i])));
+
+ map2.addAll(map);
+
+ Iterator<IColumn> iter = map2.iterator();
+ assertEquals("1st column", 1, iter.next().name().getInt(0));
+ assertEquals("2nd column", 2, iter.next().name().getInt(0));
+ assertEquals("3rd column", 3, iter.next().name().getInt(0));
+ assertEquals("4st column", 4, iter.next().name().getInt(0));
+ assertEquals("5st column", 5, iter.next().name().getInt(0));
+ assertEquals("6st column", 6, iter.next().name().getInt(0));
+ }
+
+ @Test
+ public void testGetCollection()
+ {
+ testGetCollectionInternal(false);
+ testGetCollectionInternal(true);
+ }
+
+ private void testGetCollectionInternal(boolean reversed)
+ {
+ ISortedColumns map = ArrayBackedSortedColumns.FACTORY.create(BytesType.instance, reversed);
+ int[] values = new int[]{ 1, 2, 3, 5, 9 };
+
+ List<IColumn> sorted = new ArrayList<IColumn>();
+ for (int v : values)
+ sorted.add(new Column(ByteBufferUtil.bytes(v)));
+ List<IColumn> reverseSorted = new ArrayList<IColumn>(sorted);
+ Collections.reverse(reverseSorted);
+
+ for (int i = 0; i < values.length; ++i)
+ map.addColumn(new Column(ByteBufferUtil.bytes(values[reversed ? values.length - 1 - i : i])));
+
+ assertSame(sorted, map.getSortedColumns());
+ assertSame(reverseSorted, map.getReverseSortedColumns());
+ }
+
+ @Test
+ public void testGetNames()
+ {
+ testGetNamesInternal(false);
+ testGetNamesInternal(true);
+ }
+
+ private void testGetNamesInternal(boolean reversed)
+ {
+ ISortedColumns map = ArrayBackedSortedColumns.FACTORY.create(BytesType.instance, reversed);
+ List<ByteBuffer> names = new ArrayList<ByteBuffer>();
+ int[] values = new int[]{ 1, 2, 3, 5, 9 };
+ for (int v : values)
+ names.add(ByteBufferUtil.bytes(v));
+
+ for (int i = 0; i < values.length; ++i)
+ map.addColumn(new Column(ByteBufferUtil.bytes(values[reversed ? values.length - 1 - i : i])));
+
+ assertSame(names, map.getColumnNames());
+ }
+
+ private <T> void assertSame(Collection<T> c1, Collection<T> c2)
+ {
+ Iterator<T> iter1 = c1.iterator();
+ Iterator<T> iter2 = c2.iterator();
+ while (iter1.hasNext() && iter2.hasNext())
+ assertEquals(iter1.next(), iter2.next());
+ if (iter1.hasNext() || iter2.hasNext())
+ fail("The collection don't have the same size");
+ }
+}
Modified: cassandra/trunk/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java?rev=1155426&r1=1155425&r2=1155426&view=diff
==============================================================================
--- cassandra/trunk/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java (original)
+++ cassandra/trunk/test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java Tue Aug 9 16:34:49 2011
@@ -104,12 +104,12 @@ public class ColumnFamilyStoreTest exten
QueryFilter sliceFilter = QueryFilter.getSliceFilter(Util.dk("key1"), new QueryPath("Standard2", null, null), ByteBufferUtil.EMPTY_BYTE_BUFFER, ByteBufferUtil.EMPTY_BYTE_BUFFER, false, 1);
ColumnFamily cf = store.getColumnFamily(sliceFilter);
assert cf.isMarkedForDelete();
- assert cf.getColumnsMap().isEmpty();
+ assert cf.isEmpty();
QueryFilter namesFilter = QueryFilter.getNamesFilter(Util.dk("key1"), new QueryPath("Standard2", null, null), ByteBufferUtil.bytes("a"));
cf = store.getColumnFamily(namesFilter);
assert cf.isMarkedForDelete();
- assert cf.getColumnsMap().isEmpty();
+ assert cf.isEmpty();
}
};
Modified: cassandra/trunk/test/unit/org/apache/cassandra/db/RowTest.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/test/unit/org/apache/cassandra/db/RowTest.java?rev=1155426&r1=1155425&r2=1155426&view=diff
==============================================================================
--- cassandra/trunk/test/unit/org/apache/cassandra/db/RowTest.java (original)
+++ cassandra/trunk/test/unit/org/apache/cassandra/db/RowTest.java Tue Aug 9 16:34:49 2011
@@ -43,7 +43,7 @@ public class RowTest extends SchemaLoade
cf2.delete(0, 0);
ColumnFamily cfDiff = cf1.diff(cf2);
- assertEquals(cfDiff.getColumnsMap().size(), 0);
+ assertEquals(cfDiff.getColumnCount(), 0);
assertEquals(cfDiff.getMarkedForDeleteAt(), 0);
}
Modified: cassandra/trunk/test/unit/org/apache/cassandra/service/AntiEntropyServiceTestAbstract.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/test/unit/org/apache/cassandra/service/AntiEntropyServiceTestAbstract.java?rev=1155426&r1=1155425&r2=1155426&view=diff
==============================================================================
--- cassandra/trunk/test/unit/org/apache/cassandra/service/AntiEntropyServiceTestAbstract.java (original)
+++ cassandra/trunk/test/unit/org/apache/cassandra/service/AntiEntropyServiceTestAbstract.java Tue Aug 9 16:34:49 2011
@@ -152,7 +152,7 @@ public abstract class AntiEntropyService
// add a row
validator.add(new PrecompactedRow(new DecoratedKey(mid, ByteBufferUtil.bytes("inconceivable!")),
- new ColumnFamily(DatabaseDescriptor.getCFMetaData(tablename, cfname))));
+ ColumnFamily.create(DatabaseDescriptor.getCFMetaData(tablename, cfname))));
validator.completeTree();
// confirm that the tree was validated
Modified: cassandra/trunk/test/unit/org/apache/cassandra/streaming/StreamingTransferTest.java
URL: http://svn.apache.org/viewvc/cassandra/trunk/test/unit/org/apache/cassandra/streaming/StreamingTransferTest.java?rev=1155426&r1=1155425&r2=1155426&view=diff
==============================================================================
--- cassandra/trunk/test/unit/org/apache/cassandra/streaming/StreamingTransferTest.java (original)
+++ cassandra/trunk/test/unit/org/apache/cassandra/streaming/StreamingTransferTest.java Tue Aug 9 16:34:49 2011
@@ -253,8 +253,8 @@ public class StreamingTransferTest exten
assertEquals(2, rows.size());
assert rows.get(0).key.key.equals(ByteBufferUtil.bytes("test"));
assert rows.get(1).key.key.equals(ByteBufferUtil.bytes("transfer3"));
- assert rows.get(0).cf.getColumnsMap().size() == 1;
- assert rows.get(1).cf.getColumnsMap().size() == 1;
+ assert rows.get(0).cf.getColumnCount() == 1;
+ assert rows.get(1).cf.getColumnCount() == 1;
// these keys fall outside of the ranges and should not be transferred
assert null == cfstore.getColumnFamily(QueryFilter.getIdentityFilter(Util.dk("transfer1"), new QueryPath("Standard1")));