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

Add get_slice_from functionality using column indexes for efficiency. 

Patch by Jun Rao and jbellis for CASSANDRA-172

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@777578 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2009-05-22 15:33:42 +00:00
parent a432e18077
commit 4ac9118f7c
17 changed files with 1879 additions and 141 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
interface/cassandra.thrift
View File

@ -83,9 +83,12 @@ service Cassandra {
list<column_t> get_slice_by_name_range(1:string tablename, 2:string key, 3:string columnFamily, 4:string start, 5:string end, 6:i32 count=-1)
throws (1: InvalidRequestException ire, 2: NotFoundException nfe),
list<column_t> get_slice_by_names(1:string tablename, 2:string key, 3:string columnFamily, 4:list<string> columnNames)
list<column_t> get_slice_by_names(1:string tablename, 2:string key, 3:string columnFamily, 4:list<string> columnNames)
throws (1: InvalidRequestException ire, 2: NotFoundException nfe),
list<column_t> get_slice_from(1:string tablename, 2:string key, 3:string columnFamily_column, 4:bool isAscending, 5:i32 count)
throws (1: InvalidRequestException ire, 2: NotFoundException nfe),
column_t get_column(1:string tablename, 2:string key, 3:string columnFamily_column)
throws (1: InvalidRequestException ire, 2: NotFoundException nfe),

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

File diff suppressed because it is too large Load Diff

BIN
lib/google-collect-1.0-rc1.jar Normal file
View File

Binary file not shown.

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

@ -39,37 +39,15 @@ public class ColumnComparatorFactory
public static Comparator<IColumn> getComparator(ComparatorType comparatorType)
{
Comparator<IColumn> columnComparator = timestampComparator_;
switch (comparatorType)
{
case NAME:
columnComparator = nameComparator_;
break;
case TIMESTAMP:
default:
columnComparator = timestampComparator_;
break;
}
return columnComparator;
if (comparatorType == ComparatorType.NAME)
return nameComparator_;
assert comparatorType == ComparatorType.TIMESTAMP;
return timestampComparator_;
}
public static Comparator<IColumn> getComparator(int comparatorTypeInt)
{
ComparatorType comparatorType = ComparatorType.NAME;
if (comparatorTypeInt == ComparatorType.NAME.ordinal())
{
comparatorType = ComparatorType.NAME;
}
else if (comparatorTypeInt == ComparatorType.TIMESTAMP.ordinal())
{
comparatorType = ComparatorType.TIMESTAMP;
}
return getComparator(comparatorType);
return getComparator(ComparatorType.values()[comparatorTypeInt]);
}
}

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

@ -247,6 +247,7 @@ public final class ColumnFamily
void clear()
{
columns_.clear();
size_.set(0);
}
/*
@ -301,21 +302,26 @@ public final class ColumnFamily
columns_.remove(columnName);
}
void delete(int localtime, long timestamp)
public void delete(int localtime, long timestamp)
{
localDeletionTime = localtime;
markedForDeleteAt = timestamp;
}
public void delete(ColumnFamily cf2)
{
delete(Math.max(getLocalDeletionTime(), cf2.getLocalDeletionTime()),
Math.max(getMarkedForDeleteAt(), cf2.getMarkedForDeleteAt()));
}
public boolean isMarkedForDelete()
{
return markedForDeleteAt > Long.MIN_VALUE;
}
/*
* This function will calculate the differnce between 2 column families
* the external input is considered the superset of internal
* so there are no deletes in the diff.
* This function will calculate the difference between 2 column families.
* The external input is assumed to be a superset of internal.
*/
ColumnFamily diff(ColumnFamily cfComposite)
{
@ -446,8 +452,7 @@ public final class ColumnFamily
{
assert cf.name().equals(cf2.name());
cf.addColumns(cf2);
cf.delete(Math.max(cf.getLocalDeletionTime(), cf2.getLocalDeletionTime()),
Math.max(cf.getMarkedForDeleteAt(), cf2.getMarkedForDeleteAt()));
cf.delete(cf2);
}
return cf;
}

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

@ -45,6 +45,10 @@ import org.apache.cassandra.service.StorageService;
import org.apache.cassandra.utils.*;
import org.apache.cassandra.concurrent.DebuggableThreadPoolExecutor;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.collections.IteratorUtils;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.collections.comparators.ReverseComparator;
import org.cliffc.high_scale_lib.NonBlockingHashMap;
import org.cliffc.high_scale_lib.NonBlockingHashSet;
@ -1567,4 +1571,119 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
{
return readStats_.mean();
}
/**
* get a list of columns starting from a given column, in a specified order
* only the latest version of a column is returned
*/
public ColumnFamily getSliceFrom(String key, String cfName, String startColumn, boolean isAscending, int count)
throws IOException, ExecutionException, InterruptedException
{
lock_.readLock().lock();
try
{
final ColumnFamily returnCF;
ColumnIterator iter;
List<ColumnIterator> iterators = new ArrayList<ColumnIterator>();
/* add the current memtable */
memtableLock_.readLock().lock();
try
{
iter = memtable_.getColumnIterator(key, cfName, isAscending, startColumn);
returnCF = iter.getColumnFamily();
}
finally
{
memtableLock_.readLock().unlock();
}
iterators.add(iter);
/* add the memtables being flushed */
List<Memtable> memtables = getUnflushedMemtables(cfName);
for (Memtable memtable:memtables)
{
iter = memtable.getColumnIterator(key, cfName, isAscending, startColumn);
returnCF.delete(iter.getColumnFamily());
iterators.add(iter);
}
/* add the SSTables on disk */
List<String> files = new ArrayList<String>(ssTables_);
for (String file : files)
{
// If the key is not present in the SSTable's BloomFilter, continue to the next file
if (!SSTable.isKeyInFile(key, file))
continue;
iter = new SSTableColumnIterator(file, key, cfName, startColumn, isAscending);
if (iter.hasNext())
{
returnCF.delete(iter.getColumnFamily());
iterators.add(iter);
}
}
// define a 'reduced' iterator that merges columns w/ the same name, which
// greatly simplifies computing liveColumns in the presence of tombstones.
Comparator<IColumn> comparator = new Comparator<IColumn>()
{
public int compare(IColumn c1, IColumn c2)
{
return c1.name().compareTo(c2.name());
}
};
if (!isAscending)
comparator = new ReverseComparator(comparator);
Iterator collated = IteratorUtils.collatedIterator(comparator, iterators);
if (!collated.hasNext())
return new ColumnFamily(cfName, DatabaseDescriptor.getColumnFamilyType(cfName));
List<IColumn> L = new ArrayList();
CollectionUtils.addAll(L, collated);
ReducingIterator<IColumn> reduced = new ReducingIterator<IColumn>(L.iterator())
{
ColumnFamily curCF = returnCF.cloneMeShallow();
protected Object getKey(IColumn o)
{
return o == null ? null : o.name();
}
public void reduce(IColumn current)
{
curCF.addColumn(current);
}
protected IColumn getReduced()
{
IColumn c = curCF.getAllColumns().first();
curCF.clear();
return c;
}
};
// add unique columns to the CF container
int liveColumns = 0;
for (IColumn column : reduced)
{
if (liveColumns >= count)
{
break;
}
if (!column.isMarkedForDelete())
liveColumns++;
returnCF.addColumn(column);
}
/* close remaining cursors */
for (ColumnIterator ci : iterators)
ci.close();
return removeDeleted(returnCF);
}
finally
{
lock_.readLock().unlock();
}
}
}

134
src/java/org/apache/cassandra/db/ColumnIterator.java Normal file
View File

@ -0,0 +1,134 @@
/**
* 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.ArrayList;
import java.util.Iterator;
import org.apache.cassandra.io.DataInputBuffer;
import org.apache.cassandra.io.DataOutputBuffer;
import org.apache.cassandra.io.SSTable;
import org.apache.cassandra.io.SequenceFile.ColumnGroupReader;
import org.apache.cassandra.service.StorageService;
import com.google.common.collect.AbstractIterator;
public interface ColumnIterator extends Iterator<IColumn>
{
/**
* returns the CF of the column being iterated.
* The CF is only guaranteed to be available after a call to next() or hasNext().
*/
public abstract ColumnFamily getColumnFamily();
/** clean up any open resources */
public void close() throws IOException;
}
/**
* A Column Iterator over SSTable
*/
class SSTableColumnIterator extends AbstractIterator<IColumn> implements ColumnIterator
{
protected boolean isAscending;
private String startColumn;
private DataOutputBuffer outBuf = new DataOutputBuffer();
private DataInputBuffer inBuf = new DataInputBuffer();
private int curColumnIndex;
private ColumnFamily curCF = null;
private ArrayList<IColumn> curColumns = new ArrayList<IColumn>();
private ColumnGroupReader reader;
public SSTableColumnIterator(String filename, String key, String cfName, String startColumn, boolean isAscending)
throws IOException
{
this.isAscending = isAscending;
SSTable ssTable = new SSTable(filename, StorageService.getPartitioner());
reader = ssTable.getColumnGroupReader(key, cfName, startColumn, isAscending);
this.startColumn = startColumn;
curColumnIndex = isAscending ? 0 : -1;
}
private boolean isColumnNeeded(IColumn column)
{
if (isAscending)
return (column.name().compareTo(startColumn) >= 0);
else
return (column.name().compareTo(startColumn) <= 0);
}
private void getColumnsFromBuffer() throws IOException
{
inBuf.reset(outBuf.getData(), outBuf.getLength());
ColumnFamily columnFamily = ColumnFamily.serializer().deserialize(inBuf);
if (curCF == null)
curCF = columnFamily.cloneMeShallow();
curColumns.clear();
for (IColumn column : columnFamily.getAllColumns())
if (isColumnNeeded(column))
curColumns.add(column);
if (isAscending)
curColumnIndex = 0;
else
curColumnIndex = curColumns.size() - 1;
}
public ColumnFamily getColumnFamily()
{
return curCF;
}
protected IColumn computeNext()
{
while (true)
{
if (isAscending)
{
if (curColumnIndex < curColumns.size())
{
return curColumns.get(curColumnIndex++);
}
}
else
{
if (curColumnIndex >= 0)
{
return curColumns.get(curColumnIndex--);
}
}
try
{
if (!reader.getNextBlock(outBuf))
return endOfData();
getColumnsFromBuffer();
}
catch (IOException e)
{
throw new RuntimeException(e);
}
}
}
public void close() throws IOException
{
reader.close();
}
}

29
src/java/org/apache/cassandra/db/FileStruct.java
View File

@ -28,6 +28,7 @@ import org.apache.cassandra.io.SSTable;
import org.apache.cassandra.io.Coordinate;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.log4j.Logger;
import com.google.common.collect.AbstractIterator;
public class FileStruct implements Comparable<FileStruct>, Iterator<String>
@ -171,10 +172,8 @@ public class FileStruct implements Comparable<FileStruct>, Iterator<String>
throw new UnsupportedOperationException();
}
private class FileStructIterator
private class FileStructIterator extends AbstractIterator<String>
{
String saved;
public FileStructIterator()
{
if (key == null)
@ -184,14 +183,6 @@ public class FileStruct implements Comparable<FileStruct>, Iterator<String>
forward();
}
}
if (key.equals(SSTable.blockIndexKey_))
{
saved = null;
}
else
{
saved = key;
}
}
private void forward()
@ -204,23 +195,17 @@ public class FileStruct implements Comparable<FileStruct>, Iterator<String>
{
throw new RuntimeException(e);
}
saved = isExhausted() ? null : key;
}
public boolean hasNext()
protected String computeNext()
{
return saved != null;
}
public String next()
{
if (saved == null)
if (key.equals(SSTable.blockIndexKey_))
{
throw new IndexOutOfBoundsException();
return endOfData();
}
String key = saved;
String oldKey = key;
forward();
return key;
return oldKey;
}
}
}

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

@ -18,6 +18,14 @@
package org.apache.cassandra.db;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.collections.comparators.ReverseComparator;
import org.apache.commons.lang.ArrayUtils;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.dht.IPartitioner;
import org.apache.cassandra.io.DataOutputBuffer;
@ -27,13 +35,6 @@ import org.apache.cassandra.utils.BloomFilter;
import org.apache.cassandra.utils.DestructivePQIterator;
import org.apache.log4j.Logger;
import java.io.IOException;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.collections.MapUtils;
/**
* Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
*/
@ -177,8 +178,7 @@ public class Memtable implements Comparable<Memtable>
int newObjectCount = oldCf.getColumnCount();
resolveSize(oldSize, newSize);
resolveCount(oldObjectCount, newObjectCount);
oldCf.delete(Math.max(oldCf.getLocalDeletionTime(), columnFamily.getLocalDeletionTime()),
Math.max(oldCf.getMarkedForDeleteAt(), columnFamily.getMarkedForDeleteAt()));
oldCf.delete(columnFamily);
}
else
{
@ -330,4 +330,70 @@ public class Memtable implements Comparable<Memtable>
// race conditions even though a task has been processed.
return !isDirty_;
}
/**
* obtain an iterator of columns in this memtable in the specified order starting from a given column.
*/
ColumnIterator getColumnIterator(final String key, final String cfName, final boolean isAscending, String startColumn)
{
ColumnFamily cf = columnFamilies_.get(key);
final ColumnFamily columnFamily;
if (cf != null)
columnFamily = cf.cloneMeShallow();
else
columnFamily = new ColumnFamily(cfName, DatabaseDescriptor.getColumnFamilyType(cfName));
final IColumn columns[] = (cf == null ? columnFamily : cf).getAllColumns().toArray(new IColumn[columnFamily.getAllColumns().size()]);
// TODO if we are dealing with supercolumns, we need to clone them while we have the read lock since they can be modified later
if (!isAscending)
ArrayUtils.reverse(columns);
IColumn startIColumn;
if (DatabaseDescriptor.getColumnFamilyType(cfName).equals("Standard"))
startIColumn = new Column(startColumn);
else
startIColumn = new SuperColumn(startColumn);
// can't use a ColumnComparatorFactory comparator since those compare on both name and time (and thus will fail to match
// our dummy column, since the time there is arbitrary).
Comparator<IColumn> comparator = new Comparator<IColumn>()
{
public int compare(IColumn column1, IColumn column2)
{
return column1.name().compareTo(column2.name());
}
};
if (!isAscending)
{
comparator = new ReverseComparator(comparator);
}
int index = Arrays.binarySearch(columns, startIColumn, comparator);
final int startIndex = index < 0 ? -(index + 1) : index;
return new ColumnIterator()
{
private int curIndex_ = startIndex;
public ColumnFamily getColumnFamily()
{
return columnFamily;
}
public boolean hasNext()
{
return curIndex_ < columns.length;
}
public IColumn next()
{
return columns[curIndex_++];
}
public void close() throws IOException {}
public void remove()
{
throw new UnsupportedOperationException();
}
};
}
}

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

@ -39,6 +39,8 @@ public abstract class ReadCommand
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 String EMPTY_CF = "";
private static ReadCommandSerializer serializer = new ReadCommandSerializer();
@ -96,6 +98,7 @@ class ReadCommandSerializer implements ICompactSerializer<ReadCommand>
CMD_SERIALIZER_MAP.put(ReadCommand.CMD_TYPE_GET_COLUMNS_SINCE, new ColumnsSinceReadCommandSerializer());
CMD_SERIALIZER_MAP.put(ReadCommand.CMD_TYPE_GET_SLICE, new SliceReadCommandSerializer());
CMD_SERIALIZER_MAP.put(ReadCommand.CMD_TYPE_GET_SLICE_BY_RANGE, new SliceByRangeReadCommandSerializer());
CMD_SERIALIZER_MAP.put(ReadCommand.CMD_TYPE_GET_SLICE_FROM, new SliceFromReadCommandSerializer());
}

99
src/java/org/apache/cassandra/db/SliceFromReadCommand.java Normal file
View File

@ -0,0 +1,99 @@
/**
* 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;
public class SliceFromReadCommand extends ReadCommand
{
public final String columnFamilyColumn;
public final boolean isAscending;
public final int count;
public SliceFromReadCommand(String table, String key, String columnFamilyColumn, boolean isAscending, int count)
{
super(table, key, CMD_TYPE_GET_SLICE_FROM);
this.columnFamilyColumn = columnFamilyColumn;
this.isAscending = isAscending;
this.count = count;
}
@Override
public String getColumnFamilyName()
{
return RowMutation.getColumnAndColumnFamily(columnFamilyColumn)[0];
}
@Override
public ReadCommand copy()
{
ReadCommand readCommand = new SliceFromReadCommand(table, key, columnFamilyColumn, isAscending, count);
readCommand.setDigestQuery(isDigestQuery());
return readCommand;
}
@Override
public Row getRow(Table table) throws IOException
{
return table.getSliceFrom(key, columnFamilyColumn, isAscending, count);
}
@Override
public String toString()
{
return "GetSliceReadMessage(" +
"table='" + table + '\'' +
", key='" + key + '\'' +
", columnFamily='" + columnFamilyColumn + '\'' +
", isAscending='" + isAscending + '\'' +
", count='" + count + '\'' +
')';
}
}
class SliceFromReadCommandSerializer extends ReadCommandSerializer
{
@Override
public void serialize(ReadCommand rm, DataOutputStream dos) throws IOException
{
SliceFromReadCommand realRM = (SliceFromReadCommand)rm;
dos.writeBoolean(realRM.isDigestQuery());
dos.writeUTF(realRM.table);
dos.writeUTF(realRM.key);
dos.writeUTF(realRM.columnFamilyColumn);
dos.writeBoolean(realRM.isAscending);
dos.writeInt(realRM.count);
}
@Override
public ReadCommand deserialize(DataInputStream dis) throws IOException
{
boolean isDigest = dis.readBoolean();
String table = dis.readUTF();
String key = dis.readUTF();
String columnFamily = dis.readUTF();
boolean isAscending = dis.readBoolean();
int count = dis.readInt();
SliceFromReadCommand rm = new SliceFromReadCommand(table, key, columnFamily, isAscending, count);
rm.setDigestQuery(isDigest);
return rm;
}
}

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

@ -50,10 +50,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.service.CassandraServer;
import org.apache.cassandra.utils.BasicUtilities;
import org.apache.cassandra.utils.FBUtilities;
import org.apache.cassandra.utils.FileUtils;
import org.apache.cassandra.utils.LogUtil;
import org.apache.cassandra.utils.*;
import org.apache.log4j.Logger;
/**
@ -775,6 +772,34 @@ public class Table
return row;
}
/**
* Selects a list of columns in a column family from a given column for the specified key.
*/
public Row getSliceFrom(String key, String cf, boolean isAscending, int count) throws IOException
{
Row row = new Row(key);
String[] values = RowMutation.getColumnAndColumnFamily(cf);
ColumnFamilyStore cfStore = columnFamilyStores_.get(values[0]);
long start1 = System.currentTimeMillis();
try
{
ColumnFamily columnFamily = cfStore.getSliceFrom(key, values[0], values[1], isAscending, count);
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);
}
}
/**
* This method adds the row to the Commit Log associated with this table.
* Once this happens the data associated with the individual column families
@ -950,39 +975,46 @@ public class Table
}
}
Iterator<String> collated = IteratorUtils.collatedIterator(comparator, iterators);
Iterable<String> reduced = new ReducingIterator<String>(collated) {
String current;
public void reduce(String current)
{
this.current = current;
}
protected String getReduced()
{
return current;
}
};
try
{
// pull keys out of the CollatedIterator. checking tombstone status is expensive,
// so we set an arbitrary limit on how many we'll do at once.
List<String> keys = new ArrayList<String>();
String last = null, current = null;
while (keys.size() < maxResults)
for (String current : reduced)
{
if (!collated.hasNext())
if (!stopAt.isEmpty() && comparator.compare(stopAt, current) < 0)
{
break;
}
current = collated.next();
if (!current.equals(last))
// 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)
for (String cfName : getApplicationColumnFamilies())
{
if (!stopAt.isEmpty() && comparator.compare(stopAt, current) < 0)
ColumnFamilyStore cfs = getColumnFamilyStore(cfName);
ColumnFamily cf = cfs.getColumnFamily(current, cfName, new IdentityFilter(), Integer.MAX_VALUE);
if (cf != null && cf.getColumns().size() > 0)
{
keys.add(current);
break;
}
last = current;
// 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)
for (String cfName : getApplicationColumnFamilies())
{
ColumnFamilyStore cfs = getColumnFamilyStore(cfName);
ColumnFamily cf = cfs.getColumnFamily(current, cfName, new IdentityFilter(), Integer.MAX_VALUE);
if (cf != null && cf.getColumns().size() > 0)
{
keys.add(current);
break;
}
}
}
if (keys.size() >= maxResults)
{
break;
}
}
return keys;

26
src/java/org/apache/cassandra/io/SSTable.java
View File

@ -42,6 +42,7 @@ import org.apache.cassandra.utils.BasicUtilities;
import org.apache.cassandra.utils.BloomFilter;
import org.apache.cassandra.utils.FileUtils;
import org.apache.cassandra.utils.LogUtil;
import org.apache.cassandra.io.SequenceFile.ColumnGroupReader;
/**
* This class is built on top of the SequenceFile. It stores
@ -873,4 +874,29 @@ public class SSTable
return hashtable.remove(cannonicalize(filename));
}
}
/**
* obtain a BlockReader for the getColumnSlice call.
*/
public ColumnGroupReader getColumnGroupReader(String key, String cfName,
String startColumn, boolean isAscending) throws IOException
{
ColumnGroupReader reader = null;
IFileReader dataReader = SequenceFile.reader(dataFile_);
try
{
/* Morph key into actual key based on the partition type. */
String decoratedKey = partitioner_.decorateKey(key);
Coordinate fileCoordinate = getCoordinates(decoratedKey, dataReader, partitioner_);
reader = new ColumnGroupReader(dataFile_, decoratedKey, cfName, startColumn, isAscending, fileCoordinate);
}
finally
{
if (dataReader != null)
dataReader.close();
}
return reader;
}
}

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

@ -28,6 +28,7 @@ import java.security.PrivilegedAction;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Arrays;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.utils.BloomFilter;
@ -532,6 +533,146 @@ public class SequenceFile
}
}
/**
* This is a reader that finds the block for a starting column and returns
* blocks before/after it for each next call. This function assumes that
* the CF is sorted by name and exploits the name index.
*/
public static class ColumnGroupReader extends BufferReader
{
private String key_;
private String cfName_;
private boolean isAscending_;
private List<IndexHelper.ColumnIndexInfo> columnIndexList_;
private long columnStartPosition_;
private int curRangeIndex_;
private int allColumnsSize_;
private int localDeletionTime_;
private long markedForDeleteAt_;
ColumnGroupReader(String filename, String key, String cfName, String startColumn, boolean isAscending, Coordinate section) throws IOException
{
super(filename, 128 * 1024);
this.cfName_ = cfName;
this.key_ = key;
this.isAscending_ = isAscending;
init(startColumn, section);
}
/**
* Build a list of index entries ready for search.
*/
private List<IndexHelper.ColumnIndexInfo> getFullColumnIndexList(List<IndexHelper.ColumnIndexInfo> columnIndexList, int totalNumCols)
{
if (columnIndexList.size() == 0)
{
/* if there is no column index, add an index entry that covers the full space. */
return Arrays.asList(new IndexHelper.ColumnIndexInfo[]{new IndexHelper.ColumnNameIndexInfo("", 0, totalNumCols)});
}
List<IndexHelper.ColumnIndexInfo> fullColIndexList = new ArrayList<IndexHelper.ColumnIndexInfo>();
int accumulatededCols = 0;
for (IndexHelper.ColumnIndexInfo colPosInfo : columnIndexList)
accumulatededCols += colPosInfo.count();
int remainingCols = totalNumCols - accumulatededCols;
fullColIndexList.add(new IndexHelper.ColumnNameIndexInfo("", 0, columnIndexList.get(0).count()));
for (int i = 0; i < columnIndexList.size() - 1; i++)
{
IndexHelper.ColumnNameIndexInfo colPosInfo = (IndexHelper.ColumnNameIndexInfo)columnIndexList.get(i);
fullColIndexList.add(new IndexHelper.ColumnNameIndexInfo(colPosInfo.name(),
colPosInfo.position(),
columnIndexList.get(i + 1).count()));
}
String columnName = ((IndexHelper.ColumnNameIndexInfo)columnIndexList.get(columnIndexList.size() - 1)).name();
fullColIndexList.add(new IndexHelper.ColumnNameIndexInfo(columnName,
columnIndexList.get(columnIndexList.size() - 1).position(),
remainingCols));
return fullColIndexList;
}
private void init(String startColumn, Coordinate section) throws IOException
{
String keyInDisk = null;
if (seekTo(key_, section) >= 0)
keyInDisk = file_.readUTF();
if ( keyInDisk != null && keyInDisk.equals(key_))
{
/* read off the size of this row */
int dataSize = file_.readInt();
/* skip the bloomfilter */
int totalBytesRead = IndexHelper.skipBloomFilter(file_);
/* read off the index flag, it has to be true */
boolean hasColumnIndexes = file_.readBoolean();
totalBytesRead += 1;
/* read the index */
List<IndexHelper.ColumnIndexInfo> colIndexList = new ArrayList<IndexHelper.ColumnIndexInfo>();
if (hasColumnIndexes)
totalBytesRead += IndexHelper.deserializeIndex(cfName_, file_, colIndexList);
/* need to do two things here.
* 1. move the file pointer to the beginning of the list of stored columns
* 2. calculate the size of all columns */
String cfName = file_.readUTF();
localDeletionTime_ = file_.readInt();
markedForDeleteAt_ = file_.readLong();
int totalNumCols = file_.readInt();
allColumnsSize_ = dataSize - (totalBytesRead + utfPrefix_ + cfName.length() + 4 + 8 + 4);
columnStartPosition_ = file_.getFilePointer();
columnIndexList_ = getFullColumnIndexList(colIndexList, totalNumCols);
int index = Collections.binarySearch(columnIndexList_, new IndexHelper.ColumnNameIndexInfo(startColumn));
curRangeIndex_ = index < 0 ? (++index) * (-1) - 1 : index;
}
else
{
/* no keys found in this file because of a false positive in BF */
curRangeIndex_ = -1;
columnIndexList_ = new ArrayList<IndexHelper.ColumnIndexInfo>();
}
}
private boolean getBlockFromCurIndex(DataOutputBuffer bufOut) throws IOException
{
if (curRangeIndex_ < 0 || curRangeIndex_ >= columnIndexList_.size())
return false;
IndexHelper.ColumnIndexInfo curColPostion = columnIndexList_.get(curRangeIndex_);
long start = curColPostion.position();
long end = curRangeIndex_ < columnIndexList_.size() - 1
? columnIndexList_.get(curRangeIndex_+1).position()
: allColumnsSize_;
/* seek to the correct offset to the data, and calculate the data size */
file_.seek(columnStartPosition_ + start);
long dataSize = end - start;
bufOut.reset();
// write CF info
bufOut.writeUTF(cfName_);
bufOut.writeInt(localDeletionTime_);
bufOut.writeLong(markedForDeleteAt_);
// now write the columns
bufOut.writeInt(curColPostion.count());
bufOut.write(file_, (int)dataSize);
return true;
}
public boolean getNextBlock(DataOutputBuffer outBuf) throws IOException
{
boolean result = getBlockFromCurIndex(outBuf);
if (isAscending_)
curRangeIndex_++;
else
curRangeIndex_--;
return result;
}
}
public static abstract class AbstractReader implements IFileReader
{
private static final short utfPrefix_ = 2;
@ -705,7 +846,7 @@ public class SequenceFile
* @param section indicates the location of the block index.
* @throws IOException
*/
private long seekTo(String key, Coordinate section) throws IOException
protected long seekTo(String key, Coordinate section) throws IOException
{
seek(section.end_);
long position = getPositionFromBlockIndex(key);

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

@ -35,6 +35,7 @@ import org.apache.cassandra.db.ColumnReadCommand;
import org.apache.cassandra.db.ColumnsSinceReadCommand;
import org.apache.cassandra.db.SliceByNamesReadCommand;
import org.apache.cassandra.db.SliceByRangeReadCommand;
import org.apache.cassandra.db.SliceFromReadCommand;
import org.apache.cassandra.db.SliceReadCommand;
import org.apache.cassandra.db.IColumn;
import org.apache.cassandra.db.Row;
@ -208,7 +209,26 @@ public class CassandraServer implements Cassandra.Iface
}
return thriftifyColumns(columns);
}
public List<column_t> get_slice_from(String tablename, String key, String columnFamily_column, boolean isAscending, int count) throws InvalidRequestException
{
logger.debug("get_slice_from");
String[] values = RowMutation.getColumnAndColumnFamily(columnFamily_column);
if (values.length != 2 || DatabaseDescriptor.getColumnFamilyType(values[0]) != "Standard")
throw new InvalidRequestException("get_slice_from requires a standard CF name and a starting column name");
if (count <= 0)
throw new InvalidRequestException("get_slice_from requires positive count");
if ("Name".compareTo(DatabaseDescriptor.getCFMetaData(tablename, values[0]).indexProperty_) != 0)
throw new InvalidRequestException("get_slice_from requires CF indexed by name");
ColumnFamily cfamily = readColumnFamily(new SliceFromReadCommand(tablename, key, columnFamily_column, isAscending, count));
if (cfamily == null)
{
return EMPTY_COLUMNS;
}
Collection<IColumn> columns = cfamily.getAllColumns();
return thriftifyColumns(columns);
}
public column_t get_column(String tablename, String key, String columnFamily_column) throws NotFoundException, InvalidRequestException
{
logger.debug("get_column");

59
src/java/org/apache/cassandra/utils/ReducingIterator.java Normal file
View File

@ -0,0 +1,59 @@
package org.apache.cassandra.utils;
import java.util.Iterator;
import com.google.common.collect.AbstractIterator;
/**
* reduces equal values from the source iterator to a single (optionally transformed) instance.
*/
public abstract class ReducingIterator<T> extends AbstractIterator<T> implements Iterator<T>, Iterable<T>
{
protected Iterator<T> source;
protected T last;
public ReducingIterator(Iterator<T> source)
{
this.source = source;
}
/** combine this object with the previous ones. intermediate state is up to your implementation. */
public abstract void reduce(T current);
/** return the last object computed by reduce */
protected abstract T getReduced();
/** override this if the keys you want to base the reduce on are not the same as the object itself (but can be generated from it) */
protected Object getKey(T o)
{
return o;
}
protected T computeNext()
{
if (last == null && !source.hasNext())
return endOfData();
boolean keyChanged = false;
while (!keyChanged)
{
if (last != null)
reduce(last);
if (!source.hasNext())
{
last = null;
break;
}
T current = source.next();
if (last != null && !getKey(current).equals(getKey(last)))
keyChanged = true;
last = current;
}
return getReduced();
}
public Iterator<T> iterator()
{
return this;
}
}

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

@ -18,6 +18,11 @@
package org.apache.cassandra.db;
import java.util.SortedSet;
import java.util.Arrays;
import java.util.List;
import java.util.ArrayList;
import org.junit.Test;
import static junit.framework.Assert.*;
@ -202,4 +207,117 @@ public class TableTest extends CleanupHelper
rm.add(cf);
return rm;
}
@Test
public void testGetSliceFromBasic() throws Throwable
{
Table table = Table.open(TABLE_NAME);
String ROW = "row1";
RowMutation rm = new RowMutation(TABLE_NAME, ROW);
ColumnFamily cf = new ColumnFamily("Standard1", "Standard");
cf.addColumn(new Column("col1", "val1".getBytes(), 1L));
cf.addColumn(new Column("col3", "val3".getBytes(), 1L));
cf.addColumn(new Column("col4", "val4".getBytes(), 1L));
cf.addColumn(new Column("col5", "val5".getBytes(), 1L));
cf.addColumn(new Column("col7", "val7".getBytes(), 1L));
cf.addColumn(new Column("col9", "val9".getBytes(), 1L));
rm.add(cf);
rm.apply();
rm = new RowMutation(TABLE_NAME, ROW);
rm.delete("Standard1:col4", 2L);
rm.apply();
validateGetSliceFromBasic(table, ROW);
// flush to disk
table.getColumnFamilyStore("Standard1").forceBlockingFlush();
validateGetSliceFromBasic(table, ROW);
}
@Test
public void testGetSliceFromAdvanced() throws Throwable
{
Table table = Table.open(TABLE_NAME);
String ROW = "row2";
RowMutation rm = new RowMutation(TABLE_NAME, ROW);
ColumnFamily cf = new ColumnFamily("Standard1", "Standard");
cf.addColumn(new Column("col1", "val1".getBytes(), 1L));
cf.addColumn(new Column("col2", "val2".getBytes(), 1L));
cf.addColumn(new Column("col3", "val3".getBytes(), 1L));
cf.addColumn(new Column("col4", "val4".getBytes(), 1L));
cf.addColumn(new Column("col5", "val5".getBytes(), 1L));
cf.addColumn(new Column("col6", "val6".getBytes(), 1L));
rm.add(cf);
rm.apply();
// flush to disk
table.getColumnFamilyStore("Standard1").forceBlockingFlush();
rm = new RowMutation(TABLE_NAME, ROW);
cf = new ColumnFamily("Standard1", "Standard");
cf.addColumn(new Column("col1", "valx".getBytes(), 2L));
cf.addColumn(new Column("col2", "valx".getBytes(), 2L));
cf.addColumn(new Column("col3", "valx".getBytes(), 2L));
rm.add(cf);
rm.apply();
validateGetSliceFromAdvanced(table, ROW);
// flush to disk
table.getColumnFamilyStore("Standard1").forceBlockingFlush();
validateGetSliceFromAdvanced(table, ROW);
}
private void assertColumns(ColumnFamily columnFamily, String... columnFamilyNames)
{
assertNotNull(columnFamily);
SortedSet<IColumn> columns = columnFamily.getAllColumns();
List<String> L = new ArrayList<String>();
for (IColumn column : columns)
{
L.add(column.name());
}
assert Arrays.equals(L.toArray(new String[columns.size()]), columnFamilyNames);
}
private void validateGetSliceFromAdvanced(Table table, String row) throws Throwable
{
Row result;
ColumnFamily cfres;
result = table.getSliceFrom(row, "Standard1:col2", true, 3);
cfres = result.getColumnFamily("Standard1");
assertColumns(cfres, "col2", "col3", "col4");
assertEquals(new String(cfres.getColumn("col2").value()), "valx");
assertEquals(new String(cfres.getColumn("col3").value()), "valx");
assertEquals(new String(cfres.getColumn("col4").value()), "val4");
}
private void validateGetSliceFromBasic(Table table, String row) throws Throwable
{
Row result;
ColumnFamily cf;
result = table.getSliceFrom(row, "Standard1:col5", true, 2);
cf = result.getColumnFamily("Standard1");
assertColumns(cf, "col5", "col7");
result = table.getSliceFrom(row, "Standard1:col4", true, 2);
cf = result.getColumnFamily("Standard1");
assertColumns(cf, "col4", "col5", "col7");
result = table.getSliceFrom(row, "Standard1:col5", false, 2);
cf = result.getColumnFamily("Standard1");
assertColumns(cf, "col3", "col4", "col5");
result = table.getSliceFrom(row, "Standard1:col6", false, 2);
cf = result.getColumnFamily("Standard1");
assertColumns(cf, "col3", "col4", "col5");
result = table.getSliceFrom(row, "Standard1:col95", true, 2);
cf = result.getColumnFamily("Standard1");
assertColumns(cf);
result = table.getSliceFrom(row, "Standard1:col0", false, 2);
cf = result.getColumnFamily("Standard1");
assertColumns(cf);
}
}