#At file:///G:/bzr/mysql-6.0-falcon-team/ based on revid:hky@stripped
3010 Vladislav Vaintroub 2009-02-10
Bug #42560 Falcon recovery error: wrong page type,
expected 5 got <something>
Problem:
during index creation root page number is not logged.
The page is not recreated on recovery which may result
in corrupted index.
Fix:
- Log index root page number and recreate index root
page in recovery of SRLCreateIndex. Do it in phase 2
of recovery, instead of "redo" phase.
(phase2 is responsible for recreating "physical" objects
, e.g pages)
- Also, SRLDeleteIndex will be executed in phase 2 of recovery
- Log index root page number when creating index and
recreate this page on recovery. Refactor code so
createIndexRoot can be called inside and outside recovery.
The code is more compact and better readable.
- Simplify redoIndexPage.
Do not care about left and right neighbour pages.
they are logged separately during splits, and there is no
need for careful trialReads.
- Cleanup:
remove misleadingly named IndexPage::createNewLevel(),
substitute with initRootPage().
remove setIndexRoot(), its single usage was within
redoDeleteIndex()
modified:
storage/falcon/Dbb.cpp
storage/falcon/IndexPage.cpp
storage/falcon/IndexPage.h
storage/falcon/IndexRootPage.cpp
storage/falcon/IndexRootPage.h
storage/falcon/SRLCreateIndex.cpp
storage/falcon/SRLCreateIndex.h
storage/falcon/SRLDeleteIndex.cpp
storage/falcon/SRLDeleteIndex.h
storage/falcon/SRLVersion.h
=== modified file 'storage/falcon/Dbb.cpp'
--- a/storage/falcon/Dbb.cpp 2009-02-09 05:01:44 +0000
+++ b/storage/falcon/Dbb.cpp 2009-02-10 22:50:04 +0000
@@ -405,7 +405,6 @@ int32 Dbb::findNextRecord(Section *secti
int32 Dbb::createIndex(TransId transId, int indexVersion)
{
int indexId;
-
switch (indexVersion)
{
case INDEX_VERSION_0:
@@ -420,9 +419,6 @@ int32 Dbb::createIndex(TransId transId,
ASSERT(false);
}
- if (serialLog)
- serialLog->logControl->createIndex.append(this, transId, indexId, indexVersion);
-
return indexId;
}
=== modified file 'storage/falcon/IndexPage.cpp'
--- a/storage/falcon/IndexPage.cpp 2008-07-09 12:35:48 +0000
+++ b/storage/falcon/IndexPage.cpp 2009-02-10 22:50:04 +0000
@@ -618,35 +618,6 @@ void IndexPage::printPage(Bdb * bdb, boo
printPage ((IndexPage*) bdb->buffer, bdb->pageNumber, inversion);
}
-Bdb* IndexPage::createNewLevel(Dbb* dbb, int level, int version, TransId transId)
-{
- Bdb *bdb = dbb->allocPage (PAGE_btree, transId);
- BDB_HISTORY(bdb);
- IndexPage *page = (IndexPage*) bdb->buffer;
- page->level = level;
- page->version = version;
- IndexKey dummy;
- page->length = OFFSET (IndexPage*, nodes);
- //page->addNode (dbb, &dummy, END_LEVEL);
- IndexNode node;
- node.insert(page->nodes, 0, 0, dummy.key, END_LEVEL);
- page->length += IndexNode::nodeLength(0, 0, END_LEVEL);
-
- return bdb;
-}
-
-Bdb* IndexPage::createNewLevel(Dbb * dbb, int level, int version, int32 page1, int32 page2, IndexKey* key2, TransId transId)
-{
- Bdb *parentBdb = createNewLevel(dbb, level, version, transId);
- BDB_HISTORY(parentBdb);
- IndexPage *parentPage = (IndexPage*) parentBdb->buffer;
- IndexKey dummy(key2->index);
- parentPage->addNode (dbb, &dummy, page1);
- parentPage->addNode (dbb, key2, page2);
- //parentPage->validate(NULL);
-
- return parentBdb;
-}
Bdb* IndexPage::findLevel(Dbb * dbb, int32 indexId, Bdb *bdb, int level, IndexKey *indexKey, int32 recordNumber)
{
@@ -1705,3 +1676,16 @@ void IndexPage::backup(EncodedDataStream
void IndexPage::restore(EncodedDataStream* stream)
{
}
+
+void IndexPage::initRootPage(Bdb *bdb)
+{
+ // Initialize index root page
+ IndexPage *page = (IndexPage*) bdb->buffer;
+ page->level = 0;
+ page->version = INDEX_CURRENT_VERSION;
+ IndexKey dummy;
+ page->length = OFFSET (IndexPage*, nodes);
+ IndexNode node;
+ node.insert(page->nodes, 0, 0, dummy.key, END_LEVEL);
+ page->length += IndexNode::nodeLength(0, 0, END_LEVEL);
+}
\ No newline at end of file
=== modified file 'storage/falcon/IndexPage.h'
--- a/storage/falcon/IndexPage.h 2008-06-17 17:41:54 +0000
+++ b/storage/falcon/IndexPage.h 2009-02-10 22:50:04 +0000
@@ -64,14 +64,13 @@ public:
Btn* findSupernode(int level, UCHAR *key, size_t len, int32 recordNumber, Btn *after, bool *match);
Btn* findPriorNodeForSupernode(Btn *where,IndexKey *priorKey);
Btn* getEnd(void);
+ static void initRootPage(Bdb *bdb);
void backup(EncodedDataStream* stream);
void restore(EncodedDataStream* stream);
static int computePrefix (IndexKey *key1, IndexKey *key2);
static Bdb* findLevel (Dbb *dbb, int32 indexId, Bdb *bdb, int level, IndexKey *indexKey, int32 recordNumber);
- static Bdb* createNewLevel (Dbb* dbb, int level, int version, TransId transId);
- static Bdb* createNewLevel (Dbb *dbb, int level, int version, int32 page1, int32 page2, IndexKey *key2, TransId transId);
static void printPage (Bdb *bdb, bool inversion);
static void printPage (IndexPage *page, int32 pageNumber, bool inversion);
static void printPage (IndexPage *page, int32 pageNum, bool printDetail, bool inversion);
=== modified file 'storage/falcon/IndexRootPage.cpp'
--- a/storage/falcon/IndexRootPage.cpp 2008-11-20 17:05:50 +0000
+++ b/storage/falcon/IndexRootPage.cpp 2009-02-10 22:50:04 +0000
@@ -96,15 +96,9 @@ int32 IndexRootPage::createIndex(Dbb * d
}
else
{
- sectionsBdb->mark(transId);
- Bdb *indexBdb = createIndexRoot(dbb, transId);
- BDB_HISTORY(indexBdb);
- sections->pages [slot] = indexBdb->pageNumber;
- IndexPage::logIndexPage(indexBdb, transId);
- indexBdb->release(REL_HISTORY);
+ createIndexRoot(dbb, transId,0, id, sectionsBdb);
sectionsBdb->release(REL_HISTORY);
dbb->nextIndex = (skipped) ? skipped : id + 1;
-
return id;
}
}
@@ -787,22 +781,10 @@ void IndexRootPage::debugBucket(Dbb *dbb
void IndexRootPage::redoIndexPage(Dbb* dbb, int32 pageNumber, int32 parentPage, int level, int32 priorPage, int32 nextPage, int length, const UCHAR *data, bool haveSuperNodes)
{
//Log::debug("redoIndexPage %d -> %d -> %d level %d, parent %d)\n", priorPage, pageNumber, nextPage, level, parentPage);
- Bdb *bdb = dbb->fakePage(pageNumber, PAGE_any, 0);
+ Bdb *bdb = dbb->fakePage(pageNumber, PAGE_btree, NO_TRANSACTION);
BDB_HISTORY(bdb);
- IndexPage *indexPage = (IndexPage*) bdb->buffer;
-
- // Try to read page. If it looks OK, keep it. Otherwise, rebuild it
-
- if (!dbb->trialRead(bdb) ||
- indexPage->pageType != PAGE_btree ||
- indexPage->parentPage != parentPage ||
- indexPage->level == level)
- {
- memset(indexPage, 0, dbb->pageSize);
- //indexPage->pageType = PAGE_btree;
- bdb->setPageHeader(PAGE_btree);
- }
+ IndexPage *indexPage = (IndexPage*) bdb->buffer;
indexPage->level = level;
indexPage->parentPage = parentPage;
indexPage->nextPage = nextPage;
@@ -820,7 +802,7 @@ void IndexRootPage::redoIndexPage(Dbb* d
memset(indexPage->superNodes, 0, sizeof(indexPage->superNodes));
}
- // If we have a parent page, propogate the first node upward
+ // If we have a parent page, propagate the first node upward
if (parentPage && indexPage->priorPage != 0)
{
@@ -852,77 +834,21 @@ void IndexRootPage::redoIndexPage(Dbb* d
}
bdb->release(REL_HISTORY);
-
- if (nextPage)
- {
- bdb = dbb->trialFetch(nextPage, PAGE_any, Exclusive);
- BDB_HISTORY(bdb);
-
- if (bdb)
- {
- indexPage = (IndexPage*) bdb->buffer;
-
- if (indexPage->pageType == PAGE_btree &&
- indexPage->parentPage == parentPage &&
- indexPage->level == level)
- {
- if (indexPage->priorPage != pageNumber)
- {
- bdb->mark(NO_TRANSACTION);
- indexPage->priorPage = pageNumber;
- }
- }
-
- bdb->release(REL_HISTORY);
- }
- }
-
- if (priorPage)
- {
- bdb = dbb->trialFetch(priorPage, PAGE_any, Exclusive);
- BDB_HISTORY(bdb);
-
- if (bdb)
- {
- indexPage = (IndexPage*) bdb->buffer;
-
- if (indexPage->pageType == PAGE_btree &&
- indexPage->parentPage == parentPage &&
- indexPage->level == level)
- {
- if (indexPage->nextPage != pageNumber)
- {
- bdb->mark(NO_TRANSACTION);
- indexPage->nextPage = pageNumber;
- }
- }
-
- bdb->release(REL_HISTORY);
- }
- }
}
-void IndexRootPage::setIndexRoot(Dbb* dbb, int indexId, int32 pageNumber, TransId transId)
+void IndexRootPage::redoIndexDelete(Dbb* dbb, int indexId)
{
+ ASSERT(dbb->serialLog->recovering);
int sequence = indexId / dbb->pagesPerSection;
int slot = indexId % dbb->pagesPerSection;
- Bdb *bdb = Section::getSectionPage (dbb, INDEX_ROOT, sequence, Exclusive, transId);
+ Bdb *bdb = Section::getSectionPage (dbb, INDEX_ROOT, sequence, Exclusive, NO_TRANSACTION);
BDB_HISTORY(bdb);
- bdb->mark(transId);
+ bdb->mark(NO_TRANSACTION);
SectionPage *sections = (SectionPage*) bdb->buffer;
- sections->pages[slot] = pageNumber;
-
- if (!dbb->serialLog->recovering && !dbb->noLog)
- dbb->serialLog->logControl->sectionPage.append(dbb, transId, bdb->pageNumber, pageNumber, slot, INDEX_ROOT, sequence, 0);
-
+ sections->pages[slot] =0;
bdb->release(REL_HISTORY);
}
-void IndexRootPage::redoIndexDelete(Dbb* dbb, int indexId)
-{
- setIndexRoot(dbb, indexId, 0, NO_TRANSACTION);
-}
-
void IndexRootPage::indexMerge(Dbb *dbb, int indexId, SRLUpdateIndex *logRecord, TransId transId)
{
IndexKey key;
@@ -1050,30 +976,39 @@ void IndexRootPage::indexMerge(Dbb *dbb,
//Log::debug("indexMerge: index %d, %d insertions, %d punts, %d duplicates, %d rollovers\n", indexId, insertions, punts, duplicates, rollovers);
}
-void IndexRootPage::redoCreateIndex(Dbb* dbb, int indexId)
+void IndexRootPage::redoCreateIndex(Dbb* dbb, int indexId, int pageNumber)
{
Bdb *bdb = Section::getSectionPage (dbb, INDEX_ROOT, indexId / dbb->pagesPerSection, Exclusive, NO_TRANSACTION);
BDB_HISTORY(bdb);
ASSERT(bdb);
- SectionPage *sections = (SectionPage*) bdb->buffer;
- int slot = indexId % dbb->pagesPerSection;
-
- if (!sections->pages [slot])
- {
- bdb->mark(NO_TRANSACTION);
- Bdb *pageBdb = createIndexRoot(dbb, NO_TRANSACTION);
- sections->pages [slot] = pageBdb->pageNumber;
- pageBdb->release(REL_HISTORY);
- }
+ createIndexRoot(dbb, NO_TRANSACTION, pageNumber, indexId ,bdb);
bdb->release(REL_HISTORY);
}
-Bdb* IndexRootPage::createIndexRoot(Dbb* dbb, TransId transId)
+void IndexRootPage::createIndexRoot(Dbb* dbb, TransId transId, int pageNumber, int id, Bdb *sectionsBdb)
{
- Bdb *bdb = IndexPage::createNewLevel(dbb, 0, INDEX_CURRENT_VERSION, transId);
+ Bdb *bdb;
- return bdb;
+ if (dbb->serialLog->recovering)
+ // Use given page number.
+ bdb = dbb->fakePage (pageNumber, PAGE_btree, transId);
+ else
+ {
+ // This is not recovery , allocate a new page and log it.
+ bdb = dbb->allocPage(PAGE_btree, transId);
+ dbb->serialLog->logControl->createIndex.append(dbb, transId, id, INDEX_CURRENT_VERSION,
+ bdb->pageNumber);
+ }
+ BDB_HISTORY(bdb);
+
+ IndexPage::initRootPage(bdb);
+
+ // link index root to sections
+ SectionPage *sections = (SectionPage *)sectionsBdb->buffer;
+ sections->pages [id%dbb->pagesPerSection] = bdb->pageNumber;
+ sectionsBdb->mark(transId);
+ bdb->release(REL_HISTORY);
}
void IndexRootPage::analyzeIndex(Dbb* dbb, int indexId, IndexAnalysis *indexAnalysis)
=== modified file 'storage/falcon/IndexRootPage.h'
--- a/storage/falcon/IndexRootPage.h 2008-05-06 15:52:21 +0000
+++ b/storage/falcon/IndexRootPage.h 2009-02-10 22:50:04 +0000
@@ -55,14 +55,13 @@ public:
static int32 createIndex (Dbb *dbb, TransId transId);
static void create (Dbb *dbb, TransId transId);
static void indexMerge(Dbb *dbb, int indexId, SRLUpdateIndex *indexNodes, TransId transId);
- static Bdb* createIndexRoot(Dbb* dbb, TransId transId);
+ static void createIndexRoot(Dbb* dbb, TransId transId, int pageNumber, int indexId, Bdb *sectionPage);
static void analyzeIndex(Dbb* dbb, int indexId, IndexAnalysis *indexAnalysis);
static int32 getIndexRoot(Dbb* dbb, int indexId);
static void redoIndexPage(Dbb* dbb, int32 pageNumber, int32 parentPageNumber, int level, int32 prior, int32 next, int length, const UCHAR *data, bool haveSuperNodes);
- static void setIndexRoot(Dbb* dbb, int indexId, int32 pageNumber, TransId transId);
static void redoIndexDelete(Dbb* dbb, int indexId);
- static void redoCreateIndex(Dbb* dbb, int indexId);
+ static void redoCreateIndex(Dbb* dbb, int indexId, int pageNumber);
};
#endif // !defined(AFX_INDEXROOTPAGE_H__6A019C27_A340_11D2_AB5A_0000C01D2301__INCLUDED_)
=== modified file 'storage/falcon/SRLCreateIndex.cpp'
--- a/storage/falcon/SRLCreateIndex.cpp 2007-12-04 21:16:33 +0000
+++ b/storage/falcon/SRLCreateIndex.cpp 2009-02-10 22:50:04 +0000
@@ -39,7 +39,7 @@ SRLCreateIndex::~SRLCreateIndex()
}
-void SRLCreateIndex::append(Dbb *dbb, TransId transId, int32 id, int idxVersion)
+void SRLCreateIndex::append(Dbb *dbb, TransId transId, int32 id, int idxVersion, int pageNumber)
{
START_RECORD(srlCreateIndex, "SRLCreateIndex::append");
log->getTransaction(transId);
@@ -48,6 +48,7 @@ void SRLCreateIndex::append(Dbb *dbb, Tr
putInt(id);
putInt(idxVersion);
putInt(transId);
+ putInt(pageNumber);
sync.unlock();
}
@@ -61,22 +62,31 @@ void SRLCreateIndex::read()
indexId = getInt();
indexVersion = getInt();
transactionId = getInt();
+ if (control->version >= srlVersion17)
+ pageNumber = getInt();
+ else
+ pageNumber = 0; // we'll probably suck in recovery
}
void SRLCreateIndex::pass1()
{
+ log->bumpPageIncarnation(pageNumber,tableSpaceId, objInUse);
log->bumpIndexIncarnation(indexId, tableSpaceId, objInUse);
}
-void SRLCreateIndex::redo()
+void SRLCreateIndex::pass2()
{
+ log->bumpPageIncarnation(pageNumber,tableSpaceId, objInUse);
if (!log->bumpIndexIncarnation(indexId, tableSpaceId, objInUse))
return;
+ if (!control->isPostFlush())
+ return;
+
switch (indexVersion)
{
case INDEX_VERSION_1:
- IndexRootPage::redoCreateIndex(log->getDbb(tableSpaceId), indexId);
+ IndexRootPage::redoCreateIndex(log->getDbb(tableSpaceId), indexId, pageNumber);
break;
case INDEX_VERSION_0:
@@ -88,6 +98,12 @@ void SRLCreateIndex::redo()
}
}
+void SRLCreateIndex::redo()
+{
+ log->bumpPageIncarnation(pageNumber,tableSpaceId, objInUse);
+ log->bumpIndexIncarnation(indexId, tableSpaceId, objInUse);
+}
+
void SRLCreateIndex::print()
{
logPrint("Create Index %d\n", indexId);
=== modified file 'storage/falcon/SRLCreateIndex.h'
--- a/storage/falcon/SRLCreateIndex.h 2008-11-14 02:30:11 +0000
+++ b/storage/falcon/SRLCreateIndex.h 2009-02-10 22:50:04 +0000
@@ -31,8 +31,9 @@ public:
SRLCreateIndex();
virtual ~SRLCreateIndex();
- void append (Dbb *dbb, TransId transId, int32 indexId, int indexVersion);
+ void append (Dbb *dbb, TransId transId, int32 indexId, int indexVersion, int pageNumber);
virtual void pass1();
+ virtual void pass2();
virtual void print();
virtual void redo();
virtual void read();
@@ -40,6 +41,7 @@ public:
int32 indexId;
int indexVersion;
+ int32 pageNumber;
};
#endif // !defined(AFX_SRLCREATEINDEX_H__4044BC3E_E422_4CB3_B61F_316DD6ED5B96__INCLUDED_)
=== modified file 'storage/falcon/SRLDeleteIndex.cpp'
--- a/storage/falcon/SRLDeleteIndex.cpp 2008-11-14 02:30:11 +0000
+++ b/storage/falcon/SRLDeleteIndex.cpp 2009-02-10 22:50:04 +0000
@@ -74,7 +74,7 @@ void SRLDeleteIndex::pass1()
log->bumpIndexIncarnation(indexId, tableSpaceId, objDeleted);
}
-void SRLDeleteIndex::redo()
+void SRLDeleteIndex::pass2()
{
if (!log->bumpIndexIncarnation(indexId, tableSpaceId, objDeleted))
return;
@@ -82,7 +82,9 @@ void SRLDeleteIndex::redo()
Dbb *dbb = log->findDbb(tableSpaceId);
if (!dbb)
return;
-
+ if (!control->isPostFlush())
+ return;
+
switch (indexVersion)
{
case INDEX_VERSION_0:
@@ -98,6 +100,12 @@ void SRLDeleteIndex::redo()
}
}
+void SRLDeleteIndex::redo()
+{
+ log->bumpIndexIncarnation(indexId, tableSpaceId, objDeleted);
+}
+
+
void SRLDeleteIndex::print()
{
logPrint("Delete Index %d\n", indexId);
=== modified file 'storage/falcon/SRLDeleteIndex.h'
--- a/storage/falcon/SRLDeleteIndex.h 2008-11-14 02:30:11 +0000
+++ b/storage/falcon/SRLDeleteIndex.h 2009-02-10 22:50:04 +0000
@@ -33,6 +33,7 @@ public:
virtual ~SRLDeleteIndex();
virtual void redo();
virtual void pass1();
+ virtual void pass2();
virtual void read();
virtual void commit(void);
void append(Dbb *dbb, TransId transId, int indexId, int idxVersion);
=== modified file 'storage/falcon/SRLVersion.h'
--- a/storage/falcon/SRLVersion.h 2009-01-26 18:13:45 +0000
+++ b/storage/falcon/SRLVersion.h 2009-02-10 22:50:04 +0000
@@ -43,7 +43,8 @@ static const int srlVersion13 = 13; //
static const int srlVersion14 = 14; // Added supernodes logging March 7, 2008
static const int srlVersion15 = 15; // Added tablespace parameters to SRLCreateTableSpace March 27, 2008
static const int srlVersion16 = 16; // Added SRLInventoryPage January 26, 2009
-static const int srlCurrentVersion = srlVersion16;
+static const int srlVersion17 = 17; // Log root page number in SRLCreateIndex
+static const int srlCurrentVersion = srlVersion17;
class SRLVersion : public SerialLogRecord
{
| Thread |
|---|
| • bzr commit into mysql-6.0-falcon-team branch (vvaintroub:3010)Bug#42560 | Vladislav Vaintroub | 10 Feb |
