/*
** 2012 Jan 11
**
** The author disclaims copyright to this source code.  In place of
** a legal notice, here is a blessing:
**
**    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.
*/
/* TODO(shess): THIS MODULE IS STILL EXPERIMENTAL.  DO NOT USE IT. */
/* Implements a virtual table "recover" which can be used to recover
 * data from a corrupt table.  The table is walked manually, with
 * corrupt items skipped.  Additionally, any errors while reading will
 * be skipped.
 *
 * Given a table with this definition:
 *
 * CREATE TABLE Stuff (
 *   name TEXT PRIMARY KEY,
 *   value TEXT NOT NULL
 * );
 *
 * to recover the data from teh table, you could do something like:
 *
 * -- Attach another database, the original is not trustworthy.
 * ATTACH DATABASE '/tmp/db.db' AS rdb;
 * -- Create a new version of the table.
 * CREATE TABLE rdb.Stuff (
 *   name TEXT PRIMARY KEY,
 *   value TEXT NOT NULL
 * );
 * -- This will read the original table's data.
 * CREATE VIRTUAL TABLE temp.recover_Stuff using recover(
 *   main.Stuff,
 *   name TEXT STRICT NOT NULL,  -- only real TEXT data allowed
 *   value TEXT STRICT NOT NULL
 * );
 * -- Corruption means the UNIQUE constraint may no longer hold for
 * -- Stuff, so either OR REPLACE or OR IGNORE must be used.
 * INSERT OR REPLACE INTO rdb.Stuff (rowid, name, value )
 *   SELECT rowid, name, value FROM temp.recover_Stuff;
 * DROP TABLE temp.recover_Stuff;
 * DETACH DATABASE rdb;
 * -- Move db.db to replace original db in filesystem.
 *
 *
 * Usage
 *
 * Given the goal of dealing with corruption, it would not be safe to
 * create a recovery table in the database being recovered.  So
 * recovery tables must be created in the temp database.  They are not
 * appropriate to persist, in any case.  [As a bonus, sqlite_master
 * tables can be recovered.  Perhaps more cute than useful, though.]
 *
 * The parameters are a specifier for the table to read, and a column
 * definition for each bit of data stored in that table.  The named
 * table must be convertable to a root page number by reading the
 * sqlite_master table.  Bare table names are assumed to be in
 * database 0 ("main"), other databases can be specified in db.table
 * fashion.
 *
 * Column definitions are similar to BUT NOT THE SAME AS those
 * provided to CREATE statements:
 *  column-def: column-name [type-name [STRICT] [NOT NULL]]
 *  type-name: (ANY|ROWID|INTEGER|FLOAT|NUMERIC|TEXT|BLOB)
 *
 * Only those exact type names are accepted, there is no type
 * intuition.  The only constraints accepted are STRICT (see below)
 * and NOT NULL.  Anything unexpected will cause the create to fail.
 *
 * ANY is a convenience to indicate that manifest typing is desired.
 * It is equivalent to not specifying a type at all.  The results for
 * such columns will have the type of the data's storage.  The exposed
 * schema will contain no type for that column.
 *
 * ROWID is used for columns representing aliases to the rowid
 * (INTEGER PRIMARY KEY, with or without AUTOINCREMENT), to make the
 * concept explicit.  Such columns are actually stored as NULL, so
 * they cannot be simply ignored.  The exposed schema will be INTEGER
 * for that column.
 *
 * NOT NULL causes rows with a NULL in that column to be skipped.  It
 * also adds NOT NULL to the column in the exposed schema.  If the
 * table has ever had columns added using ALTER TABLE, then those
 * columns implicitly contain NULL for rows which have not been
 * updated.  [Workaround using COALESCE() in your SELECT statement.]
 *
 * The created table is read-only, with no indices.  Any SELECT will
 * be a full-table scan, returning each valid row read from the
 * storage of the backing table.  The rowid will be the rowid of the
 * row from the backing table.  "Valid" means:
 * - The cell metadata for the row is well-formed.  Mainly this means that
 *   the cell header info describes a payload of the size indicated by
 *   the cell's payload size.
 * - The cell does not run off the page.
 * - The cell does not overlap any other cell on the page.
 * - The cell contains doesn't contain too many columns.
 * - The types of the serialized data match the indicated types (see below).
 *
 *
 * Type affinity versus type storage.
 *
 * http://www.sqlite.org/datatype3.html describes SQLite's type
 * affinity system.  The system provides for automated coercion of
 * types in certain cases, transparently enough that many developers
 * do not realize that it is happening.  Importantly, it implies that
 * the raw data stored in the database may not have the obvious type.
 *
 * Differences between the stored data types and the expected data
 * types may be a signal of corruption.  This module makes some
 * allowances for automatic coercion.  It is important to be concious
 * of the difference between the schema exposed by the module, and the
 * data types read from storage.  The following table describes how
 * the module interprets things:
 *
 * type     schema   data                     STRICT
 * ----     ------   ----                     ------
 * ANY      <none>   any                      any
 * ROWID    INTEGER  n/a                      n/a
 * INTEGER  INTEGER  integer                  integer
 * FLOAT    FLOAT    integer or float         float
 * NUMERIC  NUMERIC  integer, float, or text  integer or float
 * TEXT     TEXT     text or blob             text
 * BLOB     BLOB     blob                     blob
 *
 * type is the type provided to the recover module, schema is the
 * schema exposed by the module, data is the acceptable types of data
 * decoded from storage, and STRICT is a modification of that.
 *
 * A very loose recovery system might use ANY for all columns, then
 * use the appropriate sqlite3_column_*() calls to coerce to expected
 * types.  This doesn't provide much protection if a page from a
 * different table with the same column count is linked into an
 * inappropriate btree.
 *
 * A very tight recovery system might use STRICT to enforce typing on
 * all columns, preferring to skip rows which are valid at the storage
 * level but don't contain the right types.  Note that FLOAT STRICT is
 * almost certainly not appropriate, since integral values are
 * transparently stored as integers, when that is more efficient.
 *
 * Another option is to use ANY for all columns and inspect each
 * result manually (using sqlite3_column_*).  This should only be
 * necessary in cases where developers have used manifest typing (test
 * to make sure before you decide that you aren't using manifest
 * typing!).
 *
 *
 * Caveats
 *
 * Leaf pages not referenced by interior nodes will not be found.
 *
 * Leaf pages referenced from interior nodes of other tables will not
 * be resolved.
 *
 * Rows referencing invalid overflow pages will be skipped.
 *
 * SQlite rows have a header which describes how to interpret the rest
 * of the payload.  The header can be valid in cases where the rest of
 * the record is actually corrupt (in the sense that the data is not
 * the intended data).  This can especially happen WRT overflow pages,
 * as lack of atomic updates between pages is the primary form of
 * corruption I have seen in the wild.
 */
/* TODO(shess): It might be useful to allow DEFAULT in types to
 * specify what to do for NULL when an ALTER TABLE case comes up.
 * Unfortunately, simply adding it to the exposed schema and using
 * sqlite3_result_null() does not cause the default to be generate.
 * Handling it ourselves seems hard, unfortunately.
 */

#include <assert.h>
#include <ctype.h>
#include <stdio.h>
#include <string.h>

/* Internal things that are used:
 * u32, u64, i64 types.
 * Btree, Pager, and DbPage structs.
 * DbPage.pData, .pPager, and .pgno
 * sqlite3 struct.
 * sqlite3BtreePager() and sqlite3BtreeGetPageSize()
 * sqlite3PagerAcquire() and sqlite3PagerUnref()
 * getVarint32() and getVarint().
 */
#include "sqliteInt.h"

/* For debugging. */
#if 0
#define FNENTRY() fprintf(stderr, "In %s\n", __FUNCTION__)
#else
#define FNENTRY()
#endif

/* Generic constants and helper functions. */

/* Accepted types are specified by a mask. */
#define MASK_ROWID (1<<0)
#define MASK_INTEGER (1<<1)
#define MASK_FLOAT (1<<2)
#define MASK_TEXT (1<<3)
#define MASK_BLOB (1<<4)
#define MASK_NULL (1<<5)

/* TODO(shess): In the future, these will be used more often.  For
 * now, just pretend they're useful.
 */

/* True if iSerialType refers to a blob. */
static int SerialTypeIsBlob(u64 iSerialType){
  assert( iSerialType>=12 );
  return (iSerialType%2)==0;
}

/* Returns true if the serialized type represented by iSerialType is
 * compatible with the given type mask.
 */
static int SerialTypeIsCompatible(u64 iSerialType, unsigned char mask){
  switch( iSerialType ){
    case 0  : return (mask&MASK_NULL)!=0;
    case 1  : return (mask&MASK_INTEGER)!=0;
    case 2  : return (mask&MASK_INTEGER)!=0;
    case 3  : return (mask&MASK_INTEGER)!=0;
    case 4  : return (mask&MASK_INTEGER)!=0;
    case 5  : return (mask&MASK_INTEGER)!=0;
    case 6  : return (mask&MASK_INTEGER)!=0;
    case 7  : return (mask&MASK_FLOAT)!=0;
    case 8  : return (mask&MASK_INTEGER)!=0;
    case 9  : return (mask&MASK_INTEGER)!=0;
    case 10 : assert( !"RESERVED TYPE"); return 0;
    case 11 : assert( !"RESERVED TYPE"); return 0;
  }
  return (mask&(SerialTypeIsBlob(iSerialType) ? MASK_BLOB : MASK_TEXT));
}

/* Versions of strdup() with return values appropriate for
 * sqlite3_free().  malloc.c has sqlite3DbStrDup()/NDup(), but those
 * need sqlite3DbFree(), which seems intrusive.
 */
static char *sqlite3_strndup(const char *z, unsigned n){
  if( z==NULL ){
    return NULL;
  }

  char *zNew = sqlite3_malloc(n+1);
  if( zNew!=NULL ){
    memcpy(zNew, z, n);
    zNew[n] = '\0';
  }
  return zNew;
}
static char *sqlite3_strdup(const char *z){
  if( z==NULL ){
    return NULL;
  }
  return sqlite3_strndup(z, strlen(z));
}

/* Fetch the page number of zTable in zDb from sqlite_master in zDb,
 * and put it in *piRootPage.
 */
static int getRootPage(sqlite3 *db, const char *zDb, const char *zTable,
                       unsigned *piRootPage){
  if( strcmp(zTable, "sqlite_master")==0 ){
    *piRootPage = 1;
    return SQLITE_OK;
  }

  char *zSql = sqlite3_mprintf("SELECT rootpage FROM %s.sqlite_master "
                               "WHERE type = 'table' AND tbl_name = %Q",
                               zDb, zTable);
  if( !zSql ){
    return SQLITE_NOMEM;
  }

  sqlite3_stmt *pStmt = 0;
  int rc = sqlite3_prepare_v2(db, zSql, -1, &pStmt, 0);
  sqlite3_free(zSql);
  if( rc!=SQLITE_OK ){
    return rc;
  }

  /* Require a result. */
  rc = sqlite3_step(pStmt);
  if( rc==SQLITE_DONE ){
    rc = SQLITE_CORRUPT;
  }else if( rc==SQLITE_ROW ){
    *piRootPage = sqlite3_column_int(pStmt, 0);

    /* Require only one result. */
    rc = sqlite3_step(pStmt);
    if( rc==SQLITE_DONE ){
      rc = SQLITE_OK;
    }else if( rc==SQLITE_ROW ){
      rc = SQLITE_CORRUPT;
    }
  }
  sqlite3_finalize(pStmt);
  return rc;
}

typedef struct Recover Recover;
struct Recover {
  sqlite3_vtab base;
  sqlite3 *db;                /* Host database connection */
  char *zDb;                  /* Database containing target table */
  char *zTable;               /* Target table */
  int nCols;                  /* Number of columns in target table */
  unsigned char *pTypes;      /* Types of columns in target table */
};

/* Internal helper for deleting the module. */
static void recoverRelease(Recover *pRecover){
  sqlite3_free(pRecover->zDb);
  sqlite3_free(pRecover->zTable);
  sqlite3_free(pRecover->pTypes);
  memset(pRecover, 0xA5, sizeof(*pRecover));
  sqlite3_free(pRecover);
}

/* Helper function for initializing the module.  Forward-declared so
 * recoverCreate() and recoverConnect() can see it.
 */
static int recoverInit(
  sqlite3 *, void *, int, const char *const*, sqlite3_vtab **, char **
);

static int recoverCreate(
  sqlite3 *db,
  void *pAux,
  int argc, const char *const*argv,
  sqlite3_vtab **ppVtab,
  char **pzErr
){
  FNENTRY();
  return recoverInit(db, pAux, argc, argv, ppVtab, pzErr);
}

/* This should never be called. */
static int recoverConnect(
  sqlite3 *db,
  void *pAux,
  int argc, const char *const*argv,
  sqlite3_vtab **ppVtab,
  char **pzErr
){
  FNENTRY();
  return recoverInit(db, pAux, argc, argv, ppVtab, pzErr);
}

/* No indices supported. */
static int recoverBestIndex(sqlite3_vtab *tab, sqlite3_index_info *pIdxInfo){
  FNENTRY();
  return SQLITE_OK;
}

/* Logically, this should never be called. */
static int recoverDisconnect(sqlite3_vtab *pVtab){
  FNENTRY();
  recoverRelease((Recover*)pVtab);
  return SQLITE_OK;
}

static int recoverDestroy(sqlite3_vtab *pVtab){
  FNENTRY();
  recoverRelease((Recover*)pVtab);
  return SQLITE_OK;
}

typedef struct RecoverCursor RecoverCursor;
struct RecoverCursor {
  sqlite3_vtab_cursor base;
  i64 iRowid;  /* TODO(shess): Implement for real. */
  int bEOF;
};

static int recoverOpen(sqlite3_vtab *pVTab, sqlite3_vtab_cursor **ppCursor){
  FNENTRY();

  Recover *pRecover = (Recover*)pVTab;

  unsigned iRootPage = 0;
  int rc = getRootPage(pRecover->db, pRecover->zDb, pRecover->zTable,
                       &iRootPage);
  if( rc!=SQLITE_OK ){
    return rc;
  }

  /* TODO(shess): Implement some real stuff in here. */

  RecoverCursor *pCursor = sqlite3_malloc(sizeof(RecoverCursor));
  if( !pCursor ){
    return SQLITE_NOMEM;
  }
  memset(pCursor, 0, sizeof(*pCursor));
  pCursor->base.pVtab = pVTab;
  pCursor->iRowid = 0;

  *ppCursor = (sqlite3_vtab_cursor*)pCursor;
  return SQLITE_OK;
}

static int recoverClose(sqlite3_vtab_cursor *cur){
  FNENTRY();
  RecoverCursor *pCursor = (RecoverCursor*)cur;
  memset(pCursor, 0xA5, sizeof(*pCursor));
  sqlite3_free(cur);
  return SQLITE_OK;
}

/* TODO(shess): Some data for purposes of mocking things.  Will go
 * away.
 */
static struct {
  u64 iColType;
  const char *zTypeName;
} gMockData[] = {
  { 0, "NULL"},
  { 1, "INTEGER"},
  { 7, "FLOAT"},
  { 13, "TEXT"},
  { 12, "BLOB"},
};

static int recoverNext(sqlite3_vtab_cursor *pVtabCursor){
  FNENTRY();
  RecoverCursor *pCursor = (RecoverCursor*)pVtabCursor;
  Recover *pRecover = (Recover*)pCursor->base.pVtab;

  /* iRowid is 1-base, gMockData is 0-based. */
  while( pCursor->iRowid<ArraySize(gMockData) ){
    pCursor->iRowid++;
    if( SerialTypeIsCompatible(gMockData[pCursor->iRowid-1].iColType,
                               pRecover->pTypes[pRecover->nCols-1]) ){
      return SQLITE_OK;
    }
  }
  pCursor->bEOF = 1;
  return SQLITE_OK;
}

static int recoverFilter(
  sqlite3_vtab_cursor *pVtabCursor,
  int idxNum, const char *idxStr,
  int argc, sqlite3_value **argv
){
  FNENTRY();
  return recoverNext(pVtabCursor);
}

static int recoverEof(sqlite3_vtab_cursor *pVtabCursor){
  FNENTRY();
  RecoverCursor *pCursor = (RecoverCursor*)pVtabCursor;
  return pCursor->bEOF;
}

static int recoverColumn(sqlite3_vtab_cursor *cur, sqlite3_context *ctx, int i){
  FNENTRY();
  RecoverCursor *pCursor = (RecoverCursor*)cur;
  Recover *pRecover = (Recover*)pCursor->base.pVtab;

  if( i>=pRecover->nCols ){
    return SQLITE_ERROR;
  }

  /* ROWID alias. */
  if( (pRecover->pTypes[i]&MASK_ROWID) ){
    sqlite3_result_int64(ctx, pCursor->iRowid);
    return SQLITE_OK;
  }

  /* TODO(shess): Replace this with real code. */
  if( pCursor->iRowid<1 || pCursor->iRowid>ArraySize(gMockData)+1 ){
    return SQLITE_ERROR;
  }
  if( i==pRecover->nCols-2 ){
    sqlite3_result_text(ctx, gMockData[pCursor->iRowid-1].zTypeName, -1,
                        SQLITE_STATIC);
  }else if( i==pRecover->nCols-1 ){
    switch( gMockData[pCursor->iRowid-1].iColType ){
      case 0 : sqlite3_result_null(ctx); break;
      case 1 : sqlite3_result_int(ctx, 17); break;
      case 7 : sqlite3_result_double(ctx, 3.1415927); break;
      case 13 :
        sqlite3_result_text(ctx, "This is text", -1, SQLITE_STATIC);
        break;
      case 12 :
        sqlite3_result_blob(ctx, "This is a blob", 14, SQLITE_STATIC);
        break;
    }
  }
  return SQLITE_OK;
}

static int recoverRowid(sqlite3_vtab_cursor *pVtabCursor, sqlite_int64 *pRowid){
  FNENTRY();
  RecoverCursor *pCursor = (RecoverCursor*)pVtabCursor;
  *pRowid = pCursor->iRowid;
  return SQLITE_OK;
}

static sqlite3_module recoverModule = {
  0,                         /* iVersion */
  recoverCreate,             /* xCreate - create a table */
  recoverConnect,            /* xConnect - connect to an existing table */
  recoverBestIndex,          /* xBestIndex - Determine search strategy */
  recoverDisconnect,         /* xDisconnect - Disconnect from a table */
  recoverDestroy,            /* xDestroy - Drop a table */
  recoverOpen,               /* xOpen - open a cursor */
  recoverClose,              /* xClose - close a cursor */
  recoverFilter,             /* xFilter - configure scan constraints */
  recoverNext,               /* xNext - advance a cursor */
  recoverEof,                /* xEof */
  recoverColumn,             /* xColumn - read data */
  recoverRowid,              /* xRowid - read data */
  0,                         /* xUpdate - write data */
  0,                         /* xBegin - begin transaction */
  0,                         /* xSync - sync transaction */
  0,                         /* xCommit - commit transaction */
  0,                         /* xRollback - rollback transaction */
  0,                         /* xFindFunction - function overloading */
  0,                         /* xRename - rename the table */
};

int recoverVtableInit(sqlite3 *db){
  return sqlite3_create_module_v2(db, "recover", &recoverModule, NULL, 0);
}

/* This section of code is for parsing the create input and
 * initializing the module.
 */

/* Find the next word in zText and place the endpoints in pzWord*.
 * Returns true if the word is non-empty.  "Word" is defined as
 * alphanumeric plus '_' at this time.
 */
static int findWord(const char *zText,
                    const char **pzWordStart, const char **pzWordEnd){
  while( isspace(*zText) ){
    zText++;
  }
  *pzWordStart = zText;
  while( isalnum(*zText) || *zText=='_' ){
    zText++;
  }
  int r = zText>*pzWordStart;  /* In case pzWordStart==pzWordEnd */
  *pzWordEnd = zText;
  return r;
}

/* Return true if the next word in zText is zWord, also setting
 * *pzContinue to the character after the word.
 */
static int expectWord(const char *zText, const char *zWord,
                      const char **pzContinue){
  const char *zWordStart, *zWordEnd;
  if( findWord(zText, &zWordStart, &zWordEnd) &&
      strncasecmp(zWord, zWordStart, zWordEnd - zWordStart)==0 ){
    *pzContinue = zWordEnd;
    return 1;
  }
  return 0;
}

/* Parse the name and type information out of parameter.  In case of
 * success, *pzNameStart/End contain the name of the column,
 * *pzTypeStart/End contain the top-level type, and *pTypeMask has the
 * type mask to use for the column.
 */
static int findNameAndType(const char *parameter,
                           const char **pzNameStart, const char **pzNameEnd,
                           const char **pzTypeStart, const char **pzTypeEnd,
                           unsigned char *pTypeMask){
  if( !findWord(parameter, pzNameStart, pzNameEnd) ){
    return SQLITE_MISUSE;
  }

  /* Manifest typing, accept any storage type. */
  if( !findWord(*pzNameEnd, pzTypeStart, pzTypeEnd) ){
    *pzTypeEnd = *pzTypeStart = "";
    *pTypeMask = MASK_INTEGER | MASK_FLOAT | MASK_BLOB | MASK_TEXT | MASK_NULL;
    return SQLITE_OK;
  }

  /* strictMask is used for STRICT, strictMask|otherMask if STRICT is
   * not supplied.  zReplace provides an alternate type to expose to
   * the caller.
   */
  static struct {
    const char *zName;
    unsigned char strictMask;
    unsigned char otherMask;
    const char *zReplace;
  } kTypeInfo[] = {
    { "ANY",
      MASK_INTEGER | MASK_FLOAT | MASK_BLOB | MASK_TEXT | MASK_NULL,
      0, "",
    },
    { "ROWID",   MASK_INTEGER | MASK_ROWID,             0, "INTEGER", },
    { "INTEGER", MASK_INTEGER | MASK_NULL,              0, NULL, },
    { "FLOAT",   MASK_FLOAT | MASK_NULL,                MASK_INTEGER, NULL, },
    { "NUMERIC", MASK_INTEGER | MASK_FLOAT | MASK_NULL, MASK_TEXT, NULL, },
    { "TEXT",    MASK_TEXT | MASK_NULL,                 MASK_BLOB, NULL, },
    { "BLOB",    MASK_BLOB | MASK_NULL,                 0, NULL, },
  };

  unsigned i, nNameLen = *pzTypeEnd - *pzTypeStart;
  for( i=0; i<ArraySize(kTypeInfo); ++i ){
    if( strncasecmp(kTypeInfo[i].zName, *pzTypeStart, nNameLen)==0 ){
      break;
    }
  }
  if( i==ArraySize(kTypeInfo) ){
    return SQLITE_MISUSE;
  }

  const char *zEnd = *pzTypeEnd;
  int bStrict = 0;
  if( expectWord(zEnd, "STRICT", &zEnd) ){
    /* TODO(shess): Ick.  But I don't want another single-purpose
     * flag, either.
     */
    if( kTypeInfo[i].zReplace && !kTypeInfo[i].zReplace[0] ){
      return SQLITE_MISUSE;
    }
    bStrict = 1;
  }

  int bNotNull = 0;
  if( expectWord(zEnd, "NOT", &zEnd) ){
    if( expectWord(zEnd, "NULL", &zEnd) ){
      bNotNull = 1;
    }else{
      /* Anything other than NULL after NOT is an error. */
      return SQLITE_MISUSE;
    }
  }

  /* Anything else is an error. */
  const char *zDummy;
  if( findWord(zEnd, &zDummy, &zDummy) ){
    return SQLITE_MISUSE;
  }

  *pTypeMask = kTypeInfo[i].strictMask;
  if( !bStrict ){
    *pTypeMask |= kTypeInfo[i].otherMask;
  }
  if( bNotNull ){
    *pTypeMask &= ~MASK_NULL;
  }
  if( kTypeInfo[i].zReplace ){
    *pzTypeStart = kTypeInfo[i].zReplace;
    *pzTypeEnd = *pzTypeStart + strlen(*pzTypeStart);
  }
  return SQLITE_OK;
}

/* Parse the arguments, placing type masks in *pTypes and the exposed
 * schema in *pzCreateSql (for sqlite3_declare_vtab).
 */
static int ParseColumnsAndGenerateCreate(int nCols, const char *const *pCols,
                                         char **pzCreateSql,
                                         unsigned char *pTypes,
                                         char **pzErr){
  char *zCreateSql = sqlite3_mprintf("CREATE TABLE x(");
  if( !zCreateSql ){
    return SQLITE_NOMEM;
  }

  unsigned i;
  for( i=0; i<nCols; i++ ){
    const char *zNameStart, *zNameEnd;
    const char *zTypeStart, *zTypeEnd;
    int rc = findNameAndType(pCols[i],
                             &zNameStart, &zNameEnd,
                             &zTypeStart, &zTypeEnd,
                             &pTypes[i]);
    if( rc!=SQLITE_OK ){
      *pzErr = sqlite3_mprintf("unable to parse column %d", i);
      sqlite3_free(zCreateSql);
      return rc;
    }

    const char *zNotNull = "";
    if( !(pTypes[i]&MASK_NULL) ){
      zNotNull = " NOT NULL";
    }

    /* Add name and type to the create statement. */
    const char *zSep = (i < nCols - 1 ? ", " : ")");
    zCreateSql = sqlite3_mprintf("%z%.*s %.*s%s%s",
                                 zCreateSql,
                                 zNameEnd - zNameStart, zNameStart,
                                 zTypeEnd - zTypeStart, zTypeStart,
                                 zNotNull, zSep);
    if( !zCreateSql ){
      return SQLITE_NOMEM;
    }
  }

  *pzCreateSql = zCreateSql;
  return SQLITE_OK;
}

/* Helper function for initializing the module. */
/* TODO(shess): Since connect isn't supported, could inline into
 * recoverCreate().
 */
/* TODO(shess): Explore cases where it would make sense to set *pzErr. */
static int recoverInit(
  sqlite3 *db,                        /* Database connection */
  void *pAux,                         /* unused */
  int argc, const char *const*argv,   /* Parameters to CREATE TABLE statement */
  sqlite3_vtab **ppVtab,              /* OUT: New virtual table */
  char **pzErr                        /* OUT: Error message, if any */
){
  /* argv[0] module name
   * argv[1] db name for virtual table
   * argv[2] virtual table name
   * argv[3] backing table name
   * argv[4] columns
   */
  const int kTypeCol = 4;
  int nCols = argc - kTypeCol;

  /* Require to be in the temp database. */
  if( strcasecmp(argv[1], "temp")!=0 ){
    *pzErr = sqlite3_mprintf("recover table must be in temp database");
    return SQLITE_MISUSE;
  }

  /* Need the backing table and at least one column. */
  if( nCols<1 ){
    *pzErr = sqlite3_mprintf("no columns specified");
    return SQLITE_MISUSE;
  }

  Recover *pRecover = sqlite3_malloc(sizeof(Recover));
  if( !pRecover ){
    return SQLITE_NOMEM;
  }
  memset(pRecover, 0, sizeof(*pRecover));
  pRecover->base.pModule = &recoverModule;
  pRecover->db = db;

  /* Parse out db.table, assuming main if no dot. */
  char *zDot = strchr(argv[3], '.');
  if( !zDot ){
    pRecover->zDb = sqlite3_strdup(db->aDb[0].zName);
    pRecover->zTable = sqlite3_strdup(argv[3]);
  }else if( zDot>argv[3] && zDot[1]!='\0' ){
    pRecover->zDb = sqlite3_strndup(argv[3], zDot - argv[3]);
    pRecover->zTable = sqlite3_strdup(zDot + 1);
  }else{
    /* ".table" or "db." not allowed. */
    *pzErr = sqlite3_mprintf("ill-formed table specifier");
    recoverRelease(pRecover);
    return SQLITE_ERROR;
  }

  pRecover->nCols = nCols;
  pRecover->pTypes = sqlite3_malloc(pRecover->nCols);
  if( !pRecover->zDb || !pRecover->zTable || !pRecover->pTypes ){
    recoverRelease(pRecover);
    return SQLITE_NOMEM;
  }

  /* Require the backing table to exist. */
  /* TODO(shess): Be more pedantic about the form of the descriptor
   * string.  This already fails for poorly-formed strings, simply
   * because there won't be a root page, but it would make more sense
   * to be explicit.
   */
  unsigned iRootPage;
  int rc = getRootPage(pRecover->db, pRecover->zDb, pRecover->zTable,
                       &iRootPage);
  if( rc!=SQLITE_OK ){
    *pzErr = sqlite3_mprintf("unable to find backing table");
    recoverRelease(pRecover);
    return rc;
  }

  /* Parse the column definitions. */
  char *zCreateSql;
  rc = ParseColumnsAndGenerateCreate(nCols, argv + kTypeCol,
                                     &zCreateSql, pRecover->pTypes, pzErr);
  if( rc!=SQLITE_OK ){
    recoverRelease(pRecover);
    return rc;
  }

  rc = sqlite3_declare_vtab(db, zCreateSql);
  sqlite3_free(zCreateSql);
  if( rc!=SQLITE_OK ){
    recoverRelease(pRecover);
    return rc;
  }

  *ppVtab = (sqlite3_vtab *)pRecover;
  return SQLITE_OK;
}