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

remove table_ from CommitLog and add multitable tests. 

patch by jbellis; reviewed by goffinet for CASSANDRA-79

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@787407 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Jonathan Ellis 2009-06-22 21:18:12 +00:00
parent 5ccebd43c9
commit 509cbb85de
15 changed files with 127 additions and 97 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
conf/storage-conf.xml
View File

@ -22,7 +22,10 @@
<!--======================================================================-->
<ClusterName>Test Cluster</ClusterName>
<!-- Tables and ColumnFamilies -->
<!-- Tables and ColumnFamilies
Think of a table as a namespace, not a relational table.
(ColumnFamilies are closer in meaning to those.)
-->
<Tables>
<Table Name="Table1">
<!-- if FlushPeriodInMinutes is configured and positive, it will be
@ -32,9 +35,7 @@
<ColumnFamily ColumnSort="Name" Name="Standard1" FlushPeriodInMinutes="60"/>
<ColumnFamily ColumnSort="Name" Name="Standard2"/>
<ColumnFamily ColumnSort="Time" Name="StandardByTime1"/>
<ColumnFamily ColumnSort="Time" Name="StandardByTime2"/>
<ColumnFamily ColumnType="Super" ColumnSort="Name" Name="Super1"/>
<ColumnFamily ColumnType="Super" ColumnSort="Name" Name="Super2"/>
</Table>
</Tables>

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

@ -709,7 +709,7 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
{
if (cLogCtx.isValidContext())
{
CommitLog.open(table_).onMemtableFlush(columnFamily_, cLogCtx);
CommitLog.open().onMemtableFlush(table_, columnFamily_, cLogCtx);
}
}
@ -1667,6 +1667,9 @@ public final class ColumnFamilyStore implements ColumnFamilyStoreMBean
}
}
/**
* for testing. no effort is made to clear historical memtables.
*/
void clearUnsafe()
{
lock_.writeLock().lock();

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

@ -67,7 +67,7 @@ import java.util.concurrent.locks.ReentrantLock;
public class CommitLog
{
private static volatile int SEGMENT_SIZE = 128*1024*1024; // roll after log gets this big
private static Map<String, CommitLog> instances_ = new HashMap<String, CommitLog>();
private static volatile CommitLog instance_;
private static Lock lock_ = new ReentrantLock();
private static Logger logger_ = Logger.getLogger(CommitLog.class);
private static Map<String, CommitLogHeader> clHeaders_ = new HashMap<String, CommitLogHeader>();
@ -145,19 +145,17 @@ public class CommitLog
return SequenceFile.writer(file);
}
static CommitLog open(String table) throws IOException
static CommitLog open() throws IOException
{
CommitLog commitLog = instances_.get(table);
if ( commitLog == null )
if ( instance_ == null )
{
CommitLog.lock_.lock();
try
{
commitLog = instances_.get(table);
if ( commitLog == null )
if ( instance_ == null )
{
commitLog = new CommitLog(table, false);
instances_.put(table, commitLog);
instance_ = new CommitLog(false);
}
}
finally
@ -165,16 +163,9 @@ public class CommitLog
CommitLog.lock_.unlock();
}
}
return commitLog;
return instance_;
}
static String getTableName(String file)
{
String[] values = file.split("-");
return values[1];
}
private String table_;
/* Current commit log file */
private String logFile_;
/* header for current commit log */
@ -188,7 +179,7 @@ public class CommitLog
private void setNextFileName()
{
logFile_ = DatabaseDescriptor.getLogFileLocation() + System.getProperty("file.separator") +
"CommitLog-" + table_ + "-" + System.currentTimeMillis() + ".log";
"CommitLog-" + System.currentTimeMillis() + ".log";
}
/*
@ -197,9 +188,8 @@ public class CommitLog
* param @ recoverymode - is commit log being instantiated in
* in recovery mode.
*/
CommitLog(String table, boolean recoveryMode) throws IOException
CommitLog(boolean recoveryMode) throws IOException
{
table_ = table;
if ( !recoveryMode )
{
setNextFileName();
@ -217,7 +207,6 @@ public class CommitLog
*/
CommitLog(File logFile) throws IOException
{
table_ = CommitLog.getTableName(logFile.getName());
logFile_ = logFile.getAbsolutePath();
logWriter_ = CommitLog.createWriter(logFile_);
}
@ -242,8 +231,7 @@ public class CommitLog
*/
private void writeCommitLogHeader() throws IOException
{
Table table = Table.open(table_);
int cfSize = table.getNumberOfColumnFamilies();
int cfSize = Table.TableMetadata.getColumnFamilyCount();
/* record the beginning of the commit header */
/* write the commit log header */
clHeader_ = new CommitLogHeader(cfSize);
@ -299,8 +287,8 @@ public class CommitLog
/* read the commit log entry */
try
{
Row row = Row.serializer(table_).deserialize(bufIn);
Table table = Table.open(table_);
Row row = Row.serializer().deserialize(bufIn);
Table table = Table.open(row.getTable());
tablesRecovered.add(table);
Collection<ColumnFamily> columnFamilies = new ArrayList<ColumnFamily>(row.getColumnFamilies());
/* remove column families that have already been flushed */
@ -349,7 +337,7 @@ public class CommitLog
*/
private void maybeUpdateHeader(Row row) throws IOException
{
Table table = Table.open(table_);
Table table = Table.open(row.getTable());
for (ColumnFamily columnFamily : row.getColumnFamilies())
{
int id = table.getColumnFamilyId(columnFamily.name());
@ -382,7 +370,7 @@ public class CommitLog
{
/* serialize the row */
cfBuffer.reset();
Row.serializer(table_).serialize(row, cfBuffer);
Row.serializer().serialize(row, cfBuffer);
currentPosition = logWriter_.getCurrentPosition();
cLogCtx = new CommitLogContext(logFile_, currentPosition);
/* Update the header */
@ -410,9 +398,9 @@ public class CommitLog
* The bit flag associated with this column family is set in the
* header and this is used to decide if the log file can be deleted.
*/
synchronized void onMemtableFlush(String cf, CommitLog.CommitLogContext cLogCtx) throws IOException
synchronized void onMemtableFlush(String tableName, String cf, CommitLog.CommitLogContext cLogCtx) throws IOException
{
Table table = Table.open(table_);
Table table = Table.open(tableName);
int id = table.getColumnFamilyId(cf);
/* trying discarding old commit log files */
discard(cLogCtx, id);

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

@ -153,7 +153,7 @@ public class Memtable implements Comparable<Memtable>
try
{
Table.open(table_).getColumnFamilyStore(cfName_).switchMemtable();
enqueueFlush(CommitLog.open(table_).getContext());
enqueueFlush(CommitLog.open().getContext());
}
catch (IOException ex)
{

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

@ -59,29 +59,21 @@ private static ICompactSerializer<ReadResponse> serializer_;
return message;
}
private String table_;
private Row row_;
private byte[] digest_ = ArrayUtils.EMPTY_BYTE_ARRAY;
private boolean isDigestQuery_ = false;
public ReadResponse(String table, byte[] digest )
public ReadResponse(byte[] digest )
{
assert digest != null;
table_ = table;
digest_= digest;
}
public ReadResponse(String table, Row row)
public ReadResponse(Row row)
{
table_ = table;
row_ = row;
}
public String table()
{
return table_;
}
public Row row()
{
return row_;
@ -107,20 +99,18 @@ class ReadResponseSerializer implements ICompactSerializer<ReadResponse>
{
public void serialize(ReadResponse rm, DataOutputStream dos) throws IOException
{
dos.writeUTF(rm.table());
dos.writeInt(rm.digest().length);
dos.write(rm.digest());
dos.writeBoolean(rm.isDigestQuery());
if( !rm.isDigestQuery() && rm.row() != null )
{
Row.serializer(rm.table()).serialize(rm.row(), dos);
Row.serializer().serialize(rm.row(), dos);
}
}
public ReadResponse deserialize(DataInputStream dis) throws IOException
{
String table = dis.readUTF();
int digestSize = dis.readInt();
byte[] digest = new byte[digestSize];
dis.read(digest, 0 , digestSize);
@ -129,17 +119,17 @@ class ReadResponseSerializer implements ICompactSerializer<ReadResponse>
Row row = null;
if ( !isDigest )
{
row = Row.serializer(table).deserialize(dis);
row = Row.serializer().deserialize(dis);
}
ReadResponse rmsg = null;
if( isDigest )
{
rmsg = new ReadResponse(table, digest);
rmsg = new ReadResponse(digest);
}
else
{
rmsg = new ReadResponse(table, row);
rmsg = new ReadResponse(row);
}
rmsg.setIsDigestQuery(isDigest);
return rmsg;

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

@ -79,11 +79,11 @@ public class ReadVerbHandler implements IVerbHandler
ReadResponse readResponse = null;
if (readCommand.isDigestQuery())
{
readResponse = new ReadResponse(table.getTableName(), row.digest());
readResponse = new ReadResponse(row.digest());
}
else
{
readResponse = new ReadResponse(table.getTableName(), row);
readResponse = new ReadResponse(row);
}
readResponse.setIsDigestQuery(readCommand.isDigestQuery());
/* serialize the ReadResponseMessage. */

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

@ -55,7 +55,7 @@ public class RecoveryManager
{
File[] files = getListofCommitLogs();
Arrays.sort(files, new FileUtils.FileComparator());
new CommitLog(DatabaseDescriptor.getTables().get(0), true).recover(files);
new CommitLog(true).recover(files);
FileUtils.delete(files);
}
}

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

@ -48,9 +48,9 @@ public class Row
return table_;
}
static RowSerializer serializer(String tableName)
static RowSerializer serializer()
{
return new RowSerializer(tableName);
return new RowSerializer();
}
private String key_;
@ -197,41 +197,31 @@ public class Row
class RowSerializer implements ICompactSerializer<Row>
{
private String table_;
public RowSerializer(String tableName)
{
this.table_ = tableName;
}
public void serialize(Row row, DataOutputStream dos) throws IOException
{
dos.writeUTF(row.getTable());
dos.writeUTF(row.key());
Collection<ColumnFamily> columnFamilies = row.getColumnFamilies();
int size = columnFamilies.size();
dos.writeInt(size);
if (size > 0)
for (ColumnFamily cf : columnFamilies)
{
for (ColumnFamily cf : columnFamilies)
{
ColumnFamily.serializer().serialize(cf, dos);
}
ColumnFamily.serializer().serialize(cf, dos);
}
}
public Row deserialize(DataInputStream dis) throws IOException
{
String table = dis.readUTF();
String key = dis.readUTF();
Row row = new Row(table_, key);
Row row = new Row(table, key);
int size = dis.readInt();
if (size > 0)
for (int i = 0; i < size; ++i)
{
for (int i = 0; i < size; ++i)
{
ColumnFamily cf = ColumnFamily.serializer().deserialize(dis);
row.addColumnFamily(cf);
}
ColumnFamily cf = ColumnFamily.serializer().deserialize(dis);
row.addColumnFamily(cf);
}
return row;
}

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

@ -112,7 +112,7 @@ public class SystemTable
bufIn.readInt();
try
{
systemRow_ = Row.serializer(table_).deserialize(bufIn);
systemRow_ = Row.serializer().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(row.getTable()).serialize(row, bufOut);
Row.serializer().serialize(row, bufOut);
try
{
writer_.append(row.key(), bufOut);

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

@ -150,6 +150,11 @@ public class Table
return sb.toString();
}
public static int getColumnFamilyCount()
{
return idCfMap_.size();
}
}
/**
@ -688,7 +693,7 @@ public class Table
{
/* Add row to the commit log. */
long start = System.currentTimeMillis();
CommitLog.CommitLogContext cLogCtx = CommitLog.open(table_).add(row);
CommitLog.CommitLogContext cLogCtx = CommitLog.open().add(row);
for (ColumnFamily columnFamily : row.getColumnFamilies())
{

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

@ -87,7 +87,7 @@ public class ReadResponseResolver implements IResponseResolver<Row>
rowList.add(result.row());
endPoints.add(response.getFrom());
key = result.row().key();
table = result.table();
table = result.row().getTable();
}
else
{

4
test/conf/storage-conf.xml
View File

@ -51,6 +51,10 @@
<ColumnFamily ColumnType="Super" ColumnSort="Name" Name="Super1"/>
<ColumnFamily ColumnType="Super" ColumnSort="Name" Name="Super2"/>
</Table>
<Table Name = "Table2">
<ColumnFamily ColumnSort="Name" Name="Standard1"/>
<ColumnFamily ColumnSort="Name" Name="Standard3"/>
</Table>
</Tables>
<Seeds>
<!-- Add names of hosts that are deemed contact points -->

40
test/unit/org/apache/cassandra/db/MultitableTest.java Normal file
View File

@ -0,0 +1,40 @@
package org.apache.cassandra.db;
import java.io.IOException;
import java.util.concurrent.ExecutionException;
import org.junit.Test;
import static org.apache.cassandra.db.TableTest.assertColumns;
import org.apache.cassandra.CleanupHelper;
public class MultitableTest extends CleanupHelper
{
@Test
public void testSameCFs() throws IOException, ExecutionException, InterruptedException
{
Table table1 = Table.open("Table1");
Table table2 = Table.open("Table2");
RowMutation rm;
ColumnFamily cf;
rm = new RowMutation("Table1", "keymulti");
cf = ColumnFamily.create("Table1", "Standard1");
cf.addColumn(new Column("col1", "val1".getBytes(), 1L));
rm.add(cf);
rm.apply();
rm = new RowMutation("Table2", "keymulti");
cf = ColumnFamily.create("Table2", "Standard1");
cf.addColumn(new Column("col2", "val2".getBytes(), 1L));
rm.add(cf);
rm.apply();
table1.getColumnFamilyStore("Standard1").forceBlockingFlush();
table2.getColumnFamilyStore("Standard1").forceBlockingFlush();
assertColumns(table1.get("keymulti", "Standard1"), "col1");
assertColumns(table2.get("keymulti", "Standard1"), "col2");
}
}

11
test/unit/org/apache/cassandra/db/RecoveryManagerTest.java
View File

@ -39,20 +39,29 @@ public class RecoveryManagerTest extends CleanupHelper
public void testSomething() throws IOException, ExecutionException, InterruptedException
{
Table table1 = Table.open("Table1");
Table table2 = Table.open("Table2");
RowMutation rm;
ColumnFamily cf;
rm = new RowMutation("Table1", "keymulti");
cf = new ColumnFamily("Standard1", "Standard");
cf = ColumnFamily.create("Table1", "Standard1");
cf.addColumn(new Column("col1", "val1".getBytes(), 1L));
rm.add(cf);
rm.apply();
rm = new RowMutation("Table2", "keymulti");
cf = ColumnFamily.create("Table2", "Standard3");
cf.addColumn(new Column("col2", "val2".getBytes(), 1L));
rm.add(cf);
rm.apply();
table1.getColumnFamilyStore("Standard1").clearUnsafe();
table2.getColumnFamilyStore("Standard3").clearUnsafe();
RecoveryManager.doRecovery();
assertColumns(table1.get("keymulti", "Standard1"), "col1");
assertColumns(table2.get("keymulti", "Standard3"), "col2");
}
}

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

@ -22,6 +22,7 @@ import java.util.SortedSet;
import java.util.Arrays;
import java.util.List;
import java.util.ArrayList;
import java.util.concurrent.ExecutionException;
import java.io.IOException;
import org.apache.commons.lang.StringUtils;
@ -35,7 +36,6 @@ public class TableTest extends CleanupHelper
{
private static final String KEY2 = "key2";
private static final String TEST_KEY = "key1";
private static final String TABLE_NAME = "Table1";
interface Runner
{
@ -60,7 +60,7 @@ public class TableTest extends CleanupHelper
@Test
public void testGetRowSingleColumn() throws Throwable
{
final Table table = Table.open(TABLE_NAME);
final Table table = Table.open("Table1");
Runner setup = new Runner()
{
public void run() throws Exception
@ -88,7 +88,7 @@ public class TableTest extends CleanupHelper
@Test
public void testGetRowOffsetCount() throws Throwable
{
final Table table = Table.open(TABLE_NAME);
final Table table = Table.open("Table1");
Runner setup = new Runner()
{
@ -127,7 +127,7 @@ public class TableTest extends CleanupHelper
{
Table table = Table.open("Table1");
RowMutation rm = new RowMutation(TABLE_NAME,KEY2);
RowMutation rm = new RowMutation("Table1",KEY2);
ColumnFamily cf = ColumnFamily.create("Table1", "Standard1");
// First write 5, 6
cf.addColumn(new Column("col5", "val5".getBytes(), 1L));
@ -142,7 +142,7 @@ public class TableTest extends CleanupHelper
table.getColumnFamilyStore("Standard1").forceBlockingFlush();
// Flushed memtable to disk, we're now inserting into a new memtable
rm = new RowMutation(TABLE_NAME, KEY2);
rm = new RowMutation("Table1", KEY2);
cf = ColumnFamily.create("Table1", "Standard1");
// now write 7, 8, 4 into new memtable
cf.addColumn(new Column("col7", "val7".getBytes(), 1L));
@ -174,8 +174,8 @@ public class TableTest extends CleanupHelper
public void testGetRowSliceByRange() throws Throwable
{
String key = TEST_KEY+"slicerow";
Table table = Table.open(TABLE_NAME);
RowMutation rm = new RowMutation(TABLE_NAME,key);
Table table = Table.open("Table1");
RowMutation rm = new RowMutation("Table1",key);
ColumnFamily cf = ColumnFamily.create("Table1", "Standard1");
// First write "a", "b", "c"
cf.addColumn(new Column("a", "val1".getBytes(), 1L));
@ -201,9 +201,9 @@ public class TableTest extends CleanupHelper
@Test
public void testGetRowSuperColumnOffsetCount() throws Throwable
{
Table table = Table.open(TABLE_NAME);
Table table = Table.open("Table1");
RowMutation rm = new RowMutation(TABLE_NAME,TEST_KEY);
RowMutation rm = new RowMutation("Table1",TEST_KEY);
ColumnFamily cf = ColumnFamily.create("Table1", "Super1");
SuperColumn sc1 = new SuperColumn("sc1");
sc1.addColumn(new Column("col1","val1".getBytes(), 1L));
@ -241,7 +241,7 @@ public class TableTest extends CleanupHelper
private RowMutation makeSimpleRowMutation()
{
RowMutation rm = new RowMutation(TABLE_NAME,TEST_KEY);
RowMutation rm = new RowMutation("Table1",TEST_KEY);
ColumnFamily cf = ColumnFamily.create("Table1", "Standard1");
cf.addColumn(new Column("col1","val1".getBytes(), 1L));
cf.addColumn(new Column("col2","val2".getBytes(), 1L));
@ -253,8 +253,8 @@ public class TableTest extends CleanupHelper
@Test
public void testGetSliceNoMatch() throws Throwable
{
Table table = Table.open(TABLE_NAME);
RowMutation rm = new RowMutation(TABLE_NAME, "row1000");
Table table = Table.open("Table1");
RowMutation rm = new RowMutation("Table1", "row1000");
ColumnFamily cf = ColumnFamily.create("Table1", "Standard2");
cf.addColumn(new Column("col1", "val1".getBytes(), 1));
rm.add(cf);
@ -290,13 +290,13 @@ 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
final Table table = Table.open(TABLE_NAME);
final Table table = Table.open("Table1");
final String ROW = "row1";
Runner setup = new Runner()
{
public void run() throws Exception
{
RowMutation rm = new RowMutation(TABLE_NAME, ROW);
RowMutation rm = new RowMutation("Table1", ROW);
ColumnFamily cf = ColumnFamily.create("Table1", "Standard1");
cf.addColumn(new Column("col1", "val1".getBytes(), 1L));
cf.addColumn(new Column("col3", "val3".getBytes(), 1L));
@ -307,7 +307,7 @@ public class TableTest extends CleanupHelper
rm.add(cf);
rm.apply();
rm = new RowMutation(TABLE_NAME, ROW);
rm = new RowMutation("Table1", ROW);
rm.delete("Standard1:col4", 2L);
rm.apply();
}
@ -353,13 +353,13 @@ public class TableTest extends CleanupHelper
public void testGetSliceFromAdvanced() throws Throwable
{
// tests slicing against data from one row spread across two sstables
final Table table = Table.open(TABLE_NAME);
final Table table = Table.open("Table1");
final String ROW = "row2";
Runner setup = new Runner()
{
public void run() throws Exception
{
RowMutation rm = new RowMutation(TABLE_NAME, ROW);
RowMutation rm = new RowMutation("Table1", ROW);
ColumnFamily cf = ColumnFamily.create("Table1", "Standard1");
cf.addColumn(new Column("col1", "val1".getBytes(), 1L));
cf.addColumn(new Column("col2", "val2".getBytes(), 1L));
@ -371,7 +371,7 @@ public class TableTest extends CleanupHelper
rm.apply();
table.getColumnFamilyStore("Standard1").forceBlockingFlush();
rm = new RowMutation(TABLE_NAME, ROW);
rm = new RowMutation("Table1", ROW);
cf = ColumnFamily.create("Table1", "Standard1");
cf.addColumn(new Column("col1", "valx".getBytes(), 2L));
cf.addColumn(new Column("col2", "valx".getBytes(), 2L));
@ -404,9 +404,9 @@ public class TableTest extends CleanupHelper
public void testGetSliceFromLarge() throws Throwable
{
// tests slicing against 1000 columns in an sstable
Table table = Table.open(TABLE_NAME);
Table table = Table.open("Table1");
String ROW = "row3";
RowMutation rm = new RowMutation(TABLE_NAME, ROW);
RowMutation rm = new RowMutation("Table1", ROW);
ColumnFamily cf = ColumnFamily.create("Table1", "Standard1");
for (int i = 1000; i < 2000; i++)
cf.addColumn(new Column("col" + i, ("vvvvvvvvvvvvvvvv" + i).getBytes(), 1L));