Have .recover store all orphaned rows in a single table, with extra columns to indica...

FossilOrigin-Name: 7221f6e33ed6a5a96ec61e25f2a1f70b84aae66e503d897eb7b7ff1aec42355d

diff --git a/manifest b/manifest
index 2ea5b39962c578c5c84f262485cbb71ba18ca4c8..8404fd782bb4ede97719c4d466f7c4cc321187e9 100644 (file)
--- a/manifest
+++ b/manifest
@@ -1,5 +1,5 @@
-C Fix\sanother\sproblem\swith\sdatabase\sfreelist\shandling\sin\sthe\s".recover"\scommand.
-D 2019-04-26T15:40:27.902
+C Have\s.recover\sstore\sall\sorphaned\srows\sin\sa\ssingle\stable,\swith\sextra\scolumns\sto\sindicate\sthe\sorphaned\spage\sand\ssub-tree\sthey\swere\sdiscovered\swithin.
+D 2019-04-26T21:11:37.615
 F .fossil-settings/empty-dirs dbb81e8fc0401ac46a1491ab34a7f2c7c0452f2f06b54ebb845d024ca8283ef1
 F .fossil-settings/ignore-glob 35175cdfcf539b2318cb04a9901442804be81cd677d8b889fcc9149c21f239ea
 F LICENSE.md df5091916dbb40e6e9686186587125e1b2ff51f022cc334e886c19a0e9982724
@@ -520,7 +520,7 @@ F src/random.c 80f5d666f23feb3e6665a6ce04c7197212a88384
 F src/resolve.c 567888ee3faec14dae06519b4306201771058364a37560186a3e0e755ebc4cb8
 F src/rowset.c d977b011993aaea002cab3e0bb2ce50cf346000dff94e944d547b989f4b1fe93
 F src/select.c b7304d2f491c11a03a7fbdf34bc218282ac54052377809d4dc3b4b1e7f4bfc93
-F src/shell.c.in 44dce9f5b3c68c07e30db740aa4e635819ab6fa01a229a21356ddd4fa8f1e47e
+F src/shell.c.in 146dbc2708a314556710f8e8b13226d647fbb70778c973bb6fd655186811ce63
 F src/sqlite.h.in 38390767acc1914d58930e03149595ee4710afa4e3c43ab6c3a8aea3f1a6b8cd
 F src/sqlite3.rc 5121c9e10c3964d5755191c80dd1180c122fc3a8
 F src/sqlite3ext.h 9ecc93b8493bd20c0c07d52e2ac0ed8bab9b549c7f7955b59869597b650dd8b5
@@ -1821,7 +1821,7 @@ F vsixtest/vsixtest.tcl 6a9a6ab600c25a91a7acc6293828957a386a8a93
 F vsixtest/vsixtest.vcxproj.data 2ed517e100c66dc455b492e1a33350c1b20fbcdc
 F vsixtest/vsixtest.vcxproj.filters 37e51ffedcdb064aad6ff33b6148725226cd608e
 F vsixtest/vsixtest_TemporaryKey.pfx e5b1b036facdb453873e7084e1cae9102ccc67a0
-P afdae10424f0f3d0f10a4b73e9732aa55c5ee664814d8ca0edd372cfb17c2445
-R 5b76306d65317b1b0fee3a74fc0359d7
+P bee2652ac26370e612a8c81dd7554befc2d523442a2fbbc77dc73479e6a0d7fd
+R 52eb405413df1bbf1e7b369ddce62695
 U dan
-Z 068a6ba7f1f2eaa523c5f432cd2b8c0c
+Z 87ebd735a2a8d2115b37082d2935a00e

diff --git a/manifest.uuid b/manifest.uuid
index 4c5a22946168b81f87d6d35674bf19d23129994e..3cfd1fa139c113f6ff9e776b744862b60bfe58b2 100644 (file)
--- a/manifest.uuid
+++ b/manifest.uuid
@@ -1 +1 @@
-bee2652ac26370e612a8c81dd7554befc2d523442a2fbbc77dc73479e6a0d7fd
\ No newline at end of file
+7221f6e33ed6a5a96ec61e25f2a1f70b84aae66e503d897eb7b7ff1aec42355d
\ No newline at end of file

diff --git a/src/shell.c.in b/src/shell.c.in
index c417160d525714b314b1241e3900267cf570a5c4..0737a347a0a3d5dd2c1d69f9535151f5ab55e994 100644 (file)
--- a/src/shell.c.in
+++ b/src/shell.c.in
@@ -6226,7 +6226,7 @@ static void recoverFreeTable(RecoverTable *pTab){
     sqlite3_free(pTab->zQuoted);
     if( pTab->azlCol ){
       int i;
-      for(i=0; i<pTab->nCol; i++){
+      for(i=0; i<=pTab->nCol; i++){
         sqlite3_free(pTab->azlCol[i]);
       }
       sqlite3_free(pTab->azlCol);
@@ -6235,9 +6235,8 @@ static void recoverFreeTable(RecoverTable *pTab){
   }
 }
 
-static void recoverOldTable(
+static RecoverTable *recoverOldTable(
   int *pRc,                       /* IN/OUT: Error code */
-  RecoverTable *pTab,
   const char *zName,              /* Name of table */
   const char *zSql,               /* CREATE TABLE statement */
   int bIntkey, 
@@ -6245,12 +6244,14 @@ static void recoverOldTable(
 ){
   sqlite3 *dbtmp = 0;             /* sqlite3 handle for testing CREATE TABLE */
   int rc = *pRc;
+  RecoverTable *pTab = 0;
 
+  pTab = (RecoverTable*)shellMalloc(&rc, sizeof(RecoverTable));
   if( rc==SQLITE_OK ){
     int nSqlCol = 0;
     int bSqlIntkey = 0;
     sqlite3_stmt *pStmt = 0;
-
+    
     rc = sqlite3_open("", &dbtmp);
     if( rc==SQLITE_OK ){
       rc = sqlite3_exec(dbtmp, "PRAGMA writable_schema = on", 0, 0, 0);
@@ -6285,43 +6286,48 @@ static void recoverOldTable(
     shellFinalize(&rc, pStmt);
 
     if( bIntkey==bSqlIntkey ){
+      int i;
       const char *zPk = "_rowid_";
       sqlite3_stmt *pPkFinder = 0;
 
-      shellPreparePrintf(dbtmp, &rc, &pPkFinder, 
+      pTab->iPk = -2;
+      if( bIntkey ){
+        shellPreparePrintf(dbtmp, &rc, &pPkFinder, 
           "SELECT cid, name FROM pragma_table_info(%Q) "
           "  WHERE pk=1 AND type='integer' COLLATE nocase"
-          "  AND NOT EXISTS (SELECT cid FROM pragma_table_info(%Q) WHERE pk=2)",
-          zName, zName
-      );
-      if( rc==SQLITE_OK && SQLITE_ROW==sqlite3_step(pPkFinder) ){
-        pTab->iPk = sqlite3_column_int(pPkFinder, 0);
-        zPk = (const char*)sqlite3_column_text(pPkFinder, 1);
+          "  AND NOT EXISTS (SELECT cid FROM pragma_table_info(%Q) WHERE pk=2)"
+          , zName, zName
+        );
+        if( rc==SQLITE_OK && SQLITE_ROW==sqlite3_step(pPkFinder) ){
+          pTab->iPk = sqlite3_column_int(pPkFinder, 0);
+          zPk = (const char*)sqlite3_column_text(pPkFinder, 1);
+        }
       }
 
       pTab->zName = shellMPrintf(&rc, "%s", zName);
       pTab->zQuoted = shellMPrintf(&rc, "%Q", pTab->zName);
-      pTab->azlCol = (char**)shellMalloc(&rc, sizeof(char*) * nSqlCol);
+      pTab->azlCol = (char**)shellMalloc(&rc, sizeof(char*) * (nSqlCol+1));
       pTab->nCol = nSqlCol;
 
-      if( nSqlCol==1 && pTab->iPk==0 ){
+      if( bIntkey ){
         pTab->azlCol[0] = shellMPrintf(&rc, "%Q", zPk);
       }else{
-        shellPreparePrintf(dbtmp, &rc, &pStmt, 
-          "SELECT -1+row_number()          OVER (ORDER BY cid),"
-          "    %Q||%Q||group_concat(name, ', ') FILTER (WHERE cid!=%d) "
-          "           OVER (ORDER BY cid) "
+        pTab->azlCol[0] = shellMPrintf(&rc, "");
+      }
+      i = 1;
+      shellPreparePrintf(dbtmp, &rc, &pStmt, 
+          "SELECT %Q || group_concat(name, ', ') "
+          "  FILTER (WHERE cid!=%d) OVER (ORDER BY cid) "
           "FROM pragma_table_info(%Q)", 
-          (bIntkey ? zPk : ""), (bIntkey ? ", " : ""), 
-          pTab->iPk, zName
-        );
-        while( rc==SQLITE_OK && SQLITE_ROW==sqlite3_step(pStmt) ){
-          int idx = sqlite3_column_int(pStmt, 0);
-          const char *zText = (const char*)sqlite3_column_text(pStmt, 1);
-          pTab->azlCol[idx] = shellMPrintf(&rc, "%s", zText);
-        }
-        shellFinalize(&rc, pStmt);
+          bIntkey ? ", " : "", pTab->iPk, zName
+      );
+      while( rc==SQLITE_OK && SQLITE_ROW==sqlite3_step(pStmt) ){
+        const char *zText = (const char*)sqlite3_column_text(pStmt, 0);
+        pTab->azlCol[i] = shellMPrintf(&rc, "%s%s", pTab->azlCol[0], zText);
+        i++;
       }
+      shellFinalize(&rc, pStmt);
+
       shellFinalize(&rc, pPkFinder);
     }
   }
@@ -6329,6 +6335,11 @@ static void recoverOldTable(
  finished:
   sqlite3_close(dbtmp);
   *pRc = rc;
+  if( rc!=SQLITE_OK ){
+    recoverFreeTable(pTab);
+    pTab = 0;
+  }
+  return pTab;
 }
 
 static RecoverTable *recoverNewTable(
@@ -6336,7 +6347,8 @@ static RecoverTable *recoverNewTable(
   int *pRc,
   int iRoot,
   int bIntkey,
-  int nCol
+  int nCol,
+  int *pbNoop
 ){
   sqlite3_stmt *pStmt = 0;
   RecoverTable *pRet = 0;
@@ -6344,8 +6356,6 @@ static RecoverTable *recoverNewTable(
   const char *zSql = 0;
   const char *zName = 0;
 
-  pRet = (RecoverTable*)shellMalloc(pRc, sizeof(RecoverTable));
-  if( pRet ) pRet->iPk = -2;
 
   /* Search the recovered schema for an object with root page iRoot. */
   shellPreparePrintf(pState->db, pRc, &pStmt,
@@ -6360,55 +6370,56 @@ static RecoverTable *recoverNewTable(
     if( sqlite3_stricmp(zType, "table")==0 ){
       zName = (const char*)sqlite3_column_text(pStmt, 1);
       zSql = (const char*)sqlite3_column_text(pStmt, 2);
-      recoverOldTable(pRc, pRet, zName, zSql, bIntkey, nCol);
+      pRet = recoverOldTable(pRc, zName, zSql, bIntkey, nCol);
       break;
     }
   }
-  shellFinalize(pRc, pStmt);
-  if( bNoop ){
-    sqlite3_free(pRet);
-    return 0;
-  }
-
-  if( pRet && pRet->zName==0 ){
-    sqlite3_stmt *pStmt = 0;
 
-    pRet->zName = shellMPrintf(pRc, "orphan_%d_%d", nCol, iRoot);
-    pRet->zQuoted = shellMPrintf(pRc, "%Q", pRet->zName);
-    pRet->azlCol = (char**)shellMalloc(pRc, sizeof(char*) * nCol);
-    pRet->nCol = nCol;
+  shellFinalize(pRc, pStmt);
+  *pbNoop = bNoop;
+  return pRet;
+}
 
-    shellPreparePrintf(pState->db, pRc, &pStmt, 
-      "WITH s(i) AS ("
-      "  SELECT 1 UNION ALL SELECT i+1 FROM s WHERE i<%d"
-      ")"
-      "SELECT i-1, %Q || group_concat('c' || i, ', ') OVER (ORDER BY i) FROM s",
-      nCol, (bIntkey ? "id, " : "")
+static RecoverTable *recoverOrphanTable(
+  ShellState *pState, 
+  int *pRc, 
+  int nCol
+){
+  RecoverTable *pTab = 0;
+  if( nCol>=0 && *pRc==SQLITE_OK ){
+    int i;
+    raw_printf(pState->out, 
+        "CREATE TABLE recover_orphan(rootpgno INTEGER, "
+        "pgno INTEGER, nfield INTEGER, id INTEGER"
     );
-    while( *pRc==SQLITE_OK && SQLITE_ROW==sqlite3_step(pStmt) ){
-      int idx = sqlite3_column_int(pStmt, 0);
-      const char *zText = (const char*)sqlite3_column_text(pStmt, 1);
-      pRet->azlCol[idx] = shellMPrintf(pRc, "%s", zText);
+    for(i=0; i<nCol; i++){
+      raw_printf(pState->out, ", c%d", i);
     }
-    shellFinalize(pRc, pStmt);
+    raw_printf(pState->out, ");\n");
 
-    if( *pRc==SQLITE_OK ){
-      char *zCreate = shellMPrintf(pRc, "CREATE TABLE %Q (%s)", 
-        pRet->zName, pRet->azlCol[nCol-1]
-      );
-      if( zCreate ){
-        raw_printf(pState->out, "%s;\n", zCreate);
-        sqlite3_free(zCreate);
+    pTab = (RecoverTable*)shellMalloc(pRc, sizeof(RecoverTable));
+    if( pTab ){
+      pTab->zName = shellMPrintf(pRc, "%s", "recover_orphan");
+      pTab->zQuoted = shellMPrintf(pRc, "%Q", pTab->zName);
+      pTab->nCol = nCol;
+      pTab->iPk = -2;
+      if( nCol>0 ){
+        pTab->azlCol = (char**)shellMalloc(pRc, sizeof(char*) * (nCol+1));
+        if( pTab->azlCol ){
+          pTab->azlCol[nCol] = shellMPrintf(pRc, "");
+          for(i=nCol-1; i>=0; i--){
+            pTab->azlCol[i] = shellMPrintf(pRc, "%s, NULL", pTab->azlCol[i+1]);
+          }
+        }
       }
     }
-  }
 
-  if( *pRc!=SQLITE_OK ){
-    recoverFreeTable(pRet);
-    pRet = 0;
+    if( *pRc!=SQLITE_OK ){
+      recoverFreeTable(pTab);
+      pTab = 0;
+    }
   }
-
-  return pRet;
+  return pTab;
 }
 
 /*
@@ -6423,6 +6434,8 @@ static int recoverDatabaseCmd(ShellState *pState, int nArg, char **azArg){
   sqlite3_stmt *pCells = 0;       /* Loop through all cells in a page */
   const char *zRecoveryDb = "";   /* Name of "recovery" database */
   int i;
+  int nOrphan = -1;
+  RecoverTable *pOrphan = 0;
 
   int bFreelist = 1;              /* 0 if --freelist-corrupt is specified */
   for(i=1; i<nArg; i++){
@@ -6545,6 +6558,7 @@ static int recoverDatabaseCmd(ShellState *pState, int nArg, char **azArg){
     "  SELECT pgno FROM recovery.map WHERE root=1"
     ")"
     "GROUP BY pgno, cell;"
+    "CREATE INDEX recovery.schema_rootpage ON schema(rootpage);"
   );
 
   /* Open a transaction, then print out all non-virtual, non-"sqlite_%" 
@@ -6566,6 +6580,20 @@ static int recoverDatabaseCmd(ShellState *pState, int nArg, char **azArg){
     shellFinalize(&rc, pStmt);
   }
 
+  /* Figure out if an orphan table will be required. And if so, how many
+  ** user columns it should contain */
+  shellPrepare(pState->db, &rc, 
+      "SELECT coalesce(max(maxlen), -2) FROM recovery.map" 
+      "  WHERE root>1 AND root NOT IN (SELECT rootpage FROM recovery.schema)"
+      , &pLoop
+  );
+  if( rc==SQLITE_OK && SQLITE_ROW==sqlite3_step(pLoop) ){
+    nOrphan = sqlite3_column_int(pLoop, 0);
+  }
+  shellFinalize(&rc, pLoop);
+  pLoop = 0;
+  pOrphan = recoverOrphanTable(pState, &rc, nOrphan);
+
   shellPrepare(pState->db, &rc,
       "SELECT pgno FROM recovery.map WHERE root=?", &pPages
   );
@@ -6586,33 +6614,48 @@ static int recoverDatabaseCmd(ShellState *pState, int nArg, char **azArg){
     int iRoot = sqlite3_column_int(pLoop, 0);
     int bIntkey = sqlite3_column_int(pLoop, 1);
     int nCol = sqlite3_column_int(pLoop, 2);
+    int bNoop = 0;
     RecoverTable *pTab;
 
-    pTab = recoverNewTable(pState, &rc, iRoot, bIntkey, nCol);
-    if( pTab ){
-      if( 0==sqlite3_stricmp(pTab->zName, "sqlite_sequence") ){
-        raw_printf(pState->out, "DELETE FROM sqlite_sequence;\n");
-      }
-      sqlite3_bind_int(pPages, 1, iRoot);
-      sqlite3_bind_int(pCells, 2, pTab->iPk);
-      while( rc==SQLITE_OK && SQLITE_ROW==sqlite3_step(pPages) ){
-        sqlite3_bind_int(pCells, 1, sqlite3_column_int(pPages, 0));
-        while( rc==SQLITE_OK && SQLITE_ROW==sqlite3_step(pCells) ){
-          int iMax = sqlite3_column_int(pCells, 0);
-          const char *zVal = (const char*)sqlite3_column_text(pCells, 1);
+    pTab = recoverNewTable(pState, &rc, iRoot, bIntkey, nCol, &bNoop);
+    if( bNoop || rc ) continue;
+    if( pTab==0 ) pTab = pOrphan;
+
+    if( 0==sqlite3_stricmp(pTab->zName, "sqlite_sequence") ){
+      raw_printf(pState->out, "DELETE FROM sqlite_sequence;\n");
+    }
+    sqlite3_bind_int(pPages, 1, iRoot);
+    sqlite3_bind_int(pCells, 2, pTab->iPk);
+
+    while( rc==SQLITE_OK && SQLITE_ROW==sqlite3_step(pPages) ){
+      int iPgno = sqlite3_column_int(pPages, 0);
+      sqlite3_bind_int(pCells, 1, iPgno);
+      while( rc==SQLITE_OK && SQLITE_ROW==sqlite3_step(pCells) ){
+        int nField = sqlite3_column_int(pCells, 0);
+        const char *zVal = (const char*)sqlite3_column_text(pCells, 1);
+
+        nField = nField+1;
+        if( pTab==pOrphan ){
+          raw_printf(pState->out, 
+              "INSERT INTO %s VALUES(%d, %d, %d, %s%s%s);\n",
+              pTab->zQuoted, iRoot, iPgno, nField, 
+              bIntkey ? "" : "NULL, ", zVal, pTab->azlCol[nField]
+          );
+        }else{
           raw_printf(pState->out, "INSERT INTO %s(%s) VALUES( %s );\n", 
-              pTab->zQuoted, pTab->azlCol[iMax>0?iMax:0], zVal
+              pTab->zQuoted, pTab->azlCol[nField], zVal
           );
         }
-        shellReset(&rc, pCells);
       }
-      shellReset(&rc, pPages);
+      shellReset(&rc, pCells);
     }
-    recoverFreeTable(pTab);
+    shellReset(&rc, pPages);
+    if( pTab!=pOrphan ) recoverFreeTable(pTab);
   }
   shellFinalize(&rc, pLoop);
   shellFinalize(&rc, pPages);
   shellFinalize(&rc, pCells);
+  recoverFreeTable(pOrphan);
 
   /* The rest of the schema */
   if( rc==SQLITE_OK ){