# Check that an OTA cannot be applied to a table that has no PK.
#
# UPDATE: At one point OTA required that all tables featured either
# explicit IPK columns or were declared WITHOUT ROWID. This has been
# relaxed so that external PRIMARY KEYs on tables with automatic rowids
# are now allowed.
#



reset_db
create_ota1 ota.db
do_execsql_test 2.1 { CREATE TABLE t1(a, b, c) }
do_test 2.2 {
  sqlite3ota ota test.db ota.db
  ota step
} {SQLITE_ERROR}
do_test 2.3 {
  list [catch { ota close } msg] $msg
} {1 {SQLITE_ERROR - table t1 has no PRIMARY KEY}}
reset_db
do_execsql_test 2.4 { CREATE TABLE t1(a PRIMARY KEY, b, c) }
do_test 2.5 {
  sqlite3ota ota test.db ota.db
  ota step
} {SQLITE_OK}
do_test 2.6 {

# Check that an OTA cannot be applied to a table that has no PK.
#
# UPDATE: At one point OTA required that all tables featured either
# explicit IPK columns or were declared WITHOUT ROWID. This has been
# relaxed so that external PRIMARY KEYs on tables with automatic rowids
# are now allowed.
#
# UPDATE 2: Tables without any PRIMARY KEY declaration are now allowed.
# However the input table must feature an "ota_rowid" column.
#
reset_db
create_ota1 ota.db
do_execsql_test 2.1 { CREATE TABLE t1(a, b, c) }
do_test 2.2 {
  sqlite3ota ota test.db ota.db
  ota step
} {SQLITE_ERROR}
do_test 2.3 {
  list [catch { ota close } msg] $msg
} {1 {SQLITE_ERROR - table data_t1 requires ota_rowid column}}
reset_db
do_execsql_test 2.4 { CREATE TABLE t1(a PRIMARY KEY, b, c) }
do_test 2.5 {
  sqlite3ota ota test.db ota.db
  ota step
} {SQLITE_OK}
do_test 2.6 {

#
#    May you do good and not evil.
#    May you find forgiveness for yourself and forgive others.
#    May you share freely, never taking more than you give.
#
#***********************************************************************
#
# Test OTA with virtual tables
#

if {![info exists testdir]} {
  set testdir [file join [file dirname [info script]] .. .. test]
}
source $testdir/tester.tcl
set ::testprefix ota9
................................................................................
  12 d e f
  13 g h X
  14 x y z
}
do_execsql_test 1.2.3 { INSERT INTO f1(f1) VALUES('integrity-check') }
integrity_check 1.2.4

































































finish_test

#
#    May you do good and not evil.
#    May you find forgiveness for yourself and forgive others.
#    May you share freely, never taking more than you give.
#
#***********************************************************************
#
# Test OTA with virtual tables. And tables with no PRIMARY KEY declarations.
#

if {![info exists testdir]} {
  set testdir [file join [file dirname [info script]] .. .. test]
}
source $testdir/tester.tcl
set ::testprefix ota9
................................................................................
  12 d e f
  13 g h X
  14 x y z
}
do_execsql_test 1.2.3 { INSERT INTO f1(f1) VALUES('integrity-check') }
integrity_check 1.2.4

#-------------------------------------------------------------------------
# Tables with no PK declaration.
#

# Run the OTA in file $ota on target database $target until completion.
#
proc run_ota {target ota} {
  sqlite3ota ota $target $ota
  while { [ota step]=="SQLITE_OK" } {}
  ota close
}

foreach {tn idx} {
  1 { }
  2 { 
    CREATE INDEX i1 ON t1(a);
  }
  3 { 
    CREATE INDEX i1 ON t1(b, c);
    CREATE INDEX i2 ON t1(c, b);
    CREATE INDEX i3 ON t1(a, a, a, b, b, b, c, c, c);
  }
} {

  reset_db
  do_execsql_test 2.$tn.1 {
    CREATE TABLE t1(a, b, c);
    INSERT INTO t1 VALUES(1, 2, 3);
    INSERT INTO t1 VALUES(4, 5, 6);
    INSERT INTO t1(rowid, a, b, c) VALUES(-1, 'a', 'b', 'c');
    INSERT INTO t1(rowid, a, b, c) VALUES(-2, 'd', 'e', 'f');
  }

  db eval $idx
  
  do_test 2.$tn.2 {
    forcedelete ota.db
    sqlite3 db2 ota.db
    db2 eval {
      CREATE TABLE data_t1(ota_rowid, a, b, c, ota_control);
      INSERT INTO data_t1 VALUES(3, 'x', 'y', 'z', 0);
      INSERT INTO data_t1 VALUES(NULL, 'X', 'Y', 'Z', 0);
      INSERT INTO data_t1 VALUES('1', NULL, NULL, NULL, 1);
      INSERT INTO data_t1 VALUES(-2, NULL, NULL, 'fff', '..x');
    }
    db2 close
  } {}
  
  run_ota test.db ota.db
  
  do_execsql_test 2.$tn.3 {
    SELECT rowid, a, b, c FROM t1 ORDER BY rowid;
  } {
    -2 d e fff
    -1 a b c
     2 4 5 6
     3 x y z
     4 X Y Z
  }
  
  integrity_check 2.$tn.4
}


finish_test

*/
struct OtaObjIter {
  sqlite3_stmt *pTblIter;         /* Iterate through tables */
  sqlite3_stmt *pIdxIter;         /* Index iterator */
  int nTblCol;                    /* Size of azTblCol[] array */
  char **azTblCol;                /* Array of quoted column names */
  unsigned char *abTblPk;         /* Array of flags - true for PK columns */


  unsigned char bRowid;           /* True for implicit IPK tables */
  unsigned char bVtab;            /* True for a virtual table */


  /* Output variables. zTbl==0 implies EOF. */
  int bCleanup;                   /* True in "cleanup" state */
  const char *zTbl;               /* Name of target db table */
  const char *zIdx;               /* Name of target db index (or null) */
  int iVisit;                     /* Number of points visited, incl. current */

................................................................................
  sqlite3_stmt *pDelete;          /* Statement for DELETE ops */

  /* Last UPDATE used (for PK b-tree updates only), or NULL. */
  char *zMask;                    /* Copy of update mask used with pUpdate */
  sqlite3_stmt *pUpdate;          /* Last update statement (or NULL) */
};









/*
** OTA handle.
*/
struct sqlite3ota {
  int eStage;                     /* Value of OTA_STATE_STAGE field */
  sqlite3 *db;                    /* "main" -> target db, "ota" -> ota db */
  char *zTarget;                  /* Path to target db */
................................................................................
  }
  sqlite3_free(pIter->azTblCol);
  pIter->azTblCol = 0;
  pIter->abTblPk = 0;
  pIter->nTblCol = 0;
  sqlite3_free(pIter->zMask);
  pIter->zMask = 0;
  pIter->bRowid = 0;
  pIter->bVtab = 0;
}

/*
** Finalize all statements and free all allocations that are specific to
** the current object (table/index pair).
*/
static void otaObjIterClearStatements(OtaObjIter *pIter){
................................................................................
** an error does occur, an error code and error message are also left in 
** the OTA handle.
*/
static int otaObjIterGetCols(sqlite3ota *p, OtaObjIter *pIter){
  if( pIter->azTblCol==0 ){
    sqlite3_stmt *pStmt = 0;
    int nCol = 0;
    int bSeenPk = 0;
    int i;                        /* for() loop iterator variable */
    int rc2;                      /* sqlite3_finalize() return value */


    assert( pIter->bRowid==0 && pIter->bVtab==0 );

    /* Populate the azTblCol[] and nTblCol variables based on the columns
    ** of the input table. Ignore any input table columns that begin with
    ** "ota_".  */
    p->rc = prepareFreeAndCollectError(p->db, &pStmt, &p->zErrmsg, 
        sqlite3_mprintf("SELECT * FROM 'data_%q'", pIter->zTbl)
    );
................................................................................
    }
    for(i=0; p->rc==SQLITE_OK && i<nCol; i++){
      const char *zName = (const char*)sqlite3_column_name(pStmt, i);
      if( sqlite3_strnicmp("ota_", zName, 4) ){
        char *zCopy = otaQuoteName(zName);
        pIter->azTblCol[pIter->nTblCol++] = zCopy;
        if( zCopy==0 ) p->rc = SQLITE_NOMEM;



      }
    }
    sqlite3_finalize(pStmt);
    pStmt = 0;

    /* Check that all non-HIDDEN columns in the destination table are also
    ** present in the input table. Populate the abTblPk[] array at the
................................................................................
        p->rc = SQLITE_ERROR;
        p->zErrmsg = sqlite3_mprintf("column missing from data_%q: %s",
            pIter->zTbl, zName
        );
      }else{
        int iPk = sqlite3_column_int(pStmt, 5);
        pIter->abTblPk[i] = (iPk!=0);
        if( iPk ) bSeenPk = 1;
        if( iPk<0 ) pIter->bRowid = 1;

      }
    }
    rc2 = sqlite3_finalize(pStmt);
    if( p->rc==SQLITE_OK ) p->rc = rc2;

    if( p->rc==SQLITE_OK && bSeenPk==0 ){
      const char *zTab = pIter->zTbl;







      if( otaIsVtab(p, zTab) ){
        pIter->bVtab = 1;

      }else{
        p->zErrmsg = sqlite3_mprintf("table %s has no PRIMARY KEY", zTab);



        p->rc = SQLITE_ERROR;



      }
    }
  }

  return p->rc;
}

................................................................................
        break;
      }
    }
  }
  return zList;
}












static char *otaObjIterGetOldlist(
  sqlite3ota *p, 
  OtaObjIter *pIter,
  const char *zObj
){
  char *zList = 0;
  if( p->rc==SQLITE_OK ){
................................................................................
      if( zList==0 ){
        p->rc = SQLITE_NOMEM;
        break;
      }
    }

    /* For a table with implicit rowids, append "old._rowid_" to the list. */
    if( pIter->bRowid ){
      zList = sqlite3_mprintf("%z, %s._rowid_", zList, zObj);
    }
  }
  return zList;
}











static char *otaObjIterGetWhere(
  sqlite3ota *p, 
  OtaObjIter *pIter
){
  char *zList = 0;
  if( p->rc==SQLITE_OK ){
    if( pIter->bVtab ){
      zList = otaMPrintfAndCollectError(p, "rowid = ?%d", pIter->nTblCol+1);
    }else{
      const char *zSep = "";
      int i;
      for(i=0; i<pIter->nTblCol; i++){
        if( pIter->abTblPk[i] ){
          const char *zCol = pIter->azTblCol[i];
          zList = otaMPrintfAndCollectError(
................................................................................
      if( !zLimit ) p->rc = SQLITE_NOMEM;
    }

    if( zIdx ){
      int *aiCol;                 /* Column map */
      const char **azColl;        /* Collation sequences */

      assert( pIter->bVtab==0 );

      /* Create the index writers */
      if( p->rc==SQLITE_OK ){
        p->rc = sqlite3_index_writer(
            p->db, 0, zIdx, &pIter->pInsert, &azColl, &aiCol, &pIter->nCol
        );
      }
................................................................................
        );
      }

      /* Create the SELECT statement to read keys in sorted order */
      zCollist = otaObjIterGetCollist(p, pIter, pIter->nCol, aiCol, azColl);
      if( p->rc==SQLITE_OK ){
        char *zSql;
        if( pIter->bRowid ){
          zSql = sqlite3_mprintf(
              "SELECT %s, ota_control FROM ota.'ota_tmp_%q' ORDER BY %s%s",
              zCollist, pIter->zTbl, 
              zCollist, zLimit
          );
        }else{
          zSql = sqlite3_mprintf(
................................................................................
              zCollist, pIter->zTbl, 
              zCollist, zLimit
          );
        }
        p->rc = prepareFreeAndCollectError(p->db, &pIter->pSelect, pz, zSql);
      }
    }else{

      const char *zTbl = pIter->zTbl;
      char *zBindings = otaObjIterGetBindlist(p, pIter->nTblCol+pIter->bVtab);
      char *zWhere = otaObjIterGetWhere(p, pIter);
      char *zOldlist = otaObjIterGetOldlist(p, pIter, "old");
      char *zNewlist = otaObjIterGetOldlist(p, pIter, "new");


      zCollist = otaObjIterGetCollist(p, pIter, pIter->nTblCol, 0, 0);
      pIter->nCol = pIter->nTblCol;

      /* Create the SELECT statement to read keys from data_xxx */
      if( p->rc==SQLITE_OK ){
        p->rc = prepareFreeAndCollectError(p->db, &pIter->pSelect, pz,
            sqlite3_mprintf(
              "SELECT %s, ota_control%s FROM ota.'data_%q'%s", 
              zCollist, (pIter->bVtab ? ", ota_rowid" : ""), zTbl, zLimit
            )
        );
      }

      /* Create the INSERT statement to write to the target PK b-tree */
      if( p->rc==SQLITE_OK ){
        p->rc = prepareFreeAndCollectError(p->db, &pIter->pInsert, pz,
            sqlite3_mprintf(
              "INSERT INTO main.%Q(%s%s) VALUES(%s)", 
              zTbl, zCollist, (pIter->bVtab ? ", rowid" : ""), zBindings
            )
        );
      }

      /* Create the DELETE statement to write to the target PK b-tree */
      if( p->rc==SQLITE_OK ){
        p->rc = prepareFreeAndCollectError(p->db, &pIter->pDelete, pz,
            sqlite3_mprintf(
              "DELETE FROM main.%Q WHERE %s", zTbl, zWhere
            )
        );
      }

      if( pIter->bVtab==0 ){
        const char *zOtaRowid = (pIter->bRowid ? ", ota_rowid" : "");




        /* Create the ota_tmp_xxx table and the triggers to populate it. */
        otaMPrintfExec(p, 
            "PRAGMA ota_mode = 1;"
            "CREATE TABLE IF NOT EXISTS ota.'ota_tmp_%q' AS "
            "SELECT *%s FROM ota.'data_%q' WHERE 0;"

................................................................................
            "  INSERT INTO 'ota_tmp_%q'(ota_control, %s%s) VALUES(2, %s);"
            "END;"

            "CREATE TEMP TRIGGER ota_update2_%q AFTER UPDATE ON main.%Q "
            "BEGIN "
            "  INSERT INTO 'ota_tmp_%q'(ota_control, %s%s) VALUES(3, %s);"
            "END;"
            , zTbl, (pIter->bRowid ? ", 0 AS ota_rowid" : ""), zTbl, 

            zTbl, zTbl, zTbl, zCollist, zOtaRowid, zOldlist,
            zTbl, zTbl, zTbl, zCollist, zOtaRowid, zOldlist,
            zTbl, zTbl, zTbl, zCollist, zOtaRowid, zNewlist
        );
        if( pIter->bRowid ){
          otaMPrintfExec(p, 
              "CREATE TEMP TRIGGER ota_insert_%q AFTER INSERT ON main.%Q "
              "BEGIN "
              "  INSERT INTO 'ota_tmp_%q'(ota_control, %s, ota_rowid)"
              "  VALUES(0, %s);"
              "END;"
              , zTbl, zTbl, zTbl, zCollist, zNewlist
................................................................................

  if( res==0 ){
    otaBadControlError(p);
  }
  return res;
}










/*
** This function does the work for an sqlite3ota_step() call.
**
** The object-iterator (p->objiter) currently points to a valid object,
** and the input cursor (p->objiter.pSelect) currently points to a valid
** input row. Perform whatever processing is required and return.
**
................................................................................
      for(i=0; i<pIter->nCol; i++){
        if( eType==SQLITE_DELETE && pIter->zIdx==0 && pIter->abTblPk[i]==0 ){
          continue;
        }
        pVal = sqlite3_column_value(pIter->pSelect, i);
        sqlite3_bind_value(pWriter, i+1, pVal);
      }
      if( pIter->bVtab ){



        /* For a virtual table, the SELECT statement is:
        **
        **   SELECT <cols>, ota_control, ota_rowid FROM ....
        **
        ** Hence column_value(pIter->nCol+1).
        */

        pVal = sqlite3_column_value(pIter->pSelect, pIter->nCol+1);
        sqlite3_bind_value(pWriter, pIter->nCol+1, pVal);
      }
      sqlite3_step(pWriter);
      p->rc = resetAndCollectError(pWriter, &p->zErrmsg);
    }else if( eType==OTA_UPDATE ){
      sqlite3_value *pVal;
................................................................................
      sqlite3_stmt *pUpdate = 0;
      otaGetUpdateStmt(p, pIter, zMask, &pUpdate);
      if( pUpdate ){
        for(i=0; i<pIter->nCol; i++){
          pVal = sqlite3_column_value(pIter->pSelect, i);
          sqlite3_bind_value(pUpdate, i+1, pVal);
        }
        if( pIter->bVtab ){


          pVal = sqlite3_column_value(pIter->pSelect, pIter->nCol+1);
          sqlite3_bind_value(pUpdate, pIter->nCol+1, pVal);
        }
        sqlite3_step(pUpdate);
        p->rc = resetAndCollectError(pUpdate, &p->zErrmsg);
      }
    }else{
................................................................................
        OtaObjIter *pIter = &p->objiter;
        while( p && p->rc==SQLITE_OK && pIter->zTbl ){

          if( pIter->bCleanup ){
            /* Clean up the ota_tmp_xxx table for the previous table. It 
            ** cannot be dropped as there are currently active SQL statements.
            ** But the contents can be deleted.  */
            if( pIter->bVtab==0 ){
              otaMPrintfExec(p, "DELETE FROM ota.'ota_tmp_%q'", pIter->zTbl);
            }
          }else{
            otaObjIterPrepareAll(p, pIter, 0);

            /* Advance to the next row to process. */
            if( p->rc==SQLITE_OK ){

*/
struct OtaObjIter {
  sqlite3_stmt *pTblIter;         /* Iterate through tables */
  sqlite3_stmt *pIdxIter;         /* Index iterator */
  int nTblCol;                    /* Size of azTblCol[] array */
  char **azTblCol;                /* Array of quoted column names */
  unsigned char *abTblPk;         /* Array of flags - true for PK columns */
  int eType;
#if 0
  unsigned char bRowid;           /* True for implicit IPK tables */
  unsigned char bVtab;            /* True for a virtual table */
#endif

  /* Output variables. zTbl==0 implies EOF. */
  int bCleanup;                   /* True in "cleanup" state */
  const char *zTbl;               /* Name of target db table */
  const char *zIdx;               /* Name of target db index (or null) */
  int iVisit;                     /* Number of points visited, incl. current */

................................................................................
  sqlite3_stmt *pDelete;          /* Statement for DELETE ops */

  /* Last UPDATE used (for PK b-tree updates only), or NULL. */
  char *zMask;                    /* Copy of update mask used with pUpdate */
  sqlite3_stmt *pUpdate;          /* Last update statement (or NULL) */
};

/*
** Values for OtaObjIter.eType
*/
#define OTA_PK_REAL     1         /* Table has a real primary key */
#define OTA_PK_EXTERNAL 2         /* Table has an external primary key index */
#define OTA_PK_NONE     3         /* Table has no PK (use rowid) */
#define OTA_PK_VTAB     4         /* Table is a virtual table (use rowid) */

/*
** OTA handle.
*/
struct sqlite3ota {
  int eStage;                     /* Value of OTA_STATE_STAGE field */
  sqlite3 *db;                    /* "main" -> target db, "ota" -> ota db */
  char *zTarget;                  /* Path to target db */
................................................................................
  }
  sqlite3_free(pIter->azTblCol);
  pIter->azTblCol = 0;
  pIter->abTblPk = 0;
  pIter->nTblCol = 0;
  sqlite3_free(pIter->zMask);
  pIter->zMask = 0;
  pIter->eType = 0;               /* Invalid value */

}

/*
** Finalize all statements and free all allocations that are specific to
** the current object (table/index pair).
*/
static void otaObjIterClearStatements(OtaObjIter *pIter){
................................................................................
** an error does occur, an error code and error message are also left in 
** the OTA handle.
*/
static int otaObjIterGetCols(sqlite3ota *p, OtaObjIter *pIter){
  if( pIter->azTblCol==0 ){
    sqlite3_stmt *pStmt = 0;
    int nCol = 0;

    int i;                        /* for() loop iterator variable */
    int rc2;                      /* sqlite3_finalize() return value */
    int bOtaRowid = 0;            /* If input table has column "ota_rowid" */

    assert( pIter->eType==0 );

    /* Populate the azTblCol[] and nTblCol variables based on the columns
    ** of the input table. Ignore any input table columns that begin with
    ** "ota_".  */
    p->rc = prepareFreeAndCollectError(p->db, &pStmt, &p->zErrmsg, 
        sqlite3_mprintf("SELECT * FROM 'data_%q'", pIter->zTbl)
    );
................................................................................
    }
    for(i=0; p->rc==SQLITE_OK && i<nCol; i++){
      const char *zName = (const char*)sqlite3_column_name(pStmt, i);
      if( sqlite3_strnicmp("ota_", zName, 4) ){
        char *zCopy = otaQuoteName(zName);
        pIter->azTblCol[pIter->nTblCol++] = zCopy;
        if( zCopy==0 ) p->rc = SQLITE_NOMEM;
      }
      else if( 0==sqlite3_stricmp("ota_rowid", zName) ){
        bOtaRowid = 1;
      }
    }
    sqlite3_finalize(pStmt);
    pStmt = 0;

    /* Check that all non-HIDDEN columns in the destination table are also
    ** present in the input table. Populate the abTblPk[] array at the
................................................................................
        p->rc = SQLITE_ERROR;
        p->zErrmsg = sqlite3_mprintf("column missing from data_%q: %s",
            pIter->zTbl, zName
        );
      }else{
        int iPk = sqlite3_column_int(pStmt, 5);
        pIter->abTblPk[i] = (iPk!=0);
        if( iPk ){
          pIter->eType = (iPk<0) ? OTA_PK_EXTERNAL : OTA_PK_REAL;
        }
      }
    }
    rc2 = sqlite3_finalize(pStmt);
    if( p->rc==SQLITE_OK ) p->rc = rc2;

    if( p->rc==SQLITE_OK ){
      if( pIter->eType==0 ){
        /* This must either be a virtual table, or a regular table with no
        ** PRIMARY KEY declaration whatsoever.  */
        if( bOtaRowid==0 ){
          p->rc = SQLITE_ERROR;
          p->zErrmsg = sqlite3_mprintf(
              "table data_%q requires ota_rowid column", pIter->zTbl
          );
        }else if( otaIsVtab(p, pIter->zTbl) ){

          pIter->eType = OTA_PK_VTAB;
        }else{

          pIter->eType = OTA_PK_NONE;
        }
      }else if( bOtaRowid ){
        p->rc = SQLITE_ERROR;
        p->zErrmsg = sqlite3_mprintf(
            "table data_%q may not have ota_rowid column", pIter->zTbl
        );
      }
    }
  }

  return p->rc;
}

................................................................................
        break;
      }
    }
  }
  return zList;
}

/*
** Assuming the current table columns are "a", "b" and "c", and the zObj
** paramter is passed "old", return a string of the form:
**
**     "old.a, old.b, old.b"
**
** With the column names escaped.
**
** For tables with implicit rowids - OTA_PK_EXTERNAL and OTA_PK_NONE, append
** the text ", old._rowid_" to the returned value.
*/
static char *otaObjIterGetOldlist(
  sqlite3ota *p, 
  OtaObjIter *pIter,
  const char *zObj
){
  char *zList = 0;
  if( p->rc==SQLITE_OK ){
................................................................................
      if( zList==0 ){
        p->rc = SQLITE_NOMEM;
        break;
      }
    }

    /* For a table with implicit rowids, append "old._rowid_" to the list. */
    if( pIter->eType==OTA_PK_EXTERNAL || pIter->eType==OTA_PK_NONE ){
      zList = sqlite3_mprintf("%z, %s._rowid_", zList, zObj);
    }
  }
  return zList;
}

/*
** Return an expression that can be used in a WHERE clause to match the
** primary key of the current table. For example, if the table is:
**
**   CREATE TABLE t1(a, b, c, PRIMARY KEY(b, c));
**
** Return the string:
**
**   "b = ?1 AND c = ?2"
*/
static char *otaObjIterGetWhere(
  sqlite3ota *p, 
  OtaObjIter *pIter
){
  char *zList = 0;
  if( p->rc==SQLITE_OK ){
    if( pIter->eType==OTA_PK_VTAB || pIter->eType==OTA_PK_NONE ){
      zList = otaMPrintfAndCollectError(p, "_rowid_ = ?%d", pIter->nTblCol+1);
    }else{
      const char *zSep = "";
      int i;
      for(i=0; i<pIter->nTblCol; i++){
        if( pIter->abTblPk[i] ){
          const char *zCol = pIter->azTblCol[i];
          zList = otaMPrintfAndCollectError(
................................................................................
      if( !zLimit ) p->rc = SQLITE_NOMEM;
    }

    if( zIdx ){
      int *aiCol;                 /* Column map */
      const char **azColl;        /* Collation sequences */

      assert( pIter->eType!=OTA_PK_VTAB );

      /* Create the index writers */
      if( p->rc==SQLITE_OK ){
        p->rc = sqlite3_index_writer(
            p->db, 0, zIdx, &pIter->pInsert, &azColl, &aiCol, &pIter->nCol
        );
      }
................................................................................
        );
      }

      /* Create the SELECT statement to read keys in sorted order */
      zCollist = otaObjIterGetCollist(p, pIter, pIter->nCol, aiCol, azColl);
      if( p->rc==SQLITE_OK ){
        char *zSql;
        if( pIter->eType==OTA_PK_EXTERNAL || pIter->eType==OTA_PK_NONE ){
          zSql = sqlite3_mprintf(
              "SELECT %s, ota_control FROM ota.'ota_tmp_%q' ORDER BY %s%s",
              zCollist, pIter->zTbl,
              zCollist, zLimit
          );
        }else{
          zSql = sqlite3_mprintf(
................................................................................
              zCollist, pIter->zTbl, 
              zCollist, zLimit
          );
        }
        p->rc = prepareFreeAndCollectError(p->db, &pIter->pSelect, pz, zSql);
      }
    }else{
      int bOtaRowid = (pIter->eType==OTA_PK_VTAB || pIter->eType==OTA_PK_NONE);
      const char *zTbl = pIter->zTbl;

      char *zWhere = otaObjIterGetWhere(p, pIter);
      char *zOldlist = otaObjIterGetOldlist(p, pIter, "old");
      char *zNewlist = otaObjIterGetOldlist(p, pIter, "new");
      char *zBindings = otaObjIterGetBindlist(p, pIter->nTblCol + bOtaRowid);

      zCollist = otaObjIterGetCollist(p, pIter, pIter->nTblCol, 0, 0);
      pIter->nCol = pIter->nTblCol;

      /* Create the SELECT statement to read keys from data_xxx */
      if( p->rc==SQLITE_OK ){
        p->rc = prepareFreeAndCollectError(p->db, &pIter->pSelect, pz,
            sqlite3_mprintf(
              "SELECT %s, ota_control%s FROM ota.'data_%q'%s", 
              zCollist, (bOtaRowid ? ", ota_rowid" : ""), zTbl, zLimit
            )
        );
      }

      /* Create the INSERT statement to write to the target PK b-tree */
      if( p->rc==SQLITE_OK ){
        p->rc = prepareFreeAndCollectError(p->db, &pIter->pInsert, pz,
            sqlite3_mprintf(
              "INSERT INTO main.%Q(%s%s) VALUES(%s)", 
              zTbl, zCollist, (bOtaRowid ? ", _rowid_" : ""), zBindings
            )
        );
      }

      /* Create the DELETE statement to write to the target PK b-tree */
      if( p->rc==SQLITE_OK ){
        p->rc = prepareFreeAndCollectError(p->db, &pIter->pDelete, pz,
            sqlite3_mprintf(
              "DELETE FROM main.%Q WHERE %s", zTbl, zWhere
            )
        );
      }

      if( pIter->eType!=OTA_PK_VTAB ){
        const char *zOtaRowid = "";
        if( pIter->eType==OTA_PK_EXTERNAL || pIter->eType==OTA_PK_NONE ){
          zOtaRowid = ", ota_rowid";
        }

        /* Create the ota_tmp_xxx table and the triggers to populate it. */
        otaMPrintfExec(p, 
            "PRAGMA ota_mode = 1;"
            "CREATE TABLE IF NOT EXISTS ota.'ota_tmp_%q' AS "
            "SELECT *%s FROM ota.'data_%q' WHERE 0;"

................................................................................
            "  INSERT INTO 'ota_tmp_%q'(ota_control, %s%s) VALUES(2, %s);"
            "END;"

            "CREATE TEMP TRIGGER ota_update2_%q AFTER UPDATE ON main.%Q "
            "BEGIN "
            "  INSERT INTO 'ota_tmp_%q'(ota_control, %s%s) VALUES(3, %s);"
            "END;"
            , zTbl, (pIter->eType==OTA_PK_EXTERNAL ? ", 0 AS ota_rowid" : "")
            , zTbl, 
            zTbl, zTbl, zTbl, zCollist, zOtaRowid, zOldlist,
            zTbl, zTbl, zTbl, zCollist, zOtaRowid, zOldlist,
            zTbl, zTbl, zTbl, zCollist, zOtaRowid, zNewlist
        );
        if( pIter->eType==OTA_PK_EXTERNAL || pIter->eType==OTA_PK_NONE ){
          otaMPrintfExec(p, 
              "CREATE TEMP TRIGGER ota_insert_%q AFTER INSERT ON main.%Q "
              "BEGIN "
              "  INSERT INTO 'ota_tmp_%q'(ota_control, %s, ota_rowid)"
              "  VALUES(0, %s);"
              "END;"
              , zTbl, zTbl, zTbl, zCollist, zNewlist
................................................................................

  if( res==0 ){
    otaBadControlError(p);
  }
  return res;
}

#ifdef SQLITE_DEBUG
static void assertColumnName(sqlite3_stmt *pStmt, int iCol, const char *zName){
  const char *zCol = sqlite3_column_name(pStmt, iCol);
  assert( 0==sqlite3_stricmp(zName, zCol) );
}
#else
# define assertColumnName(x,y,z)
#endif

/*
** This function does the work for an sqlite3ota_step() call.
**
** The object-iterator (p->objiter) currently points to a valid object,
** and the input cursor (p->objiter.pSelect) currently points to a valid
** input row. Perform whatever processing is required and return.
**
................................................................................
      for(i=0; i<pIter->nCol; i++){
        if( eType==SQLITE_DELETE && pIter->zIdx==0 && pIter->abTblPk[i]==0 ){
          continue;
        }
        pVal = sqlite3_column_value(pIter->pSelect, i);
        sqlite3_bind_value(pWriter, i+1, pVal);
      }
      if( pIter->zIdx==0
       && (pIter->eType==OTA_PK_VTAB || pIter->eType==OTA_PK_NONE) 
      ){
        /* For a virtual table, or a table with no primary key, the 
        ** SELECT statement is:
        **
        **   SELECT <cols>, ota_control, ota_rowid FROM ....
        **
        ** Hence column_value(pIter->nCol+1).
        */
        assertColumnName(pIter->pSelect, pIter->nCol+1, "ota_rowid");
        pVal = sqlite3_column_value(pIter->pSelect, pIter->nCol+1);
        sqlite3_bind_value(pWriter, pIter->nCol+1, pVal);
      }
      sqlite3_step(pWriter);
      p->rc = resetAndCollectError(pWriter, &p->zErrmsg);
    }else if( eType==OTA_UPDATE ){
      sqlite3_value *pVal;
................................................................................
      sqlite3_stmt *pUpdate = 0;
      otaGetUpdateStmt(p, pIter, zMask, &pUpdate);
      if( pUpdate ){
        for(i=0; i<pIter->nCol; i++){
          pVal = sqlite3_column_value(pIter->pSelect, i);
          sqlite3_bind_value(pUpdate, i+1, pVal);
        }
        if( pIter->eType==OTA_PK_VTAB || pIter->eType==OTA_PK_NONE ){
          /* Bind the ota_rowid value to column _rowid_ */
          assertColumnName(pIter->pSelect, pIter->nCol+1, "ota_rowid");
          pVal = sqlite3_column_value(pIter->pSelect, pIter->nCol+1);
          sqlite3_bind_value(pUpdate, pIter->nCol+1, pVal);
        }
        sqlite3_step(pUpdate);
        p->rc = resetAndCollectError(pUpdate, &p->zErrmsg);
      }
    }else{
................................................................................
        OtaObjIter *pIter = &p->objiter;
        while( p && p->rc==SQLITE_OK && pIter->zTbl ){

          if( pIter->bCleanup ){
            /* Clean up the ota_tmp_xxx table for the previous table. It 
            ** cannot be dropped as there are currently active SQL statements.
            ** But the contents can be deleted.  */
            if( pIter->eType!=OTA_PK_VTAB ){
              otaMPrintfExec(p, "DELETE FROM ota.'ota_tmp_%q'", pIter->zTbl);
            }
          }else{
            otaObjIterPrepareAll(p, pIter, 0);

            /* Advance to the next row to process. */
            if( p->rc==SQLITE_OK ){