Merge lp:~oontvoo/akiban-server/bug_drop_fulltext_index into lp:~akiban-technologies/akiban-server/trunk

[Vy Nguyen]

fix drop fulltext index bug (bug 1163026)

[Nathan Williams (nwilliams)]

What is the IndexDef needed for?

I'm guessing bug1154336 is quite similar too. Perhaps dropTableInternal() needs to be in on the game.

review: Needs Information

[Vy Nguyen (oontvoo)]

PersisitStore.java - line 1554 (+/- a few lines)

It looks like the index needs to be 'attached' to an IndexDef.

[Nathan Williams (nwilliams)]

Only to delete the tree, which full texts don't have.

DDLFunctions has separate methods for dropping table and group indexes. Something different needs to happen for full texts (i.e. FullTextIndexService#dropIndex), so another entry point would be consistent and make sense.

[Nathan Williams (nwilliams)]

Actually, handling the distinction in PS#deleteIndexes() would let dropTableInternal() use it as well, so I'm fine with that too.

[lp:~oontvoo/akiban-server/bug_drop_fulltext_index]

2621. By Vy Nguyen on 2013-04-12

skip fulltext index in persistitStore.deleteIndexes()

[Vy Nguyen (oontvoo)]

skip fulltext index in persistitStore.deleteIndexes()

review: Needs Resubmitting

[Nathan Williams (nwilliams)]

Since we're here, can you please fix the other bug I mentioned. We just need to call FTIS#dropIndex() for them.

review: Needs Fixing

[lp:~oontvoo/akiban-server/bug_drop_fulltext_index]

2622. By Vy Nguyen on 2013-04-12

pass FullTextIndexService to PersistitStore for deleteIndexes() to use

2623. By Vy Nguyen on 2013-04-12

bind FTS to FTSImpl

[Vy Nguyen (oontvoo)]

I am having a compilation-error in

src/test/java/com/akiban/server/test/FailureOnStartupIT.java:[85,12] error: constructor OperatorStore in class OperatorStore cannot be applied to given types;

(I declared FTS as an argument in PersistStore's ctor, and other places where it's called.)

What did I miss?

review: Needs Information

[Vy Nguyen (oontvoo)]

Hm, never mind, found it!

[Nathan Williams (nwilliams)]

Look at FailureOnStartupIT.java, line 85. It extends OperatorStore so its constructor needs adjusted too.

[lp:~oontvoo/akiban-server/bug_drop_fulltext_index]

2624. By Vy Nguyen on 2013-04-12

have FullTextIndexService attache itself to PersistStore rather than passing it to the constructor because that'd be circular dependency

2625. By Vy Nguyen on 2013-04-14

revert back to 2621

[Vy Nguyen (oontvoo)]

Actually, it's more complicated than just calling FTS.dropIndex()

+ By the time deleteIndexes() is called, the index is no longer in AIS for FullTextIndexShared.forAIS() to look up. But that's not the 'complicated' part.

+ In order for dropIndex() to work, the index needs to be in 'indexes' map, which none of the current fulltext index is because no one puts it in there. The createIndex() method, which would do this, needs to be called manually.
This problem is solved in the do-maintenance branch. And I can certainly have it fix the bug you mentioned. (In fact, I tried merging this branch into it and the delete works correctly, which it wouldn't in this branch).

review: Needs Resubmitting

[Nathan Williams (nwilliams)]

> By the time deleteIndexes() is called, the index is no
> longer in AIS for FullTextIndexShared.forAIS() to look up

It does not need to be in the current AIS and probably shouldn't be, given that it's being removed. Look at the flow of the other drop methods (e.g. table, index). The AIS changes are performed through the SchemaManager before tree deletion is done. You can either hang onto the 'old' AIS or get it through the Index reference.

> In order for dropIndex() to work, the index needs to be in 'indexes' map,

It should be, and sounds like will be once your other branch is in, but doesn't need to be. The getIndex() method will populate the indexes map on demand. It creates a value cached on the AIS, which is going away in this case, but will work just fine.

review: Needs Fixing

[lp:~oontvoo/akiban-server/bug_drop_fulltext_index]

2626. By Vy Nguyen on 2013-04-15

fix bug1154336

2627. By Vy Nguyen on 2013-04-15

remove formatting changes

2628. By Vy Nguyen on 2013-04-15

clean up

2629. By Vy Nguyen on 2013-04-15

2630. By Vy Nguyen on 2013-04-15

more tests

2631. By Vy Nguyen on 2013-04-15

require FulLTetIndexService for testing

2632. By Vy Nguyen on 2013-04-15

basicddl

2633. By Vy Nguyen on 2013-04-15

add tests on droping table with ftindex. removeTrees should also delete fulltext index, even if these don't really have a tree

2634. By Vy Nguyen on 2013-04-15

add the original testcase from the bug report

2635. By Vy Nguyen on 2013-04-15

2636. By Vy Nguyen on 2013-04-17

merge trunk

2637. By Vy Nguyen on 2013-04-17

use wait() in test instead of blindly sleep for a long time

2638. By Vy Nguyen on 2013-04-22

merge trunk, move configs to TestConfigService

2639. By Vy Nguyen on 2013-04-22

edge cases in removeTrees()

2640. By Vy Nguyen on 2013-04-22

remove unused imports in FullTextIndexServiceIT

2641. By Vy Nguyen on 2013-04-22

use getownFullTextIndex

2642. By Vy Nguyen on 2013-04-22

clean up

2643. By Vy Nguyen on 2013-04-22

clean up

2644. By Vy Nguyen on 2013-04-22

pass Index as an argument to dropIndex to correctly determine the treename

2645. By Vy Nguyen on 2013-04-22

delete fulltext index

2646. By Vy Nguyen on 2013-04-26

use default settings in OptimizerTestBase

2647. By Vy Nguyen on 2013-04-29

use the old AIS when neccessary

2648. By Vy Nguyen on 2013-04-29

remove unused import accidentally added

Unmerged revisions

=== modified file 'src/main/java/com/akiban/rest/resources/FullTextResource.java'
--- src/main/java/com/akiban/rest/resources/FullTextResource.java	2013-03-22 20:05:57 +0000
+++ src/main/java/com/akiban/rest/resources/FullTextResource.java	2013-04-15 17:23:26 +0000
@@ -30,7 +30,6 @@
 import javax.ws.rs.QueryParam;
 import javax.ws.rs.core.Context;
 import javax.ws.rs.core.Response;
-import javax.ws.rs.core.UriInfo;
 import java.io.PrintWriter;
 
 import static com.akiban.rest.resources.ResourceHelper.MEDIATYPE_JSON_JAVASCRIPT;
@@ -66,22 +65,4 @@
                 })
                 .build();
     }
-
-    @POST
-    @Produces(MEDIATYPE_JSON_JAVASCRIPT)
-    public Response textSearch(@Context final HttpServletRequest request,
-                               @PathParam("table") String table,
-                               @PathParam("index") String index) throws Exception {
-        final IndexName indexName = new IndexName(ResourceHelper.parseTableName(request, table), index);
-        ResourceHelper.checkSchemaAccessible(reqs.securityService, request, indexName.getSchemaName());
-        return RestResponseBuilder
-                .forRequest(request)
-                .body(new RestResponseBuilder.BodyGenerator() {
-                    @Override
-                    public void write(PrintWriter writer) throws Exception {
-                        reqs.restDMLService.refreshFullTextIndex(writer, indexName);
-                    }
-                })
-                .build();
-    }
 }

=== modified file 'src/main/java/com/akiban/server/rowdata/RowDefCache.java'
--- src/main/java/com/akiban/server/rowdata/RowDefCache.java	2013-03-22 20:05:57 +0000
+++ src/main/java/com/akiban/server/rowdata/RowDefCache.java	2013-04-15 17:23:26 +0000
@@ -38,6 +38,7 @@
 
 import com.akiban.ais.model.AkibanInformationSchema;
 import com.akiban.ais.model.Column;
+import com.akiban.ais.model.FullTextIndex;
 import com.akiban.ais.model.IndexColumn;
 import com.akiban.ais.model.Join;
 import com.akiban.ais.model.JoinColumn;
@@ -184,6 +185,7 @@
         for (TableIndex index : table.getIndexesIncludingInternal()) {
             List<IndexColumn> indexColumns = index.getKeyColumns();
             if(!indexColumns.isEmpty()) {
+                
                 new IndexDef(rowDef, index);
                 if (index.isPrimaryKey()) {
                     indexList.add(0, index);
@@ -193,7 +195,7 @@
             }
             //else Don't create IndexDef for empty, autogenerated indexes
         }
-
+        
         // Group indexes
         final List<GroupIndex> groupIndexList = new ArrayList<>();
         for (GroupIndex index : table.getGroupIndexes()) {

=== modified file 'src/main/java/com/akiban/server/service/dxl/BasicDDLFunctions.java'
--- src/main/java/com/akiban/server/service/dxl/BasicDDLFunctions.java	2013-04-11 18:35:20 +0000
+++ src/main/java/com/akiban/server/service/dxl/BasicDDLFunctions.java	2013-04-15 17:23:26 +0000
@@ -1031,7 +1031,14 @@
             for(String indexName : indexNamesToDrop) {
                 Index index = table.getIndex(indexName);
                 if(index == null) {
-                    throw new NoSuchIndexException (indexName);
+                    if (table instanceof UserTable)
+                    {
+                        index = ((UserTable)table).getFullTextIndex(indexName);
+                        if (index == null)
+                            throw new NoSuchIndexException (indexName);
+                    }
+                    else
+                        throw new NoSuchIndexException (indexName);
                 }
                 if(index.isPrimaryKey()) {
                     throw new ProtectedIndexException(indexName, table.getName());

=== modified file 'src/main/java/com/akiban/server/service/restdml/RestDMLService.java'
--- src/main/java/com/akiban/server/service/restdml/RestDMLService.java	2013-04-15 14:38:06 +0000
+++ src/main/java/com/akiban/server/service/restdml/RestDMLService.java	2013-04-15 17:23:26 +0000
@@ -51,8 +51,6 @@
                               TableName procName, Map<String,List<String>> queryParams, String content) throws SQLException;
 
     public void fullTextSearch(PrintWriter writer, IndexName indexName, Integer depth, String query, Integer limit);
-    // TODO: Temporary.
-    public void refreshFullTextIndex(PrintWriter writer, IndexName indexName);
 
     public String jonquilToSQL(TableName tableName, String jonquil) throws IOException;
 }

=== modified file 'src/main/java/com/akiban/server/service/restdml/RestDMLServiceImpl.java'
--- src/main/java/com/akiban/server/service/restdml/RestDMLServiceImpl.java	2013-04-15 14:38:06 +0000
+++ src/main/java/com/akiban/server/service/restdml/RestDMLServiceImpl.java	2013-04-15 17:23:26 +0000
@@ -649,14 +649,4 @@
                                             true);
         }
     }
-
-    // TODO: Temporary.
-    public void refreshFullTextIndex(PrintWriter writer, IndexName indexName) {
-        long count;
-        try (Session session = sessionService.createSession()) {
-            count = fullTextService.createIndex(session, indexName);
-        }
-        writer.write(String.format("{\"count\":%d}", count));
-    }
-
 }

=== modified file 'src/main/java/com/akiban/server/service/text/FullTextCursor.java'
--- src/main/java/com/akiban/server/service/text/FullTextCursor.java	2013-03-22 20:05:57 +0000
+++ src/main/java/com/akiban/server/service/text/FullTextCursor.java	2013-04-15 17:23:26 +0000
@@ -103,7 +103,7 @@
         catch (IOException ex) {
             throw new AkibanInternalException("Error reading document", ex);
         }
-        HKey hkey = hkey(doc.getBinaryValue(IndexedField.KEY_FIELD));
+        HKey hkey = hkey(doc.get(IndexedField.KEY_FIELD));
         Row row = new HKeyRow(rowType, hkey, hKeyCache);
         logger.debug("FullTextCursor: yield {}", row);
         return row;
@@ -152,11 +152,12 @@
 
     /* Allocate a new <code>PersistitHKey</code> and copy the given
      * key bytes into it. */
-    protected HKey hkey(BytesRef keyBytes) {
+    protected HKey hkey(String encoded) {
         PersistitHKey hkey = (PersistitHKey)context.getStore().newHKey(rowType.hKey());
         Key key = hkey.key();
-        key.setEncodedSize(keyBytes.length);
-        System.arraycopy(keyBytes.bytes, keyBytes.offset, key.getEncodedBytes(), 0, keyBytes.length);
+        byte decodedBytes[] = RowIndexer.decodeString(encoded);
+        key.setEncodedSize(decodedBytes.length);
+        System.arraycopy(decodedBytes, 0, key.getEncodedBytes(), 0, decodedBytes.length);
         return hkey;
     }
 

=== modified file 'src/main/java/com/akiban/server/service/text/FullTextIndexInfo.java'
--- src/main/java/com/akiban/server/service/text/FullTextIndexInfo.java	2013-03-22 20:05:57 +0000
+++ src/main/java/com/akiban/server/service/text/FullTextIndexInfo.java	2013-04-15 17:23:26 +0000
@@ -20,12 +20,16 @@
 import com.akiban.ais.model.AkibanInformationSchema;
 import com.akiban.ais.model.Column;
 import com.akiban.ais.model.FullTextIndex;
+import com.akiban.ais.model.Group;
 import com.akiban.ais.model.IndexColumn;
 import com.akiban.ais.model.IndexName;
 import com.akiban.ais.model.UserTable;
+import com.akiban.qp.expression.IndexKeyRange;
 import com.akiban.qp.operator.API;
 import com.akiban.qp.operator.Operator;
+import com.akiban.qp.row.HKeyRow;
 import com.akiban.qp.rowtype.HKeyRowType;
+import com.akiban.qp.rowtype.IndexRowType;
 import com.akiban.qp.rowtype.RowType;
 import com.akiban.qp.rowtype.Schema;
 import com.akiban.qp.rowtype.UserTableRowType;
@@ -50,7 +54,8 @@
     private Map<Column,IndexedField> fieldsByColumn;
     private Map<RowType,List<IndexedField>> fieldsByRowType;
     private String defaultFieldName;
-
+    private Operator plan;
+    
     public FullTextIndexInfo(FullTextIndexShared shared) {
         this.shared = shared;
     }
@@ -63,7 +68,9 @@
         }
         index = table.getFullTextIndex(name.getName());
         if (index == null) {
-            throw new NoSuchIndexException(name.getName());
+            NoSuchIndexException ret =  new NoSuchIndexException(name.getName());
+            ret.printStackTrace();
+            throw ret;
         }
         schema = SchemaCache.globalSchema(ais);
         indexedRowType = schema.userTableRowType(table);
@@ -87,6 +94,8 @@
             }
             fields.add(entry.getValue());
         }
+        
+        plan = computePlan();
     }
 
     public FullTextIndex getIndex() {
@@ -139,6 +148,74 @@
         plan = API.filter_Default(plan, rowTypes);
         return plan;
     }
+    
+    
+    
+    private Operator computePlan()
+    {
+        Operator ret = null;
+
+        Group group = indexedRowType.userTable().getGroup();
+        Set<UserTableRowType> ancestors = new HashSet<>();
+        boolean hasDesc = false;
+
+        for (IndexColumn ic : index.getKeyColumns())
+        {
+            UserTable colUserTable = ic.getColumn().getUserTable();
+
+            if (!hasDesc && !colUserTable.equals(indexedRowType.userTable()))
+                // if any column in the index def belongs to a table
+                // that is a descendant of this indexed row's table
+                // (meaning this indexed row has descendant(s))
+                hasDesc = colUserTable.isDescendantOf(indexedRowType.userTable());
+            
+            // if the indexed table is a child of this column's table
+            // (meaning this indexed row has parent(s))
+            // collect all ancestor's rowtype
+            if (indexedRowType.userTable().isDescendantOf(colUserTable))
+                ancestors.add(schema.userTableRowType(colUserTable));
+        }
+
+        if (hasDesc)
+        {
+            ancestors.remove(indexedRowType);
+
+            ret = API.branchLookup_Nested(group, 
+                                           hKeyRowType,
+                                           indexedRowType,
+                                           API.InputPreservationOption.DISCARD_INPUT,
+                                           0);
+            if (!ancestors.isEmpty())
+            {
+                
+                ret = API.ancestorLookup_Default(ret,
+                                                 group,
+                                                 indexedRowType, 
+                                                 ancestors, 
+                                                 API.InputPreservationOption.KEEP_INPUT);
+            }
+        }
+        else
+        {
+            // has at least one ancestor (which is itself)
+            ancestors.add(indexedRowType);
+            ret = API.ancestorLookup_Nested(group,
+                                            hKeyRowType,
+                                            ancestors,
+                                            0);
+        }
+          
+        return ret;
+    }
+    /**
+     * 
+     * @param row
+     * @return the operator plan to get to every row related to this index row
+     */
+    public Operator getOperator()
+    {
+        return plan;
+    }
 
     public Analyzer getAnalyzer() {
         Analyzer analyzer;

=== modified file 'src/main/java/com/akiban/server/service/text/FullTextIndexInfosImpl.java'
--- src/main/java/com/akiban/server/service/text/FullTextIndexInfosImpl.java	2013-03-22 20:05:57 +0000
+++ src/main/java/com/akiban/server/service/text/FullTextIndexInfosImpl.java	2013-04-15 17:23:26 +0000
@@ -23,6 +23,7 @@
 import com.akiban.qp.rowtype.RowType;
 import com.akiban.qp.rowtype.Schema;
 import com.akiban.server.error.FullTextQueryParseException;
+import com.akiban.server.error.NoSuchIndexException;
 import com.akiban.server.service.session.Session;
 
 import org.apache.lucene.queryparser.flexible.core.QueryNodeException;
@@ -65,6 +66,21 @@
     protected abstract AkibanInformationSchema getAIS(Session session);
     protected abstract File getIndexPath();
 
+    protected FullTextIndexInfo getIndexToDrop(IndexName name)
+    {
+        synchronized(indexes)
+        {
+            FullTextIndexShared shared = indexes.get(name);
+            if (shared != null)
+            {
+                return shared.valueFor();
+            }
+            else
+            {
+                throw new NoSuchIndexException(name.toString());
+            }
+        }
+    }
     protected FullTextIndexInfo getIndex(Session session, IndexName name) {
         AkibanInformationSchema ais = getAIS(session);
         FullTextIndexInfo info;

=== modified file 'src/main/java/com/akiban/server/service/text/FullTextIndexService.java'
--- src/main/java/com/akiban/server/service/text/FullTextIndexService.java	2013-03-22 20:05:57 +0000
+++ src/main/java/com/akiban/server/service/text/FullTextIndexService.java	2013-04-15 17:23:26 +0000
@@ -27,7 +27,21 @@
 
 /** Full service that does index maintenance and querying. */
 public interface FullTextIndexService extends FullTextIndexInfos {
-    public long createIndex(Session session, IndexName name);
+        
+    /**
+     * This is a 'promise' to populate the given index, which is *being*
+     * created.
+     * @param name 
+     */
+    public void schedulePopulate(String schema, String table, String index);
+
+    /**
+     * Update the given index based on the changedRow 
+     * 
+     * @param name
+     * @param changedRow 
+     */
+    public void updateIndex(Session session, IndexName name, Iterable<byte[]> rows);
     public void dropIndex(Session session, IndexName name);
     public Cursor searchIndex(QueryContext context, IndexName name, 
                               Query query, int limit);

=== modified file 'src/main/java/com/akiban/server/service/text/FullTextIndexServiceImpl.java'
--- src/main/java/com/akiban/server/service/text/FullTextIndexServiceImpl.java	2013-03-22 20:05:57 +0000
+++ src/main/java/com/akiban/server/service/text/FullTextIndexServiceImpl.java	2013-04-15 17:23:26 +0000
@@ -18,39 +18,59 @@
 package com.akiban.server.service.text;
 
 import com.akiban.ais.model.AkibanInformationSchema;
+import com.akiban.ais.model.Index;
 import com.akiban.ais.model.IndexName;
+import com.akiban.ais.model.TableName;
+import com.akiban.ais.model.UserTable;
 import com.akiban.qp.operator.API;
 import com.akiban.qp.operator.Cursor;
 import com.akiban.qp.operator.Operator;
 import com.akiban.qp.operator.QueryContext;
 import com.akiban.qp.operator.SimpleQueryContext;
 import com.akiban.qp.operator.StoreAdapter;
-import com.akiban.qp.rowtype.RowType;
-import com.akiban.qp.rowtype.Schema;
 import com.akiban.qp.persistitadapter.PersistitAdapter;
+import com.akiban.qp.persistitadapter.PersistitHKey;
+import com.akiban.qp.row.HKeyRow;
+import com.akiban.qp.rowtype.HKeyRowType;
+import com.akiban.qp.util.HKeyCache;
 import com.akiban.server.error.AkibanInternalException;
-import com.akiban.server.error.DuplicateIndexException;
 import com.akiban.server.service.Service;
 import com.akiban.server.service.config.ConfigurationService;
 import com.akiban.server.service.dxl.DXLService;
 import com.akiban.server.service.session.Session;
+import com.akiban.server.service.session.SessionService;
+import com.akiban.server.service.session.SessionServiceImpl;
 import com.akiban.server.service.transaction.TransactionService;
 import com.akiban.server.service.tree.TreeService;
+import com.akiban.server.store.PersistitStore;
+import com.akiban.server.store.PersistitStore.HKeyBytesStream;
 import com.akiban.server.store.Store;
 
-import org.apache.lucene.analysis.Analyzer;
 import org.apache.lucene.index.IndexWriter;
 import org.apache.lucene.search.Query;
 
 import com.google.inject.Inject;
+import com.persistit.Exchange;
+import com.persistit.exception.PersistitException;
+import com.persistit.Key;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.*;
-import java.util.*;
+import java.util.Timer;
+import java.util.TimerTask;
+
 
 public class FullTextIndexServiceImpl extends FullTextIndexInfosImpl implements FullTextIndexService, Service {
     public static final String INDEX_PATH_PROPERTY = "akserver.text.indexpath";
+    public static final String UPDATE_INTERVAL = "akserver.text.maintenanceInterval";
+    public static final String POPULATE_DELAY_INTERVAL = "akserver.text.populateDelayInterval";
+
+    private static final String POPULATE_SCHEMA = "populate";
+    private static final String FULL_TEXT_TABLE = "full_text_populate";
+    
+    private final SessionService sessionService = new SessionServiceImpl();
+
 
     private final ConfigurationService configService;
     private final DXLService dxlService;
@@ -59,7 +79,15 @@
     private final TreeService treeService;
 
     private File indexPath;
-    
+
+    private volatile Timer maintenanceTimer;
+    private long maintenanceInterval;
+    private TimerTask updateWorker;
+    private PersistitStore persistitStore;
+
+    private volatile Timer populateTimer;
+    private long populateDelayInterval;
+
     private static final Logger logger = LoggerFactory.getLogger(FullTextIndexServiceImpl.class);
 
     @Inject
@@ -72,12 +100,14 @@
         this.store = store;
         this.transactionService = transactionService;
         this.treeService = treeService;
+
     }
 
     /* FullTextIndexService */
 
-    @Override
-    public long createIndex(Session session, IndexName name) {
+    private long createIndex(Session session, IndexName name) {
+        if (session == null)
+            session = sessionService.createSession();
         FullTextIndexInfo index = getIndex(session, name);
         try {
             return populateIndex(session, index);
@@ -89,10 +119,27 @@
 
     @Override
     public void dropIndex(Session session, IndexName name) {
-        FullTextIndexInfo index = getIndex(session, name);
-        index.deletePath();
-        synchronized (indexes) {
-            indexes.remove(name);
+        
+        try
+        {
+            // see if there exists a promise for populating this index
+            Exchange ex = getPopulateExchange(session);
+            ex.clear().append(name.getSchemaName())
+                      .append(name.getTableName())
+                      .append(name.getName());
+            
+            if (ex.traverse(Key.Direction.EQ, true, 0))
+                ex.fetchAndRemove();
+
+            FullTextIndexInfo index = getIndexToDrop(name);
+            index.deletePath();
+            synchronized (indexes) {
+                indexes.remove(name);
+            }
+        }
+        catch (PersistitException e)
+        {
+            throw new AkibanInternalException("Error while removing index", e);
         }
     }
 
@@ -112,6 +159,10 @@
     
     @Override
     public void start() {
+        persistitStore = store.getPersistitStore();
+        persistitStore.setFullTextService(this);
+        enableUpdateWorker();
+        enablePopulateWorker();
     }
 
     @Override
@@ -120,6 +171,8 @@
             for (FullTextIndexShared index : indexes.values()) {
                 index.close();
             }
+            disableUpdateWorker();
+            disablePopulateWorker();
         }
         catch (IOException ex) {
             throw new AkibanInternalException("Error closing index", ex);
@@ -193,4 +246,282 @@
         }
     }
 
+    @Override
+    public void updateIndex(Session session, IndexName name, Iterable<byte[]> rows)         
+    {
+        try
+        {
+            FullTextIndexInfo indexInfo = getIndex(session, name);
+            StoreAdapter adapter = session.get(StoreAdapter.STORE_ADAPTER_KEY);
+            if (adapter == null)
+                adapter = new PersistitAdapter(indexInfo.getSchema(),
+                                               store, treeService, 
+                                               session, configService);
+            QueryContext queryContext = new SimpleQueryContext(adapter);
+            HKeyCache<com.akiban.qp.row.HKey> cache = new HKeyCache<>(adapter);
+            IndexWriter writer = indexInfo.getIndexer().getWriter();
+            RowIndexer rowIndexer = new RowIndexer(indexInfo, writer, true);
+            Cursor cursor = null;
+            boolean success = false;
+            try
+            {
+                Operator operator = indexInfo.getOperator();
+                int n = 0;
+                for (byte row[] : rows)
+                {
+                    HKeyRow hkeyRow = toHKeyRow(row, indexInfo.getHKeyRowType(),
+                                                adapter, cache);
+                    queryContext.setRow(0, hkeyRow);
+                    cursor = API.cursor(operator, queryContext);
+                    rowIndexer.updateDocument(cursor, row);
+                }
+                success = true;
+            }
+            finally
+            {
+                rowIndexer.close();
+                if (cursor != null)
+                    cursor.destroy();
+
+                try
+                {
+                    if (success)
+                        writer.commit();
+                    else
+                        writer.rollback();
+                }
+                catch (IOException ex)
+                {
+                    throw new AkibanInternalException("Error commit writer", ex);
+                }
+            }
+        }
+        catch (IOException e)
+        {
+            throw new AkibanInternalException("Error updating index", e);
+        }
+    }
+
+    private class DefaultUpdateWorker extends TimerTask
+    {
+        @Override
+        // 'sync' because only one worker can work at a time?
+        public synchronized void run()
+        {
+            Session session = sessionService.createSession();
+            boolean transaction = true;
+            try
+            {
+                transactionService.beginTransaction(session);
+                HKeyBytesStream rows = persistitStore.getChangedRows(session);
+                if (rows != null) // if tree is not empty
+                {
+                    do
+                    {
+                        updateIndex(session,
+                                    rows.getIndexName(),
+                                    rows);
+                    }
+                    while (rows.nextIndex());
+                    rows.removeAll(); // done updating. remove all entries
+                }
+                transaction= false;
+            }
+            catch(PersistitException e)
+            {
+                throw new AkibanInternalException("Error while maintaning full_text indices");
+            }
+            finally
+            {
+                if (transaction)
+                    transactionService.rollbackTransaction(session);
+                else
+                    transactionService.commitTransaction(session);
+                session.close();
+            }
+        }
+    }; 
+    
+ 
+    @Override
+    public void schedulePopulate(String schema, String table, String index)
+    {   
+        Session session = sessionService.createSession();
+        boolean success = false;
+        try
+        {
+            transactionService.beginTransaction(session);
+            if(addPopulate(session, schema, table, index) && !hasScheduled && populateEnabled)
+            {
+                populateTimer.schedule(populateWorker(), populateDelayInterval);
+                hasScheduled = true;
+            }
+
+            success = true;            
+        }
+        catch (PersistitException ex)
+        {
+            throw new AkibanInternalException("Error while scheduling index population", ex);
+        }
+        finally
+        {
+            if (success)
+                transactionService.commitTransaction(session);
+            else
+                transactionService.rollbackTransaction(session);
+            session.close();
+        }
+    }
+    
+    private class DefaultPopulateWorker extends TimerTask
+    {
+        @Override
+        public synchronized void run()
+        {
+            Session session = sessionService.createSession();
+            try
+            {
+                Exchange ex = getPopulateExchange(session);
+                IndexName toPopulate;
+                while ((toPopulate = nextInQueue(ex)) != null)
+                {
+                    if (stillExists(session, toPopulate))
+                        createIndex(session, toPopulate);
+                    else
+                        logger.debug("FullTextIndex " + toPopulate + " deleted before population");
+                }
+                ex.removeAll();
+                hasScheduled = false;
+            }
+            catch (PersistitException ex1)
+            {
+                throw new AkibanInternalException("Error while populating full_text indices", ex1);
+            }
+            finally
+            {
+                session.close();
+            }
+        }
+        
+        private boolean stillExists(Session session, IndexName indexName)
+        {
+            AkibanInformationSchema ais = getAIS(session);
+            UserTable table = ais.getUserTable(indexName.getFullTableName());
+            return !(table == null ||  table.getFullTextIndex(indexName.getName()) == null);
+        }
+
+    };
+
+    // ---------- for testing ---------------
+    void disableUpdateWorker()
+    {
+        if (maintenanceTimer == null)
+        {
+            logger.debug("maintenance worker ALREADY disabled");
+            return;
+        }
+        updateWorker.cancel();
+        maintenanceTimer.cancel();
+        maintenanceTimer.purge();
+        maintenanceTimer = null;
+        updateWorker = null;
+    }
+    
+    protected void enableUpdateWorker()
+    {
+        maintenanceInterval = Long.parseLong(configService.getProperty(UPDATE_INTERVAL));
+        maintenanceTimer = new Timer();
+        updateWorker = new DefaultUpdateWorker();
+        maintenanceTimer.scheduleAtFixedRate(updateWorker, maintenanceInterval, maintenanceInterval);
+    }
+    
+    protected void enablePopulateWorker()
+    {
+        populateDelayInterval = Long.parseLong(configService.getProperty(POPULATE_DELAY_INTERVAL));
+        populateTimer = new Timer();
+        populateTimer.schedule(populateWorker(), populateDelayInterval);
+        populateEnabled = true;
+        hasScheduled = true;
+    }
+    
+    void disablePopulateWorker()
+    {
+        if (populateTimer == null)
+        {
+            logger.debug("populate worker already disabled");
+            return;
+        }
+        populateTimer.cancel();
+        populateTimer.purge();
+        hasScheduled = false;
+        populateEnabled = false;
+        populateTimer = null;
+    }
+    
+    private TimerTask populateWorker()
+    {
+        return new DefaultPopulateWorker();
+    }
+
+    //----------- private helpers -----------
+    protected IndexName nextInQueue(Exchange ex) throws PersistitException
+    {
+        Key key = ex.getKey();
+
+        if (ex.next(true)) // empty tree?
+        {
+            key.reset();
+            IndexName ret = new IndexName(new TableName(key.decodeString(),
+                                                        key.decodeString()),
+                                          key.decodeString());
+            return ret;
+        }
+        else
+            return null;
+    }
+    
+    protected Exchange getPopulateExchange(Session session) throws PersistitException
+    {
+        Exchange ret = treeService.getExchange(session,
+                                               treeService.treeLink(POPULATE_SCHEMA,
+                                                                    FULL_TEXT_TABLE));
+        // start at the first entry
+        ret.append(Key.BEFORE);
+        return ret;
+    }
+
+    private volatile boolean hasScheduled = false;
+    private volatile boolean populateEnabled = false;
+
+    private synchronized boolean addPopulate(Session session,
+                           String schema,
+                           String table,
+                           String index) throws PersistitException
+    {
+        Exchange ex = getPopulateExchange(session);
+
+        // Assumption: There is not any existing entry about this index
+        // (Because they should have been removed in dropIndex())
+        // KEY: schema | table | indexName
+        ex.getKey().clear()
+                   .append(schema)
+                   .append(table)
+                   .append(index);
+
+        // VALUE: <empty>
+
+        ex.store();
+        return true;
+    }
+
+    private HKeyRow toHKeyRow(byte rowBytes[], HKeyRowType hKeyRowType,
+                              StoreAdapter store, HKeyCache<com.akiban.qp.row.HKey> cache)
+    {
+        PersistitHKey hkey = (PersistitHKey)store.newHKey(hKeyRowType.hKey());
+        Key key = hkey.key();
+        key.setEncodedSize(rowBytes.length);
+        System.arraycopy(rowBytes, 0, key.getEncodedBytes(), 0, rowBytes.length);
+        return new HKeyRow(hKeyRowType, hkey, cache);
+    }
+
 }

=== modified file 'src/main/java/com/akiban/server/service/text/FullTextIndexShared.java'
--- src/main/java/com/akiban/server/service/text/FullTextIndexShared.java	2013-03-22 20:05:57 +0000
+++ src/main/java/com/akiban/server/service/text/FullTextIndexShared.java	2013-04-15 17:23:26 +0000
@@ -103,6 +103,11 @@
         return ais.getCachedValue(this, this);
     }
 
+    public FullTextIndexInfo valueFor()
+    {
+        return new FullTextIndexInfo(this);
+    }
+
     @Override
     public FullTextIndexInfo valueFor(AkibanInformationSchema ais) {
         FullTextIndexInfo result = new FullTextIndexInfo(this);

=== modified file 'src/main/java/com/akiban/server/service/text/Indexer.java'
--- src/main/java/com/akiban/server/service/text/Indexer.java	2013-03-22 20:05:57 +0000
+++ src/main/java/com/akiban/server/service/text/Indexer.java	2013-04-15 17:23:26 +0000
@@ -26,6 +26,7 @@
 
 import java.io.Closeable;
 import java.io.IOException;
+import org.apache.lucene.index.IndexReader;
 
 public class Indexer implements Closeable
 {
@@ -34,8 +35,14 @@
     
     public Indexer(FullTextIndexShared index, Analyzer analyzer) throws IOException {
         this.index = index;
+         
+        
         IndexWriterConfig iwc = new IndexWriterConfig(Version.LUCENE_40, analyzer);
-        this.writer = new IndexWriter(index.open(), iwc);
+        iwc.setMaxBufferedDeleteTerms(1); // The deletion needs to be reflected immediately (on disk)
+        
+        this.writer = new IndexWriter(index.open(),  iwc);
+        IndexReader r;
+        
     }
 
     public FullTextIndexShared getIndex() {

=== modified file 'src/main/java/com/akiban/server/service/text/RowIndexer.java'
--- src/main/java/com/akiban/server/service/text/RowIndexer.java	2013-03-22 20:05:57 +0000
+++ src/main/java/com/akiban/server/service/text/RowIndexer.java	2013-04-15 17:23:26 +0000
@@ -20,9 +20,11 @@
 import com.akiban.ais.model.AkibanInformationSchema;
 import com.akiban.qp.operator.Cursor;
 import com.akiban.qp.persistitadapter.PersistitHKey;
+import com.akiban.qp.row.HKey;
 import com.akiban.qp.row.Row;
 import com.akiban.qp.rowtype.RowType;
 import com.akiban.qp.rowtype.UserTableRowType;
+import com.akiban.qp.util.PersistitKey;
 import com.akiban.server.types3.pvalue.PValueSource;
 import com.akiban.util.ShareHolder;
 import com.persistit.Key;
@@ -39,6 +41,11 @@
 
 import java.io.*;
 import java.util.*;
+import org.apache.commons.codec.binary.Base64;
+import org.apache.lucene.document.Field.Store;
+import org.apache.lucene.document.StringField;
+import org.apache.lucene.index.DirectoryReader;
+import org.apache.lucene.index.IndexReader;
 
 /** Given <code>Row</code>s in hkey order, create <code>Document</code>s. */
 public class RowIndexer implements Closeable
@@ -50,7 +57,8 @@
     private IndexWriter writer;
     private Document currentDocument;
     private long documentCount;
-    private BytesRef keyBytes;
+    //private BytesRef keyBytes;
+    private String keyEncodedString;
     private boolean updating;
     
     private static final Logger logger = LoggerFactory.getLogger(RowIndexer.class);
@@ -134,10 +142,21 @@
         return documentCount;
     }
 
+    protected void updateDocument(Cursor cursor, byte hkeyBytes[]) throws IOException
+    {
+        if (indexRows(cursor) == 0)
+        {
+            String encoded = encodeBytes(hkeyBytes, 0, hkeyBytes.length);
+            writer.deleteDocuments(new Term(IndexedField.KEY_FIELD, encoded));
+            logger.debug("Deleted documents with encoded byptes: " + encoded);
+        }
+    }
+
     protected void addDocument() throws IOException {
         if (currentDocument != null) {
             if (updating) {
-                writer.updateDocument(new Term(IndexedField.KEY_FIELD, keyBytes), 
+                
+                writer.updateDocument(new Term(IndexedField.KEY_FIELD, keyEncodedString), 
                                       currentDocument);
                 logger.debug("Updated {}", currentDocument);
             }
@@ -152,8 +171,8 @@
 
     protected void getKeyBytes(Row row) {
         Key key = ((PersistitHKey)row.hKey()).key();
-        keyBytes = new BytesRef(key.getEncodedBytes(), 0, key.getEncodedSize());
-        Field field = new StoredField(IndexedField.KEY_FIELD, keyBytes);
+        keyEncodedString = encodeBytes(key.getEncodedBytes(), 0, key.getEncodedSize());
+        Field field = new StringField(IndexedField.KEY_FIELD, keyEncodedString, Store.YES);
         currentDocument.add(field);
     }
 
@@ -166,6 +185,18 @@
         }
     }
 
+    static String encodeBytes(byte bytes[], int offset, int length)
+    {
+        // TODO: needs to be more efficient?
+        String ret = Base64.encodeBase64String(Arrays.copyOfRange(bytes, offset, length));
+        return ret;
+    }
+    
+    static byte[] decodeString(String st)
+    {
+        return Base64.decodeBase64(st);
+    }
+
     @Override
     public void close() {
         for (ShareHolder<Row> holder : ancestors) {

=== modified file 'src/main/java/com/akiban/server/store/PersistitStore.java'
--- src/main/java/com/akiban/server/store/PersistitStore.java	2013-04-11 06:15:45 +0000
+++ src/main/java/com/akiban/server/store/PersistitStore.java	2013-04-15 17:23:26 +0000
@@ -18,6 +18,7 @@
 package com.akiban.server.store;
 
 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;
@@ -41,6 +42,7 @@
 import com.akiban.server.service.config.ConfigurationService;
 import com.akiban.server.service.lock.LockService;
 import com.akiban.server.service.session.Session;
+import com.akiban.server.service.text.FullTextIndexService;
 import com.akiban.server.service.transaction.TransactionService;
 import com.akiban.server.service.tree.TreeLink;
 import com.akiban.server.service.tree.TreeService;
@@ -95,7 +97,9 @@
     private static final String BULKLOAD_PK_BUFFER_ALLOC = "akserver.bulkload.bufferalloc.pk";
     private static final String BULKLOAD_GROUP_BUFFER_ALLOC = "akserver.bulkload.bufferalloc.group";
     private static final String WRITE_LOCK_ENABLED_CONFIG = "akserver.write_lock_enabled";
-    
+    private static final String MAINTENANCE_SCHEMA = "maintenance";
+    private static final String FULL_TEXT_TABLE = "full_text";
+
     private final static int MAX_ROW_SIZE = 5000000;
 
     private final static int MAX_INDEX_TRANCHE_SIZE = 10 * 1024 * 1024;
@@ -128,8 +132,14 @@
 
     private int deferredIndexKeyLimit = MAX_INDEX_TRANCHE_SIZE;
 
+    private FullTextIndexService fullTextService;
+
     private RowDataValueCoder valueCoder;
 
+    // Each row change has a 'uniqueChangeId', stored in the 'maintenance.full_text' table
+    // The number would be reset after all maintenance is done
+    private volatile AtomicLong uniqueChangeId = new AtomicLong(0); // TODO: change this back to MIN_VAL
+    
     private static final StorageAction DIRECT_STORAGE = new StorageAction() {
         @Override
         public void doStore(Exchange exchange) throws PersistitException {
@@ -151,6 +161,368 @@
         this.transactionService = transactionService;
     }
 
+    public void setFullTextService(FullTextIndexService service)
+    {
+        fullTextService = service;
+    }
+
+     // --- for tracking changes 
+    private Exchange getChangeExchange(Session session) throws PersistitException
+    {   
+        return treeService.getExchange(session,
+                                       treeService.treeLink(MAINTENANCE_SCHEMA,
+                                                            FULL_TEXT_TABLE));
+    }
+
+    private void addChange(Session session,
+                           String schema,
+                           String table,
+                           String index,
+                           Integer indexId,
+                           Key rowHKey) throws PersistitException
+    {
+        Exchange ex = getChangeExchange(session);
+
+        // KEY: schema | table | indexName | indexId | unique_num 
+        ex.clear().append(schema).append(table).append(index).append(indexId).append(uniqueChangeId.getAndIncrement());
+
+        // VALUE: rowHKey's bytes | indexId
+        ex.getValue().clear().putByteArray(rowHKey.getEncodedBytes(),
+                                           0,
+                                           rowHKey.getEncodedSize());
+        ex.store();
+    }
+
+    private void addChangeFor(UserTable tb, Session session, Key hKey) throws PersistitException
+    {
+        for (Index index : tb.getFullTextIndexes())
+        {
+            IndexName idxName = index.getIndexName();
+            addChange(session,
+                      idxName.getSchemaName(),
+                      idxName.getTableName(),
+                      idxName.getName(),
+                      index.getIndexId(),
+                      hKey);
+        }
+    }
+    // --- for reporting changes
+    /**
+     * Collect all 'HKeyRow's from the list of key-value pairs that have
+     * the same schema.table.indexName,
+     * also fill schema[0],
+     *           table[0],
+     *      and  indexName[0]
+     *  with correct values
+     * 
+     * @param schema
+     * @param table
+     * @param indexName
+     * @param ex
+     * @return 
+     */
+    public HKeyBytesStream getChangedRows(Session session) throws PersistitException
+    {
+        Exchange ex = getChangeExchange(session);
+        ex.append(Key.BEFORE);
+        HKeyBytesStream ret;
+        
+        if (ex.hasNext(true)                                           // if the tree is not empty
+                && (ret = new HKeyBytesStream(ex, session)).hasNext()) // and does not contain only invalid indices
+        {
+            return ret;
+        }
+        else
+        {
+            return null;
+        }
+    }
+
+ 
+    /**
+     * 
+     * This is supposed to behave similar* to Scheme's stream,
+     * (ie., each element is not computed until it absolutely has to be.)
+     * 
+     * In other words, the byte arrays are not decoded before hand
+     */
+    public class HKeyBytesStream implements Iterable<byte[]>
+    {
+
+        private final Exchange ex;
+        private IndexName indexName;
+        private int modCount = 0;
+        private boolean moreRows; // more rows of this index
+        private final Session session;
+        private boolean eot; // end of tree;
+        private HKeyBytesStream()
+        {
+            ex = null;
+            session = null;
+        }
+
+        // 'private' because this should not be constructed
+        // anywhere other than PersistitStore.getChangedRows()
+        // The class itself, however, is public, as it can be used anywhere
+        private HKeyBytesStream (Exchange ex, Session session) throws PersistitException
+        {
+            this.ex = ex;
+            this.session = session;
+            eot = false;
+            moreRows = true;
+            nextIndex(); // at the beginning of the tree now. look for the first legit index
+        }
+
+        public final boolean nextIndex() throws PersistitException
+        {
+            if (eot)
+                return false;
+
+            boolean nextIndex = ex.next(true);
+            if (nextIndex)
+            {
+                modCount = 0;
+                indexName = buildName(ex);
+                if (eot = !ignoreDeleted(ex, indexName, true)) // reach the end after ignoring all deleted indexId
+                    return moreRows = false;                   // hence no more rows (or indices)
+            }
+            else
+            {
+                eot = true;
+                moreRows = false;
+            }
+
+            return nextIndex;
+        }
+
+        public boolean hasNext()
+        {
+            return !eot;
+        }
+
+        public IndexName getIndexName()
+        {
+            return indexName;
+        }
+
+        @Override
+        public Iterator<byte[]> iterator()
+        {
+            return new StreamIterator(moreRows);
+        }
+
+        /**
+         * remove all change-entries 
+         * 
+         * 
+         */
+        public void removeAll() throws PersistitException
+        {
+            ex.removeAll(); 
+            ++modCount;
+        }
+
+        private class StreamIterator implements Iterator<byte[]>
+        {
+
+            private boolean hasNext;
+            private int innerModCount = modCount;
+
+            private StreamIterator (boolean hasNext)
+            {
+                this.hasNext = hasNext;
+            }
+
+            @Override
+            public boolean hasNext()
+            {
+                return hasNext;
+            }
+
+            @Override
+            public byte[] next()
+            {
+                try
+                {
+                    if (innerModCount != modCount)
+                        throw new ConcurrentModificationException();
+
+                    byte ret[] = ex.getValue().getByteArray();
+                    boolean seeNewIndex = false;
+                    boolean hasMore;
+                    hasNext = (hasMore = ex.next(true)) 
+                                        && !(seeNewIndex = seeNewIndex(indexName.getSchemaName(),
+                                                                       indexName.getTableName(),
+                                                                       indexName.getName(),
+                                                                       ex.getKey())
+                                        // and following this row is at least one row of
+                                        // this index whose index-id is valid
+                                        && ignoreDeleted(ex, indexName, false));
+
+                    // if there are no more entries,
+                    // set eot so the 'outer loop', which takes care of each index
+                    // does not attempt to do 'next(true)'
+                    // (because surpringly, after the end has been reached
+                    //  the key will be set to BEFORE, therefore next(true)
+                    /// will return 'true'. ==> infinite loop!)
+                    eot = !hasMore;
+
+                    // Take caution in order NOT to go to the next index (ie., different name)
+                    if (seeNewIndex)  
+                        // back up one entry, because we have read past the last
+                        // entry that has the same schema.table.indexName
+                        ex.previous(); 
+                    ++innerModCount;
+                    ++modCount;
+
+                    return ret;
+                }
+                catch (PersistitException ex)
+                {
+                    throw new AkibanInternalException("Error retrieving rows from Exchange", ex);
+                }
+            }
+
+            @Override
+            public void remove()
+            {
+                throw new UnsupportedOperationException("Not supported yet.");
+            }   
+        }
+
+        /**
+         * Skip all entries whose indexName is that of a non-existing index,
+         * or whose indexId is not the same as that of the the current index (with the same name)
+         * 
+         * @param ex
+         * @param indexName
+         * @param lookPastNewIndex whether to go beyond the current index
+         * @return <FALSE> iff after ignoring all deleted indices, there's nothing left
+         *         <TRUE> otherwise
+         * @throws PersistitException 
+         */
+        private boolean ignoreDeleted(Exchange ex,
+                                      IndexName indexName,
+                                      boolean lookPastNewIndex)
+                        throws PersistitException
+        {
+            // KEY: Schema | Table | indexName | indexID | ...
+            Key key = ex.getKey();
+            key.reset();
+            
+            key.reset();
+            assert indexName.getSchemaName().equals(key.decodeString()) 
+                    : "Unexpected schema" ;
+            assert indexName.getTableName().equals(key.decodeString())
+                    : "Unexpected table";
+            assert indexName.getName().equals(key.decodeString())
+                    : "Unexpected name";
+
+            Integer indexId = key.decodeInt();
+            while (sameIdAsCurrent(session, indexName, indexId) != NOCHANGE)
+            // there was a DELETE (and possibly RECREATE)
+            // ignore all changes in the old id(s)
+            {
+                boolean hasMore = true;
+                boolean seeNewIndex = false;
+
+                do
+                { /*do nothing (skipping deleted 'index')*/}
+                while ((hasMore = ex.next(true))
+                                 && !(seeNewIndex = seeNewIndex(indexName.getSchemaName(),
+                                                                indexName.getTableName(),
+                                                                indexName.getName(),
+                                                                ex.getKey()))
+                                 && !seeNewIndexId(ex, indexId));
+
+                if (eot = !hasMore)  // reach the end! (no more rows or indices!)
+                    return false;    // set 'eot' for the same reason in StreamIterator.next() 
+
+                else if (seeNewIndex) // back up one entry in order not to 
+                {                     // go past the last pair
+                    ex.previous(true);
+                    
+                    // Saw new index,
+                    // hence do the checking again,
+                    // if we're allowed to look past new index
+                    // (otherwise, meaning no more rows with good ids of this
+                    //   index, return false)
+                    return lookPastNewIndex ? nextIndex() : false;
+                }
+                else // ie., see new IndexId
+                {
+                    Key k = ex.getKey();
+                    k.reset();
+                    k.decodeString();
+                    k.decodeString();
+                    k.decodeString();
+
+                    indexId = k.decodeInt();
+                }
+            }
+            return true;
+        }
+
+        private boolean seeNewIndexId(Exchange ex, Integer oldId)
+        {
+            Key key = ex.getKey();
+            key.reset();
+            // skip uninteresting parts
+            key.decodeString();
+            key.decodeString();
+            key.decodeString();
+
+            return !oldId.equals(key.decodeInt());
+            
+        }
+
+        /**
+         * Check if the given id is the same as the current index's id.
+         * 
+         * (If the index with the given name no longer exists, return false)
+         * @param schema
+         * @param table
+         * @param indexName
+         * @param id
+         * @return 
+         */
+        private int sameIdAsCurrent(Session session, IndexName indexName, Integer id)
+        {
+            AkibanInformationSchema ais = getAIS(session);
+            UserTable table = ais.getUserTable(indexName.getFullTableName());
+            Index index;
+            if (table == null || (index = table.getFullTextIndex(indexName.getName())) == null)
+                return DELETED;
+            
+            return index.getIndexId() != id ? RECREATED : NOCHANGE;
+        }
+
+        private boolean seeNewIndex(String schema, String table, String indexName, Key k)
+        {
+            k.reset();
+            return !k.decodeString().equals(schema)
+                    || !k.decodeString().equals(table)
+                    || !k.decodeString().equals(indexName);
+        }
+
+        private IndexName buildName(Exchange e)
+        {
+            Key key = e.getKey();
+            key.reset();
+            return new IndexName(new TableName(key.decodeString(),
+                                               key.decodeString()),
+                                 key.decodeString());
+        }
+        
+        // status of index
+        private static final int NOCHANGE = 0;
+        private static final int DELETED = 1;
+        private static final int RECREATED = 2;
+        
+    }
+
+    //------ end fulltext index maintenance services ------
+
     @Override
     public synchronized void start() {
         try {
@@ -444,6 +816,7 @@
             }
 
             PersistitIndexRowBuffer indexRow = new PersistitIndexRowBuffer(adapter(session));
+            addChangeFor(rowDef.userTable(), session, hEx.getKey());
             for (Index index : rowDef.getIndexes()) {
                 insertIntoIndex(session, index, rowData, hEx.getKey(), indexRow, deferIndexes);
             }
@@ -668,6 +1041,7 @@
     {
         RowDef rowDef = writeCheck(session, rowData, false);
         Exchange hEx = null;
+
         DELETE_ROW_TAP.in();
         try {
             hEx = getExchange(session, rowDef);
@@ -681,12 +1055,16 @@
             if (!hEx.getValue().isDefined()) {
                 throw new NoSuchRowException(hEx.getKey());
             }
+            // record the deletion of the old index row
+            if (deleteIndexes)
+                addChangeFor(rowDef.userTable(), session, hEx.getKey());
 
             // Remove the h-row
             hEx.remove();
             rowDef.getTableStatus().rowDeleted();
 
             // Remove the indexes, including the PK index
+
             PersistitIndexRowBuffer indexRow = new PersistitIndexRowBuffer(adapter(session));
             if(deleteIndexes) {
                 for (Index index : rowDef.getIndexes()) {
@@ -738,6 +1116,7 @@
         RowDef newRowDef = rowDefFromExplicitOrId(session, newRowData);
         PersistitAdapter adapter = adapter(session);
         Exchange hEx = null;
+
         UPDATE_ROW_TAP.in();
         try {
             hEx = getExchange(session, rowDef);
@@ -751,6 +1130,7 @@
             if (!hEx.getValue().isDefined()) {
                 throw new NoSuchRowException (hEx.getKey());
             }
+
             // Combine current version of row with the version coming in
             // on the update request.
             // This is done by taking only the values of columns listed
@@ -770,7 +1150,9 @@
                 packRowData(hEx, newRowDef, mergedRowData);
                 // Store the h-row
                 hEx.store();
-                // Update the indexes
+                // Update the indexes (new row)
+                addChangeFor(newRowDef.userTable(), session, hEx.getKey());
+                
                 PersistitIndexRowBuffer indexRowBuffer = new PersistitIndexRowBuffer(adapter);
                 Index[] indexes = (indexesToMaintain == null) ? rowDef.getIndexes() : indexesToMaintain;
                 for (Index index : indexes) {
@@ -874,6 +1256,10 @@
             int descendentOrdinal = descendentRowDef.getOrdinal();
             if ((tablesRequiringHKeyMaintenance == null || tablesRequiringHKeyMaintenance.get(descendentOrdinal))) {
                 PROPAGATE_HKEY_CHANGE_ROW_REPLACE_TAP.hit();
+                
+                // record the change
+                addChangeFor(descendentRowDef.userTable(), session, exchange.getKey());
+                
                 // Delete the current row from the tree. Don't call deleteRow, because we don't need to recompute
                 // the hkey.
                 exchange.remove();
@@ -1554,6 +1940,13 @@
     public void deleteIndexes(final Session session, final Collection<? extends Index> indexes) {
         List<TreeLink> links = new ArrayList<>(indexes.size());
         for(Index index : indexes) {
+            // no trees to drop
+            if (index.getIndexType() == IndexType.FULL_TEXT)
+            {
+                fullTextService.dropIndex(session, index.getIndexName());
+                indexes.remove(index);
+                continue;
+            }
             final IndexDef indexDef = index.indexDef();
             if(indexDef == null) {
                 throw new IllegalStateException("indexDef is null for index: " + index);

=== added file 'src/main/java/com/akiban/server/types3/mcompat/mfuncs/MSleep.java'
--- src/main/java/com/akiban/server/types3/mcompat/mfuncs/MSleep.java	1970-01-01 00:00:00 +0000
+++ src/main/java/com/akiban/server/types3/mcompat/mfuncs/MSleep.java	2013-04-15 17:23:26 +0000
@@ -0,0 +1,71 @@
+/**
+ * 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.types3.mcompat.mfuncs;
+
+import com.akiban.server.error.AkibanInternalException;
+import com.akiban.server.types3.LazyList;
+import com.akiban.server.types3.TExecutionContext;
+import com.akiban.server.types3.TOverloadResult;
+import com.akiban.server.types3.TScalar;
+import com.akiban.server.types3.mcompat.mtypes.MNumeric;
+import com.akiban.server.types3.pvalue.PValueSource;
+import com.akiban.server.types3.pvalue.PValueTarget;
+import com.akiban.server.types3.texpressions.TInputSetBuilder;
+import com.akiban.server.types3.texpressions.TScalarBase;
+
+public class MSleep extends TScalarBase
+{
+
+    public static final TScalar SLEEP = new MSleep();
+    
+    private MSleep() {}
+    @Override
+    protected void buildInputSets(TInputSetBuilder builder)
+    {
+        builder.covers(MNumeric.INT, 0);
+    }
+
+    @Override
+    protected void doEvaluate(TExecutionContext context, LazyList<? extends PValueSource> inputs, PValueTarget output)
+    {
+        try
+        {
+            int value = inputs.get(0).getInt32();
+            Thread.sleep(value);
+            output.putInt32(value);
+        }
+        catch (InterruptedException ex)
+        {
+            output.putNull();
+            throw new AkibanInternalException("InteruptedException " + ex);
+        }
+    }
+
+    @Override
+    public String displayName()
+    {
+        return "sleep";
+    }
+
+    @Override
+    public TOverloadResult resultType()
+    {
+        return TOverloadResult.fixed(MNumeric.INT);
+    }
+    
+}

=== modified file 'src/main/java/com/akiban/sql/aisddl/AISDDL.java'
--- src/main/java/com/akiban/sql/aisddl/AISDDL.java	2013-03-22 20:05:57 +0000
+++ src/main/java/com/akiban/sql/aisddl/AISDDL.java	2013-04-15 17:23:26 +0000
@@ -84,7 +84,7 @@
                              server.getBinderContext(), context);
             return;
         case NodeTypes.CREATE_INDEX_NODE:
-            IndexDDL.createIndex(ddlFunctions, session, schema, (CreateIndexNode)ddl);
+            IndexDDL.createIndex(ddlFunctions, session, schema, (CreateIndexNode)ddl, server.getServiceManager());
             return;
         case NodeTypes.DROP_INDEX_NODE:
             IndexDDL.dropIndex(ddlFunctions, session, schema, (DropIndexNode)ddl, context);

=== modified file 'src/main/java/com/akiban/sql/aisddl/AlterTableDDL.java'
--- src/main/java/com/akiban/sql/aisddl/AlterTableDDL.java	2013-03-22 20:05:57 +0000
+++ src/main/java/com/akiban/sql/aisddl/AlterTableDDL.java	2013-04-15 17:23:26 +0000
@@ -254,7 +254,7 @@
         TableName newName = tableCopy.getName();
         for(ConstraintDefinitionNode cdn : conDefNodes) {
             assert cdn.getConstraintType() != ConstraintType.DROP : cdn;
-            String name = TableDDL.addIndex(indexNamer, builder, cdn, newName.getSchemaName(), newName.getTableName());
+            String name = TableDDL.addIndex(indexNamer, builder, cdn, newName.getSchemaName(), newName.getTableName(), context);
             indexChanges.add(TableChange.createAdd(name));
         }
 

=== modified file 'src/main/java/com/akiban/sql/aisddl/IndexDDL.java'
--- src/main/java/com/akiban/sql/aisddl/IndexDDL.java	2013-03-22 20:05:57 +0000
+++ src/main/java/com/akiban/sql/aisddl/IndexDDL.java	2013-04-15 17:23:26 +0000
@@ -36,6 +36,8 @@
 import com.akiban.server.service.session.Session;
 
 import com.akiban.qp.operator.QueryContext;
+import com.akiban.server.service.ServiceManager;
+import com.akiban.server.service.text.FullTextIndexService;
 
 /** DDL operations on Indices */
 public class IndexDDL
@@ -87,7 +89,7 @@
             }
             // if we can't find the index, set tableName to null
             // to flag not a user table index. 
-            if (table.getIndex(indexName) == null) {
+            if (table.getIndex(indexName) == null && table.getFullTextIndex(indexName) == null) {
                 tableName = null;
             }
             // Check the group associated to the table for the 
@@ -148,17 +150,18 @@
     public static void createIndex(DDLFunctions ddlFunctions,
                                    Session session,
                                    String defaultSchemaName,
-                                   CreateIndexNode createIndex)  {
+                                   CreateIndexNode createIndex,
+                                   ServiceManager sm)  {
         AkibanInformationSchema ais = ddlFunctions.getAIS(session);
         
         Collection<Index> indexesToAdd = new LinkedList<>();
 
-        indexesToAdd.add(buildIndex(ais, defaultSchemaName, createIndex));
+        indexesToAdd.add(buildIndex(ais, defaultSchemaName, createIndex, sm));
         
         ddlFunctions.createIndexes(session, indexesToAdd);
     }
     
-    protected static Index buildIndex (AkibanInformationSchema ais, String defaultSchemaName, CreateIndexNode index) {
+    protected static Index buildIndex (AkibanInformationSchema ais, String defaultSchemaName, CreateIndexNode index, ServiceManager sm) {
         final String schemaName = index.getObjectName().getSchemaName() != null ? index.getObjectName().getSchemaName() : defaultSchemaName;
         final String indexName = index.getObjectName().getTableName();
 
@@ -173,7 +176,7 @@
         
         if (index.getColumnList().functionType() == IndexColumnList.FunctionType.FULL_TEXT) {
             logger.debug ("Building Full text index on table {}", tableName) ;
-            tableIndex = buildFullTextIndex (builder, tableName, indexName, index);
+            tableIndex = buildFullTextIndex (builder, tableName, indexName, index, sm);
         } else if (checkIndexType (index, tableName) == Index.IndexType.TABLE) {
             logger.debug ("Building Table index on table {}", tableName) ;
             tableIndex = buildTableIndex (builder, tableName, indexName, index);
@@ -328,9 +331,10 @@
         return builder.akibanInformationSchema().getGroup(groupName).getIndex(indexName);
     }
 
-    protected static Index buildFullTextIndex (AISBuilder builder, TableName tableName, String indexName, IndexDefinition index) {
+    protected static Index buildFullTextIndex (AISBuilder builder, TableName tableName, String indexName, IndexDefinition index, ServiceManager sm) {
         UserTable table = builder.akibanInformationSchema().getUserTable(tableName);
-
+        FullTextIndexService fullTextService = sm.getServiceByClass(FullTextIndexService.class);
+        
         if (index.getJoinType() != null) {
             throw new TableIndexJoinTypeException();
         }
@@ -363,6 +367,8 @@
             }
             
             builder.fullTextIndexColumn(tableName, indexName, schemaName, columnTable.getTableName(), columnName, i);
+            // populate index
+            fullTextService.schedulePopulate(schemaName, tableName.getTableName(), indexName);
             i++;
         }
         return builder.akibanInformationSchema().getUserTable(tableName).getFullTextIndex(indexName);

=== modified file 'src/main/java/com/akiban/sql/aisddl/TableDDL.java'
--- src/main/java/com/akiban/sql/aisddl/TableDDL.java	2013-04-11 05:21:42 +0000
+++ src/main/java/com/akiban/sql/aisddl/TableDDL.java	2013-04-15 17:23:26 +0000
@@ -195,7 +195,7 @@
                 }
             }
             else if (tableElement instanceof ConstraintDefinitionNode) {
-                addIndex (namer, builder, (ConstraintDefinitionNode)tableElement, schemaName, tableName);
+                addIndex (namer, builder, (ConstraintDefinitionNode)tableElement, schemaName, tableName, context);
             }
         }
         builder.basicSchemaIsComplete();
@@ -312,7 +312,7 @@
 
 
     public static String addIndex(IndexNameGenerator namer, AISBuilder builder, ConstraintDefinitionNode cdn,
-                                  String schemaName, String tableName)  {
+                                  String schemaName, String tableName, QueryContext context)  {
         // We don't (yet) have a constraint representation so override any provided
         UserTable table = builder.akibanInformationSchema().getUserTable(schemaName, tableName);
         final String constraint;
@@ -330,7 +330,7 @@
         } 
         // Indexes do things a little differently because they need to support Group indexes, Full Text and Geospacial
         else if (cdn.getConstraintType() == ConstraintDefinitionNode.ConstraintType.INDEX) {
-            return generateTableIndex(namer, builder, cdn, table);
+            return generateTableIndex(namer, builder, cdn, table, context);
         } else {
             throw new UnsupportedCheckConstraintException ();
         }
@@ -474,7 +474,8 @@
     private static String generateTableIndex(IndexNameGenerator namer, 
             AISBuilder builder, 
             ConstraintDefinitionNode cdn, 
-            Table table) {
+            Table table,
+            QueryContext context) {
         IndexDefinition id = ((IndexConstraintDefinitionNode)cdn);
         IndexColumnList columnList = id.getIndexColumnList();
         Index tableIndex;
@@ -485,7 +486,7 @@
 
         if (columnList.functionType() == IndexColumnList.FunctionType.FULL_TEXT) {
             logger.debug ("Building Full text index on table {}", table.getName()) ;
-            tableIndex = IndexDDL.buildFullTextIndex (builder, table.getName(), indexName, id);
+            tableIndex = IndexDDL.buildFullTextIndex (builder, table.getName(), indexName, id, context.getServiceManager());
         } else if (IndexDDL.checkIndexType (id, table.getName()) == Index.IndexType.TABLE) {
             logger.debug ("Building Table index on table {}", table.getName()) ;
             tableIndex = IndexDDL.buildTableIndex (builder, table.getName(), indexName, id);

=== modified file 'src/main/resources/com/akiban/server/service/config/configuration-defaults.properties'
--- src/main/resources/com/akiban/server/service/config/configuration-defaults.properties	2013-04-06 05:07:23 +0000
+++ src/main/resources/com/akiban/server/service/config/configuration-defaults.properties	2013-04-15 17:23:26 +0000
@@ -36,6 +36,9 @@
 akserver.bulkload.bufferalloc.pk=0.17
 akserver.bulkload.bufferalloc.group=0.33
 akserver.restrict_user_schema=false
+akserver.text.indexpath=/tmp/aktext
+akserver.text.maintenanceInterval=3000
+akserver.text.populateDelayInterval=1000
 # EXPERIMENTAL
 akserver.indexRowPooling = false
 akserver.pt.osc.hook=disabled

=== modified file 'src/test/java/com/akiban/server/rowdata/SchemaFactory.java'
--- src/test/java/com/akiban/server/rowdata/SchemaFactory.java	2013-04-11 18:35:20 +0000
+++ src/test/java/com/akiban/server/rowdata/SchemaFactory.java	2013-04-15 17:23:26 +0000
@@ -31,6 +31,7 @@
 import com.akiban.server.api.DDLFunctions;
 import com.akiban.server.api.ddl.DDLFunctionsMockBase;
 import com.akiban.server.error.PersistitAdapterException;
+import com.akiban.server.service.ServiceManager;
 import com.akiban.server.service.routines.MockRoutineLoader;
 import com.akiban.server.service.session.Session;
 import com.akiban.sql.StandardException;
@@ -76,7 +77,12 @@
     }
 
     public AkibanInformationSchema ais(String... ddl) {
-        return ais(new AkibanInformationSchema(), ddl);
+        return ais((ServiceManager)null, ddl);
+    }
+    
+    public AkibanInformationSchema ais(ServiceManager sm, String... ddl)
+    {
+        return ais(sm, new AkibanInformationSchema(), ddl);
     }
     
     public static AkibanInformationSchema loadAIS(File fromFile, String defaultSchema) {
@@ -89,11 +95,20 @@
         }
     }
     
+    public AkibanInformationSchema ais(ServiceManager sm, AkibanInformationSchema baseAIS, String... ddl)
+    {
+        return ais(sm, new CreateOnlyDDLMock(baseAIS), null, ddl);
+    }
     public AkibanInformationSchema ais(AkibanInformationSchema baseAIS, String... ddl) {
         return ais(new CreateOnlyDDLMock(baseAIS), null, ddl);
     }
 
-    public AkibanInformationSchema ais(DDLFunctions ddlFunctions, Session session, String... ddl) {
+    public AkibanInformationSchema ais(DDLFunctions ddlFunctions, Session session, String... ddl)
+    {
+        return ais(null, ddlFunctions, session, ddl);
+    }
+    
+    public AkibanInformationSchema ais(ServiceManager sm, DDLFunctions ddlFunctions, Session session, String... ddl) {
         StringBuilder buffer = new StringBuilder();
         for (String line : ddl) {
             buffer.append(line);
@@ -110,7 +125,7 @@
             if (stmt instanceof CreateTableNode) {
                 TableDDL.createTable(ddlFunctions , session , defaultSchema, (CreateTableNode) stmt, null);
             } else if (stmt instanceof CreateIndexNode) {
-                IndexDDL.createIndex(ddlFunctions, session, defaultSchema, (CreateIndexNode) stmt);
+                IndexDDL.createIndex(ddlFunctions, session, defaultSchema, (CreateIndexNode) stmt, sm);
             } else if (stmt instanceof CreateViewNode) {
                 ViewDDL.createView(ddlFunctions, session, defaultSchema, (CreateViewNode) stmt,
                                    new AISBinderContext(ddlFunctions.getAIS(session), defaultSchema), null);

=== modified file 'src/test/java/com/akiban/server/service/text/FullTextIndexServiceIT.java'
--- src/test/java/com/akiban/server/service/text/FullTextIndexServiceIT.java	2013-03-22 20:05:57 +0000
+++ src/test/java/com/akiban/server/service/text/FullTextIndexServiceIT.java	2013-04-15 17:23:26 +0000
@@ -18,6 +18,7 @@
 package com.akiban.server.service.text;
 
 import com.akiban.ais.model.FullTextIndex;
+import com.akiban.ais.model.IndexName;
 import com.akiban.qp.operator.Operator;
 import com.akiban.qp.operator.QueryContext;
 import static com.akiban.qp.operator.API.cursor;
@@ -27,9 +28,13 @@
 import com.akiban.qp.rowtype.Schema;
 import com.akiban.qp.util.SchemaCache;
 import com.akiban.server.service.servicemanager.GuicedServiceManager;
+import com.akiban.server.service.session.Session;
+import com.akiban.server.service.session.SessionServiceImpl;
 import com.akiban.server.test.it.ITBase;
 import com.akiban.server.test.it.qp.TestRow;
 
+import com.persistit.Exchange;
+import com.persistit.exception.PersistitException;
 import org.junit.Before;
 import org.junit.Test;
 import static org.junit.Assert.*;
@@ -39,12 +44,18 @@
 public class FullTextIndexServiceIT extends ITBase
 {
     public static final String SCHEMA = "test";
-
+    private static final long updateInterval = 1000L; // for testing
+    private static final long populateDelayInterval = 1000L;
     protected FullTextIndexService fullText;
     protected Schema schema;
     protected PersistitAdapter adapter;
     protected QueryContext queryContext;
 
+    private int c;
+    private int o;
+    private int i;
+    private int a;
+    
     @Override
     protected GuicedServiceManager.BindingsConfigurationProvider serviceBindingsProvider() {
         return super.serviceBindingsProvider()
@@ -55,29 +66,31 @@
     protected Map<String, String> startupConfigProperties() {
         Map<String, String> properties = new HashMap<>();
         properties.put("akserver.text.indexpath", "/tmp/aktext");
+        properties.put(FullTextIndexServiceImpl.UPDATE_INTERVAL, Long.toString(updateInterval));
+        properties.put(FullTextIndexServiceImpl.POPULATE_DELAY_INTERVAL, Long.toString(populateDelayInterval));
         return properties;
     }
 
     @Before
     public void createData() {
-        int c = createTable(SCHEMA, "c",
-                            "cid INT PRIMARY KEY NOT NULL",
-                            "name VARCHAR(128) COLLATE en_us_ci");
-        int o = createTable(SCHEMA, "o",
-                            "oid INT PRIMARY KEY NOT NULL",
-                            "cid INT NOT NULL",
-                            "GROUPING FOREIGN KEY(cid) REFERENCES c(cid)",
-                            "order_date DATE");
-        int i = createTable(SCHEMA, "i",
-                            "iid INT PRIMARY KEY NOT NULL",
-                            "oid INT NOT NULL",
-                            "GROUPING FOREIGN KEY(oid) REFERENCES o(oid)",
-                            "sku VARCHAR(10) NOT NULL");
-        int a = createTable(SCHEMA, "a",
-                            "aid INT PRIMARY KEY NOT NULL",
-                            "cid INT NOT NULL",
-                            "GROUPING FOREIGN KEY(cid) REFERENCES c(cid)",
-                            "state CHAR(2)");
+        c = createTable(SCHEMA, "c",
+                        "cid INT PRIMARY KEY NOT NULL",
+                        "name VARCHAR(128) COLLATE en_us_ci");
+        o = createTable(SCHEMA, "o",
+                        "oid INT PRIMARY KEY NOT NULL",
+                        "cid INT NOT NULL",
+                        "GROUPING FOREIGN KEY(cid) REFERENCES c(cid)",
+                        "order_date DATE");
+        i = createTable(SCHEMA, "i",
+                        "iid INT PRIMARY KEY NOT NULL",
+                        "oid INT NOT NULL",
+                        "GROUPING FOREIGN KEY(oid) REFERENCES o(oid)",
+                        "sku VARCHAR(10) NOT NULL");
+        a = createTable(SCHEMA, "a",
+                        "aid INT PRIMARY KEY NOT NULL",
+                        "cid INT NOT NULL",
+                        "GROUPING FOREIGN KEY(cid) REFERENCES c(cid)",
+                        "state CHAR(2)");
         writeRow(c, 1, "Fred Flintstone");
         writeRow(o, 101, 1, "2012-12-12");
         writeRow(i, 10101, 101, "ABCD");
@@ -99,11 +112,286 @@
     }
 
     @Test
-    public void cDown() {
-        FullTextIndex index = createFullTextIndex(SCHEMA, "c", "idx_c", 
-                                                  "name", "i.sku", "a.state");
-        fullText.createIndex(session(), index.getIndexName());
-
+    public void testPopulateScheduling() throws InterruptedException, PersistitException
+    {
+        // This test is specifically for FullTextIndexServiceImpl.java
+        assertEquals(FullTextIndexServiceImpl.class, fullText.getClass());
+        FullTextIndexServiceImpl fullTextImpl = (FullTextIndexServiceImpl)fullText;
+        
+        // disable the populate worker (so it doesn't read all the entries
+        // out before we get a chance to look at the tree.
+        fullTextImpl.disablePopulateWorker();
+        
+        // create 3 indices
+        final FullTextIndex expecteds[] = new FullTextIndex[]
+        {
+            createFullTextIndex(serviceManager(),
+                                SCHEMA, "c", "idx1_c",
+                                "name"),
+        
+            createFullTextIndex(serviceManager(),
+                                SCHEMA, "c", "idx2_c",
+                                "i.sku"),
+            createFullTextIndex(serviceManager(),
+                                SCHEMA, "c", "idx3_c",
+                                "name", "i.sku")
+        };
+
+        
+        // read the entries out
+        traverse(fullTextImpl,
+                 new Visitor()
+                 {
+                    int n = 0;
+                    public void visit(IndexName idx)
+                    {
+                        assertEquals("entry[" + n + "]", expecteds[n++].getIndexName(),
+                                                         idx);
+                    }
+
+                    public void endOfTree()
+                    {
+                        assertEquals(expecteds.length, n);
+                    }
+                 });
+        
+
+        // let the worker do its job.
+        // (After it is done, the tree had better be empty)
+        fullTextImpl.enablePopulateWorker();
+        Thread.sleep(populateDelayInterval * 5);
+
+        traverse(fullTextImpl,
+                 new Visitor()
+                 {
+                     int n = 0;
+                    
+                     @Override
+                     public void visit(IndexName idx)
+                     {
+                         ++n;
+                     }
+
+                     @Override
+                     public void endOfTree()
+                     {
+                         assertEquals (0, n);
+                     } 
+                 });
+        
+        // drop all the indices
+        Session session = new SessionServiceImpl().createSession();
+        for (FullTextIndex idx : expecteds)
+            fullTextImpl.dropIndex(session, idx.getIndexName());
+        session.close();
+    }
+
+
+    @Test
+    public void testDeleteIndex() throws PersistitException, InterruptedException
+    {
+     
+        // This test is specifically for FullTextIndexServiceImpl.java
+        assertEquals(FullTextIndexServiceImpl.class, fullText.getClass());
+        FullTextIndexServiceImpl fullTextImpl = (FullTextIndexServiceImpl)fullText;
+         
+        // <1> disable worker
+        fullTextImpl.disablePopulateWorker();
+
+        // <2> create 3 new indices
+         // create 3 indices
+        final FullTextIndex expecteds[] = new FullTextIndex[]
+        {
+            createFullTextIndex(serviceManager(),
+                                SCHEMA, "c", "idx4_c",
+                                "a.state"),
+        
+            createFullTextIndex(serviceManager(),
+                                SCHEMA, "c", "idx5_c",
+                                "i.sku", "a.state"),
+            createFullTextIndex(serviceManager(),
+                                SCHEMA, "c", "idx6_c",
+                                "name", "i.sku")
+        };
+
+        // <3> delete 2 of them
+        Session session = new SessionServiceImpl().createSession();
+        boolean sawNPE = false;
+        try
+        {
+            fullTextImpl.dropIndex(session, expecteds[0].getIndexName());
+        }
+        catch (NullPointerException e) // NPE is expected because, the index hasn't been
+        {                              // populated yet. But we're not testing that!
+            sawNPE = true;
+        }
+        assertTrue("NPE should have happened", sawNPE);
+        
+
+        sawNPE = false;
+        try
+        {
+            fullTextImpl.dropIndex(session, expecteds[1].getIndexName());
+        }
+        catch (NullPointerException e) // NPE is expected because, the index hasn't been
+        {                              // populated yet. But we're not testing that!
+            sawNPE = true;
+        }
+        assertTrue("NPE should have happened", sawNPE);
+        
+        // <4> check that the tree only has one entry now (ie., epxecteds2[2]
+        traverse(fullTextImpl,
+                 new Visitor()
+                 {
+                     int n = 0;
+                    
+                     @Override
+                     public void visit(IndexName idx)
+                     {
+                         assertEquals("entry[" + n++ + "]", expecteds[2].getIndexName(),
+                                                            idx);
+                     }
+
+                     @Override
+                     public void endOfTree()
+                     {
+                         assertEquals (1, n);
+                     } 
+                 });
+
+        // wake the worker up to do its job
+        fullTextImpl.enablePopulateWorker();
+        Thread.sleep(populateDelayInterval * 5);
+        
+        // drop the remaining index
+        fullTextImpl.dropIndex(session, expecteds[2].getIndexName());
+        session.close();
+    }
+
+    private static interface Visitor
+    {
+        void visit(IndexName idx);
+        void endOfTree();
+    }
+    
+    
+    private static void traverse(FullTextIndexServiceImpl serv,
+                                 Visitor visitor) throws PersistitException
+    {
+         Session session = new SessionServiceImpl().createSession();
+
+         try
+         {
+             Exchange ex = serv.getPopulateExchange(session);
+             IndexName toPopulate;
+             while ((toPopulate = serv.nextInQueue(ex)) != null)
+                 visitor.visit(toPopulate);
+             visitor.endOfTree();
+         }
+         finally
+         {
+             session.close();
+         }
+    }
+
+    @Test
+    public void testUpdate() throws InterruptedException
+    {
+        
+        /*
+            Test plans:
+            1 - do a search, confirm that the rows come back as expected
+            2 - (With update worker enabled)
+                + insert new rows
+                + sleep for sometime
+                + do the search again and confirm that the new rows are found
+            3 - (With update worker NOT enable)
+                + disable update worker
+                + insert new rows
+                + do the search again and confirm that the new rows are NOT found
+
+         */
+
+        //CREATE INDEX cust_ft ON customers(FULL_TEXT(name, addresses.state, items.sku))
+
+        // part 1
+        FullTextIndex index = createFullTextIndex(serviceManager(),
+                                                  SCHEMA, "c", "idx_c", 
+                                                  "name", "i.sku", "a.state");
+        //fullText.createIndex(session(), index.getIndexName());
+        Thread.sleep(populateDelayInterval * 5);
+        RowType rowType = rowType("c");
+        RowBase[] expected1 = new RowBase[]
+        {
+            row(rowType, 1L),
+            row(rowType, 3L)
+        };
+        FullTextQueryBuilder builder = new FullTextQueryBuilder(index, ais(), queryContext);
+        Operator plan = builder.scanOperator("flintstone", 15);
+        compareRows(expected1, cursor(plan, queryContext));
+
+        plan = builder.scanOperator("state:MA", 15);
+        compareRows(expected1, cursor(plan, queryContext));
+        
+        
+        // part 2
+        // write new rows
+        writeRow(c, 4, "John Watson");
+        writeRow(c, 5, "Sherlock Flintstone");
+        writeRow(c, 6, "Mycroft Holmes");
+        writeRow(c, 7, "Flintstone Lestrade");
+        
+        // sleep a bit (wait for the worker to do the update)
+        Thread.sleep(updateInterval * 5);
+        RowBase expected2[] = new RowBase[]
+        {
+            row(rowType, 1L),
+            row(rowType, 3L),
+            row(rowType, 5L),
+            row(rowType, 7L)
+        };
+
+        // confirm new changes
+        plan = builder.scanOperator("flintstone", 15);
+        compareRows(expected2, cursor(plan, queryContext));
+        
+        // part 3
+        ((FullTextIndexServiceImpl)fullText).disableUpdateWorker();
+        
+        writeRow(c, 8, "Flintstone Hudson");
+        writeRow(c, 9, "Jim Flintstone");
+        
+        Thread.sleep(updateInterval * 2);
+        
+        // confirm that new rows are not found (ie., expected2 still works)
+        plan = builder.scanOperator("flintstone", 15);
+        compareRows(expected2, cursor(plan, queryContext));
+        
+        ((FullTextIndexServiceImpl)fullText).enableUpdateWorker();
+        Thread.sleep(updateInterval * 5);
+        
+        // now the rows should be seen.
+        // (Because disabling the worker does not stop the changes fron being recorded)
+        RowBase expected3[] = new RowBase[]
+        {
+            row(rowType, 1L),
+            row(rowType, 3L),
+            row(rowType, 5L),
+            row(rowType, 7L),
+            row(rowType, 8L),
+            row(rowType, 9L)
+        };
+
+        plan = builder.scanOperator("flintstone", 15);
+        compareRows(expected3, cursor(plan, queryContext));
+    }
+
+    @Test
+    public void cDown() throws InterruptedException {
+        FullTextIndex index = createFullTextIndex(serviceManager(),
+                                                  SCHEMA, "c", "idx_c", 
+                                                  "name", "i.sku", "a.state");
+        Thread.sleep(populateDelayInterval * 5);
         RowType rowType = rowType("c");
         RowBase[] expected = new RowBase[] {
             row(rowType, 1L),
@@ -115,14 +403,15 @@
 
         plan = builder.scanOperator("state:MA", 10);
         compareRows(expected, cursor(plan, queryContext));
+        
     }
 
     @Test
-    public void oUpDown() {
-        FullTextIndex index = createFullTextIndex(SCHEMA, "o", "idx_o",
+    public void oUpDown() throws InterruptedException {
+        FullTextIndex index = createFullTextIndex(serviceManager(),
+                                                  SCHEMA, "o", "idx_o",
                                                   "c.name", "i.sku");
-        fullText.createIndex(session(), index.getIndexName());
-
+        Thread.sleep(populateDelayInterval * 5);
         RowType rowType = rowType("o");
         RowBase[] expected = new RowBase[] {
             row(rowType, 1L, 101L)

=== modified file 'src/test/java/com/akiban/server/test/ApiTestBase.java'
--- src/test/java/com/akiban/server/test/ApiTestBase.java	2013-03-22 20:05:57 +0000
+++ src/test/java/com/akiban/server/test/ApiTestBase.java	2013-04-15 17:23:26 +0000
@@ -520,16 +520,16 @@
      * overriding the {@link #startupConfigProperties()} and/or
      * {@link #serviceBindingsProvider()} methods.
      */
-    protected static Map<String, String> uniqueStartupConfigProperties(Class clazz) {
+    protected Map<String, String> uniqueStartupConfigProperties(Class clazz) {
         return Collections.singletonMap(
                 "test.services",
                 clazz.getName()
         );
     }
 
-    protected AkibanInformationSchema createFromDDL(String schema, String ddl) {
+    protected AkibanInformationSchema createFromDDL(ServiceManager sm, String schema, String ddl) {
         SchemaFactory schemaFactory = new SchemaFactory(schema);
-        return schemaFactory.ais(ddl().getAIS(session()), ddl);
+        return schemaFactory.ais(sm, ddl().getAIS(session()), ddl);
     }
 
     protected static final class SimpleColumn {
@@ -602,7 +602,7 @@
 
     protected final int createTable(String schema, String table, String definition) throws InvalidOperationException {
         String ddl = String.format("CREATE TABLE \"%s\" (%s)", table, definition);
-        AkibanInformationSchema tempAIS = createFromDDL(schema, ddl);
+        AkibanInformationSchema tempAIS = createFromDDL(serviceManager(), schema, ddl);
         UserTable tempTable = tempAIS.getUserTable(schema, table);
         ddl().createTable(session(), tempTable);
         updateAISGeneration();
@@ -621,7 +621,7 @@
     
     protected final void createSequence (String schema, String name, String definition) {
         String ddl = String.format("CREATE SEQUENCE %s %s", name, definition);
-        AkibanInformationSchema tempAIS = createFromDDL(schema, ddl);
+        AkibanInformationSchema tempAIS = createFromDDL(serviceManager(), schema, ddl);
         Sequence sequence = tempAIS.getSequence(new TableName(schema, name));
         ddl().createSequence(session(), sequence);
         updateAISGeneration();
@@ -629,7 +629,7 @@
 
     protected final void createView(String schema, String name, String definition) {
         String ddl = String.format("CREATE VIEW %s AS %s", name, definition);
-        AkibanInformationSchema tempAIS = createFromDDL(schema, ddl);
+        AkibanInformationSchema tempAIS = createFromDDL(serviceManager(), schema, ddl);
         View view = tempAIS.getView(new TableName(schema, name));
         ddl().createView(session(), view);
         updateAISGeneration();
@@ -664,7 +664,7 @@
                                    schema,
                                    table,
                                    Strings.join(Arrays.asList(indexCols), ","));
-        return createFromDDL(schema, ddl);
+        return createFromDDL(serviceManager(), schema, ddl);
     }
 
     protected final TableIndex createIndex(String schema, String table, String indexName, String... indexCols) {
@@ -767,7 +767,7 @@
         return ddl().getAIS(session()).getGroup(groupName).getIndex(indexName);
     }
 
-    protected final FullTextIndex createFullTextIndex(String schema, String table, String indexName, String... indexCols) {
+    protected final FullTextIndex createFullTextIndex(ServiceManager sm, String schema, String table, String indexName, String... indexCols) {
         AkibanInformationSchema tempAIS = createIndexInternal(schema, table, indexName, "FULL_TEXT(" + Strings.join(Arrays.asList(indexCols), ",") + ")");
         Index tempIndex = tempAIS.getUserTable(schema, table).getFullTextIndex(indexName);
         ddl().createIndexes(session(), Collections.singleton(tempIndex));
@@ -779,10 +779,10 @@
         SchemaFactory schemaFactory = new SchemaFactory(schema);
         
         // Insert DDL into the System 
-        AkibanInformationSchema ais = schemaFactory.ais(ddl(), session(), ddl);
+        AkibanInformationSchema ais = schemaFactory.ais(serviceManager(), ddl(), session(), ddl);
         
         // sort DDL to find first root table of the user schema
-        ais = schemaFactory.ais (ddl);
+        ais = schemaFactory.ais (serviceManager(), ddl);
         List<UserTable> tables = new ArrayList<>(ais.getUserTables().values());
         Collections.sort(tables, new Comparator<UserTable>() {
             @Override

=== modified file 'src/test/java/com/akiban/server/test/it/bugs/bug701614/MissingColumnsIT.java'
--- src/test/java/com/akiban/server/test/it/bugs/bug701614/MissingColumnsIT.java	2013-03-22 20:05:57 +0000
+++ src/test/java/com/akiban/server/test/it/bugs/bug701614/MissingColumnsIT.java	2013-04-15 17:23:26 +0000
@@ -38,7 +38,7 @@
 
     private int loadBlocksTable() throws InvalidOperationException, IOException {
         final String blocksDDL = Strings.join(Strings.dumpResource(getClass(), "blocks-table.ddl"));
-        AkibanInformationSchema tempAIS = createFromDDL("drupal", blocksDDL);
+        AkibanInformationSchema tempAIS = createFromDDL(serviceManager(), "drupal", blocksDDL);
         ddl().createTable(session(), tempAIS.getUserTable("drupal", "blocks"));
         updateAISGeneration();
         AkibanInformationSchema ais = ddl().getAIS(session());

=== modified file 'src/test/java/com/akiban/sql/optimizer/OptimizerTestBase.java'
--- src/test/java/com/akiban/sql/optimizer/OptimizerTestBase.java	2013-03-22 20:05:57 +0000
+++ src/test/java/com/akiban/sql/optimizer/OptimizerTestBase.java	2013-04-15 17:23:26 +0000
@@ -25,18 +25,26 @@
 import com.akiban.sql.views.ViewDefinition;
 
 import com.akiban.ais.model.AkibanInformationSchema;
+import com.akiban.server.service.ServiceManager;
+import com.akiban.server.service.config.TestConfigService;
 import com.akiban.server.service.functions.FunctionsRegistryImpl;
 
+import com.akiban.server.service.servicemanager.GuicedServiceManager;
+import com.akiban.server.service.text.FullTextIndexService;
+import com.akiban.server.service.text.FullTextIndexServiceImpl;
 import org.junit.Before;
 import org.junit.Ignore;
 
 import java.io.File;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 
 @Ignore
 public class OptimizerTestBase extends ASTTransformTestBase
 {
+    
     protected OptimizerTestBase(String caseName, String sql, 
                                 String expected, String error) {
         super(caseName, sql, expected, error);
@@ -47,13 +55,17 @@
                  + OptimizerTestBase.class.getPackage().getName().replace('.', '/'));
     public static final String DEFAULT_SCHEMA = "test";
 
+    protected static ServiceManager sm = prepareServices();
+    protected static int populateDelayInterval = 1000;
+    protected static int updateInterval = 3000;
+    
     // Base class has all possible transformers for convenience.
     protected AISBinder binder;
     protected TypeComputer typeComputer;
     protected BooleanNormalizer booleanNormalizer;
     protected SubqueryFlattener subqueryFlattener;
     protected DistinctEliminator distinctEliminator;
-
+    
     @Before
     public void makeTransformers() throws Exception {
         parser = new SQLParser();
@@ -65,13 +77,19 @@
         distinctEliminator = new DistinctEliminator(parser);
     }
 
+    protected static ServiceManager serviceManager()
+    {
+        return sm;
+    }
+
     public static AkibanInformationSchema parseSchema(List<File> ddls) throws Exception {
         StringBuilder ddlBuilder = new StringBuilder();
         for (File ddl : ddls)
             ddlBuilder.append(fileContents(ddl));
         String sql = ddlBuilder.toString();
         SchemaFactory schemaFactory = new SchemaFactory(DEFAULT_SCHEMA);
-        return schemaFactory.ais(sql);
+        AkibanInformationSchema ret = schemaFactory.ais(serviceManager(), sql);
+        return ret;
     }
 
     public static AkibanInformationSchema parseSchema(File ddl) throws Exception {
@@ -98,4 +116,35 @@
         }
     }
 
+     
+    protected static ServiceManager createServiceManager(Map<String, String> startupConfigProperties)
+    {
+        return new GuicedServiceManager(serviceBindingsProvider());
+    }
+
+    protected static GuicedServiceManager.BindingsConfigurationProvider serviceBindingsProvider()
+    {
+        return GuicedServiceManager
+                    .testUrls()
+                        .bindAndRequire(FullTextIndexService.class,
+                                        FullTextIndexServiceImpl.class);
+    }
+    
+
+    protected static Map<String, String> startupConfigProperties() {
+        Map<String, String> properties = new HashMap<>();
+        properties.put("akserver.text.indexpath", "/tmp/aktext");
+        properties.put(FullTextIndexServiceImpl.UPDATE_INTERVAL, Long.toString(updateInterval));
+        properties.put(FullTextIndexServiceImpl.POPULATE_DELAY_INTERVAL, Long.toString(populateDelayInterval));
+        return properties;
+    }
+    
+    private static ServiceManager prepareServices()
+    {
+        System.setProperty("akiban.home", System.getProperty("user.home"));
+        ServiceManager ret = createServiceManager(startupConfigProperties());
+        ret.startServices();
+        
+        return ret;
+    }
 }

=== added file 'src/test/resources/com/akiban/sql/pg/yaml/functional/test-drop-fulltext.yaml'
--- src/test/resources/com/akiban/sql/pg/yaml/functional/test-drop-fulltext.yaml	1970-01-01 00:00:00 +0000
+++ src/test/resources/com/akiban/sql/pg/yaml/functional/test-drop-fulltext.yaml	2013-04-15 17:23:26 +0000
@@ -0,0 +1,8 @@
+# test drop full_text index
+---
+- CreateTable: t(id int, name varchar(224));
+---
+- Statement: CREATE INDEX idx on t(full_text(name));
+---
+- Statement: DROP INDEX t.idx;
+...
\ No newline at end of file

=== added file 'src/test/resources/com/akiban/sql/pg/yaml/functional/test-ft-with-inherited-key.yaml'
--- src/test/resources/com/akiban/sql/pg/yaml/functional/test-ft-with-inherited-key.yaml	1970-01-01 00:00:00 +0000
+++ src/test/resources/com/akiban/sql/pg/yaml/functional/test-ft-with-inherited-key.yaml	2013-04-15 17:23:26 +0000
@@ -0,0 +1,28 @@
+# test full-text updates with inherited keys
+---
+- CreateTable: t1(id INT NOT NULL PRIMARY KEY);
+---
+- CreateTable: t2(id INT NOT NULL PRIMARY KEY, pid INT NOT NULL, GROUPING FOREIGN KEY(pid) REFERENCES t1(id));
+---
+- CreateTable: t3(id INT NOT NULL PRIMARY KEY, pid INT NOT NULL, GROUPING FOREIGN KEY(pid) REFERENCES t2(id), name VARCHAR(16) COLLATE en_ci);
+---
+- Statement: INSERT INTO t1 VALUES(1),(2);
+---
+- Statement: INSERT INTO t2 VALUES(11, 1);
+---
+- Statement: INSERT INTO t3 VALUES(111, 11, 'Fred'),(112, 11, 'Barney');
+---
+- Statement: CREATE INDEX t3_ft ON t3(FULL_TEXT(name));
+---
+- Statement: SELECT SLEEP(7000);
+---
+- Statement: SELECT * FROM t3 WHERE FULL_TEXT_SEARCH(name, 'fred');
+- output: [[111, 11, 'Fred']]
+---
+- Statement: UPDATE t2 SET pid = 2 WHERE pid = 1;
+---
+- Statement: SELECT SLEEP(7000);
+---
+- Statement: SELECT * FROM t3 WHERE FULL_TEXT_SEARCH(name, 'fred');
+- output: [[111, 11, 'Fred']]
+...
\ No newline at end of file

=== added file 'src/test/resources/com/akiban/sql/pg/yaml/functional/test-ft-with-pk.yaml'
--- src/test/resources/com/akiban/sql/pg/yaml/functional/test-ft-with-pk.yaml	1970-01-01 00:00:00 +0000
+++ src/test/resources/com/akiban/sql/pg/yaml/functional/test-ft-with-pk.yaml	2013-04-15 17:23:26 +0000
@@ -0,0 +1,22 @@
+# Tests with changed primary key
+---
+- CreateTable: t (id int primary key not null, name varchar(224));
+---
+- Statement: INSERT INTO t values (1, 'foo1');
+---
+- Statement: INSERT INTO t values (2, 'foo2');
+---
+- Statement: create index idx1 on t (full_text(name));
+---
+- Statement: SELECT SLEEP(7000);
+---
+- Statement: SELECT id FROM t where full_text_search(name = 'foo1');
+- output: [[1]]
+---
+- Statement: UPDATE t SET id = 3 where id = 1;
+---
+- Statement: SELECT SLEEP(7000);
+---
+- Statement: SELECT id FROM t where full_text_search(name = 'foo1');
+- output: [[3]]
+...
\ No newline at end of file

=== added file 'src/test/resources/com/akiban/sql/pg/yaml/functional/test-fulltext-maintenance.yaml'
--- src/test/resources/com/akiban/sql/pg/yaml/functional/test-fulltext-maintenance.yaml	1970-01-01 00:00:00 +0000
+++ src/test/resources/com/akiban/sql/pg/yaml/functional/test-fulltext-maintenance.yaml	2013-04-15 17:23:26 +0000
@@ -0,0 +1,49 @@
+# test that full text indices are updated (and populated) properly
+---
+- CreateTable: t (id int, name varchar(224));
+---
+- Statement: INSERT INTO  t VALUES (1, 'bar1');
+---
+- Statement: INSERT INTO  t VALUES (2, 'bar1');
+---
+- Statement: INSERT INTO  t VALUES (3, 'bar3');
+---
+- Statement: INSERT INTO  t VALUES (4, 'bar4');
+---
+- Statement: CREATE INDEX idx ON t (full_text(name));
+--- # should be long enough! (Could query the actual interval from AIS, but we'd need to parse the string!)
+- Statement: SELECT SLEEP (7000);
+---
+- Statement: select id from t where full_text_search(name = 'bar1');
+- output: [[1], [2]]
+---
+- Statement: UPDATE t set name='foo' WHERE id = 1;
+---
+- Statement: SELECT SLEEP(7000);
+---
+- Statement: select id from t where full_text_search(name = 'bar1');
+- output: [[2]]
+---
+- Statement: SELECT id from t where full_text_search(name = 'bar4');
+- output: [[4]]
+---
+- Statement: DELETE FROM t where id = 4;
+---
+- Statement: SELECT SLEEP(7000);
+---
+- Statement: SELECT id from t where full_text_search(name = 'bar4');
+- row_count: 0
+---
+- Create
+...
+
+CREATE TABLE t1(id INT NOT NULL PRIMARY KEY);
+CREATE TABLE t2(id INT NOT NULL PRIMARY KEY, pid INT NOT NULL, GROUPING FOREIGN KEY(pid) REFERENCES t1(id));
+CREATE TABLE t3(id INT NOT NULL PRIMARY KEY, pid INT NOT NULL, GROUPING FOREIGN KEY(pid) REFERENCES t2(id), name VARCHAR(16) COLLATE en_ci);
+INSERT INTO t1 VALUES(1),(2);
+INSERT INTO t2 VALUES(11, 1);
+INSERT INTO t3 VALUES(111, 11, 'Fred'),(112, 11, 'Barney');
+CREATE INDEX t3_ft ON t3(FULL_TEXT(name));
+SELECT * FROM t3 WHERE FULL_TEXT_SEARCH(name, 'fred');
+UPDATE t2 SET pid = 2 WHERE pid = 1;
+SELECT * FROM t3 WHERE FULL_TEXT_SEARCH(name, 'fred');