natas_hw
/
cassandra
mirror of https://github.com/apache/cassandra
0
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

add IdentityQueryFilter and finish removing IFilter. 

patch by jbellis; reviewed by Jun Rao for CASSANDRA-287

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@793055 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2009-07-10 17:49:18 +00:00
parent 7b6fb2b36c
commit dd4f4be2b4
26 changed files with 125 additions and 750 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
src/java/org/apache/cassandra/db/AbstractColumnFactory.java
View File

@ -22,6 +22,7 @@ import java.util.HashMap;
import java.util.Map;
import java.util.StringTokenizer;
import org.apache.cassandra.io.ICompactSerializer;
/**
@ -51,7 +52,7 @@ abstract class AbstractColumnFactory
public abstract IColumn createColumn(String name, byte[] value);
public abstract IColumn createColumn(String name, byte[] value, long timestamp);
public abstract IColumn createColumn(String name, byte[] value, long timestamp, boolean deleted);
public abstract ICompactSerializer2<IColumn> createColumnSerializer();
public abstract ICompactSerializer<IColumn> createColumnSerializer();
}
class ColumnFactory extends AbstractColumnFactory
@ -75,7 +76,7 @@ class ColumnFactory extends AbstractColumnFactory
return new Column(name, value, timestamp, deleted);
}
public ICompactSerializer2<IColumn> createColumnSerializer()
public ICompactSerializer<IColumn> createColumnSerializer()
{
return Column.serializer();
}
@ -130,7 +131,7 @@ class SuperColumnFactory extends AbstractColumnFactory
return superColumn;
}
public ICompactSerializer2<IColumn> createColumnSerializer()
public ICompactSerializer<IColumn> createColumnSerializer()
{
return SuperColumn.serializer();
}

77
src/java/org/apache/cassandra/db/Column.java
View File

@ -22,12 +22,12 @@ import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.util.Collection;
import java.util.Map;
import java.nio.ByteBuffer;
import org.apache.commons.lang.ArrayUtils;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.io.ICompactSerializer;
/**
@ -212,7 +212,7 @@ public final class Column implements IColumn
}
}
class ColumnSerializer implements ICompactSerializer2<IColumn>
class ColumnSerializer implements ICompactSerializer<IColumn>
{
public void serialize(IColumn column, DataOutputStream dos) throws IOException
{
@ -223,81 +223,14 @@ class ColumnSerializer implements ICompactSerializer2<IColumn>
dos.write(column.value());
}
private IColumn defreeze(DataInputStream dis, String name) throws IOException
public IColumn deserialize(DataInputStream dis) throws IOException
{
IColumn column = null;
String name = dis.readUTF();
boolean delete = dis.readBoolean();
long ts = dis.readLong();
int size = dis.readInt();
byte[] value = new byte[size];
dis.readFully(value);
column = new Column(name, value, ts, delete);
return column;
}
public IColumn deserialize(DataInputStream dis) throws IOException
{
String name = dis.readUTF();
return defreeze(dis, name);
}
/**
* Here we need to get the column and apply the filter.
*/
public IColumn deserialize(DataInputStream dis, IFilter filter) throws IOException
{
assert dis.available() > 0;
String name = dis.readUTF();
IColumn column = new Column(name);
column = filter.filter(column, dis);
if ( column != null )
{
column = defreeze(dis, name);
}
else
{
/* Skip a boolean and the timestamp */
dis.skip(DBConstants.boolSize_ + DBConstants.tsSize_);
int size = dis.readInt();
dis.skip(size);
}
return column;
}
/**
* We know the name of the column here so just return it.
* Filter is pretty much useless in this call and is ignored.
*/
public IColumn deserialize(DataInputStream dis, String columnName, IFilter filter) throws IOException
{
assert dis.available() > 0;
IColumn column = null;
String name = dis.readUTF();
if ( name.equals(columnName) )
{
column = defreeze(dis, name);
}
else
{
/* Skip a boolean and the timestamp */
dis.skip(DBConstants.boolSize_ + DBConstants.tsSize_);
int size = dis.readInt();
dis.skip(size);
}
return column;
}
public void skip(DataInputStream dis) throws IOException
{
/* read the column name */
dis.readUTF();
/* boolean indicating if the column is deleted */
dis.readBoolean();
/* timestamp associated with the column */
dis.readLong();
/* size of the column */
int size = dis.readInt();
dis.skip(size);
return new Column(name, value, ts, delete);
}
}

83
src/java/org/apache/cassandra/db/ColumnFamily.java
View File

@ -37,6 +37,7 @@ import org.apache.log4j.Logger;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.io.ICompactSerializer;
/**
* Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
@ -44,7 +45,7 @@ import org.apache.cassandra.utils.FBUtilities;
public final class ColumnFamily
{
/* The column serializer for this Column Family. Create based on config. */
private static ICompactSerializer2<ColumnFamily> serializer_;
private static ICompactSerializer<ColumnFamily> serializer_;
public static final short utfPrefix_ = 2;
private static Logger logger_ = Logger.getLogger( ColumnFamily.class );
@ -64,7 +65,7 @@ public final class ColumnFamily
indexTypes_.put("Time", "Time");
}
public static ICompactSerializer2<ColumnFamily> serializer()
public static ICompactSerializer<ColumnFamily> serializer()
{
return serializer_;
}
@ -73,9 +74,9 @@ public final class ColumnFamily
* This method returns the serializer whose methods are
* preprocessed by a dynamic proxy.
*/
public static ICompactSerializer2<ColumnFamily> serializerWithIndexes()
public static ICompactSerializer<ColumnFamily> serializerWithIndexes()
{
return (ICompactSerializer2<ColumnFamily>)Proxy.newProxyInstance( ColumnFamily.class.getClassLoader(), new Class[]{ICompactSerializer2.class}, new CompactSerializerInvocationHandler<ColumnFamily>(serializer_) );
return (ICompactSerializer<ColumnFamily>)Proxy.newProxyInstance( ColumnFamily.class.getClassLoader(), new Class[]{ICompactSerializer.class}, new CompactSerializerInvocationHandler<ColumnFamily>(serializer_) );
}
public static String getColumnType(String key)
@ -112,7 +113,7 @@ public final class ColumnFamily
private String name_;
private transient ICompactSerializer2<IColumn> columnSerializer_;
private transient ICompactSerializer<IColumn> columnSerializer_;
private long markedForDeleteAt = Long.MIN_VALUE;
private int localDeletionTime = Integer.MIN_VALUE;
private AtomicInteger size_ = new AtomicInteger(0);
@ -165,7 +166,7 @@ public final class ColumnFamily
}
}
public ICompactSerializer2<IColumn> getColumnSerializer()
public ICompactSerializer<IColumn> getColumnSerializer()
{
return columnSerializer_;
}
@ -449,7 +450,7 @@ public final class ColumnFamily
return cf;
}
public static class ColumnFamilySerializer implements ICompactSerializer2<ColumnFamily>
public static class ColumnFamilySerializer implements ICompactSerializer<ColumnFamily>
{
/*
* We are going to create indexes, and write out that information as well. The format
@ -492,22 +493,12 @@ public final class ColumnFamily
}
}
/*
* Use this method to create a bare bones Column Family. This column family
* does not have any of the Column information.
*/
private ColumnFamily defreezeColumnFamily(DataInputStream dis) throws IOException
public ColumnFamily deserialize(DataInputStream dis) throws IOException
{
ColumnFamily cf = new ColumnFamily(dis.readUTF(),
dis.readUTF(),
ColumnComparatorFactory.ComparatorType.values()[dis.readInt()]);
cf.delete(dis.readInt(), dis.readLong());
return cf;
}
public ColumnFamily deserialize(DataInputStream dis) throws IOException
{
ColumnFamily cf = defreezeColumnFamily(dis);
int size = dis.readInt();
IColumn column;
for (int i = 0; i < size; ++i)
@ -517,62 +508,6 @@ public final class ColumnFamily
}
return cf;
}
/*
* This version of deserialize is used when we need a specific set if columns for
* a column family specified in the name cfName parameter.
*/
public ColumnFamily deserialize(DataInputStream dis, IFilter filter) throws IOException
{
ColumnFamily cf = defreezeColumnFamily(dis);
int size = dis.readInt();
IColumn column = null;
for ( int i = 0; i < size; ++i )
{
column = cf.getColumnSerializer().deserialize(dis, filter);
if(column != null)
{
cf.addColumn(column);
}
}
return cf;
}
/*
* Deserialize a particular column or super column or the entire columnfamily given a : seprated name
* name could be of the form cf:superColumn:column or cf:column or cf
*/
public ColumnFamily deserialize(DataInputStream dis, String name, IFilter filter) throws IOException
{
String[] names = RowMutation.getColumnAndColumnFamily(name);
String columnName = "";
if ( names.length == 1 )
return deserialize(dis, filter);
if( names.length == 2 )
columnName = names[1];
if( names.length == 3 )
columnName = names[1]+ ":" + names[2];
ColumnFamily cf = defreezeColumnFamily(dis);
/* read the number of columns */
int size = dis.readInt();
for ( int i = 0; i < size; ++i )
{
IColumn column = cf.getColumnSerializer().deserialize(dis, columnName, filter);
if ( column != null )
{
cf.addColumn(column);
break;
}
}
return cf;
}
public void skip(DataInputStream dis) throws IOException
{
throw new UnsupportedOperationException("This operation is not yet supported.");
}
}
}

155
src/java/org/apache/cassandra/db/ColumnFamilyStore.java
View File

@ -508,116 +508,6 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
writeStats_.add(System.currentTimeMillis() - start);
}
public ColumnFamily getColumnFamily(String key, String columnFamilyColumn, IFilter filter) throws IOException
{
long start = System.currentTimeMillis();
List<ColumnFamily> columnFamilies = getColumnFamilies(key, columnFamilyColumn, filter);
ColumnFamily cf = resolveAndRemoveDeleted(columnFamilies);
readStats_.add(System.currentTimeMillis() - start);
return cf;
}
public ColumnFamily getColumnFamily(String key, String columnFamilyColumn, IFilter filter, int gcBefore) throws IOException
{
long start = System.currentTimeMillis();
List<ColumnFamily> columnFamilies = getColumnFamilies(key, columnFamilyColumn, filter);
ColumnFamily cf = removeDeleted(ColumnFamily.resolve(columnFamilies), gcBefore);
readStats_.add(System.currentTimeMillis() - start);
return cf;
}
/**
* Get the column family in the most efficient order.
* 1. Memtable
* 2. Sorted list of files
*/
List<ColumnFamily> getColumnFamilies(String key, String columnFamilyColumn, IFilter filter) throws IOException
{
List<ColumnFamily> columnFamilies = new ArrayList<ColumnFamily>();
/* Get the ColumnFamily from Memtable */
getColumnFamilyFromCurrentMemtable(key, columnFamilyColumn, filter, columnFamilies);
/* Check if MemtableManager has any historical information */
getUnflushedColumnFamily(key, columnFamily_, columnFamilyColumn, filter, columnFamilies);
long start = System.currentTimeMillis();
getColumnFamilyFromDisk(key, columnFamilyColumn, columnFamilies, filter);
diskReadStats_.add(System.currentTimeMillis() - start);
return columnFamilies;
}
/**
* Fetch from disk files and go in sorted order to be efficient
* This function exits as soon as the required data is found.
*
* @param key
* @param cf
* @param columnFamilies
* @param filter
* @throws IOException
*/
private void getColumnFamilyFromDisk(String key, String cf, List<ColumnFamily> columnFamilies, IFilter filter) throws IOException
{
sstableLock_.readLock().lock();
try
{
for (SSTableReader sstable : ssTables_.values())
{
ColumnFamily columnFamily = null;
try
{
columnFamily = fetchColumnFamily(key, cf, filter, sstable);
}
catch (IOException e)
{
// annotate exception w/ more information about context
throw new IOException("Error fetching " + key + ":" + cf + " from " + sstable, e);
}
if (columnFamily != null)
{
columnFamilies.add(columnFamily);
}
}
}
finally
{
sstableLock_.readLock().unlock();
}
}
private ColumnFamily fetchColumnFamily(String key, String cf, IFilter filter, SSTableReader ssTable) throws IOException
{
DataInputBuffer bufIn;
bufIn = filter.next(key, cf, ssTable);
if (bufIn.getLength() == 0)
{
return null;
}
ColumnFamily columnFamily = ColumnFamily.serializer().deserialize(bufIn, cf, filter);
if (columnFamily == null)
{
return null;
}
return columnFamily;
}
private void getColumnFamilyFromCurrentMemtable(String key, String cf, IFilter filter, List<ColumnFamily> columnFamilies)
{
ColumnFamily columnFamily;
memtableLock_.readLock().lock();
try
{
columnFamily = memtable_.getLocalCopy(key, cf, filter);
}
finally
{
memtableLock_.readLock().unlock();
}
if (columnFamily != null)
{
columnFamilies.add(columnFamily);
}
}
/**
* like resolve, but leaves the resolved CF as the only item in the list
*/
@ -1404,26 +1294,6 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
return memtables;
}
/*
* Retrieve column family from the list of Memtables that have been
* submitted for flush but have not yet been flushed.
* It also filters out unneccesary columns based on the passed in filter.
*/
void getUnflushedColumnFamily(String key, String cfName, String cf, IFilter filter, List<ColumnFamily> columnFamilies)
{
List<Memtable> memtables = getUnflushedMemtables(cfName);
Collections.sort(memtables);
int size = memtables.size();
for ( int i = size - 1; i >= 0; --i )
{
ColumnFamily columnFamily = memtables.get(i).getLocalCopy(key, cf, filter);
if ( columnFamily != null )
{
columnFamilies.add(columnFamily);
}
}
}
/* Submit memtables to be flushed to disk */
public static void submitFlush(final Memtable memtable, final CommitLog.CommitLogContext cLogCtx)
{
@ -1554,11 +1424,11 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
}
/**
* get a list of columns starting from a given column, in a specified order
* only the latest version of a column is returned
* get a list of columns starting from a given column, in a specified order.
* only the latest version of a column is returned.
* @return null if there is no data and no tombstones; otherwise a ColumnFamily
*/
public ColumnFamily getColumnFamily(QueryFilter filter)
throws IOException
public ColumnFamily getColumnFamily(QueryFilter filter) throws IOException
{
String[] values = RowMutation.getColumnAndColumnFamily(filter.columnFamilyColumn);
@ -1567,9 +1437,12 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
{
QueryFilter nameFilter = new NamesQueryFilter(filter.key, values[0], values[1]);
ColumnFamily cf = getColumnFamily(nameFilter);
for (IColumn column : cf.getAllColumns())
if (cf != null)
{
filter.filterSuperColumn((SuperColumn) column);
for (IColumn column : cf.getAllColumns())
{
filter.filterSuperColumn((SuperColumn) column);
}
}
return removeDeleted(cf);
}
@ -1609,21 +1482,17 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
for (SSTableReader sstable : sstables)
{
iter = filter.getSSTableColumnIterator(sstable);
if (iter.hasNext())
if (iter.hasNext()) // initializes iter.CF
{
returnCF.delete(iter.getColumnFamily());
iterators.add(iter);
}
else
{
iter.close();
}
iterators.add(iter);
}
Comparator<IColumn> comparator = filter.getColumnComparator();
Iterator collated = IteratorUtils.collatedIterator(comparator, iterators);
if (!collated.hasNext())
return ColumnFamily.create(table_, columnFamily_);
return null;
filter.collectColumns(returnCF, collated);

11
src/java/org/apache/cassandra/db/ColumnReadCommand.java
View File

@ -21,6 +21,8 @@ package org.apache.cassandra.db;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.util.TreeSet;
import java.util.Arrays;
public class ColumnReadCommand extends ReadCommand
{
@ -49,7 +51,14 @@ public class ColumnReadCommand extends ReadCommand
@Override
public Row getRow(Table table) throws IOException
{
return table.getRow(key, columnFamilyColumn);
String[] values = RowMutation.getColumnAndColumnFamily(columnFamilyColumn);
assert values.length > 1 && values.length <= 3;
if (values.length == 2)
{
return table.getRow(key, values[0], new TreeSet<String>(Arrays.asList(values[1])));
}
assert values.length == 3 : columnFamilyColumn;
return table.getRow(key, values[0] + ":" + values[1], new TreeSet<String>(Arrays.asList(values[2])));
}
@Override

7
src/java/org/apache/cassandra/db/CompactSerializerInvocationHandler.java
View File

@ -22,19 +22,20 @@ import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import org.apache.cassandra.io.DataOutputBuffer;
import org.apache.cassandra.io.ICompactSerializer;
/*
* This is the abstraction that pre-processes calls to implementations
* of the ICompactSerializer2 serialize() via dynamic proxies.
* of the ICompactSerializer serialize() via dynamic proxies.
* Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
*/
public class CompactSerializerInvocationHandler<T> implements InvocationHandler
{
private ICompactSerializer2<T> serializer_;
private ICompactSerializer<T> serializer_;
public CompactSerializerInvocationHandler(ICompactSerializer2<T> serializer)
public CompactSerializerInvocationHandler(ICompactSerializer<T> serializer)
{
serializer_ = serializer;
}

3
src/java/org/apache/cassandra/db/HintedHandOffManager.java
View File

@ -37,6 +37,7 @@ import org.apache.cassandra.net.EndPoint;
import org.apache.cassandra.net.Message;
import org.apache.cassandra.net.MessagingService;
import org.apache.cassandra.service.*;
import org.apache.cassandra.db.filter.IdentityQueryFilter;
/**
@ -159,7 +160,7 @@ public class HintedHandOffManager
// 7. I guess we are done
for (String tableName : DatabaseDescriptor.getTables())
{
ColumnFamily hintColumnFamily = ColumnFamilyStore.removeDeleted(hintStore.getColumnFamily(tableName, HINTS_CF, new IdentityFilter()), Integer.MAX_VALUE);
ColumnFamily hintColumnFamily = ColumnFamilyStore.removeDeleted(hintStore.getColumnFamily(new IdentityQueryFilter(tableName, HINTS_CF)), Integer.MAX_VALUE);
if (hintColumnFamily == null)
{
continue;

65
src/java/org/apache/cassandra/db/ICompactSerializer2.java
View File

@ -1,65 +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.db;
import java.io.DataInputStream;
import java.io.IOException;
import java.util.List;
import org.apache.cassandra.io.ICompactSerializer;
import org.apache.cassandra.utils.BloomFilter;
/**
* This interface is an extension of the ICompactSerializer which allows for partial deserialization
* of a type.
* Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
*/
public interface ICompactSerializer2<T> extends ICompactSerializer<T>
{
/**
* Returns an instance of an IColumn which contains only the
* columns that are required. This is specified in the <i>columnNames</i>
* argument.
*
* @param dis DataInput from which we need to deserialize.
* @throws IOException
* @return type which contains the specified items.
*/
public T deserialize(DataInputStream dis, IFilter filter) throws IOException;
/**
* This method is used to deserialize just the specified field from
* the serialized stream.
*
* @param dis DataInput from which we need to deserialize.
* @param name name of the desired field.
* @throws IOException
* @return the deserialized type.
*/
public T deserialize(DataInputStream dis, String name, IFilter filter) throws IOException;
/**
*
* @param dis
* @throws IOException
*/
public void skip(DataInputStream dis) throws IOException;
}

33
src/java/org/apache/cassandra/db/IFilter.java
View File

@ -1,33 +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.db;
import java.io.DataInputStream;
import java.io.IOException;
import org.apache.cassandra.io.DataInputBuffer;
import org.apache.cassandra.io.SSTableReader;
public interface IFilter
{
public ColumnFamily filter(String cfName, ColumnFamily cf);
public IColumn filter(IColumn column, DataInputStream dis) throws IOException;
public DataInputBuffer next(String key, String cf, SSTableReader ssTable) throws IOException;
}

44
src/java/org/apache/cassandra/db/IdentityFilter.java
View File

@ -1,44 +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.db;
import java.io.DataInputStream;
import java.io.IOException;
import org.apache.cassandra.io.DataInputBuffer;
import org.apache.cassandra.io.SSTableReader;
public class IdentityFilter implements IFilter
{
public ColumnFamily filter(String cfString, ColumnFamily columnFamily)
{
return columnFamily;
}
public IColumn filter(IColumn column, DataInputStream dis) throws IOException
{
return column;
}
public DataInputBuffer next(String key, String cf, SSTableReader ssTable) throws IOException
{
return ssTable.next(key, cf);
}
}

42
src/java/org/apache/cassandra/db/Memtable.java
View File

@ -204,48 +204,6 @@ public class Memtable implements Comparable<Memtable>
return builder.toString();
}
ColumnFamily getLocalCopy(String key, String columnFamilyColumn, IFilter filter)
{
String[] values = RowMutation.getColumnAndColumnFamily(columnFamilyColumn);
ColumnFamily columnFamily = null;
if(values.length == 1 )
{
columnFamily = columnFamilies_.get(key);
}
else
{
ColumnFamily cFamily = columnFamilies_.get(key);
if (cFamily == null) return null;
if (values.length == 2) {
IColumn column = cFamily.getColumn(values[1]); // super or normal column
if (column != null )
{
columnFamily = cFamily.cloneMeShallow();
columnFamily.addColumn(column);
}
}
else
{
assert values.length == 3;
SuperColumn superColumn = (SuperColumn)cFamily.getColumn(values[1]);
if (superColumn != null)
{
IColumn subColumn = superColumn.getSubColumn(values[2]);
if (subColumn != null)
{
columnFamily = cFamily.cloneMeShallow();
SuperColumn container = superColumn.cloneMeShallow();
container.addColumn(subColumn);
columnFamily.addColumn(container);
}
}
}
}
/* Filter unnecessary data from the column based on the provided filter */
return filter.filter(columnFamilyColumn, columnFamily);
}
void flush(CommitLog.CommitLogContext cLogCtx) throws IOException
{
logger_.info("Flushing " + this);

13
src/java/org/apache/cassandra/db/ReadCommand.java
View File

@ -33,13 +33,10 @@ import org.apache.cassandra.service.StorageService;
public abstract class ReadCommand
{
public static final String DO_REPAIR = "READ-REPAIR";
public static final byte CMD_TYPE_GET_ROW=1;
public static final byte CMD_TYPE_GET_COLUMN=2;
public static final byte CMD_TYPE_GET_SLICE_BY_NAMES=3;
public static final byte CMD_TYPE_GET_COLUMNS_SINCE=4;
public static final byte CMD_TYPE_GET_SLICE=5;
public static final byte CMD_TYPE_GET_SLICE_BY_RANGE = 6;
public static final byte CMD_TYPE_GET_SLICE_FROM=7;
public static final byte CMD_TYPE_GET_COLUMN=1;
public static final byte CMD_TYPE_GET_SLICE_BY_NAMES=2;
public static final byte CMD_TYPE_GET_COLUMNS_SINCE=3;
public static final byte CMD_TYPE_GET_SLICE =4;
public static final String EMPTY_CF = "";
@ -95,7 +92,7 @@ class ReadCommandSerializer implements ICompactSerializer<ReadCommand>
CMD_SERIALIZER_MAP.put(ReadCommand.CMD_TYPE_GET_COLUMN, new ColumnReadCommandSerializer());
CMD_SERIALIZER_MAP.put(ReadCommand.CMD_TYPE_GET_SLICE_BY_NAMES, new SliceByNamesReadCommandSerializer());
CMD_SERIALIZER_MAP.put(ReadCommand.CMD_TYPE_GET_COLUMNS_SINCE, new ColumnsSinceReadCommandSerializer());
CMD_SERIALIZER_MAP.put(ReadCommand.CMD_TYPE_GET_SLICE_FROM, new SliceFromReadCommandSerializer());
CMD_SERIALIZER_MAP.put(ReadCommand.CMD_TYPE_GET_SLICE, new SliceFromReadCommandSerializer());
}

2
src/java/org/apache/cassandra/db/SliceFromReadCommand.java
View File

@ -31,7 +31,7 @@ public class SliceFromReadCommand extends ReadCommand
public SliceFromReadCommand(String table, String key, String columnFamilyColumn, String start, String finish, boolean isAscending, int offset, int count)
{
super(table, key, CMD_TYPE_GET_SLICE_FROM);
super(table, key, CMD_TYPE_GET_SLICE);
this.columnFamilyColumn = columnFamilyColumn;
this.start = start;
this.finish = finish;

114
src/java/org/apache/cassandra/db/SuperColumn.java
View File

@ -24,7 +24,6 @@ import java.io.IOException;
import java.io.Serializable;
import java.util.Collection;
import java.util.Set;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.lang.ArrayUtils;
@ -32,6 +31,7 @@ import org.apache.commons.lang.StringUtils;
import org.apache.log4j.Logger;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.io.ICompactSerializer;
/**
* Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
@ -319,7 +319,7 @@ public final class SuperColumn implements IColumn, Serializable
}
}
class SuperColumnSerializer implements ICompactSerializer2<IColumn>
class SuperColumnSerializer implements ICompactSerializer<IColumn>
{
public void serialize(IColumn column, DataOutputStream dos) throws IOException
{
@ -339,37 +339,11 @@ class SuperColumnSerializer implements ICompactSerializer2<IColumn>
}
}
/*
* Use this method to create a bare bones Super Column. This super column
* does not have any of the Column information.
*/
private SuperColumn defreezeSuperColumn(DataInputStream dis) throws IOException
public IColumn deserialize(DataInputStream dis) throws IOException
{
String name = dis.readUTF();
SuperColumn superColumn = new SuperColumn(name);
superColumn.markForDeleteAt(dis.readInt(), dis.readLong());
return superColumn;
}
public IColumn deserialize(DataInputStream dis) throws IOException
{
SuperColumn superColumn = defreezeSuperColumn(dis);
fillSuperColumn(superColumn, dis);
return superColumn;
}
public void skip(DataInputStream dis) throws IOException
{
defreezeSuperColumn(dis);
/* read the number of columns stored */
dis.readInt();
/* read the size of all columns to skip */
int size = dis.readInt();
dis.skip(size);
}
private void fillSuperColumn(IColumn superColumn, DataInputStream dis) throws IOException
{
assert dis.available() > 0;
/* read the number of columns */
@ -381,87 +355,7 @@ class SuperColumnSerializer implements ICompactSerializer2<IColumn>
IColumn subColumn = Column.serializer().deserialize(dis);
superColumn.addColumn(subColumn);
}
}
public IColumn deserialize(DataInputStream dis, IFilter filter) throws IOException
{
assert dis.available() > 0;
IColumn superColumn = defreezeSuperColumn(dis);
superColumn = filter.filter(superColumn, dis);
if(superColumn != null)
{
fillSuperColumn(superColumn, dis);
return superColumn;
}
else
{
/* read the number of columns stored */
dis.readInt();
/* read the size of all columns to skip */
int size = dis.readInt();
dis.skip(size);
return null;
}
}
/*
* Deserialize a particular column since the name is in the form of
* superColumn:column.
*/
public IColumn deserialize(DataInputStream dis, String name, IFilter filter) throws IOException
{
assert dis.available() > 0;
String[] names = RowMutation.getColumnAndColumnFamily(name);
if ( names.length == 1 )
{
IColumn superColumn = defreezeSuperColumn(dis);
if(name.equals(superColumn.name()))
{
/* read the number of columns stored */
int size = dis.readInt();
/* read the size of all columns */
dis.readInt();
IColumn column = null;
for ( int i = 0; i < size; ++i )
{
column = Column.serializer().deserialize(dis, filter);
if(column != null)
{
superColumn.addColumn(column);
}
}
return superColumn;
}
else
{
/* read the number of columns stored */
dis.readInt();
/* read the size of all columns to skip */
int size = dis.readInt();
dis.skip(size);
return null;
}
}
SuperColumn superColumn = defreezeSuperColumn(dis);
int size = dis.readInt();
/* skip the size of the columns */
dis.readInt();
if ( size > 0 )
{
for ( int i = 0; i < size; ++i )
{
IColumn subColumn = Column.serializer().deserialize(dis, names[1], filter);
if ( subColumn != null )
{
superColumn.addColumn(subColumn);
break;
}
}
}
return superColumn;
}
}

6
src/java/org/apache/cassandra/db/SystemTable.java
View File

@ -26,6 +26,8 @@ import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.dht.Token;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.utils.BasicUtilities;
import org.apache.cassandra.db.filter.IdentityQueryFilter;
import org.apache.cassandra.db.filter.NamesQueryFilter;
/**
* Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
@ -47,7 +49,7 @@ public class SystemTable
IPartitioner p = StorageService.getPartitioner();
Table table = Table.open(Table.SYSTEM_TABLE);
/* Retrieve the "LocationInfo" column family */
ColumnFamily cf = table.getColumnFamilyStore(LOCATION_CF).getColumnFamily(LOCATION_KEY, LOCATION_KEY + ":" + TOKEN, new IdentityFilter());
ColumnFamily cf = table.getColumnFamilyStore(LOCATION_CF).getColumnFamily(new NamesQueryFilter(LOCATION_KEY, LOCATION_KEY, TOKEN));
long oldTokenColumnTimestamp = cf.getColumn(SystemTable.TOKEN).timestamp();
/* create the "Token" whose value is the new token. */
IColumn tokenColumn = new Column(SystemTable.TOKEN, p.getTokenFactory().toByteArray(token), oldTokenColumnTimestamp + 1);
@ -71,7 +73,7 @@ public class SystemTable
{
/* Read the system table to retrieve the storage ID and the generation */
Table table = Table.open(Table.SYSTEM_TABLE);
ColumnFamily cf = table.getColumnFamilyStore(LOCATION_CF).getColumnFamily(LOCATION_KEY, LOCATION_KEY + ":" + GENERATION, new IdentityFilter());
ColumnFamily cf = table.getColumnFamilyStore(LOCATION_CF).getColumnFamily(new NamesQueryFilter(LOCATION_KEY, LOCATION_KEY, GENERATION));
IPartitioner p = StorageService.getPartitioner();
if (cf == null)

72
src/java/org/apache/cassandra/db/Table.java
View File

@ -43,10 +43,7 @@ import org.apache.cassandra.net.io.IStreamComplete;
import org.apache.cassandra.net.io.StreamContextManager;
import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.*;
import org.apache.cassandra.db.filter.QueryFilter;
import org.apache.cassandra.db.filter.SliceQueryFilter;
import org.apache.cassandra.db.filter.NamesQueryFilter;
import org.apache.cassandra.db.filter.TimeQueryFilter;
import org.apache.cassandra.db.filter.*;
import org.apache.log4j.Logger;
@ -502,23 +499,16 @@ public class Table
*/
@Deprecated // CF should be our atom of work, not Row
public Row get(String key) throws IOException
{
{
Row row = new Row(table_, key);
Set<String> columnFamilies = tableMetadata_.getColumnFamilies();
long start = System.currentTimeMillis();
for ( String columnFamily : columnFamilies )
for (String columnFamily : getColumnFamilies())
{
ColumnFamilyStore cfStore = columnFamilyStores_.get(columnFamily);
if ( cfStore != null )
{
ColumnFamily cf = cfStore.getColumnFamily(key, columnFamily, new IdentityFilter());
if ( cf != null )
row.addColumnFamily(cf);
ColumnFamily cf = get(key, columnFamily);
if (cf != null)
{
row.addColumnFamily(cf);
}
}
long timeTaken = System.currentTimeMillis() - start;
dbAnalyticsSource_.updateReadStatistics(timeTaken);
return row;
}
@ -527,25 +517,22 @@ public class Table
* Selects the specified column family for the specified key.
*/
@Deprecated // single CFs could be larger than memory
public ColumnFamily get(String key, String columnFamilyColumn) throws IOException
public ColumnFamily get(String key, String cfName) throws IOException
{
String[] values = RowMutation.getColumnAndColumnFamily(columnFamilyColumn);
long start = System.currentTimeMillis();
assert !cfName.contains(":") : cfName;
String[] values = RowMutation.getColumnAndColumnFamily(cfName);
ColumnFamilyStore cfStore = columnFamilyStores_.get(values[0]);
assert cfStore != null : "Column family " + columnFamilyColumn + " has not been defined";
ColumnFamily columnFamily = cfStore.getColumnFamily(key, columnFamilyColumn, new IdentityFilter());
long timeTaken = System.currentTimeMillis() - start;
dbAnalyticsSource_.updateReadStatistics(timeTaken);
return columnFamily;
assert cfStore != null : "Column family " + cfName + " has not been defined";
return cfStore.getColumnFamily(new IdentityQueryFilter(key, cfName));
}
/**
* Selects only the specified column family for the specified key.
*/
public Row getRow(String key, String cf) throws IOException
public Row getRow(String key, String cfName) throws IOException
{
Row row = new Row(table_, key);
ColumnFamily columnFamily = get(key, cf);
ColumnFamily columnFamily = get(key, cfName);
if ( columnFamily != null )
row.addColumnFamily(columnFamily);
return row;
@ -581,28 +568,14 @@ public class Table
return getRow(key, filter);
}
private Row getRow(String key, QueryFilter filter) throws IOException
public Row getRow(String key, QueryFilter filter) throws IOException
{
ColumnFamilyStore cfStore = columnFamilyStores_.get(filter.getColumnFamilyName());
Row row = new Row(table_, key);
long start1 = System.currentTimeMillis();
try
{
ColumnFamily columnFamily = cfStore.getColumnFamily(filter);
if (columnFamily != null)
row.addColumnFamily(columnFamily);
long timeTaken = System.currentTimeMillis() - start1;
dbAnalyticsSource_.updateReadStatistics(timeTaken);
return row;
}
catch (InterruptedException e)
{
throw new RuntimeException(e);
}
catch (ExecutionException e)
{
throw new RuntimeException(e);
}
ColumnFamily columnFamily = cfStore.getColumnFamily(filter);
if (columnFamily != null)
row.addColumnFamily(columnFamily);
return row;
}
/**
@ -612,8 +585,6 @@ public class Table
*/
void apply(Row row) throws IOException
{
/* Add row to the commit log. */
long start = System.currentTimeMillis();
CommitLog.CommitLogContext cLogCtx = CommitLog.open().add(row);
for (ColumnFamily columnFamily : row.getColumnFamilies())
@ -621,9 +592,6 @@ public class Table
ColumnFamilyStore cfStore = columnFamilyStores_.get(columnFamily.name());
cfStore.apply(row.key(), columnFamily, cLogCtx);
}
long timeTaken = System.currentTimeMillis() - start;
dbAnalyticsSource_.updateWriteStatistics(timeTaken);
}
void applyNow(Row row) throws IOException
@ -772,7 +740,7 @@ public class Table
}
// make sure there is actually non-tombstone content associated w/ this key
// TODO record the key source(s) somehow and only check that source (e.g., memtable or sstable)
if (cfs.getColumnFamily(current, columnFamily, new IdentityFilter(), Integer.MAX_VALUE) != null)
if (cfs.getColumnFamily(new SliceQueryFilter(current, columnFamily, "", "", true, 0, 1)) != null)
{
keys.add(current);
}

20
src/java/org/apache/cassandra/db/filter/IdentityQueryFilter.java Normal file
View File

@ -0,0 +1,20 @@
package org.apache.cassandra.db.filter;
import org.apache.cassandra.db.SuperColumn;
public class IdentityQueryFilter extends SliceQueryFilter
{
/**
* Only for use in testing; will read entire CF into memory.
*/
public IdentityQueryFilter(String key, String columnFamilyColumn)
{
super(key, columnFamilyColumn, "", "", true, 0, Integer.MAX_VALUE);
}
@Override
public void filterSuperColumn(SuperColumn superColumn)
{
// no filtering done, deliberately
}
}

70
src/java/org/apache/cassandra/io/SequenceFile.java
View File

@ -829,74 +829,4 @@ public class SequenceFile
{
return new BufferReader(filename, size);
}
/**
* Efficiently writes a UTF8 string to the buffer.
* Assuming all Strings that are passed in have length
* that can be represented as a short i.e length of the
* string is <= 65535
*
* @param buffer buffer to write the serialize version into
* @param str string to serialize
*/
protected static void writeUTF(ByteBuffer buffer, String str)
{
int strlen = str.length();
int utflen = 0;
int c, count = 0;
/* use charAt instead of copying String to char array */
for (int i = 0; i < strlen; i++)
{
c = str.charAt(i);
if ((c >= 0x0001) && (c <= 0x007F))
{
utflen++;
}
else if (c > 0x07FF)
{
utflen += 3;
}
else
{
utflen += 2;
}
}
byte[] bytearr = new byte[utflen + 2];
bytearr[count++] = (byte) ((utflen >>> 8) & 0xFF);
bytearr[count++] = (byte) ((utflen >>> 0) & 0xFF);
int i = 0;
for (i = 0; i < strlen; i++)
{
c = str.charAt(i);
if (!((c >= 0x0001) && (c <= 0x007F)))
break;
bytearr[count++] = (byte) c;
}
for (; i < strlen; i++)
{
c = str.charAt(i);
if ((c >= 0x0001) && (c <= 0x007F))
{
bytearr[count++] = (byte) c;
}
else if (c > 0x07FF)
{
bytearr[count++] = (byte) (0xE0 | ((c >> 12) & 0x0F));
bytearr[count++] = (byte) (0x80 | ((c >> 6) & 0x3F));
bytearr[count++] = (byte) (0x80 | ((c >> 0) & 0x3F));
}
else
{
bytearr[count++] = (byte) (0xC0 | ((c >> 6) & 0x1F));
bytearr[count++] = (byte) (0x80 | ((c >> 0) & 0x3F));
}
}
buffer.put(bytearr, 0, utflen + 2);
}
}

3
test/unit/org/apache/cassandra/db/ColumnFamilyStoreTest.java
View File

@ -30,6 +30,7 @@ import org.junit.Test;
import static junit.framework.Assert.assertEquals;
import org.apache.cassandra.CleanupHelper;
import org.apache.cassandra.db.filter.IdentityQueryFilter;
import org.apache.cassandra.io.SSTableReader;
public class ColumnFamilyStoreTest extends CleanupHelper
@ -115,7 +116,7 @@ public class ColumnFamilyStoreTest extends CleanupHelper
List<SSTableReader> ssTables = table.getAllSSTablesOnDisk();
assertEquals(1, ssTables.size());
ssTables.get(0).forceBloomFilterFailures();
ColumnFamily cf = store.getColumnFamily("key2", "Standard1:Column1", new IdentityFilter());
ColumnFamily cf = store.getColumnFamily(new IdentityQueryFilter("key2", "Standard1:Column1"));
assertNull(cf);
}
}

3
test/unit/org/apache/cassandra/db/RemoveColumnFamilyTest.java
View File

@ -24,6 +24,7 @@ import java.util.concurrent.ExecutionException;
import org.junit.Test;
import static junit.framework.Assert.assertNull;
import org.apache.cassandra.db.filter.IdentityQueryFilter;
public class RemoveColumnFamilyTest
{
@ -44,7 +45,7 @@ public class RemoveColumnFamilyTest
rm.delete("Standard1", 1);
rm.apply();
ColumnFamily retrieved = store.getColumnFamily("key1", "Standard1:Column1", new IdentityFilter());
ColumnFamily retrieved = store.getColumnFamily(new IdentityQueryFilter("key1", "Standard1:Column1"));
assert retrieved.isMarkedForDelete();
assertNull(retrieved.getColumn("Column1"));
assertNull(ColumnFamilyStore.removeDeleted(retrieved, Integer.MAX_VALUE));

3
test/unit/org/apache/cassandra/db/RemoveColumnFamilyWithFlush1Test.java
View File

@ -24,6 +24,7 @@ import java.util.concurrent.ExecutionException;
import org.junit.Test;
import static junit.framework.Assert.assertNull;
import org.apache.cassandra.db.filter.IdentityQueryFilter;
public class RemoveColumnFamilyWithFlush1Test
{
@ -46,7 +47,7 @@ public class RemoveColumnFamilyWithFlush1Test
rm.delete("Standard1", 1);
rm.apply();
ColumnFamily retrieved = store.getColumnFamily("key1", "Standard1", new IdentityFilter());
ColumnFamily retrieved = store.getColumnFamily(new IdentityQueryFilter("key1", "Standard1"));
assert retrieved.isMarkedForDelete();
assertNull(retrieved.getColumn("Column1"));
assertNull(ColumnFamilyStore.removeDeleted(retrieved, Integer.MAX_VALUE));

3
test/unit/org/apache/cassandra/db/RemoveColumnFamilyWithFlush2Test.java
View File

@ -24,6 +24,7 @@ import java.util.concurrent.ExecutionException;
import org.junit.Test;
import static junit.framework.Assert.assertNull;
import org.apache.cassandra.db.filter.IdentityQueryFilter;
public class RemoveColumnFamilyWithFlush2Test
{
@ -44,7 +45,7 @@ public class RemoveColumnFamilyWithFlush2Test
rm.apply();
store.forceBlockingFlush();
ColumnFamily retrieved = store.getColumnFamily("key1", "Standard1:Column1", new IdentityFilter());
ColumnFamily retrieved = store.getColumnFamily(new IdentityQueryFilter("key1", "Standard1:Column1"));
assert retrieved.isMarkedForDelete();
assertNull(retrieved.getColumn("Column1"));
assertNull(ColumnFamilyStore.removeDeleted(retrieved, Integer.MAX_VALUE));

5
test/unit/org/apache/cassandra/db/RemoveColumnTest.java
View File

@ -24,6 +24,8 @@ import java.util.concurrent.ExecutionException;
import org.junit.Test;
import static junit.framework.Assert.assertNull;
import org.apache.cassandra.db.filter.IdentityQueryFilter;
import org.apache.cassandra.db.filter.NamesQueryFilter;
public class RemoveColumnTest
{
@ -45,8 +47,9 @@ public class RemoveColumnTest
rm.delete("Standard1:Column1", 1);
rm.apply();
ColumnFamily retrieved = store.getColumnFamily("key1", "Standard1", new IdentityFilter());
ColumnFamily retrieved = store.getColumnFamily(new NamesQueryFilter("key1", "Standard1", "Column1"));
assert retrieved.getColumn("Column1").isMarkedForDelete();
assertNull(ColumnFamilyStore.removeDeleted(retrieved, Integer.MAX_VALUE));
assertNull(ColumnFamilyStore.removeDeleted(store.getColumnFamily(new IdentityQueryFilter("key1", "Standard1")), Integer.MAX_VALUE));
}
}

3
test/unit/org/apache/cassandra/db/RemoveSubColumnTest.java
View File

@ -24,6 +24,7 @@ import java.util.concurrent.ExecutionException;
import org.junit.Test;
import static junit.framework.Assert.assertNull;
import org.apache.cassandra.db.filter.IdentityQueryFilter;
public class RemoveSubColumnTest
{
@ -45,7 +46,7 @@ public class RemoveSubColumnTest
rm.delete("Super1:SC1:Column1", 1);
rm.apply();
ColumnFamily retrieved = store.getColumnFamily("key1", "Super1:SC1", new IdentityFilter());
ColumnFamily retrieved = store.getColumnFamily(new IdentityQueryFilter("key1", "Super1:SC1"));
assert retrieved.getColumn("SC1").getSubColumn("Column1").isMarkedForDelete();
assertNull(ColumnFamilyStore.removeDeleted(retrieved, Integer.MAX_VALUE));
}

23
test/unit/org/apache/cassandra/db/RemoveSuperColumnTest.java
View File

@ -24,12 +24,13 @@ import java.util.concurrent.Future;
import java.util.List;
import java.util.Collection;
import java.util.Arrays;
import java.util.TreeSet;
import org.apache.commons.lang.StringUtils;
import org.junit.Test;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertEquals;
import org.apache.cassandra.db.filter.IdentityQueryFilter;
import org.apache.cassandra.db.filter.NamesQueryFilter;
public class RemoveSuperColumnTest
@ -65,24 +66,17 @@ public class RemoveSuperColumnTest
private void validateRemoveTwoSources() throws IOException
{
ColumnFamilyStore store = Table.open("Table1").getColumnFamilyStore("Super1");
List<ColumnFamily> families = store.getColumnFamilies("key1", "Super1", new IdentityFilter());
assert families.size() == 2 : StringUtils.join(families, ", ");
assert families.get(0).getAllColumns().first().getMarkedForDeleteAt() == 1; // delete marker, just added
assert !families.get(1).getAllColumns().first().isMarkedForDelete(); // flushed old version
ColumnFamily resolved = ColumnFamily.resolve(families);
ColumnFamily resolved = store.getColumnFamily(new NamesQueryFilter("key1", "Super1", "SC1"));
assert resolved.getAllColumns().first().getMarkedForDeleteAt() == 1;
Collection<IColumn> subColumns = resolved.getAllColumns().first().getSubColumns();
assert subColumns.size() == 1;
assert subColumns.iterator().next().timestamp() == 0;
assert resolved.getAllColumns().first().getSubColumns().size() == 0;
assertNull(ColumnFamilyStore.removeDeleted(resolved, Integer.MAX_VALUE));
assertNull(ColumnFamilyStore.removeDeleted(store.getColumnFamily(new IdentityQueryFilter("key1", "Super1")), Integer.MAX_VALUE));
}
private void validateRemoveCompacted() throws IOException
{
ColumnFamilyStore store = Table.open("Table1").getColumnFamilyStore("Super1");
List<ColumnFamily> families = store.getColumnFamilies("key1", "Super1", new IdentityFilter());
assert families.size() == 1 : StringUtils.join(families, ", ");
ColumnFamily resolved = families.get(0);
ColumnFamily resolved = store.getColumnFamily(new NamesQueryFilter("key1", "Super1", "SC1"));
assert resolved.getAllColumns().first().getMarkedForDeleteAt() == 1;
Collection<IColumn> subColumns = resolved.getAllColumns().first().getSubColumns();
assert subColumns.size() == 0;
@ -124,10 +118,7 @@ public class RemoveSuperColumnTest
private void validateRemoveWithNewData() throws IOException
{
ColumnFamilyStore store = Table.open("Table1").getColumnFamilyStore("Super2");
List<ColumnFamily> families;
ColumnFamily resolved;
resolved = store.getColumnFamily(new NamesQueryFilter("key1", "Super2:SC1", "Column2"));
ColumnFamily resolved = store.getColumnFamily(new NamesQueryFilter("key1", "Super2:SC1", "Column2"));
validateNewDataFamily(resolved);
}

8
test/unit/org/apache/cassandra/db/TableTest.java
View File

@ -26,6 +26,7 @@ import org.junit.Test;
import static junit.framework.Assert.*;
import org.apache.cassandra.CleanupHelper;
import org.apache.cassandra.db.filter.NamesQueryFilter;
import org.apache.cassandra.io.SSTableReader;
public class TableTest extends CleanupHelper
@ -71,10 +72,10 @@ public class TableTest extends CleanupHelper
{
Row result;
result = table.getRow(TEST_KEY, "Standard1:col1");
result = table.getRow(TEST_KEY, new NamesQueryFilter(TEST_KEY, "Standard1", "col1"));
assertColumns(result.getColumnFamily("Standard1"), "col1");
result = table.getRow(TEST_KEY, "Standard1:col3");
result = table.getRow(TEST_KEY, new NamesQueryFilter(TEST_KEY, "Standard1", "col3"));
assertColumns(result.getColumnFamily("Standard1"), "col3");
}
};
@ -367,8 +368,7 @@ public class TableTest extends CleanupHelper
public static void assertColumns(ColumnFamily columnFamily, String... columnNames)
{
assertNotNull(columnFamily);
SortedSet<IColumn> columns = columnFamily.getAllColumns();
SortedSet<IColumn> columns = columnFamily == null ? new TreeSet<IColumn>() : columnFamily.getAllColumns();
List<String> L = new ArrayList<String>();
for (IColumn column : columns)
{