From 3673e2818603a1c77ae22f8e2180fc42415f5151 Mon Sep 17 00:00:00 2001 From: Jonathan Ellis Date: Mon, 22 Jun 2009 21:17:46 +0000 Subject: [PATCH] multitable support. fix CF table dependencies, making CommitLogTest pass. Also fixes CASSANDRA-188 in passing. fix tablename being null bugs by always parsing tablename from filename. fix last part of SequenceFile. patch by jbellis; reviewed by goffinet for CASSANDRA-79 git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@787404 13f79535-47bb-0310-9956-ffa450edef68 --- .../cassandra/config/DatabaseDescriptor.java | 10 +- .../apache/cassandra/db/BinaryMemtable.java | 2 +- .../org/apache/cassandra/db/ColumnFamily.java | 9 +- .../cassandra/db/ColumnFamilyStore.java | 11 +- .../apache/cassandra/db/ColumnIndexer.java | 33 +-- .../apache/cassandra/db/ColumnIterator.java | 2 +- .../org/apache/cassandra/db/CommitLog.java | 2 +- .../org/apache/cassandra/db/Memtable.java | 2 +- .../org/apache/cassandra/db/SystemTable.java | 2 +- src/java/org/apache/cassandra/db/Table.java | 6 +- .../org/apache/cassandra/db/TypeInfo.java | 120 -------- .../org/apache/cassandra/io/IndexHelper.java | 32 +- src/java/org/apache/cassandra/io/SSTable.java | 28 +- .../org/apache/cassandra/io/SequenceFile.java | 80 ++--- .../org/apache/cassandra/db/TableTest.java | 278 +++++++++++------- .../org/apache/cassandra/io/SSTableTest.java | 4 +- 16 files changed, 264 insertions(+), 357 deletions(-) delete mode 100644 src/java/org/apache/cassandra/db/TypeInfo.java diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java index 35398f5e35..b7271a4ee1 100644 --- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java +++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java @@ -358,6 +358,10 @@ public class DatabaseDescriptor { throw new ConfigurationException("invalid column sort value " + rawColumnIndexType); } + if ("Super".equals(columnType) && !"Name".equals(columnIndexType)) + { + throw new ConfigurationException("Super columnfamilies may only be name-sorted"); + } // see if flush period is set String flushPeriodInMinutes = XMLUtils.getAttributeValue(columnFamily, "FlushPeriodInMinutes"); @@ -816,17 +820,17 @@ public class DatabaseDescriptor return dataFileDirectory; } - public static TypeInfo getTypeInfo(String tableName, String cfName) + public static ColumnComparatorFactory.ComparatorType getTypeInfo(String tableName, String cfName) { assert tableName != null; CFMetaData cfMetadata = DatabaseDescriptor.getCFMetaData(tableName, cfName); if ( cfMetadata.indexProperty_.equals("Name") ) { - return TypeInfo.STRING; + return ColumnComparatorFactory.ComparatorType.NAME; } else { - return TypeInfo.LONG; + return ColumnComparatorFactory.ComparatorType.TIMESTAMP; } } diff --git a/src/java/org/apache/cassandra/db/BinaryMemtable.java b/src/java/org/apache/cassandra/db/BinaryMemtable.java index 2ccb6e46f7..6f209ff87d 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, null, StorageService.getPartitioner()); + SSTable ssTable = new SSTable(directory, filename, 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/ColumnFamily.java b/src/java/org/apache/cassandra/db/ColumnFamily.java index 617a2f8aa3..0ada4224e9 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamily.java +++ b/src/java/org/apache/cassandra/db/ColumnFamily.java @@ -96,8 +96,7 @@ public final class ColumnFamily { Comparator comparator; String columnType = DatabaseDescriptor.getColumnFamilyType(tableName, cfName); - if ("Super".equals(columnType) - || DatabaseDescriptor.isNameSortingEnabled(tableName, cfName)) + if (DatabaseDescriptor.isNameSortingEnabled(tableName, cfName)) { comparator = ColumnComparatorFactory.getComparator(ColumnComparatorFactory.ComparatorType.NAME); } @@ -299,10 +298,6 @@ public final class ColumnFamily 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. @@ -349,7 +344,7 @@ public final class ColumnFamily return columns_.getComparator(); } - private ColumnComparatorFactory.ComparatorType getComparatorType() + public ColumnComparatorFactory.ComparatorType getComparatorType() { return getComparator() == ColumnComparatorFactory.nameComparator_ ? ColumnComparatorFactory.ComparatorType.NAME diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java index bcd346bf6c..4174bfae43 100644 --- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java +++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java @@ -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, table_); + SSTable.onStart(filenames); 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, null, StorageService.getPartitioner()); + SSTable ssTable = new SSTable(ssTableFile, StorageService.getPartitioner()); DataInputBuffer bufIn; bufIn = filter.next(key, cf, ssTable); if (bufIn.getLength() == 0) @@ -1119,15 +1119,12 @@ 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, tableName, StorageService.getPartitioner()); + ssTableRange = new SSTable(rangeFileLocation, mergedFileName, StorageService.getPartitioner()); } try { @@ -1323,7 +1320,7 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean if (ssTable == null) { - ssTable = new SSTable(compactionFileLocation, mergedFileName, null, StorageService.getPartitioner()); + ssTable = new SSTable(compactionFileLocation, mergedFileName, StorageService.getPartitioner()); } ssTable.append(lastkey, bufOut); diff --git a/src/java/org/apache/cassandra/db/ColumnIndexer.java b/src/java/org/apache/cassandra/db/ColumnIndexer.java index 6c0c2356eb..ef76c829ab 100644 --- a/src/java/org/apache/cassandra/db/ColumnIndexer.java +++ b/src/java/org/apache/cassandra/db/ColumnIndexer.java @@ -57,8 +57,7 @@ public class ColumnIndexer dos.write(bufOut.getData(), 0, bufOut.getLength()); /* Do the indexing */ - TypeInfo typeInfo = DatabaseDescriptor.getTypeInfo(columnFamily.getTable(), columnFamily.name()); - doIndexing(typeInfo, columns, dos); + doIndexing(columnFamily.getComparatorType(), columns, dos); } /** @@ -92,30 +91,14 @@ public class ColumnIndexer return bf; } - private static IndexHelper.ColumnIndexInfo getColumnIndexInfo(TypeInfo typeInfo, IColumn column) + private static IndexHelper.ColumnIndexInfo getColumnIndexInfo(ColumnComparatorFactory.ComparatorType typeInfo, IColumn column) { IndexHelper.ColumnIndexInfo cIndexInfo = null; - if ( column instanceof SuperColumn ) - { - cIndexInfo = IndexHelper.ColumnIndexFactory.instance(TypeInfo.STRING); - cIndexInfo.set(column.name()); - } - else - { - cIndexInfo = IndexHelper.ColumnIndexFactory.instance(typeInfo); - switch(typeInfo) - { - case STRING: - cIndexInfo.set(column.name()); - break; - - case LONG: - cIndexInfo.set(column.timestamp()); - break; - } - } - + cIndexInfo = IndexHelper.ColumnIndexFactory.instance(typeInfo); + cIndexInfo.set(typeInfo == ColumnComparatorFactory.ComparatorType.NAME + ? column.name() : column.timestamp()); + return cIndexInfo; } @@ -124,13 +107,11 @@ public class ColumnIndexer * the name index is generated and written into the provided * stream * @param columns for whom the name index needs to be generated - * @param bf bloom filter that summarizes the columns that make - * up the column family. * @param dos stream into which the serialized name index needs * to be written. * @throws IOException */ - private static void doIndexing(TypeInfo typeInfo, Collection columns, DataOutputStream dos) throws IOException + private static void doIndexing(ColumnComparatorFactory.ComparatorType typeInfo, Collection columns, DataOutputStream dos) throws IOException { /* we are going to write column indexes */ int numColumns = 0; diff --git a/src/java/org/apache/cassandra/db/ColumnIterator.java b/src/java/org/apache/cassandra/db/ColumnIterator.java index 04b9bda51c..fab378dc98 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, null, StorageService.getPartitioner()); + SSTable ssTable = new SSTable(filename, 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 0f9a8749fe..7c41eb93fb 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(table_, file.getAbsolutePath()); + IFileReader reader = SequenceFile.reader(file.getAbsolutePath()); try { CommitLogHeader clHeader = readCommitLogHeader(reader); diff --git a/src/java/org/apache/cassandra/db/Memtable.java b/src/java/org/apache/cassandra/db/Memtable.java index 95c27d46a2..5024bd6d1a 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, table_, StorageService.getPartitioner()); + SSTable ssTable = new SSTable(directory, filename, StorageService.getPartitioner()); // sort keys in the order they would be in when decorated final IPartitioner partitioner = StorageService.getPartitioner(); diff --git a/src/java/org/apache/cassandra/db/SystemTable.java b/src/java/org/apache/cassandra/db/SystemTable.java index 22a84c7514..de6c3f85ff 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, table); + reader_ = SequenceFile.reader(systemTable); } private String getFileName() diff --git a/src/java/org/apache/cassandra/db/Table.java b/src/java/org/apache/cassandra/db/Table.java index 9aee9e6aa6..c5513c162c 100644 --- a/src/java/org/apache/cassandra/db/Table.java +++ b/src/java/org/apache/cassandra/db/Table.java @@ -92,7 +92,7 @@ public class Table { String file = getFileName(table); writer_ = SequenceFile.writer(file); - reader_ = SequenceFile.reader(file, table); + reader_ = SequenceFile.reader(file); Table.TableMetadata.load(table); metadata = new Table.TableMetadata(); @@ -118,7 +118,7 @@ public class Table if ( reader_ == null ) { - reader_ = SequenceFile.reader(file, table); + reader_ = SequenceFile.reader(file); } while ( !reader_.isEOF() ) @@ -962,7 +962,7 @@ public class Table // sstables for (String filename : cfs.getSSTableFilenames()) { - FileStruct fs = new FileStruct(SequenceFile.reader(filename, table_), StorageService.getPartitioner()); + FileStruct fs = new FileStruct(SequenceFile.reader(filename), StorageService.getPartitioner()); fs.seekTo(startWith); iterators.add(fs); } diff --git a/src/java/org/apache/cassandra/db/TypeInfo.java b/src/java/org/apache/cassandra/db/TypeInfo.java deleted file mode 100644 index 863fe9f620..0000000000 --- a/src/java/org/apache/cassandra/db/TypeInfo.java +++ /dev/null @@ -1,120 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you under the Apache License, Version 2.0 (the - * "License"); you may not use this file except in compliance - * with the License. You may obtain a copy of the License at - * - * http://www.apache.org/licenses/LICENSE-2.0 - * - * Unless required by applicable law or agreed to in writing, software - * distributed under the License is distributed on an "AS IS" BASIS, - * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. - * See the License for the specific language governing permissions and - * limitations under the License. - */ -package org.apache.cassandra.db; - -public enum TypeInfo -{ - BYTE, - CHAR, - SHORT, - INT, - LONG, - DOUBLE, - FLOAT, - STRING, - BLOB; - - public static byte toByte(TypeInfo ti) - { - byte value = 0; - switch(ti) - { - case BYTE: - value = 1; - break; - - case CHAR: - value = 2; - break; - - case SHORT: - value = 3; - break; - - case INT: - value = 4; - break; - - case LONG: - value = 5; - break; - - case DOUBLE: - value = 6; - break; - - case FLOAT: - value = 7; - break; - - case STRING: - value = 8; - break; - - case BLOB: - value = 9; - break; - } - - return value; - } - - public static TypeInfo fromByte(byte b) - { - TypeInfo ti = null; - switch(b) - { - case 1: - ti = TypeInfo.BYTE; - break; - - case 2: - ti = TypeInfo.CHAR; - break; - - case 3: - ti = TypeInfo.SHORT; - break; - - case 4: - ti = TypeInfo.INT; - break; - - case 5: - ti = TypeInfo.LONG; - break; - - case 6: - ti = TypeInfo.DOUBLE; - break; - - case 7: - ti = TypeInfo.FLOAT; - break; - - case 8: - ti = TypeInfo.STRING; - break; - - case 9: - ti = TypeInfo.BLOB; - break; - } - return ti; - } -} diff --git a/src/java/org/apache/cassandra/io/IndexHelper.java b/src/java/org/apache/cassandra/io/IndexHelper.java index 1f64bb8359..4617e53743 100644 --- a/src/java/org/apache/cassandra/io/IndexHelper.java +++ b/src/java/org/apache/cassandra/io/IndexHelper.java @@ -26,7 +26,7 @@ import java.util.*; import org.apache.cassandra.config.DatabaseDescriptor; import org.apache.cassandra.db.IColumn; -import org.apache.cassandra.db.TypeInfo; +import org.apache.cassandra.db.ColumnComparatorFactory; import org.apache.cassandra.utils.FBUtilities; @@ -152,12 +152,8 @@ public class IndexHelper DataInputBuffer indexIn = new DataInputBuffer(); indexIn.reset(indexOut.getData(), indexOut.getLength()); - TypeInfo typeInfo = DatabaseDescriptor.getTypeInfo(tableName, cfName); - if ( DatabaseDescriptor.getColumnFamilyType(tableName, cfName).equals("Super") || DatabaseDescriptor.isNameSortingEnabled(tableName, cfName) ) - { - typeInfo = TypeInfo.STRING; - } - + ColumnComparatorFactory.ComparatorType typeInfo = DatabaseDescriptor.getTypeInfo(tableName, cfName); + while(indexIn.available() > 0) { ColumnIndexInfo cIndexInfo = ColumnIndexFactory.instance(typeInfo); @@ -281,7 +277,7 @@ public class IndexHelper * binary search. */ Comparator comparator = Collections.reverseOrder(); - IndexHelper.ColumnIndexInfo rhs = IndexHelper.ColumnIndexFactory.instance(TypeInfo.LONG); + IndexHelper.ColumnIndexInfo rhs = IndexHelper.ColumnIndexFactory.instance(ColumnComparatorFactory.ComparatorType.TIMESTAMP); rhs.set(timeRange.rhs()); int index = Collections.binarySearch(columnIndexList, rhs, comparator); if ( index < 0 ) @@ -307,7 +303,7 @@ public class IndexHelper { int chunks = columnIndexList.size(); /* Index info for the lower bound of the time range */ - IndexHelper.ColumnIndexInfo lhs = IndexHelper.ColumnIndexFactory.instance(TypeInfo.LONG); + IndexHelper.ColumnIndexInfo lhs = IndexHelper.ColumnIndexFactory.instance(ColumnComparatorFactory.ComparatorType.TIMESTAMP); lhs.set(timeRange.lhs()); int i = index + 1; for ( ; i < chunks; ++i ) @@ -339,21 +335,11 @@ public class IndexHelper public static class ColumnIndexFactory { - public static ColumnIndexInfo instance(TypeInfo typeInfo) + public static ColumnIndexInfo instance(ColumnComparatorFactory.ComparatorType typeInfo) { - ColumnIndexInfo cIndexInfo = null; - switch(typeInfo) - { - case STRING: - cIndexInfo = new ColumnNameIndexInfo(); - break; - - case LONG: - cIndexInfo = new ColumnTimestampIndexInfo(); - break; - } - return cIndexInfo; - } + return typeInfo == ColumnComparatorFactory.ComparatorType.NAME + ? new ColumnNameIndexInfo() : new ColumnTimestampIndexInfo(); + } } /** diff --git a/src/java/org/apache/cassandra/io/SSTable.java b/src/java/org/apache/cassandra/io/SSTable.java index dfc47a6158..1d28e28616 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, String tableName) throws IOException + public static void onStart(List filenames) throws IOException { for (String filename : filenames) { try { - new SSTable(filename, tableName, StorageService.getPartitioner()); + new SSTable(filename, StorageService.getPartitioner()); } catch (IOException ex) { @@ -205,17 +205,15 @@ 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, String tableName, IPartitioner partitioner) throws IOException + public SSTable(String dataFileName, IPartitioner partitioner) throws IOException { dataFile_ = dataFileName; partitioner_ = partitioner; - table_ = tableName; /* * this is to prevent multiple threads from * loading the same index files multiple times @@ -237,15 +235,20 @@ 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, String tableName, IPartitioner partitioner) throws IOException + public SSTable(String directory, String filename, 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); } + static String parseTableName(String filename) + { + String[] parts = new File(filename).getName().split("-"); // table, cf, index, [filetype] + return parts[0]; + } + private void loadBloomFilter() throws IOException { assert bfs_.get(dataFile_) == null; @@ -477,7 +480,7 @@ public class SSTable IFileReader dataReader = null; try { - dataReader = SequenceFile.reader(dataFile_, table_); + dataReader = SequenceFile.reader(dataFile_); String decoratedKey = partitioner_.decorateKey(clientKey); long position = getPosition(decoratedKey, dataReader, partitioner_); @@ -559,22 +562,19 @@ public class SSTable */ public ColumnGroupReader getColumnGroupReader(String key, String cfName, String startColumn, boolean isAscending) throws IOException { - ColumnGroupReader reader = null; - IFileReader dataReader = SequenceFile.reader(table_, dataFile_); + IFileReader dataReader = SequenceFile.reader(dataFile_); try { /* Morph key into actual key based on the partition type. */ String decoratedKey = partitioner_.decorateKey(key); long position = getPosition(decoratedKey, dataReader, partitioner_); - reader = new ColumnGroupReader(dataFile_, decoratedKey, cfName, startColumn, isAscending, position); + return new ColumnGroupReader(dataFile_, decoratedKey, cfName, startColumn, isAscending, position); } finally { - if (dataReader != null) - dataReader.close(); + dataReader.close(); } - return reader; } /** obviously only for testing */ diff --git a/src/java/org/apache/cassandra/io/SequenceFile.java b/src/java/org/apache/cassandra/io/SequenceFile.java index 4db226ccef..34ba7e0e01 100644 --- a/src/java/org/apache/cassandra/io/SequenceFile.java +++ b/src/java/org/apache/cassandra/io/SequenceFile.java @@ -516,6 +516,8 @@ public class SequenceFile { private String key_; private String cfName_; + private String cfType_; + private int indexType_; private boolean isAscending_; private List columnIndexList_; @@ -585,16 +587,18 @@ public class SequenceFile /* read the index */ List colIndexList = new ArrayList(); if (hasColumnIndexes) - totalBytesRead += IndexHelper.deserializeIndex(null, cfName_, file_, colIndexList); + totalBytesRead += IndexHelper.deserializeIndex(getTableName(), cfName_, file_, colIndexList); /* need to do two things here. * 1. move the file pointer to the beginning of the list of stored columns * 2. calculate the size of all columns */ String cfName = file_.readUTF(); + cfType_ = file_.readUTF(); + indexType_ = file_.readInt(); localDeletionTime_ = file_.readInt(); markedForDeleteAt_ = file_.readLong(); int totalNumCols = file_.readInt(); - allColumnsSize_ = dataSize - (totalBytesRead + utfPrefix_ + cfName.length() + 4 + 8 + 4); + allColumnsSize_ = dataSize - (totalBytesRead + 2 * utfPrefix_ + cfName.length() + cfType_.length() + 4 + 4 + 8 + 4); columnStartPosition_ = file_.getFilePointer(); columnIndexList_ = getFullColumnIndexList(colIndexList, totalNumCols); @@ -627,6 +631,8 @@ public class SequenceFile bufOut.reset(); // write CF info bufOut.writeUTF(cfName_); + bufOut.writeUTF(cfType_); + bufOut.writeInt(indexType_); bufOut.writeInt(localDeletionTime_); bufOut.writeLong(markedForDeleteAt_); // now write the columns @@ -651,12 +657,15 @@ public class SequenceFile private static final short utfPrefix_ = 2; protected RandomAccessFile file_; protected String filename_; - private String table_; - AbstractReader(String filename, String tableName) + AbstractReader(String filename) { filename_ = filename; - table_ = tableName; + } + + String getTableName() + { + return SSTable.parseTableName(filename_); } public String getFileName() @@ -703,10 +712,11 @@ public class SequenceFile /* if we do then deserialize the index */ if (hasColumnIndexes) { - if (DatabaseDescriptor.isNameSortingEnabled(table_, cfName) || DatabaseDescriptor.getColumnFamilyType(table_, cfName).equals("Super")) + String tableName = getTableName(); + if (DatabaseDescriptor.isNameSortingEnabled(tableName, cfName)) { /* read the index */ - totalBytesRead += IndexHelper.deserializeIndex(table_, cfName, file_, columnIndexList); + totalBytesRead += IndexHelper.deserializeIndex(tableName, cfName, file_, columnIndexList); } else { @@ -734,7 +744,7 @@ public class SequenceFile if (DatabaseDescriptor.isTimeSortingEnabled(null, cfName)) { /* read the index */ - totalBytesRead += IndexHelper.deserializeIndex(table_, cfName, file_, columnIndexList); + totalBytesRead += IndexHelper.deserializeIndex(getTableName(), cfName, file_, columnIndexList); } else { @@ -832,6 +842,12 @@ public class SequenceFile String cfName = file_.readUTF(); dataSize -= (utfPrefix_ + cfName.length()); + String cfType = file_.readUTF(); + dataSize -= (utfPrefix_ + cfType.length()); + + int indexType = file_.readInt(); + dataSize -= 4; + /* read local deletion time */ int localDeletionTime = file_.readInt(); dataSize -=4; @@ -852,19 +868,13 @@ public class SequenceFile file_.skipBytes((int) coordinate.start_); dataSize = (int) (coordinate.end_ - coordinate.start_); - /* - * write the number of columns in the column family we are returning: - * dataSize that we are reading + - * length of column family name + - * one booleanfor deleted or not + - * one int for number of columns - */ - bufOut.writeInt(dataSize + utfPrefix_ + cfName.length() + 4 + 8 + 4); - /* write the column family name */ + // returned data size + bufOut.writeInt(dataSize + utfPrefix_ * 2 + cfName.length() + cfType.length() + 4 + 4 + 8 + 4); + // echo back the CF data we read bufOut.writeUTF(cfName); - /* write local deletion time */ + bufOut.writeUTF(cfType); + bufOut.writeInt(indexType); bufOut.writeInt(localDeletionTime); - /* write if this cf is marked for delete */ bufOut.writeLong(markedForDeleteAt); /* write number of columns */ bufOut.writeInt(columnRange.count()); @@ -911,6 +921,12 @@ public class SequenceFile String cfName = file_.readUTF(); dataSize -= (utfPrefix_ + cfName.length()); + String cfType = file_.readUTF(); + dataSize -= (utfPrefix_ + cfType.length()); + + int indexType = file_.readInt(); + dataSize -= 4; + /* read local deletion time */ int localDeletionTime = file_.readInt(); dataSize -=4; @@ -940,19 +956,13 @@ public class SequenceFile dataSizeReturned += coordinate.end_ - coordinate.start_; } - /* - * write the number of columns in the column family we are returning: - * dataSize that we are reading + - * length of column family name + - * one booleanfor deleted or not + - * one int for number of columns - */ - bufOut.writeInt(dataSizeReturned + utfPrefix_ + cfName.length() + 4 + 8 + 4); - /* write the column family name */ + // returned data size + bufOut.writeInt(dataSizeReturned + utfPrefix_ * 2 + cfName.length() + cfType.length() + 4 + 4 + 8 + 4); + // echo back the CF data we read bufOut.writeUTF(cfName); - /* write local deletion time */ + bufOut.writeUTF(cfType); + bufOut.writeInt(indexType); bufOut.writeInt(localDeletionTime); - /* write if this cf is marked for delete */ bufOut.writeLong(markedForDeleteAt); /* write number of columns */ bufOut.writeInt(numColsReturned); @@ -1012,9 +1022,9 @@ public class SequenceFile public static class Reader extends AbstractReader { - Reader(String filename, String tableName) throws IOException + Reader(String filename) throws IOException { - super(filename, tableName); + super(filename); init(filename); } @@ -1077,7 +1087,7 @@ public class SequenceFile BufferReader(String filename, int size) throws IOException { - super(filename, null); + super(filename); size_ = size; } @@ -1106,9 +1116,9 @@ public class SequenceFile return new FastConcurrentWriter(filename, size); } - public static IFileReader reader(String filename, String tableName) throws IOException + public static IFileReader reader(String filename) throws IOException { - return new Reader(filename, tableName); + return new Reader(filename); } public static IFileReader bufferedReader(String filename, int size) throws IOException diff --git a/test/unit/org/apache/cassandra/db/TableTest.java b/test/unit/org/apache/cassandra/db/TableTest.java index 823d331c94..02b994662a 100644 --- a/test/unit/org/apache/cassandra/db/TableTest.java +++ b/test/unit/org/apache/cassandra/db/TableTest.java @@ -37,6 +37,19 @@ public class TableTest extends CleanupHelper private static final String TEST_KEY = "key1"; private static final String TABLE_NAME = "Table1"; + interface Runner + { + public void run() throws Exception; + } + + private void reTest(Runner setup, ColumnFamilyStore cfs, Runner verify) throws Exception + { + setup.run(); + verify.run(); + cfs.forceBlockingFlush(); + verify.run(); + } + @Test public void testOpen() throws Throwable { Table table = Table.open("Mailbox"); @@ -47,40 +60,66 @@ public class TableTest extends CleanupHelper @Test public void testGetRowSingleColumn() throws Throwable { - Table table = Table.open(TABLE_NAME); - RowMutation rm = makeSimpleRowMutation(); - rm.apply(); - Row result = table.getRow(TEST_KEY, "Standard1:col1"); - ColumnFamily cres = result.getColumnFamily("Standard1"); - assertNotNull(cres); - assertEquals(1, cres.getColumnCount()); - assertNotNull(cres.getColumn("col1")); + final Table table = Table.open(TABLE_NAME); + Runner setup = new Runner() + { + public void run() throws Exception + { + RowMutation rm = makeSimpleRowMutation(); + rm.apply(); + } + }; + Runner verify = new Runner() + { + public void run() throws Exception + { + Row result; + + result = table.getRow(TEST_KEY, "Standard1:col1"); + assertColumns(result.getColumnFamily("Standard1"), "col1"); + + result = table.getRow(TEST_KEY, "Standard1:col3"); + assertColumns(result.getColumnFamily("Standard1"), "col3"); + } + }; + reTest(setup, table.getColumnFamilyStore("Standard1"), verify); } @Test public void testGetRowOffsetCount() throws Throwable { - Table table = Table.open(TABLE_NAME); - - RowMutation rm = makeSimpleRowMutation(); //inserts col1, col2, col3 + final Table table = Table.open(TABLE_NAME); - - rm.apply(); - Row result = table.getRow(TEST_KEY, "Standard1", 0, 2); - ColumnFamily cres = result.getColumnFamily("Standard1"); - assertNotNull(cres); - assertEquals(cres.getColumnCount(), 2); - // should have col1 and col2 - assertNotNull(cres.getColumn("col1")); - assertNotNull(cres.getColumn("col2")); + Runner setup = new Runner() + { + public void run() throws Exception + { + RowMutation rm = makeSimpleRowMutation(); //inserts col1, col2, col3 + rm.apply(); + } + }; + Runner verify = new Runner() + { + public void run() throws Exception + { + Row result = table.getRow(TEST_KEY, "Standard1", 0, 2); + ColumnFamily cres = result.getColumnFamily("Standard1"); + assertNotNull(cres); + assertEquals(cres.getColumnCount(), 2); + // should have col1 and col2 + assertNotNull(cres.getColumn("col1")); + assertNotNull(cres.getColumn("col2")); - result = table.getRow(TEST_KEY, "Standard1", 1, 2); - cres = result.getColumnFamily("Standard1"); - assertNotNull(cres); - assertEquals(2, cres.getColumnCount()); - // offset is 1, so we should have col2 and col3 - assertNotNull(cres.getColumn("col2")); - assertNotNull(cres.getColumn("col3")); + result = table.getRow(TEST_KEY, "Standard1", 1, 2); + cres = result.getColumnFamily("Standard1"); + assertNotNull(cres); + assertEquals(2, cres.getColumnCount()); + // offset is 1, so we should have col2 and col3 + assertNotNull(cres.getColumn("col2")); + assertNotNull(cres.getColumn("col3")); + } + }; + reTest(setup, table.getColumnFamilyStore("Standard1"), verify); } @Test @@ -251,57 +290,114 @@ public class TableTest extends CleanupHelper public void testGetSliceFromBasic() throws Throwable { // tests slicing against data from one row in a memtable and then flushed to an sstable - Table table = Table.open(TABLE_NAME); - String ROW = "row1"; - RowMutation rm = new RowMutation(TABLE_NAME, ROW); - 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)); - cf.addColumn(new Column("col5", "val5".getBytes(), 1L)); - cf.addColumn(new Column("col7", "val7".getBytes(), 1L)); - cf.addColumn(new Column("col9", "val9".getBytes(), 1L)); - rm.add(cf); - rm.apply(); - - rm = new RowMutation(TABLE_NAME, ROW); - rm.delete("Standard1:col4", 2L); - rm.apply(); + final Table table = Table.open(TABLE_NAME); + final String ROW = "row1"; + Runner setup = new Runner() + { + public void run() throws Exception + { + RowMutation rm = new RowMutation(TABLE_NAME, ROW); + 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)); + cf.addColumn(new Column("col5", "val5".getBytes(), 1L)); + cf.addColumn(new Column("col7", "val7".getBytes(), 1L)); + cf.addColumn(new Column("col9", "val9".getBytes(), 1L)); + rm.add(cf); + rm.apply(); - validateGetSliceFromBasic(table, ROW); - table.getColumnFamilyStore("Standard1").forceBlockingFlush(); - validateGetSliceFromBasic(table, ROW); + rm = new RowMutation(TABLE_NAME, ROW); + rm.delete("Standard1:col4", 2L); + rm.apply(); + } + }; + + Runner verify = new Runner() + { + public void run() throws Exception + { + Row result; + ColumnFamily cf; + + result = table.getSliceFrom(ROW, "Standard1:col5", true, 2); + cf = result.getColumnFamily("Standard1"); + assertColumns(cf, "col5", "col7"); + + result = table.getSliceFrom(ROW, "Standard1:col4", true, 2); + cf = result.getColumnFamily("Standard1"); + assertColumns(cf, "col4", "col5", "col7"); + + result = table.getSliceFrom(ROW, "Standard1:col5", false, 2); + cf = result.getColumnFamily("Standard1"); + assertColumns(cf, "col3", "col4", "col5"); + + result = table.getSliceFrom(ROW, "Standard1:col6", false, 2); + cf = result.getColumnFamily("Standard1"); + assertColumns(cf, "col3", "col4", "col5"); + + result = table.getSliceFrom(ROW, "Standard1:col95", true, 2); + cf = result.getColumnFamily("Standard1"); + assertColumns(cf); + + result = table.getSliceFrom(ROW, "Standard1:col0", false, 2); + cf = result.getColumnFamily("Standard1"); + assertColumns(cf); + } + }; + + reTest(setup, table.getColumnFamilyStore("Standard1"), verify); } @Test public void testGetSliceFromAdvanced() throws Throwable { // tests slicing against data from one row spread across two sstables - Table table = Table.open(TABLE_NAME); - String ROW = "row2"; - RowMutation rm = new RowMutation(TABLE_NAME, ROW); - 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)); - cf.addColumn(new Column("col4", "val4".getBytes(), 1L)); - cf.addColumn(new Column("col5", "val5".getBytes(), 1L)); - cf.addColumn(new Column("col6", "val6".getBytes(), 1L)); - rm.add(cf); - rm.apply(); - table.getColumnFamilyStore("Standard1").forceBlockingFlush(); - - rm = new RowMutation(TABLE_NAME, ROW); - 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)); - rm.add(cf); - rm.apply(); + final Table table = Table.open(TABLE_NAME); + final String ROW = "row2"; + Runner setup = new Runner() + { + public void run() throws Exception + { + RowMutation rm = new RowMutation(TABLE_NAME, ROW); + 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)); + cf.addColumn(new Column("col4", "val4".getBytes(), 1L)); + cf.addColumn(new Column("col5", "val5".getBytes(), 1L)); + cf.addColumn(new Column("col6", "val6".getBytes(), 1L)); + rm.add(cf); + rm.apply(); + table.getColumnFamilyStore("Standard1").forceBlockingFlush(); - validateGetSliceFromAdvanced(table, ROW); - table.getColumnFamilyStore("Standard1").forceBlockingFlush(); - validateGetSliceFromAdvanced(table, ROW); + rm = new RowMutation(TABLE_NAME, ROW); + 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)); + rm.add(cf); + rm.apply(); + } + }; + + Runner verify = new Runner() + { + public void run() throws Exception + { + Row result; + ColumnFamily cfres; + + result = table.getSliceFrom(ROW, "Standard1:col2", true, 3); + cfres = result.getColumnFamily("Standard1"); + assertColumns(cfres, "col2", "col3", "col4"); + assertEquals(new String(cfres.getColumn("col2").value()), "valx"); + assertEquals(new String(cfres.getColumn("col3").value()), "valx"); + assertEquals(new String(cfres.getColumn("col4").value()), "val4"); + } + }; + + reTest(setup, table.getColumnFamilyStore("Standard1"), verify); } @Test @@ -362,46 +458,4 @@ public class TableTest extends CleanupHelper : "Columns [" + StringUtils.join(columns, ", ") + "] is not expected [" + StringUtils.join(columnNames, ", ") + "]"; } - private void validateGetSliceFromAdvanced(Table table, String row) throws Throwable - { - Row result; - ColumnFamily cfres; - - result = table.getSliceFrom(row, "Standard1:col2", true, 3); - cfres = result.getColumnFamily("Standard1"); - assertColumns(cfres, "col2", "col3", "col4"); - assertEquals(new String(cfres.getColumn("col2").value()), "valx"); - assertEquals(new String(cfres.getColumn("col3").value()), "valx"); - assertEquals(new String(cfres.getColumn("col4").value()), "val4"); - } - - private void validateGetSliceFromBasic(Table table, String row) throws Throwable - { - Row result; - ColumnFamily cf; - - result = table.getSliceFrom(row, "Standard1:col5", true, 2); - cf = result.getColumnFamily("Standard1"); - assertColumns(cf, "col5", "col7"); - - result = table.getSliceFrom(row, "Standard1:col4", true, 2); - cf = result.getColumnFamily("Standard1"); - assertColumns(cf, "col4", "col5", "col7"); - - result = table.getSliceFrom(row, "Standard1:col5", false, 2); - cf = result.getColumnFamily("Standard1"); - assertColumns(cf, "col3", "col4", "col5"); - - result = table.getSliceFrom(row, "Standard1:col6", false, 2); - cf = result.getColumnFamily("Standard1"); - assertColumns(cf, "col3", "col4", "col5"); - - result = table.getSliceFrom(row, "Standard1:col95", true, 2); - cf = result.getColumnFamily("Standard1"); - assertColumns(cf); - - result = table.getSliceFrom(row, "Standard1:col0", false, 2); - cf = result.getColumnFamily("Standard1"); - assertColumns(cf); - } } diff --git a/test/unit/org/apache/cassandra/io/SSTableTest.java b/test/unit/org/apache/cassandra/io/SSTableTest.java index 5255ef4e17..873465085c 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", "Table1", new OrderPreservingPartitioner()); + SSTable ssTable = new SSTable(f.getPath() + "-Data.db", 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", "Table1", new OrderPreservingPartitioner()); + SSTable ssTable = new SSTable(f.getPath() + "-Data.db", new OrderPreservingPartitioner()); FileStruct fs = new FileStruct(SequenceFile.bufferedReader(ssTable.dataFile_, 128 * 1024), new OrderPreservingPartitioner()); for (String key : keys) {