List:Commits« Previous MessageNext Message »
From:Vladislav Vaintroub Date:February 10 2009 7:13pm
Subject:bzr commit into mysql-6.0-falcon-team branch (vvaintroub:3010)
Bug#42560
View as plain text  
#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)
      
      
      - 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/SRLVersion.h

per-file messages:
  storage/falcon/Dbb.cpp
    move logging of create index somewhere to createRootPage
  storage/falcon/IndexPage.cpp
    remove misleadingly named IndexPage::createNewLevel(). 
    If is only used when creating the root page, move 
    corresponding code to new initRootPage() function.
  storage/falcon/IndexPage.h
     - remove createNewLevel()
     - add initRootPage()
  storage/falcon/IndexRootPage.cpp
    - refactor createIndex and redoCreateIndex - use function createIndexRoot in both cases
    which takes care for page allocation and logging.
    
    - simplify redoIndexPage - no need to carefully read/modify left and right neighbors-
    those pages are always logged separately on split, and this should guarantee index
    consistency
  storage/falcon/IndexRootPage.h
    createIndexRoot and redoCreateIndex take additional parameters
  storage/falcon/SRLCreateIndex.cpp
    Log index root page number during "create index".
  storage/falcon/SRLCreateIndex.h
    Log index root page number during "create index".
  storage/falcon/SRLVersion.h
    Increment serial log version due to changed logging of SRLCreateIndex
=== 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 19:12:48 +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 19:12:48 +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 19:12:48 +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 19:12:48 +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] = 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 19:12:48 +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 19:12:48 +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 19:12:48 +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/SRLVersion.h'
--- a/storage/falcon/SRLVersion.h	2009-01-26 18:13:45 +0000
+++ b/storage/falcon/SRLVersion.h	2009-02-10 19:12:48 +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#42560Vladislav Vaintroub10 Feb