diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index 98347ab152..35398f5e35 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -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 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 getTableMetaData(String table) + public static Map 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 cfInfo = tableToCFMetaDataMap_.get(table); + assert tableName != null; + Map 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 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; diff --git a/src/java/org/apache/cassandra/db/BinaryMemtable.java b/src/java/org/apache/cassandra/db/BinaryMemtable.java index 6f209ff87d..2ccb6e46f7 100644 --- a/src/java/org/apache/cassandra/db/BinaryMemtable.java +++ b/src/java/org/apache/cassandra/db/BinaryMemtable.java @@ -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 keys = new ArrayList( columnFamilies_.keySet() ); Collections.sort(keys); /* Use this BloomFilter to decide if a key exists in a SSTable */ diff --git a/src/java/org/apache/cassandra/db/ColumnComparatorFactory.java b/src/java/org/apache/cassandra/db/ColumnComparatorFactory.java index 17b1409f6b..5277f58fcc 100644 --- a/src/java/org/apache/cassandra/db/ColumnComparatorFactory.java +++ b/src/java/org/apache/cassandra/db/ColumnComparatorFactory.java @@ -34,8 +34,8 @@ public class ColumnComparatorFactory TIMESTAMP } - private static Comparator nameComparator_ = new ColumnNameComparator(); - private static Comparator timestampComparator_ = new ColumnTimestampComparator(); + public static final Comparator nameComparator_ = new ColumnNameComparator(); + public static final Comparator timestampComparator_ = new ColumnTimestampComparator(); public static Comparator getComparator(ComparatorType comparatorType) { diff --git a/src/java/org/apache/cassandra/db/ColumnFamily.java b/src/java/org/apache/cassandra/db/ColumnFamily.java index 9724ec827e..617a2f8aa3 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamily.java +++ b/src/java/org/apache/cassandra/db/ColumnFamily.java @@ -51,6 +51,7 @@ public final class ColumnFamily private static Map columnTypes_ = new HashMap(); private static Map indexTypes_ = new HashMap(); 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 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 columnComparator_; - - private Comparator 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 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 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 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 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 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."); } } + } diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index bca4a5674e..bcd346bf6c 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -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 reduced = new ReducingIterator(collated) { ColumnFamily curCF = returnCF.cloneMeShallow(); diff --git a/src/java/org/apache/cassandra/db/ColumnIndexer.java b/src/java/org/apache/cassandra/db/ColumnIndexer.java index d08da7e168..6c0c2356eb 100644 --- a/src/java/org/apache/cassandra/db/ColumnIndexer.java +++ b/src/java/org/apache/cassandra/db/ColumnIndexer.java @@ -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); } diff --git a/src/java/org/apache/cassandra/db/ColumnIterator.java b/src/java/org/apache/cassandra/db/ColumnIterator.java index fab378dc98..04b9bda51c 100644 --- a/src/java/org/apache/cassandra/db/ColumnIterator.java +++ b/src/java/org/apache/cassandra/db/ColumnIterator.java @@ -58,7 +58,7 @@ class SSTableColumnIterator extends AbstractIterator 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; diff --git a/src/java/org/apache/cassandra/db/CommitLog.java b/src/java/org/apache/cassandra/db/CommitLog.java index fff4434d28..0f9a8749fe 100644 --- a/src/java/org/apache/cassandra/db/CommitLog.java +++ b/src/java/org/apache/cassandra/db/CommitLog.java @@ -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 columnFamilies = new ArrayList(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 */ diff --git a/src/java/org/apache/cassandra/db/DBManager.java b/src/java/org/apache/cassandra/db/DBManager.java index 5b7a3f07dd..5c9e9312b5 100644 --- a/src/java/org/apache/cassandra/db/DBManager.java +++ b/src/java/org/apache/cassandra/db/DBManager.java @@ -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); diff --git a/src/java/org/apache/cassandra/db/HintedHandOffManager.java b/src/java/org/apache/cassandra/db/HintedHandOffManager.java index 3b7f2fe6c4..84347689dd 100644 --- a/src/java/org/apache/cassandra/db/HintedHandOffManager.java +++ b/src/java/org/apache/cassandra/db/HintedHandOffManager.java @@ -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 quorumResponseHandler = new QuorumResponseHandler(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 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 keys = hintColumnFamily.getAllColumns(); - if (keys == null) - { - return; - } - - for (IColumn keyColumn : keys) - { - Collection 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 keys = hintColumnFamily.getAllColumns(); + if (keys == null) + { + return; + } + + for (IColumn keyColumn : keys) + { + Collection 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 keys = hintedColumnFamily.getAllColumns(); - if (keys == null) - { - return; - } - - for (IColumn keyColumn : keys) - { - Collection 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 keys = hintedColumnFamily.getAllColumns(); + if (keys == null) + { + return; + } + + for (IColumn keyColumn : keys) + { + Collection 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()); + } } } diff --git a/src/java/org/apache/cassandra/db/Memtable.java b/src/java/org/apache/cassandra/db/Memtable.java index 24125b3c98..95c27d46a2 100644 --- a/src/java/org/apache/cassandra/db/Memtable.java +++ b/src/java/org/apache/cassandra/db/Memtable.java @@ -253,7 +253,7 @@ public class Memtable implements Comparable 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 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); diff --git a/src/java/org/apache/cassandra/db/ReadResponse.java b/src/java/org/apache/cassandra/db/ReadResponse.java index e6e069f29a..0ea6bccd84 100644 --- a/src/java/org/apache/cassandra/db/ReadResponse.java +++ b/src/java/org/apache/cassandra/db/ReadResponse.java @@ -114,7 +114,7 @@ class ReadResponseSerializer implements ICompactSerializer 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 Row row = null; if ( !isDigest ) { - row = Row.serializer().deserialize(dis); + row = Row.serializer(table).deserialize(dis); } ReadResponse rmsg = null; diff --git a/src/java/org/apache/cassandra/db/Row.java b/src/java/org/apache/cassandra/db/Row.java index 97e31b7974..8e56e0708b 100644 --- a/src/java/org/apache/cassandra/db/Row.java +++ b/src/java/org/apache/cassandra/db/Row.java @@ -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(columnFamilies_); return row; } @@ -188,6 +197,12 @@ public class Row class RowSerializer implements ICompactSerializer { + 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 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) diff --git a/src/java/org/apache/cassandra/db/RowMutation.java b/src/java/org/apache/cassandra/db/RowMutation.java index b2c13a3555..cc9607d6aa 100644 --- a/src/java/org/apache/cassandra/db/RowMutation.java +++ b/src/java/org/apache/cassandra/db/RowMutation.java @@ -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(), ", ") + "]" + ')'; } diff --git a/src/java/org/apache/cassandra/db/SystemTable.java b/src/java/org/apache/cassandra/db/SystemTable.java index 589b96e819..22a84c7514 100644 --- a/src/java/org/apache/cassandra/db/SystemTable.java +++ b/src/java/org/apache/cassandra/db/SystemTable.java @@ -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); diff --git a/src/java/org/apache/cassandra/db/Table.java b/src/java/org/apache/cassandra/db/Table.java index a2a3d83e53..9aee9e6aa6 100644 --- a/src/java/org/apache/cassandra/db/Table.java +++ b/src/java/org/apache/cassandra/db/Table.java @@ -67,6 +67,7 @@ public class Table { /* Name of the column family */ public final static String cfName_ = "TableMetadata"; + private String table_; private static ICompactSerializer 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 tableMetadataMap_ = new HashMap(); + + 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 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 cfIdMap_ = new HashMap(); private Map idCfMap_ = new HashMap(); - 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 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 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 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 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 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); } diff --git a/src/java/org/apache/cassandra/io/IndexHelper.java b/src/java/org/apache/cassandra/io/IndexHelper.java index 888a59a7ba..1f64bb8359 100644 --- a/src/java/org/apache/cassandra/io/IndexHelper.java +++ b/src/java/org/apache/cassandra/io/IndexHelper.java @@ -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 columnIndexList) throws IOException + static int deserializeIndex(String tableName, String cfName, DataInput in, List 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; } diff --git a/src/java/org/apache/cassandra/io/SSTable.java b/src/java/org/apache/cassandra/io/SSTable.java index 0212e9113b..dfc47a6158 100644 --- a/src/java/org/apache/cassandra/io/SSTable.java +++ b/src/java/org/apache/cassandra/io/SSTable.java @@ -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 filenames) throws IOException + public static void onStart(List 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 { diff --git a/src/java/org/apache/cassandra/io/SequenceFile.java b/src/java/org/apache/cassandra/io/SequenceFile.java index 571c14a01f..4db226ccef 100644 --- a/src/java/org/apache/cassandra/io/SequenceFile.java +++ b/src/java/org/apache/cassandra/io/SequenceFile.java @@ -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 colIndexList = new ArrayList(); 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 diff --git a/src/java/org/apache/cassandra/loader/Loader.java b/src/java/org/apache/cassandra/loader/Loader.java index 6e2532be58..01679b1921 100644 --- a/src/java/org/apache/cassandra/loader/Loader.java +++ b/src/java/org/apache/cassandra/loader/Loader.java @@ -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 ssTables = Table.open(table).getAllSSTablesOnDisk(); - /* Load the indexes into memory */ - for ( String df : ssTables ) + List 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 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 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 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 diff --git a/src/java/org/apache/cassandra/loader/PreLoad.java b/src/java/org/apache/cassandra/loader/PreLoad.java index 861f0b9246..e4e0bbe1f9 100644 --- a/src/java/org/apache/cassandra/loader/PreLoad.java +++ b/src/java/org/apache/cassandra/loader/PreLoad.java @@ -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 ssTables = Table.open(table).getAllSSTablesOnDisk(); - /* Load the indexes into memory */ - for ( String df : ssTables ) + List 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 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 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 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 ..."); } diff --git a/src/java/org/apache/cassandra/net/http/HttpRequestHandler.java b/src/java/org/apache/cassandra/net/http/HttpRequestHandler.java index e0572677ae..c6f38b1d1d 100644 --- a/src/java/org/apache/cassandra/net/http/HttpRequestHandler.java +++ b/src/java/org/apache/cassandra/net/http/HttpRequestHandler.java @@ -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
\n", df); - if ( tableStats.length() == 0 ) - return; + List tables = DatabaseDescriptor.getTables(); + for ( String table:tables ) + { + String tableStats = Table.open( table ).tableStats("\n
\n", df); - formatter.appendLine("DB statistics:"); - formatter.appendLine("
"); - formatter.appendLine("
"); + if ( tableStats.length() == 0 ) + return; - formatter.appendLine(tableStats); - formatter.appendLine("
"); - formatter.appendLine("
"); + formatter.appendLine("DB statistics: " + table); + formatter.appendLine("
"); + formatter.appendLine("
"); + + formatter.appendLine(tableStats); + formatter.appendLine("
"); + formatter.appendLine("
"); + } } private String handlePageDisplay(String queryFormData, String insertFormData, String scriptFormData) diff --git a/src/java/org/apache/cassandra/service/CassandraServer.java b/src/java/org/apache/cassandra/service/CassandraServer.java index b7a5c41a89..f32ad4cbb7 100644 --- a/src/java/org/apache/cassandra/service/CassandraServer.java +++ b/src/java/org/apache/cassandra/service/CassandraServer.java @@ -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) { diff --git a/src/java/org/apache/cassandra/service/ReadResponseResolver.java b/src/java/org/apache/cassandra/service/ReadResponseResolver.java index 0e99201285..9cf391144d 100644 --- a/src/java/org/apache/cassandra/service/ReadResponseResolver.java +++ b/src/java/org/apache/cassandra/service/ReadResponseResolver.java @@ -121,7 +121,7 @@ public class ReadResponseResolver implements IResponseResolver } /* 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)); diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index de2a2f28fa..4a754bda5e 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -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); /* diff --git a/test/unit/org/apache/cassandra/db/ColumnFamilyTest.java b/test/unit/org/apache/cassandra/db/ColumnFamilyTest.java index d81f4c866d..78f69abd1f 100644 --- a/test/unit/org/apache/cassandra/db/ColumnFamilyTest.java +++ b/test/unit/org/apache/cassandra/db/ColumnFamilyTest.java @@ -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; } } diff --git a/test/unit/org/apache/cassandra/db/RangeFilterTest.java b/test/unit/org/apache/cassandra/db/RangeFilterTest.java index acdaa27a0e..78c2360b82 100644 --- a/test/unit/org/apache/cassandra/db/RangeFilterTest.java +++ b/test/unit/org/apache/cassandra/db/RangeFilterTest.java @@ -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"); diff --git a/test/unit/org/apache/cassandra/db/RowTest.java b/test/unit/org/apache/cassandra/db/RowTest.java index 22e22ba796..48ac84aa91 100644 --- a/test/unit/org/apache/cassandra/db/RowTest.java +++ b/test/unit/org/apache/cassandra/db/RowTest.java @@ -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); diff --git a/test/unit/org/apache/cassandra/db/TableTest.java b/test/unit/org/apache/cassandra/db/TableTest.java index 61b12ca753..823d331c94 100644 --- a/test/unit/org/apache/cassandra/db/TableTest.java +++ b/test/unit/org/apache/cassandra/db/TableTest.java @@ -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); diff --git a/test/unit/org/apache/cassandra/io/SSTableTest.java b/test/unit/org/apache/cassandra/io/SSTableTest.java index 873465085c..5255ef4e17 100644 --- a/test/unit/org/apache/cassandra/io/SSTableTest.java +++ b/test/unit/org/apache/cassandra/io/SSTableTest.java @@ -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 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) {