phoenix-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From jamestay...@apache.org
Subject [3/3] phoenix git commit: PHOENIX-4095 Prevent index from getting out of sync with data table during partial rebuild
Date Fri, 18 Aug 2017 00:03:59 GMT
PHOENIX-4095 Prevent index from getting out of sync with data table during partial rebuild


Project: http://git-wip-us.apache.org/repos/asf/phoenix/repo
Commit: http://git-wip-us.apache.org/repos/asf/phoenix/commit/572d242d
Tree: http://git-wip-us.apache.org/repos/asf/phoenix/tree/572d242d
Diff: http://git-wip-us.apache.org/repos/asf/phoenix/diff/572d242d

Branch: refs/heads/4.x-HBase-1.2
Commit: 572d242dca11c93fcc9d2dca0bf5c836a05ce36b
Parents: 9219bc2
Author: James Taylor <jamestaylor@apache.org>
Authored: Thu Aug 17 15:47:11 2017 -0700
Committer: James Taylor <jamestaylor@apache.org>
Committed: Thu Aug 17 17:02:22 2017 -0700

----------------------------------------------------------------------
 .../phoenix/end2end/ConcurrentMutationsIT.java  |   3 +-
 .../phoenix/end2end/OutOfOrderMutationsIT.java  |  19 +-
 .../end2end/index/PartialIndexRebuilderIT.java  | 197 +++++++++--
 .../apache/phoenix/util/IndexScrutinyIT.java    |  89 +++++
 .../org/apache/phoenix/util/TestUtilIT.java     |  89 -----
 .../UngroupedAggregateRegionObserver.java       |   5 +
 .../org/apache/phoenix/hbase/index/Indexer.java | 171 ++++------
 .../hbase/index/covered/NonTxIndexBuilder.java  |  93 +----
 .../covered/example/CoveredColumnIndexer.java   |   5 +-
 .../hbase/index/util/IndexManagementUtil.java   |  94 +++++
 .../hbase/index/covered/CoveredColumnsTest.java |  46 +++
 .../index/covered/LocalTableStateTest.java      | 293 ++++++++++++++++
 .../index/covered/NonTxIndexBuilderTest.java    | 341 +++++++++++++++++++
 .../hbase/index/covered/TestCoveredColumns.java |  46 ---
 .../index/covered/TestLocalTableState.java      | 293 ----------------
 .../index/covered/TestNonTxIndexBuilder.java    | 335 ------------------
 .../org/apache/phoenix/util/IndexScrutiny.java  | 144 ++++++++
 .../java/org/apache/phoenix/util/TestUtil.java  | 109 +-----
 18 files changed, 1289 insertions(+), 1083 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/phoenix/blob/572d242d/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsIT.java
index d3e3761..e674d8f 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/ConcurrentMutationsIT.java
@@ -47,6 +47,7 @@ import org.apache.phoenix.jdbc.PhoenixConnection;
 import org.apache.phoenix.query.ConnectionQueryServices;
 import org.apache.phoenix.query.QueryServices;
 import org.apache.phoenix.query.QueryServicesOptions;
+import org.apache.phoenix.util.IndexScrutiny;
 import org.apache.phoenix.util.PhoenixRuntime;
 import org.apache.phoenix.util.PropertiesUtil;
 import org.apache.phoenix.util.ReadOnlyProps;
@@ -274,7 +275,7 @@ public class ConcurrentMutationsIT extends BaseUniqueNamesOwnClusterIT {
         }
         
         assertTrue("Ran out of time", doneSignal.await(120, TimeUnit.SECONDS));
-        long actualRowCount = TestUtil.scrutinizeIndex(conn, tableName, indexName);
+        long actualRowCount = IndexScrutiny.scrutinizeIndex(conn, tableName, indexName);
         assertEquals(nRows, actualRowCount);
     }
 

http://git-wip-us.apache.org/repos/asf/phoenix/blob/572d242d/phoenix-core/src/it/java/org/apache/phoenix/end2end/OutOfOrderMutationsIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/OutOfOrderMutationsIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/OutOfOrderMutationsIT.java
index 5cdc1ee..e8adf6b 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/OutOfOrderMutationsIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/OutOfOrderMutationsIT.java
@@ -38,6 +38,7 @@ import org.apache.hadoop.hbase.client.ResultScanner;
 import org.apache.hadoop.hbase.client.Scan;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.phoenix.jdbc.PhoenixConnection;
+import org.apache.phoenix.util.IndexScrutiny;
 import org.apache.phoenix.util.PhoenixRuntime;
 import org.apache.phoenix.util.PropertiesUtil;
 import org.apache.phoenix.util.TestUtil;
@@ -91,7 +92,7 @@ public class OutOfOrderMutationsIT extends ParallelStatsDisabledIT {
         props.setProperty(PhoenixRuntime.CURRENT_SCN_ATTRIB, Long.toString(ts));
         conn = DriverManager.getConnection(getUrl(), props);
         
-        TestUtil.scrutinizeIndex(conn, tableName, indexName);        
+        IndexScrutiny.scrutinizeIndex(conn, tableName, indexName);        
         assertNoTimeStampAt(conn, indexName, 1030);
         conn.close();
         
@@ -175,7 +176,7 @@ public class OutOfOrderMutationsIT extends ParallelStatsDisabledIT {
         props.setProperty(PhoenixRuntime.CURRENT_SCN_ATTRIB, Long.toString(ts));
         conn = DriverManager.getConnection(getUrl(), props);
         
-        TestUtil.scrutinizeIndex(conn, tableName, indexName);        
+        IndexScrutiny.scrutinizeIndex(conn, tableName, indexName);        
         
         ResultSet rs = conn.createStatement().executeQuery("SELECT /*+ NO_INDEX */ ts FROM " + tableName);
         assertTrue(rs.next());
@@ -265,7 +266,7 @@ public class OutOfOrderMutationsIT extends ParallelStatsDisabledIT {
         TestUtil.dumpTable(conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes(tableName)));
         TestUtil.dumpTable(conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes(indexName)));
 
-        TestUtil.scrutinizeIndex(conn, tableName, indexName);        
+        IndexScrutiny.scrutinizeIndex(conn, tableName, indexName);        
         
         ResultSet rs = conn.createStatement().executeQuery("SELECT /*+ NO_INDEX */ ts FROM " + tableName);
         assertTrue(rs.next());
@@ -331,7 +332,7 @@ public class OutOfOrderMutationsIT extends ParallelStatsDisabledIT {
         TestUtil.dumpTable(conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes(tableName)));
         TestUtil.dumpTable(conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes(indexName)));
 
-        TestUtil.scrutinizeIndex(conn, tableName, indexName);        
+        IndexScrutiny.scrutinizeIndex(conn, tableName, indexName);        
         
         ResultSet rs = conn.createStatement().executeQuery("SELECT /*+ NO_INDEX */ ts FROM " + tableName);
         assertTrue(rs.next());
@@ -393,7 +394,7 @@ public class OutOfOrderMutationsIT extends ParallelStatsDisabledIT {
         TestUtil.dumpTable(conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes(tableName)));
         TestUtil.dumpTable(conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes(indexName)));
 
-        TestUtil.scrutinizeIndex(conn, tableName, indexName);        
+        IndexScrutiny.scrutinizeIndex(conn, tableName, indexName);        
         
         ResultSet rs = conn.createStatement().executeQuery("SELECT /*+ NO_INDEX */ ts,v FROM " + tableName);
         assertTrue(rs.next());
@@ -458,7 +459,7 @@ public class OutOfOrderMutationsIT extends ParallelStatsDisabledIT {
         TestUtil.dumpTable(conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes(tableName)));
         TestUtil.dumpTable(conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes(indexName)));
 
-        TestUtil.scrutinizeIndex(conn, tableName, indexName);        
+        IndexScrutiny.scrutinizeIndex(conn, tableName, indexName);        
         
         ResultSet rs = conn.createStatement().executeQuery("SELECT /*+ NO_INDEX */ ts,v FROM " + tableName);
         assertTrue(rs.next());
@@ -523,7 +524,7 @@ public class OutOfOrderMutationsIT extends ParallelStatsDisabledIT {
         TestUtil.dumpTable(conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes(tableName)));
         TestUtil.dumpTable(conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes(indexName)));
 
-        TestUtil.scrutinizeIndex(conn, tableName, indexName);        
+        IndexScrutiny.scrutinizeIndex(conn, tableName, indexName);        
         
         ResultSet rs = conn.createStatement().executeQuery("SELECT /*+ NO_INDEX */ ts,v FROM " + tableName);
         assertTrue(rs.next());
@@ -587,7 +588,7 @@ public class OutOfOrderMutationsIT extends ParallelStatsDisabledIT {
         TestUtil.dumpTable(conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes(tableName)));
         TestUtil.dumpTable(conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes(indexName)));
 
-        long rowCount = TestUtil.scrutinizeIndex(conn, tableName, indexName);
+        long rowCount = IndexScrutiny.scrutinizeIndex(conn, tableName, indexName);
         assertEquals(0,rowCount);
         
         conn.close();
@@ -640,7 +641,7 @@ public class OutOfOrderMutationsIT extends ParallelStatsDisabledIT {
         TestUtil.dumpTable(conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes(tableName)));
         TestUtil.dumpTable(conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes(indexName)));
 
-        long rowCount = TestUtil.scrutinizeIndex(conn, tableName, indexName);
+        long rowCount = IndexScrutiny.scrutinizeIndex(conn, tableName, indexName);
         assertEquals(0,rowCount);
         
         conn.close();

http://git-wip-us.apache.org/repos/asf/phoenix/blob/572d242d/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/PartialIndexRebuilderIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/PartialIndexRebuilderIT.java b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/PartialIndexRebuilderIT.java
index 8c43e3f..bc0dda8 100644
--- a/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/PartialIndexRebuilderIT.java
+++ b/phoenix-core/src/it/java/org/apache/phoenix/end2end/index/PartialIndexRebuilderIT.java
@@ -18,44 +18,203 @@
 package org.apache.phoenix.end2end.index;
 
 import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.fail;
+import static org.junit.Assert.assertTrue;
 
 import java.sql.Connection;
 import java.sql.DriverManager;
+import java.sql.SQLException;
 import java.util.Map;
 import java.util.Properties;
+import java.util.Random;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
 
 import org.apache.hadoop.hbase.client.HTableInterface;
-import org.apache.hadoop.hbase.util.Bytes;
 import org.apache.phoenix.end2end.BaseUniqueNamesOwnClusterIT;
 import org.apache.phoenix.jdbc.PhoenixConnection;
 import org.apache.phoenix.jdbc.PhoenixDatabaseMetaData;
 import org.apache.phoenix.query.QueryServices;
 import org.apache.phoenix.schema.PIndexState;
+import org.apache.phoenix.schema.PMetaData;
+import org.apache.phoenix.schema.PTable;
+import org.apache.phoenix.schema.PTableKey;
+import org.apache.phoenix.schema.TableNotFoundException;
 import org.apache.phoenix.util.EnvironmentEdgeManager;
+import org.apache.phoenix.util.IndexScrutiny;
 import org.apache.phoenix.util.IndexUtil;
 import org.apache.phoenix.util.PhoenixRuntime;
 import org.apache.phoenix.util.PropertiesUtil;
 import org.apache.phoenix.util.ReadOnlyProps;
+import org.apache.phoenix.util.Repeat;
+import org.apache.phoenix.util.RunUntilFailure;
 import org.apache.phoenix.util.SchemaUtil;
 import org.apache.phoenix.util.TestUtil;
 import org.junit.BeforeClass;
 import org.junit.Test;
+import org.junit.runner.RunWith;
 
 import com.google.common.collect.Maps;
 
+@RunWith(RunUntilFailure.class)
 public class PartialIndexRebuilderIT extends BaseUniqueNamesOwnClusterIT {
+    private static final Random RAND = new Random(5);
+
     @BeforeClass
     public static void doSetup() throws Exception {
         Map<String, String> serverProps = Maps.newHashMapWithExpectedSize(10);
         serverProps.put(QueryServices.INDEX_FAILURE_HANDLING_REBUILD_ATTRIB, Boolean.TRUE.toString());
         serverProps.put(QueryServices.INDEX_FAILURE_HANDLING_REBUILD_INTERVAL_ATTRIB, "1000");
         serverProps.put(QueryServices.INDEX_REBUILD_DISABLE_TIMESTAMP_THRESHOLD, "30000"); // give up rebuilding after 30 seconds
-        serverProps.put(QueryServices.INDEX_FAILURE_HANDLING_REBUILD_OVERLAP_FORWARD_TIME_ATTRIB, Long.toString(1000));
+        serverProps.put(QueryServices.INDEX_FAILURE_HANDLING_REBUILD_OVERLAP_FORWARD_TIME_ATTRIB, Long.toString(2000));
         setUpTestDriver(new ReadOnlyProps(serverProps.entrySet().iterator()), ReadOnlyProps.EMPTY_PROPS);
     }
 
+    private static boolean mutateRandomly(Connection conn, String fullTableName, int nRows) throws Exception {
+        return mutateRandomly(conn, fullTableName, nRows, false);
+    }
+    
+    private static boolean hasInactiveIndex(PMetaData metaCache, PTableKey key) throws TableNotFoundException {
+        PTable table = metaCache.getTableRef(key).getTable();
+        for (PTable index : table.getIndexes()) {
+            if (index.getIndexState() == PIndexState.INACTIVE) {
+                return true;
+            }
+        }
+        return false;
+    }
+    
+    private static boolean isAllActiveIndex(PMetaData metaCache, PTableKey key) throws TableNotFoundException {
+        PTable table = metaCache.getTableRef(key).getTable();
+        for (PTable index : table.getIndexes()) {
+            if (index.getIndexState() != PIndexState.ACTIVE) {
+                return false;
+            }
+        }
+        return true;
+    }
+    
+    private static boolean mutateRandomly(Connection conn, String fullTableName, int nRows, boolean checkForInactive) throws SQLException, InterruptedException {
+        PTableKey key = new PTableKey(null,fullTableName);
+        PMetaData metaCache = conn.unwrap(PhoenixConnection.class).getMetaDataCache();
+        boolean hasInactiveIndex = false;
+        int batchSize = checkForInactive && !isAllActiveIndex(metaCache, key) ? 1 : 200;
+        for (int i = 0; i < 10000; i++) {
+            int pk = Math.abs(RAND.nextInt()) % nRows;
+            int v1 = Math.abs(RAND.nextInt()) % nRows;
+            int v2 = Math.abs(RAND.nextInt()) % nRows;
+            if (checkForInactive && hasInactiveIndex(metaCache, key)) {
+                checkForInactive = false;
+                hasInactiveIndex = true;
+                batchSize = 200;
+            }
+            conn.createStatement().execute("UPSERT INTO " + fullTableName + " VALUES(" + pk + "," + v1 + "," + v2 + ")");
+            if (i % batchSize == 0) {
+                conn.commit();
+                if (checkForInactive) Thread.sleep(100);
+            }
+        }
+        conn.commit();
+        for (int i = 0; i < 10000; i++) {
+            int pk = Math.abs(RAND.nextInt()) % nRows;
+            if (checkForInactive && hasInactiveIndex(metaCache, key)) {
+                checkForInactive = false;
+                hasInactiveIndex = true;
+                batchSize = 200;
+            }
+            conn.createStatement().execute("DELETE FROM " + fullTableName + " WHERE k= " + pk);
+            if (i % batchSize == 0) {
+                conn.commit();
+            }
+        }
+        conn.commit();
+        for (int i = 0; i < 10000; i++) {
+            int pk = Math.abs(RAND.nextInt()) % nRows;
+            int v1 = Math.abs(RAND.nextInt()) % nRows;
+            int v2 = Math.abs(RAND.nextInt()) % nRows;
+            if (checkForInactive && hasInactiveIndex(metaCache, key)) {
+                checkForInactive = false;
+                hasInactiveIndex = true;
+                batchSize = 200;
+            }
+            conn.createStatement().execute("UPSERT INTO " + fullTableName + " VALUES(" + pk + "," + v1 + "," + v2 + ")");
+            if (i % batchSize == 0) {
+                conn.commit();
+            }
+        }
+        conn.commit();
+        return hasInactiveIndex;
+    }
+    
+    @Test
+    @Repeat(20)
+    public void testDeleteAndUpsertAfterFailure() throws Throwable {
+        final int nRows = 10;
+        String schemaName = generateUniqueName();
+        String tableName = generateUniqueName();
+        String indexName = generateUniqueName();
+        final String fullTableName = SchemaUtil.getTableName(schemaName, tableName);
+        String fullIndexName = SchemaUtil.getTableName(schemaName, indexName);
+        try (Connection conn = DriverManager.getConnection(getUrl())) {
+            conn.createStatement().execute("CREATE TABLE " + fullTableName + "(k INTEGER PRIMARY KEY, v1 INTEGER, v2 INTEGER) COLUMN_ENCODED_BYTES = 0, STORE_NULLS=true");
+            conn.createStatement().execute("CREATE INDEX " + indexName + " ON " + fullTableName + " (v1) INCLUDE (v2)");
+            mutateRandomly(conn, fullTableName, nRows);
+            long disableTS = EnvironmentEdgeManager.currentTimeMillis();
+            HTableInterface metaTable = conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME_BYTES);
+            IndexUtil.updateIndexState(fullIndexName, disableTS, metaTable, PIndexState.DISABLE);
+            mutateRandomly(conn, fullTableName, nRows);
+            TestUtil.waitForIndexRebuild(conn, fullIndexName, PIndexState.ACTIVE);
+            
+            long actualRowCount = IndexScrutiny.scrutinizeIndex(conn, fullTableName, fullIndexName);
+            assertEquals(nRows,actualRowCount);
+       }
+    }
+    
     @Test
+    public void testWriteWhileRebuilding() throws Throwable {
+        final int nRows = 10;
+        String schemaName = generateUniqueName();
+        String tableName = generateUniqueName();
+        String indexName = generateUniqueName();
+        final String fullTableName = SchemaUtil.getTableName(schemaName, tableName);
+        String fullIndexName = SchemaUtil.getTableName(schemaName, indexName);
+        try (Connection conn = DriverManager.getConnection(getUrl())) {
+            conn.createStatement().execute("CREATE TABLE " + fullTableName + "(k INTEGER PRIMARY KEY, v1 INTEGER, v2 INTEGER) COLUMN_ENCODED_BYTES = 0, STORE_NULLS=true");
+            conn.createStatement().execute("CREATE INDEX " + indexName + " ON " + fullTableName + " (v1) INCLUDE (v2)");
+            mutateRandomly(conn, fullTableName, nRows);
+            long disableTS = EnvironmentEdgeManager.currentTimeMillis();
+            HTableInterface metaTable = conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(PhoenixDatabaseMetaData.SYSTEM_CATALOG_NAME_BYTES);
+            IndexUtil.updateIndexState(fullIndexName, disableTS, metaTable, PIndexState.DISABLE);
+            mutateRandomly(conn, fullTableName, nRows);
+            final boolean[] hasInactiveIndex = new boolean[1];
+            final CountDownLatch doneSignal = new CountDownLatch(1);
+            Runnable r = new Runnable() {
+
+                @Override
+                public void run() {
+                    try {
+                        Connection conn = DriverManager.getConnection(getUrl());
+                        hasInactiveIndex[0] = mutateRandomly(conn, fullTableName, nRows, true);
+                    } catch (Exception e) {
+                        throw new RuntimeException(e);
+                    } finally {
+                        doneSignal.countDown();
+                    }
+                }
+                
+            };
+            Thread t = new Thread(r);
+            t.setDaemon(true);
+            t.start();
+            TestUtil.waitForIndexRebuild(conn, fullIndexName, PIndexState.ACTIVE);
+            doneSignal.await(120, TimeUnit.SECONDS);
+            assertTrue(hasInactiveIndex[0]);
+            
+            long actualRowCount = IndexScrutiny.scrutinizeIndex(conn, fullTableName, fullIndexName);
+            assertEquals(nRows,actualRowCount);
+       }
+    }
+
+      @Test
     public void testMultiVersionsAfterFailure() throws Throwable {
         String schemaName = generateUniqueName();
         String tableName = generateUniqueName();
@@ -77,10 +236,8 @@ public class PartialIndexRebuilderIT extends BaseUniqueNamesOwnClusterIT {
             conn.createStatement().execute("UPSERT INTO " + fullTableName + " VALUES('a','eeeee')");
             conn.commit();
             TestUtil.waitForIndexRebuild(conn, fullIndexName, PIndexState.ACTIVE);
-            TestUtil.dumpTable(conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes(fullTableName)));
-            TestUtil.dumpTable(conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes(fullIndexName)));
 
-            TestUtil.scrutinizeIndex(conn, fullTableName, fullIndexName);
+            IndexScrutiny.scrutinizeIndex(conn, fullTableName, fullIndexName);
         }
     }
     
@@ -106,10 +263,8 @@ public class PartialIndexRebuilderIT extends BaseUniqueNamesOwnClusterIT {
             conn.createStatement().execute("UPSERT INTO " + fullTableName + " VALUES('a','ccc')");
             conn.commit();
             TestUtil.waitForIndexRebuild(conn, fullIndexName, PIndexState.ACTIVE);
-            TestUtil.dumpTable(conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes(fullTableName)));
-            TestUtil.dumpTable(conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes(fullIndexName)));
 
-            TestUtil.scrutinizeIndex(conn, fullTableName, fullIndexName);
+            IndexScrutiny.scrutinizeIndex(conn, fullTableName, fullIndexName);
         }
     }
     
@@ -135,10 +290,8 @@ public class PartialIndexRebuilderIT extends BaseUniqueNamesOwnClusterIT {
             conn.createStatement().execute("UPSERT INTO " + fullTableName + " VALUES('a',null)");
             conn.commit();
             TestUtil.waitForIndexRebuild(conn, fullIndexName, PIndexState.ACTIVE);
-            TestUtil.dumpTable(conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes(fullTableName)));
-            TestUtil.dumpTable(conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes(fullIndexName)));
 
-            TestUtil.scrutinizeIndex(conn, fullTableName, fullIndexName);
+            IndexScrutiny.scrutinizeIndex(conn, fullTableName, fullIndexName);
         }
     }
     
@@ -162,10 +315,8 @@ public class PartialIndexRebuilderIT extends BaseUniqueNamesOwnClusterIT {
             conn.createStatement().execute("DELETE FROM " + fullTableName);
             conn.commit();
             TestUtil.waitForIndexRebuild(conn, fullIndexName, PIndexState.ACTIVE);
-            TestUtil.dumpTable(conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes(fullTableName)));
-            TestUtil.dumpTable(conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes(fullIndexName)));
 
-            TestUtil.scrutinizeIndex(conn, fullTableName, fullIndexName);
+            IndexScrutiny.scrutinizeIndex(conn, fullTableName, fullIndexName);
        }
     }
     
@@ -189,10 +340,8 @@ public class PartialIndexRebuilderIT extends BaseUniqueNamesOwnClusterIT {
             conn.createStatement().execute("UPSERT INTO " + fullTableName + " VALUES('a','b')");
             conn.commit();
             TestUtil.waitForIndexRebuild(conn, fullIndexName, PIndexState.ACTIVE);
-            TestUtil.dumpTable(conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes(fullTableName)));
-            TestUtil.dumpTable(conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes(fullIndexName)));
 
-            TestUtil.scrutinizeIndex(conn, fullTableName, fullIndexName);
+            IndexScrutiny.scrutinizeIndex(conn, fullTableName, fullIndexName);
         }
     }
     
@@ -220,10 +369,8 @@ public class PartialIndexRebuilderIT extends BaseUniqueNamesOwnClusterIT {
                 conn2.commit();
             }
             TestUtil.waitForIndexRebuild(conn, fullIndexName, PIndexState.ACTIVE);
-            TestUtil.dumpTable(conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes(fullTableName)));
-            TestUtil.dumpTable(conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes(fullIndexName)));
 
-            TestUtil.scrutinizeIndex(conn, fullTableName, fullIndexName);
+            IndexScrutiny.scrutinizeIndex(conn, fullTableName, fullIndexName);
         }
     }
     
@@ -251,10 +398,8 @@ public class PartialIndexRebuilderIT extends BaseUniqueNamesOwnClusterIT {
                 conn2.commit();
             }
             TestUtil.waitForIndexRebuild(conn, fullIndexName, PIndexState.ACTIVE);
-            TestUtil.dumpTable(conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes(fullTableName)));
-            TestUtil.dumpTable(conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes(fullIndexName)));
 
-            TestUtil.scrutinizeIndex(conn, fullTableName, fullIndexName);
+            IndexScrutiny.scrutinizeIndex(conn, fullTableName, fullIndexName);
         }
     }
     
@@ -282,10 +427,8 @@ public class PartialIndexRebuilderIT extends BaseUniqueNamesOwnClusterIT {
                 conn2.commit();
             }
             TestUtil.waitForIndexRebuild(conn, fullIndexName, PIndexState.ACTIVE);
-            TestUtil.dumpTable(conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes(fullTableName)));
-            TestUtil.dumpTable(conn.unwrap(PhoenixConnection.class).getQueryServices().getTable(Bytes.toBytes(fullIndexName)));
 
-            TestUtil.scrutinizeIndex(conn, fullTableName, fullIndexName);
+            IndexScrutiny.scrutinizeIndex(conn, fullTableName, fullIndexName);
         }
     }
 }

http://git-wip-us.apache.org/repos/asf/phoenix/blob/572d242d/phoenix-core/src/it/java/org/apache/phoenix/util/IndexScrutinyIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/util/IndexScrutinyIT.java b/phoenix-core/src/it/java/org/apache/phoenix/util/IndexScrutinyIT.java
new file mode 100644
index 0000000..a703294
--- /dev/null
+++ b/phoenix-core/src/it/java/org/apache/phoenix/util/IndexScrutinyIT.java
@@ -0,0 +1,89 @@
+package org.apache.phoenix.util;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+import java.sql.Connection;
+import java.sql.DriverManager;
+
+import org.apache.phoenix.end2end.ParallelStatsDisabledIT;
+import org.junit.Test;
+
+public class IndexScrutinyIT extends ParallelStatsDisabledIT {
+    @Test
+    public void testRowCountIndexScrutiny() throws Throwable {
+        String schemaName = generateUniqueName();
+        String tableName = generateUniqueName();
+        String indexName = generateUniqueName();
+        String fullTableName = SchemaUtil.getTableName(schemaName, tableName);
+        String fullIndexName = SchemaUtil.getTableName(schemaName, indexName);
+        try (Connection conn = DriverManager.getConnection(getUrl())) {
+            conn.createStatement().execute("CREATE TABLE " + fullTableName + "(k VARCHAR PRIMARY KEY, v VARCHAR) COLUMN_ENCODED_BYTES = 0, STORE_NULLS=true");
+            conn.createStatement().execute("CREATE INDEX " + indexName + " ON " + fullTableName + " (v)");
+            conn.createStatement().execute("UPSERT INTO " + fullTableName + " VALUES('b','bb')");
+            conn.createStatement().execute("UPSERT INTO " + fullTableName + " VALUES('a','ccc')");
+            conn.commit();
+            
+            int count = conn.createStatement().executeUpdate("DELETE FROM " + fullIndexName + " WHERE \":K\"='a' AND \"0:V\"='ccc'");
+            assertEquals(1,count);
+            conn.commit();
+            try {
+                IndexScrutiny.scrutinizeIndex(conn, fullTableName, fullIndexName);
+                fail();
+            } catch (AssertionError e) {
+                assertEquals(e.getMessage(),"Expected data table row count to match expected:<2> but was:<1>");
+            }
+        }
+    }
+    @Test
+    public void testExtraRowIndexScrutiny() throws Throwable {
+        String schemaName = generateUniqueName();
+        String tableName = generateUniqueName();
+        String indexName = generateUniqueName();
+        String fullTableName = SchemaUtil.getTableName(schemaName, tableName);
+        String fullIndexName = SchemaUtil.getTableName(schemaName, indexName);
+        try (Connection conn = DriverManager.getConnection(getUrl())) {
+            conn.createStatement().execute("CREATE TABLE " + fullTableName + "(k VARCHAR PRIMARY KEY, v VARCHAR, v2 VARCHAR) COLUMN_ENCODED_BYTES = 0, STORE_NULLS=true");
+            conn.createStatement().execute("CREATE INDEX " + indexName + " ON " + fullTableName + " (v) INCLUDE (v2)");
+            conn.createStatement().execute("UPSERT INTO " + fullTableName + " VALUES('b','bb','0')");
+            conn.createStatement().execute("UPSERT INTO " + fullTableName + " VALUES('a','ccc','1')");
+            conn.commit();
+            
+            conn.createStatement().executeUpdate("UPSERT INTO " + fullIndexName + " VALUES ('bbb','x','0')");
+            conn.commit();
+            try {
+                IndexScrutiny.scrutinizeIndex(conn, fullTableName, fullIndexName);
+                fail();
+            } catch (AssertionError e) {
+                assertEquals(e.getMessage(),"Expected to find PK in data table: ('x')");
+            }
+        }
+    }
+    
+    @Test
+    public void testValueIndexScrutiny() throws Throwable {
+        String schemaName = generateUniqueName();
+        String tableName = generateUniqueName();
+        String indexName = generateUniqueName();
+        String fullTableName = SchemaUtil.getTableName(schemaName, tableName);
+        String fullIndexName = SchemaUtil.getTableName(schemaName, indexName);
+        try (Connection conn = DriverManager.getConnection(getUrl())) {
+            conn.createStatement().execute("CREATE TABLE " + fullTableName + "(k VARCHAR PRIMARY KEY, v VARCHAR, v2 VARCHAR) COLUMN_ENCODED_BYTES = 0, STORE_NULLS=true");
+            conn.createStatement().execute("CREATE INDEX " + indexName + " ON " + fullTableName + " (v) INCLUDE (v2)");
+            conn.createStatement().execute("UPSERT INTO " + fullTableName + " VALUES('b','bb','0')");
+            conn.createStatement().execute("UPSERT INTO " + fullTableName + " VALUES('a','ccc','1')");
+            conn.commit();
+            
+            conn.createStatement().executeUpdate("UPSERT INTO " + fullIndexName + " VALUES ('ccc','a','2')");
+            conn.commit();
+            try {
+                IndexScrutiny.scrutinizeIndex(conn, fullTableName, fullIndexName);
+                fail();
+            } catch (AssertionError e) {
+                assertEquals(e.getMessage(),"Expected equality for V2, but '2'!='1'");
+            }
+        }
+    }
+
+
+}

http://git-wip-us.apache.org/repos/asf/phoenix/blob/572d242d/phoenix-core/src/it/java/org/apache/phoenix/util/TestUtilIT.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/it/java/org/apache/phoenix/util/TestUtilIT.java b/phoenix-core/src/it/java/org/apache/phoenix/util/TestUtilIT.java
deleted file mode 100644
index d26d9f6..0000000
--- a/phoenix-core/src/it/java/org/apache/phoenix/util/TestUtilIT.java
+++ /dev/null
@@ -1,89 +0,0 @@
-package org.apache.phoenix.util;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.fail;
-
-import java.sql.Connection;
-import java.sql.DriverManager;
-
-import org.apache.phoenix.end2end.ParallelStatsDisabledIT;
-import org.junit.Test;
-
-public class TestUtilIT extends ParallelStatsDisabledIT {
-    @Test
-    public void testRowCountIndexScrutiny() throws Throwable {
-        String schemaName = generateUniqueName();
-        String tableName = generateUniqueName();
-        String indexName = generateUniqueName();
-        String fullTableName = SchemaUtil.getTableName(schemaName, tableName);
-        String fullIndexName = SchemaUtil.getTableName(schemaName, indexName);
-        try (Connection conn = DriverManager.getConnection(getUrl())) {
-            conn.createStatement().execute("CREATE TABLE " + fullTableName + "(k VARCHAR PRIMARY KEY, v VARCHAR) COLUMN_ENCODED_BYTES = 0, STORE_NULLS=true");
-            conn.createStatement().execute("CREATE INDEX " + indexName + " ON " + fullTableName + " (v)");
-            conn.createStatement().execute("UPSERT INTO " + fullTableName + " VALUES('b','bb')");
-            conn.createStatement().execute("UPSERT INTO " + fullTableName + " VALUES('a','ccc')");
-            conn.commit();
-            
-            int count = conn.createStatement().executeUpdate("DELETE FROM " + fullIndexName + " WHERE \":K\"='a' AND \"0:V\"='ccc'");
-            assertEquals(1,count);
-            conn.commit();
-            try {
-                TestUtil.scrutinizeIndex(conn, fullTableName, fullIndexName);
-                fail();
-            } catch (AssertionError e) {
-                assertEquals(e.getMessage(),"Expected data table row count to match expected:<2> but was:<1>");
-            }
-        }
-    }
-    @Test
-    public void testExtraRowIndexScrutiny() throws Throwable {
-        String schemaName = generateUniqueName();
-        String tableName = generateUniqueName();
-        String indexName = generateUniqueName();
-        String fullTableName = SchemaUtil.getTableName(schemaName, tableName);
-        String fullIndexName = SchemaUtil.getTableName(schemaName, indexName);
-        try (Connection conn = DriverManager.getConnection(getUrl())) {
-            conn.createStatement().execute("CREATE TABLE " + fullTableName + "(k VARCHAR PRIMARY KEY, v VARCHAR, v2 VARCHAR) COLUMN_ENCODED_BYTES = 0, STORE_NULLS=true");
-            conn.createStatement().execute("CREATE INDEX " + indexName + " ON " + fullTableName + " (v) INCLUDE (v2)");
-            conn.createStatement().execute("UPSERT INTO " + fullTableName + " VALUES('b','bb','0')");
-            conn.createStatement().execute("UPSERT INTO " + fullTableName + " VALUES('a','ccc','1')");
-            conn.commit();
-            
-            conn.createStatement().executeUpdate("UPSERT INTO " + fullIndexName + " VALUES ('bbb','x','0')");
-            conn.commit();
-            try {
-                TestUtil.scrutinizeIndex(conn, fullTableName, fullIndexName);
-                fail();
-            } catch (AssertionError e) {
-                assertEquals(e.getMessage(),"Expected to find PK in data table: ('x')");
-            }
-        }
-    }
-    
-    @Test
-    public void testValueIndexScrutiny() throws Throwable {
-        String schemaName = generateUniqueName();
-        String tableName = generateUniqueName();
-        String indexName = generateUniqueName();
-        String fullTableName = SchemaUtil.getTableName(schemaName, tableName);
-        String fullIndexName = SchemaUtil.getTableName(schemaName, indexName);
-        try (Connection conn = DriverManager.getConnection(getUrl())) {
-            conn.createStatement().execute("CREATE TABLE " + fullTableName + "(k VARCHAR PRIMARY KEY, v VARCHAR, v2 VARCHAR) COLUMN_ENCODED_BYTES = 0, STORE_NULLS=true");
-            conn.createStatement().execute("CREATE INDEX " + indexName + " ON " + fullTableName + " (v) INCLUDE (v2)");
-            conn.createStatement().execute("UPSERT INTO " + fullTableName + " VALUES('b','bb','0')");
-            conn.createStatement().execute("UPSERT INTO " + fullTableName + " VALUES('a','ccc','1')");
-            conn.commit();
-            
-            conn.createStatement().executeUpdate("UPSERT INTO " + fullIndexName + " VALUES ('ccc','a','2')");
-            conn.commit();
-            try {
-                TestUtil.scrutinizeIndex(conn, fullTableName, fullIndexName);
-                fail();
-            } catch (AssertionError e) {
-                assertEquals(e.getMessage(),"Expected equality for V2, but '2'!='1'");
-            }
-        }
-    }
-
-
-}

http://git-wip-us.apache.org/repos/asf/phoenix/blob/572d242d/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java
index a07b5d0..298f9e9 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/coprocessor/UngroupedAggregateRegionObserver.java
@@ -52,6 +52,7 @@ import org.apache.hadoop.hbase.HTableDescriptor;
 import org.apache.hadoop.hbase.KeyValue;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.Delete;
+import org.apache.hadoop.hbase.client.Durability;
 import org.apache.hadoop.hbase.client.HTable;
 import org.apache.hadoop.hbase.client.Mutation;
 import org.apache.hadoop.hbase.client.Put;
@@ -897,6 +898,8 @@ public class UngroupedAggregateRegionObserver extends BaseScannerRegionObserver
                                     put.setAttribute(BaseScannerRegionObserver.IGNORE_NEWER_MUTATIONS,
                                             PDataType.TRUE_BYTES);
                                     mutations.add(put);
+                                    // Since we're replaying existing mutations, it makes no sense to write them to the wal
+                                    put.setDurability(Durability.SKIP_WAL);
                                 }
                                 put.add(cell);
                             } else {
@@ -907,6 +910,8 @@ public class UngroupedAggregateRegionObserver extends BaseScannerRegionObserver
                                     del.setAttribute(BaseScannerRegionObserver.IGNORE_NEWER_MUTATIONS,
                                             PDataType.TRUE_BYTES);
                                     mutations.add(del);
+                                    // Since we're replaying existing mutations, it makes no sense to write them to the wal
+                                    del.setDurability(Durability.SKIP_WAL);
                                 }
                                 del.addDeleteMarker(cell);
                             }

http://git-wip-us.apache.org/repos/asf/phoenix/blob/572d242d/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java
index d017bc1..02b7a0b 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/Indexer.java
@@ -123,6 +123,11 @@ public class Indexer extends BaseRegionObserver {
   protected IndexBuildManager builder;
   private LockManager lockManager;
 
+  // Hack to get around not being able to save any state between
+  // coprocessor calls. TODO: remove after HBASE-18127 when available
+  private ThreadLocal<Collection<Pair<Mutation, byte[]>>> indexUpdates =
+          new ThreadLocal<Collection<Pair<Mutation, byte[]>>>();
+  
   /** Configuration key for the {@link IndexBuilder} to use */
   public static final String INDEX_BUILDER_CONF_KEY = "index.builder";
 
@@ -483,9 +488,12 @@ public class Indexer extends BaseRegionObserver {
               }
     
               // Only copy mutations if we found duplicate rows
-              // (which is pretty much never for Phoenix)
+              // which only occurs when we're partially rebuilding
+              // the index (since we'll potentially have both a
+              // Put and a Delete mutation for the same row).
               if (copyMutations) {
                   // Add the mutation to the batch set
+
                   ImmutableBytesPtr row = new ImmutableBytesPtr(m.getRow());
                   MultiMutation stored = mutationsMap.get(row);
                   // we haven't seen this row before, so add it
@@ -508,6 +516,9 @@ public class Indexer extends BaseRegionObserver {
           miniBatchOp.setWalEdit(0, edit);
       }
   
+      if (copyMutations) {
+          mutations = IndexManagementUtil.flattenMutationsByTimestamp(mutations);
+      }
 
       // get the current span, or just use a null-span to avoid a bunch of if statements
       try (TraceScope scope = Trace.startSpan("Starting to build index updates")) {
@@ -531,42 +542,29 @@ public class Indexer extends BaseRegionObserver {
           metricSource.updateIndexPrepareTime(duration);
           current.addTimelineAnnotation("Built index updates, doing preStep");
           TracingUtils.addAnnotation(current, "index update count", indexUpdates.size());
-
-          // write them, either to WAL or the index tables
-          doPre(indexUpdates, edit, durability);
+          if (!indexUpdates.isEmpty()) {
+              setIndexUpdates(c, indexUpdates);
+              // write index updates to WAL
+              if (durability != Durability.SKIP_WAL) {
+                  // we have all the WAL durability, so we just update the WAL entry and move on
+                  for (Pair<Mutation, byte[]> entry : indexUpdates) {
+                    edit.add(new IndexedKeyValue(entry.getSecond(), entry.getFirst()));
+                  }              
+              }
+          }
       }
   }
 
-  /**
-   * Add the index updates to the WAL, or write to the index table, if the WAL has been disabled
-   * @return <tt>true</tt> if the WAL has been updated.
-   * @throws IOException
-   */
-  private boolean doPre(Collection<Pair<Mutation, byte[]>> indexUpdates, final WALEdit edit,
-      final Durability durability) throws IOException {
-    // no index updates, so we are done
-    if (indexUpdates == null || indexUpdates.size() == 0) {
-      return false;
-    }
-
-    // if writing to wal is disabled, we never see the WALEdit updates down the way, so do the index
-    // update right away
-    if (durability == Durability.SKIP_WAL) {
-      try {
-        this.writer.write(indexUpdates, false);
-        return false;
-      } catch (Throwable e) {
-        LOG.error("Failed to update index with entries:" + indexUpdates, e);
-        IndexManagementUtil.rethrowIndexingException(e);
-      }
-    }
-
-    // we have all the WAL durability, so we just update the WAL entry and move on
-    for (Pair<Mutation, byte[]> entry : indexUpdates) {
-      edit.add(new IndexedKeyValue(entry.getSecond(), entry.getFirst()));
-    }
-
-    return true;
+  private void setIndexUpdates(ObserverContext<RegionCoprocessorEnvironment> c, Collection<Pair<Mutation, byte[]>> indexUpdates) {
+      this.indexUpdates.set(indexUpdates);
+  }
+  
+  private Collection<Pair<Mutation, byte[]>> getIndexUpdates(ObserverContext<RegionCoprocessorEnvironment> c) {
+      return this.indexUpdates.get();
+  }
+  
+  private void removeIndexUpdates(ObserverContext<RegionCoprocessorEnvironment> c) {
+      this.indexUpdates.remove();
   }
 
   @Override
@@ -589,8 +587,7 @@ public class Indexer extends BaseRegionObserver {
           if (success) { // if miniBatchOp was successfully written, write index updates
               //each batch operation, only the first one will have anything useful, so we can just grab that
               Mutation mutation = miniBatchOp.getOperation(0);
-              WALEdit edit = miniBatchOp.getWalEdit(0);
-              doPost(edit, mutation, mutation.getDurability());
+              doPost(c, mutation);
           }
        } finally {
            long duration = EnvironmentEdgeManager.currentTimeMillis() - start;
@@ -604,22 +601,21 @@ public class Indexer extends BaseRegionObserver {
        }
   }
 
-  private void doPost(WALEdit edit, Mutation m, final Durability durability) throws IOException {
-    try {
-      doPostWithExceptions(edit, m, durability);
-      return;
-    } catch (Throwable e) {
-      rethrowIndexingException(e);
+  private void doPost(ObserverContext<RegionCoprocessorEnvironment> c, Mutation m) throws IOException {
+      try {
+        doPostWithExceptions(c,m);
+        return;
+      } catch (Throwable e) {
+        rethrowIndexingException(e);
+      }
+      throw new RuntimeException(
+          "Somehow didn't complete the index update, but didn't return succesfully either!");
     }
-    throw new RuntimeException(
-        "Somehow didn't complete the index update, but didn't return succesfully either!");
-  }
 
-  private void doPostWithExceptions(WALEdit edit, Mutation m, final Durability durability)
-          throws Exception {
+  private void doPostWithExceptions(ObserverContext<RegionCoprocessorEnvironment> c, Mutation m)
+          throws IOException {
       //short circuit, if we don't need to do any work
-      if (durability == Durability.SKIP_WAL || !this.builder.isEnabled(m) || edit == null) {
-          // already did the index update in prePut, so we are done
+      if (!this.builder.isEnabled(m)) {
           return;
       }
 
@@ -631,59 +627,38 @@ public class Indexer extends BaseRegionObserver {
           }
           long start = EnvironmentEdgeManager.currentTimeMillis();
 
-          // there is a little bit of excess here- we iterate all the non-indexed kvs for this check first
-          // and then do it again later when getting out the index updates. This should be pretty minor
-          // though, compared to the rest of the runtime
-          IndexedKeyValue ikv = getFirstIndexedKeyValue(edit);
-
-          /*
-           * early exit - we have nothing to write, so we don't need to do anything else. NOTE: we don't
-           * release the WAL Rolling lock (INDEX_UPDATE_LOCK) since we never take it in doPre if there are
-           * no index updates.
-           */
-          if (ikv == null) {
+          
+          Collection<Pair<Mutation, byte[]>> indexUpdates = getIndexUpdates(c);
+          if (indexUpdates == null) {
               return;
           }
 
-          /*
-           * only write the update if we haven't already seen this batch. We only want to write the batch
-           * once (this hook gets called with the same WALEdit for each Put/Delete in a batch, which can
-           * lead to writing all the index updates for each Put/Delete).
-           */
-          if (!ikv.getBatchFinished()) {
-              Collection<Pair<Mutation, byte[]>> indexUpdates = extractIndexUpdate(edit);
-
-              // the WAL edit is kept in memory and we already specified the factory when we created the
-              // references originally - therefore, we just pass in a null factory here and use the ones
-              // already specified on each reference
-              try {
-        		  current.addTimelineAnnotation("Actually doing index update for first time");
-                  Collection<Pair<Mutation, byte[]>> localUpdates =
-                          new ArrayList<Pair<Mutation, byte[]>>();
-                  Collection<Pair<Mutation, byte[]>> remoteUpdates =
-                          new ArrayList<Pair<Mutation, byte[]>>();
-        		  for (Pair<Mutation, byte[]> mutation : indexUpdates) {
-        			  if (Bytes.toString(mutation.getSecond()).equals(
-        					  environment.getRegion().getTableDesc().getNameAsString())) {
-        				  localUpdates.add(mutation);
-        			  } else {
-                          remoteUpdates.add(mutation);
-        			  }
-        		  }
-                  if(!remoteUpdates.isEmpty()) {
-                      writer.writeAndKillYourselfOnFailure(remoteUpdates, false);
-                  }
-                  if(!localUpdates.isEmpty()) {
-                      writer.writeAndKillYourselfOnFailure(localUpdates, true);
-                  }
-              } finally {                  // With a custom kill policy, we may throw instead of kill the server.
-                  // Without doing this in a finally block (at least with the mini cluster),
-                  // the region server never goes down.
-
-                  // mark the batch as having been written. In the single-update case, this never gets check
-                  // again, but in the batch case, we will check it again (see above).
-                  ikv.markBatchFinished();
+          // the WAL edit is kept in memory and we already specified the factory when we created the
+          // references originally - therefore, we just pass in a null factory here and use the ones
+          // already specified on each reference
+          try {
+    		  current.addTimelineAnnotation("Actually doing index update for first time");
+              Collection<Pair<Mutation, byte[]>> localUpdates =
+                      new ArrayList<Pair<Mutation, byte[]>>();
+              Collection<Pair<Mutation, byte[]>> remoteUpdates =
+                      new ArrayList<Pair<Mutation, byte[]>>();
+    		  for (Pair<Mutation, byte[]> mutation : indexUpdates) {
+    			  if (Bytes.toString(mutation.getSecond()).equals(
+    					  environment.getRegion().getTableDesc().getNameAsString())) {
+    				  localUpdates.add(mutation);
+    			  } else {
+                      remoteUpdates.add(mutation);
+    			  }
+    		  }
+              if(!remoteUpdates.isEmpty()) {
+                  writer.writeAndKillYourselfOnFailure(remoteUpdates, false);
+              }
+              if(!localUpdates.isEmpty()) {
+                  writer.writeAndKillYourselfOnFailure(localUpdates, true);
               }
+          } finally { // With a custom kill policy, we may throw instead of kill the server.
+              // mark the batch as having been written so it won't be written again
+              removeIndexUpdates(c);
           }
 
           long duration = EnvironmentEdgeManager.currentTimeMillis() - start;

http://git-wip-us.apache.org/repos/asf/phoenix/blob/572d242d/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/NonTxIndexBuilder.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/NonTxIndexBuilder.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/NonTxIndexBuilder.java
index 8d2bd83..c013b5b 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/NonTxIndexBuilder.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/NonTxIndexBuilder.java
@@ -12,24 +12,17 @@ package org.apache.phoenix.hbase.index.covered;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collection;
-import java.util.Collections;
-import java.util.Comparator;
-import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
 import java.util.Set;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.hbase.Cell;
-import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.KeyValue;
 import org.apache.hadoop.hbase.KeyValueUtil;
 import org.apache.hadoop.hbase.client.Delete;
 import org.apache.hadoop.hbase.client.Mutation;
 import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
-import org.apache.hadoop.hbase.util.Bytes;
-import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
 import org.apache.hadoop.hbase.util.Pair;
 import org.apache.phoenix.hbase.index.builder.BaseIndexBuilder;
 import org.apache.phoenix.hbase.index.covered.data.LocalHBaseState;
@@ -39,7 +32,6 @@ import org.apache.phoenix.hbase.index.covered.update.IndexUpdateManager;
 import org.apache.phoenix.hbase.index.covered.update.IndexedColumnGroup;
 
 import com.google.common.collect.Lists;
-import com.google.common.primitives.Longs;
 
 /**
  * Build covered indexes for phoenix updates.
@@ -94,76 +86,27 @@ public class NonTxIndexBuilder extends BaseIndexBuilder {
      * @throws IOException
      */
     private void batchMutationAndAddUpdates(IndexUpdateManager manager, LocalTableState state, Mutation m, IndexMetaData indexMetaData) throws IOException {
-        // split the mutation into timestamp-based batches
-        Collection<Batch> batches = createTimestampBatchesFromMutation(m);
-
-        // go through each batch of keyvalues and build separate index entries for each
-        boolean cleanupCurrentState = !indexMetaData.isImmutableRows();
-        for (Batch batch : batches) {
-            /*
-             * We have to split the work between the cleanup and the update for each group because when we update the
-             * current state of the row for the current batch (appending the mutations for the current batch) the next
-             * group will see that as the current state, which will can cause the a delete and a put to be created for
-             * the next group.
-             */
-            if (addMutationsForBatch(manager, batch, state, cleanupCurrentState, indexMetaData)) {
-                cleanupCurrentState = false;
-            }
-        }
-    }
-
-    /**
-     * Batch all the {@link KeyValue}s in a {@link Mutation} by timestamp. Updates any {@link KeyValue} with a timestamp
-     * == {@link HConstants#LATEST_TIMESTAMP} to the timestamp at the time the method is called.
-     * 
-     * @param m
-     *            {@link Mutation} from which to extract the {@link KeyValue}s
-     * @return the mutation, broken into batches and sorted in ascending order (smallest first)
-     */
-    protected Collection<Batch> createTimestampBatchesFromMutation(Mutation m) {
-        Map<Long, Batch> batches = new HashMap<Long, Batch>();
+        // The cells of a mutation are broken up into time stamp batches prior to this call (in Indexer).
+        long ts = m.getFamilyCellMap().values().iterator().next().iterator().next().getTimestamp();
+        Batch batch = new Batch(ts);
         for (List<Cell> family : m.getFamilyCellMap().values()) {
-            List<KeyValue> familyKVs = KeyValueUtil.ensureKeyValues(family);
-            createTimestampBatchesFromKeyValues(familyKVs, batches);
-        }
-        // sort the batches
-        List<Batch> sorted = new ArrayList<Batch>(batches.values());
-        Collections.sort(sorted, new Comparator<Batch>() {
-            @Override
-            public int compare(Batch o1, Batch o2) {
-                return Longs.compare(o1.getTimestamp(), o2.getTimestamp());
+            List<KeyValue> kvs = KeyValueUtil.ensureKeyValues(family);
+            for (KeyValue kv : kvs) {
+                batch.add(kv);
+                assert(ts == kv.getTimestamp());
             }
-        });
-        return sorted;
-    }
-
-    /**
-     * Batch all the {@link KeyValue}s in a collection of kvs by timestamp. Updates any {@link KeyValue} with a
-     * timestamp == {@link HConstants#LATEST_TIMESTAMP} to the timestamp at the time the method is called.
-     * 
-     * @param kvs
-     *            {@link KeyValue}s to break into batches
-     * @param batches
-     *            to update with the given kvs
-     */
-    protected void createTimestampBatchesFromKeyValues(Collection<KeyValue> kvs, Map<Long, Batch> batches) {
-        long now = EnvironmentEdgeManager.currentTime();
-        byte[] nowBytes = Bytes.toBytes(now);
+        }
 
-        // batch kvs by timestamp
-        for (KeyValue kv : kvs) {
-            long ts = kv.getTimestamp();
-            // override the timestamp to the current time, so the index and primary tables match
-            // all the keys with LATEST_TIMESTAMP will then be put into the same batch
-            if (kv.updateLatestStamp(nowBytes)) {
-                ts = now;
-            }
-            Batch batch = batches.get(ts);
-            if (batch == null) {
-                batch = new Batch(ts);
-                batches.put(ts, batch);
-            }
-            batch.add(kv);
+        // go through each batch of keyvalues and build separate index entries for each
+        boolean cleanupCurrentState = !indexMetaData.isImmutableRows();
+        /*
+         * We have to split the work between the cleanup and the update for each group because when we update the
+         * current state of the row for the current batch (appending the mutations for the current batch) the next
+         * group will see that as the current state, which will can cause the a delete and a put to be created for
+         * the next group.
+         */
+        if (addMutationsForBatch(manager, batch, state, cleanupCurrentState, indexMetaData)) {
+            cleanupCurrentState = false;
         }
     }
 

http://git-wip-us.apache.org/repos/asf/phoenix/blob/572d242d/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/example/CoveredColumnIndexer.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/example/CoveredColumnIndexer.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/example/CoveredColumnIndexer.java
index c830e54..925bcbb 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/example/CoveredColumnIndexer.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/covered/example/CoveredColumnIndexer.java
@@ -40,6 +40,7 @@ import org.apache.phoenix.hbase.index.covered.IndexMetaData;
 import org.apache.phoenix.hbase.index.covered.LocalTableState;
 import org.apache.phoenix.hbase.index.covered.NonTxIndexBuilder;
 import org.apache.phoenix.hbase.index.covered.update.IndexUpdateManager;
+import org.apache.phoenix.hbase.index.util.IndexManagementUtil;
 
 /**
  * Index maintainer that maintains multiple indexes based on '{@link ColumnGroup}s'. Each group is a
@@ -141,7 +142,7 @@ public class CoveredColumnIndexer extends NonTxIndexBuilder {
       }
 
       // do the usual thing as for deletes
-      Collection<Batch> timeBatch = createTimestampBatchesFromMutation(p);
+      Collection<Batch> timeBatch = IndexManagementUtil.createTimestampBatchesFromMutation(p);
       LocalTableState state = new LocalTableState(env, localTable, p);
       for (Batch entry : timeBatch) {
         //just set the timestamp on the table - it already has all the future state
@@ -158,7 +159,7 @@ public class CoveredColumnIndexer extends NonTxIndexBuilder {
    */
   private Collection<Batch>  batchByRow(Collection<KeyValue> filtered) {
     Map<Long, Batch> batches = new HashMap<Long, Batch>();
-    createTimestampBatchesFromKeyValues(filtered, batches);
+    IndexManagementUtil.createTimestampBatchesFromKeyValues(filtered, batches);
     return batches.values();
   }
 }
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/phoenix/blob/572d242d/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/util/IndexManagementUtil.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/util/IndexManagementUtil.java b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/util/IndexManagementUtil.java
index 6582c8a..697caef 100644
--- a/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/util/IndexManagementUtil.java
+++ b/phoenix-core/src/main/java/org/apache/phoenix/hbase/index/util/IndexManagementUtil.java
@@ -18,22 +18,39 @@
 package org.apache.phoenix.hbase.index.util;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.Cell;
+import org.apache.hadoop.hbase.CellUtil;
 import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.KeyValue.Type;
+import org.apache.hadoop.hbase.KeyValueUtil;
+import org.apache.hadoop.hbase.client.Delete;
+import org.apache.hadoop.hbase.client.Mutation;
+import org.apache.hadoop.hbase.client.Put;
 import org.apache.hadoop.hbase.client.Scan;
 import org.apache.hadoop.hbase.regionserver.wal.WALCellCodec;
 import org.apache.phoenix.hbase.index.ValueGetter;
 import org.apache.phoenix.hbase.index.builder.IndexBuildingFailureException;
+import org.apache.phoenix.hbase.index.covered.Batch;
 import org.apache.phoenix.hbase.index.covered.data.LazyValueGetter;
 import org.apache.phoenix.hbase.index.covered.update.ColumnReference;
 import org.apache.phoenix.hbase.index.scanner.ScannerBuilder.CoveredDeleteScanner;
 
+import com.google.common.collect.Lists;
+import com.google.common.primitives.Longs;
+
 /**
  * Utility class to help manage indexes
  */
@@ -189,4 +206,81 @@ public class IndexManagementUtil {
             conf.setInt(key, value);
         }
     }
+
+    /**
+     * Batch all the {@link KeyValue}s in a collection of kvs by timestamp. Updates any {@link KeyValue} with a
+     * timestamp == {@link HConstants#LATEST_TIMESTAMP} to the timestamp at the time the method is called.
+     * 
+     * @param kvs {@link KeyValue}s to break into batches
+     * @param batches to update with the given kvs
+     */
+    public static void createTimestampBatchesFromKeyValues(Collection<KeyValue> kvs, Map<Long, Batch> batches) {
+        // batch kvs by timestamp
+        for (KeyValue kv : kvs) {
+            long ts = kv.getTimestamp();
+            Batch batch = batches.get(ts);
+            if (batch == null) {
+                batch = new Batch(ts);
+                batches.put(ts, batch);
+            }
+            batch.add(kv);
+        }
+    }
+
+    /**
+     * Batch all the {@link KeyValue}s in a {@link Mutation} by timestamp. Updates any {@link KeyValue} with a timestamp
+     * == {@link HConstants#LATEST_TIMESTAMP} to the timestamp at the time the method is called.
+     * 
+     * @param m {@link Mutation} from which to extract the {@link KeyValue}s
+     * @return the mutation, broken into batches and sorted in ascending order (smallest first)
+     */
+    public static Collection<Batch> createTimestampBatchesFromMutation(Mutation m) {
+        Map<Long, Batch> batches = new HashMap<Long, Batch>();
+        for (List<Cell> family : m.getFamilyCellMap().values()) {
+            List<KeyValue> familyKVs = KeyValueUtil.ensureKeyValues(family);
+            createTimestampBatchesFromKeyValues(familyKVs, batches);
+        }
+        // sort the batches
+        List<Batch> sorted = new ArrayList<Batch>(batches.values());
+        Collections.sort(sorted, new Comparator<Batch>() {
+            @Override
+            public int compare(Batch o1, Batch o2) {
+                return Longs.compare(o1.getTimestamp(), o2.getTimestamp());
+            }
+        });
+        return sorted;
+    }
+
+    public static Collection<? extends Mutation> flattenMutationsByTimestamp(Collection<? extends Mutation> mutations) {
+          List<Mutation> flattenedMutations = Lists.newArrayListWithExpectedSize(mutations.size() * 10);
+          for (Mutation m : mutations) {
+              byte[] row = m.getRow();
+              Collection<Batch> batches = createTimestampBatchesFromMutation(m);
+              for (Batch batch : batches) {
+                  Mutation mWithSameTS;
+                  Cell firstCell = batch.getKvs().get(0);
+                  if (KeyValue.Type.codeToType(firstCell.getTypeByte()) == KeyValue.Type.Put) {
+                      mWithSameTS = new Put(row);
+                  } else {
+                      mWithSameTS = new Delete(row);
+                  }
+                  if (m.getAttributesMap() != null) {
+                      for (Map.Entry<String,byte[]> entry : m.getAttributesMap().entrySet()) {
+                          mWithSameTS.setAttribute(entry.getKey(), entry.getValue());
+                      }
+                  }
+                  for (Cell cell : batch.getKvs()) {
+                      byte[] fam = CellUtil.cloneFamily(cell);
+                      List<Cell> famCells = mWithSameTS.getFamilyCellMap().get(fam);
+                      if (famCells == null) {
+                          famCells = Lists.newArrayList();
+                          mWithSameTS.getFamilyCellMap().put(fam, famCells);
+                      }
+                      famCells.add(cell);
+                  }
+                  flattenedMutations.add(mWithSameTS);
+              }
+          }
+          return flattenedMutations;
+      }
 }
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/phoenix/blob/572d242d/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/covered/CoveredColumnsTest.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/covered/CoveredColumnsTest.java b/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/covered/CoveredColumnsTest.java
new file mode 100644
index 0000000..db7d838
--- /dev/null
+++ b/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/covered/CoveredColumnsTest.java
@@ -0,0 +1,46 @@
+/*
+ * 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.phoenix.hbase.index.covered;
+
+import static org.junit.Assert.assertEquals;
+
+import java.util.Arrays;
+
+import org.apache.hadoop.hbase.util.Bytes;
+import org.junit.Test;
+
+import org.apache.phoenix.hbase.index.covered.CoveredColumns;
+import org.apache.phoenix.hbase.index.covered.update.ColumnReference;
+
+public class CoveredColumnsTest {
+
+  private static final byte[] fam = Bytes.toBytes("fam");
+  private static final byte[] qual = Bytes.toBytes("qual");
+
+  @Test
+  public void testCovering() {
+    ColumnReference ref = new ColumnReference(fam, qual);
+    CoveredColumns columns = new CoveredColumns();
+    assertEquals("Should have only found a single column to cover", 1, columns
+        .findNonCoveredColumns(Arrays.asList(ref)).size());
+
+    columns.addColumn(ref);
+    assertEquals("Shouldn't have any columns to cover", 0,
+      columns.findNonCoveredColumns(Arrays.asList(ref)).size());
+  }
+}
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/phoenix/blob/572d242d/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/covered/LocalTableStateTest.java
----------------------------------------------------------------------
diff --git a/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/covered/LocalTableStateTest.java b/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/covered/LocalTableStateTest.java
new file mode 100644
index 0000000..dcf330f
--- /dev/null
+++ b/phoenix-core/src/test/java/org/apache/phoenix/hbase/index/covered/LocalTableStateTest.java
@@ -0,0 +1,293 @@
+/*
+ * 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.phoenix.hbase.index.covered;
+
+import static org.junit.Assert.assertEquals;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.KeyValue;
+import org.apache.hadoop.hbase.KeyValue.Type;
+import org.apache.hadoop.hbase.KeyValueUtil;
+import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hbase.client.Scan;
+import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
+import org.apache.hadoop.hbase.regionserver.Region;
+import org.apache.hadoop.hbase.regionserver.RegionScanner;
+import org.apache.hadoop.hbase.util.Bytes;
+import org.apache.hadoop.hbase.util.Pair;
+import org.apache.phoenix.hbase.index.covered.data.LocalHBaseState;
+import org.apache.phoenix.hbase.index.covered.data.LocalTable;
+import org.apache.phoenix.hbase.index.covered.update.ColumnReference;
+import org.apache.phoenix.hbase.index.scanner.Scanner;
+import org.apache.phoenix.hbase.index.scanner.ScannerBuilder.CoveredDeleteScanner;
+import org.junit.Test;
+import org.mockito.Mockito;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
+/**
+ *
+ */
+public class LocalTableStateTest {
+
+  private static final byte[] row = Bytes.toBytes("row");
+  private static final byte[] fam = Bytes.toBytes("fam");
+  private static final byte[] qual = Bytes.toBytes("qual");
+  private static final byte[] val = Bytes.toBytes("val");
+  private static final long ts = 10;
+  private static final IndexMetaData indexMetaData = new IndexMetaData() {
+
+    @Override
+    public boolean isImmutableRows() {
+        return false;
+    }
+
+    @Override
+    public boolean ignoreNewerMutations() {
+        return false;
+    }
+      
+  };
+
+  @SuppressWarnings("unchecked")
+  @Test
+  public void testCorrectOrderingWithLazyLoadingColumns() throws Exception {
+    Put m = new Put(row);
+    m.add(fam, qual, ts, val);
+    // setup mocks
+    Configuration conf = new Configuration(false);
+    RegionCoprocessorEnvironment env = Mockito.mock(RegionCoprocessorEnvironment.class);
+    Mockito.when(env.getConfiguration()).thenReturn(conf);
+
+    Region region = Mockito.mock(Region.class);
+    Mockito.when(env.getRegion()).thenReturn(region);
+    RegionScanner scanner = Mockito.mock(RegionScanner.class);
+    Mockito.when(region.getScanner(Mockito.any(Scan.class))).thenReturn(scanner);
+    final byte[] stored = Bytes.toBytes("stored-value");
+    Mockito.when(scanner.next(Mockito.any(List.class))).thenAnswer(new Answer<Boolean>() {
+      @Override
+      public Boolean answer(InvocationOnMock invocation) throws Throwable {
+        List<KeyValue> list = (List<KeyValue>) invocation.getArguments()[0];
+        KeyValue kv = new KeyValue(row, fam, qual, ts, Type.Put, stored);
+        kv.setSequenceId(0);
+        list.add(kv);
+        return false;
+      }
+    });
+
+
+    LocalHBaseState state = new LocalTable(env);
+    LocalTableState table = new LocalTableState(env, state, m);
+    //add the kvs from the mutation
+    table.addPendingUpdates(KeyValueUtil.ensureKeyValues(m.get(fam, qual)));
+
+    // setup the lookup
+    ColumnReference col = new ColumnReference(fam, qual);
+    table.setCurrentTimestamp(ts);
+    //check that our value still shows up first on scan, even though this is a lazy load
+    Pair<CoveredDeleteScanner, IndexUpdate> p = table.getIndexedColumnsTableState(Arrays.asList(col), false, false, indexMetaData);
+    Scanner s = p.getFirst();
+    assertEquals("Didn't get the pending mutation's value first", m.get(fam, qual).get(0), s.next());
+  }
+
+  public static final class ScannerCreatedException extends RuntimeException {
+      ScannerCreatedException(String msg) {
+          super(msg);
+      }
+  }
+  
+  @Test(expected = ScannerCreatedException.class)
+  public void testScannerForMutableRows() throws Exception {
+      IndexMetaData indexMetaData = new IndexMetaData() {
+
+          @Override
+          public boolean isImmutableRows() {
+              return false;
+          }
+
+          @Override
+          public boolean ignoreNewerMutations() {
+              return false;
+          }
+            
+        };
+    Put m = new Put(row);
+    m.add(fam, qual, ts, val);
+    // setup mocks
+    Configuration conf = new Configuration(false);
+    RegionCoprocessorEnvironment env = Mockito.mock(RegionCoprocessorEnvironment.class);
+    Mockito.when(env.getConfiguration()).thenReturn(conf);
+
+    Region region = Mockito.mock(Region.class);
+    Mockito.when(env.getRegion()).thenReturn(region);
+    Mockito.when(region.getScanner(Mockito.any(Scan.class))).thenThrow(new ScannerCreatedException("Should not open scanner when data is immutable"));
+
+    LocalHBaseState state = new LocalTable(env);
+    LocalTableState table = new LocalTableState(env, state, m);
+    //add the kvs from the mutation
+    table.addPendingUpdates(KeyValueUtil.ensureKeyValues(m.get(fam, qual)));
+
+    // setup the lookup
+    ColumnReference col = new ColumnReference(fam, qual);
+    table.setCurrentTimestamp(ts);
+    table.getIndexedColumnsTableState(Arrays.asList(col), false, false, indexMetaData);
+  }
+
+  @Test
+  public void testNoScannerForImmutableRows() throws Exception {
+      IndexMetaData indexMetaData = new IndexMetaData() {
+
+          @Override
+          public boolean isImmutableRows() {
+              return true;
+          }
+
+          @Override
+          public boolean ignoreNewerMutations() {
+              return false;
+          }
+            
+        };
+    Put m = new Put(row);
+    m.add(fam, qual, ts, val);
+    // setup mocks
+    Configuration conf = new Configuration(false);
+    RegionCoprocessorEnvironment env = Mockito.mock(RegionCoprocessorEnvironment.class);
+    Mockito.when(env.getConfiguration()).thenReturn(conf);
+
+    Region region = Mockito.mock(Region.class);
+    Mockito.when(env.getRegion()).thenReturn(region);
+    Mockito.when(region.getScanner(Mockito.any(Scan.class))).thenThrow(new ScannerCreatedException("Should not open scanner when data is immutable"));
+
+    LocalHBaseState state = new LocalTable(env);
+    LocalTableState table = new LocalTableState(env, state, m);
+    //add the kvs from the mutation
+    table.addPendingUpdates(KeyValueUtil.ensureKeyValues(m.get(fam, qual)));
+
+    // setup the lookup
+    ColumnReference col = new ColumnReference(fam, qual);
+    table.setCurrentTimestamp(ts);
+    //check that our value still shows up first on scan, even though this is a lazy load
+    Pair<CoveredDeleteScanner, IndexUpdate> p = table.getIndexedColumnsTableState(Arrays.asList(col), false, false, indexMetaData);
+    Scanner s = p.getFirst();
+    assertEquals("Didn't get the pending mutation's value first", m.get(fam, qual).get(0), s.next());
+  }
+
+  /**
+   * Test that we correctly rollback the state of keyvalue
+   * @throws Exception
+   */
+  @Test
+  @SuppressWarnings("unchecked")
+  public void testCorrectRollback() throws Exception {
+    Put m = new Put(row);
+    m.add(fam, qual, ts, val);
+    // setup mocks
+    RegionCoprocessorEnvironment env = Mockito.mock(RegionCoprocessorEnvironment.class);
+
+    Region region = Mockito.mock(Region.class);
+    Mockito.when(env.getRegion()).thenReturn(region);
+    RegionScanner scanner = Mockito.mock(RegionScanner.class);
+    Mockito.when(region.getScanner(Mockito.any(Scan.class))).thenReturn(scanner);
+    final byte[] stored = Bytes.toBytes("stored-value");
+    final KeyValue storedKv = new KeyValue(row, fam, qual, ts, Type.Put, stored);
+    storedKv.setSequenceId(2);
+    Mockito.when(scanner.next(Mockito.any(List.class))).thenAnswer(new Answer<Boolean>() {
+      @Override
+      public Boolean answer(InvocationOnMock invocation) throws Throwable {
+        List<KeyValue> list = (List<KeyValue>) invocation.getArguments()[0];
+
+        list.add(storedKv);
+        return false;
+      }
+    });
+    LocalHBaseState state = new LocalTable(env);
+    LocalTableState table = new LocalTableState(env, state, m);
+    // add the kvs from the mutation
+    KeyValue kv = KeyValueUtil.ensureKeyValue(m.get(fam, qual).get(0));
+    kv.setSequenceId(0);
+    table.addPendingUpdates(kv);
+
+    // setup the lookup
+    ColumnReference col = new ColumnReference(fam, qual);
+    table.setCurrentTimestamp(ts);
+    // check that the value is there
+    Pair<CoveredDeleteScanner, IndexUpdate> p = table.getIndexedColumnsTableState(Arrays.asList(col), false, false, indexMetaData);
+    Scanner s = p.getFirst();
+    assertEquals("Didn't get the pending mutation's value first", kv, s.next());
+
+    // rollback that value
+    table.rollback(Arrays.asList(kv));
+    p = table.getIndexedColumnsTableState(Arrays.asList(col), false, false, indexMetaData);
+    s = p.getFirst();
+    assertEquals("Didn't correctly rollback the row - still found it!", null, s.next());
+    Mockito.verify(env, Mockito.times(1)).getRegion();
+    Mockito.verify(region, Mockito.times(1)).getScanner(Mockito.any(Scan.class));
+  }
+
+  @SuppressWarnings("unchecked")
+  @Test
+  public void testOnlyLoadsRequestedColumns() throws Exception {
+    // setup mocks
+    RegionCoprocessorEnvironment env = Mockito.mock(RegionCoprocessorEnvironment.class);
+
+    Region region = Mockito.mock(Region.class);
+    Mockito.when(env.getRegion()).thenReturn(region);
+    RegionScanner scanner = Mockito.mock(RegionScanner.class);
+    Mockito.when(region.getScanner(Mockito.any(Scan.class))).thenReturn(scanner);
+    final KeyValue storedKv =
+        new KeyValue(row, fam, qual, ts, Type.Put, Bytes.toBytes("stored-value"));
+    storedKv.setSequenceId(2);
+    Mockito.when(scanner.next(Mockito.any(List.class))).thenAnswer(new Answer<Boolean>() {
+      @Override
+      public Boolean answer(InvocationOnMock invocation) throws Throwable {
+        List<KeyValue> list = (List<KeyValue>) invocation.getArguments()[0];
+
+        list.add(storedKv);
+        return false;
+      }
+    });
+    LocalHBaseState state = new LocalTable(env);
+    Put pendingUpdate = new Put(row);
+    pendingUpdate.add(fam, qual, ts, val);
+    LocalTableState table = new LocalTableState(env, state, pendingUpdate);
+
+    // do the lookup for the given column
+    ColumnReference col = new ColumnReference(fam, qual);
+    table.setCurrentTimestamp(ts);
+    // check that the value is there
+    Pair<CoveredDeleteScanner, IndexUpdate> p = table.getIndexedColumnsTableState(Arrays.asList(col), false, false, indexMetaData);
+    Scanner s = p.getFirst();
+    // make sure it read the table the one time
+    assertEquals("Didn't get the stored keyvalue!", storedKv, s.next());
+
+    // on the second lookup it shouldn't access the underlying table again - the cached columns
+    // should know they are done
+    p = table.getIndexedColumnsTableState(Arrays.asList(col), false, false, indexMetaData);
+    s = p.getFirst();
+    assertEquals("Lost already loaded update!", storedKv, s.next());
+    Mockito.verify(env, Mockito.times(1)).getRegion();
+    Mockito.verify(region, Mockito.times(1)).getScanner(Mockito.any(Scan.class));
+  }
+
+  // TODO add test here for making sure multiple column references with the same column family don't
+  // cause an infinite loop
+}


Mime
View raw message