Merge lp:~nwilliams/akiban-server/abstract-store into lp:~akiban-technologies/akiban-server/trunk

[Nathan Williams]

Pull general methods from PersistitStore into AbstractStore and cleanup PersistitException on interfaces.

Particularly, don't declare to throw anything on TableStatus or Store. Minor cascading cleanup that removes try/catch/wrap from higher level classes down into implementations.

Additionally, remove deferred indexing methods as they have been broken for a long time (bug767737) and are unused.

[Mike McMahon (mmcm)]

Looks as described.

review: Approve

[Thomas Jones-Low (tjoneslo)]

There were 2 failures during build/test:

* job server-build failed at build number 4030: http://172.16.20.104:8080/job/server-build/4030/

* view must-pass failed: server-build is yellow

review: Needs Fixing

[Thomas Jones-Low (tjoneslo)]

Build system failure, not merge proposal failure. Trying again.

[Thomas Jones-Low (tjoneslo)]

There were 2 failures during build/test:

* job server-build failed at build number 4032: http://172.16.20.104:8080/job/server-build/4032/

* view must-pass failed: server-build is yellow

review: Needs Fixing

[Nathan Williams (nwilliams)]

Manually ran ITs on sleepy.

=== modified file 'src/main/java/com/akiban/ais/model/UserTable.java'
--- src/main/java/com/akiban/ais/model/UserTable.java	2013-04-07 01:22:07 +0000
+++ src/main/java/com/akiban/ais/model/UserTable.java	2013-05-20 14:47:34 +0000
@@ -168,6 +168,10 @@
         return Collections.unmodifiableList(candidateChildJoins);
     }
 
+    public boolean hasChildren() {
+        return !getCandidateChildJoins().isEmpty();
+    }
+
     public Join getParentJoin()
     {
         Join parentJoin = null;

=== modified file 'src/main/java/com/akiban/qp/persistitadapter/OneTableRowCollector.java'
--- src/main/java/com/akiban/qp/persistitadapter/OneTableRowCollector.java	2013-03-22 20:05:57 +0000
+++ src/main/java/com/akiban/qp/persistitadapter/OneTableRowCollector.java	2013-05-20 14:47:34 +0000
@@ -26,15 +26,13 @@
 import com.akiban.server.api.dml.scan.NewRow;
 import com.akiban.server.rowdata.RowData;
 import com.akiban.server.rowdata.RowDef;
-import com.akiban.server.service.config.ConfigurationService;
 import com.akiban.server.service.session.Session;
-import com.akiban.server.store.PersistitStore;
+import com.akiban.server.store.Store;
 
 public class OneTableRowCollector extends OperatorBasedRowCollector
 {
-    OneTableRowCollector(ConfigurationService config,
-                         Session session,
-                         PersistitStore store,
+    OneTableRowCollector(Session session,
+                         Store store,
                          RowDef rowDef,
                          int indexId,
                          int scanFlags,
@@ -43,7 +41,7 @@
                          RowData end,
                          ColumnSelector endColumns)
     {
-        super(store, session, config);
+        super(store, session);
         // rootmostQueryTable
         queryRootTable = rowDef.userTable();
         queryRootType = schema.userTableRowType(queryRootTable);

=== modified file 'src/main/java/com/akiban/qp/persistitadapter/OperatorBasedRowCollector.java'
--- src/main/java/com/akiban/qp/persistitadapter/OperatorBasedRowCollector.java	2013-04-30 22:35:43 +0000
+++ src/main/java/com/akiban/qp/persistitadapter/OperatorBasedRowCollector.java	2013-05-20 14:47:34 +0000
@@ -23,6 +23,8 @@
 import com.akiban.qp.operator.Limit;
 import com.akiban.qp.operator.Operator;
 import com.akiban.qp.operator.SimpleQueryContext;
+import com.akiban.qp.operator.StoreAdapter;
+import com.akiban.qp.row.AbstractRow;
 import com.akiban.qp.row.Row;
 import com.akiban.qp.rowtype.*;
 import com.akiban.qp.rowtype.Schema;
@@ -38,6 +40,7 @@
 import com.akiban.server.service.session.Session;
 import com.akiban.server.store.PersistitStore;
 import com.akiban.server.store.RowCollector;
+import com.akiban.server.store.Store;
 import com.akiban.server.types3.Types3Switch;
 import com.akiban.util.GrowableByteBuffer;
 import com.akiban.util.ShareHolder;
@@ -78,12 +81,12 @@
          // the current row across these two invocations.
         boolean wasHeld = false;
         boolean wroteToPayload = false;
-        PersistitGroupRow row;
+        AbstractRow row;
         if (currentRow.isEmpty()) {
-            row = (PersistitGroupRow) cursor.next();
+            row = (AbstractRow) cursor.next();
         } else {
             wasHeld = true;
-            row = (PersistitGroupRow) currentRow.get();
+            row = (AbstractRow) currentRow.get();
             currentRow.release();
         }
         if (row == null) {
@@ -117,7 +120,7 @@
     public RowData collectNextRow()
     {
         RowData rowData = null;
-        PersistitGroupRow row = (PersistitGroupRow) cursor.next();
+        AbstractRow row = (AbstractRow) cursor.next();
         if (row == null) {
             close();
         } else {
@@ -196,9 +199,8 @@
 
     // OperatorBasedRowCollector interface
 
-    public static OperatorBasedRowCollector newCollector(ConfigurationService config,
-                                                         Session session,
-                                                         PersistitStore store,
+    public static OperatorBasedRowCollector newCollector(Session session,
+                                                         Store store,
                                                          int scanFlags,
                                                          RowDef rowDef,
                                                          int indexId,
@@ -217,8 +219,7 @@
             throw new IllegalArgumentException("Must scan a UserTable: " + rowDef);
         }
         OperatorBasedRowCollector rowCollector =
-              new OneTableRowCollector(config,
-                                       session,
+              new OneTableRowCollector(session,
                                        store,
                                        rowDef,
                                        indexId,
@@ -234,12 +235,10 @@
         return rowCollector;
     }
     
-    protected OperatorBasedRowCollector(PersistitStore store, Session session, ConfigurationService config)
+    protected OperatorBasedRowCollector(Store store, Session session)
     {
         this.schema = SchemaCache.globalSchema(store.getAIS(session));
-        // Passing null to PersistitAdapter's TreeService argument. TreeService is only needed for sorting,
-        // which OBRC doesn't use.
-        this.adapter = new PersistitAdapter(schema, store, null, session, config);
+        this.adapter = store.createAdapter(session, schema);
         this.rowCollectorId = idCounter.getAndIncrement();
     }
 
@@ -397,7 +396,7 @@
 
     private long rowCollectorId;
     protected final Schema schema;
-    protected PersistitAdapter adapter;
+    protected StoreAdapter adapter;
     protected UserTable queryRootTable;
     protected UserTableRowType queryRootType;
     protected TableIndex predicateIndex;

=== modified file 'src/main/java/com/akiban/qp/persistitadapter/OperatorStore.java'
--- src/main/java/com/akiban/qp/persistitadapter/OperatorStore.java	2013-04-30 22:35:43 +0000
+++ src/main/java/com/akiban/qp/persistitadapter/OperatorStore.java	2013-05-20 14:47:34 +0000
@@ -48,7 +48,6 @@
 import com.akiban.server.store.SchemaManager;
 import com.akiban.server.types.ToObjectValueTarget;
 import com.akiban.server.types.ValueSource;
-import com.akiban.server.types3.Types3Switch;
 import com.akiban.sql.optimizer.rule.PlanGenerator;
 import com.akiban.util.tap.InOutTap;
 import com.akiban.util.tap.PointTap;
@@ -86,7 +85,6 @@
 
     @Override
     public void updateRow(Session session, RowData oldRowData, RowData newRowData, ColumnSelector columnSelector, Index[] indexes)
-        throws PersistitException
     {
         if(indexes != null) {
             throw new IllegalStateException("Unexpected indexes: " + Arrays.toString(indexes));
@@ -169,7 +167,7 @@
     }
 
     @Override
-    public void writeRow(Session session, RowData rowData) throws PersistitException {
+    public void writeRow(Session session, RowData rowData) {
         INSERT_TOTAL.in();
         INSERT_MAINTENANCE.in();
         try {
@@ -194,7 +192,7 @@
     }
 
     @Override
-    public void deleteRow(Session session, RowData rowData, boolean deleteIndexes, boolean cascadeDelete) throws PersistitException {
+    public void deleteRow(Session session, RowData rowData, boolean deleteIndexes, boolean cascadeDelete) {
         DELETE_TOTAL.in();
         DELETE_MAINTENANCE.in();
         try {
@@ -288,14 +286,14 @@
             BitSet columnDifferences,
             OperatorStoreGIHandler handler,
             OperatorStoreGIHandler.Action action)
-    throws PersistitException
     {
         UserTable userTable = ais.getUserTable(rowData.getRowDefId());
         if(canSkipMaintenance(userTable)) {
             return;
         }
-        Exchange hEx = adapter.takeExchange(userTable.getGroup());
+        Exchange hEx = null;
         try {
+            hEx = adapter.takeExchange(userTable.getGroup());
             // the "false" at the end of constructHKey toggles whether the RowData should be modified to increment
             // the hidden PK field, if there is one. For PK-less rows, this field have already been incremented by now,
             // so we don't want to increment it again
@@ -316,8 +314,12 @@
                     SKIP_MAINTENANCE.hit();
                 }
             }
+        } catch(PersistitException e) {
+            throw PersistitAdapter.wrapPersistitException(session, e);
         } finally {
-            adapter.returnExchange(hEx);
+            if(hEx != null) {
+                adapter.returnExchange(hEx);
+            }
         }
     }
 
@@ -357,7 +359,6 @@
             AkibanInformationSchema ais, 
             PersistitAdapter adapter, 
             RowData rowData)
-    throws PersistitException
     {
         UserTable uTable = ais.getUserTable(rowData.getRowDefId());
 

=== modified file 'src/main/java/com/akiban/qp/persistitadapter/PersistitAdapter.java'
--- src/main/java/com/akiban/qp/persistitadapter/PersistitAdapter.java	2013-04-11 05:51:16 +0000
+++ src/main/java/com/akiban/qp/persistitadapter/PersistitAdapter.java	2013-05-20 14:47:34 +0000
@@ -129,10 +129,6 @@
         } catch (InvalidOperationException e) {
             rollbackIfNeeded(e);
             throw e;
-        } catch (PersistitException e) {
-            rollbackIfNeeded(e);
-            handlePersistitException(e);
-            assert false;
         }
         finally {
             leaveUpdateStep(oldStep);
@@ -149,10 +145,6 @@
         } catch (InvalidOperationException e) {
             rollbackIfNeeded(e);
             throw e;
-        } catch (PersistitException e) {
-            rollbackIfNeeded(e);
-            handlePersistitException(e);
-            assert false;
         }
         finally {
             leaveUpdateStep(oldStep);
@@ -170,10 +162,6 @@
         } catch (InvalidOperationException e) {
             rollbackIfNeeded(e);
             throw e;
-        } catch (PersistitException e) {
-            rollbackIfNeeded(e);
-            handlePersistitException(e);
-            assert false;
         }
         finally {
             leaveUpdateStep(oldStep);
@@ -203,10 +191,6 @@
         } catch (InvalidOperationException e) {
             rollbackIfNeeded(e);
             throw e;
-        } catch (PersistitException e) {
-            rollbackIfNeeded(e);
-            handlePersistitException(e);
-            assert false;
         }
         finally {
             leaveUpdateStep(oldStep);
@@ -216,11 +200,7 @@
     @Override
     public long rowCount(RowType tableType) {
         RowDef rowDef = tableType.userTable().rowDef();
-        try {
-            return rowDef.getTableStatus().getRowCount();
-        } catch(PersistitInterruptedException e) {
-            throw new QueryCanceledException(getSession());
-        }
+        return rowDef.getTableStatus().getRowCount();
     }
 
     @Override
@@ -235,7 +215,7 @@
             key = persistitKeyValueSource.key();
             depth = persistitKeyValueSource.depth();
         } else {
-            key = persistit.getKey();
+            key = persistit.createKey();
             collator.append(key, valueSource.getString());
             depth = 0;
         }
@@ -325,7 +305,10 @@
     public static boolean isFromInterruption(Exception e) {
         Throwable cause = e.getCause();
         return (e instanceof PersistitInterruptedException) ||
-               ((cause != null) && (cause instanceof InterruptedIOException || cause instanceof InterruptedException));
+               ((cause != null) &&
+                (cause instanceof PersistitInterruptedException ||
+                 cause instanceof InterruptedIOException ||
+                 cause instanceof InterruptedException));
     }
 
     public static RuntimeException wrapPersistitException(Session session, PersistitException e)
@@ -418,7 +401,10 @@
 
     // For use by this class
     private void rollbackIfNeeded(Exception e) {
-        if((e instanceof DuplicateKeyException) || (e instanceof PersistitException) || isFromInterruption(e)) {
+        if((e instanceof DuplicateKeyException) ||
+           (e instanceof PersistitException) ||
+           (e instanceof PersistitAdapterException) ||
+           isFromInterruption(e)) {
             Transaction txn = transaction();
             if(txn.isActive()) {
                 txn.rollback();

=== modified file 'src/main/java/com/akiban/qp/persistitadapter/PersistitGroupRow.java'
--- src/main/java/com/akiban/qp/persistitadapter/PersistitGroupRow.java	2013-03-22 20:05:57 +0000
+++ src/main/java/com/akiban/qp/persistitadapter/PersistitGroupRow.java	2013-05-20 14:47:34 +0000
@@ -159,6 +159,7 @@
         } while (exception != null);
     }
 
+    @Override
     public RowData rowData()
     {
         return rowData;

=== modified file 'src/main/java/com/akiban/qp/persistitadapter/indexrow/PersistitIndexRow.java'
--- src/main/java/com/akiban/qp/persistitadapter/indexrow/PersistitIndexRow.java	2013-03-22 20:05:57 +0000
+++ src/main/java/com/akiban/qp/persistitadapter/indexrow/PersistitIndexRow.java	2013-05-20 14:47:34 +0000
@@ -128,7 +128,7 @@
     protected PersistitIndexRow(PersistitAdapter adapter, IndexRowType indexRowType)
     {
         super(adapter);
-        this.keyState = adapter.persistit().getKey();
+        this.keyState = adapter.persistit().createKey();
         resetForWrite(indexRowType.index(), keyState);
         this.indexRowType = indexRowType;
         this.leafmostTable = (UserTable) index.leafMostTable();

=== modified file 'src/main/java/com/akiban/qp/row/AbstractRow.java'
--- src/main/java/com/akiban/qp/row/AbstractRow.java	2013-03-22 20:05:57 +0000
+++ src/main/java/com/akiban/qp/row/AbstractRow.java	2013-05-20 14:47:34 +0000
@@ -22,6 +22,7 @@
 import com.akiban.qp.rowtype.IndexRowType;
 import com.akiban.qp.rowtype.RowType;
 import com.akiban.server.Quote;
+import com.akiban.server.rowdata.RowData;
 import com.akiban.server.types.ValueSource;
 import com.akiban.server.types.util.ValueSources;
 import com.akiban.server.types3.TClass;
@@ -166,6 +167,10 @@
         return builder.toString();
     }
 
+    public RowData rowData() {
+        throw new UnsupportedOperationException();
+    }
+
     // for use by subclasses
     protected void afterRelease() {}
     protected void beforeAcquire() {}

=== modified file 'src/main/java/com/akiban/server/PersistitAccumulatorTableStatusCache.java'
--- src/main/java/com/akiban/server/PersistitAccumulatorTableStatusCache.java	2013-04-11 06:15:45 +0000
+++ src/main/java/com/akiban/server/PersistitAccumulatorTableStatusCache.java	2013-05-20 14:47:34 +0000
@@ -18,6 +18,7 @@
 package com.akiban.server;
 
 import com.akiban.qp.memoryadapter.MemoryTableFactory;
+import com.akiban.qp.persistitadapter.PersistitAdapter;
 import com.akiban.server.error.PersistitAdapterException;
 import com.akiban.server.rowdata.IndexDef;
 import com.akiban.server.rowdata.RowDef;
@@ -97,18 +98,30 @@
         }
 
         @Override
-        public long getAutoIncrement() throws PersistitInterruptedException {
-            return autoIncrement.getSnapshot();
-        }
-
-        @Override
-        public int getOrdinal() throws PersistitInterruptedException {
-            return (int) ordinal.getSnapshot();
-        }
-
-        @Override
-        public long getRowCount() throws PersistitInterruptedException {
-            return rowCount.getSnapshot();
+        public long getAutoIncrement() {
+            try {
+                return autoIncrement.getSnapshot();
+            } catch(PersistitInterruptedException e) {
+                throw PersistitAdapter.wrapPersistitException(null, e);
+            }
+        }
+
+        @Override
+        public int getOrdinal() {
+            try {
+                return (int) ordinal.getSnapshot();
+            } catch(PersistitInterruptedException e) {
+                throw PersistitAdapter.wrapPersistitException(null, e);
+            }
+        }
+
+        @Override
+        public long getRowCount() {
+            try {
+                return rowCount.getSnapshot();
+            } catch(PersistitInterruptedException e) {
+                throw PersistitAdapter.wrapPersistitException(null, e);
+            }
         }
 
         @Override
@@ -127,8 +140,12 @@
         }
 
         @Override
-        public long getUniqueID() throws PersistitInterruptedException {
-            return uniqueID.getSnapshot();
+        public long getUniqueID() {
+            try {
+                return uniqueID.getSnapshot();
+            } catch(PersistitInterruptedException e) {
+                throw PersistitAdapter.wrapPersistitException(null, e);
+            }
         }
 
         @Override
@@ -151,23 +168,31 @@
             rowCount.sumAdd(count);
         }
 
-        public void setOrdinal(int ordinal) throws PersistitInterruptedException {
-            this.ordinal.set(ordinal);
+        public void setOrdinal(int ordinal) {
+            try {
+                this.ordinal.set(ordinal);
+            } catch(PersistitInterruptedException e) {
+                throw PersistitAdapter.wrapPersistitException(null, e);
+            }
         }
 
         @Override
-        public long createNewUniqueID() throws PersistitInterruptedException {
+        public long createNewUniqueID() {
             return uniqueID.seqAllocate();
         }
 
         @Override
-        public void truncate() throws PersistitInterruptedException {
-            internalSetRowCount(0);
-            internalSetAutoIncrement(0, true);
+        public void truncate() {
+            try {
+                internalSetRowCount(0);
+                internalSetAutoIncrement(0, true);
+            } catch(PersistitInterruptedException e) {
+                throw PersistitAdapter.wrapPersistitException(null, e);
+            }
         }
 
         @Override
-        public void setAutoIncrement(long value) throws PersistitInterruptedException {
+        public void setAutoIncrement(long value) {
             internalSetAutoIncrement(value, false);
         }
 
@@ -189,8 +214,12 @@
             rowCount.set(rowCountValue);
         }
 
-        private void internalSetAutoIncrement(long autoIncrementValue, boolean evenIfLess) throws PersistitInterruptedException {
-            autoIncrement.set(autoIncrementValue, evenIfLess);
+        private void internalSetAutoIncrement(long autoIncrementValue, boolean evenIfLess) {
+            try {
+                autoIncrement.set(autoIncrementValue, evenIfLess);
+            } catch(PersistitInterruptedException e) {
+                throw PersistitAdapter.wrapPersistitException(null, e);
+            }
         }
     }
 

=== modified file 'src/main/java/com/akiban/server/TableStatus.java'
--- src/main/java/com/akiban/server/TableStatus.java	2013-04-11 06:15:45 +0000
+++ src/main/java/com/akiban/server/TableStatus.java	2013-05-20 14:47:34 +0000
@@ -34,34 +34,34 @@
     void rowsWritten(long count);
 
     /** Reset, but do not remove, the state of a table. */
-    void truncate() throws PersistitInterruptedException;
+    void truncate();
 
     /** Set the auto-increment value of a given table. */
-    void setAutoIncrement(long value) throws PersistitInterruptedException;
+    void setAutoIncrement(long value);
 
     /** Set the RowDef of a given table.*/
     void setRowDef(RowDef rowDef);
 
     /** Create a brand new, unique ID for the given table. */
-    long createNewUniqueID() throws PersistitInterruptedException;
+    long createNewUniqueID();
 
     /** Set the ordinal value of a given table. */
-    void setOrdinal(int value) throws PersistitInterruptedException;
+    void setOrdinal(int value);
 
     /**
      * @return Current auto-increment value of the assocated table.
      */
-    long getAutoIncrement() throws PersistitInterruptedException;
+    long getAutoIncrement();
 
     /**
      * @return Ordinal of the associated table.
      */
-    int getOrdinal() throws PersistitInterruptedException;
+    int getOrdinal();
 
     /**
      * @return Current number of rows in the associated table.
      */
-    long getRowCount() throws PersistitInterruptedException;
+    long getRowCount();
 
     /**
      * @return Approximate number of rows in the associated table.
@@ -71,7 +71,7 @@
     /**
      * @return The <b>last</b> unique value used for the associated table.
      */
-    long getUniqueID() throws PersistitInterruptedException;
+    long getUniqueID();
 
     /** @return The table ID this status is for */
     int getTableID();

=== modified file 'src/main/java/com/akiban/server/manage/ManageMXBean.java'
--- src/main/java/com/akiban/server/manage/ManageMXBean.java	2013-03-22 20:05:57 +0000
+++ src/main/java/com/akiban/server/manage/ManageMXBean.java	2013-05-20 14:47:34 +0000
@@ -25,15 +25,9 @@
 
     int getJmxPort();
 
-    boolean isDeferIndexesEnabled();
-
-    void setDeferIndexes(final boolean defer);
-
     void buildIndexes(final String arg, final boolean deferIndexes);
 
     void deleteIndexes(final String arg);
 
-    void flushIndexes();
-
     String getVersionString();
 }

=== modified file 'src/main/java/com/akiban/server/manage/ManageMXBeanImpl.java'
--- src/main/java/com/akiban/server/manage/ManageMXBeanImpl.java	2013-03-22 20:05:57 +0000
+++ src/main/java/com/akiban/server/manage/ManageMXBeanImpl.java	2013-05-20 14:47:34 +0000
@@ -52,16 +52,6 @@
     }
 
     @Override
-    public boolean isDeferIndexesEnabled() {
-        return getStore().isDeferIndexes();
-    }
-
-    @Override
-    public void setDeferIndexes(final boolean defer) {
-        getStore().setDeferIndexes(defer);
-    }
-
-    @Override
     public void buildIndexes(final String arg, final boolean deferIndexes) {
         Session session = createSession();
         try {
@@ -88,18 +78,6 @@
     }
 
     @Override
-    public void flushIndexes() {
-        Session session = createSession();
-        try {
-            getStore().flushIndexes(session);
-        } catch(Exception t) {
-            throw new RuntimeException(t);
-        } finally {
-            session.close();
-        }
-    }
-
-    @Override
     public String getVersionString() {
         return AkServer.VERSION_STRING;
     }

=== modified file 'src/main/java/com/akiban/server/service/dxl/BasicDDLFunctions.java'
--- src/main/java/com/akiban/server/service/dxl/BasicDDLFunctions.java	2013-04-22 02:00:55 +0000
+++ src/main/java/com/akiban/server/service/dxl/BasicDDLFunctions.java	2013-05-20 14:47:34 +0000
@@ -1161,7 +1161,7 @@
             if (expected != actual) {
                 PersistitStore pStore = this.store().getPersistitStore();
                 if (index.isTableIndex()) {
-                    pStore.getTableStatus(((TableIndex) index).getTable()).setRowCount(actual);
+                    index.leafMostTable().rowDef().getTableStatus().setRowCount(actual);
                 }
                 else {
                     final Exchange ex = pStore.getExchange(session, index);

=== modified file 'src/main/java/com/akiban/server/service/dxl/BasicDMLFunctions.java'
--- src/main/java/com/akiban/server/service/dxl/BasicDMLFunctions.java	2013-03-22 20:05:57 +0000
+++ src/main/java/com/akiban/server/service/dxl/BasicDMLFunctions.java	2013-05-20 14:47:34 +0000
@@ -581,22 +581,14 @@
     {
         logger.trace("writing a row");
         final RowData rowData = niceRowToRowData(row);
-        try {
-            store().writeRow(session, rowData);
-        } catch (PersistitException ex) {
-            throw new PersistitAdapterException(ex);
-        }
+        store().writeRow(session, rowData);
     }
 
     @Override
     public void writeRows(Session session, List<RowData> rows) {
         logger.trace("writing {} rows", rows.size());
-        try {
-            for(RowData rowData : rows) {
-                store().writeRow(session, rowData);
-            }
-        } catch (PersistitException ex) {
-            throw new PersistitAdapterException(ex);
+        for(RowData rowData : rows) {
+            store().writeRow(session, rowData);
         }
     }
 
@@ -605,11 +597,7 @@
     {
         logger.trace("deleting a row (cascade: {})", cascadeDelete);
         final RowData rowData = niceRowToRowData(row);
-        try {
-            store().deleteRow(session, rowData, true, cascadeDelete);
-        } catch (PersistitException ex) {
-            throw new PersistitAdapterException(ex);
-        }
+        store().deleteRow(session, rowData, true, cascadeDelete);
     }
 
     private RowData niceRowToRowData(NewRow row) 
@@ -636,12 +624,7 @@
                 tableId
         );
 
-        try {
-            store().updateRow(session, oldData, newData, columnSelector, null);
-        } catch (PersistitException ex) {
-            throw new PersistitAdapterException(ex);
-        }
-            
+        store().updateRow(session, oldData, newData, columnSelector, null);
     }
 
     private void checkForModifiedCursors(
@@ -749,12 +732,7 @@
         final UserTable utable = table.isUserTable() ? (UserTable)table : null;
 
         if(utable == null || canFastTruncate(session, utable)) {
-            try {
-                store().truncateGroup(session, table.getGroup());
-            } catch (PersistitException ex) {
-                throw new PersistitAdapterException(ex);
-            }
-                
+            store().truncateGroup(session, table.getGroup());
             return;
         }
 
@@ -810,11 +788,7 @@
                 thrown = e;
             }
         }
-        try {
-            store().truncateTableStatus(session, tableId);
-        } catch (PersistitException ex) {
-            throw new PersistitAdapterException(ex);
-        }
+        store().truncateTableStatus(session, tableId);
         if (thrown != null) {
             throw thrown;
         }

=== added file 'src/main/java/com/akiban/server/store/AbstractStore.java'
--- src/main/java/com/akiban/server/store/AbstractStore.java	1970-01-01 00:00:00 +0000
+++ src/main/java/com/akiban/server/store/AbstractStore.java	2013-05-20 14:47:34 +0000
@@ -0,0 +1,487 @@
+/**
+ * Copyright (C) 2009-2013 Akiban Technologies, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package com.akiban.server.store;
+
+import com.akiban.ais.model.Group;
+import com.akiban.ais.model.Index;
+import com.akiban.ais.model.NopVisitor;
+import com.akiban.ais.model.Table;
+import com.akiban.ais.model.TableName;
+import com.akiban.ais.model.UserTable;
+import com.akiban.qp.persistitadapter.OperatorBasedRowCollector;
+import com.akiban.server.TableStatistics;
+import com.akiban.server.TableStatus;
+import com.akiban.server.api.dml.ColumnSelector;
+import com.akiban.server.api.dml.scan.ScanLimit;
+import com.akiban.server.error.CursorCloseBadException;
+import com.akiban.server.error.CursorIsUnknownException;
+import com.akiban.server.error.NoSuchTableException;
+import com.akiban.server.error.RowDefNotFoundException;
+import com.akiban.server.rowdata.IndexDef;
+import com.akiban.server.rowdata.RowData;
+import com.akiban.server.rowdata.RowDef;
+import com.akiban.server.service.session.Session;
+import com.akiban.server.service.tree.TreeLink;
+import com.akiban.server.store.statistics.Histogram;
+import com.akiban.server.store.statistics.HistogramEntry;
+import com.akiban.server.store.statistics.IndexStatistics;
+import com.akiban.server.store.statistics.IndexStatisticsService;
+import com.akiban.util.tap.InOutTap;
+import com.akiban.util.tap.Tap;
+import com.persistit.Key;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.BitSet;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.List;
+
+public abstract class AbstractStore implements Store {
+    private static final Logger LOG = LoggerFactory.getLogger(AbstractStore.class.getName());
+    private static final InOutTap NEW_COLLECTOR_TAP = Tap.createTimer("read: new_collector");
+    private static final Session.MapKey<Integer,List<RowCollector>> COLLECTORS = Session.MapKey.mapNamed("collectors");
+
+    protected IndexStatisticsService indexStatisticsService;
+
+    //
+    // AbstractStore
+    //
+
+    protected RowDef rowDefFromExplicitOrId(Session session, RowData rowData) {
+        RowDef rowDef = rowData.getExplicitRowDef();
+        if(rowDef == null) {
+            rowDef = getRowDef(session, rowData.getRowDefId());
+        }
+        return rowDef;
+    }
+
+    protected boolean hasNullIndexSegments(RowData rowData, Index index)
+    {
+        IndexDef indexDef = index.indexDef();
+        assert indexDef.getRowDef().getRowDefId() == rowData.getRowDefId();
+        for (int i : indexDef.getFields()) {
+            if (rowData.isNull(i)) {
+                return true;
+            }
+        }
+        return false;
+    }
+
+    /** Convert from new-format histogram to old for adapter. */
+    protected TableStatistics.Histogram indexStatisticsToHistogram(Session session, Index index, Key key) {
+        IndexStatistics stats = indexStatisticsService.getIndexStatistics(session, index);
+        if (stats == null) {
+            return null;
+        }
+        Histogram fromHistogram = stats.getHistogram(0, index.getKeyColumns().size());
+        if (fromHistogram == null) {
+            return null;
+        }
+        IndexDef indexDef = index.indexDef();
+        RowDef indexRowDef = indexDef.getRowDef();
+        TableStatistics.Histogram toHistogram = new TableStatistics.Histogram(index.getIndexId());
+        RowData indexRowData = new RowData(new byte[4096]);
+        Object[] indexValues = new Object[indexRowDef.getFieldCount()];
+        long count = 0;
+        for (HistogramEntry entry : fromHistogram.getEntries()) {
+            // Decode the key.
+            int keylen = entry.getKeyBytes().length;
+            System.arraycopy(entry.getKeyBytes(), 0, key.getEncodedBytes(), 0, keylen);
+            key.setEncodedSize(keylen);
+            key.indexTo(0);
+            int depth = key.getDepth();
+            // Copy key fields to index row.
+            for (int field : indexDef.getFields()) {
+                if (--depth >= 0) {
+                    indexValues[field] = key.decode();
+                } else {
+                    indexValues[field] = null;
+                }
+            }
+            indexRowData.createRow(indexRowDef, indexValues);
+            // Partial counts to running total less than key.
+            count += entry.getLessCount();
+            toHistogram.addSample(new TableStatistics.HistogramSample(indexRowData.copy(), count));
+            count += entry.getEqualCount();
+        }
+        // Add final entry with all nulls.
+        Arrays.fill(indexValues, null);
+        indexRowData.createRow(indexRowDef, indexValues);
+        toHistogram.addSample(new TableStatistics.HistogramSample(indexRowData.copy(), count));
+        return toHistogram;
+    }
+
+    protected static boolean bytesEqual(byte[] a, int aoffset, int asize, byte[] b, int boffset, int bsize) {
+        if (asize != bsize) {
+            return false;
+        }
+        for (int i = 0; i < asize; i++) {
+            if (a[i + aoffset] != b[i + boffset]) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    protected static boolean fieldsEqual(RowDef rowDef, RowData a, RowData b, int[] fieldIndexes)
+    {
+        for (int fieldIndex : fieldIndexes) {
+            long aloc = rowDef.fieldLocation(a, fieldIndex);
+            long bloc = rowDef.fieldLocation(b, fieldIndex);
+            if (!bytesEqual(a.getBytes(), (int) aloc, (int) (aloc >>> 32),
+                            b.getBytes(), (int) bloc, (int) (bloc >>> 32))) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    protected static boolean fieldEqual(RowDef rowDef, RowData a, RowData b, int fieldPosition)
+    {
+        long aloc = rowDef.fieldLocation(a, fieldPosition);
+        long bloc = rowDef.fieldLocation(b, fieldPosition);
+        return bytesEqual(a.getBytes(), (int) aloc, (int) (aloc >>> 32),
+                          b.getBytes(), (int) bloc, (int) (bloc >>> 32));
+    }
+
+    protected BitSet analyzeFieldChanges(Session session, RowDef rowDef, RowData oldRow, RowData newRow)
+    {
+        BitSet tablesRequiringHKeyMaintenance;
+        assert oldRow.getRowDefId() == newRow.getRowDefId();
+        int fields = rowDef.getFieldCount();
+        // Find the PK and FK fields
+        BitSet keyField = new BitSet(fields);
+        for (int pkFieldPosition : rowDef.getPKIndex().indexDef().getFields()) {
+            keyField.set(pkFieldPosition, true);
+        }
+        for (int fkFieldPosition : rowDef.getParentJoinFields()) {
+            keyField.set(fkFieldPosition, true);
+        }
+        // Find whether and where key fields differ
+        boolean allEqual = true;
+        for (int keyFieldPosition = keyField.nextSetBit(0);
+             allEqual && keyFieldPosition >= 0;
+             keyFieldPosition = keyField.nextSetBit(keyFieldPosition + 1)) {
+            boolean fieldEqual = fieldEqual(rowDef, oldRow, newRow, keyFieldPosition);
+            if (!fieldEqual) {
+                allEqual = false;
+            }
+        }
+        if (allEqual) {
+            tablesRequiringHKeyMaintenance = null;
+        } else {
+            // A PK or FK field has changed, so the update has to be done as delete/insert. To minimize hkey
+            // propagation work, find which tables (descendents of the updated table) are affected by hkey
+            // changes.
+            tablesRequiringHKeyMaintenance = hKeyDependentTableOrdinals(session, oldRow.getRowDefId());
+        }
+        return tablesRequiringHKeyMaintenance;
+    }
+
+    private BitSet hKeyDependentTableOrdinals(Session session, int rowDefId)
+    {
+        RowDef rowDef = getRowDef(session, rowDefId);
+        UserTable table = rowDef.userTable();
+        BitSet ordinals = new BitSet();
+        for (UserTable hKeyDependentTable : table.hKeyDependentTables()) {
+            int ordinal = hKeyDependentTable.rowDef().getOrdinal();
+            ordinals.set(ordinal, true);
+        }
+        return ordinals;
+    }
+
+
+    //
+    // Store methods
+    //
+
+    @Override
+    public RowDef getRowDef(Session session, int rowDefID) {
+        Table table = getAIS(session).getUserTable(rowDefID);
+        if(table == null) {
+            throw new RowDefNotFoundException(rowDefID);
+        }
+        return table.rowDef();
+    }
+
+    @Override
+    public RowDef getRowDef(Session session, TableName tableName) {
+        Table table = getAIS(session).getTable(tableName);
+        if(table == null) {
+            throw new NoSuchTableException(tableName);
+        }
+        return table.rowDef();
+    }
+
+    @Override
+    public RowCollector newRowCollector(Session session,
+                                        int scanFlags,
+                                        int rowDefId,
+                                        int indexId,
+                                        byte[] columnBitMap,
+                                        RowData start,
+                                        ColumnSelector startColumns,
+                                        RowData end,
+                                        ColumnSelector endColumns,
+                                        ScanLimit scanLimit)
+    {
+        NEW_COLLECTOR_TAP.in();
+        RowCollector rc;
+        try {
+            if(start != null && startColumns == null) {
+                startColumns = createNonNullFieldSelector(start);
+            }
+            if(end != null && endColumns == null) {
+                endColumns = createNonNullFieldSelector(end);
+            }
+            RowDef rowDef = checkRequest(session, rowDefId, start, startColumns, end, endColumns);
+            rc = OperatorBasedRowCollector.newCollector(session,
+                                                        this,
+                                                        scanFlags,
+                                                        rowDef,
+                                                        indexId,
+                                                        columnBitMap,
+                                                        start,
+                                                        startColumns,
+                                                        end,
+                                                        endColumns,
+                                                        scanLimit);
+        } finally {
+            NEW_COLLECTOR_TAP.out();
+        }
+        return rc;
+    }
+
+    @Override
+    public void addSavedRowCollector(final Session session,
+                                     final RowCollector rc) {
+        final Integer tableId = rc.getTableId();
+        final List<RowCollector> list = collectorsForTableId(session, tableId);
+        if (!list.isEmpty()) {
+            LOG.debug("Note: Nested RowCollector on tableId={} depth={}", tableId, list.size() + 1);
+            assert list.get(list.size() - 1) != rc : "Redundant call";
+            //
+            // This disallows the patch because we agreed not to fix the
+            // bug. However, these changes fix a memory leak, which is
+            // important for robustness.
+            //
+            // throw new StoreException(122, "Bug 255 workaround is disabled");
+        }
+        list.add(rc);
+    }
+
+    @Override
+    public RowCollector getSavedRowCollector(final Session session,
+                                             final int tableId) throws CursorIsUnknownException {
+        final List<RowCollector> list = collectorsForTableId(session, tableId);
+        if (list.isEmpty()) {
+            LOG.debug("Nested RowCollector on tableId={} depth={}", tableId, (list.size() + 1));
+            throw new CursorIsUnknownException(tableId);
+        }
+        return list.get(list.size() - 1);
+    }
+
+    @Override
+    public void removeSavedRowCollector(final Session session,
+                                        final RowCollector rc) throws CursorIsUnknownException {
+        final Integer tableId = rc.getTableId();
+        final List<RowCollector> list = collectorsForTableId(session, tableId);
+        if (list.isEmpty()) {
+            throw new CursorIsUnknownException (tableId);
+        }
+        final RowCollector removed = list.remove(list.size() - 1);
+        if (removed != rc) {
+            throw new CursorCloseBadException(tableId);
+        }
+    }
+
+    // This is to avoid circular dependencies in Guicer.
+    // TODO: There is still a functional circularity: store needs
+    // stats to clear them when deleting a group; stats need store to
+    // persist the stats. It would be better to separate out the
+    // higher level store functions from what other services require.
+    @Override
+    public void setIndexStatistics(IndexStatisticsService indexStatisticsService) {
+        this.indexStatisticsService = indexStatisticsService;
+    }
+
+    @Override
+    public TableStatistics getTableStatistics(final Session session, int tableId) {
+        final RowDef rowDef = getRowDef(session, tableId);
+        final TableStatistics ts = new TableStatistics(tableId);
+        final TableStatus status = rowDef.getTableStatus();
+        ts.setAutoIncrementValue(status.getAutoIncrement());
+        ts.setRowCount(status.getRowCount());
+        // TODO - get correct values
+        ts.setMeanRecordLength(100);
+        ts.setBlockSize(8192);
+        for(Index index : rowDef.getIndexes()) {
+            if(index.isSpatial()) {
+                continue;
+            }
+            TableStatistics.Histogram histogram = indexStatisticsToHistogram(session, index, createKey());
+            if(histogram != null) {
+                ts.addHistogram(histogram);
+            }
+        }
+        return ts;
+    }
+
+    @Override
+    public long getRowCount(Session session, boolean exact, RowData start, RowData end, byte[] columnBitMap) {
+        // TODO: Compute a reasonable value. The value 2 is special because it is not 0 or 1 but will
+        // still induce MySQL to use an index rather than a full table scan.
+        return 2;
+    }
+
+    @Override
+    public void dropGroup(final Session session, Group group) {
+        group.getRoot().traverseTableAndDescendants(new NopVisitor() {
+            @Override
+            public void visitUserTable(UserTable table) {
+                removeTrees(session, table);
+            }
+        });
+    }
+
+    @Override
+    public void truncateGroup(final Session session, final Group group) {
+        final List<Index> indexes = new ArrayList<>();
+        // Collect indexes, truncate table statuses
+        group.getRoot().traverseTableAndDescendants(new NopVisitor() {
+            @Override
+            public void visitUserTable(UserTable table) {
+                indexes.addAll(table.getIndexesIncludingInternal());
+                table.rowDef().getTableStatus().truncate();
+            }
+        });
+
+        indexes.addAll(group.getIndexes());
+        truncateIndexes(session, indexes);
+
+        // Truncate the group tree
+        truncateTree(session, group);
+    }
+
+    @Override
+    public void truncateTableStatus(final Session session, final int rowDefId) {
+        getRowDef(session, rowDefId).getTableStatus().truncate();
+    }
+
+    @Override
+    public void truncateIndexes(Session session, Collection<? extends Index> indexes) {
+        for(Index index : indexes) {
+            truncateTree(session, index.indexDef());
+        }
+        // Delete any statistics associated with index.
+        indexStatisticsService.deleteIndexStatistics(session, indexes);
+    }
+
+    @Override
+    public void deleteIndexes(final Session session, final Collection<? extends Index> indexes) {
+        for(Index index : indexes) {
+            final IndexDef indexDef = index.indexDef();
+            if(indexDef != null) {
+                removeTree(session, indexDef);
+            }
+        }
+        indexStatisticsService.deleteIndexStatistics(session, indexes);
+    }
+
+    @Override
+    public void removeTrees(Session session, UserTable table) {
+        // Table indexes
+        for(Index index : table.getIndexesIncludingInternal()) {
+            removeTree(session, index.indexDef());
+        }
+        indexStatisticsService.deleteIndexStatistics(session, table.getIndexesIncludingInternal());
+
+        // Group indexes
+        for(Index index : table.getGroupIndexes()) {
+            removeTree(session, index.indexDef());
+        }
+        indexStatisticsService.deleteIndexStatistics(session, table.getGroupIndexes());
+
+        // Sequence
+        if(table.getIdentityColumn() != null) {
+            deleteSequences(session, Collections.singleton(table.getIdentityColumn().getIdentityGenerator()));
+        }
+
+        // And the group tree
+        removeTree(session, table.getGroup());
+    }
+
+    @Override
+    public void removeTrees(Session session, Collection<? extends TreeLink> treeLinks) {
+        for(TreeLink link : treeLinks) {
+            removeTree(session, link);
+        }
+    }
+
+
+    //
+    // Internal
+    //
+
+    private List<RowCollector> collectorsForTableId(final Session session, final int tableId) {
+        List<RowCollector> list = session.get(COLLECTORS, tableId);
+        if (list == null) {
+            list = new ArrayList<>();
+            session.put(COLLECTORS, tableId, list);
+        }
+        return list;
+    }
+
+    private RowDef checkRequest(Session session, int rowDefId, RowData start, ColumnSelector startColumns,
+                                RowData end, ColumnSelector endColumns) throws IllegalArgumentException {
+        if (start != null) {
+            if (startColumns == null) {
+                throw new IllegalArgumentException("non-null start row requires non-null ColumnSelector");
+            }
+            if( start.getRowDefId() != rowDefId) {
+                throw new IllegalArgumentException("Start and end RowData must specify the same rowDefId");
+            }
+        }
+        if (end != null) {
+            if (endColumns == null) {
+                throw new IllegalArgumentException("non-null end row requires non-null ColumnSelector");
+            }
+            if (end.getRowDefId() != rowDefId) {
+                throw new IllegalArgumentException("Start and end RowData must specify the same rowDefId");
+            }
+        }
+        final RowDef rowDef = getRowDef(session, rowDefId);
+        if (rowDef == null) {
+            throw new IllegalArgumentException("No RowDef for rowDefId " + rowDefId);
+        }
+        return rowDef;
+    }
+
+    private static ColumnSelector createNonNullFieldSelector(final RowData rowData) {
+        assert rowData != null;
+        return new ColumnSelector() {
+            @Override
+            public boolean includesColumn(int columnPosition) {
+                return !rowData.isNull(columnPosition);
+            }
+        };
+    }
+}

=== modified file 'src/main/java/com/akiban/server/store/DelegatingStore.java'
--- src/main/java/com/akiban/server/store/DelegatingStore.java	2013-03-22 20:05:57 +0000
+++ src/main/java/com/akiban/server/store/DelegatingStore.java	2013-05-20 14:47:34 +0000
@@ -17,10 +17,14 @@
 
 package com.akiban.server.store;
 
+import com.akiban.ais.model.AkibanInformationSchema;
 import com.akiban.ais.model.Group;
 import com.akiban.ais.model.Index;
 import com.akiban.ais.model.Sequence;
-import com.akiban.ais.model.Table;
+import com.akiban.ais.model.TableName;
+import com.akiban.ais.model.UserTable;
+import com.akiban.qp.operator.StoreAdapter;
+import com.akiban.qp.rowtype.Schema;
 import com.akiban.server.rowdata.RowData;
 import com.akiban.server.rowdata.RowDef;
 import com.akiban.server.TableStatistics;
@@ -30,7 +34,8 @@
 import com.akiban.server.service.Service;
 import com.akiban.server.service.session.Session;
 import com.akiban.server.service.tree.TreeLink;
-import com.persistit.exception.PersistitException;
+import com.akiban.server.store.statistics.IndexStatisticsService;
+import com.persistit.Key;
 
 import java.util.Collection;
 
@@ -42,7 +47,8 @@
         this.delegate = delegate;
     }
 
-    protected S getDelegate() {
+    // TODO: Ditch DelegatingStore altogether
+    public S getDelegate() {
         return delegate;
     }
 
@@ -62,11 +68,22 @@
         delegate.crash();
     }
 
+    @Override
+    public AkibanInformationSchema getAIS(Session session) {
+        return delegate.getAIS(session);
+    }
+
+    @Override
     public RowDef getRowDef(Session session, int rowDefID) {
         return delegate.getRowDef(session, rowDefID);
     }
 
     @Override
+    public RowDef getRowDef(Session session, TableName tableName) {
+        return delegate.getRowDef(session, tableName);
+    }
+
+    @Override
     public void startBulkLoad(Session session) {
         delegate.startBulkLoad(session);
     }
@@ -81,15 +98,15 @@
         delegate.finishBulkLoad(session);
     }
 
-    public void writeRow(Session session, RowData rowData) throws PersistitException {
+    public void writeRow(Session session, RowData rowData) {
         delegate.writeRow(session, rowData);
     }
 
-    public void deleteRow(Session session, RowData rowData, boolean deleteIndexes, boolean cascadeDelete) throws PersistitException {
+    public void deleteRow(Session session, RowData rowData, boolean deleteIndexes, boolean cascadeDelete) {
         delegate.deleteRow(session, rowData, deleteIndexes, cascadeDelete);
     }
 
-    public void updateRow(Session session, RowData oldRowData, RowData newRowData, ColumnSelector columnSelector, Index[] indexes) throws PersistitException {
+    public void updateRow(Session session, RowData oldRowData, RowData newRowData, ColumnSelector columnSelector, Index[] indexes) {
         delegate.updateRow(session, oldRowData, newRowData, columnSelector, indexes);
     }
 
@@ -97,11 +114,11 @@
         delegate.dropGroup(session, group);
     }
 
-    public void truncateGroup(Session session, Group group) throws PersistitException {
+    public void truncateGroup(Session session, Group group) {
         delegate.truncateGroup(session, group);
     }
 
-    public void truncateTableStatus(Session session, int rowDefId) throws PersistitException {
+    public void truncateTableStatus(Session session, int rowDefId) {
         delegate.truncateTableStatus(session, rowDefId);
     }
 
@@ -117,11 +134,6 @@
         delegate.removeSavedRowCollector(session, rc);
     }
 
-    @SuppressWarnings("deprecation")
-    public RowCollector newRowCollector(Session session, int rowDefId, int indexId, int scanFlags, RowData start, RowData end, byte[] columnBitMap, ScanLimit scanLimit) {
-        return delegate.newRowCollector(session, rowDefId, indexId, scanFlags, start, end, columnBitMap, scanLimit);
-    }
-
     public RowCollector newRowCollector(Session session, int scanFlags, int rowDefId, int indexId, byte[] columnBitMap, RowData start, ColumnSelector startColumns, RowData end, ColumnSelector endColumns, ScanLimit scanLimit) {
         return delegate.newRowCollector(session, scanFlags, rowDefId, indexId, columnBitMap, start, startColumns, end, endColumns, scanLimit);
     }
@@ -134,10 +146,6 @@
         return delegate.getTableStatistics(session, tableId);
     }
 
-    public void flushIndexes(Session session) {
-        delegate.flushIndexes(session);
-    }
-
     public void buildIndexes(Session session, Collection<? extends Index> indexes, boolean defer) {
         delegate.buildIndexes(session, indexes, defer);
     }
@@ -146,22 +154,11 @@
         delegate.deleteIndexes(session, indexes);
     }
 
-    public void deleteSequences (Session session, Collection<? extends Sequence> sequences) {
-        delegate.deleteSequences(session, sequences);
-    }
-    
-    public void removeTrees(Session session, Table table) {
+    @Override
+    public void removeTrees(Session session, UserTable table) {
         delegate.removeTrees(session, table);
     }
 
-    public boolean isDeferIndexes() {
-        return delegate.isDeferIndexes();
-    }
-
-    public void setDeferIndexes(boolean defer) {
-        delegate.setDeferIndexes(defer);
-    }
-
     @Override
     public void truncateIndexes(Session session, Collection<? extends Index> indexes) {
         delegate.truncateIndexes(session, indexes);
@@ -171,4 +168,34 @@
     public void removeTrees(Session session, Collection<? extends TreeLink> treeLinks) {
         delegate.removeTrees(session, treeLinks);
     }
+
+    @Override
+    public StoreAdapter createAdapter(Session session, Schema schema) {
+        return delegate.createAdapter(session, schema);
+    }
+
+    @Override
+    public void setIndexStatistics(IndexStatisticsService indexStatistics) {
+        delegate.setIndexStatistics(indexStatistics);
+    }
+
+    @Override
+    public void truncateTree(Session session, TreeLink treeLink) {
+        delegate.truncateTree(session, treeLink);
+    }
+
+    @Override
+    public Key createKey() {
+        return delegate.createKey();
+    }
+
+    @Override
+    public void deleteSequences(Session session, Collection<? extends Sequence> sequences) {
+        delegate.deleteSequences(session, sequences);
+    }
+
+    @Override
+    public void removeTree(Session session, TreeLink treeLink) {
+        delegate.removeTree(session, treeLink);
+    }
 }

=== modified file 'src/main/java/com/akiban/server/store/PersistitStore.java'
--- src/main/java/com/akiban/server/store/PersistitStore.java	2013-05-03 00:59:49 +0000
+++ src/main/java/com/akiban/server/store/PersistitStore.java	2013-05-20 14:47:34 +0000
@@ -20,19 +20,18 @@
 import com.akiban.ais.model.*;
 import com.akiban.ais.model.Index.IndexType;
 import com.akiban.qp.operator.StoreAdapter;
-import com.akiban.qp.persistitadapter.OperatorBasedRowCollector;
 import com.akiban.qp.persistitadapter.PersistitAdapter;
 import com.akiban.qp.persistitadapter.PersistitHKey;
 import com.akiban.qp.persistitadapter.indexrow.PersistitIndexRow;
 import com.akiban.qp.persistitadapter.indexrow.PersistitIndexRowBuffer;
 import com.akiban.qp.rowtype.IndexRowType;
+import com.akiban.qp.rowtype.Schema;
 import com.akiban.qp.util.SchemaCache;
 import com.akiban.server.*;
 import com.akiban.server.api.dml.ColumnSelector;
 import com.akiban.server.api.dml.scan.LegacyRowWrapper;
 import com.akiban.server.api.dml.scan.NewRow;
 import com.akiban.server.api.dml.scan.NiceRow;
-import com.akiban.server.api.dml.scan.ScanLimit;
 import com.akiban.server.collation.CString;
 import com.akiban.server.collation.CStringKeyCoder;
 import com.akiban.server.error.*;
@@ -46,10 +45,6 @@
 import com.akiban.server.service.transaction.TransactionService;
 import com.akiban.server.service.tree.TreeLink;
 import com.akiban.server.service.tree.TreeService;
-import com.akiban.server.store.statistics.Histogram;
-import com.akiban.server.store.statistics.HistogramEntry;
-import com.akiban.server.store.statistics.IndexStatistics;
-import com.akiban.server.store.statistics.IndexStatisticsService;
 import com.akiban.util.tap.InOutTap;
 import com.akiban.util.tap.PointTap;
 import com.akiban.util.tap.Tap;
@@ -69,9 +64,8 @@
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.concurrent.atomic.AtomicReference;
 
-public class PersistitStore implements Store, Service {
-
-    private static final Session.MapKey<Integer, List<RowCollector>> COLLECTORS = Session.MapKey.mapNamed("collectors");
+public class PersistitStore extends AbstractStore implements Service
+{
     private static final AtomicReference<Bulkload> activeBulkload = new AtomicReference<>();
 
     private static final Logger LOG = LoggerFactory
@@ -85,8 +79,6 @@
 
     private static final InOutTap TABLE_INDEX_MAINTENANCE_TAP = Tap.createTimer("index: maintain_table");
 
-    private static final InOutTap NEW_COLLECTOR_TAP = Tap.createTimer("read: new_collector");
-
     // an InOutTap would be nice, but pre-propagateDownGroup optimization, propagateDownGroup was called recursively
     // (via writeRow). PointTap handles this correctly, InOutTap does not, currently.
     private static final PointTap PROPAGATE_HKEY_CHANGE_TAP = Tap.createCount("write: propagate_hkey_change");
@@ -102,10 +94,6 @@
 
     private final static int MAX_ROW_SIZE = 5000000;
 
-    private final static int MAX_INDEX_TRANCHE_SIZE = 10 * 1024 * 1024;
-
-    private final static int KEY_STATE_SIZE_OVERHEAD = 50;
-
     private final static byte[] EMPTY_BYTE_ARRAY = new byte[0];
     
     private boolean writeLockEnabled;
@@ -126,12 +114,6 @@
 
     private DisplayFilter originalDisplayFilter;
 
-    private volatile IndexStatisticsService indexStatistics;
-
-    private final Map<Tree, SortedSet<KeyState>> deferredIndexKeys = new HashMap<>();
-
-    private int deferredIndexKeyLimit = MAX_INDEX_TRANCHE_SIZE;
-
     private FullTextIndexService fullTextService;
 
     private RowDataValueCoder valueCoder;
@@ -161,6 +143,12 @@
         this.transactionService = transactionService;
     }
 
+
+    //
+    // FullText change tracking
+    // TODO: Move out of PersistitStore
+    //
+
     public void setFullTextService(FullTextIndexService service)
     {
         fullTextService = service;
@@ -579,14 +567,15 @@
     }
 
     @Override
+    public Key createKey() {
+        return treeService.createKey();
+    }
+
+    @Override
     public PersistitStore getPersistitStore() {
         return this;
     }
 
-    public TreeService treeService() {
-        return treeService;
-    }
-
     public Persistit getDb() {
         return treeService.getDb();
     }
@@ -603,11 +592,6 @@
         return treeService.getExchange(session, index.indexDef());
     }
 
-    public Key getKey()
-    {
-        return treeService.getKey();
-    }
-
     public void releaseExchange(final Session session, final Exchange exchange) {
         treeService.releaseExchange(session, exchange);
     }
@@ -744,6 +728,7 @@
 
     // --------------------- Implement Store interface --------------------
 
+    @Override
     public AkibanInformationSchema getAIS(Session session) {
         Bulkload bulkload = activeBulkload.get();
         if (bulkload != null)
@@ -751,32 +736,18 @@
         return schemaManager.getAis(session);
     }
 
-    public RowDef getRowDef(Session session, TableName tableName) {
-        Table table = getAIS(session).getTable(tableName);
-        if(table == null) {
-            throw new NoSuchTableException(tableName);
-        }
-        return table.rowDef();
-    }
-
-    @Override
-    public RowDef getRowDef(Session session, int rowDefID) {
-        Table table = getAIS(session).getUserTable(rowDefID);
-        if(table == null) {
-            throw new RowDefNotFoundException(rowDefID);
-        }
-        return table.rowDef();
-    }
-
     @Override
     public void writeRow(Session session, RowData rowData)
-        throws PersistitException
     {
         Bulkload bulkload = activeBulkload.get();
-        if (bulkload != null)
-            writeRowBulk(session, rowData, bulkload);
-        else
-            writeRowStandard(session, rowData, null, true);
+        try {
+            if (bulkload != null)
+                writeRowBulk(session, rowData, bulkload);
+            else
+                writeRowStandard(session, rowData, null, true);
+        } catch(PersistitException e) {
+            throw PersistitAdapter.wrapPersistitException(session, e);
+        }
     }
 
     private void writeRowStandard(Session session,
@@ -804,7 +775,7 @@
             // in a good position to report a meaningful uniqueness violation, e.g. on the PK, since we don't have
             // the PK value handy. Instead, rely on PK validation when indexes are maintained.
 
-            packRowData(hEx, rowDef, rowData);
+            packRowData(hEx, rowData);
             // Store the h-row
             hEx.store();
             if (rowDef.isAutoIncrement()) {
@@ -824,7 +795,7 @@
             // bug1112940: Bump row count *after* uniqueness checks in insertIntoIndex
             rowDef.getTableStatus().rowsWritten(1);
 
-            if (propagateHKeyChanges && hasChildren(rowDef.userTable())) {
+            if (propagateHKeyChanges && rowDef.userTable().hasChildren()) {
                 // The row being inserted might be the parent of orphan rows
                 // already present. The hkeys of these
                 // orphan rows need to be maintained. The hkeys of interest
@@ -860,17 +831,13 @@
                 }
                 propagateDownGroup(session, hEx, tablesRequiringHKeyMaintenance, indexRow, true, false);
             }
-
-            if (deferredIndexKeyLimit <= 0) {
-                putAllDeferredIndexKeys(session);
-            }
         } finally {
             WRITE_ROW_TAP.out();
             releaseExchange(session, hEx);
         }
     }
 
-    private void writeRowBulk(Session session, RowData rowData, Bulkload bulkload) throws PersistitException {
+    private void writeRowBulk(Session session, RowData rowData, Bulkload bulkload) {
         final RowDef rowDef = writeRowCheck(session, rowData, true);
         if(session.get(StoreAdapter.STORE_ADAPTER_KEY) == null) {
             // Attaches itself to the session
@@ -908,25 +875,27 @@
         // Group table
         Key groupTableKey = bulkload.groupTableKey.get();
         Value groupTableValue = bulkload.groupTableValue.get();
-        constructHKey(session, groupTableKey, rowDef, rowData, true, hiddenPk); // invokes key.clear()
-        groupTableValue.clear();
-        packRowData(groupTableValue, rowDef, rowData);
         try {
+            constructHKey(session, groupTableKey, rowDef, rowData, true, hiddenPk); // invokes key.clear()
+            groupTableValue.clear();
+            packRowData(groupTableValue, rowData);
             Tree tree = treeService.populateTreeCache(rowDef.getGroup()).getTree();
             bulkload.groupBuilder.treeBuilder.store(tree, groupTableKey, groupTableValue);
+
+            PersistitIndexRowBuffer indexRow = new PersistitIndexRowBuffer(adapter(session));
+            for (Index index : rowDef.getIndexes()) {
+                StorageAction action = index.isPrimaryKey() ? bulkload.pkStorage : bulkload.groupBuilder;
+                insertIntoIndex(session, index, rowData, groupTableKey, indexRow, deferIndexes, action);
+            }
         } catch (InvalidOperationException e) {
             throw e;
+        } catch (PersistitException e) {
+            throw PersistitAdapter.wrapPersistitException(session, e);
         } catch (Exception e) {
             LOG.error("while merging PKs", e);
             throw new BulkloadException("unknown exception (see log): " + e.getMessage());
         }
 
-        PersistitIndexRowBuffer indexRow = new PersistitIndexRowBuffer(adapter(session));
-        for (Index index : rowDef.getIndexes()) {
-            StorageAction action = index.isPrimaryKey() ? bulkload.pkStorage : bulkload.groupBuilder;
-            insertIntoIndex(session, index, rowData, groupTableKey, indexRow, deferIndexes, action);
-        }
-
         insertedRowsCount.incrementAndGet();
     }
 
@@ -1030,18 +999,17 @@
     }
 
     @Override
-    public void deleteRow(Session session, RowData rowData, boolean deleteIndexes, boolean cascadeDelete) throws PersistitException
+    public void deleteRow(Session session, RowData rowData, boolean deleteIndexes, boolean cascadeDelete)
     {
         deleteRow(session, rowData, deleteIndexes, cascadeDelete, null, true);
     }
-    
-    private void deleteRow(Session session, RowData rowData, boolean deleteIndexes, boolean cascadeDelete, 
+
+    private void deleteRow(Session session, RowData rowData, boolean deleteIndexes, boolean cascadeDelete,
                            BitSet tablesRequiringHKeyMaintenance, boolean propagateHKeyChanges)
-        throws PersistitException
     {
         RowDef rowDef = writeCheck(session, rowData, false);
         Exchange hEx = null;
-
+        
         DELETE_ROW_TAP.in();
         try {
             hEx = getExchange(session, rowDef);
@@ -1075,9 +1043,11 @@
             // The row being deleted might be the parent of rows that
             // now become orphans. The hkeys
             // of these rows need to be maintained.
-            if(propagateHKeyChanges && hasChildren(rowDef.userTable())) {
+            if(propagateHKeyChanges && rowDef.userTable().hasChildren()) {
                 propagateDownGroup(session, hEx, tablesRequiringHKeyMaintenance, indexRow, deleteIndexes, cascadeDelete);
             }
+        } catch(PersistitException e) {
+            throw PersistitAdapter.wrapPersistitException(session, e);
         } finally {
             DELETE_ROW_TAP.out();
             releaseExchange(session, hEx);
@@ -1090,7 +1060,6 @@
                           RowData newRowData,
                           ColumnSelector columnSelector,
                           Index[] indexes)
-        throws PersistitException
     {
         updateRow(session, oldRowData, newRowData, columnSelector, indexes, (indexes != null), true);
     }
@@ -1102,7 +1071,6 @@
                            Index[] indexesToMaintain,
                            boolean indexesAsInsert,
                            boolean propagateHKeyChanges)
-        throws PersistitException
     {
         int rowDefId = oldRowData.getRowDefId();
         if (newRowData.getRowDefId() != rowDefId) {
@@ -1147,7 +1115,7 @@
                     : null;
             if (tablesRequiringHKeyMaintenance == null) {
                 // No PK or FK fields have changed. Just update the row.
-                packRowData(hEx, newRowDef, mergedRowData);
+                packRowData(hEx, mergedRowData);
                 // Store the h-row
                 hEx.store();
                 // Update the indexes (new row)
@@ -1169,58 +1137,14 @@
                 deleteRow(session, oldRowData, true, false, tablesRequiringHKeyMaintenance, true);
                 writeRowStandard(session, mergedRowData, tablesRequiringHKeyMaintenance, true); // May throw DuplicateKeyException
             }
+        } catch(PersistitException e) {
+            throw PersistitAdapter.wrapPersistitException(session, e);
         } finally {
             UPDATE_ROW_TAP.out();
             releaseExchange(session, hEx);
         }
     }
 
-    private BitSet analyzeFieldChanges(Session session, RowDef rowDef, RowData oldRow, RowData newRow)
-    {
-        BitSet tablesRequiringHKeyMaintenance;
-        assert oldRow.getRowDefId() == newRow.getRowDefId();
-        int fields = rowDef.getFieldCount();
-        // Find the PK and FK fields
-        BitSet keyField = new BitSet(fields);
-        for (int pkFieldPosition : rowDef.getPKIndex().indexDef().getFields()) {
-            keyField.set(pkFieldPosition, true);
-        }
-        for (int fkFieldPosition : rowDef.getParentJoinFields()) {
-            keyField.set(fkFieldPosition, true);
-        }
-        // Find whether and where key fields differ
-        boolean allEqual = true;
-        for (int keyFieldPosition = keyField.nextSetBit(0);
-             allEqual && keyFieldPosition >= 0;
-             keyFieldPosition = keyField.nextSetBit(keyFieldPosition + 1)) {
-            boolean fieldEqual = fieldEqual(rowDef, oldRow, newRow, keyFieldPosition);
-            if (!fieldEqual) {
-                allEqual = false;
-            }
-        }
-        if (allEqual) {
-            tablesRequiringHKeyMaintenance = null;
-        } else {
-            // A PK or FK field has changed, so the update has to be done as delete/insert. To minimize hkey
-            // propagation work, find which tables (descendents of the updated table) are affected by hkey
-            // changes.
-            tablesRequiringHKeyMaintenance = hKeyDependentTableOrdinals(session, oldRow.getRowDefId());
-        }
-        return tablesRequiringHKeyMaintenance;
-    }
-
-    private BitSet hKeyDependentTableOrdinals(Session session, int rowDefId)
-    {
-        RowDef rowDef = getRowDef(session, rowDefId);
-        UserTable table = rowDef.userTable();
-        BitSet ordinals = new BitSet();
-        for (UserTable hKeyDependentTable : table.hKeyDependentTables()) {
-            int ordinal = hKeyDependentTable.rowDef().getOrdinal();
-            ordinals.set(ordinal, true);
-        }
-        return ordinals;
-    }
-
     private void checkNoGroupIndexes(Table table) {
         if (updateGroupIndexes && !table.getGroupIndexes().isEmpty()) {
             throw new UnsupportedOperationException("PersistitStore can't update group indexes; found on " + table);
@@ -1278,308 +1202,42 @@
     }
 
     @Override
-    public void dropGroup(Session session, Group group) {
-        for(Table table : group.getRoot().getAIS().getUserTables().values()) {
-            if(table.getGroup() == group) {
-                removeTrees(session, table);
-            }
-        }
-        // tableStatusCache entries updated elsewhere
-    }
-
-    @Override
-    public void truncateGroup(final Session session, final Group group) throws PersistitException {
+    public void truncateGroup(final Session session, final Group group) {
         List<Index> indexes = new ArrayList<>();
-        // Collect indexes, truncate table statuses
-        for(UserTable table : group.getRoot().getAIS().getUserTables().values()) {
-            if(table.getGroup() == group) {
-                indexes.addAll(table.getIndexesIncludingInternal());
-                table.rowDef().getTableStatus().truncate();
-            }
-        }
-        indexes.addAll(group.getIndexes());
-        truncateIndexes(session, indexes);
-
         // Truncate the group tree
         final Exchange hEx = getExchange(session, group);
-        hEx.removeAll();
-        releaseExchange(session, hEx);
-    }
+        try {
+            // Collect indexes, truncate table statuses
+            for(UserTable table : group.getRoot().getAIS().getUserTables().values()) {
+                if(table.getGroup() == group) {
+                    indexes.addAll(table.getIndexesIncludingInternal());
+                    table.rowDef().getTableStatus().truncate();
+                }
+            }
+            indexes.addAll(group.getIndexes());
+            truncateIndexes(session, indexes);
 
-    // This is to avoid circular dependencies in Guicer.  
-    // TODO: There is still a functional circularity: store needs
-    // stats to clear them when deleting a group; stats need store to
-    // persist the stats. It would be better to separate out the
-    // higher level store functions from what other services require.
-    public void setIndexStatistics(IndexStatisticsService indexStatistics) {
-        this.indexStatistics = indexStatistics;
+            hEx.removeAll();
+        } catch(PersistitException e) {
+            throw PersistitAdapter.wrapPersistitException(session, e);
+        } finally {
+            releaseExchange(session, hEx);
+        }
     }
 
     @Override
     public void truncateIndexes(Session session, Collection<? extends Index> indexes) {
+        super.truncateIndexes(session, indexes);
         for(Index index : indexes) {
-            Exchange iEx = getExchange(session, index);
-            try {
-                iEx.removeAll();
-                if (index.isGroupIndex()) {
-                    new AccumulatorAdapter(AccumulatorAdapter.AccumInfo.ROW_COUNT, iEx.getTree()).set(0);
-                }
-            } catch (PersistitException e) {
-                throw new PersistitAdapterException(e);
-            }
-            releaseExchange(session, iEx);
-        }
-        // Delete any statistics associated with index.
-        indexStatistics.deleteIndexStatistics(session, indexes);
-    }
-
-    @Override
-    public void truncateTableStatus(final Session session, final int rowDefId) throws PersistitException {
-        getRowDef(session, rowDefId).getTableStatus().truncate();
-    }
-
-    @Override
-    public RowCollector getSavedRowCollector(final Session session,
-            final int tableId) throws CursorIsUnknownException {
-        final List<RowCollector> list = collectorsForTableId(session, tableId);
-        if (list.isEmpty()) {
-            LOG.debug("Nested RowCollector on tableId={} depth={}", tableId, (list.size() + 1));
-            throw new CursorIsUnknownException(tableId);
-        }
-        return list.get(list.size() - 1);
-    }
-
-    @Override
-    public void addSavedRowCollector(final Session session,
-            final RowCollector rc) {
-        final Integer tableId = rc.getTableId();
-        final List<RowCollector> list = collectorsForTableId(session, tableId);
-        if (!list.isEmpty()) {
-            LOG.debug("Note: Nested RowCollector on tableId={} depth={}", tableId, list.size() + 1);
-            assert list.get(list.size() - 1) != rc : "Redundant call";
-            //
-            // This disallows the patch because we agreed not to fix the
-            // bug. However, these changes fix a memory leak, which is
-            // important for robustness.
-            //
-            // throw new StoreException(122, "Bug 255 workaround is disabled");
-        }
-        list.add(rc);
-    }
-
-    @Override
-    public void removeSavedRowCollector(final Session session,
-            final RowCollector rc) throws CursorIsUnknownException {
-        final Integer tableId = rc.getTableId();
-        final List<RowCollector> list = collectorsForTableId(session, tableId);
-        if (list.isEmpty()) {
-            throw new CursorIsUnknownException (tableId);
-        }
-        final RowCollector removed = list.remove(list.size() - 1);
-        if (removed != rc) {
-            throw new CursorCloseBadException (tableId);
-        }
-    }
-
-    private List<RowCollector> collectorsForTableId(final Session session,
-            final int tableId) {
-        List<RowCollector> list = session.get(COLLECTORS, tableId);
-        if (list == null) {
-            list = new ArrayList<>();
-            session.put(COLLECTORS, tableId, list);
-        }
-        return list;
-    }
-
-    private RowDef checkRequest(Session session, int rowDefId, RowData start, ColumnSelector startColumns,
-                                RowData end, ColumnSelector endColumns) throws IllegalArgumentException {
-        if (start != null) {
-            if (startColumns == null) {
-                throw new IllegalArgumentException("non-null start row requires non-null ColumnSelector");
-            }
-            if( start.getRowDefId() != rowDefId) {
-                throw new IllegalArgumentException("Start and end RowData must specify the same rowDefId");
-            }
-        }
-        if (end != null) {
-            if (endColumns == null) {
-                throw new IllegalArgumentException("non-null end row requires non-null ColumnSelector");
-            }
-            if (end.getRowDefId() != rowDefId) {
-                throw new IllegalArgumentException("Start and end RowData must specify the same rowDefId");
-            }
-        }
-        final RowDef rowDef = getRowDef(session, rowDefId);
-        if (rowDef == null) {
-            throw new IllegalArgumentException("No RowDef for rowDefId " + rowDefId);
-        }
-        return rowDef;
-    }
-
-    private static ColumnSelector createNonNullFieldSelector(final RowData rowData) {
-        assert rowData != null;
-        return new ColumnSelector() {
-            @Override
-            public boolean includesColumn(int columnPosition) {
-                return !rowData.isNull(columnPosition);
-            }
-        };
-    }
-
-    @Override
-    public RowCollector newRowCollector(Session session,
-                                        int rowDefId,
-                                        int indexId,
-                                        int scanFlags,
-                                        RowData start,
-                                        RowData end,
-                                        byte[] columnBitMap,
-                                        ScanLimit scanLimit)
-    {
-        return newRowCollector(session, scanFlags, rowDefId, indexId, columnBitMap, start, null, end, null, scanLimit);
-    }
-
-    @Override
-    public RowCollector newRowCollector(Session session,
-                                        int scanFlags,
-                                        int rowDefId,
-                                        int indexId,
-                                        byte[] columnBitMap,
-                                        RowData start,
-                                        ColumnSelector startColumns,
-                                        RowData end,
-                                        ColumnSelector endColumns,
-                                        ScanLimit scanLimit)
-    {
-        NEW_COLLECTOR_TAP.in();
-        RowCollector rc;
-        try {
-            if(start != null && startColumns == null) {
-                startColumns = createNonNullFieldSelector(start);
-            }
-            if(end != null && endColumns == null) {
-                endColumns = createNonNullFieldSelector(end);
-            }
-            RowDef rowDef = checkRequest(session, rowDefId, start, startColumns, end, endColumns);
-            rc = OperatorBasedRowCollector.newCollector(config,
-                                                        session,
-                                                        this,
-                                                        scanFlags,
-                                                        rowDef,
-                                                        indexId,
-                                                        columnBitMap,
-                                                        start,
-                                                        startColumns,
-                                                        end,
-                                                        endColumns,
-                                                        scanLimit);
-        } finally {
-            NEW_COLLECTOR_TAP.out();
-        }
-        return rc;
-    }
-
-    public final static long HACKED_ROW_COUNT = 2;
-
-    @Override
-    public long getRowCount(final Session session, final boolean exact,
-            final RowData start, final RowData end, final byte[] columnBitMap) {
-        //
-        // TODO: Compute a reasonable value. The value "2" is a hack -
-        // special because it's not 0 or 1, but small enough to induce
-        // MySQL to use an index rather than full table scan.
-        //
-        return HACKED_ROW_COUNT; // TODO: delete the HACKED_ROW_COUNT field when
-                                 // this gets fixed
-        // final int tableId = start.getRowDefId();
-        // final TableStatus status = tableManager.getTableStatus(tableId);
-        // return status.getRowCount();
-    }
-
-    @Override
-    public TableStatistics getTableStatistics(final Session session, int tableId) {
-        final RowDef rowDef = getRowDef(session, tableId);
-        final TableStatistics ts = new TableStatistics(tableId);
-        final TableStatus status = rowDef.getTableStatus();
-        try {
-            ts.setAutoIncrementValue(status.getAutoIncrement());
-            ts.setRowCount(status.getRowCount());
-            // TODO - get correct values
-            ts.setMeanRecordLength(100);
-            ts.setBlockSize(8192);
-        } catch (PersistitException e) {
-            throw new PersistitAdapterException(e);
-        }
-        for (Index index : rowDef.getIndexes()) {
-            if (index.isSpatial())
-                continue;
-            TableStatistics.Histogram histogram = indexStatisticsToHistogram(session, 
-                                                                             index);
-            if (histogram != null) {
-                ts.addHistogram(histogram);
-            }
-        }
-        return ts;
-    }
-
-    /** Convert from new-format histogram to old for adapter. */
-    protected TableStatistics.Histogram indexStatisticsToHistogram(Session session,
-                                                                   Index index) {
-        IndexStatistics stats = indexStatistics.getIndexStatistics(session, index);
-        if (stats == null) {
-            return null;
-        }
-        Histogram fromHistogram = stats.getHistogram(0, index.getKeyColumns().size());
-        if (fromHistogram == null) {
-            return null;
-        }
-        IndexDef indexDef = index.indexDef();
-        RowDef indexRowDef = indexDef.getRowDef();
-        TableStatistics.Histogram toHistogram = new TableStatistics.Histogram(index.getIndexId());
-        Key key = treeService.createKey();
-        RowData indexRowData = new RowData(new byte[4096]);
-        Object[] indexValues = new Object[indexRowDef.getFieldCount()];
-        long count = 0;
-        for (HistogramEntry entry : fromHistogram.getEntries()) {
-            // Decode the key.
-            int keylen = entry.getKeyBytes().length;
-            System.arraycopy(entry.getKeyBytes(), 0, key.getEncodedBytes(), 0, keylen);
-            key.setEncodedSize(keylen);
-            key.indexTo(0);
-            int depth = key.getDepth();
-            // Copy key fields to index row.
-            for (int field : indexDef.getFields()) {
-                if (--depth >= 0) {
-                    indexValues[field] = key.decode();
-                } else {
-                    indexValues[field] = null;
-                }
-            }
-            indexRowData.createRow(indexRowDef, indexValues);
-            // Partial counts to running total less than key.
-            count += entry.getLessCount();
-            toHistogram.addSample(new TableStatistics.HistogramSample(indexRowData.copy(),
-                                                                      count));
-            count += entry.getEqualCount();
-        }
-        // Add final entry with all nulls.
-        Arrays.fill(indexValues, null);
-        indexRowData.createRow(indexRowDef, indexValues);
-        toHistogram.addSample(new TableStatistics.HistogramSample(indexRowData.copy(),
-                                                                  count));
-        return toHistogram;
-    }
-
-    boolean hasNullIndexSegments(RowData rowData, Index index)
-    {
-        IndexDef indexDef = index.indexDef();
-        assert indexDef.getRowDef().getRowDefId() == rowData.getRowDefId();
-        for (int i : indexDef.getFields()) {
-            if (rowData.isNull(i)) {
-                return true;
-            }
-        }
-        return false;
+            if(index.isGroupIndex()) {
+                try {
+                    Tree tree = index.indexDef().getTreeCache().getTree();
+                    new AccumulatorAdapter(AccumulatorAdapter.AccumInfo.ROW_COUNT, tree).set(0);
+                } catch(PersistitInterruptedException e) {
+                    throw PersistitAdapter.wrapPersistitException(session, e);
+                }
+            }
+        }
     }
 
     private void checkNotGroupIndex(Index index) {
@@ -1607,24 +1265,13 @@
     {
         checkNotGroupIndex(index);
         Exchange iEx = getExchange(session, index);
-        constructIndexRow(iEx, rowData, index, hkey, indexRow, true);
-        checkUniqueness(index, rowData, iEx);
-        if (deferIndexes) {
-            // TODO: bug767737, deferred indexing does not handle uniqueness
-            synchronized (deferredIndexKeys) {
-                SortedSet<KeyState> keySet = deferredIndexKeys.get(iEx.getTree());
-                if (keySet == null) {
-                    keySet = new TreeSet<>();
-                    deferredIndexKeys.put(iEx.getTree(), keySet);
-                }
-                KeyState ks = new KeyState(iEx.getKey());
-                keySet.add(ks);
-                deferredIndexKeyLimit -= (ks.getBytes().length + KEY_STATE_SIZE_OVERHEAD);
-            }
-        } else {
+        try {
+            constructIndexRow(iEx, rowData, index, hkey, indexRow, true);
+            checkUniqueness(index, rowData, iEx);
             storageAction.store(iEx);
+        } finally {
+            releaseExchange(session, iEx);
         }
-        releaseExchange(session, iEx);
     }
 
     private void checkUniqueness(Index index, RowData rowData, Exchange iEx) throws PersistitException
@@ -1661,22 +1308,6 @@
         return keyExistsInIndex;
     }
 
-    private void putAllDeferredIndexKeys(final Session session) {
-        synchronized (deferredIndexKeys) {
-            for (final Map.Entry<Tree, SortedSet<KeyState>> entry : deferredIndexKeys
-                    .entrySet()) {
-                final Exchange iEx = treeService.getExchange(session, entry.getKey());
-                try {
-                    buildIndexAddKeys(entry.getValue(), iEx);
-                    entry.getValue().clear();
-                } finally {
-                    treeService.releaseExchange(session, iEx);
-                }
-            }
-            deferredIndexKeyLimit = MAX_INDEX_TRANCHE_SIZE;
-        }
-    }
-
     private void updateIndex(Session session,
                              Index index,
                              RowDef rowDef,
@@ -1761,47 +1392,11 @@
         releaseExchange(session, iEx);
     }
 
-    static boolean bytesEqual(byte[] a, int aoffset, int asize,
-                              byte[] b, int boffset, int bsize) {
-        if (asize != bsize) {
-            return false;
-        }
-        for (int i = 0; i < asize; i++) {
-            if (a[i + aoffset] != b[i + boffset]) {
-                return false;
-            }
-        }
-        return true;
-    }
-
-    public static boolean fieldsEqual(RowDef rowDef, RowData a, RowData b, int[] fieldIndexes)
-    {
-        for (int fieldIndex : fieldIndexes) {
-            long aloc = rowDef.fieldLocation(a, fieldIndex);
-            long bloc = rowDef.fieldLocation(b, fieldIndex);
-            if (!bytesEqual(a.getBytes(), (int) aloc, (int) (aloc >>> 32),
-                            b.getBytes(), (int) bloc, (int) (bloc >>> 32))) {
-                return false;
-            }
-        }
-        return true;
-    }
-
-    public static boolean fieldEqual(RowDef rowDef, RowData a, RowData b, int fieldPosition)
-    {
-        long aloc = rowDef.fieldLocation(a, fieldPosition);
-        long bloc = rowDef.fieldLocation(b, fieldPosition);
-        return bytesEqual(a.getBytes(), (int) aloc, (int) (aloc >>> 32),
-                          b.getBytes(), (int) bloc, (int) (bloc >>> 32));
-    }
-
-    public void packRowData(final Exchange hEx, final RowDef rowDef,
-            final RowData rowData) {
-        packRowData(hEx.getValue(), rowDef, rowData);
-    }
-
-    public void packRowData(final Value value, final RowDef rowDef,
-                            final RowData rowData) {
+    public void packRowData(final Exchange hEx,final RowData rowData) {
+        packRowData(hEx.getValue(), rowData);
+    }
+
+    public void packRowData(final Value value, final RowData rowData) {
         value.directPut(valueCoder, rowData, null);
     }
 
@@ -1818,8 +1413,8 @@
         // rowData.prepareRow(0);
     }
 
+    @Override
     public void buildIndexes(Session session, Collection<? extends Index> indexes, boolean defer) {
-        flushIndexes(session);
         Set<Group> groups = new HashSet<>();
         Map<Integer,RowDef> userRowDefs = new HashMap<>();
         Set<Index> indexesToBuild = new HashSet<>();
@@ -1852,103 +1447,37 @@
                                 indexKeyCount++;
                             }
                         }
-                        if (deferredIndexKeyLimit <= 0) {
-                            putAllDeferredIndexKeys(session);
-                        }
                     }
                 }
             } catch (PersistitException e) {
                 throw new PersistitAdapterException(e);
             }
-            flushIndexes(session);
             LOG.debug("Inserted {} index keys into group {}", indexKeyCount, group.getName());
         }
     }
 
     @Override
-    public void removeTrees(Session session, Collection<? extends TreeLink> treeLinks) {
-        try {
-            for(TreeLink link : treeLinks) {
-                if(!schemaManager.treeRemovalIsDelayed()) {
-                    Exchange ex = treeService.getExchange(session, link);
-                    ex.removeTree();
-                    // Do not releaseExchange, causes caching and leak for now unused tree
-                }
-                schemaManager.treeWasRemoved(session, link.getSchemaName(), link.getTreeName());
-            }
-        } catch (PersistitException e) {
-            LOG.debug("Exception removing tree from Persistit", e);
-            throw new PersistitAdapterException(e);
+    public void removeTrees(Session session, UserTable table) {
+        super.removeTrees(session, table);
+
+        // TODO: Generalize. Knowing about FullTextService is wrong.
+        for(FullTextIndex idx : table.getOwnFullTextIndexes()) {
+            fullTextService.dropIndex(session, idx);
         }
     }
 
     @Override
-    public void removeTrees(Session session, Table table) {
-        Collection<TreeLink> treeLinks = new ArrayList<>();
-
-        // delete all fulltext indexes
-        if (table.isUserTable())
-        {
-            for (FullTextIndex idx : ((UserTable)table).getOwnFullTextIndexes())
-                fullTextService.dropIndex(session, idx);
-        }
-
-        // Add all index trees
-        final Collection<TableIndex> tableIndexes = table.isUserTable() ? ((UserTable)table).getIndexesIncludingInternal() : table.getIndexes();
-        final Collection<GroupIndex> groupIndexes = table.getGroupIndexes();
-        for(Index index : tableIndexes) {
-            treeLinks.add(index.indexDef());
-        }
-        for(Index index : groupIndexes) {
-            treeLinks.add(index.indexDef());
-        }
-        // Drop the sequence trees too. 
-        if (table.isUserTable() && ((UserTable)table).getIdentityColumn() != null) {
-            treeLinks.add(((UserTable)table).getIdentityColumn().getIdentityGenerator());
-        } else if (table.isGroupTable()) {
-            for (UserTable userTable : table.getAIS().getUserTables().values()) {
-                if (userTable.getGroup() == table.getGroup() &&
-                        userTable.getIdentityColumn() != null) {
-                    treeLinks.add(userTable.getIdentityColumn().getIdentityGenerator());
-                }
-            }
-        }
-        
-        // And the group tree
-        treeLinks.add(table.getGroup());
-        // And drop them all
-        removeTrees(session, treeLinks);
-        indexStatistics.deleteIndexStatistics(session, tableIndexes);
-        indexStatistics.deleteIndexStatistics(session, groupIndexes);
-    }
-
-    public void flushIndexes(final Session session) {
-        try {
-            putAllDeferredIndexKeys(session);
-        } catch (PersistitAdapterException e) {
-            LOG.debug("Exception while trying to flush deferred index keys", e);
-            throw e;
-        }
-    }
-
     public void deleteIndexes(final Session session, final Collection<? extends Index> indexes) {
-        List<TreeLink> links = new ArrayList<>(indexes.size());
+        super.deleteIndexes(session, indexes);
+        // TODO: Generalize. Knowing about FullTextService is wrong.
         for(Index index : indexes) {
             // no trees to drop
             if (index.getIndexType() == IndexType.FULL_TEXT)
             {
                 fullTextService.dropIndex(session, (FullTextIndex)index);
                 indexes.remove(index);
-                continue;
-            }
-            final IndexDef indexDef = index.indexDef();
-            if(indexDef == null) {
-                throw new IllegalStateException("indexDef is null for index: " + index);
-            }
-            links.add(indexDef);
+            }
         }
-        removeTrees(session, links);
-        indexStatistics.deleteIndexStatistics(session, indexes);
     }
     
     @Override
@@ -1956,28 +1485,6 @@
         removeTrees(session, sequences);
     }
 
-    private void buildIndexAddKeys(final SortedSet<KeyState> keys,
-            final Exchange iEx) {
-        final long start = System.nanoTime();
-        try {
-            for (final KeyState keyState : keys) {
-                keyState.copyTo(iEx.getKey());
-                iEx.store();
-            }
-        } catch (PersistitException e) {
-            LOG.error(e.getMessage());
-            throw new PersistitAdapterException(e);
-        }
-        final long elapsed = System.nanoTime() - start;
-        if (LOG.isInfoEnabled()) {
-            LOG.debug("Index builder inserted {} keys into index tree {} in {} seconds", new Object[]{
-                    keys.size(),
-                    iEx.getTree().getName(),
-                    elapsed / 1000000000
-            });
-        }
-    }
-
     private RowData mergeRows(RowDef rowDef, RowData currentRow, RowData newRowData, ColumnSelector columnSelector) {
         NewRow mergedRow = NiceRow.fromRowData(currentRow, rowDef);
         NewRow newRow = new LegacyRowWrapper(rowDef, newRowData);
@@ -1990,24 +1497,6 @@
         return mergedRow.toRowData();
     }
 
-    private RowDef rowDefFromExplicitOrId(Session session, RowData rowData) {
-        RowDef rowDef = rowData.getExplicitRowDef();
-        if(rowDef == null) {
-            rowDef = getRowDef(session, rowData.getRowDefId());
-        }
-        return rowDef;
-    }
-
-    @Override
-    public boolean isDeferIndexes() {
-        return deferIndexes;
-    }
-
-    @Override
-    public void setDeferIndexes(final boolean defer) {
-        deferIndexes = defer;
-    }
-
     private void lockAndCheckVersion(Session session, RowDef rowDef) {
         final LockService.Mode mode = LockService.Mode.SHARED;
         final int tableID = rowDef.getRowDefId();
@@ -2070,24 +1559,11 @@
         return visitor;
     }
 
-    public TableStatus getTableStatus(Table table) {
-        TableStatus ts = null;
-        if(table.rowDef() != null) {
-            ts = table.rowDef().getTableStatus();
-        }
-        return ts;
-    }
-
     private static PersistitAdapter adapter(Session session)
     {
         return (PersistitAdapter) session.get(StoreAdapter.STORE_ADAPTER_KEY);
     }
 
-    private static boolean hasChildren(UserTable table) {
-        // At runtime, getCandidateChildJoins() = getChildJoins() and doesn't involve building a temp list
-        return !table.getCandidateChildJoins().isEmpty();
-    }
-
     private TreeBuilder createTreeBuilder(String name, float bufferBoolFraction) {
         TreeBuilder tb = new TreeBuilder(getDb(), name, -1, bufferBoolFraction)
         // TODO: throw an Akiban dup-key exception once we can handle them
@@ -2153,6 +1629,38 @@
         lockKeyAppender.clear();
     }
 
+    @Override
+    public StoreAdapter createAdapter(Session session, Schema schema) {
+        return new PersistitAdapter(schema, this, treeService, session, config);
+    }
+
+    @Override
+    public void truncateTree(Session session, TreeLink treeLink) {
+        Exchange iEx = treeService.getExchange(session, treeLink);
+        try {
+            iEx.removeAll();
+        } catch (PersistitException e) {
+            throw PersistitAdapter.wrapPersistitException(session, e);
+        } finally {
+            releaseExchange(session, iEx);
+        }
+    }
+
+    @Override
+    public void removeTree(Session session, TreeLink treeLink) {
+        try {
+            if(!schemaManager.treeRemovalIsDelayed()) {
+                Exchange ex = treeService.getExchange(session, treeLink);
+                ex.removeTree();
+                // Do not releaseExchange, causes caching and leak for now unused tree
+            }
+            schemaManager.treeWasRemoved(session, treeLink.getSchemaName(), treeLink.getTreeName());
+        } catch (PersistitException e) {
+            LOG.debug("Exception removing tree from Persistit", e);
+            throw PersistitAdapter.wrapPersistitException(session, e);
+        }
+    }
+
     private class Bulkload {
 
         Bulkload(final Persistit persistit, AkibanInformationSchema ais) {

=== modified file 'src/main/java/com/akiban/server/store/PersistitStoreSchemaManager.java'
--- src/main/java/com/akiban/server/store/PersistitStoreSchemaManager.java	2013-04-11 18:35:20 +0000
+++ src/main/java/com/akiban/server/store/PersistitStoreSchemaManager.java	2013-05-20 14:47:34 +0000
@@ -474,11 +474,7 @@
         for(ChangedTableDescription desc : alteredTables) {
             if(desc.isNewGroup()) {
                 UserTable oldTable = oldAIS.getUserTable(desc.getOldName());
-                try {
-                    oldTable.rowDef().getTableStatus().setOrdinal(0);
-                } catch(PersistitException e) {
-                    throw wrapPersistitException(session, e);
-                }
+                oldTable.rowDef().getTableStatus().setOrdinal(0);
             }
         }
 

=== modified file 'src/main/java/com/akiban/server/store/Store.java'
--- src/main/java/com/akiban/server/store/Store.java	2013-04-14 00:26:23 +0000
+++ src/main/java/com/akiban/server/store/Store.java	2013-05-20 14:47:34 +0000
@@ -17,57 +17,40 @@
 
 package com.akiban.server.store;
 
+import com.akiban.ais.model.AkibanInformationSchema;
 import com.akiban.ais.model.Group;
 import com.akiban.ais.model.Index;
 import com.akiban.ais.model.Sequence;
-import com.akiban.ais.model.Table;
+import com.akiban.ais.model.TableName;
+import com.akiban.ais.model.UserTable;
+import com.akiban.qp.operator.StoreAdapter;
+import com.akiban.qp.rowtype.Schema;
 import com.akiban.server.TableStatistics;
 import com.akiban.server.api.dml.ColumnSelector;
 import com.akiban.server.api.dml.scan.ScanLimit;
 import com.akiban.server.rowdata.RowData;
 import com.akiban.server.rowdata.RowDef;
 import com.akiban.server.service.session.Session;
+import com.akiban.server.service.tree.KeyCreator;
 import com.akiban.server.service.tree.TreeLink;
-import com.persistit.exception.PersistitException;
-import com.persistit.exception.RollbackException;
+import com.akiban.server.store.statistics.IndexStatisticsService;
 
 import java.util.Collection;
 
-/**
- * An abstraction for a layer that stores and retrieves data
- *
- * @author peter
- *
- */
-public interface Store {
+public interface Store extends KeyCreator {
 
     /** Get the RowDef for the given ID. Note, a transaction should be active before calling this. */
     RowDef getRowDef(Session session, int rowDefID);
-
-    void writeRow(Session session, RowData rowData) throws PersistitException;
-
-    void deleteRow(Session session, RowData rowData, boolean deleteIndexes, boolean cascadeDelete) throws PersistitException;
+    RowDef getRowDef(Session session, TableName tableName);
+    AkibanInformationSchema getAIS(Session session);
+
+    void writeRow(Session session, RowData rowData);
+
+    void deleteRow(Session session, RowData rowData, boolean deleteIndexes, boolean cascadeDelete);
 
     void updateRow(Session session, RowData oldRowData,
                    RowData newRowData,
-                   ColumnSelector columnSelector, Index[] indexes) throws PersistitException;
-
-    /**
-     * See {@link #newRowCollector(Session, int, int, int, byte[], RowData, ColumnSelector, RowData, ColumnSelector, ScanLimit)}
-     * for parameter descriptions.
-     * @throws Exception 
-     *
-     * @deprecated This constructor is ambiguous and may not return the expected rows. Fields from <code>start</code>
-     * and <code>end</code> that are <code>NULL</code> are considered to be <b>unset</b>.
-     */
-    RowCollector newRowCollector(Session session,
-                                 int rowDefId,
-                                 int indexId,
-                                 int scanFlags,
-                                 RowData start,
-                                 RowData end,
-                                 byte[] columnBitMap,
-                                 ScanLimit scanLimit);
+                   ColumnSelector columnSelector, Index[] indexes);
 
     /**
      * Create a new RowCollector.
@@ -134,13 +117,10 @@
      * Truncate the given group. This includes indexes from all tables, group
      * indexes, the group itself, and all table statuses.
      */
-    void truncateGroup(Session session, Group group) throws PersistitException;
-
-    void truncateTableStatus(Session session, int rowDefId) throws RollbackException, PersistitException;
-
-    boolean isDeferIndexes();
-    void setDeferIndexes(boolean b);
-    void flushIndexes(Session session);
+    void truncateGroup(Session session, Group group);
+
+    void truncateTableStatus(Session session, int rowDefId);
+
     void deleteIndexes(Session session, Collection<? extends Index> indexes);
     void buildIndexes(Session session, Collection<? extends Index> indexes, boolean deferIndexes);
 
@@ -149,10 +129,11 @@
      * Remove all trees, and their contents, associated with the given table.
      * @param session Session
      * @param table Table
-     * @throws PersistitException 
-     * @throws Exception 
+     * @throws Exception
      */
-    void removeTrees(Session session, Table table);
+    void removeTrees(Session session, UserTable table);
+    void removeTree(Session session, TreeLink treeLink);
+    void truncateTree(Session session, TreeLink treeLink);
 
     /**
      * Low level operation. Removes the given trees and <i>only</i> the given trees.
@@ -171,4 +152,8 @@
     void finishBulkLoad(Session session);
 
     boolean isBulkloading();
+
+    void setIndexStatistics(IndexStatisticsService indexStatistics);
+
+    StoreAdapter createAdapter(Session session, Schema schema);
 }

=== modified file 'src/main/java/com/akiban/server/store/statistics/IndexStatisticsServiceImpl.java'
--- src/main/java/com/akiban/server/store/statistics/IndexStatisticsServiceImpl.java	2013-04-11 05:51:16 +0000
+++ src/main/java/com/akiban/server/store/statistics/IndexStatisticsServiceImpl.java	2013-05-20 14:47:34 +0000
@@ -146,7 +146,7 @@
             if (table.hasMemoryTableFactory()) {
                 return table.getMemoryTableFactory().rowCount();
             } else {
-                return store.getTableStatus(table).getRowCount();
+                return table.rowDef().getTableStatus().getRowCount();
             }
         }
         final Exchange ex = store.getExchange(session, index);
@@ -161,7 +161,7 @@
     @Override
     public long countEntriesApproximate(Session session, Index index) {
         if (index.isTableIndex()) {
-            return store.getTableStatus(((TableIndex)index).getTable()).getApproximateRowCount();
+            return index.leafMostTable().rowDef().getTableStatus().getApproximateRowCount();
         }
         final Exchange ex = store.getExchange(session, index);
         try {

=== modified file 'src/main/java/com/akiban/server/store/statistics/PersistitStoreIndexStatistics.java'
--- src/main/java/com/akiban/server/store/statistics/PersistitStoreIndexStatistics.java	2013-03-22 20:05:57 +0000
+++ src/main/java/com/akiban/server/store/statistics/PersistitStoreIndexStatistics.java	2013-05-20 14:47:34 +0000
@@ -284,6 +284,9 @@
         RowData rowData = new RowData(new byte[INITIAL_ROW_SIZE]);
         RowDef indexStatisticsRowDef = getIndexStatsRowDef(session);
         RowDef indexStatisticsEntryRowDef = getIndexStatsEntryRowDef(session);
+        if(index.indexDef() == null) {
+            return;
+        }
         int tableId = index.indexDef().getRowDef().getRowDefId();
         int indexId = index.getIndexId();
         // Delete index_statistics_entry rows.

=== modified file 'src/test/java/com/akiban/server/test/it/keyupdate/FixCountStarIT.java'
--- src/test/java/com/akiban/server/test/it/keyupdate/FixCountStarIT.java	2013-04-11 05:51:16 +0000
+++ src/test/java/com/akiban/server/test/it/keyupdate/FixCountStarIT.java	2013-05-20 14:47:34 +0000
@@ -95,7 +95,7 @@
             public void run() {
                 if (index.isTableIndex()) {
                     TableIndex tIndex = (TableIndex) index;
-                    store().getPersistitStore().getTableStatus(tIndex.getTable()).setRowCount(newVal);
+                    tIndex.leafMostTable().rowDef().getTableStatus().setRowCount(newVal);
                 }
                 else {
                     PersistitStore store = store().getPersistitStore();

=== modified file 'src/test/java/com/akiban/server/test/it/rowtests/ObjectToKeyIT.java'
--- src/test/java/com/akiban/server/test/it/rowtests/ObjectToKeyIT.java	2013-03-22 20:05:57 +0000
+++ src/test/java/com/akiban/server/test/it/rowtests/ObjectToKeyIT.java	2013-05-20 14:47:34 +0000
@@ -38,7 +38,7 @@
     }
 
     private void testObjectToKey(FieldDef field, Object... testValues) throws PersistitException {
-        Key key = persistitStore().getKey();
+        Key key = persistitStore().createKey();
         PersistitKeyAppender appender = PersistitKeyAppender.create(key);
         for(Object inObj : testValues) {
             key.clear();

=== modified file 'src/test/java/com/akiban/server/test/it/store/AbstractScanBase.java'
--- src/test/java/com/akiban/server/test/it/store/AbstractScanBase.java	2013-03-27 04:42:10 +0000
+++ src/test/java/com/akiban/server/test/it/store/AbstractScanBase.java	2013-05-20 14:47:34 +0000
@@ -139,8 +139,8 @@
             final byte[] columnBitMap, final int indexId) throws Exception {
         int scanCount = 0;
         result.clear();
-        final RowCollector rc = store().newRowCollector(session(), rowDefId, indexId,
-                scanFlags, start, end, columnBitMap, null);
+        final RowCollector rc = store().newRowCollector(session(), scanFlags, rowDefId, indexId, columnBitMap,
+                                                        start, null, end, null, null);
         if (VERBOSE) {
             System.out.println("Test " + test);
         }