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

Pass and write table parameter as needed throughout the codebase to add multitable support. 

(Many places cheated and just assumed that tables.get(0) was the only table.)
Patch builds but does not yet pass tests.

Patch by goffinet; reviewed by jbellis for CASSANDRA-79

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@787403 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2009-06-22 21:17:59 +00:00
parent 1a66c10695
commit 9aeef94db8
30 changed files with 431 additions and 360 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -459,16 +459,16 @@ public class DatabaseDescriptor
for ( String table : tables )
{
Table.TableMetadata tmetadata = Table.TableMetadata.instance();
Table.TableMetadata tmetadata = Table.TableMetadata.instance(table);
if (tmetadata.isEmpty())
{
tmetadata = Table.TableMetadata.instance();
tmetadata = Table.TableMetadata.instance(table);
/* Column families associated with this table */
Map<String, CFMetaData> columnFamilies = tableToCFMetaDataMap_.get(table);
for (String columnFamily : columnFamilies.keySet())
{
tmetadata.add(columnFamily, idGenerator.getAndIncrement(), DatabaseDescriptor.getColumnType(columnFamily));
tmetadata.add(columnFamily, idGenerator.getAndIncrement(), DatabaseDescriptor.getColumnType(table, columnFamily));
}
/*
@ -580,9 +580,10 @@ public class DatabaseDescriptor
return sb.toString();
}
public static Map<String, CFMetaData> getTableMetaData(String table)
public static Map<String, CFMetaData> getTableMetaData(String tableName)
{
return tableToCFMetaDataMap_.get(table);
assert tableName != null;
return tableToCFMetaDataMap_.get(tableName);
}
/*
@ -590,19 +591,20 @@ public class DatabaseDescriptor
* meta data. If the table name or column family name is not valid
* this function returns null.
*/
public static CFMetaData getCFMetaData(String table, String cfName)
public static CFMetaData getCFMetaData(String tableName, String cfName)
{
Map<String, CFMetaData> cfInfo = tableToCFMetaDataMap_.get(table);
assert tableName != null;
Map<String, CFMetaData> cfInfo = tableToCFMetaDataMap_.get(tableName);
if (cfInfo == null)
return null;
return cfInfo.get(cfName);
}
public static String getColumnType(String cfName)
public static String getColumnType(String tableName, String cfName)
{
String table = getTables().get(0);
CFMetaData cfMetaData = getCFMetaData(table, cfName);
assert tableName != null;
CFMetaData cfMetaData = getCFMetaData(tableName, cfName);
if (cfMetaData == null)
return null;
@ -611,6 +613,7 @@ public class DatabaseDescriptor
public static int getFlushPeriod(String tableName, String columnFamilyName)
{
assert tableName != null;
CFMetaData cfMetaData = getCFMetaData(tableName, columnFamilyName);
if (cfMetaData == null)
@ -618,10 +621,10 @@ public class DatabaseDescriptor
return cfMetaData.flushPeriodInMinutes;
}
public static boolean isNameSortingEnabled(String cfName)
public static boolean isNameSortingEnabled(String tableName, String cfName)
{
String table = getTables().get(0);
CFMetaData cfMetaData = getCFMetaData(table, cfName);
assert tableName != null;
CFMetaData cfMetaData = getCFMetaData(tableName, cfName);
if (cfMetaData == null)
return false;
@ -629,23 +632,29 @@ public class DatabaseDescriptor
return "Name".equals(cfMetaData.indexProperty_);
}
public static boolean isTimeSortingEnabled(String cfName)
public static boolean isTimeSortingEnabled(String tableName, String cfName)
{
String table = getTables().get(0);
CFMetaData cfMetaData = getCFMetaData(table, cfName);
assert tableName != null;
CFMetaData cfMetaData = getCFMetaData(tableName, cfName);
if (cfMetaData == null)
return false;
return "Time".equals(cfMetaData.indexProperty_);
}
public static List<String> getTables()
{
return tables_;
}
public static String getTable(String tableName)
{
assert tableName != null;
int index = getTables().indexOf(tableName);
return index >= 0 ? getTables().get(index) : null;
}
public static void setTables(String table)
{
tables_.add(table);
@ -764,9 +773,10 @@ public class DatabaseDescriptor
return seeds_;
}
public static String getColumnFamilyType(String cfName)
public static String getColumnFamilyType(String tableName, String cfName)
{
String cfType = getColumnType(cfName);
assert tableName != null;
String cfType = getColumnType(tableName, cfName);
if ( cfType == null )
cfType = "Standard";
return cfType;
@ -806,10 +816,10 @@ public class DatabaseDescriptor
return dataFileDirectory;
}
public static TypeInfo getTypeInfo(String cfName)
public static TypeInfo getTypeInfo(String tableName, String cfName)
{
String table = DatabaseDescriptor.getTables().get(0);
CFMetaData cfMetadata = DatabaseDescriptor.getCFMetaData(table, cfName);
assert tableName != null;
CFMetaData cfMetadata = DatabaseDescriptor.getCFMetaData(tableName, cfName);
if ( cfMetadata.indexProperty_.equals("Name") )
{
return TypeInfo.STRING;

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

@ -145,7 +145,7 @@ public class BinaryMemtable
* Use the SSTable to write the contents of the TreeMap
* to disk.
*/
SSTable ssTable = new SSTable(directory, filename, StorageService.getPartitioner());
SSTable ssTable = new SSTable(directory, filename, null, StorageService.getPartitioner());
List<String> keys = new ArrayList<String>( columnFamilies_.keySet() );
Collections.sort(keys);
/* Use this BloomFilter to decide if a key exists in a SSTable */

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

@ -34,8 +34,8 @@ public class ColumnComparatorFactory
TIMESTAMP
}
private static Comparator<IColumn> nameComparator_ = new ColumnNameComparator();
private static Comparator<IColumn> timestampComparator_ = new ColumnTimestampComparator();
public static final Comparator<IColumn> nameComparator_ = new ColumnNameComparator();
public static final Comparator<IColumn> timestampComparator_ = new ColumnTimestampComparator();
public static Comparator<IColumn> getComparator(ComparatorType comparatorType)
{

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

@ -51,6 +51,7 @@ public final class ColumnFamily
private static Map<String, String> columnTypes_ = new HashMap<String, String>();
private static Map<String, String> indexTypes_ = new HashMap<String, String>();
private String type_;
private String table_;
static
{
@ -91,6 +92,23 @@ public final class ColumnFamily
return indexTypes_.get(columnIndexProperty);
}
public static ColumnFamily create(String tableName, String cfName)
{
Comparator<IColumn> comparator;
String columnType = DatabaseDescriptor.getColumnFamilyType(tableName, cfName);
if ("Super".equals(columnType)
|| 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);
}
return new ColumnFamily(cfName, columnType, comparator);
}
private transient AbstractColumnFactory columnFactory_;
private String name_;
@ -101,66 +119,24 @@ public final class ColumnFamily
private AtomicInteger size_ = new AtomicInteger(0);
private EfficientBidiMap columns_;
private Comparator<IColumn> columnComparator_;
private Comparator<IColumn> getColumnComparator(String cfName, String columnType)
{
if(columnComparator_ == null)
{
/*
* if this columnfamily has supercolumns or there is an index on the column name,
* then sort by name
*/
if("Super".equals(columnType) || DatabaseDescriptor.isNameSortingEnabled(cfName))
{
columnComparator_ = ColumnComparatorFactory.getComparator(ColumnComparatorFactory.ComparatorType.NAME);
}
/* if this columnfamily has simple columns, and no index on name sort by timestamp */
else
{
columnComparator_ = ColumnComparatorFactory.getComparator(ColumnComparatorFactory.ComparatorType.TIMESTAMP);
}
}
return columnComparator_;
}
public ColumnFamily(String cfName, String columnType)
public ColumnFamily(String cfName, String columnType, Comparator<IColumn> comparator)
{
name_ = cfName;
type_ = columnType;
createColumnFactoryAndColumnSerializer(columnType);
columnFactory_ = AbstractColumnFactory.getColumnFactory(columnType);
columnSerializer_ = columnFactory_.createColumnSerializer();
if(columns_ == null)
columns_ = new EfficientBidiMap(comparator);
}
void createColumnFactoryAndColumnSerializer(String columnType)
public ColumnFamily(String cfName, String columnType, ColumnComparatorFactory.ComparatorType indexType)
{
if ( columnFactory_ == null )
{
columnFactory_ = AbstractColumnFactory.getColumnFactory(columnType);
columnSerializer_ = columnFactory_.createColumnSerializer();
if(columns_ == null)
columns_ = new EfficientBidiMap(getColumnComparator(name_, columnType));
}
}
void createColumnFactoryAndColumnSerializer()
{
String columnType = DatabaseDescriptor.getColumnFamilyType(name_);
if ( columnType == null )
{
List<String> tables = DatabaseDescriptor.getTables();
if ( tables.size() > 0 )
{
String table = tables.get(0);
columnType = Table.open(table).getColumnFamilyType(name_);
}
}
createColumnFactoryAndColumnSerializer(columnType);
this(cfName, columnType, ColumnComparatorFactory.getComparator(indexType));
}
ColumnFamily cloneMeShallow()
{
ColumnFamily cf = new ColumnFamily(name_, type_);
ColumnFamily cf = new ColumnFamily(name_, type_, getComparator());
cf.markedForDeleteAt = markedForDeleteAt;
cf.localDeletionTime = localDeletionTime;
return cf;
@ -192,7 +168,6 @@ public final class ColumnFamily
public ICompactSerializer2<IColumn> getColumnSerializer()
{
createColumnFactoryAndColumnSerializer();
return columnSerializer_;
}
@ -320,13 +295,21 @@ public final class ColumnFamily
return markedForDeleteAt > Long.MIN_VALUE;
}
public String getTable() {
return table_;
}
public void setTable_(String table_) {
this.table_ = table_;
}
/*
* 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)
{
ColumnFamily cfDiff = new ColumnFamily(cfComposite.name(), cfComposite.type_);
ColumnFamily cfDiff = new ColumnFamily(cfComposite.name(), cfComposite.type_, getComparator());
if (cfComposite.getMarkedForDeleteAt() > getMarkedForDeleteAt())
{
cfDiff.delete(cfComposite.getLocalDeletionTime(), cfComposite.getMarkedForDeleteAt());
@ -361,6 +344,18 @@ public final class ColumnFamily
return null;
}
private Comparator<IColumn> getComparator()
{
return columns_.getComparator();
}
private ColumnComparatorFactory.ComparatorType getComparatorType()
{
return getComparator() == ColumnComparatorFactory.nameComparator_
? ColumnComparatorFactory.ComparatorType.NAME
: ColumnComparatorFactory.ComparatorType.TIMESTAMP;
}
int size()
{
if ( size_.get() == 0 )
@ -489,6 +484,8 @@ public final class ColumnFamily
Collection<IColumn> columns = columnFamily.getAllColumns();
dos.writeUTF(columnFamily.name());
dos.writeUTF(columnFamily.type_);
dos.writeInt(columnFamily.getComparatorType().ordinal());
dos.writeInt(columnFamily.localDeletionTime);
dos.writeLong(columnFamily.markedForDeleteAt);
@ -505,8 +502,9 @@ public final class ColumnFamily
*/
private ColumnFamily defreezeColumnFamily(DataInputStream dis) throws IOException
{
String name = dis.readUTF();
ColumnFamily cf = new ColumnFamily(name, DatabaseDescriptor.getColumnFamilyType(name));
ColumnFamily cf = new ColumnFamily(dis.readUTF(),
dis.readUTF(),
ColumnComparatorFactory.ComparatorType.values()[dis.readInt()]);
cf.delete(dis.readInt(), dis.readLong());
return cf;
}
@ -515,11 +513,11 @@ public final class ColumnFamily
{
ColumnFamily cf = defreezeColumnFamily(dis);
int size = dis.readInt();
IColumn column = null;
for ( int i = 0; i < size; ++i )
IColumn column;
for (int i = 0; i < size; ++i)
{
column = cf.getColumnSerializer().deserialize(dis);
if(column != null)
if (column != null)
{
cf.addColumn(column);
}
@ -587,5 +585,6 @@ public final class ColumnFamily
throw new UnsupportedOperationException("This operation is not yet supported.");
}
}
}

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

@ -135,7 +135,7 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
Collections.sort(indices);
int value = (indices.size() > 0) ? (indices.get(indices.size() - 1)) : 0;
ColumnFamilyStore cfs = new ColumnFamilyStore(table, columnFamily, "Super".equals(DatabaseDescriptor.getColumnType(columnFamily)), value);
ColumnFamilyStore cfs = new ColumnFamilyStore(table, columnFamily, "Super".equals(DatabaseDescriptor.getColumnType(table, columnFamily)), value);
MBeanServer mbs = ManagementFactory.getPlatformMBeanServer();
try
@ -188,7 +188,7 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
/* There are no files to compact just add to the list of SSTables */
ssTables_.addAll(filenames);
/* Load the index files and the Bloom Filters associated with them. */
SSTable.onStart(filenames);
SSTable.onStart(filenames, table_);
MinorCompactionManager.instance().submit(ColumnFamilyStore.this);
if (columnFamily_.equals(Table.hints_))
{
@ -577,7 +577,7 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
private ColumnFamily fetchColumnFamily(String key, String cf, IFilter filter, String ssTableFile) throws IOException
{
SSTable ssTable = new SSTable(ssTableFile, StorageService.getPartitioner());
SSTable ssTable = new SSTable(ssTableFile, null, StorageService.getPartitioner());
DataInputBuffer bufIn;
bufIn = filter.next(key, cf, ssTable);
if (bufIn.getLength() == 0)
@ -1118,12 +1118,16 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
{
if (ssTableRange == null)
{
String [] temp = null;
String tableName;
temp = rangeFileLocation.split("-");
tableName = temp[0];
if (target != null)
{
rangeFileLocation = rangeFileLocation + System.getProperty("file.separator") + "bootstrap";
}
FileUtils.createDirectory(rangeFileLocation);
ssTableRange = new SSTable(rangeFileLocation, mergedFileName, StorageService.getPartitioner());
ssTableRange = new SSTable(rangeFileLocation, mergedFileName, tableName, StorageService.getPartitioner());
}
try
{
@ -1319,7 +1323,7 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
if (ssTable == null)
{
ssTable = new SSTable(compactionFileLocation, mergedFileName, StorageService.getPartitioner());
ssTable = new SSTable(compactionFileLocation, mergedFileName, null, StorageService.getPartitioner());
}
ssTable.append(lastkey, bufOut);
@ -1539,6 +1543,10 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
{
return readStats_.mean();
}
public String getTableName()
{
return table_;
}
/**
* get a list of columns starting from a given column, in a specified order
@ -1604,7 +1612,7 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
comparator = new ReverseComparator(comparator);
Iterator collated = IteratorUtils.collatedIterator(comparator, iterators);
if (!collated.hasNext())
return new ColumnFamily(cfName, DatabaseDescriptor.getColumnFamilyType(cfName));
return ColumnFamily.create(table_, cfName);
ReducingIterator<IColumn> reduced = new ReducingIterator<IColumn>(collated)
{
ColumnFamily curCF = returnCF.cloneMeShallow();

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

@ -57,7 +57,7 @@ public class ColumnIndexer
dos.write(bufOut.getData(), 0, bufOut.getLength());
/* Do the indexing */
TypeInfo typeInfo = DatabaseDescriptor.getTypeInfo(columnFamily.name());
TypeInfo typeInfo = DatabaseDescriptor.getTypeInfo(columnFamily.getTable(), columnFamily.name());
doIndexing(typeInfo, columns, dos);
}

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

@ -58,7 +58,7 @@ class SSTableColumnIterator extends AbstractIterator<IColumn> implements ColumnI
throws IOException
{
this.isAscending = isAscending;
SSTable ssTable = new SSTable(filename, StorageService.getPartitioner());
SSTable ssTable = new SSTable(filename, null, StorageService.getPartitioner());
reader = ssTable.getColumnGroupReader(key, cfName, startColumn, isAscending);
this.startColumn = startColumn;
curColumnIndex = isAscending ? 0 : -1;

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

@ -272,7 +272,7 @@ public class CommitLog
for (File file : clogs)
{
// IFileReader reader = SequenceFile.bufferedReader(file.getAbsolutePath(), DatabaseDescriptor.getLogFileSizeThreshold());
IFileReader reader = SequenceFile.reader(file.getAbsolutePath());
IFileReader reader = SequenceFile.reader(table_, file.getAbsolutePath());
try
{
CommitLogHeader clHeader = readCommitLogHeader(reader);
@ -299,7 +299,7 @@ public class CommitLog
/* read the commit log entry */
try
{
Row row = Row.serializer().deserialize(bufIn);
Row row = Row.serializer(table_).deserialize(bufIn);
Table table = Table.open(table_);
tablesRecovered.add(table);
Collection<ColumnFamily> columnFamilies = new ArrayList<ColumnFamily>(row.getColumnFamilies());
@ -382,7 +382,7 @@ public class CommitLog
{
/* serialize the row */
cfBuffer.reset();
Row.serializer().serialize(row, cfBuffer);
Row.serializer(table_).serialize(row, cfBuffer);
currentPosition = logWriter_.getCurrentPosition();
cLogCtx = new CommitLogContext(logFile_, currentPosition);
/* Update the header */

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

@ -108,8 +108,8 @@ public class DBManager
int generation = 1;
String key = FBUtilities.getHostAddress();
row = new Row(key);
ColumnFamily cf = new ColumnFamily(SystemTable.cfName_, "Standard");
row = new Row(SystemTable.name_, key);
ColumnFamily cf = ColumnFamily.create("system", SystemTable.cfName_); // TODO create system table
cf.addColumn(new Column(SystemTable.token_, p.getTokenFactory().toByteArray(token)));
cf.addColumn(new Column(SystemTable.generation_, BasicUtilities.intToByteArray(generation)) );
row.addColumnFamily(cf);

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

@ -76,7 +76,7 @@ public class HintedHandOffManager
return instance_;
}
private static boolean sendMessage(String endpointAddress, String key) throws DigestMismatchException, TimeoutException, IOException, InvalidRequestException
private static boolean sendMessage(String endpointAddress, String tableName, String key) throws DigestMismatchException, TimeoutException, IOException, InvalidRequestException
{
EndPoint endPoint = new EndPoint(endpointAddress, DatabaseDescriptor.getStoragePort());
if (!FailureDetector.instance().isAlive(endPoint))
@ -84,14 +84,14 @@ public class HintedHandOffManager
return false;
}
Table table = Table.open(DatabaseDescriptor.getTables().get(0));
Table table = Table.open(tableName);
Row row = table.get(key);
Row purgedRow = new Row(key);
Row purgedRow = new Row(tableName,key);
for (ColumnFamily cf : row.getColumnFamilies())
{
purgedRow.addColumnFamily(ColumnFamilyStore.removeDeleted(cf));
}
RowMutation rm = new RowMutation(DatabaseDescriptor.getTables().get(0), purgedRow);
RowMutation rm = new RowMutation(tableName, purgedRow);
Message message = rm.makeRowMutationMessage();
QuorumResponseHandler<Boolean> quorumResponseHandler = new QuorumResponseHandler<Boolean>(1, new WriteResponseResolver());
MessagingService.getMessagingInstance().sendRR(message, new EndPoint[]{ endPoint }, quorumResponseHandler);
@ -99,14 +99,14 @@ public class HintedHandOffManager
return quorumResponseHandler.get();
}
private static void deleteEndPoint(String endpointAddress, String key, long timestamp) throws Exception
private static void deleteEndPoint(String endpointAddress, String tableName, String key, long timestamp) throws Exception
{
RowMutation rm = new RowMutation(DatabaseDescriptor.getTables().get(0), key_);
RowMutation rm = new RowMutation(tableName, key_);
rm.delete(Table.hints_ + ":" + key + ":" + endpointAddress, timestamp);
rm.apply();
}
private static void deleteHintedData(String key) throws Exception
private static void deleteHintedData(String tableName, String key) throws Exception
{
// delete the row from Application CFs: find the largest timestamp in any of
// the data columns, and delete the entire CF with that value for the tombstone.
@ -116,8 +116,8 @@ public class HintedHandOffManager
// This is sub-optimal but okay, since HH is just an effort to make a recovering
// node more consistent than it would have been; we can rely on the other
// consistency mechanisms to finish the job in this corner case.
RowMutation rm = new RowMutation(DatabaseDescriptor.getTables().get(0), key_);
Table table = Table.open(DatabaseDescriptor.getTables().get(0));
RowMutation rm = new RowMutation(tableName, key_);
Table table = Table.open(tableName);
Row row = table.get(key); // not necessary to do removeDeleted here
Collection<ColumnFamily> cfs = row.getColumnFamilies();
for (ColumnFamily cf : cfs)
@ -155,48 +155,51 @@ public class HintedHandOffManager
// 5. Now force a flush
// 6. Do major compaction to clean up all deletes etc.
// 7. I guess we r done
Table table = Table.open(DatabaseDescriptor.getTables().get(0));
try
for ( String tableName:DatabaseDescriptor.getTables() )
{
ColumnFamily hintColumnFamily = ColumnFamilyStore.removeDeleted(table.get(key_, Table.hints_), Integer.MAX_VALUE);
if (hintColumnFamily == null)
Table table = Table.open(tableName);
try
{
columnFamilyStore.forceFlush();
return;
}
Collection<IColumn> keys = hintColumnFamily.getAllColumns();
if (keys == null)
{
return;
}
for (IColumn keyColumn : keys)
{
Collection<IColumn> endpoints = keyColumn.getSubColumns();
int deleted = 0;
for (IColumn endpoint : endpoints)
ColumnFamily hintColumnFamily = ColumnFamilyStore.removeDeleted(table.get(key_, Table.hints_), Integer.MAX_VALUE);
if (hintColumnFamily == null)
{
if (sendMessage(endpoint.name(), keyColumn.name()))
columnFamilyStore.forceFlush();
return;
}
Collection<IColumn> keys = hintColumnFamily.getAllColumns();
if (keys == null)
{
return;
}
for (IColumn keyColumn : keys)
{
Collection<IColumn> endpoints = keyColumn.getSubColumns();
int deleted = 0;
for (IColumn endpoint : endpoints)
{
deleteEndPoint(endpoint.name(), keyColumn.name(), keyColumn.timestamp());
deleted++;
if (sendMessage(endpoint.name(), tableName, keyColumn.name()))
{
deleteEndPoint(endpoint.name(), tableName, keyColumn.name(), keyColumn.timestamp());
deleted++;
}
}
if (deleted == endpoints.size())
{
deleteHintedData(tableName, keyColumn.name());
}
}
if (deleted == endpoints.size())
{
deleteHintedData(keyColumn.name());
}
columnFamilyStore.forceFlush();
columnFamilyStore.forceCompaction(null, null, 0, null);
}
catch (Exception ex)
{
logger_.error(ex.getMessage());
}
finally
{
logger_.debug("Finished hinted handoff of " + columnFamilyStore.columnFamily_);
}
columnFamilyStore.forceFlush();
columnFamilyStore.forceCompaction(null, null, 0, null);
}
catch (Exception ex)
{
logger_.error(ex.getMessage());
}
finally
{
logger_.debug("Finished hinted handoff of " + columnFamilyStore.columnFamily_);
}
}
@ -207,43 +210,46 @@ public class HintedHandOffManager
// 1. Scan through all the keys that we need to handoff
// 2. For each key read the list of recepients if teh endpoint matches send
// 3. Delete that recepient from the key if write was successful
Table table = Table.open(DatabaseDescriptor.getTables().get(0));
try
for ( String tableName:DatabaseDescriptor.getTables() )
{
ColumnFamily hintedColumnFamily = table.get(key_, Table.hints_);
if (hintedColumnFamily == null)
Table table = Table.open(tableName);
try
{
return;
}
Collection<IColumn> keys = hintedColumnFamily.getAllColumns();
if (keys == null)
{
return;
}
for (IColumn keyColumn : keys)
{
Collection<IColumn> endpoints = keyColumn.getSubColumns();
for (IColumn endpoint : endpoints)
ColumnFamily hintedColumnFamily = table.get(key_, Table.hints_);
if (hintedColumnFamily == null)
{
if (endpoint.name().equals(endPoint.getHost()) && sendMessage(endpoint.name(), keyColumn.name()))
return;
}
Collection<IColumn> keys = hintedColumnFamily.getAllColumns();
if (keys == null)
{
return;
}
for (IColumn keyColumn : keys)
{
Collection<IColumn> endpoints = keyColumn.getSubColumns();
for (IColumn endpoint : endpoints)
{
deleteEndPoint(endpoint.name(), keyColumn.name(), keyColumn.timestamp());
if (endpoints.size() == 1)
if (endpoint.name().equals(endPoint.getHost()) && sendMessage(endpoint.name(), null, keyColumn.name()))
{
deleteHintedData(keyColumn.name());
deleteEndPoint(endpoint.name(), tableName, keyColumn.name(), keyColumn.timestamp());
if (endpoints.size() == 1)
{
deleteHintedData(tableName, keyColumn.name());
}
}
}
}
}
}
catch (Exception ex)
{
logger_.error(ex.getMessage());
}
finally
{
logger_.debug("Finished hinted handoff for endpoint " + endPoint.getHost());
catch (Exception ex)
{
logger_.error(ex.getMessage());
}
finally
{
logger_.debug("Finished hinted handoff for endpoint " + endPoint.getHost());
}
}
}

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

@ -253,7 +253,7 @@ public class Memtable implements Comparable<Memtable>
String directory = DatabaseDescriptor.getDataFileLocation();
String filename = cfStore.getTempFileName();
SSTable ssTable = new SSTable(directory, filename, StorageService.getPartitioner());
SSTable ssTable = new SSTable(directory, filename, table_, StorageService.getPartitioner());
// sort keys in the order they would be in when decorated
final IPartitioner partitioner = StorageService.getPartitioner();
@ -333,14 +333,14 @@ public class Memtable implements Comparable<Memtable>
if (cf != null)
columnFamily = cf.cloneMeShallow();
else
columnFamily = new ColumnFamily(cfName, DatabaseDescriptor.getColumnFamilyType(cfName));
columnFamily = ColumnFamily.create(table_, 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"))
if (DatabaseDescriptor.getColumnFamilyType(table_, cfName).equals("Standard"))
startIColumn = new Column(startColumn);
else
startIColumn = new SuperColumn(startColumn);

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

@ -114,7 +114,7 @@ class ReadResponseSerializer implements ICompactSerializer<ReadResponse>
if( !rm.isDigestQuery() && rm.row() != null )
{
Row.serializer().serialize(rm.row(), dos);
Row.serializer(rm.table()).serialize(rm.row(), dos);
}
}
@ -129,7 +129,7 @@ class ReadResponseSerializer implements ICompactSerializer<ReadResponse>
Row row = null;
if ( !isDigest )
{
row = Row.serializer().deserialize(dis);
row = Row.serializer(table).deserialize(dis);
}
ReadResponse rmsg = null;

27
src/java/org/apache/cassandra/db/Row.java
View File

@ -36,12 +36,21 @@ import org.apache.cassandra.utils.FBUtilities;
public class Row
{
private static RowSerializer serializer_ = new RowSerializer();
private static Logger logger_ = Logger.getLogger(Row.class);
private String table_;
static RowSerializer serializer()
public Row(String table_, String key) {
this.table_ = table_;
this.key_ = key;
}
public String getTable() {
return table_;
}
static RowSerializer serializer(String tableName)
{
return serializer_;
return new RowSerializer(tableName);
}
private String key_;
@ -130,7 +139,7 @@ public class Row
*/
public Row diff(Row rowComposite)
{
Row rowDiff = new Row(key_);
Row rowDiff = new Row(table_, key_);
for (ColumnFamily cfComposite : rowComposite.getColumnFamilies())
{
@ -152,7 +161,7 @@ public class Row
public Row cloneMe()
{
Row row = new Row(key_);
Row row = new Row(table_, key_);
row.columnFamilies_ = new HashMap<String, ColumnFamily>(columnFamilies_);
return row;
}
@ -188,6 +197,12 @@ public class Row
class RowSerializer implements ICompactSerializer<Row>
{
private String table_;
public RowSerializer(String tableName)
{
this.table_ = tableName;
}
public void serialize(Row row, DataOutputStream dos) throws IOException
{
dos.writeUTF(row.key());
@ -207,7 +222,7 @@ class RowSerializer implements ICompactSerializer<Row>
public Row deserialize(DataInputStream dis) throws IOException
{
String key = dis.readUTF();
Row row = new Row(key);
Row row = new Row(table_, key);
int size = dis.readInt();
if (size > 0)

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

@ -162,7 +162,7 @@ public class RowMutation implements Serializable
{
if ( columnFamily == null )
{
columnFamily = new ColumnFamily(values[0], ColumnFamily.getColumnType("Standard"));
columnFamily = ColumnFamily.create(table_, values[0]);
}
columnFamily.addColumn(values[1], value, timestamp);
}
@ -170,7 +170,7 @@ public class RowMutation implements Serializable
{
if ( columnFamily == null )
{
columnFamily = new ColumnFamily(values[0], ColumnFamily.getColumnType("Super"));
columnFamily = ColumnFamily.create(table_, values[0]);
}
columnFamily.addColumn(values[1]+ ":" + values[2], value, timestamp);
}
@ -193,7 +193,7 @@ public class RowMutation implements Serializable
ColumnFamily columnFamily = modifications_.get(cfName);
if (columnFamily == null)
columnFamily = new ColumnFamily(cfName, DatabaseDescriptor.getColumnType(cfName));
columnFamily = ColumnFamily.create(table_, cfName);
if (values.length == 2)
{
if (columnFamily.isSuper())
@ -229,7 +229,7 @@ public class RowMutation implements Serializable
*/
public void apply() throws IOException
{
Row row = new Row(key_);
Row row = new Row(table_, key_);
apply(row);
}
@ -329,7 +329,8 @@ public class RowMutation implements Serializable
public String toString()
{
return "RowMutation(" +
"key='" + key_ + '\'' +
"table='" + table_ + '\'' +
", key='" + key_ + '\'' +
", modifications=[" + StringUtils.join(modifications_.values(), ", ") + "]" +
')';
}

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

@ -82,7 +82,7 @@ public class SystemTable
table_ = table;
String systemTable = getFileName();
writer_ = SequenceFile.writer(systemTable);
reader_ = SequenceFile.reader(systemTable);
reader_ = SequenceFile.reader(systemTable, table);
}
private String getFileName()
@ -106,13 +106,13 @@ public class SystemTable
* This buffer contains key and value so we need to strip
* certain parts
*/
// read the key
// read the key
bufIn.readUTF();
// read the data length and then deserialize
bufIn.readInt();
try
{
systemRow_ = Row.serializer().deserialize(bufIn);
systemRow_ = Row.serializer(table_).deserialize(bufIn);
}
catch ( IOException e )
{
@ -133,7 +133,7 @@ public class SystemTable
String file = getFileName();
long currentPos = writer_.getCurrentPosition();
DataOutputBuffer bufOut = new DataOutputBuffer();
Row.serializer().serialize(row, bufOut);
Row.serializer(row.getTable()).serialize(row, bufOut);
try
{
writer_.append(row.key(), bufOut);

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

@ -67,6 +67,7 @@ public class Table
{
/* Name of the column family */
public final static String cfName_ = "TableMetadata";
private String table_;
private static ICompactSerializer<TableMetadata> serializer_;
static
{
@ -77,19 +78,28 @@ public class Table
/* Use the following writer/reader to write/read to Metadata table */
private static IFileWriter writer_;
private static IFileReader reader_;
public static Table.TableMetadata instance() throws IOException
private static HashMap<String,TableMetadata> tableMetadataMap_ = new HashMap<String,TableMetadata>();
private static TableMetadata getTableMetadata(String table)
{
if ( tableMetadata_ == null )
return tableMetadataMap_.get(table);
}
public static Table.TableMetadata instance(String table) throws IOException
{
TableMetadata metadata = getTableMetadata(table);
if ( metadata == null )
{
String file = getFileName();
String file = getFileName(table);
writer_ = SequenceFile.writer(file);
reader_ = SequenceFile.reader(file);
Table.TableMetadata.load();
if ( tableMetadata_ == null )
tableMetadata_ = new Table.TableMetadata();
reader_ = SequenceFile.reader(file, table);
Table.TableMetadata.load(table);
metadata = new Table.TableMetadata();
metadata.table_ = table;
tableMetadataMap_.put(table,metadata);
}
return tableMetadata_;
return metadata;
}
static ICompactSerializer<TableMetadata> serializer()
@ -97,9 +107,9 @@ public class Table
return serializer_;
}
private static void load() throws IOException
private static void load(String table) throws IOException
{
String file = Table.TableMetadata.getFileName();
String file = Table.TableMetadata.getFileName(table);
File f = new File(file);
if ( f.exists() )
{
@ -108,7 +118,7 @@ public class Table
if ( reader_ == null )
{
reader_ = SequenceFile.reader(file);
reader_ = SequenceFile.reader(file, table);
}
while ( !reader_.isEOF() )
@ -132,9 +142,8 @@ public class Table
private Map<String, Integer> cfIdMap_ = new HashMap<String, Integer>();
private Map<Integer, String> idCfMap_ = new HashMap<Integer, String>();
private static String getFileName()
private static String getFileName(String table)
{
String table = DatabaseDescriptor.getTables().get(0);
return DatabaseDescriptor.getMetadataDirectory() + System.getProperty("file.separator") + table + "-Metadata.db";
}
@ -187,12 +196,11 @@ public class Table
public void apply() throws IOException
{
String table = DatabaseDescriptor.getTables().get(0);
DataOutputBuffer bufOut = new DataOutputBuffer();
Table.TableMetadata.serializer_.serialize(this, bufOut);
try
{
writer_.append(table, bufOut);
writer_.append(table_, bufOut);
}
catch ( IOException ex )
{
@ -262,13 +270,17 @@ public class Table
{
File file = new File( streamContext.getTargetFile() );
String fileName = file.getName();
String [] temp = null;
String tableName;
temp = fileName.split("-");
tableName = temp[0];
/*
* If the file is a Data File we need to load the indicies associated
* with this file. We also need to cache the file name in the SSTables
* list of the associated Column Family. Also merge the CBF into the
* sampler.
*/
new SSTable(streamContext.getTargetFile(), StorageService.getPartitioner());
new SSTable(streamContext.getTargetFile(), tableName, StorageService.getPartitioner());
logger_.debug("Merging the counting bloom filter in the sampler ...");
String[] peices = FBUtilities.strip(fileName, "-");
Table.open(peices[0]).getColumnFamilyStore(peices[1]).addToList(streamContext.getTargetFile());
@ -348,17 +360,19 @@ public class Table
for ( StreamContextManager.StreamContext streamContext : streamContexts )
{
String[] peices = FBUtilities.strip(streamContext.getTargetFile(), "-");
distinctEntries.add(peices[1] + "-" + peices[2]);
distinctEntries.add(peices[0] + "-" + peices[1] + "-" + peices[2]);
}
/* Generate unique file names per entry */
Table table = Table.open( DatabaseDescriptor.getTables().get(0) );
Map<String, ColumnFamilyStore> columnFamilyStores = table.getColumnFamilyStores();
for ( String distinctEntry : distinctEntries )
{
String tableName;
String[] peices = FBUtilities.strip(distinctEntry, "-");
ColumnFamilyStore cfStore = columnFamilyStores.get(peices[0]);
tableName = peices[0];
Table table = Table.open( tableName );
Map<String, ColumnFamilyStore> columnFamilyStores = table.getColumnFamilyStores();
ColumnFamilyStore cfStore = columnFamilyStores.get(peices[1]);
logger_.debug("Generating file name for " + distinctEntry + " ...");
fileNames.put(distinctEntry, cfStore.getNextFileName());
}
@ -566,7 +580,7 @@ public class Table
dbAnalyticsSource_ = new DBAnalyticsSource();
try
{
tableMetadata_ = Table.TableMetadata.instance();
tableMetadata_ = Table.TableMetadata.instance(table);
Set<String> columnFamilies = tableMetadata_.getColumnFamilies();
for ( String columnFamily : columnFamilies )
{
@ -614,7 +628,7 @@ public class Table
*/
public Row get(String key) throws IOException
{
Row row = new Row(key);
Row row = new Row(table_, key);
Set<String> columnFamilies = tableMetadata_.getColumnFamilies();
long start = System.currentTimeMillis();
for ( String columnFamily : columnFamilies )
@ -654,7 +668,7 @@ public class Table
*/
public Row getRow(String key, String cf) throws IOException
{
Row row = new Row(key);
Row row = new Row(table_, key);
ColumnFamily columnFamily = get(key, cf);
if ( columnFamily != null )
row.addColumnFamily(columnFamily);
@ -666,7 +680,7 @@ public class Table
*/
public Row getRow(String key, String cf, int start, int count) throws IOException
{
Row row = new Row(key);
Row row = new Row(table_, key);
String[] values = RowMutation.getColumnAndColumnFamily(cf);
ColumnFamilyStore cfStore = columnFamilyStores_.get(values[0]);
long start1 = System.currentTimeMillis();
@ -693,7 +707,7 @@ public class Table
public Row getRow(String key, String cf, String startColumn, String endColumn, int count) throws IOException
{
Row row = new Row(key);
Row row = new Row(table_, key);
String[] values = RowMutation.getColumnAndColumnFamily(cf);
ColumnFamilyStore cfStore = columnFamilyStores_.get(values[0]);
long start1 = System.currentTimeMillis();
@ -712,7 +726,7 @@ public class Table
public Row getRow(String key, String cf, long sinceTimeStamp) throws IOException
{
Row row = new Row(key);
Row row = new Row(table_, key);
String[] values = RowMutation.getColumnAndColumnFamily(cf);
ColumnFamilyStore cfStore = columnFamilyStores_.get(values[0]);
long start1 = System.currentTimeMillis();
@ -735,7 +749,7 @@ public class Table
*/
public Row getRow(String key, String cf, List<String> columns) throws IOException
{
Row row = new Row(key);
Row row = new Row(table_, key);
String[] values = RowMutation.getColumnAndColumnFamily(cf);
ColumnFamilyStore cfStore = columnFamilyStores_.get(values[0]);
@ -948,7 +962,7 @@ public class Table
// sstables
for (String filename : cfs.getSSTableFilenames())
{
FileStruct fs = new FileStruct(SequenceFile.reader(filename), StorageService.getPartitioner());
FileStruct fs = new FileStruct(SequenceFile.reader(filename, table_), StorageService.getPartitioner());
fs.seekTo(startWith);
iterators.add(fs);
}

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

@ -131,12 +131,12 @@ public class IndexHelper
/**
* Deserialize the index into a structure and return the number of bytes read.
* @param in Input from which the serialized form of the index is read
* @param columnIndexList the structure which is filled in with the deserialized index
* @return number of bytes read from the input
* @param tableName
*@param in Input from which the serialized form of the index is read
* @param columnIndexList the structure which is filled in with the deserialized index @return number of bytes read from the input
* @throws IOException
*/
static int deserializeIndex(String cfName, DataInput in, List<ColumnIndexInfo> columnIndexList) throws IOException
static int deserializeIndex(String tableName, String cfName, DataInput in, List<ColumnIndexInfo> columnIndexList) throws IOException
{
/* read only the column index list */
int columnIndexSize = in.readInt();
@ -152,8 +152,8 @@ public class IndexHelper
DataInputBuffer indexIn = new DataInputBuffer();
indexIn.reset(indexOut.getData(), indexOut.getLength());
TypeInfo typeInfo = DatabaseDescriptor.getTypeInfo(cfName);
if ( DatabaseDescriptor.getColumnFamilyType(cfName).equals("Super") || DatabaseDescriptor.isNameSortingEnabled(cfName) )
TypeInfo typeInfo = DatabaseDescriptor.getTypeInfo(tableName, cfName);
if ( DatabaseDescriptor.getColumnFamilyType(tableName, cfName).equals("Super") || DatabaseDescriptor.isNameSortingEnabled(tableName, cfName) )
{
typeInfo = TypeInfo.STRING;
}

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

@ -150,13 +150,13 @@ public class SSTable
* associated with these files. Also caches the file handles
* associated with these files.
*/
public static void onStart(List<String> filenames) throws IOException
public static void onStart(List<String> filenames, String tableName) throws IOException
{
for (String filename : filenames)
{
try
{
new SSTable(filename, StorageService.getPartitioner());
new SSTable(filename, tableName, StorageService.getPartitioner());
}
catch (IOException ex)
{
@ -205,15 +205,17 @@ public class SSTable
private String lastWrittenKey_;
private IPartitioner partitioner_;
private String table_;
/**
* This ctor basically gets passed in the full path name
* of the data file associated with this SSTable. Use this
* ctor to read the data in this file.
*/
public SSTable(String dataFileName, IPartitioner partitioner) throws IOException
public SSTable(String dataFileName, String tableName, IPartitioner partitioner) throws IOException
{
dataFile_ = dataFileName;
partitioner_ = partitioner;
table_ = tableName;
/*
* this is to prevent multiple threads from
* loading the same index files multiple times
@ -235,11 +237,12 @@ public class SSTable
* This ctor is used for writing data into the SSTable. Use this
* version for non DB writes to the SSTable.
*/
public SSTable(String directory, String filename, IPartitioner partitioner) throws IOException
public SSTable(String directory, String filename, String tableName, IPartitioner partitioner) throws IOException
{
dataFile_ = directory + System.getProperty("file.separator") + filename + "-Data.db";
partitioner_ = partitioner;
dataWriter_ = SequenceFile.bufferedWriter(dataFile_, 4 * 1024 * 1024);
table_ = tableName;
indexRAF_ = new BufferedRandomAccessFile(indexFilename(), "rw", 1024 * 1024);
}
@ -474,7 +477,7 @@ public class SSTable
IFileReader dataReader = null;
try
{
dataReader = SequenceFile.reader(dataFile_);
dataReader = SequenceFile.reader(dataFile_, table_);
String decoratedKey = partitioner_.decorateKey(clientKey);
long position = getPosition(decoratedKey, dataReader, partitioner_);
@ -557,7 +560,7 @@ public class SSTable
public ColumnGroupReader getColumnGroupReader(String key, String cfName, String startColumn, boolean isAscending) throws IOException
{
ColumnGroupReader reader = null;
IFileReader dataReader = SequenceFile.reader(dataFile_);
IFileReader dataReader = SequenceFile.reader(table_, dataFile_);
try
{

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

@ -506,7 +506,7 @@ 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
@ -585,7 +585,7 @@ public class SequenceFile
/* read the index */
List<IndexHelper.ColumnIndexInfo> colIndexList = new ArrayList<IndexHelper.ColumnIndexInfo>();
if (hasColumnIndexes)
totalBytesRead += IndexHelper.deserializeIndex(cfName_, file_, colIndexList);
totalBytesRead += IndexHelper.deserializeIndex(null, cfName_, file_, colIndexList);
/* need to do two things here.
* 1. move the file pointer to the beginning of the list of stored columns
@ -651,10 +651,12 @@ public class SequenceFile
private static final short utfPrefix_ = 2;
protected RandomAccessFile file_;
protected String filename_;
private String table_;
AbstractReader(String filename)
AbstractReader(String filename, String tableName)
{
filename_ = filename;
table_ = tableName;
}
public String getFileName()
@ -701,10 +703,10 @@ public class SequenceFile
/* if we do then deserialize the index */
if (hasColumnIndexes)
{
if (DatabaseDescriptor.isNameSortingEnabled(cfName) || DatabaseDescriptor.getColumnFamilyType(cfName).equals("Super"))
if (DatabaseDescriptor.isNameSortingEnabled(table_, cfName) || DatabaseDescriptor.getColumnFamilyType(table_, cfName).equals("Super"))
{
/* read the index */
totalBytesRead += IndexHelper.deserializeIndex(cfName, file_, columnIndexList);
totalBytesRead += IndexHelper.deserializeIndex(table_, cfName, file_, columnIndexList);
}
else
{
@ -729,10 +731,10 @@ public class SequenceFile
/* if we do then deserialize the index */
if (hasColumnIndexes)
{
if (DatabaseDescriptor.isTimeSortingEnabled(cfName))
if (DatabaseDescriptor.isTimeSortingEnabled(null, cfName))
{
/* read the index */
totalBytesRead += IndexHelper.deserializeIndex(cfName, file_, columnIndexList);
totalBytesRead += IndexHelper.deserializeIndex(table_, cfName, file_, columnIndexList);
}
else
{
@ -829,7 +831,7 @@ public class SequenceFile
/* read the column family name */
String cfName = file_.readUTF();
dataSize -= (utfPrefix_ + cfName.length());
/* read local deletion time */
int localDeletionTime = file_.readInt();
dataSize -=4;
@ -997,10 +999,10 @@ public class SequenceFile
/*
* If we have read the bloom filter in the data
* file we know we are at the end of the file
* file we know we are at the end of the file
* and no further key processing is required. So
* we return -1 indicating we are at the end of
* the file.
* the file.
*/
if (key.equals(SequenceFile.marker_))
bytesRead = -1L;
@ -1010,9 +1012,9 @@ public class SequenceFile
public static class Reader extends AbstractReader
{
Reader(String filename) throws IOException
Reader(String filename, String tableName) throws IOException
{
super(filename);
super(filename, tableName);
init(filename);
}
@ -1075,7 +1077,7 @@ public class SequenceFile
BufferReader(String filename, int size) throws IOException
{
super(filename);
super(filename, null);
size_ = size;
}
@ -1104,9 +1106,9 @@ public class SequenceFile
return new FastConcurrentWriter(filename, size);
}
public static IFileReader reader(String filename) throws IOException
public static IFileReader reader(String filename, String tableName) throws IOException
{
return new Reader(filename);
return new Reader(filename, tableName);
}
public static IFileReader bufferedReader(String filename, int size) throws IOException

81
src/java/org/apache/cassandra/loader/Loader.java
View File

@ -294,46 +294,51 @@ public class Loader
void preLoad(File rootDirectory) throws Throwable
{
String table = DatabaseDescriptor.getTables().get(0);
String cfName = Table.recycleBin_ + ":" + "Keys";
/* populate just the keys. */
preParse(rootDirectory, table, cfName);
/* dump the memtables */
Table.open(table).flush(false);
/* force a compaction of the files. */
Table.open(table).forceCompaction(null,null,null);
/*
* This is a hack to let everyone finish. Just sleep for
* a couple of minutes.
*/
logger_.info("Taking a nap after forcing a compaction ...");
Thread.sleep(Loader.siesta_);
/* Figure out the keys in the index file to relocate the node */
List<String> ssTables = Table.open(table).getAllSSTablesOnDisk();
/* Load the indexes into memory */
for ( String df : ssTables )
List<String> tables = DatabaseDescriptor.getTables();
for ( String table:tables )
{
new SSTable(df, StorageService.getPartitioner());
String cfName = Table.recycleBin_ + ":" + "Keys";
/* populate just the keys. */
preParse(rootDirectory, table, cfName);
/* dump the memtables */
Table.open(table).flush(false);
/* force a compaction of the files. */
Table.open(table).forceCompaction(null,null,null);
/*
* This is a hack to let everyone finish. Just sleep for
* a couple of minutes.
*/
logger_.info("Taking a nap after forcing a compaction ...");
Thread.sleep(Loader.siesta_);
/* Figure out the keys in the index file to relocate the node */
List<String> ssTables = Table.open(table).getAllSSTablesOnDisk();
/* Load the indexes into memory */
for ( String df : ssTables )
{
new SSTable(df, table, StorageService.getPartitioner());
}
/* We should have only one file since we just compacted. */
List<String> indexedKeys = SSTable.getIndexedKeys();
storageService_.relocate(indexedKeys.toArray( new String[0]) );
/*
* This is a hack to let everyone relocate and learn about
* each other. Just sleep for a couple of minutes.
*/
logger_.info("Taking a nap after relocating ...");
Thread.sleep(Loader.siesta_);
/*
* Do the cleanup necessary. Delete all commit logs and
* the SSTables and reset the load state in the StorageService.
*/
// TODO Hmm need to double check here
SSTable.delete(ssTables.get(0));
logger_.info("Finished all the requisite clean up ...");
}
/* We should have only one file since we just compacted. */
List<String> indexedKeys = SSTable.getIndexedKeys();
storageService_.relocate(indexedKeys.toArray( new String[0]) );
/*
* This is a hack to let everyone relocate and learn about
* each other. Just sleep for a couple of minutes.
*/
logger_.info("Taking a nap after relocating ...");
Thread.sleep(Loader.siesta_);
/*
* Do the cleanup necessary. Delete all commit logs and
* the SSTables and reset the load state in the StorageService.
*/
SSTable.delete(ssTables.get(0));
logger_.info("Finished all the requisite clean up ...");
}
void load(String xmlFile) throws Throwable

77
src/java/org/apache/cassandra/loader/PreLoad.java
View File

@ -77,45 +77,48 @@ public class PreLoad
void run(String userFile) throws Throwable
{
String table = DatabaseDescriptor.getTables().get(0);
String cfName = Table.recycleBin_ + ":" + "Keys";
/* populate just the keys. */
preParse(userFile, table, cfName);
/* dump the memtables */
Table.open(table).flush(false);
/* force a compaction of the files. */
Table.open(table).forceCompaction(null, null,null);
/*
* This is a hack to let everyone finish. Just sleep for
* a couple of minutes.
*/
logger_.info("Taking a nap after forcing a compaction ...");
Thread.sleep(PreLoad.siesta_);
/* Figure out the keys in the index file to relocate the node */
List<String> ssTables = Table.open(table).getAllSSTablesOnDisk();
/* Load the indexes into memory */
for ( String df : ssTables )
List<String> tables = DatabaseDescriptor.getTables();
for ( String table:tables )
{
new SSTable(df, StorageService.getPartitioner());
String cfName = Table.recycleBin_ + ":" + "Keys";
/* populate just the keys. */
preParse(userFile, table, cfName);
/* dump the memtables */
Table.open(table).flush(false);
/* force a compaction of the files. */
Table.open(table).forceCompaction(null, null,null);
/*
* This is a hack to let everyone finish. Just sleep for
* a couple of minutes.
*/
logger_.info("Taking a nap after forcing a compaction ...");
Thread.sleep(PreLoad.siesta_);
/* Figure out the keys in the index file to relocate the node */
List<String> ssTables = Table.open(table).getAllSSTablesOnDisk();
/* Load the indexes into memory */
for ( String df : ssTables )
{
new SSTable(df, table, StorageService.getPartitioner());
}
/* We should have only one file since we just compacted. */
List<String> indexedKeys = SSTable.getIndexedKeys();
storageService_.relocate(indexedKeys.toArray( new String[0]) );
/*
* This is a hack to let everyone relocate and learn about
* each other. Just sleep for a couple of minutes.
*/
logger_.info("Taking a nap after relocating ...");
Thread.sleep(PreLoad.siesta_);
/*
* Do the cleanup necessary. Delete all commit logs and
* the SSTables and reset the load state in the StorageService.
*/
SSTable.delete(ssTables.get(0));
}
/* We should have only one file since we just compacted. */
List<String> indexedKeys = SSTable.getIndexedKeys();
storageService_.relocate(indexedKeys.toArray( new String[0]) );
/*
* This is a hack to let everyone relocate and learn about
* each other. Just sleep for a couple of minutes.
*/
logger_.info("Taking a nap after relocating ...");
Thread.sleep(PreLoad.siesta_);
/*
* Do the cleanup necessary. Delete all commit logs and
* the SSTables and reset the load state in the StorageService.
*/
SSTable.delete(ssTables.get(0));
logger_.info("Finished all the requisite clean up ...");
}

23
src/java/org/apache/cassandra/net/http/HttpRequestHandler.java
View File

@ -250,18 +250,23 @@ public class HttpRequestHandler implements Runnable
private void displayDBStatistics(HTMLFormatter formatter, java.text.DecimalFormat df)
{
String tableStats = Table.open( DatabaseDescriptor.getTables().get(0) ).tableStats("\n<br>\n", df);
if ( tableStats.length() == 0 )
return;
List<String> tables = DatabaseDescriptor.getTables();
for ( String table:tables )
{
String tableStats = Table.open( table ).tableStats("\n<br>\n", df);
formatter.appendLine("DB statistics:");
formatter.appendLine("<br>");
formatter.appendLine("<br>");
if ( tableStats.length() == 0 )
return;
formatter.appendLine(tableStats);
formatter.appendLine("<br>");
formatter.appendLine("<br>");
formatter.appendLine("DB statistics: " + table);
formatter.appendLine("<br>");
formatter.appendLine("<br>");
formatter.appendLine(tableStats);
formatter.appendLine("<br>");
formatter.appendLine("<br>");
}
}
private String handlePageDisplay(String queryFormData, String insertFormData, String scriptFormData)

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

@ -205,7 +205,7 @@ public class CassandraServer implements Cassandra.Iface
{
logger.debug("get_slice_from");
String[] values = RowMutation.getColumnAndColumnFamily(columnParent);
if (values.length != 2 || DatabaseDescriptor.getColumnFamilyType(values[0]) != "Standard")
if (values.length != 2 || DatabaseDescriptor.getColumnFamilyType(tablename, 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");
@ -228,7 +228,7 @@ public class CassandraServer implements Cassandra.Iface
{
throw new InvalidRequestException("get_column requires non-empty columnfamily");
}
if (DatabaseDescriptor.getColumnFamilyType(values[0]).equals("Standard"))
if (DatabaseDescriptor.getColumnFamilyType(null, values[0]).equals("Standard"))
{
if (values.length != 2)
{

2
src/java/org/apache/cassandra/service/ReadResponseResolver.java
View File

@ -121,7 +121,7 @@ public class ReadResponseResolver implements IResponseResolver<Row>
}
/* Now calculate the resolved row */
retRow = new Row(key);
retRow = new Row(table, key);
for (int i = 0 ; i < rowList.size(); i++)
{
retRow.repair(rowList.get(i));

2
src/java/org/apache/cassandra/service/StorageProxy.java
View File

@ -596,7 +596,7 @@ public class StorageProxy implements StorageProxyMBean
endpoints.remove(StorageService.getLocalStorageEndPoint());
// TODO: throw a thrift exception if we do not have N nodes
Table table = Table.open(DatabaseDescriptor.getTables().get(0));
Table table = Table.open(command.table);
Row row = command.getRow(table);
/*

16
test/unit/org/apache/cassandra/db/ColumnFamilyTest.java
View File

@ -41,7 +41,7 @@ public class ColumnFamilyTest
random.nextBytes(bytes);
ColumnFamily cf;
cf = new ColumnFamily("Standard1", "Standard");
cf = ColumnFamily.create("Table1", "Standard1");
cf.addColumn("C", bytes, 1);
DataOutputBuffer bufOut = new DataOutputBuffer();
ColumnFamily.serializer().serialize(cf, bufOut);
@ -66,7 +66,7 @@ public class ColumnFamilyTest
}
// write
cf = new ColumnFamily("Standard1", "Standard");
cf = ColumnFamily.create("Table1", "Standard1");
DataOutputBuffer bufOut = new DataOutputBuffer();
for (String cName : map.navigableKeySet())
{
@ -89,7 +89,7 @@ public class ColumnFamilyTest
@Test
public void testGetColumnCount()
{
ColumnFamily cf = new ColumnFamily("Standard1", "Standard");
ColumnFamily cf = ColumnFamily.create("Table1", "Standard1");
byte val[] = "sample value".getBytes();
cf.addColumn("col1", val, 1);
@ -103,7 +103,7 @@ public class ColumnFamilyTest
@Test
public void testTimestamp()
{
ColumnFamily cf = new ColumnFamily("Standard1", "Standard");
ColumnFamily cf = ColumnFamily.create("Table1", "Standard1");
byte val1[] = "sample 1".getBytes();
byte val2[] = "sample 2".getBytes();
byte val3[] = "sample 3".getBytes();
@ -118,9 +118,9 @@ public class ColumnFamilyTest
@Test
public void testMergeAndAdd()
{
ColumnFamily cf_new = new ColumnFamily("Standard1", "Standard");
ColumnFamily cf_old = new ColumnFamily("Standard1", "Standard");
ColumnFamily cf_result = new ColumnFamily("Standard1", "Standard");
ColumnFamily cf_new = ColumnFamily.create("Table1", "Standard1");
ColumnFamily cf_old = ColumnFamily.create("Table1", "Standard1");
ColumnFamily cf_result = ColumnFamily.create("Table1", "Standard1");
byte val[] = "sample value".getBytes();
byte val2[] = "x value ".getBytes();
@ -141,7 +141,7 @@ public class ColumnFamilyTest
@Test
public void testEmptyDigest()
{
ColumnFamily cf = new ColumnFamily("Standard1", "Standard");
ColumnFamily cf = ColumnFamily.create("Table1", "Standard1");
assert cf.digest().length == 0;
}
}

6
test/unit/org/apache/cassandra/db/RangeFilterTest.java
View File

@ -29,7 +29,7 @@ public class RangeFilterTest
@Test
public void testRangeFilterOnColumns() throws IOException
{
ColumnFamily cf = new ColumnFamily("Standard1", "Standard");
ColumnFamily cf = ColumnFamily.create("Table1", "Standard1");
byte[] val = "test value".getBytes();
cf.addColumn(new Column("a", val, System.currentTimeMillis()));
cf.addColumn(new Column("b", val, System.currentTimeMillis()));
@ -47,7 +47,7 @@ public class RangeFilterTest
@Test
public void testRangeFilterOnColumnsWithCount() throws IOException
{
ColumnFamily cf = new ColumnFamily("Standard1", "Standard");
ColumnFamily cf = ColumnFamily.create("Table1", "Standard1");
byte[] val = "test value".getBytes();
cf.addColumn(new Column("a", val, System.currentTimeMillis()));
cf.addColumn(new Column("b", val, System.currentTimeMillis()));
@ -65,7 +65,7 @@ public class RangeFilterTest
@Test
public void testRangeFilterOnSuperColumns() throws IOException
{
ColumnFamily cf = new ColumnFamily("Super1", "Super");
ColumnFamily cf = ColumnFamily.create("Table1", "Super1");
byte[] val = "test value".getBytes();
SuperColumn sc = null;
sc = new SuperColumn("a");

10
test/unit/org/apache/cassandra/db/RowTest.java
View File

@ -29,10 +29,10 @@ public class RowTest
@Test
public void testDiffColumnFamily()
{
ColumnFamily cf1 = new ColumnFamily("Standard1", "Standard");
ColumnFamily cf1 = ColumnFamily.create("Table1", "Standard1");
cf1.addColumn("one", "onev".getBytes(), 0);
ColumnFamily cf2 = new ColumnFamily("Standard1", "Standard");
ColumnFamily cf2 = ColumnFamily.create("Table1", "Standard1");
cf2.delete(0, 0);
ColumnFamily cfDiff = cf1.diff(cf2);
@ -58,15 +58,15 @@ public class RowTest
public void testRepair()
{
Row row1 = new Row();
ColumnFamily cf1 = new ColumnFamily("Standard1", "Standard");
ColumnFamily cf1 = ColumnFamily.create("Table1", "Standard1");
cf1.addColumn("one", "A".getBytes(), 0);
row1.addColumnFamily(cf1);
Row row2 = new Row();
ColumnFamily cf2 = new ColumnFamily("Standard1", "Standard");
ColumnFamily cf2 = ColumnFamily.create("Table1", "Standard1");
cf2.addColumn("one", "B".getBytes(), 1);
cf2.addColumn("two", "C".getBytes(), 1);
ColumnFamily cf3 = new ColumnFamily("Standard2", "Standard");
ColumnFamily cf3 = ColumnFamily.create("Table2", "Standard2");
cf3.addColumn("three", "D".getBytes(), 1);
row2.addColumnFamily(cf2);
row2.addColumnFamily(cf3);

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

@ -89,7 +89,7 @@ public class TableTest extends CleanupHelper
Table table = Table.open("Table1");
RowMutation rm = new RowMutation(TABLE_NAME,KEY2);
ColumnFamily cf = new ColumnFamily("Standard1","Standard");
ColumnFamily cf = ColumnFamily.create("Table1", "Standard1");
// First write 5, 6
cf.addColumn(new Column("col5", "val5".getBytes(), 1L));
cf.addColumn(new Column("col6", "val6".getBytes(), 1L));
@ -104,7 +104,7 @@ public class TableTest extends CleanupHelper
// Flushed memtable to disk, we're now inserting into a new memtable
rm = new RowMutation(TABLE_NAME, KEY2);
cf = new ColumnFamily("Standard1","Standard");
cf = ColumnFamily.create("Table1", "Standard1");
// now write 7, 8, 4 into new memtable
cf.addColumn(new Column("col7", "val7".getBytes(), 1L));
cf.addColumn(new Column("col8", "val8".getBytes(), 1L));
@ -137,7 +137,7 @@ public class TableTest extends CleanupHelper
String key = TEST_KEY+"slicerow";
Table table = Table.open(TABLE_NAME);
RowMutation rm = new RowMutation(TABLE_NAME,key);
ColumnFamily cf = new ColumnFamily("Standard1","Standard");
ColumnFamily cf = ColumnFamily.create("Table1", "Standard1");
// First write "a", "b", "c"
cf.addColumn(new Column("a", "val1".getBytes(), 1L));
cf.addColumn(new Column("b", "val2".getBytes(), 1L));
@ -165,7 +165,7 @@ public class TableTest extends CleanupHelper
Table table = Table.open(TABLE_NAME);
RowMutation rm = new RowMutation(TABLE_NAME,TEST_KEY);
ColumnFamily cf = new ColumnFamily("Super1","Super");
ColumnFamily cf = ColumnFamily.create("Table1", "Super1");
SuperColumn sc1 = new SuperColumn("sc1");
sc1.addColumn(new Column("col1","val1".getBytes(), 1L));
sc1.addColumn(new Column("col2","val2".getBytes(), 1L));
@ -203,7 +203,7 @@ public class TableTest extends CleanupHelper
private RowMutation makeSimpleRowMutation()
{
RowMutation rm = new RowMutation(TABLE_NAME,TEST_KEY);
ColumnFamily cf = new ColumnFamily("Standard1", "Standard");
ColumnFamily cf = ColumnFamily.create("Table1", "Standard1");
cf.addColumn(new Column("col1","val1".getBytes(), 1L));
cf.addColumn(new Column("col2","val2".getBytes(), 1L));
cf.addColumn(new Column("col3","val3".getBytes(), 1L));
@ -216,7 +216,7 @@ public class TableTest extends CleanupHelper
{
Table table = Table.open(TABLE_NAME);
RowMutation rm = new RowMutation(TABLE_NAME, "row1000");
ColumnFamily cf = new ColumnFamily("Standard2", "Standard");
ColumnFamily cf = ColumnFamily.create("Table1", "Standard2");
cf.addColumn(new Column("col1", "val1".getBytes(), 1));
rm.add(cf);
rm.apply();
@ -254,7 +254,7 @@ public class TableTest extends CleanupHelper
Table table = Table.open(TABLE_NAME);
String ROW = "row1";
RowMutation rm = new RowMutation(TABLE_NAME, ROW);
ColumnFamily cf = new ColumnFamily("Standard1", "Standard");
ColumnFamily cf = ColumnFamily.create("Table1", "Standard1");
cf.addColumn(new Column("col1", "val1".getBytes(), 1L));
cf.addColumn(new Column("col3", "val3".getBytes(), 1L));
cf.addColumn(new Column("col4", "val4".getBytes(), 1L));
@ -280,7 +280,7 @@ public class TableTest extends CleanupHelper
Table table = Table.open(TABLE_NAME);
String ROW = "row2";
RowMutation rm = new RowMutation(TABLE_NAME, ROW);
ColumnFamily cf = new ColumnFamily("Standard1", "Standard");
ColumnFamily cf = ColumnFamily.create("Table1", "Standard1");
cf.addColumn(new Column("col1", "val1".getBytes(), 1L));
cf.addColumn(new Column("col2", "val2".getBytes(), 1L));
cf.addColumn(new Column("col3", "val3".getBytes(), 1L));
@ -292,7 +292,7 @@ public class TableTest extends CleanupHelper
table.getColumnFamilyStore("Standard1").forceBlockingFlush();
rm = new RowMutation(TABLE_NAME, ROW);
cf = new ColumnFamily("Standard1", "Standard");
cf = ColumnFamily.create("Table1", "Standard1");
cf.addColumn(new Column("col1", "valx".getBytes(), 2L));
cf.addColumn(new Column("col2", "valx".getBytes(), 2L));
cf.addColumn(new Column("col3", "valx".getBytes(), 2L));
@ -311,7 +311,7 @@ public class TableTest extends CleanupHelper
Table table = Table.open(TABLE_NAME);
String ROW = "row3";
RowMutation rm = new RowMutation(TABLE_NAME, ROW);
ColumnFamily cf = new ColumnFamily("Standard1", "Standard");
ColumnFamily cf = ColumnFamily.create("Table1", "Standard1");
for (int i = 1000; i < 2000; i++)
cf.addColumn(new Column("col" + i, ("vvvvvvvvvvvvvvvv" + i).getBytes(), 1L));
rm.add(cf);

4
test/unit/org/apache/cassandra/io/SSTableTest.java
View File

@ -56,7 +56,7 @@ public class SSTableTest extends CleanupHelper
private void verifySingle(File f, byte[] bytes, String key) throws IOException
{
SSTable ssTable = new SSTable(f.getPath() + "-Data.db", new OrderPreservingPartitioner());
SSTable ssTable = new SSTable(f.getPath() + "-Data.db", "Table1", new OrderPreservingPartitioner());
FileStruct fs = new FileStruct(SequenceFile.bufferedReader(ssTable.dataFile_, 128 * 1024), new OrderPreservingPartitioner());
fs.seekTo(key);
int size = fs.getBufIn().readInt();
@ -95,7 +95,7 @@ public class SSTableTest extends CleanupHelper
{
List<String> keys = new ArrayList(map.keySet());
Collections.shuffle(keys);
SSTable ssTable = new SSTable(f.getPath() + "-Data.db", new OrderPreservingPartitioner());
SSTable ssTable = new SSTable(f.getPath() + "-Data.db", "Table1", new OrderPreservingPartitioner());
FileStruct fs = new FileStruct(SequenceFile.bufferedReader(ssTable.dataFile_, 128 * 1024), new OrderPreservingPartitioner());
for (String key : keys)
{