Add a new algorithm for handling INSERT which reduces fragmentation on

a VACUUM.  Ticket #2075.  More testing needed. (CVS 3643)

FossilOrigin-Name: 9f56a878cbbc715262b3a48ee696148dbd7bf1d2

1 files changed, 453 insertions, 52 deletions

diff --git a/src/insert.c b/src/insert.c
index e10867316..d96780663 100644
--- a/src/insert.c
+++ b/src/insert.c
@@ -12,7 +12,7 @@
 ** This file contains C code routines that are called by the parser
 ** to handle INSERT statements in SQLite.
 **
-** $Id: insert.c,v 1.172 2006/08/29 18:46:14 drh Exp $
+** $Id: insert.c,v 1.173 2007/02/13 15:01:11 drh Exp $
 */
 #include "sqliteInt.h"
 
@@ -118,6 +118,117 @@ static int selectReadsTable(Select *p, Schema *pSchema, int iTab){
   return 0;
 }
 
+#ifndef SQLITE_OMIT_AUTOINCREMENT
+/*
+** Write out code to initialize the autoincrement logic.  This code
+** looks up the current autoincrement value in the sqlite_sequence
+** table and stores that value in a memory cell.  Code generated by
+** autoIncStep() will keep that memory cell holding the largest
+** rowid value.  Code generated by autoIncEnd() will write the new
+** largest value of the counter back into the sqlite_sequence table.
+**
+** This routine returns the index of the mem[] cell that contains
+** the maximum rowid counter.
+**
+** Two memory cells are allocated.  The next memory cell after the
+** one returned holds the rowid in sqlite_sequence where we will
+** write back the revised maximum rowid.
+*/
+static int autoIncBegin(
+  Parse *pParse,      /* Parsing context */
+  int iDb,            /* Index of the database holding pTab */
+  Table *pTab         /* The table we are writing to */
+){
+  int memId = 0;
+  if( pTab->autoInc ){
+    Vdbe *v = pParse->pVdbe;
+    Db *pDb = &pParse->db->aDb[iDb];
+    int iCur = pParse->nTab;
+    int addr;
+    assert( v );
+    addr = sqlite3VdbeCurrentAddr(v);
+    memId = pParse->nMem+1;
+    pParse->nMem += 2;
+    sqlite3OpenTable(pParse, iCur, iDb, pDb->pSchema->pSeqTab, OP_OpenRead);
+    sqlite3VdbeAddOp(v, OP_Rewind, iCur, addr+13);
+    sqlite3VdbeAddOp(v, OP_Column, iCur, 0);
+    sqlite3VdbeOp3(v, OP_String8, 0, 0, pTab->zName, 0);
+    sqlite3VdbeAddOp(v, OP_Ne, 0x100, addr+12);
+    sqlite3VdbeAddOp(v, OP_Rowid, iCur, 0);
+    sqlite3VdbeAddOp(v, OP_MemStore, memId-1, 1);
+    sqlite3VdbeAddOp(v, OP_Column, iCur, 1);
+    sqlite3VdbeAddOp(v, OP_MemStore, memId, 1);
+    sqlite3VdbeAddOp(v, OP_Goto, 0, addr+13);
+    sqlite3VdbeAddOp(v, OP_Next, iCur, addr+4);
+    sqlite3VdbeAddOp(v, OP_Close, iCur, 0);
+  }
+  return memId;
+}
+
+/*
+** Update the maximum rowid for an autoincrement calculation.
+**
+** This routine should be called when the top of the stack holds a
+** new rowid that is about to be inserted.  If that new rowid is
+** larger than the maximum rowid in the memId memory cell, then the
+** memory cell is updated.  The stack is unchanged.
+*/
+static void autoIncStep(Parse *pParse, int memId){
+  if( memId>0 ){
+    sqlite3VdbeAddOp(pParse->pVdbe, OP_MemMax, memId, 0);
+  }
+}
+
+/*
+** After doing one or more inserts, the maximum rowid is stored
+** in mem[memId].  Generate code to write this value back into the
+** the sqlite_sequence table.
+*/
+static void autoIncEnd(
+  Parse *pParse,     /* The parsing context */
+  int iDb,           /* Index of the database holding pTab */
+  Table *pTab,       /* Table we are inserting into */
+  int memId          /* Memory cell holding the maximum rowid */
+){
+  if( pTab->autoInc ){
+    int iCur = pParse->nTab;
+    Vdbe *v = pParse->pVdbe;
+    Db *pDb = &pParse->db->aDb[iDb];
+    int addr;
+    assert( v );
+    addr = sqlite3VdbeCurrentAddr(v);
+    sqlite3OpenTable(pParse, iCur, iDb, pDb->pSchema->pSeqTab, OP_OpenWrite);
+    sqlite3VdbeAddOp(v, OP_MemLoad, memId-1, 0);
+    sqlite3VdbeAddOp(v, OP_NotNull, -1, addr+7);
+    sqlite3VdbeAddOp(v, OP_Pop, 1, 0);
+    sqlite3VdbeAddOp(v, OP_NewRowid, iCur, 0);
+    sqlite3VdbeOp3(v, OP_String8, 0, 0, pTab->zName, 0);
+    sqlite3VdbeAddOp(v, OP_MemLoad, memId, 0);
+    sqlite3VdbeAddOp(v, OP_MakeRecord, 2, 0);
+    sqlite3VdbeAddOp(v, OP_Insert, iCur, 0);
+    sqlite3VdbeAddOp(v, OP_Close, iCur, 0);
+  }
+}
+#else
+/*
+** If SQLITE_OMIT_AUTOINCREMENT is defined, then the three routines
+** above are all no-ops
+*/
+# define autoIncBegin(A,B,C) (0)
+# define autoIncStep(A,B)
+# define autoIncEnd(A,B,C,D)
+#endif /* SQLITE_OMIT_AUTOINCREMENT */
+
+
+/* Forward declaration */
+static int xferOptimization(
+  Parse *pParse,        /* Parser context */
+  Table *pDest,         /* The table we are inserting into */
+  Select *pSelect,      /* A SELECT statement to use as the data source */
+  int onError,          /* How to handle constraint errors */
+  int iDbDest           /* The database of pDest */
+);
+
 /*
 ** This routine is call to handle SQL of the following forms:
 **
@@ -133,7 +244,7 @@ static int selectReadsTable(Select *p, Schema *pSchema, int iTab){
 ** NULL and pSelect is a pointer to the select statement used to generate
 ** data for the insert.
 **
-** The code generated follows one of three templates.  For a simple
+** The code generated follows one of four templates.  For a simple
 ** select with data coming from a VALUES clause, the code executes
 ** once straight down through.  The template looks like this:
 **
@@ -142,16 +253,37 @@ static int selectReadsTable(Select *p, Schema *pSchema, int iTab){
 **         write the resulting record into <table>
 **         cleanup
 **
-** If the statement is of the form
+** The three remaining templates assume the statement is of the form
 **
 **   INSERT INTO <table> SELECT ...
 **
-** And the SELECT clause does not read from <table> at any time, then
-** the generated code follows this template:
+** If the SELECT clause is of the restricted form "SELECT * FROM <table2>" -
+** in other words if the SELECT pulls all columns from a single table
+** and there is no WHERE or LIMIT or GROUP BY or ORDER BY clauses, and
+** if <table2> and <table1> are distinct tables but have identical
+** schemas, including all the same indices, then a special optimization
+** is invoked that copies raw records from <table2> over to <table1>.
+** See the xferOptimization() function for the implementation of this
+** template.  This is the second template.
+**
+**         open a write cursor to <table>
+**         open read cursor on <table2>
+**         transfer all records in <table2> over to <table>
+**         close cursors
+**         foreach index on <table>
+**           open a write cursor on the <table> index
+**           open a read cursor on the corresponding <table2> index
+**           transfer all records from the read to the write cursors
+**           close cursors
+**         end foreach
+**
+** The third template is for when the second template does not apply
+** and the SELECT clause does not read from <table> at any time.
+** The generated code follows this template:
 **
 **         goto B
 **      A: setup for the SELECT
-**         loop over the tables in the SELECT
+**         loop over the rows in the SELECT
 **           gosub C
 **         end loop
 **         cleanup after the SELECT
@@ -162,7 +294,7 @@ static int selectReadsTable(Select *p, Schema *pSchema, int iTab){
 **         return
 **      D: cleanup
 **
-** The third template is used if the insert statement takes its
+** The fourth template is used if the insert statement takes its
 ** values from a SELECT but the data is being inserted into a table
 ** that is also read as part of the SELECT.  In the third form,
 ** we have to use a intermediate table to store the results of
@@ -221,10 +353,6 @@ void sqlite3Insert(
   int triggers_exist = 0;     /* True if there are FOR EACH ROW triggers */
 #endif
 
-#ifndef SQLITE_OMIT_AUTOINCREMENT
-  int counterRowid = 0;  /* Memory cell holding rowid of autoinc counter */
-#endif
-
   if( pParse->nErr || sqlite3MallocFailed() ){
     goto insert_cleanup;
   }
@@ -291,31 +419,27 @@ void sqlite3Insert(
     newIdx = pParse->nTab++;
   }
 
-#ifndef SQLITE_OMIT_AUTOINCREMENT
+#ifndef SQLITE_OMIT_XFER_OPT
+  /* If the statement is of the form
+  **
+  **       INSERT INTO <table1> SELECT * FROM <table2>;
+  **
+  ** Then special optimizations can be applied that make the transfer
+  ** very fast and which reduce fragmentation of indices.
+  */
+  if( pColumn==0 && xferOptimization(pParse, pTab, pSelect, onError, iDb) ){
+    assert( !triggers_exist );
+    assert( pList==0 );
+    goto insert_cleanup;
+  }
+#endif /* SQLITE_OMIT_XFER_OPT */
+
   /* If this is an AUTOINCREMENT table, look up the sequence number in the
   ** sqlite_sequence table and store it in memory cell counterMem.  Also
   ** remember the rowid of the sqlite_sequence table entry in memory cell
   ** counterRowid.
   */
-  if( pTab->autoInc ){
-    int iCur = pParse->nTab;
-    int addr = sqlite3VdbeCurrentAddr(v);
-    counterRowid = pParse->nMem++;
-    counterMem = pParse->nMem++;
-    sqlite3OpenTable(pParse, iCur, iDb, pDb->pSchema->pSeqTab, OP_OpenRead);
-    sqlite3VdbeAddOp(v, OP_Rewind, iCur, addr+13);
-    sqlite3VdbeAddOp(v, OP_Column, iCur, 0);
-    sqlite3VdbeOp3(v, OP_String8, 0, 0, pTab->zName, 0);
-    sqlite3VdbeAddOp(v, OP_Ne, 0x100, addr+12);
-    sqlite3VdbeAddOp(v, OP_Rowid, iCur, 0);
-    sqlite3VdbeAddOp(v, OP_MemStore, counterRowid, 1);
-    sqlite3VdbeAddOp(v, OP_Column, iCur, 1);
-    sqlite3VdbeAddOp(v, OP_MemStore, counterMem, 1);
-    sqlite3VdbeAddOp(v, OP_Goto, 0, addr+13);
-    sqlite3VdbeAddOp(v, OP_Next, iCur, addr+4);
-    sqlite3VdbeAddOp(v, OP_Close, iCur, 0);
-  }
-#endif /* SQLITE_OMIT_AUTOINCREMENT */
+  counterMem = autoIncBegin(pParse, iDb, pTab);
 
   /* Figure out how many columns of data are supplied.  If the data
   ** is coming from a SELECT statement, then this step also generates
@@ -591,11 +715,7 @@ void sqlite3Insert(
     }else{
       sqlite3VdbeAddOp(v, OP_NewRowid, base, counterMem);
     }
-#ifndef SQLITE_OMIT_AUTOINCREMENT
-    if( pTab->autoInc ){
-      sqlite3VdbeAddOp(v, OP_MemMax, counterMem, 0);
-    }
-#endif /* SQLITE_OMIT_AUTOINCREMENT */
+    autoIncStep(pParse, counterMem);
 
     /* Push onto the stack, data for all columns of the new entry, beginning
     ** with the first column.
@@ -688,26 +808,11 @@ void sqlite3Insert(
     }
   }
 
-#ifndef SQLITE_OMIT_AUTOINCREMENT
   /* Update the sqlite_sequence table by storing the content of the
   ** counter value in memory counterMem back into the sqlite_sequence
   ** table.
   */
-  if( pTab->autoInc ){
-    int iCur = pParse->nTab;
-    int addr = sqlite3VdbeCurrentAddr(v);
-    sqlite3OpenTable(pParse, iCur, iDb, pDb->pSchema->pSeqTab, OP_OpenWrite);
-    sqlite3VdbeAddOp(v, OP_MemLoad, counterRowid, 0);
-    sqlite3VdbeAddOp(v, OP_NotNull, -1, addr+7);
-    sqlite3VdbeAddOp(v, OP_Pop, 1, 0);
-    sqlite3VdbeAddOp(v, OP_NewRowid, iCur, 0);
-    sqlite3VdbeOp3(v, OP_String8, 0, 0, pTab->zName, 0);
-    sqlite3VdbeAddOp(v, OP_MemLoad, counterMem, 0);
-    sqlite3VdbeAddOp(v, OP_MakeRecord, 2, 0);
-    sqlite3VdbeAddOp(v, OP_Insert, iCur, 0);
-    sqlite3VdbeAddOp(v, OP_Close, iCur, 0);
-  }
-#endif
+  autoIncEnd(pParse, iDb, pTab, counterMem);
 
   /*
   ** Return the number of rows inserted. If this routine is 
@@ -1140,3 +1245,299 @@ void sqlite3OpenTableAndIndices(
     pParse->nTab = base+i;
   }
 }
+
+#ifndef SQLITE_OMIT_XFER_OPT
+/*
+** Check to collation names to see if they are compatible.
+*/
+static int xferCompatibleCollation(const char *z1, const char *z2){
+  if( z1==0 ){
+    return z2==0;
+  }
+  if( z2==0 ){
+    return 0;
+  }
+  return sqlite3StrICmp(z1, z2)==0;
+}
+
+
+/*
+** Check to see if index pSrc is compatible as a source of data
+** for index pDest in an insert transfer optimization.  The rules
+** for a compatible index:
+**
+**    *   The index is over the same set of columns
+**    *   The same DESC and ASC markings occurs on all columns
+**    *   The same onError processing (OE_Abort, OE_Ignore, etc)
+**    *   The same collating sequence on each column
+*/
+static int xferCompatibleIndex(Index *pDest, Index *pSrc){
+  int i;
+  assert( pDest && pSrc );
+  assert( pDest->pTable!=pSrc->pTable );
+  if( pDest->nColumn!=pSrc->nColumn ){
+    return 0;   /* Different number of columns */
+  }
+  if( pDest->onError!=pSrc->onError ){
+    return 0;   /* Different conflict resolution strategies */
+  }
+  for(i=0; i<pSrc->nColumn; i++){
+    if( pSrc->aiColumn[i]!=pDest->aiColumn[i] ){
+      return 0;   /* Different columns indexed */
+    }
+    if( pSrc->aSortOrder[i]!=pDest->aSortOrder[i] ){
+      return 0;   /* Different sort orders */
+    }
+    if( pSrc->azColl[i]!=pDest->azColl[i] ){
+      return 0;   /* Different sort orders */
+    }
+  }
+
+  /* If no test above fails then the indices must be compatible */
+  return 1;
+}
+
+/*
+** Attempt the transfer optimization on INSERTs of the form
+**
+**     INSERT INTO tab1 SELECT * FROM tab2;
+**
+** This optimization is only attempted if
+**
+**    (1)  tab1 and tab2 have identical schemas including all the
+**         same indices
+**
+**    (2)  tab1 and tab2 are different tables
+**
+**    (3)  There must be no triggers on tab1
+**
+**    (4)  The result set of the SELECT statement is "*"
+**
+**    (5)  The SELECT statement has no WHERE, HAVING, ORDER BY, GROUP BY,
+**         or LIMIT clause.
+**
+**    (6)  The SELECT statement is a simple (not a compound) select that
+**         contains only tab2 in its FROM clause
+**
+** This method for implementing the INSERT transfers raw records from
+** tab2 over to tab1.  The columns are not decoded.  Raw records from
+** the indices of tab2 are transfered to tab1 as well.  In so doing,
+** the resulting tab1 has much less fragmentation.
+**
+** This routine returns TRUE if the optimization is attempted.  If any
+** of the conditions above fail so that the optimization should not
+** be attempted, then this routine returns FALSE.
+*/
+static int xferOptimization(
+  Parse *pParse,        /* Parser context */
+  Table *pDest,         /* The table we are inserting into */
+  Select *pSelect,      /* A SELECT statement to use as the data source */
+  int onError,          /* How to handle constraint errors */
+  int iDbDest           /* The database of pDest */
+){
+  ExprList *pEList;                /* The result set of the SELECT */
+  Table *pSrc;                     /* The table in the FROM clause of SELECT */
+  Index *pSrcIdx, *pDestIdx;       /* Source and destination indices */
+  struct SrcList_item *pItem;      /* An element of pSelect->pSrc */
+  int i;                           /* Loop counter */
+  int iDbSrc;                      /* The database of pSrc */
+  int iSrc, iDest;                 /* Cursors from source and destination */
+  int addr1, addr2;                /* Loop addresses */
+  int emptyDestTest;               /* Address of test for empty pDest */
+  int emptySrcTest;                /* Address of test for empty pSrc */
+  int memRowid;                    /* A memcell containing a rowid from pSrc */
+  Vdbe *v;                         /* The VDBE we are building */
+  KeyInfo *pKey;                   /* Key information for an index */
+  int counterMem;                  /* Memory register used by AUTOINC */
+
+  if( pSelect==0 ){
+    return 0;   /* Must be of the form  INSERT INTO ... SELECT ... */
+  }
+  if( pDest->pTrigger ){
+    return 0;   /* tab1 must not have triggers */
+  }
+#ifndef SQLITE_OMIT_VIRTUALTABLE
+  if( pDest->isVirtual ){
+    return 0;   /* tab1 must not be a virtual table */
+  }
+#endif
+  if( onError==OE_Default ){
+    onError = OE_Abort;
+  }
+  if( onError!=OE_Abort && onError!=OE_Rollback ){
+    return 0;   /* Cannot do OR REPLACE or OR IGNORE or OR FAIL */
+  }
+  if( pSelect->pSrc==0 ){
+    return 0;   /* SELECT must have a FROM clause */
+  }
+  if( pSelect->pSrc->nSrc!=1 ){
+    return 0;   /* FROM clause must have exactly one term */
+  }
+  if( pSelect->pSrc->a[0].pSelect ){
+    return 0;   /* FROM clause cannot contain a subquery */
+  }
+  if( pSelect->pWhere ){
+    return 0;   /* SELECT may not have a WHERE clause */
+  }
+  if( pSelect->pOrderBy ){
+    return 0;   /* SELECT may not have an ORDER BY clause */
+  }
+  if( pSelect->pHaving ){
+    return 0;   /* SELECT may not have a HAVING clause */
+  }
+  if( pSelect->pGroupBy ){
+    return 0;   /* SELECT may not have a GROUP BY clause */
+  }
+  if( pSelect->pLimit ){
+    return 0;   /* SELECT may not have a LIMIT clause */
+  }
+  if( pSelect->pOffset ){
+    return 0;   /* SELECT may not have an OFFSET clause */
+  }
+  if( pSelect->pPrior ){
+    return 0;   /* SELECT may not be a compound query */
+  }
+  if( pSelect->isDistinct ){
+    return 0;   /* SELECT may not be DISTINCT */
+  }
+  pEList = pSelect->pEList;
+  assert( pEList!=0 );
+  if( pEList->nExpr!=1 ){
+    return 0;   /* The result set must have exactly one column */
+  }
+  assert( pEList->a[0].pExpr );
+  if( pEList->a[0].pExpr->op!=TK_ALL ){
+    return 0;   /* The result set must be the special operator "*" */
+  }
+
+  /* At this point we have established that the statement is of the
+  ** correct syntactic form to participate in this optimization.  Now
+  ** we have to check the semantics.
+  */
+  pItem = pSelect->pSrc->a;
+  pSrc = sqlite3LocateTable(pParse, pItem->zName, pItem->zDatabase);
+  if( pSrc==0 ){
+    return 0;   /* FROM clause does not contain a real table */
+  }
+  if( pSrc==pDest ){
+    return 0;   /* tab1 and tab2 may not be the same table */
+  }
+#ifndef SQLITE_OMIT_VIRTUALTABLE
+  if( pSrc->isVirtual ){
+    return 0;   /* tab2 must not be a virtual table */
+  }
+#endif
+  if( pSrc->pSelect ){
+    return 0;   /* tab2 may not be a view */
+  }
+  if( pDest->nCol!=pSrc->nCol ){
+    return 0;   /* Number of columns must be the same in tab1 and tab2 */
+  }
+  if( pDest->iPKey!=pSrc->iPKey ){
+    return 0;   /* Both tables must have the same INTEGER PRIMARY KEY */
+  }
+  for(i=0; i<pDest->nCol; i++){
+    if( pDest->aCol[i].affinity!=pSrc->aCol[i].affinity ){
+      return 0;    /* Affinity must be the same on all columns */
+    }
+    if( !xferCompatibleCollation(pDest->aCol[i].zColl, pSrc->aCol[i].zColl) ){
+      return 0;    /* Collating sequence must be the same on all columns */
+    }
+    if( pDest->aCol[i].notNull && !pSrc->aCol[i].notNull ){
+      return 0;    /* tab2 must be NOT NULL if tab1 is */
+    }
+  }
+  for(pDestIdx=pDest->pIndex; pDestIdx; pDestIdx=pDestIdx->pNext){
+    for(pSrcIdx=pSrc->pIndex; pSrcIdx; pSrcIdx=pSrcIdx->pNext){
+      if( xferCompatibleIndex(pDestIdx, pSrcIdx) ) break;
+    }
+    if( pSrcIdx==0 ){
+      return 0;    /* pDestIdx has no corresponding index in pSrc */
+    }
+  }
+
+  /* If we get this far, it means either:
+  **
+  **    *   We can always do the transfer if the table contains an
+  **        an integer primary key
+  **
+  **    *   We can conditionally do the transfer if the destination
+  **        table is empty.
+  */
+  iDbSrc = sqlite3SchemaToIndex(pParse->db, pSrc->pSchema);
+  v = sqlite3GetVdbe(pParse);
+  iSrc = pParse->nTab++;
+  iDest = pParse->nTab++;
+  counterMem = autoIncBegin(pParse, iDbDest, pDest);
+  sqlite3OpenTable(pParse, iDest, iDbDest, pDest, OP_OpenWrite);
+  if( pDest->iPKey<0 ){
+    /* The tables do not have an INTEGER PRIMARY KEY so that
+    ** transfer optimization is only allowed if the destination
+    ** table is initially empty
+    */
+    addr1 = sqlite3VdbeAddOp(v, OP_Rewind, iDest, 0);
+    emptyDestTest = sqlite3VdbeAddOp(v, OP_Goto, 0, 0);
+    sqlite3VdbeJumpHere(v, addr1);
+  }else{
+    emptyDestTest = 0;
+  }
+  sqlite3OpenTable(pParse, iSrc, iDbSrc, pSrc, OP_OpenRead);
+  emptySrcTest = sqlite3VdbeAddOp(v, OP_Rewind, iSrc, 0);
+  memRowid = pParse->nMem++;
+  sqlite3VdbeAddOp(v, OP_Rowid, iSrc, 0);
+  sqlite3VdbeAddOp(v, OP_MemStore, memRowid, 1);
+  addr1 = sqlite3VdbeAddOp(v, OP_Rowid, iSrc, 0);
+  sqlite3VdbeAddOp(v, OP_Dup, 0, 0);
+  addr2 = sqlite3VdbeAddOp(v, OP_NotExists, iDest, 0);
+  sqlite3VdbeOp3(v, OP_Halt, SQLITE_CONSTRAINT, onError, 
+                    "PRIMARY KEY must be unique", P3_STATIC);
+  sqlite3VdbeJumpHere(v, addr2);
+  autoIncStep(pParse, counterMem);
+  sqlite3VdbeAddOp(v, OP_RowData, iSrc, 0);
+  sqlite3VdbeOp3(v, OP_Insert, iDest, OPFLAG_NCHANGE|OPFLAG_LASTROWID,
+                    pDest->zName, 0);
+  sqlite3VdbeAddOp(v, OP_Next, iSrc, addr1);
+  autoIncEnd(pParse, iDbDest, pDest, counterMem);
+  for(pDestIdx=pDest->pIndex; pDestIdx; pDestIdx=pDestIdx->pNext){
+    for(pSrcIdx=pSrc->pIndex; pSrcIdx; pSrcIdx=pSrcIdx->pNext){
+      if( xferCompatibleIndex(pDestIdx, pSrcIdx) ) break;
+    }
+    assert( pSrcIdx );
+    sqlite3VdbeAddOp(v, OP_Close, iSrc, 0);
+    sqlite3VdbeAddOp(v, OP_Close, iDest, 0);
+    sqlite3VdbeAddOp(v, OP_Integer, iDbSrc, 0);
+    pKey = sqlite3IndexKeyinfo(pParse, pSrcIdx);
+    VdbeComment((v, "# %s", pSrcIdx->zName));
+    sqlite3VdbeOp3(v, OP_OpenRead, iSrc, pSrcIdx->tnum, 
+                   (char*)pKey, P3_KEYINFO_HANDOFF);
+    sqlite3VdbeAddOp(v, OP_Integer, iDbDest, 0);
+    pKey = sqlite3IndexKeyinfo(pParse, pDestIdx);
+    VdbeComment((v, "# %s", pDestIdx->zName));
+    sqlite3VdbeOp3(v, OP_OpenWrite, iDest, pDestIdx->tnum, 
+                   (char*)pKey, P3_KEYINFO_HANDOFF);
+    addr1 = sqlite3VdbeAddOp(v, OP_Rewind, iSrc, 0);
+    sqlite3VdbeAddOp(v, OP_RowKey, iSrc, 0);
+    if( pDestIdx->onError!=OE_None ){
+      sqlite3VdbeAddOp(v, OP_MemLoad, memRowid, 0);
+      addr2 = sqlite3VdbeAddOp(v, OP_IsUnique, iDest, 0);
+      sqlite3VdbeOp3(v, OP_Halt, SQLITE_CONSTRAINT, onError, 
+                    "UNIQUE constraint failed", P3_STATIC);
+      sqlite3VdbeJumpHere(v, addr2);
+    }
+    sqlite3VdbeAddOp(v, OP_IdxInsert, iDest, 0);
+    sqlite3VdbeAddOp(v, OP_Next, iSrc, addr1+1);
+    sqlite3VdbeJumpHere(v, addr1);
+  }
+  sqlite3VdbeJumpHere(v, emptySrcTest);
+  sqlite3VdbeAddOp(v, OP_Close, iSrc, 0);
+  sqlite3VdbeAddOp(v, OP_Close, iDest, 0);
+  if( emptyDestTest ){
+    sqlite3VdbeAddOp(v, OP_Halt, SQLITE_OK, 0);
+    sqlite3VdbeJumpHere(v, emptyDestTest);
+    sqlite3VdbeAddOp(v, OP_Close, iDest, 0);
+    return 0;
+  }else{
+    return 1;
+  }
+}
+#endif /* SQLITE_OMIT_XFER_OPT */