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

r/m time-sorted columns. 

patch by jbellis; reviewed by Sandeep Tata for CASSANDRA-185

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@796107 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2009-07-21 01:36:10 +00:00
parent 7435fc82c0
commit 8ff63a92ba
21 changed files with 87 additions and 1649 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
interface/cassandra.thrift
View File

@ -134,9 +134,6 @@ service Cassandra {
void remove(1:string table, 2:string key, 3:ColumnPathOrParent column_path_or_parent, 4:i64 timestamp, 5:i32 block_for=0)
throws (1: InvalidRequestException ire, 2: UnavailableException ue),
list<Column> get_columns_since(1:string table, 2:string key, 3:ColumnParent column_parent, 4:i64 timeStamp)
throws (1: InvalidRequestException ire, 2: NotFoundException nfe),
list<SuperColumn> get_slice_super(1:string table, 2:string key, 3:string column_family, 4:string start, 5:string finish, 6:bool is_ascending, 7:i32 count=100)
throws (1: InvalidRequestException ire),

1006
interface/gen-java/org/apache/cassandra/service/Cassandra.java
View File

File diff suppressed because it is too large Load Diff

21
src/java/org/apache/cassandra/config/DatabaseDescriptor.java
View File

@ -587,17 +587,6 @@ public class DatabaseDescriptor
return cfMetaData.flushPeriodInMinutes;
}
public static boolean isNameSortingEnabled(String tableName, String cfName)
{
assert tableName != null;
CFMetaData cfMetaData = getCFMetaData(tableName, cfName);
if (cfMetaData == null)
return false;
return "Name".equals(cfMetaData.indexProperty_);
}
public static boolean isTimeSortingEnabled(String tableName, String cfName)
{
assert tableName != null;
@ -773,15 +762,7 @@ public class DatabaseDescriptor
public static ColumnComparatorFactory.ComparatorType getTypeInfo(String tableName, String cfName)
{
assert tableName != null;
CFMetaData cfMetadata = DatabaseDescriptor.getCFMetaData(tableName, cfName);
if ( cfMetadata.indexProperty_.equals("Name") )
{
return ColumnComparatorFactory.ComparatorType.NAME;
}
else
{
return ColumnComparatorFactory.ComparatorType.TIMESTAMP;
}
return ColumnComparatorFactory.ComparatorType.NAME;
}
public static Map<String, Map<String, CFMetaData>> getTableToColumnFamilyMap()

38
src/java/org/apache/cassandra/db/ColumnComparatorFactory.java
View File

@ -31,18 +31,14 @@ public class ColumnComparatorFactory
public static enum ComparatorType
{
NAME,
TIMESTAMP
}
public static final Comparator<IColumn> nameComparator_ = new ColumnNameComparator();
public static final Comparator<IColumn> timestampComparator_ = new ColumnTimestampComparator();
public static Comparator<IColumn> getComparator(ComparatorType comparatorType)
{
if (comparatorType == ComparatorType.NAME)
return nameComparator_;
assert comparatorType == ComparatorType.TIMESTAMP;
return timestampComparator_;
assert comparatorType == ComparatorType.NAME;
return nameComparator_;
}
public static Comparator<IColumn> getComparator(int comparatorTypeInt)
@ -67,36 +63,6 @@ abstract class AbstractColumnComparator implements Comparator<IColumn>, Serializ
}
}
class ColumnTimestampComparator extends AbstractColumnComparator
{
ColumnTimestampComparator()
{
super(ColumnComparatorFactory.ComparatorType.TIMESTAMP);
}
/* if the time-stamps are the same then sort by names */
public int compare(IColumn column1, IColumn column2)
{
assert column1.getClass() == column2.getClass();
/* inverse sort by time to get hte latest first */
long result = column2.timestamp() - column1.timestamp();
int finalResult = 0;
if (result == 0)
{
result = column1.name().compareTo(column2.name());
}
if (result > 0)
{
finalResult = 1;
}
if (result < 0)
{
finalResult = -1;
}
return finalResult;
}
}
class ColumnNameComparator extends AbstractColumnComparator
{
ColumnNameComparator()

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

@ -96,17 +96,8 @@ public final class ColumnFamily
public static ColumnFamily create(String tableName, String cfName)
{
Comparator<IColumn> comparator;
String columnType = DatabaseDescriptor.getColumnFamilyType(tableName, cfName);
if (DatabaseDescriptor.isNameSortingEnabled(tableName, cfName))
{
comparator = ColumnComparatorFactory.getComparator(ColumnComparatorFactory.ComparatorType.NAME);
}
/* if this columnfamily has simple columns, and no index on name sort by timestamp */
else
{
comparator = ColumnComparatorFactory.getComparator(ColumnComparatorFactory.ComparatorType.TIMESTAMP);
}
Comparator<IColumn> comparator = ColumnComparatorFactory.getComparator(ColumnComparatorFactory.ComparatorType.NAME);
return new ColumnFamily(cfName, columnType, comparator);
}
@ -343,9 +334,7 @@ public final class ColumnFamily
public ColumnComparatorFactory.ComparatorType getComparatorType()
{
return getComparator() == ColumnComparatorFactory.nameComparator_
? ColumnComparatorFactory.ComparatorType.NAME
: ColumnComparatorFactory.ComparatorType.TIMESTAMP;
return ColumnComparatorFactory.ComparatorType.NAME;
}
int size()

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

@ -1416,11 +1416,6 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
return getColumnFamily(new SliceQueryFilter(key, path, start, finish, isAscending, limit));
}
public ColumnFamily getColumnFamily(String key, QueryPath columnParent, long since) throws IOException
{
return getColumnFamily(new TimeQueryFilter(key, columnParent, since));
}
public ColumnFamily getColumnFamily(QueryFilter filter) throws IOException
{
return getColumnFamily(filter, getDefaultGCBefore());

104
src/java/org/apache/cassandra/db/ColumnsSinceReadCommand.java
View File

@ -1,104 +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.DataOutputStream;
import java.io.IOException;
import org.apache.cassandra.service.ColumnParent;
import org.apache.cassandra.db.filter.QueryPath;
import org.apache.cassandra.db.filter.QueryFilter;
import org.apache.cassandra.db.filter.TimeQueryFilter;
public class ColumnsSinceReadCommand extends ReadCommand
{
public final QueryPath columnParent;
public final long sinceTimestamp;
public ColumnsSinceReadCommand(String table, String key, ColumnParent column_parent, long sinceTimestamp)
{
this(table, key, new QueryPath(column_parent), sinceTimestamp);
}
public ColumnsSinceReadCommand(String table, String key, QueryPath columnParent, long sinceTimestamp)
{
super(table, key, CMD_TYPE_GET_COLUMNS_SINCE);
this.columnParent = columnParent;
this.sinceTimestamp = sinceTimestamp;
}
@Override
public String getColumnFamilyName()
{
return columnParent.columnFamilyName;
}
@Override
public ReadCommand copy()
{
ReadCommand readCommand = new ColumnsSinceReadCommand(table, key, columnParent, sinceTimestamp);
readCommand.setDigestQuery(isDigestQuery());
return readCommand;
}
@Override
public Row getRow(Table table) throws IOException
{
return table.getRow(new TimeQueryFilter(key, columnParent, sinceTimestamp));
}
@Override
public String toString()
{
return "ColumnsSinceReadCommand(" +
"table='" + table + '\'' +
", key='" + key + '\'' +
", columnParent='" + columnParent + '\'' +
", sinceTimestamp='" + sinceTimestamp + '\'' +
')';
}
}
class ColumnsSinceReadCommandSerializer extends ReadCommandSerializer
{
@Override
public void serialize(ReadCommand rm, DataOutputStream dos) throws IOException
{
ColumnsSinceReadCommand realRM = (ColumnsSinceReadCommand)rm;
dos.writeBoolean(realRM.isDigestQuery());
dos.writeUTF(realRM.table);
dos.writeUTF(realRM.key);
realRM.columnParent.serialize(dos);
dos.writeLong(realRM.sinceTimestamp);
}
@Override
public ReadCommand deserialize(DataInputStream dis) throws IOException
{
boolean isDigest = dis.readBoolean();
String table = dis.readUTF();
String key = dis.readUTF();
QueryPath columnParent = QueryPath.deserialize(dis);
long sinceTimestamp = dis.readLong();
ColumnsSinceReadCommand rm = new ColumnsSinceReadCommand(table, key, columnParent, sinceTimestamp);
rm.setDigestQuery(isDigest);
return rm;
}
}

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

@ -353,39 +353,7 @@ public class Memtable implements Comparable<Memtable>
}
};
}
public ColumnIterator getTimeIterator(final TimeQueryFilter filter)
{
final ColumnFamily cf = columnFamilies_.get(filter.key);
final ColumnFamily columnFamily = cf == null ? ColumnFamily.create(table_, filter.getColumnFamilyName()) : cf.cloneMeShallow();
return new SimpleAbstractColumnIterator()
{
private Iterator<IColumn> iter = cf == null ? null : cf.getAllColumns().iterator();
public ColumnFamily getColumnFamily()
{
return columnFamily;
}
protected IColumn computeNext()
{
if (iter == null)
{
return endOfData();
}
while (iter.hasNext())
{
IColumn column = iter.next();
if (column.timestamp() < filter.since)
break;
return column;
}
return endOfData();
}
};
}
void clearUnsafe()
{
columnFamilies_.clear();

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

@ -34,8 +34,7 @@ public abstract class ReadCommand
{
public static final String DO_REPAIR = "READ-REPAIR";
public static final byte CMD_TYPE_GET_SLICE_BY_NAMES = 1;
public static final byte CMD_TYPE_GET_COLUMNS_SINCE = 2;
public static final byte CMD_TYPE_GET_SLICE = 3;
public static final byte CMD_TYPE_GET_SLICE = 2;
public static final String EMPTY_CF = "";
@ -89,7 +88,6 @@ class ReadCommandSerializer implements ICompactSerializer<ReadCommand>
static
{
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, new SliceFromReadCommandSerializer());
}

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

@ -95,13 +95,6 @@ public class RowMutation implements Serializable
modifications_ = modifications;
}
/** trailing empty patch fragments ("" or "CF:") will be removed,
* so caller doesn't have to check for those */
public static String[] getColumnAndColumnFamily(String cf)
{
return cf.split(":");
}
public String table()
{
return table_;

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

@ -24,7 +24,9 @@ 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 java.util.concurrent.ConcurrentSkipListMap;
import org.apache.commons.lang.ArrayUtils;
import org.apache.commons.lang.StringUtils;
@ -41,7 +43,6 @@ public final class SuperColumn implements IColumn, Serializable
{
private static Logger logger_ = Logger.getLogger(SuperColumn.class);
private static SuperColumnSerializer serializer_ = new SuperColumnSerializer();
private final static String seperator_ = ":";
static SuperColumnSerializer serializer()
{
@ -49,7 +50,7 @@ public final class SuperColumn implements IColumn, Serializable
}
private String name_;
private EfficientBidiMap columns_ = new EfficientBidiMap(ColumnComparatorFactory.getComparator(ColumnComparatorFactory.ComparatorType.TIMESTAMP));
private ConcurrentSkipListMap<String, IColumn> columns_ = new ConcurrentSkipListMap<String, IColumn>();
private int localDeletionTime = Integer.MIN_VALUE;
private long markedForDeleteAt = Long.MIN_VALUE;
private AtomicInteger size_ = new AtomicInteger(0);
@ -82,7 +83,7 @@ public final class SuperColumn implements IColumn, Serializable
public Collection<IColumn> getSubColumns()
{
return columns_.getSortedColumns();
return columns_.values();
}
public IColumn getSubColumn(String columnName)
@ -277,12 +278,11 @@ public final class SuperColumn implements IColumn, Serializable
public byte[] digest()
{
Set<IColumn> columns = columns_.getSortedColumns();
byte[] xorHash = ArrayUtils.EMPTY_BYTE_ARRAY;
if(name_ == null)
return xorHash;
xorHash = name_.getBytes();
for(IColumn column : columns)
for(IColumn column : columns_.values())
{
xorHash = FBUtilities.xor(xorHash, column.digest());
}

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

@ -517,9 +517,7 @@ public class Table
@Deprecated // single CFs could be larger than memory
public ColumnFamily get(String key, String cfName) throws IOException
{
assert !cfName.contains(":") : cfName;
String[] values = RowMutation.getColumnAndColumnFamily(cfName);
ColumnFamilyStore cfStore = columnFamilyStores_.get(values[0]);
ColumnFamilyStore cfStore = columnFamilyStores_.get(cfName);
assert cfStore != null : "Column family " + cfName + " has not been defined";
return cfStore.getColumnFamily(new IdentityQueryFilter(key, new QueryPath(cfName)));
}

48
src/java/org/apache/cassandra/db/filter/SSTableTimeIterator.java
View File

@ -1,48 +0,0 @@
package org.apache.cassandra.db.filter;
import java.util.Iterator;
import java.io.IOException;
import org.apache.cassandra.db.ColumnFamily;
import org.apache.cassandra.db.IColumn;
import org.apache.cassandra.io.SSTableReader;
import org.apache.cassandra.io.DataInputBuffer;
import org.apache.cassandra.io.IndexHelper;
public class SSTableTimeIterator extends SimpleAbstractColumnIterator
{
private ColumnFamily cf;
private Iterator<IColumn> iter;
public final long since;
public SSTableTimeIterator(String filename, String key, String cfName, long since) throws IOException
{
this.since = since;
SSTableReader ssTable = SSTableReader.open(filename);
DataInputBuffer buffer = ssTable.next(key, cfName, null, new IndexHelper.TimeRange(since, Long.MAX_VALUE));
if (buffer.getLength() > 0)
{
cf = ColumnFamily.serializer().deserialize(buffer);
iter = cf.getAllColumns().iterator();
}
}
public ColumnFamily getColumnFamily()
{
return cf;
}
protected IColumn computeNext()
{
if (iter == null)
return endOfData();
while (iter.hasNext())
{
IColumn c = iter.next();
if (c.timestamp() < since)
break;
return c;
}
return endOfData();
}
}

55
src/java/org/apache/cassandra/db/filter/TimeQueryFilter.java
View File

@ -1,55 +0,0 @@
package org.apache.cassandra.db.filter;
import java.io.IOException;
import java.util.Comparator;
import org.apache.cassandra.db.*;
import org.apache.cassandra.io.SSTableReader;
import org.apache.cassandra.utils.ReducingIterator;
public class TimeQueryFilter extends QueryFilter
{
public final long since;
public TimeQueryFilter(String key, QueryPath columnParent, long since)
{
super(key, columnParent);
this.since = since;
}
public ColumnIterator getMemColumnIterator(Memtable memtable)
{
return memtable.getTimeIterator(this);
}
public ColumnIterator getSSTableColumnIterator(SSTableReader sstable) throws IOException
{
return new SSTableTimeIterator(sstable.getFilename(), key, getColumnFamilyName(), since);
}
@Override
public Comparator<IColumn> getColumnComparator()
{
return ColumnComparatorFactory.timestampComparator_;
}
public void collectColumns(ColumnFamily returnCF, ReducingIterator<IColumn> reducedColumns, int gcBefore)
{
for (IColumn column : reducedColumns)
{
if (!column.isMarkedForDelete() || column.getLocalDeletionTime() > gcBefore)
returnCF.addColumn(column);
}
}
public void filterSuperColumn(SuperColumn superColumn)
{
for (IColumn column : superColumn.getSubColumns())
{
if (column.timestamp() < since)
{
superColumn.remove(column.name());
}
}
}
}

8
src/java/org/apache/cassandra/io/IFileReader.java
View File

@ -72,14 +72,8 @@ public interface IFileReader
* @param columnFamilyName The name of the column family only without the ":"
* @param columnNames - The list of columns in the cfName column family
* that we want to return
* OR
* @param timeRange - time range we are interested in
* @param position
* @throws IOException
* @return number of bytes read.
*
*/
public long next(String key, DataOutputBuffer bufOut, String columnFamilyName, SortedSet<String> columnNames, IndexHelper.TimeRange timeRange, long position) throws IOException;
public long next(String key, DataOutputBuffer bufOut, String columnFamilyName, SortedSet<String> columnNames, long position) throws IOException;
/**
* Close the file after reading.

121
src/java/org/apache/cassandra/io/IndexHelper.java
View File

@ -246,93 +246,7 @@ public class IndexHelper
return columnRanges;
}
/**
* Returns the range in which a given column falls in the index. This
* is used when time range queries are in play. For instance if we are
* looking for columns in the range [t, t2]
* @param cIndexInfo the time we are interested in.
* @param columnIndexList the in-memory representation of the column index
* @param dataSize the total size of the data
* @param totalNumCols total number of columns
* @return an object describing a subrange in which the column is serialized
*/
static ColumnRange getColumnRangeFromTimeIndex(IndexHelper.TimeRange timeRange, List<IndexHelper.ColumnIndexInfo> columnIndexList, int dataSize, int totalNumCols)
{
/* if column indexes were not present for this column family, the handle accordingly */
if(columnIndexList.size() == 0)
{
return new ColumnRange(0, dataSize, totalNumCols);
}
/* find the offset for the column */
int size = columnIndexList.size();
long start = 0;
long end = dataSize;
int numColumns = 0;
/*
* Time indices are sorted in descending order. So
* we need to apply a reverse comparator for the
* binary search.
*/
Comparator<IndexHelper.ColumnIndexInfo> comparator = Collections.reverseOrder();
IndexHelper.ColumnIndexInfo rhs = IndexHelper.ColumnIndexFactory.instance(ColumnComparatorFactory.ComparatorType.TIMESTAMP);
rhs.set(timeRange.rhs());
int index = Collections.binarySearch(columnIndexList, rhs, comparator);
if ( index < 0 )
{
/* We are here which means that the requested column is not an index. */
index = (++index)*(-1);
}
else
{
++index;
}
/*
* Calculate the starting offset from which we have to read. So
* we achieve this by performing the probe using the bigger timestamp
* and then scanning the column position chunks till we reach the
* lower timestamp in the time range.
*/
start = (index == 0) ? 0 : columnIndexList.get(index - 1).position();
/* add the number of columns in the first chunk. */
numColumns += (index ==0) ? columnIndexList.get(0).count() : columnIndexList.get(index - 1).count();
if( index < size )
{
int chunks = columnIndexList.size();
/* Index info for the lower bound of the time range */
IndexHelper.ColumnIndexInfo lhs = IndexHelper.ColumnIndexFactory.instance(ColumnComparatorFactory.ComparatorType.TIMESTAMP);
lhs.set(timeRange.lhs());
int i = index + 1;
for ( ; i < chunks; ++i )
{
IndexHelper.ColumnIndexInfo cIndexInfo2 = columnIndexList.get(i);
if ( cIndexInfo2.compareTo(lhs) < 0 )
{
numColumns += cIndexInfo2.count();
break;
}
numColumns += cIndexInfo2.count();
}
end = columnIndexList.get(i).position();
}
else
{
end = dataSize;
int totalColsIndexed = 0;
for( IndexHelper.ColumnIndexInfo colPosInfo : columnIndexList )
{
totalColsIndexed += colPosInfo.count();
}
numColumns = totalNumCols - totalColsIndexed;
}
return new ColumnRange(start, end, numColumns);
}
public static class ColumnIndexFactory
{
public static ColumnIndexInfo instance(ColumnComparatorFactory.ComparatorType typeInfo)
@ -341,37 +255,8 @@ public class IndexHelper
? new ColumnNameIndexInfo() : new ColumnTimestampIndexInfo();
}
}
/**
* Encapsulates a time range. Queries use
* this abstraction for indicating start
* and end regions of a time filter.
*
* @author alakshman
*
*/
public static class TimeRange
{
private long lhs_;
private long rhs_;
public TimeRange(long lhs, long rhs)
{
lhs_ = lhs;
rhs_ = rhs;
}
public long lhs()
{
return lhs_;
}
public long rhs()
{
return rhs_;
}
}
/**
* A column range containing the start and end
* offset of the appropriate column index chunk

7
src/java/org/apache/cassandra/io/SSTableReader.java
View File

@ -286,11 +286,6 @@ public class SSTableReader extends SSTable
}
public DataInputBuffer next(final String clientKey, String cfName, SortedSet<String> columnNames) throws IOException
{
return next(clientKey, cfName, columnNames, null);
}
public DataInputBuffer next(final String clientKey, String cfName, SortedSet<String> columnNames, IndexHelper.TimeRange timeRange) throws IOException
{
IFileReader dataReader = null;
try
@ -301,7 +296,7 @@ public class SSTableReader extends SSTable
DataOutputBuffer bufOut = new DataOutputBuffer();
DataInputBuffer bufIn = new DataInputBuffer();
long bytesRead = dataReader.next(decoratedKey, bufOut, cfName, columnNames, timeRange, position);
long bytesRead = dataReader.next(decoratedKey, bufOut, cfName, columnNames, position);
if (bytesRead != -1L)
{
if (bufOut.getLength() > 0)

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

@ -426,15 +426,8 @@ public class SequenceFile
if (hasColumnIndexes)
{
String tableName = getTableName();
if (DatabaseDescriptor.isNameSortingEnabled(tableName, cfName))
{
/* read the index */
totalBytesRead += IndexHelper.deserializeIndex(tableName, cfName, file_, columnIndexList);
}
else
{
totalBytesRead += IndexHelper.skipIndex(file_);
}
/* read the index */
totalBytesRead += IndexHelper.deserializeIndex(tableName, cfName, file_, columnIndexList);
}
return totalBytesRead;
}
@ -476,15 +469,10 @@ public class SequenceFile
* @param bufOut DataOutputStream that needs to be filled.
* @param columnFamilyName name of the columnFamily
* @param columnNames columnNames we are interested in
* OR
* @param timeRange time range we are interested in
* @param position
* @return number of bytes that were read.
* @throws IOException
*/
public long next(String key, DataOutputBuffer bufOut, String columnFamilyName, SortedSet<String> columnNames, IndexHelper.TimeRange timeRange, long position) throws IOException
public long next(String key, DataOutputBuffer bufOut, String columnFamilyName, SortedSet<String> columnNames, long position) throws IOException
{
assert timeRange == null || columnNames == null; // at most one may be non-null
assert columnNames != null;
long bytesRead = -1L;
if (isEOF() || seekTo(position) < 0)
@ -510,11 +498,7 @@ public class SequenceFile
*/
if (keyInDisk.equals(key))
{
if (timeRange == null) {
readColumns(key, bufOut, columnFamilyName, columnNames);
} else {
readTimeRange(key, bufOut, columnFamilyName, timeRange);
}
readColumns(key, bufOut, columnFamilyName, columnNames);
}
else
{
@ -530,70 +514,6 @@ public class SequenceFile
return bytesRead;
}
private void readTimeRange(String key, DataOutputBuffer bufOut, String columnFamilyName, IndexHelper.TimeRange timeRange)
throws IOException
{
int dataSize = file_.readInt();
/* write the key into buffer */
bufOut.writeUTF(key);
int bytesSkipped = IndexHelper.skipBloomFilter(file_);
/*
* read the correct number of bytes for the column family and
* write data into buffer. Subtract from dataSize the bloom
* filter size.
*/
dataSize -= bytesSkipped;
List<IndexHelper.ColumnIndexInfo> columnIndexList = new ArrayList<IndexHelper.ColumnIndexInfo>();
/* Read the times indexes if present */
int totalBytesRead = handleColumnTimeIndexes(columnFamilyName, columnIndexList);
dataSize -= totalBytesRead;
/* read the column family name */
String cfName = file_.readUTF();
dataSize -= (utfPrefix_ + cfName.length());
String cfType = file_.readUTF();
dataSize -= (utfPrefix_ + cfType.length());
int indexType = file_.readInt();
dataSize -= 4;
/* read local deletion time */
int localDeletionTime = file_.readInt();
dataSize -=4;
/* read if this cf is marked for delete */
long markedForDeleteAt = file_.readLong();
dataSize -= 8;
/* read the total number of columns */
int totalNumCols = file_.readInt();
dataSize -= 4;
/* get the column range we have to read */
IndexHelper.ColumnRange columnRange = IndexHelper.getColumnRangeFromTimeIndex(timeRange, columnIndexList, dataSize, totalNumCols);
Coordinate coordinate = columnRange.coordinate();
/* seek to the correct offset to the data, and calculate the data size */
file_.skipBytes((int) coordinate.start_);
dataSize = (int) (coordinate.end_ - coordinate.start_);
// returned data size
bufOut.writeInt(dataSize + utfPrefix_ * 2 + cfName.length() + cfType.length() + 4 + 4 + 8 + 4);
// echo back the CF data we read
bufOut.writeUTF(cfName);
bufOut.writeUTF(cfType);
bufOut.writeInt(indexType);
bufOut.writeInt(localDeletionTime);
bufOut.writeLong(markedForDeleteAt);
/* write number of columns */
bufOut.writeInt(columnRange.count());
/* now write the columns */
bufOut.write(file_, dataSize);
}
private void readColumns(String key, DataOutputBuffer bufOut, String columnFamilyName, SortedSet<String> cNames)
throws IOException
{

19
src/java/org/apache/cassandra/service/CassandraServer.java
View File

@ -133,15 +133,6 @@ public class CassandraServer implements Cassandra.Iface
return thriftifyColumns(cfamily.getAllColumns());
}
public List<Column> get_columns_since(String table, String key, ColumnParent column_parent, long timeStamp)
throws InvalidRequestException, NotFoundException
{
logger.debug("get_columns_since");
ThriftValidation.validateColumnParent(table, column_parent);
return getSlice(new ColumnsSinceReadCommand(table, key, column_parent, timeStamp));
}
public List<Column> get_slice_by_names(String table, String key, ColumnParent column_parent, List<String> column_names)
throws InvalidRequestException, NotFoundException
{
@ -221,15 +212,7 @@ public class CassandraServer implements Cassandra.Iface
}
ColumnFamily cfamily;
if (DatabaseDescriptor.isNameSortingEnabled(table, column_parent.column_family)
&& column_parent.super_column == null)
{
cfamily = readColumnFamily(new SliceFromReadCommand(table, key, column_parent, "", "", true, Integer.MAX_VALUE));
}
else
{
cfamily = readColumnFamily(new ColumnsSinceReadCommand(table, key, column_parent, Long.MIN_VALUE));
}
cfamily = readColumnFamily(new SliceFromReadCommand(table, key, column_parent, "", "", true, Integer.MAX_VALUE));
if (cfamily == null)
{
return 0;

4
test/unit/org/apache/cassandra/db/ReadMessageTest.java
View File

@ -46,10 +46,6 @@ public class ReadMessageTest
rm2 = serializeAndDeserializeReadMessage(rm);
assert rm2.toString().equals(rm.toString());
rm = new ColumnsSinceReadCommand("Table1", "row1", new QueryPath("foo"), 1);
rm2 = serializeAndDeserializeReadMessage(rm);
assert rm2.toString().equals(rm.toString());
rm = new SliceFromReadCommand("Table1", "row1", new QueryPath("foo"), "", "", true, 2);
rm2 = serializeAndDeserializeReadMessage(rm);
assert rm2.toString().equals(rm.toString());

133
test/unit/org/apache/cassandra/db/TimeSortTest.java
View File

@ -1,133 +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.IOException;
import java.util.concurrent.ExecutionException;
import java.util.SortedSet;
import java.util.Iterator;
import org.apache.commons.lang.ArrayUtils;
import org.junit.Test;
import org.apache.cassandra.CleanupHelper;
import org.apache.cassandra.db.filter.QueryPath;
public class TimeSortTest extends CleanupHelper
{
@Test
public void testMixedSources() throws IOException, ExecutionException, InterruptedException
{
Table table = Table.open("Table1");
ColumnFamilyStore cfStore = table.getColumnFamilyStore("StandardByTime1");
RowMutation rm;
rm = new RowMutation("Table1", "key0");
rm.add(new QueryPath("StandardByTime1", null, "C0"), "a".getBytes(), 100);
rm.apply();
cfStore.forceBlockingFlush();
rm = new RowMutation("Table1", "key0");
rm.add(new QueryPath("StandardByTime1", null, "C1"), "b".getBytes(), 0);
rm.apply();
ColumnFamily cf = cfStore.getColumnFamily("key0", new QueryPath("StandardByTime1"), 10);
SortedSet<IColumn> columns = cf.getAllColumns();
assert columns.size() == 1;
}
@Test
public void testTimeSort() throws IOException, ExecutionException, InterruptedException
{
Table table = Table.open("Table1");
ColumnFamilyStore cfStore = table.getColumnFamilyStore("StandardByTime1");
for (int i = 900; i < 1000; ++i)
{
String key = Integer.toString(i);
RowMutation rm = new RowMutation("Table1", key);
for (int j = 0; j < 8; ++j)
{
byte[] bytes = j % 2 == 0 ? "a".getBytes() : "b".getBytes();
rm.add(new QueryPath("StandardByTime1", null, "Column-" + j), bytes, j * 2);
}
rm.apply();
}
validateTimeSort(table);
cfStore.forceBlockingFlush();
validateTimeSort(table);
// interleave some new data to test memtable + sstable
String key = "900";
RowMutation rm = new RowMutation("Table1", key);
for (int j = 0; j < 4; ++j)
{
rm.add(new QueryPath("StandardByTime1", null, "Column+" + j), ArrayUtils.EMPTY_BYTE_ARRAY, j * 2 + 1);
}
rm.apply();
// and some overwrites
rm = new RowMutation("Table1", key);
for (int j = 4; j < 8; ++j)
{
rm.add(new QueryPath("StandardByTime1", null, "Column-" + j), ArrayUtils.EMPTY_BYTE_ARRAY, j * 3);
}
rm.apply();
// verify
ColumnFamily cf = cfStore.getColumnFamily(key, new QueryPath("StandardByTime1"), 0);
SortedSet<IColumn> columns = cf.getAllColumns();
assert columns.size() == 12;
Iterator<IColumn> iter = columns.iterator();
IColumn column;
for (int j = 7; j >= 4; j--)
{
column = iter.next();
assert column.name().equals("Column-" + j);
assert column.timestamp() == j * 3;
assert column.value().length == 0;
}
for (int j = 3; j >= 0; j--)
{
column = iter.next();
assert column.name().equals("Column+" + j);
column = iter.next();
assert column.name().equals("Column-" + j);
}
}
private void validateTimeSort(Table table) throws IOException
{
for (int i = 900; i < 1000; ++i)
{
String key = Integer.toString(i);
for (int j = 0; j < 8; j += 3)
{
ColumnFamily cf = table.getColumnFamilyStore("StandardByTime1").getColumnFamily(key, new QueryPath("StandardByTime1"), j * 2);
SortedSet<IColumn> columns = cf.getAllColumns();
assert columns.size() == 8 - j;
int k = 7;
for (IColumn c : columns)
{
assert c.timestamp() == (k--) * 2;
}
}
}
}
}