The complete work-in-progress SQLite recover virtual table.

third_party/sqlite/src/src/recover.c

Index: third_party/sqlite/src/src/recover.c
diff --git a/third_party/sqlite/src/src/recover.c b/third_party/sqlite/src/src/recover.c
new file mode 100644
index 0000000000000000000000000000000000000000..503aec84ad3c111e8492e712ada448505c7a915f
--- /dev/null
+++ b/third_party/sqlite/src/src/recover.c
@@ -0,0 +1,2116 @@
+/* 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.
+ */
+/* Also provided are simple functions which can be used to explore a
+ * SQLite database file.  hexdump(pageno) provides a hex/ascii dump of
+ * page #pageno, with all-0x00 lines elided.  dumppage(pageno)
+ * attempts to provide a descriptive dump of all data on that page,
+ * while dumppage(pageno, "header") only dumps the header info.
+ *
+ * TODO(shess): These functions are incomplete!
+ */
+/* The implementation is via a series of nested cursors.  The
+ * cursors are implemented with this pattern:
+ *
+ * // Creates the cursor using various initialization info.
+ * int cursorCreate(...);
+ *
+ * // Returns 1 if there is no more data, 0 otherwise.
+ * int cursorEOF(Cursor *pCursor);
+ *
+ * // Various accessors can be used if not at EOF.
+ *
+ * // Move to the next item.
+ * int cursorNext(Cursor *pCursor);
+ *
+ * // Destroy the memory associated with the cursor.
+ * void cursorDestroy(Cursor *pCursor);
+ *
+ * References in the following ar to sections at
+ * http://www.sqlite.org/fileformat2.html .
+ *
+ * RecoverLeafCursor iterates the records in a leaf table node
+ * described in section 1.5 "B-tree Pages".
+ *
+ * RecoverInteriorCursor iterates the child pages in an interior table
+ * node described in section 1.5 "B-tree Pages".
+ *
+ * RecoverCursor pulls these together to iterate the rows of a table,
+ * returning results via the SQLite virtual table interface.
+ */
+/* TODO(shess): Some unlinked leaf pages could potentially be found by
+ * looking at the pointer map pages to find pages which reference an
+ * interior node but which that node does not reference.
+ */
+/* TODO(shess): This code assumes UTF8 databases.  Add handling for
+ * UTF16le and UTF16be.
+ */
+/* TODO(shess): The code doesn't monitor the freelist, so it can't
+ * enforce that cells don't overlap free areas.
+ */
+/* TODO(shess): The code is pretty lax WRT interior nodes.  It only
+ * reads the 32-bit child page number from the cell, ignoring the
+ * rowid.  This is probably fine, though.
+ */
+/* TODO(shess): Use auto_vacuum information to figure out which table
+ * a page is from.
+ */
+/* TODO(shess): Review to convert recursive routines into iterative as
+ * appropriate.
+ */
+/* TODO(shess): It might be useful to allow DEFAULT in types to
+ * specify what to do for NULL.  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. */
+
+static const unsigned char kTableLeafPage = 0x0D;
+static const unsigned char kTableInteriorPage = 0x05;
+
+/* Accepted types are specified by a mask.  Conveniently, SQLite's
+ * data-storage types use small integers from 1 to 5...
+ */
+static const unsigned char kMaskNull = (1<<SQLITE_NULL);
+static const unsigned char kMaskInteger = (1<<SQLITE_INTEGER);
+static const unsigned char kMaskFloat = (1<<SQLITE_FLOAT);
+static const unsigned char kMaskBlob = (1<<SQLITE_BLOB);
+static const unsigned char kMaskText = (1<<SQLITE_TEXT);
+static const unsigned char kMaskRowid = 1;
+
+/* Helpers to decode fixed-size fields. */
+static unsigned decodeUnsigned16(const unsigned char *pData){
+  return (pData[0]<<8) + pData[1];
+}
+static unsigned decodeUnsigned32(const unsigned char *pData){
+  return (pData[0]<<24) + (pData[1]<<16) + (pData[2]<<8) + pData[3];
+}
+static int decodeInt8(const unsigned char *pData){
+  return (int)(*pData);
+}
+static int decodeInt16(const unsigned char *pData){
+  return (decodeInt8(pData)<<8) + pData[1];
+}
+static int decodeInt24(const unsigned char *pData){
+  return (decodeInt16(pData)<<8) + pData[2];
+}
+static int decodeInt32(const unsigned char *pData){
+  return (decodeInt24(pData)<<8) + pData[3];
+}
+static i64 decodeInt48(const unsigned char *pData){
+  return (((i64)decodeInt32(pData))<<16) + (pData[4]<<8) + pData[5];
+}
+static i64 decodeInt64(const unsigned char *pData){
+  return (decodeInt48(pData)<<16) + (pData[6]<<8) + pData[7];
+}
+static double decodeFloat64(const unsigned char *pData){
+  u64 x = decodeInt64(pData);
+  return *((double*)(&x));
+}
+
+/* Return 1 if c varints can be read from pData/nData. */
+static int checkVarints(const unsigned char *pData, unsigned nData,
+                        unsigned n){
+  /* In the worst case the decoder takes all 8 bits of the 9th byte. */
+  if( nData>=9*n ){
+    return 1;
+  }
+
+  unsigned nCur = 0, nFound = 0;
+  unsigned i;
+  for( i=0; nFound<n && i<nData; ++i ){
+    nCur++;
+    if( nCur==9 || !(pData[i]&0x80) ){
+      nFound++;
+      nCur = 0;
+    }
+  }
+
+  return nFound==n;
+}
+static int checkVarint(const unsigned char *pData, unsigned nData){
+  /* In the worst case the decoder takes all 8 bits of the 9th byte. */
+  if( nData>=9 ){
+    return 1;
+  }
+
+  /* Look for a high-bit-clear byte in what's left. */
+  unsigned i;
+  for( i=0; i<nData; ++i ){
+    if( !(pData[i]&0x80) ){
+      return 1;
+    }
+  }
+
+  /* Cannot decode in the space given. */
+  return 0;
+}
+
+static const unsigned char *PageData(DbPage *pPage, unsigned iOffset){
+  assert( iOffset<=pPage->nPageSize );
+  return pPage->pData + iOffset;
+}
+static const unsigned char *PageHeader(DbPage *pPage){
+  if( pPage->pgno==1 ){
+    return PageData(pPage, 100);
+  }else{
+    return PageData(pPage, 0);
+  }
+}
+
+/* Helper to fetch the pager and page size for the named database. */
+static int GetPager(sqlite3 *db, const char *zName,
+                    Pager **pPager, unsigned *pnPageSize){
+  Btree *pBt = NULL;
+  int i;
+  for( i=0; i<db->nDb; ++i ){
+    if( !strcasecmp(db->aDb[i].zName, zName) ){
+      pBt = db->aDb[i].pBt;
+      break;
+    }
+  }
+  if( !pBt ){
+    return SQLITE_ERROR;
+  }
+
+  *pPager = sqlite3BtreePager(pBt);
+  *pnPageSize = sqlite3BtreeGetPageSize(pBt);
+  return SQLITE_OK;
+}
+
+/* iSerialType is a type read from a record header.  See "2.1 Record Format".
+ * http://www.sqlite.org/fileformat2.html#record_format
+ */
+static u64 SerialTypeLength(u64 iSerialType){
+  switch( iSerialType ){
+    case 0 : return 0;  /* NULL */
+    case 1 : return 1;  /* Various integers. */
+    case 2 : return 2;
+    case 3 : return 3;
+    case 4 : return 4;
+    case 5 : return 6;
+    case 6 : return 8;
+    case 7 : return 8;  /* 64-bit float. */
+    case 8 : return 0;  /* Constant 0. */
+    case 9 : return 0;  /* Constant 1. */
+    case 10 : case 11 : assert( !"RESERVED TYPE"); return 0;
+    default :           /* TEXT or BLOB. */
+      return (iSerialType>>1) - 6;
+  }
+  assert( !"UNKNOWN TYPE");
+  return 0;
+}
+/* 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){
+  unsigned char check = 0;
+  switch( iSerialType ){
+    case 0 : check = kMaskNull; break;
+
+    case 1 : case 2 : case 3 : case 4 : case 5 : case 6 :
+    case 8 : case 9 : check = kMaskInteger; break;
+
+    case 7 : check = kMaskFloat; break;
+
+    case 10 : case 11 : assert( !"RESERVED TYPE"); break;
+
+    default :
+      check = SerialTypeIsBlob(iSerialType) ? kMaskBlob : kMaskText;
+      break;
+  }
+  return (mask&check)!=0;
+}
+
+/* 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") ){
+    *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;
+}
+
+/* Cursor for iterating the table interior nodes.  Each cell contains
+ * a page number and a rowid.  The child page contains items left of
+ * the rowid (less than).  The rightmost page of the subtree is stored
+ * in the header.
+ *
+ * interiorCursorDestroy - release all resources associated with the
+ *                         cursor and any parent cursors.
+ * interiorCursorCreate - create a cursor with the given parent and page.
+ * interiorCursorEOF - returns true if neither the cursor nor the
+ *                     parent cursors can return any more data.
+ * interiorCursorNextPage - fetch the next child page from the cursor.
+ *
+ * interiorCursorNextPage() returns the next child page at the
+ * cursor's level in the table tree.  If the cursor runs out of cells,
+ * the parent is called to provide new nodes at the same level, until
+ * all levels are exhausted.  Non-table pages at the cursor's depth
+ * are skipped, as are pages which would cause a loop, unexpected leaf
+ * table pages are returned to the caller, interior table pages are
+ * loaded and interated.  Returns SQLITE_ROW if a child page is
+ * returned.  SQLITE_DONE can be returned even if interiorCursorEOF()
+ * returned false, for instance if a parent cursor returns a final
+ * non-table page.
+ *
+ * Note that while interiorCursorNextPage() will refuse to follow
+ * loops, it does not keep track of pages returned for purposes of
+ * preventing duplication.
+ */
+typedef struct RecoverInteriorCursor RecoverInteriorCursor;
+struct RecoverInteriorCursor {
+  RecoverInteriorCursor *pParent; /* Parent node to this node. */
+  DbPage *pPage;                  /* Reference to leaf page. */
+  unsigned nPageSize;             /* Size of page. */
+  unsigned nChildren;             /* Number of children on the page. */
+  unsigned iChild;                /* Index of next child to return. */
+};
+
+static void interiorCursorDestroy(RecoverInteriorCursor *pCursor){
+  while( pCursor ){
+    RecoverInteriorCursor *p = pCursor;
+    pCursor = pCursor->pParent;
+
+    if( p->pPage ){
+      sqlite3PagerUnref(p->pPage);
+      p->pPage = NULL;
+    }
+
+    memset(p, 0xA5, sizeof(*p));
+    sqlite3_free(p);
+  }
+}
+
+/* Internal helper.  Reset storage in preparation for iterating pPage. */
+static void interiorCursorSetPage(RecoverInteriorCursor *pCursor,
+                                  DbPage *pPage){
+  if( pCursor->pPage ){
+    sqlite3PagerUnref(pCursor->pPage);
+    pCursor->pPage = NULL;
+  }
+  pCursor->pPage = pPage;
+  pCursor->iChild = 0;
+
+  /* A child for each cell, plus one in the header. */
+  pCursor->nChildren = decodeUnsigned16(PageHeader(pPage) + 3) + 1;
+}
+
+static int interiorCursorCreate(RecoverInteriorCursor *pParent,
+                                DbPage *pPage, int nPageSize,
+                                RecoverInteriorCursor **ppCursor){
+  RecoverInteriorCursor *pCursor =
+    sqlite3_malloc(sizeof(RecoverInteriorCursor));
+  if( !pCursor ){
+    return SQLITE_NOMEM;
+  }
+
+  memset(pCursor, 0, sizeof(*pCursor));
+  pCursor->pParent = pParent;
+  pCursor->nPageSize = nPageSize;
+  interiorCursorSetPage(pCursor, pPage);
+  *ppCursor = pCursor;
+  return SQLITE_OK;
+}
+
+/* Internal helper.  Return the child page number at iChild. */
+static unsigned interiorCursorChildPage(RecoverInteriorCursor *pCursor){
+  assert( pCursor->iChild<pCursor->nChildren );
+
+  /* Rightmost child is in the header. */
+  const unsigned char *pPageHeader = PageHeader(pCursor->pPage);
+  if( pCursor->iChild==pCursor->nChildren-1 ){
+    return decodeUnsigned32(pPageHeader + 8);
+  }
+
+  /* Each cell is a 4-byte integer page number and a varint rowid
+   * which is greater than the rowid of items in that sub-tree.  This
+   * code ignores the latter constraint.  Note that the offset is from
+   * the beginning of the page, not from the header.
+   */
+  const unsigned char *pOffsets = pPageHeader + 12;
+  unsigned offset = decodeUnsigned16(pOffsets + pCursor->iChild*2);
+  if( offset<=pCursor->nPageSize-4 ){
+    return decodeUnsigned32(PageData(pCursor->pPage, offset));
+  }
+
+  /* If the offset is broken, return an invalid page number. */
+  return 0;
+}
+
+static int interiorCursorEOF(RecoverInteriorCursor *pCursor){
+  while( pCursor ){
+    if( pCursor->iChild<pCursor->nChildren ){
+      return 0;
+    }
+    pCursor = pCursor->pParent;
+  }
+  return 1;
+}
+
+/* Internal helper.  Used to detect if iPage would cause a loop. */
+static int interiorCursorPageInUse(RecoverInteriorCursor *pCursor,
+                                   unsigned iPage){
+  while( pCursor ){
+    if( pCursor->pPage->pgno==iPage ){
+      return 1;
+    }
+    pCursor = pCursor->pParent;
+  }
+  return 0;
+}
+
+/* NOTE(shess): This code may recurse up to the parent.  The stack
+ * depth should be proportional to the log of the table size, so it
+ * has not seemed worthwhile to convert it to an iterative form.
+ */
+static int interiorCursorNextPage(RecoverInteriorCursor *pCursor,
+                                  DbPage **ppPage){
+  while( !interiorCursorEOF(pCursor) ){
+    /* This page has no more children.  Get next page from parent. */
+    while( pCursor->iChild>=pCursor->nChildren ){
+      if( !pCursor->pParent ){
+        return SQLITE_DONE;
+      }
+
+      int rc = interiorCursorNextPage(pCursor->pParent, ppPage);
+      if( rc!=SQLITE_ROW ){
+        return rc;
+      }
+
+      const unsigned char *pPageHeader = PageHeader(*ppPage);
+      if( *pPageHeader==kTableLeafPage ){
+        return SQLITE_ROW;
+      }else if( *pPageHeader==kTableInteriorPage ){
+        interiorCursorSetPage(pCursor, *ppPage);
+        *ppPage = NULL;
+      }else{
+        sqlite3PagerUnref(*ppPage);
+        *ppPage = NULL;
+      }
+    }
+
+    /* Find a valid child page which isn't in use. */
+    while( pCursor->iChild<pCursor->nChildren ){
+      const unsigned iPage = interiorCursorChildPage(pCursor);
+      pCursor->iChild++;
+      if( interiorCursorPageInUse(pCursor, iPage) ){
+        fprintf(stderr, "Loop detected at %d\n", iPage);
+      }else{
+        int rc = sqlite3PagerAcquire(pCursor->pPage->pPager, iPage, ppPage, 0);
+        if( rc==SQLITE_OK ){
+          return SQLITE_ROW;
+        }
+      }
+    }
+  }
+  return SQLITE_DONE;
+}
+
+/* Structure for dealing with overflow pages.  The row's main page
+ * stores an overflow page number after the local payload, with
+ * overflow pages forming a linked list forward from there.
+ *
+ * overflowDestroy - releases all resources associated with the structure.
+ * overflowMaybeCreate - create the overflow structure if it is needed
+ *                       to represent the given record.  See function comment.
+ * overflowGetSegment - fetch a segment from the record, accounting
+ *                      for overflow pages.  Segments which are not
+ *                      entirely contained with a page are constructed
+ *                      into a buffer which is returned.  See function comment.
+ */
+/* TODO(shess): Since overflowMaybeCreate() can pass back a NULL, it
+ * might make sense to have overflowDestroy() and overflowGetSegment()
+ * accept a NULL and make appropriate decisions.
+ */
+typedef struct RecoverOverflow RecoverOverflow;
+struct RecoverOverflow {
+  RecoverOverflow *pNextOverflow;
+  DbPage *pPage;
+  unsigned nPageSize;
+};
+
+static void overflowDestroy(RecoverOverflow *pOverflow){
+  while( pOverflow ){
+    RecoverOverflow *p = pOverflow;
+    pOverflow = p->pNextOverflow;
+
+    if( p->pPage ){
+      sqlite3PagerUnref(p->pPage);
+      p->pPage = NULL;
+    }
+
+    memset(p, 0xA5, sizeof(*p));
+    sqlite3_free(p);
+  }
+}
+
+/* The target record paylod begins at iOffset on pPage.  If nBytes can
+ * be satisfied entire in-page, then no overflow pages are needed and
+ * *pnLocalBytes is set to nBytes.  Otherwise, *ppOverflow is set to
+ * the head of a list of overflow pages, and *pnLocalBytes is set to
+ * the number of bytes local to pPage.
+ */
+static int overflowMaybeCreate(DbPage *pPage, unsigned nPageSize,
+                               unsigned iOffset, unsigned nBytes,
+                               unsigned *pnLocalBytes,
+                               RecoverOverflow **ppOverflow){
+  /* Calculations from section 1.5 of
+   * http://www.sqlite.org/fileformat2.html .
+   */
+  if( nBytes<=nPageSize-35 ){
+    *pnLocalBytes = nBytes;
+    *ppOverflow = NULL;
+    return SQLITE_OK;
+  }
+
+  unsigned m = ((nPageSize-12)*32/255)-23;
+  unsigned nLocalBytes = m + ((nBytes - m)%(nPageSize - 4));
+  if( nPageSize-35<nLocalBytes ){
+    nLocalBytes = nPageSize - 35;
+  }
+
+  /* Don't read off the end of the page. */
+  if( iOffset+nLocalBytes+4>nPageSize ){
+    return SQLITE_CORRUPT;
+  }
+
+  unsigned iNextPage = decodeUnsigned32(PageData(pPage, iOffset) + nLocalBytes);
+  int rc = SQLITE_OK;
+  nBytes -= nLocalBytes;
+
+  /* Ends of a linked list of overflow pages, and number of pages. */
+  RecoverOverflow *pFirstOverflow = NULL;
+  RecoverOverflow *pLastOverflow = NULL;
+  unsigned nPages = 0;
+
+  /* While there are more pages to read, and more pages are needed. */
+  while( iNextPage && nPages*(nPageSize-4)<nBytes ){
+    rc = sqlite3PagerAcquire(pPage->pPager, iNextPage, &pPage, 0);
+    if( rc!=SQLITE_OK ){
+      break;
+    }
+
+    RecoverOverflow *pOverflow = sqlite3_malloc(sizeof(RecoverOverflow));
+    if( !pOverflow ){
+      sqlite3PagerUnref(pPage);
+      rc = SQLITE_NOMEM;
+      break;
+    }
+    memset(pOverflow, 0, sizeof(*pOverflow));
+    pOverflow->pPage = pPage;
+    pOverflow->nPageSize = nPageSize;
+
+    if( !pFirstOverflow ){
+      pFirstOverflow = pOverflow;
+    }else{
+      pLastOverflow->pNextOverflow = pOverflow;
+    }
+    pLastOverflow = pOverflow;
+    nPages++;
+    iNextPage = decodeUnsigned32(pPage->pData);
+  }
+
+  /* If there were not enough pages, or too many, things are corrupt. */
+  if( rc==SQLITE_OK && (nPages*(nPageSize-4)<nBytes || iNextPage) ){
+    rc = SQLITE_CORRUPT;
+  }
+
+  if( rc==SQLITE_OK ){
+    *ppOverflow = pFirstOverflow;
+    *pnLocalBytes = nLocalBytes;
+  }else if( pFirstOverflow ){
+    overflowDestroy(pFirstOverflow);
+  }
+  return rc;
+}
+
+/* Gets a record segment, taking into account overflow pages.  pPage
+ * is the initial database page which contains nBytes of row data at
+ * iOffset. iReqOffset and nReqBytes give the target range desired,
+ * offset from the beginning of the record (at iOffset).  The segment
+ * is returned in *ppBase, possibly in an allocated buffer.  If so,
+ * *pbFree is set true and *ppBase should eventually be freed using
+ * sqlite3_free().
+ *
+ * If the request can be satisfied from pPage, then pOverflow can
+ * safely be NULL.
+ */
+static int overflowGetSegment(DbPage *pPage, unsigned iOffset, unsigned nBytes,
+                              RecoverOverflow *pOverflow,
+                              unsigned iReqOffset, unsigned nReqBytes,
+                              unsigned char **ppBase, int *pbFree){
+  /* Skip past initial pages. */
+  while( iReqOffset>=nBytes && pOverflow ){
+    iReqOffset -= nBytes;
+    pPage = pOverflow->pPage;
+    iOffset = 4;
+    nBytes = pOverflow->nPageSize - 4;
+    pOverflow = pOverflow->pNextOverflow;
+  }
+
+  /* If the requested data cannot be satisfied. */
+  if( iReqOffset+nReqBytes>iOffset+nBytes && !pOverflow ){
+    return SQLITE_ERROR;
+  }
+
+  /* If the requested segment is entirely within the local segment,
+   * return an internal pointer.
+   */
+  if( iReqOffset+nReqBytes<=nBytes ){
+    *ppBase = pPage->pData + iOffset + iReqOffset;
+    *pbFree = 0;
+    return SQLITE_OK;
+  }
+
+  /* Construct a buffer by copying pieces from multiple pages. */
+  unsigned char *pBase = sqlite3_malloc(nReqBytes);
+  if( !pBase ){
+    return SQLITE_NOMEM;
+  }
+  unsigned nBase = 0;
+  while( nBase<nReqBytes ){
+    unsigned nCopyBytes = nBytes - iReqOffset;
+    if( nReqBytes-nBase<nCopyBytes ){
+      nCopyBytes = nReqBytes - nBase;
+    }
+    memcpy(pBase + nBase, pPage->pData + iOffset + iReqOffset, nCopyBytes);
+    nBase += nCopyBytes;
+    if( pOverflow ){
+      iReqOffset = 0;
+      pPage = pOverflow->pPage;
+      iOffset = 4;
+      nBytes = pOverflow->nPageSize - 4;
+      pOverflow = pOverflow->pNextOverflow;
+    }else if( nBase<nReqBytes ){
+      /* More data is wanted, but no further overflow pages. */
+      sqlite3_free(pBase);
+      return SQLITE_ERROR;
+    }
+  }
+  assert( nBase==nReqBytes );
+  *ppBase = pBase;
+  *pbFree = 1;
+  return SQLITE_OK;
+}
+
+/* Primary structure for iterating the contents of a table.
+ *
+ * leafCursorDestroy - release all resources associated with the cursor.
+ * leafCursorCreate - create a cursor to iterate items from tree at
+ *                    the provided root page.
+ * leafCursorNextValidCell - get the cursor ready to return data from
+ *                           the next valid cell in the table.
+ * leafCursorCellRowid - get the current cell's rowid.
+ * leafCursorCellColumns - get current cell's column count.
+ * leafCursorCellColInfo - get type and data for a column in current cell.
+ *
+ * leafCursorNextValidCell skips cells which fail simple integrity
+ * checks, such as overlapping other cells, or being located at
+ * impossible offsets, or header data doesn't correctly describe
+ * payload data.  Returns SQLITE_ROW if a valid cell is found,
+ * SQLITE_DONE if all pages in the tree were exhausted.
+ *
+ * leafCursorCellColInfo() accounts for overflow pages using
+ * overflowGetSegment().
+ */
+typedef struct RecoverLeafCursor RecoverLeafCursor;
+struct RecoverLeafCursor {
+  RecoverInteriorCursor *pParent;  /* Parent node to this node. */
+  DbPage *pPage;                   /* Reference to leaf page. */
+  unsigned nPageSize;              /* Size of left page. */
+  unsigned nCells;                 /* Number of cells on the page. */
+  unsigned iCell;                  /* Current cell. */
+
+  /* Info parsed from the current cell's data.  pHeader is necessary
+   * because the header can extend into overflow, though it is usually
+   * in-page.
+   */
+  i64 iRowid;                      /* rowid parsed. */
+  unsigned nCols;                  /* how many data items in the cell. */
+  unsigned iPayloadOffset;         /* offset to payload. */
+  unsigned nPayloadBytes;          /* Size of payload. */
+  unsigned nLocalBytes;            /* In-page portion of payload. */
+  unsigned nHeaderBytes;           /* Size of payload header. */
+  unsigned char *pHeader;          /* Header data for payload. */
+  int bFreeHeader;                 /* True if header needs to be freed. */
+  RecoverOverflow *pOverflow;      /* Cell overflow info, if needed. */
+};
+
+/* Internal helper.  Returns SQLITE_OK if the page is accepted (in
+ * which case the caller no longer owns it).  pPage may be NULL after
+ * this call, in which case the passed page was either an interior
+ * page (in which case it adds a new parent cursor), or a non-table
+ * page.
+ */
+static int leafCursorLoadPage(RecoverLeafCursor *pCursor, DbPage *pPage){
+  if( pCursor->pPage ){
+    sqlite3PagerUnref(pCursor->pPage);
+    pCursor->pPage = NULL;
+  }
+
+  /* If the page is an unexpected interior node, inject a new stack
+   * layer and try again from there.
+   */
+  const unsigned char *pPageHeader = PageHeader(pPage);
+  if( pPageHeader[0]==kTableInteriorPage ){
+    RecoverInteriorCursor *pParent;
+    int rc = interiorCursorCreate(pCursor->pParent, pPage, pCursor->nPageSize,
+                                  &pParent);
+    if( rc!=SQLITE_OK ){
+      return rc;
+    }
+    pCursor->pParent = pParent;
+    return SQLITE_OK;
+  }
+
+  /* If the page is not a leaf node, skip it. */
+  if( pPageHeader[0]!=kTableLeafPage ){
+    sqlite3PagerUnref(pPage);
+    return SQLITE_OK;
+  }
+
+  /* Take ownership of the page and start decoding. */
+  pCursor->pPage = pPage;
+  pCursor->iCell = 0;
+  pCursor->nCells = decodeUnsigned16(pPageHeader + 3);
+  return SQLITE_OK;
+}
+
+static int leafCursorNextPage(RecoverLeafCursor *pCursor){
+  if( !pCursor->pParent ){
+    return SQLITE_DONE;
+  }
+
+  /* Get the next page from the parent and load it, until one sticks
+   * or there are no more pages.
+   */
+  do {
+    DbPage *pNextPage;
+    int rc = interiorCursorNextPage(pCursor->pParent, &pNextPage);
+    if( rc!=SQLITE_ROW ){
+      assert( rc==SQLITE_DONE );
+      return rc;
+    }
+
+    rc = leafCursorLoadPage(pCursor, pNextPage);
+    if( rc!=SQLITE_OK ){
+      sqlite3PagerUnref(pNextPage);
+      return rc;
+    }
+  } while( !pCursor->pPage );
+
+  return SQLITE_ROW;
+}
+
+static void leafCursorDestroy(RecoverLeafCursor *pCursor){
+  if( pCursor->pHeader && pCursor->bFreeHeader ){
+    sqlite3_free(pCursor->pHeader);
+    pCursor->pHeader = NULL;
+  }
+
+  if( pCursor->pOverflow ){
+    overflowDestroy(pCursor->pOverflow);
+    pCursor->pOverflow = NULL;
+  }
+
+  if( pCursor->pParent ){
+    interiorCursorDestroy(pCursor->pParent);
+    pCursor->pParent = NULL;
+  }
+
+  if( pCursor->pPage ){
+    sqlite3PagerUnref(pCursor->pPage);
+    pCursor->pPage = NULL;
+  }
+
+  memset(pCursor, 0xA5, sizeof(*pCursor));
+  sqlite3_free(pCursor);
+}
+
+static int leafCursorCreate(Pager *pPager, int nPageSize,
+                            int iRootPage, RecoverLeafCursor **ppCursor){
+  /* Start out with the root page. */
+  DbPage *pPage;
+  int rc = sqlite3PagerAcquire(pPager, iRootPage, &pPage, 0);
+  if( rc!=SQLITE_OK ){
+    return rc;
+  }
+
+  RecoverLeafCursor *pCursor = sqlite3_malloc(sizeof(RecoverLeafCursor));
+  if( !pCursor ){
+    sqlite3PagerUnref(pPage);
+    return SQLITE_NOMEM;
+  }
+  memset(pCursor, 0, sizeof(*pCursor));
+
+  pCursor->nPageSize = nPageSize;
+
+  rc = leafCursorLoadPage(pCursor, pPage);
+  if( rc!=SQLITE_OK ){
+    leafCursorDestroy(pCursor);
+    return rc;
+  }
+  if( !pCursor->pPage ){
+    rc = leafCursorNextPage(pCursor);
+    if( rc!=SQLITE_DONE && rc!=SQLITE_ROW ){
+      leafCursorDestroy(pCursor);
+      return rc;
+    }
+  }
+  /* TODO(shess): What should happen if leafCursorNextPage() returned
+   * SQLITE_DONE?  It means there were no leaf pages, so probably the
+   * cursor should just go directly to EOF.  Without having tested it,
+   * it looks like iCell == nCells == 0, so leafCursorNextValidCell()
+   * will return SQLITE_DONE.  But leafCursorCellSetup() called from
+   * recoverOpen() will not go well.
+   */
+  *ppCursor = pCursor;
+  return SQLITE_OK;
+}
+
+static int ValidateError(){
+  return SQLITE_ERROR;
+}
+
+static int leafCursorCellSetup(RecoverLeafCursor *pCursor){
+  assert( pCursor->iCell<pCursor->nCells );
+
+  if( pCursor->pOverflow ){
+    overflowDestroy(pCursor->pOverflow);
+    pCursor->pOverflow = NULL;
+  }
+
+  /* Find the offset to the row. */
+  const unsigned char *pPageHeader = PageHeader(pCursor->pPage);
+  const unsigned char *pOffsets = pPageHeader + 8;
+  const unsigned iOffset = decodeUnsigned16(pOffsets + pCursor->iCell*2);
+  if( iOffset>=pCursor->nPageSize ){
+    return ValidateError();
+  }
+
+  const unsigned char *pCell = PageData(pCursor->pPage, iOffset);
+  const unsigned nCellMaxBytes = pCursor->nPageSize - iOffset;
+  unsigned l = 0;
+
+  /* B-tree leaf cells lead with varint payload size and varint rowid.
+   * Then there is the varint header size.
+   */
+  /* TODO(shess): The smallest page size is 512 bytes, which has an m
+   * of 39.  Three varints need at most 27 bytes to encode.  I think.
+   */
+  if( !checkVarints(pCell + l, nCellMaxBytes - l, 3) ){
+    return ValidateError();
+  }
+
+  u32 nPayloadBytes;
+  l += getVarint32(pCell + l, nPayloadBytes);
+  assert( iOffset+l<=pCursor->nPageSize );
+  pCursor->nPayloadBytes = nPayloadBytes;
+
+  u64 iRowid;  /* Ignored */
+  l += getVarint(pCell + l, &iRowid);
+  assert( iOffset+l<=pCursor->nPageSize );
+  pCursor->iRowid = (i64)iRowid;
+
+  pCursor->iPayloadOffset = iOffset + l;
+
+  int rc = overflowMaybeCreate(pCursor->pPage, pCursor->nPageSize,
+                               iOffset + l, nPayloadBytes,
+                               &pCursor->nLocalBytes, &pCursor->pOverflow);
+  if( rc!=SQLITE_OK ){
+    return ValidateError();
+  }
+
+  /* Check that no other cell overlaps this cell. */
+  int i;
+  const unsigned iEndOffset = iOffset + l + pCursor->nLocalBytes;
+  for( i=0; i<pCursor->nCells; ++i ){
+    const unsigned iOtherOffset = decodeUnsigned16(pOffsets + i*2);
+    if( iOtherOffset>iOffset && iOtherOffset<iEndOffset ){
+      return ValidateError();
+    }
+  }
+
+  unsigned pl = 0;
+  u64 nHeaderBytes;
+  pl = getVarint(pCell + l, &nHeaderBytes);
+  assert( nHeaderBytes<=nPayloadBytes );
+  pCursor->nHeaderBytes = nHeaderBytes;
+
+  /* The header potentially could be large enough to overflow. */
+  rc = overflowGetSegment(pCursor->pPage, iOffset + l, pCursor->nLocalBytes,
+                          pCursor->pOverflow, 0, nHeaderBytes,
+                          &pCursor->pHeader, &pCursor->bFreeHeader);
+  if( rc!=SQLITE_OK ){
+    return ValidateError();
+  }
+
+  u64 nRecordBytes = 0;
+  unsigned nCols = 0;
+  while( pl<nHeaderBytes ){
+    if( !checkVarint(pCursor->pHeader + pl, nHeaderBytes - pl) ){
+      return ValidateError();
+    }
+    u64 iSerialType;
+    pl += getVarint(pCursor->pHeader + pl, &iSerialType);
+    if( iSerialType==10 || iSerialType==11 ){
+      return ValidateError();
+    }
+    nRecordBytes += SerialTypeLength(iSerialType);
+    nCols++;
+  }
+  pCursor->nCols = nCols;
+
+  /* Parsing the header used as many bytes as expected. */
+  if( pl!=nHeaderBytes ){
+    return ValidateError();
+  }
+
+  /* Calculated payload is size of expected payload. */
+  if( nHeaderBytes+nRecordBytes!=nPayloadBytes ){
+    return ValidateError();
+  }
+
+  return SQLITE_OK;
+}
+
+static i64 leafCursorCellRowid(RecoverLeafCursor *pCursor){
+  return pCursor->iRowid;
+}
+
+static unsigned leafCursorCellColumns(RecoverLeafCursor *pCursor){
+  return pCursor->nCols;
+}
+
+/* If ppBase is non-NULL, it and pbFree will be appropriately set
+ * using overflowGetSegment().  If *pbFree is set true, *ppBase must
+ * be freed using sqlite3_free().
+ *
+ * Pass NULL for ppBase to prevent retrieving the data segment.
+ */
+static int leafCursorCellColInfo(RecoverLeafCursor *pCursor,
+                                 unsigned iCol, u64 *piColType,
+                                 unsigned char **ppBase, int *pbFree){
+  /* Implicit NULL for columns past the end.  This case happens when
+   * rows have not been updated since an ALTER TABLE added columns.
+   * It is more convenient to address here than in callers.
+   */
+  if( iCol>=pCursor->nCols ){
+    *piColType = 0;
+    if( ppBase ){
+      *ppBase = 0;
+      *pbFree = 0;
+    }
+    return SQLITE_OK;
+  }
+
+  /* Must be able to decode header size. */
+  const unsigned char *pHeader = pCursor->pHeader;
+  if( !checkVarint(pHeader, pCursor->nHeaderBytes) ){
+    return SQLITE_CORRUPT;
+  }
+
+  u64 nHeaderBytes;
+  unsigned l = getVarint(pHeader, &nHeaderBytes);
+  assert( nHeaderBytes==pCursor->nHeaderBytes );
+
+  u64 nRecordBytes = 0;
+  unsigned iColsSkipped = 0;
+  u64 iSerialType;
+  if( !checkVarint(pHeader + l, nHeaderBytes - l) ){
+    return SQLITE_CORRUPT;
+  }
+  l += getVarint(pHeader + l, &iSerialType);
+  while( iColsSkipped<iCol && l<nHeaderBytes ){
+    nRecordBytes += SerialTypeLength(iSerialType);
+    if( !checkVarint(pHeader + l, nHeaderBytes - l) ){
+      return SQLITE_CORRUPT;
+    }
+    l += getVarint(pHeader + l, &iSerialType);
+    iColsSkipped++;
+  }
+
+  /* Column's data extends past payload end. */
+  const unsigned nColBytes = SerialTypeLength(iSerialType);
+  if( nHeaderBytes+nRecordBytes+nColBytes>pCursor->nPayloadBytes ){
+    return SQLITE_CORRUPT;
+  }
+
+  *piColType = iSerialType;
+  if( ppBase ){
+    return overflowGetSegment(pCursor->pPage,
+                              pCursor->iPayloadOffset + nHeaderBytes,
+                              pCursor->nLocalBytes - nHeaderBytes,
+                              pCursor->pOverflow,
+                              nRecordBytes, nColBytes,
+                              ppBase, pbFree);
+  }
+  return SQLITE_OK;
+}
+
+static int leafCursorNextValidCell(RecoverLeafCursor *pCursor){
+  while( 1 ){
+    /* Move to the next cell. */
+    pCursor->iCell++;
+
+    /* If the leaf is done, get the next leaf. */
+    if( pCursor->iCell>=pCursor->nCells ){
+      int rc = leafCursorNextPage(pCursor);
+      if( rc!=SQLITE_ROW ){
+        return rc;
+      }
+    }
+
+    /* If the cell is valid, indicate that a row is available. */
+    int rc = leafCursorCellSetup(pCursor);
+    if( rc==SQLITE_OK ){
+      return SQLITE_ROW;
+    }
+
+    /* Iterate until done or a valid row is found. */
+    fprintf(stderr, "Skipping invalid cell\n");
+  }
+  return SQLITE_ERROR;
+}
+
+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;
+  RecoverLeafCursor *pLeafCursor;
+  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;
+  }
+
+  unsigned nPageSize;
+  Pager *pPager;
+  rc = GetPager(pRecover->db, pRecover->zDb, &pPager, &nPageSize);
+  if( rc!=SQLITE_OK ){
+    return rc;
+  }
+
+  RecoverLeafCursor *pLeafCursor;
+  rc = leafCursorCreate(pPager, nPageSize, iRootPage, &pLeafCursor);
+  if( rc!=SQLITE_OK ){
+    return rc;
+  }
+
+  RecoverCursor *pCursor = sqlite3_malloc(sizeof(RecoverCursor));
+  if( !pCursor ){
+    leafCursorDestroy(pLeafCursor);
+    return SQLITE_NOMEM;
+  }
+  memset(pCursor, 0, sizeof(*pCursor));
+  pCursor->base.pVtab = pVTab;
+  pCursor->pLeafCursor = pLeafCursor;
+
+  *ppCursor = (sqlite3_vtab_cursor*)pCursor;
+  return SQLITE_OK;
+}
+
+static int recoverClose(sqlite3_vtab_cursor *cur){
+  FNENTRY();
+  RecoverCursor *pCursor = (RecoverCursor*)cur;
+  if( pCursor->pLeafCursor ){
+    leafCursorDestroy(pCursor->pLeafCursor);
+    pCursor->pLeafCursor = NULL;
+  }
+  memset(pCursor, 0xA5, sizeof(*pCursor));
+  sqlite3_free(cur);
+  return SQLITE_OK;
+}
+
+/* Helpful place to set a breakpoint. */
+static int RecoverInvalidCell(){
+  return SQLITE_ERROR;
+}
+
+static int recoverValidateLeafCell(Recover *pRecover, RecoverCursor *pCursor){
+  /* If the row's storage has too many columns, skip it. */
+  if( leafCursorCellColumns(pCursor->pLeafCursor)>pRecover->nCols ){
+    return RecoverInvalidCell();
+  }
+
+  /* Skip rows with unexpected types. */
+  unsigned i;
+  for( i=0; i<pRecover->nCols; ++i ){
+    /* ROWID alias. */
+    if( (pRecover->pTypes[i]&kMaskRowid) ){
+      continue;
+    }
+
+    u64 iType;
+    int rc = leafCursorCellColInfo(pCursor->pLeafCursor, i, &iType, NULL, NULL);
+    assert( rc==SQLITE_OK );
+    if( rc!=SQLITE_OK || !SerialTypeIsCompatible(iType, pRecover->pTypes[i]) ){
+      return RecoverInvalidCell();
+    }
+  }
+
+  return SQLITE_OK;
+}
+
+static int recoverNext(sqlite3_vtab_cursor *pVtabCursor){
+  FNENTRY();
+  RecoverCursor *pCursor = (RecoverCursor*)pVtabCursor;
+  Recover *pRecover = (Recover*)pCursor->base.pVtab;
+  int rc;
+
+  /* Scan forward to the next cell with valid storage, then check that
+   * the stored data matches the schema.
+   */
+  while( (rc = leafCursorNextValidCell(pCursor->pLeafCursor))==SQLITE_ROW ){
+    if( recoverValidateLeafCell(pRecover, pCursor)==SQLITE_OK ){
+      return SQLITE_OK;
+    }
+  }
+
+  if( rc==SQLITE_DONE ){
+    pCursor->bEOF = 1;
+    return SQLITE_OK;
+  }
+
+  assert( rc!=SQLITE_OK );
+  return rc;
+}
+
+static int recoverFilter(
+  sqlite3_vtab_cursor *pVtabCursor,
+  int idxNum, const char *idxStr,
+  int argc, sqlite3_value **argv
+){
+  FNENTRY();
+  RecoverCursor *pCursor = (RecoverCursor*)pVtabCursor;
+  Recover *pRecover = (Recover*)pCursor->base.pVtab;
+
+  /* Load the first cell, and iterate forward if it's not valid. */
+  /* TODO(shess): What happens if no cells at all are valid? */
+  int rc = leafCursorCellSetup(pCursor->pLeafCursor);
+  if( rc!=SQLITE_OK || recoverValidateLeafCell(pRecover, pCursor)!=SQLITE_OK ){
+    return recoverNext(pVtabCursor);
+  }
+
+  return SQLITE_OK;
+}
+
+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]&kMaskRowid) ){
+    sqlite3_result_int64(ctx, leafCursorCellRowid(pCursor->pLeafCursor));
+    return SQLITE_OK;
+  }
+
+  u64 iColType;
+  unsigned char *pColData = NULL;
+  int shouldFree = 0;
+  int rc = leafCursorCellColInfo(pCursor->pLeafCursor, i, &iColType,
+                                 &pColData, &shouldFree);
+  if( rc!=SQLITE_OK ){
+    return rc;
+  }
+  if( !SerialTypeIsCompatible(iColType, pRecover->pTypes[i]) ){
+    if( shouldFree ){
+      sqlite3_free(pColData);
+    }
+    return SQLITE_ERROR;
+  }
+
+  switch( iColType ){
+    case 0 : sqlite3_result_null(ctx); break;
+    case 1 : sqlite3_result_int(ctx, decodeInt8(pColData)); break;
+    case 2 : sqlite3_result_int(ctx, decodeInt16(pColData)); break;
+    case 3 : sqlite3_result_int(ctx, decodeInt24(pColData)); break;
+      /* TODO(shess): Make sure that this is valid for large values. */
+    case 4 : sqlite3_result_int(ctx, decodeInt32(pColData)); break;
+    case 5 : sqlite3_result_int64(ctx, decodeInt48(pColData)); break;
+    case 6 : sqlite3_result_int64(ctx, decodeInt64(pColData)); break;
+    case 7 : sqlite3_result_double(ctx, decodeFloat64(pColData)); break;
+    case 8 : sqlite3_result_int(ctx, 0); break;
+    case 9 : sqlite3_result_int(ctx, 1); break;
+    case 10 : assert( iColType!=10 ); break;
+    case 11 : assert( iColType!=11 ); break;
+
+    default : {
+      /* If pColData was already allocated, arrange to pass ownership. */
+      sqlite3_destructor_type pFn = SQLITE_TRANSIENT;
+      if( shouldFree ){
+        pFn = sqlite3_free;
+        shouldFree = 0;
+      }
+
+      u64 l = SerialTypeLength(iColType);
+      if( SerialTypeIsBlob(iColType) ){
+        sqlite3_result_blob(ctx, pColData, l, pFn);
+      }else{
+        sqlite3_result_text(ctx, (const char*)pColData, l, pFn);
+      }
+    } break;
+  }
+  if( shouldFree ){
+    sqlite3_free(pColData);
+  }
+  return SQLITE_OK;
+}
+
+static int recoverRowid(sqlite3_vtab_cursor *pVtabCursor, sqlite_int64 *pRowid){
+  FNENTRY();
+  RecoverCursor *pCursor = (RecoverCursor*)pVtabCursor;
+  *pRowid = leafCursorCellRowid(pCursor->pLeafCursor);
+  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 is zText, 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) ){
+    *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 = kMaskInteger | kMaskFloat | kMaskBlob | kMaskText | kMaskNull;
+    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.
+   */
+  struct {
+    const char *zName;
+    unsigned char strictMask;
+    unsigned char otherMask;
+    const char *zReplace;
+  } typeInfo[7] = {
+    { "ANY",
+      kMaskInteger | kMaskFloat | kMaskBlob | kMaskText | kMaskNull,
+      0, "",
+    },
+    { "ROWID",   kMaskInteger | kMaskRowid,             0, "INTEGER", },
+    { "INTEGER", kMaskInteger | kMaskNull,              0, NULL, },
+    { "FLOAT",   kMaskFloat | kMaskNull,                kMaskInteger, NULL, },
+    { "NUMERIC", kMaskInteger | kMaskFloat | kMaskNull, kMaskText, NULL, },
+    { "TEXT",    kMaskText | kMaskNull,                 kMaskBlob, NULL, },
+    { "BLOB",    kMaskBlob | kMaskNull,                 0, NULL, },
+  };
+
+  unsigned i, nNameLen = *pzTypeEnd - *pzTypeStart;
+  for( i=0; i<ArraySize(typeInfo); ++i ){
+    if( !strncasecmp(typeInfo[i].zName, *pzTypeStart, nNameLen) ){
+      break;
+    }
+  }
+  if( i==ArraySize(typeInfo) ){
+    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( typeInfo[i].zReplace && !typeInfo[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 = typeInfo[i].strictMask;
+  if( !bStrict ){
+    *pTypeMask |= typeInfo[i].otherMask;
+  }
+  if( bNotNull ){
+    *pTypeMask &= ~kMaskNull;
+  }
+  if( typeInfo[i].zReplace ){
+    *pzTypeStart = typeInfo[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 ParseAndGenerateCreate(int nCols, const char *const *pCols,
+                                  char **pzCreateSql, unsigned char *pTypes){
+  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;
+    unsigned char iTypeMask;
+    int rc = findNameAndType(pCols[i],
+                             &zNameStart, &zNameEnd,
+                             &zTypeStart, &zTypeEnd,
+                             &iTypeMask);
+    if( rc!=SQLITE_OK ){
+      sqlite3_free(zCreateSql);
+      return rc;
+    }
+    pTypes[i] = iTypeMask;
+
+    const char *zNotNull = "";
+    if( !(iTypeMask&kMaskNull) ){
+      zNotNull = " NOT NULL";
+    }
+
+    /* Add name and type to the create statement. */
+    zCreateSql = sqlite3_mprintf("%z%.*s %.*s%s, ",
+                                 zCreateSql,
+                                 zNameEnd - zNameStart, zNameStart,
+                                 zTypeEnd - zTypeStart, zTypeStart,
+                                 zNotNull);
+    if( !zCreateSql ){
+      return SQLITE_NOMEM;
+    }
+  }
+
+  /* Strip trailing ", ". */
+  assert( nCols>0 );
+  zCreateSql[strlen(zCreateSql)-2] = '\0';
+
+  zCreateSql = sqlite3_mprintf("%z)", zCreateSql);
+  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") ){
+    return SQLITE_MISUSE;
+  }
+
+  /* Need the backing table and at least one column. */
+  if( nCols<1 ){
+    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 *dot = strchr(argv[3], '.');
+  if( dot && dot>argv[3] ){
+    pRecover->zDb = sqlite3_strndup(argv[3], dot - argv[3]);
+    pRecover->zTable = sqlite3_strdup(dot + 1);
+  }else{
+    pRecover->zDb = sqlite3_strdup(db->aDb[0].zName);
+    pRecover->zTable = sqlite3_strdup(argv[3]);
+  }
+
+  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 ){
+    recoverRelease(pRecover);
+    return rc;
+  }
+
+  /* Parse the column definitions. */
+  char *zCreateSql;
+  rc = ParseAndGenerateCreate(nCols, argv + kTypeCol,
+                              &zCreateSql, pRecover->pTypes);
+  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;
+}
+
+/* Generate a line of hexdump-style output in b using min(16,n) bytes
+ * from pc.  b must have room for at least 57 bytes (16*2 nibbles with
+ * 3 separator spaces, 16 ASCII chars with 3 separator spaces, 2
+ * spaces between sections, and a trailing nul).
+ */
+static void hexline(char *b, const unsigned char *pc, int n){
+  const char digits[16] = "0123456789ABCDEF";
+  int i;
+  /* 2*16 for hex, 16 for ASCII, 2*3 inline spaces, 2 spaces between
+   * sections.
+   */
+  memset(b, ' ', 3*16 + 8);
+  for( i=0; i<n && i<16; i++ ){
+    b[i*2 + i/4    ] = digits[(pc[i]>>4)&0xF];
+    b[i*2 + i/4 + 1] = digits[(pc[i]>>0)&0xF];
+  }
+  for( i=0; i<n && i<16; i++ ){
+    b[2*16 + 5 + i + i/4] = (isprint(pc[i]) ? pc[i] : '.');
+  }
+  b[3*16 + 8] = '\0';
+}
+
+/* Output nData bytes from pData to stderr, hexdump-style.  Example:
+ * 0x0000:53514C69 74652066 6F726D61 74203300  SQLi te f orma t 3.
+ * 0x0010:04000101 00402020 00011362 000039C6  .... .@   ...b ..9.
+ *
+ * Consecutive lines consisting of entirely nul bytes are elided.
+ * Refuses to dump more than 4096 bytes.
+ */
+static void hexdump(const unsigned char *pData, unsigned nData){
+  unsigned iOffset = 0, bSkipped = 0;
+  assert( nData<=4096 );
+  if( nData>4096 ){
+    nData = 4096;
+  }
+  while( nData>0 ){
+    unsigned i, nCur = (nData<16 ? nData : 16);
+    for( i=0; i<nCur && !pData[iOffset+i]; i++ );
+    if( !iOffset || nCur==nData || i<16 ){
+      char buf[200];
+      hexline(buf, pData + iOffset, nCur);
+      fprintf(stderr, "0x%04x:%s\n", iOffset, buf);
+      bSkipped = 0;
+    }else if( !bSkipped ){
+      fprintf(stderr, "...\n");
+      bSkipped = 1;
+    }
+    iOffset += nCur;
+    nData -= nCur;
+  }
+}
+
+static int GetPageFromContext(sqlite3_context *ctx,
+                              const char *zName, unsigned iPage,
+                              DbPage **ppPage, unsigned *pnPageSize){
+  sqlite3 *db = (sqlite3 *)sqlite3_user_data(ctx);
+  Pager *pPager;
+  /* TODO(shess) This needs to acquire the shared lock before it can
+   * get the page.
+   */
+  int rc = GetPager(db, zName, &pPager, pnPageSize);
+  if( rc==SQLITE_OK ){
+    /* TODO(shess): This doesn't fail if the page doesn't exist. */
+    rc = sqlite3PagerAcquire(pPager, iPage, ppPage, 0);
+  }
+  return rc;
+}
+
+static void HexdumpFn(sqlite3_context *ctx, int nArg, sqlite3_value **aArg){
+  DbPage *pPage;
+  unsigned nPageSize, iPage = sqlite3_value_int(aArg[0]);
+  int rc = GetPageFromContext(ctx, "main", iPage, &pPage, &nPageSize);
+  if( rc!=SQLITE_OK ){
+    sqlite3_result_error(ctx, "Error acquiring page", rc);
+  }else{
+    const unsigned char *pData = PageData(pPage, 0);
+    fprintf(stderr, "page %u size %u:\n", iPage, nPageSize);
+    hexdump(pData, nPageSize);
+    sqlite3PagerUnref(pPage);
+    sqlite3_result_text(ctx, "OK", -1, SQLITE_STATIC);
+  }
+}
+
+static void DumpFn(sqlite3_context *ctx, int nArg, sqlite3_value **aArg){
+  /* TODO(shess): Consider options like:
+   * header - show only the page header info.
+   * cells - also show cell meta-info.
+   * records - also show summary record data.
+   * all - show complete record contents.
+   */
+  int headerOnly = 0;
+  if( nArg>1 ){
+    if( !strcasecmp((char*)sqlite3_value_text(aArg[1]), "header") ){
+      headerOnly = 1;
+    }
+  }else if( nArg==1 ){
+    headerOnly = 1;
+  }
+
+  DbPage *pPage;
+  unsigned nPageSize, iPage = sqlite3_value_int(aArg[0]);
+  int rc = GetPageFromContext(ctx, "main", iPage, &pPage, &nPageSize);
+  if( rc!=SQLITE_OK ){
+    sqlite3_result_error(ctx, "Error acquiring page", rc);
+    return;
+  }
+  fprintf(stderr, "page %u size %u:\n", iPage, nPageSize);
+
+  const unsigned char *pPageHeader = PageHeader(pPage);
+  if( *pPageHeader==0x02 || *pPageHeader==kTableInteriorPage || *pPageHeader==0x0A || *pPageHeader==kTableLeafPage ){
+    const int bTable = ((*pPageHeader)&5)==5;
+    const int bLeaf = ((*pPageHeader)&8)==8;
+    fprintf(stderr, "[0] 0x%02X - %s %s b-tree page.\n", *pPageHeader,
+            (bLeaf ? "Leaf" : "Interior"), (bTable ? "table" : "index"));
+    fprintf(stderr, "[1,2] %u - Offset to first freeblock.\n",
+            decodeUnsigned16(pPageHeader + 1));
+    fprintf(stderr, "[3,4] %u - Number of cells.\n",
+            decodeUnsigned16(pPageHeader + 3));
+    fprintf(stderr, "[5,6] %u - Offset to first byte of cell content area.\n",
+            decodeUnsigned16(pPageHeader + 5));
+    fprintf(stderr, "[7] %u - Fragmented free bytes.\n", pPageHeader[7]);
+    if( !bLeaf ){
+      fprintf(stderr, "[8-11] %u - Right-most child page.\n",
+              decodeUnsigned32(pPageHeader + 8));
+    }
+    if( !headerOnly ){
+      if( *pPageHeader==0x02 ){
+        const unsigned char *pOffsets = pPageHeader + 12;
+        unsigned iCell, nCells = decodeUnsigned16(pPageHeader + 3);
+        for( iCell=0; iCell<nCells; ++iCell ){
+          unsigned iOffset = decodeUnsigned16(pOffsets + iCell*2);
+          const unsigned char *pCell = PageData(pPage, iOffset);
+          unsigned iChildPage = decodeUnsigned32(pCell);
+          u64 nPayloadBytes;
+          getVarint(pCell + 4, &nPayloadBytes);
+          fprintf(stderr, "cell %u offset %u child %u payload %llu\n",
+                  iCell, iOffset, iChildPage, nPayloadBytes);
+          /* TODO(shess): Partially print the payload. */
+        }
+      }else if( *pPageHeader==kTableInteriorPage ){
+        const unsigned char *pOffsets = pPageHeader + 12;
+        unsigned iCell, nCells = decodeUnsigned16(pPageHeader + 3);
+        for( iCell=0; iCell<nCells; ++iCell ){
+          unsigned iOffset = decodeUnsigned16(pOffsets + iCell*2);
+          const unsigned char *pCell = PageData(pPage, iOffset);
+          unsigned iChildPage = decodeUnsigned32(pCell);
+          u64 iRowid;
+          getVarint(pCell + 4, &iRowid);
+          fprintf(stderr, "cell %u offset %u child %u rowid %lld\n",
+                  iCell, iOffset, iChildPage, iRowid);
+        }
+      }else if( *pPageHeader==kTableLeafPage || *pPageHeader==0x0A ){
+        const unsigned char *pOffsets = pPageHeader + 8;
+        unsigned iCell, nCells = decodeUnsigned16(pPageHeader + 3);
+        for( iCell=0; iCell<nCells; ++iCell ){
+          unsigned iOffset = decodeUnsigned16(pOffsets + iCell*2);
+
+          const unsigned char *pCell = PageData(pPage, iOffset);
+          int l = 0;
+
+          u64 nPayloadBytes;
+          l += getVarint(pCell + l, &nPayloadBytes);
+
+          u64 iRowid;
+          if( bTable ){
+            l += getVarint(pCell + l, &iRowid);
+          }
+
+          int pl = 0;
+          u64 nHeaderBytes;
+          pl += getVarint(pCell + l + pl, &nHeaderBytes);
+          if( bTable ){
+            fprintf(stderr,
+                    "cell %u offset %u rowid %lld payload %llu header %llu\n",
+                    iCell, iOffset, iRowid, nPayloadBytes, nHeaderBytes);
+          }else{
+            fprintf(stderr, "cell %u offset %u payload %llu header %llu\n",
+                    iCell, iOffset, nPayloadBytes, nHeaderBytes);
+          }
+
+          RecoverOverflow *pOverflow = NULL;
+          unsigned nLocalBytes = nPayloadBytes;
+          int rc = overflowMaybeCreate(pPage, nPageSize,
+                                       iOffset + l, nPayloadBytes,
+                                       &nLocalBytes, &pOverflow);
+          if( rc!=SQLITE_OK ){
+            fprintf(stderr, "SQLITE IS NOT OK\n");
+          }
+
+          /* TODO(shess): Pull this inner loop out to a helper
+           * function to improve nesting.
+           */
+          u64 nRecordBytes = 0;
+          unsigned iCol = 0;
+          while( pl<nHeaderBytes ){
+            u64 iSerialType;
+            pl += getVarint(pCell + l + pl, &iSerialType);
+
+            unsigned nBytes = 200;
+            if( nBytes>nPayloadBytes - (nHeaderBytes + nRecordBytes) ){
+              nBytes = nPayloadBytes - (nHeaderBytes + nRecordBytes);
+            }
+            unsigned char *pRecord = NULL;
+            int bFree = 0;
+            int rc = overflowGetSegment(pPage, iOffset + l + nHeaderBytes,
+                                        nLocalBytes - nHeaderBytes, pOverflow,
+                                        nRecordBytes, nBytes,
+                                        &pRecord, &bFree);
+            if( rc!=SQLITE_OK ){
+              fprintf(stderr, "SQLITE IS NOT OK.\n");
+            }
+            switch( iSerialType ){
+              case 0 :
+                fprintf(stderr, "  col[%u] NULL\n", iCol);
+                break;
+              case 1 :
+                fprintf(stderr, "  col[%u] i8 %d\n", iCol, decodeInt8(pRecord));
+                break;
+              case 2 :
+                fprintf(stderr, "  col[%u] i16 %d\n", iCol, decodeInt16(pRecord));
+                break;
+              case 3 :
+                fprintf(stderr, "  col[%u] i24 %d\n", iCol, decodeInt24(pRecord));
+                break;
+              case 4 :
+                fprintf(stderr, "  col[%u] i32 %d\n", iCol, decodeInt32(pRecord));
+                break;
+              case 5 :
+                fprintf(stderr, "  col[%u] i48 %lld\n", iCol, decodeInt48(pRecord));
+                break;
+              case 6 :
+                fprintf(stderr, "  col[%u] i64 %lld\n", iCol, decodeInt64(pRecord));
+                break;
+              case 7 :
+                fprintf(stderr, "  col[%u] FLOAT %f\n", iCol, decodeFloat64(pRecord));
+                break;
+              case 8 :
+                fprintf(stderr, "  col[%u] 0\n", iCol);
+                break;
+              case 9 :
+                fprintf(stderr, "  col[%u] 1\n", iCol);
+                break;
+              case 10 : case 11 :
+                fprintf(stderr, "  col[%u] RESERVED\n", iCol);
+                break;
+              default :
+                if( SerialTypeIsBlob(iSerialType) ){
+                  char buf[200];
+                  hexline(buf, pRecord, SerialTypeLength(iSerialType));
+                  fprintf(stderr, "  col[%u] blob[%llu] %s\n",
+                          iCol, SerialTypeLength(iSerialType), buf);
+                }else{
+                  char buf[200];
+                  u64 nBytes = SerialTypeLength(iSerialType);
+                  int i;
+                  for( i=0; i<60 && i<nBytes; ++i ){
+                    if( isprint(pRecord[i]) ){
+                      buf[i] = pRecord[i];
+                    }else{
+                      buf[i] = '.';
+                    }
+                  }
+                  if( i<nBytes ){
+                    strcpy(buf + i, "...");
+                  }else{
+                    buf[i] = '\0';
+                  }
+                  fprintf(stderr, "  col[%u] string[%llu] %s\n",
+                          iCol, SerialTypeLength(iSerialType), buf);
+                  /* TODO(shess): Support emitting the full record. */
+                }
+                break;
+            }
+            nRecordBytes += SerialTypeLength(iSerialType);
+            iCol++;
+            if( bFree ){
+              sqlite3_free(pRecord);
+              pRecord = NULL;
+              bFree = 0;
+            }
+          }
+          if( nRecordBytes+nHeaderBytes<nPayloadBytes ){
+            fprintf(stderr, "  %llu unused payload bytes\n",
+                    nPayloadBytes - nRecordBytes - nHeaderBytes);
+          }
+          if( pOverflow ){
+            overflowDestroy(pOverflow);
+          }
+        }
+      }
+    }
+  }else{
+    fprintf(stderr, "page %u size %u is unknown type:\n", iPage, nPageSize);
+    hexdump(PageData(pPage, 0), nPageSize);
+  }
+  sqlite3PagerUnref(pPage);
+  sqlite3_result_text(ctx, "OK", -1, SQLITE_STATIC);
+}
+
+int dumpFnsInit(sqlite3 *db){
+  int rc;
+  rc = sqlite3_create_function_v2(db, "hexdump", 1, SQLITE_ANY, db,
+                                  HexdumpFn, 0, 0, NULL);
+  if( rc!=SQLITE_OK ){
+    return rc;
+  }
+
+  rc = sqlite3_create_function_v2(db, "dumppage", 1, SQLITE_ANY, db,
+                                  DumpFn, 0, 0, NULL);
+  if( rc!=SQLITE_OK ){
+    return rc;
+  }
+
+  rc = sqlite3_create_function_v2(db, "dumppage", 2, SQLITE_ANY, db,
+                                  DumpFn, 0, 0, NULL);
+  if( rc!=SQLITE_OK ){
+    return rc;
+  }
+
+  return SQLITE_OK;
+}