From fe708caa69cccecf5f373624fbf8e73a00a18983 Mon Sep 17 00:00:00 2001
From: Jonathan Ellis <jbellis@apache.org>
Date: Fri, 27 Mar 2009 16:39:00 +0000
Subject: [PATCH] clean up getLocalCopy

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@759222 13f79535-47bb-0310-9956-ffa450edef68
---
 src/org/apache/cassandra/db/Memtable.java | 57 +++++++++++------------
 1 file changed, 28 insertions(+), 29 deletions(-)

diff --git a/src/org/apache/cassandra/db/Memtable.java b/src/org/apache/cassandra/db/Memtable.java
index 653e04d79a..54c532329d 100644
--- a/src/org/apache/cassandra/db/Memtable.java
+++ b/src/org/apache/cassandra/db/Memtable.java
@@ -36,6 +36,8 @@ import java.util.concurrent.locks.Condition;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 
+import org.apache.log4j.Logger;
+
 import org.apache.cassandra.concurrent.DebuggableThreadPoolExecutor;
 import org.apache.cassandra.concurrent.ThreadFactoryImpl;
 import org.apache.cassandra.config.DatabaseDescriptor;
@@ -45,7 +47,6 @@ import org.apache.cassandra.service.PartitionerType;
 import org.apache.cassandra.service.StorageService;
 import org.apache.cassandra.utils.BloomFilter;
 import org.apache.cassandra.utils.LogUtil;
-import org.apache.log4j.Logger;
 
 /**
  * Author : Avinash Lakshman ( alakshman@facebook.com) & Prashant Malik ( pmalik@facebook.com )
@@ -326,46 +327,44 @@ public class Memtable implements MemtableMBean, Comparable<Memtable>
         	resolve(key, columnFamily);
     }
 
-    ColumnFamily getLocalCopy(String key, String cfName, IFilter filter)
+    ColumnFamily getLocalCopy(String key, String columnFamilyColumn, IFilter filter)
     {
-    	String[] values = RowMutation.getColumnAndColumnFamily(cfName);
+    	String[] values = RowMutation.getColumnAndColumnFamily(columnFamilyColumn);
     	ColumnFamily columnFamily = null;
         if(values.length == 1 )
         {
-        	columnFamily = columnFamilies_.get(key);        	
+        	columnFamily = columnFamilies_.get(key);
         }
         else
         {
         	ColumnFamily cFamily = columnFamilies_.get(key);
-        	if(cFamily == null)
-        		return null;
-        	IColumn column = null;
-        	if(values.length == 2)
-        	{
-        		column = cFamily.getColumn(values[1]);
-        		if(column != null )
-        		{
-        			columnFamily = new ColumnFamily(cfName_);
-        			columnFamily.addColumn(column);
-        		}
+        	if (cFamily == null) return null;
+
+        	if (values.length == 2) {
+                IColumn column = cFamily.getColumn(values[1]); // super or normal column
+                if (column != null )
+                {
+                    columnFamily = new ColumnFamily(cfName_);
+                    columnFamily.addColumn(column);
+                }
         	}
-        	else
-        	{
-        		column = cFamily.getColumn(values[1]);
-        		if(column != null )
-        		{
-        			 
-        			IColumn subColumn = ((SuperColumn)column).getSubColumn(values[2]);
-        			if(subColumn != null)
-        			{
-	        			columnFamily = new ColumnFamily(cfName_);
-	            		columnFamily.addColumn(values[1] + ":" + values[2], subColumn.value(), subColumn.timestamp());
-        			}
-        		}
+            else
+            {
+                assert values.length == 3;
+                SuperColumn superColumn = (SuperColumn)cFamily.getColumn(values[1]);
+                if (superColumn != null)
+                {
+                    IColumn subColumn = superColumn.getSubColumn(values[2]);
+                    if (subColumn != null)
+                    {
+                        columnFamily = new ColumnFamily(cfName_);
+                        columnFamily.addColumn(values[1] + ":" + values[2], subColumn.value(), subColumn.timestamp(), subColumn.isMarkedForDelete());
+                    }
+                }
         	}
         }
         /* Filter unnecessary data from the column based on the provided filter */
-        return filter.filter(cfName, columnFamily);
+        return filter.filter(columnFamilyColumn, columnFamily);
     }
 
     ColumnFamily get(String key, String cfName)