From 509cbb85de257a9c67311c7dd2fea18ed1b8f096 Mon Sep 17 00:00:00 2001
From: Jonathan Ellis <jbellis@apache.org>
Date: Mon, 22 Jun 2009 21:18:12 +0000
Subject: [PATCH] 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
---
 conf/storage-conf.xml                         |  7 +--
 .../cassandra/db/ColumnFamilyStore.java       |  5 ++-
 .../org/apache/cassandra/db/CommitLog.java    | 44 +++++++------------
 .../org/apache/cassandra/db/Memtable.java     |  2 +-
 .../org/apache/cassandra/db/ReadResponse.java | 22 +++-------
 .../apache/cassandra/db/ReadVerbHandler.java  |  4 +-
 .../apache/cassandra/db/RecoveryManager.java  |  2 +-
 src/java/org/apache/cassandra/db/Row.java     | 30 +++++--------
 .../org/apache/cassandra/db/SystemTable.java  |  4 +-
 src/java/org/apache/cassandra/db/Table.java   |  7 ++-
 .../service/ReadResponseResolver.java         |  2 +-
 test/conf/storage-conf.xml                    |  4 ++
 .../apache/cassandra/db/MultitableTest.java   | 40 +++++++++++++++++
 .../cassandra/db/RecoveryManagerTest.java     | 11 ++++-
 .../org/apache/cassandra/db/TableTest.java    | 40 ++++++++---------
 15 files changed, 127 insertions(+), 97 deletions(-)
 create mode 100644 test/unit/org/apache/cassandra/db/MultitableTest.java

diff --git a/conf/storage-conf.xml b/conf/storage-conf.xml
index 3316f17560..a962e3806e 100644
--- a/conf/storage-conf.xml
+++ b/conf/storage-conf.xml
@@ -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>
 
diff --git a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
index 4174bfae43..460c19187f 100644
--- a/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
+++ b/src/java/org/apache/cassandra/db/ColumnFamilyStore.java
@@ -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();
diff --git a/src/java/org/apache/cassandra/db/CommitLog.java b/src/java/org/apache/cassandra/db/CommitLog.java
index 7c41eb93fb..955a203779 100644
--- a/src/java/org/apache/cassandra/db/CommitLog.java
+++ b/src/java/org/apache/cassandra/db/CommitLog.java
@@ -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);
diff --git a/src/java/org/apache/cassandra/db/Memtable.java b/src/java/org/apache/cassandra/db/Memtable.java
index 5024bd6d1a..7218497b3d 100644
--- a/src/java/org/apache/cassandra/db/Memtable.java
+++ b/src/java/org/apache/cassandra/db/Memtable.java
@@ -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)
         {
diff --git a/src/java/org/apache/cassandra/db/ReadResponse.java b/src/java/org/apache/cassandra/db/ReadResponse.java
index 0ea6bccd84..92f7e9b618 100644
--- a/src/java/org/apache/cassandra/db/ReadResponse.java
+++ b/src/java/org/apache/cassandra/db/ReadResponse.java
@@ -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;
diff --git a/src/java/org/apache/cassandra/db/ReadVerbHandler.java b/src/java/org/apache/cassandra/db/ReadVerbHandler.java
index 82e113c432..a400837bf5 100644
--- a/src/java/org/apache/cassandra/db/ReadVerbHandler.java
+++ b/src/java/org/apache/cassandra/db/ReadVerbHandler.java
@@ -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. */
diff --git a/src/java/org/apache/cassandra/db/RecoveryManager.java b/src/java/org/apache/cassandra/db/RecoveryManager.java
index 92a9e30083..9a69e56406 100644
--- a/src/java/org/apache/cassandra/db/RecoveryManager.java
+++ b/src/java/org/apache/cassandra/db/RecoveryManager.java
@@ -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);
     }
 }
diff --git a/src/java/org/apache/cassandra/db/Row.java b/src/java/org/apache/cassandra/db/Row.java
index 8e56e0708b..54d3aa2552 100644
--- a/src/java/org/apache/cassandra/db/Row.java
+++ b/src/java/org/apache/cassandra/db/Row.java
@@ -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;
     }
diff --git a/src/java/org/apache/cassandra/db/SystemTable.java b/src/java/org/apache/cassandra/db/SystemTable.java
index de6c3f85ff..615f5f58f4 100644
--- a/src/java/org/apache/cassandra/db/SystemTable.java
+++ b/src/java/org/apache/cassandra/db/SystemTable.java
@@ -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);
diff --git a/src/java/org/apache/cassandra/db/Table.java b/src/java/org/apache/cassandra/db/Table.java
index c15a1a78ee..34e9427b8b 100644
--- a/src/java/org/apache/cassandra/db/Table.java
+++ b/src/java/org/apache/cassandra/db/Table.java
@@ -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())
         {
diff --git a/src/java/org/apache/cassandra/service/ReadResponseResolver.java b/src/java/org/apache/cassandra/service/ReadResponseResolver.java
index 9cf391144d..f302d344d9 100644
--- a/src/java/org/apache/cassandra/service/ReadResponseResolver.java
+++ b/src/java/org/apache/cassandra/service/ReadResponseResolver.java
@@ -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
     			{
diff --git a/test/conf/storage-conf.xml b/test/conf/storage-conf.xml
index d5985d214a..d3a72d9ee6 100644
--- a/test/conf/storage-conf.xml
+++ b/test/conf/storage-conf.xml
@@ -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 -->
diff --git a/test/unit/org/apache/cassandra/db/MultitableTest.java b/test/unit/org/apache/cassandra/db/MultitableTest.java
new file mode 100644
index 0000000000..b9223fdd0f
--- /dev/null
+++ b/test/unit/org/apache/cassandra/db/MultitableTest.java
@@ -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");
+    }
+}
diff --git a/test/unit/org/apache/cassandra/db/RecoveryManagerTest.java b/test/unit/org/apache/cassandra/db/RecoveryManagerTest.java
index e2edc37b64..10efac3745 100644
--- a/test/unit/org/apache/cassandra/db/RecoveryManagerTest.java
+++ b/test/unit/org/apache/cassandra/db/RecoveryManagerTest.java
@@ -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");
     }
 }
diff --git a/test/unit/org/apache/cassandra/db/TableTest.java b/test/unit/org/apache/cassandra/db/TableTest.java
index 02b994662a..42ac909247 100644
--- a/test/unit/org/apache/cassandra/db/TableTest.java
+++ b/test/unit/org/apache/cassandra/db/TableTest.java
@@ -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));