List:Commits« Previous MessageNext Message »
From:Vladislav Vaintroub Date:March 2 2009 6:36pm
Subject:bzr commit into mysql-6.0-falcon-team branch (vvaintroub:3043)
Bug#42344 Bug#42903
View as plain text  
#At file:///G:/bzr/falcon_IndexPage_cleanup/ based on revid:vvaintroub@stripped

 3043 Vladislav Vaintroub	2009-03-02
      Bug#42344  Falcon recovery Exception: read error on page 
      Bug#42903  Falcon: Inconsistent tablespace after recovery
      
      The exception is caused during recovery of an index page, when 
      trying to  insert a first record to the parent page of the current 
      page. Parent page was not existing - this caused reads behind the 
      EOF of falcon_user.fts 
      
      The problem is that parent page itself was not logged.
      Another (worse) problem that can result into corrupted index is that 
      that parent pointer is not reliable and often can be wrong for 
      non-leaf pages. When non-leaf page is split, half or its nodes are 
      transferred to the newly created page. For  consistency, parentPage
      of corresponding child pages had to be changed and change had to be 
      logged and this was not done.
      
      The fix is to remove all occurenced parentPage pointer from the 
      Falcon code. Having a link to parent page was an idea that is not 
      implemented consistently. Also, this patch removed all occurenced 
      to priorPage from the Falcon code.This was another inconsistently 
      implemented idea.
      
      IndexPage::priorPage and IndexPage::parentPage are renamed to  
      int32 IndexPage::unused[2], for future use and to prevent the need
      to upgrade for existing databases.
      
      Recovery is changed to perform careful writes in the right order 
      during the index split (newly created page, then page that overflowed, 
      then parent page or the newly creted page).
      
      Doing recovery like this could result in a slightly detuned index, if 
      crash or kill occures before parent page is logged. Detuned means
      there is a page without a parent. Index traversal will still find it
      when traversing the page level via IndexPage::next links. 
      
      "Detuned" problem can be fixed  by logging  all pages involved in index 
      split at once. This is something to consider in the future.

    modified:
      storage/falcon/Dbb.cpp
      storage/falcon/Debug.cpp
      storage/falcon/Index2Page.cpp
      storage/falcon/IndexPage.cpp
      storage/falcon/IndexPage.h
      storage/falcon/IndexRootPage.cpp
      storage/falcon/IndexRootPage.h
      storage/falcon/PageInventoryPage.cpp
      storage/falcon/SRLIndexPage.cpp
      storage/falcon/SRLIndexPage.h
      storage/falcon/SRLVersion.h
=== modified file 'storage/falcon/Dbb.cpp'
--- a/storage/falcon/Dbb.cpp	2009-02-28 16:41:07 +0000
+++ b/storage/falcon/Dbb.cpp	2009-03-02 18:36:32 +0000
@@ -1091,8 +1091,8 @@ void Dbb::printPage(Bdb* bdb)
 			//IndexPage::printPage (bdb, false);
 			{
 			IndexPage *indexPage = (IndexPage*) page;
-			Log::debug ("Page %d is index page, parent %d, prior %d, next %d, lvl %d\n", 
-						 pageNumber, indexPage->parentPage, indexPage->priorPage, indexPage->nextPage, indexPage->level);
+			Log::debug ("Page %d is index page, next %d, lvl %d\n", 
+						 pageNumber, indexPage->nextPage, indexPage->level);
 			}
 			break;
 

=== modified file 'storage/falcon/Debug.cpp'
--- a/storage/falcon/Debug.cpp	2009-02-28 16:41:07 +0000
+++ b/storage/falcon/Debug.cpp	2009-03-02 18:36:32 +0000
@@ -210,15 +210,6 @@ void Debug::execute(IndexPage *indexPage
 		case Right:
 			fetch(indexPage->nextPage);
 			break;
-
-		case Left:
-			fetch(indexPage->priorPage);
-			break;
-
-		case Up:
-			fetch(indexPage->parentPage);
-			break;
-
 		default:
 			break;
 		}

=== modified file 'storage/falcon/Index2Page.cpp'
--- a/storage/falcon/Index2Page.cpp	2009-02-28 16:41:07 +0000
+++ b/storage/falcon/Index2Page.cpp	2009-03-02 18:36:32 +0000
@@ -996,8 +996,7 @@ void Index2Page::logIndexPage(Bdb *bdb, 
 	ASSERT(bdb->useCount > 0);
 	Index2Page *indexPage = (Index2Page*) bdb->buffer;
 	dbb->serialLog->logControl->indexPage.append(dbb, transId, INDEX_VERSION_0, bdb->pageNumber, indexPage->level, 
-												 indexPage->parentPage, indexPage->priorPage, indexPage->nextPage,  
-												 indexPage->length - OFFSET (Index2Page*, nodes), 
+												 indexPage->nextPage,  indexPage->length - OFFSET (Index2Page*, nodes), 
 												 (const UCHAR*) indexPage->nodes);
 }
 

=== modified file 'storage/falcon/IndexPage.cpp'
--- a/storage/falcon/IndexPage.cpp	2009-02-28 16:41:07 +0000
+++ b/storage/falcon/IndexPage.cpp	2009-03-02 18:36:32 +0000
@@ -424,8 +424,6 @@ Bdb* IndexPage::splitIndexPageMiddle(Dbb
 		printPage (splitBdb, false);
 		}
 
-	logIndexPage(splitBdb, transId);
-
 	return splitBdb;
 }
 
@@ -654,8 +652,7 @@ Bdb* IndexPage::findLevel(Dbb * dbb, int
 
 		if (dbb->debug & (DEBUG_PAGES | DEBUG_FIND_LEVEL))
 			page->printPage (bdb, false);
-		
-		page->parentPage = parentPageNumber;
+
 		}
 
 	return bdb;
@@ -770,13 +767,11 @@ void IndexPage::printPage(IndexPage * pa
 	UCHAR	key [MAX_PHYSICAL_KEY_LENGTH];
 	int		length;
 
-	Log::debug ("Btree Page %d, level %d, length %d, prior %d, next %d, parent %d\n", 
+	Log::debug ("Btree Page %d, level %d, length %d, next %d\n", 
 			pageNumber,
 			page->level,
 			page->length,
-			page->priorPage,
-			page->nextPage,
-			page->parentPage);
+			page->nextPage);
 
 	Btn *end = (Btn*) ((UCHAR*) page + page->length);
 	IndexNode node (page);
@@ -818,13 +813,11 @@ void IndexPage::printPage(IndexPage *pag
 	int32	pageNumber = pageNum;
 #endif	
 
-	Log::debug ("Page: %d  level: %d  length: %d  prior: %d  next: %d  parent: %d BEGIN\n", 
+	Log::debug ("Page: %d  level: %d  length: %d next: %d BEGIN\n", 
 			pageNumber,
 			page->level,
 			page->length,
-			page->priorPage,
-			page->nextPage,
-			page->parentPage);
+			page->nextPage);
 
 	Btn *end = (Btn*) ((UCHAR*) page + page->length);
 	IndexNode node (page);
@@ -851,13 +844,10 @@ void IndexPage::printPage(IndexPage *pag
 		}
 
 //	if (printDetail)
-		Log::debug ("Page: %d  level: %d  length: %d  prior: %d  next: %d  parent: %d END\n", 
+		Log::debug ("Page: %d  level: %d  length: %d  next: %d  END\n", 
 				pageNumber,
 				page->level,
-				page->length,
-				page->priorPage,
-				page->nextPage,
-				page->parentPage);
+				page->length);
 
 	int length = (int)((UCHAR*) node.node - (UCHAR*) page);
 
@@ -907,13 +897,11 @@ void IndexPage::printNode(IndexPage *pag
 		
 	{
 	//LogLock logLock;
-	Log::debug("Page: %d  level: %d  length: %d  prior: %d  next: %d  parent: %d\n", 
+	Log::debug("Page: %d  level: %d  length: %d  next: %d  parent: %d\n", 
 			pageNumber,
 			page->level,
 			page->length,
-			page->priorPage,
-			page->nextPage,
-			page->parentPage);
+			page->nextPage);
 		
 	printNode(i, page, pageNumber, tmp, inversion);
 	}
@@ -1024,12 +1012,6 @@ void IndexPage::validateNodes(Dbb *dbb, 
 				}
 			else if (validation->isPageType (bdb, PAGE_btree, "IndexPage"))
 				{
-				IndexPage *indexPage = (IndexPage*) bdb->buffer;
-				
-				if (indexPage->parentPage == parentPageNumber)
-					indexPage->validate (dbb, validation, children, bdb->pageNumber);
-				else
-					validation->error ("lost index child page %d, index id %d has wrong parent", pageNumber, validation->indexId);
 				}
 				
 			bdb->release(REL_HISTORY);
@@ -1163,8 +1145,6 @@ Bdb* IndexPage::splitIndexPageEnd(Dbb *d
 		printPage (splitBdb, false);
 		}
 
-	logIndexPage(splitBdb, transId);
-
 	return splitBdb;
 }
 
@@ -1197,21 +1177,8 @@ Bdb* IndexPage::splitPage(Dbb *dbb, Bdb 
 
 	split->level = level;
 	split->version = version;
-	split->priorPage = bdb->pageNumber;
-	split->parentPage = parentPage;
 	split->length = OFFSET(IndexPage*, nodes);
-	
-	// Link page into right sibling
-
-	if ((split->nextPage = nextPage))
-		{
-		Bdb *nextBdb = dbb->fetchPage (split->nextPage, PAGE_btree, Exclusive);
-		BDB_HISTORY(bdb);
-		IndexPage *next = (IndexPage*) nextBdb->buffer;
-		nextBdb->mark(transId);
-		next->priorPage = splitBdb->pageNumber;
-		nextBdb->release(REL_HISTORY);
-		}
+	split->nextPage = nextPage;
 
 	nextPage = splitBdb->pageNumber;
 
@@ -1247,7 +1214,7 @@ void IndexPage::logIndexPage(Bdb *bdb, T
 	***/
 	
 	dbb->serialLog->logControl->indexPage.append(dbb, transId, INDEX_VERSION_1, bdb->pageNumber, indexPage->level, 
-												 indexPage->parentPage, indexPage->priorPage, indexPage->nextPage,  
+												 indexPage->nextPage,  
 												 indexPage->length - OFFSET (IndexPage*, superNodes), 
 												 (const UCHAR*) indexPage->superNodes);
 }

=== modified file 'storage/falcon/IndexPage.h'
--- a/storage/falcon/IndexPage.h	2009-02-10 22:50:04 +0000
+++ b/storage/falcon/IndexPage.h	2009-03-02 18:36:32 +0000
@@ -79,8 +79,7 @@ public:
 	static int32	getRecordNumber(const UCHAR *ptr);
 	static void		logIndexPage (Bdb *bdb, TransId transId);
 
-	int32	parentPage;
-	int32	priorPage;
+	int32	unused[2]; // used to be parent and prior pages
 	int32	nextPage;
 	//short	level;
 	UCHAR	level;

=== modified file 'storage/falcon/IndexRootPage.cpp'
--- a/storage/falcon/IndexRootPage.cpp	2009-02-28 16:41:07 +0000
+++ b/storage/falcon/IndexRootPage.cpp	2009-03-02 18:36:32 +0000
@@ -124,7 +124,8 @@ bool IndexRootPage::addIndexEntry(Dbb * 
 		{
 		/* Find insert page and position on page */
 
-		Bdb *bdb = findInsertionLeaf(dbb, indexId, &searchKey, recordNumber, transId);
+		bool isRoot;
+		Bdb *bdb = findInsertionLeaf(dbb, indexId, &searchKey, recordNumber, transId,&isRoot);
 
 		if (!bdb)
 			return false;
@@ -142,6 +143,7 @@ bool IndexRootPage::addIndexEntry(Dbb * 
 
 			ASSERT (bdb->pageNumber != page->nextPage);
 			bdb = dbb->handoffPage (bdb, page->nextPage, PAGE_btree, Exclusive);
+			isRoot = false;
 			BDB_HISTORY(bdb);
 			}
 
@@ -158,12 +160,11 @@ bool IndexRootPage::addIndexEntry(Dbb * 
 			if (result != NextPage)
 				break;
 
-			int32 parentPageNumber = page->parentPage;
 			bdb = dbb->handoffPage(bdb, page->nextPage, PAGE_btree, Exclusive);
+			isRoot = false;
 			BDB_HISTORY(bdb);
 			bdb->mark(transId);
 			page = (IndexPage*) bdb->buffer;
-			page->parentPage = parentPageNumber;
 			}
 
 		if (result == NodeAdded || result == Duplicate)
@@ -177,11 +178,11 @@ bool IndexRootPage::addIndexEntry(Dbb * 
 			return true;
 			}
 
-		/* Node didn't fit.  Split the page and propogate the
+		/* Node didn't fit.  Split the page and propagate the
 		   split upward.  Sooner or later we'll go back and re-try
 		   the original insertion */
 
-		if (splitIndexPage (dbb, indexId, bdb, transId, result, key, recordNumber))
+		if (splitIndexPage (dbb, indexId, bdb, transId, result, key, recordNumber, isRoot))
 			return true;
 
 #ifdef _DEBUG
@@ -254,13 +255,16 @@ Bdb* IndexRootPage::findLeaf(Dbb *dbb, i
 
 	return bdb;
 }
-Bdb* IndexRootPage::findInsertionLeaf(Dbb *dbb, int32 indexId, IndexKey *indexKey, int32 recordNumber, TransId transId)
+Bdb* IndexRootPage::findInsertionLeaf(Dbb *dbb, int32 indexId, IndexKey *indexKey, int32 recordNumber, TransId transId, bool *isRoot)
 {
+	int rootPageNumber;
+
 	Bdb *bdb = findRoot(dbb, indexId, 0, Shared, transId);
 	BDB_HISTORY(bdb);
 
 	if (!bdb)
 		return NULL;
+	rootPageNumber = bdb->pageNumber;
 
 	IndexPage *page = (IndexPage*) bdb->buffer;
 	
@@ -279,6 +283,8 @@ Bdb* IndexRootPage::findInsertionLeaf(Db
 		page = (IndexPage*) bdb->buffer;
 		
 		if (page->level == 0)
+			if (isRoot)
+				*isRoot = true;
 			return bdb;
 		}
 		
@@ -316,9 +322,10 @@ Bdb* IndexRootPage::findInsertionLeaf(Db
 		if (dbb->debug & (DEBUG_PAGES | DEBUG_FIND_LEAF))
 			page->printPage (bdb, false);
 		
-		page->parentPage = parentPage;
+		// page->parentPage = parentPage;
 		}
-
+	if (isRoot)
+		*isRoot = (bdb->pageNumber == rootPageNumber);
 	return bdb;
 }
 
@@ -481,23 +488,15 @@ void IndexRootPage::scanIndex  (Dbb *dbb
 }
 
 bool IndexRootPage::splitIndexPage(Dbb * dbb, int32 indexId, Bdb * bdb, TransId transId,
-								   AddNodeResult addResult, IndexKey *indexKey, int recordNumber)
+								   AddNodeResult addResult, IndexKey *indexKey, int recordNumber, bool isRoot)
 {
-	// Start by splitting page (allocating new page)
 
 	IndexPage *page = (IndexPage*) bdb->buffer;
 	IndexKey splitKey(indexKey->index);
-	bool inserted = false;
 	Bdb *splitBdb;
 	
 	if (addResult == SplitEnd)
-		{
-		// int prevDebug = dbb->debug;
-		// dbb->debug = DEBUG_PAGES;
 		splitBdb = page->splitIndexPageEnd (dbb, bdb, transId, indexKey, recordNumber);
-		//dbb->debug = prevDebug;
-		inserted = true;
-		}
 	else
 		splitBdb = page->splitIndexPageMiddle (dbb, bdb, &splitKey, transId);
 
@@ -512,23 +511,18 @@ bool IndexRootPage::splitIndexPage(Dbb *
 		splitKey.appendRecordNumber(splitRecordNumber);
 		}
 
-	// If there isn't a parent page, we need to create a new level.  Do so.
-
-	if (!page->parentPage)
+	// If splitting root page, we need to create a new level
+	if (isRoot)
 		{
 		// Allocate and copy a new left-most index page
-		
 		Bdb *leftBdb = dbb->allocPage(PAGE_btree, transId);
 		BDB_HISTORY(leftBdb);
 		IndexPage *leftPage = (IndexPage*) leftBdb->buffer;
 		memcpy(leftPage, page, page->length);
 		leftBdb->setPageHeader(leftPage->pageType);
-//		leftPage->pageNumber = leftBdb->pageNumber;
-		splitPage->priorPage = leftBdb->pageNumber;
-		
+
 		// Create a node referencing the leftmost page. Assign to it a null key
 		// and record number 0 to ensure that all nodes are inserted after it.
-		
 		IndexKey leftKey(indexKey->index);
 
 		if (leftPage->level == 0)
@@ -547,22 +541,12 @@ bool IndexRootPage::splitIndexPage(Dbb *
 		page->addNode(dbb, &leftKey, leftBdb->pageNumber);
 		page->addNode(dbb, &splitKey, splitBdb->pageNumber);
 
-		leftPage->parentPage = bdb->pageNumber;
-		splitPage->parentPage = bdb->pageNumber;
-
-		// the order of adding these to the serial log is important.
+		// The order of adding these to the serial log is important.
 		// Recovery must write them in this order incase recovery itself crashes.
-
 		IndexPage::logIndexPage(splitBdb, transId);
 		IndexPage::logIndexPage(leftBdb, transId);
 		IndexPage::logIndexPage(bdb, transId);
 		
-		/***
-		IndexPage::printPage(bdb, false);
-		IndexPage::printPage(leftBdb, false);
-		IndexPage::printPage(splitBdb, false);
-		***/
-		
 		if (dbb->debug & DEBUG_PAGE_LEVEL)
 			{
 			Log::debug("\n***** splitIndexPage(%d, %d) - NEW LEVEL: Parent page = %d\n", recordNumber, indexId, bdb->pageNumber);
@@ -580,46 +564,50 @@ bool IndexRootPage::splitIndexPage(Dbb *
 		return false;
 		}
 
-	// We need to propogate the split upward.  Start over from the top
-	// to find the insertion point.  Try to insert.  If successful, be happy
+	
+	int splitPageLevel = splitPage->level;
+	int splitPageNumber = splitBdb->pageNumber;
+
+	IndexPage::logIndexPage(splitBdb, transId);
+	splitBdb->release(REL_HISTORY);
 
-	int level = splitPage->level + 1;
 	IndexPage::logIndexPage(bdb, transId);
 	bdb->release(REL_HISTORY);
-	int splitPageNumber = splitBdb->pageNumber;
-	splitBdb->release(REL_HISTORY);
-		
+
+	// We need to insert the first key of the newly created parent page
+	// into the parent page.
 	for (;;)
 		{
-		bdb = findRoot(dbb, indexId, 0, Exclusive, transId);
-		BDB_HISTORY(bdb);
-		page = (IndexPage*) bdb->buffer;
-		bdb = IndexPage::findLevel(dbb, indexId, bdb, level, &splitKey, splitRecordNumber);
-		BDB_HISTORY(bdb);
-		bdb->mark(transId);
-		page = (IndexPage*) bdb->buffer;
-
-		// If we can add the node, we're happy
-		
-		AddNodeResult result = page->addNode(dbb, &splitKey, splitPageNumber);
+		Bdb *rootBdb = findRoot(dbb, indexId, 0, Exclusive, transId);
+		int rootPageNumber = rootBdb->pageNumber;
+		BDB_HISTORY(rootBdb);
+
+		// Find parent page
+		Bdb *parentBdb = 
+			IndexPage::findLevel(dbb, indexId, rootBdb, splitPageLevel + 1, &splitKey, splitRecordNumber);
+		BDB_HISTORY(parentBdb);
+		parentBdb->mark(transId);
+		IndexPage *parentPage = (IndexPage*) parentBdb->buffer;
+		AddNodeResult result = parentPage->addNode(dbb, &splitKey, splitPageNumber);
 
 		if (result == NodeAdded || result == Duplicate)
 			{
-			splitBdb = dbb->fetchPage (splitPageNumber, PAGE_btree, Exclusive);
-			BDB_HISTORY(splitBdb);
-			splitBdb->mark (transId);
-			splitPage = (IndexPage*) splitBdb->buffer;
-			splitPage->parentPage = bdb->pageNumber;
-			bdb->release(REL_HISTORY);
-			splitBdb->release(REL_HISTORY);
+			// Node added to parent page
+			// Log parent page.
+			if (result == NodeAdded)
+				{
+				IndexPage::logIndexPage(parentBdb,transId);
+				}
 
+			parentBdb->release(REL_HISTORY);
 			return false;
 			}
 
-		// That page needs to be split.  Recurse
+		// Parent page needs to be split.Recurse
+		ASSERT(result == SplitMiddle || result == SplitEnd);
 
-		if (splitIndexPage (dbb, indexId, bdb, transId, 
-							result, &splitKey, splitPageNumber))
+		if (splitIndexPage (dbb, indexId, parentBdb, transId, result, &splitKey,
+				splitPageNumber, (parentBdb->pageNumber == rootPageNumber)))
 			return true;
 		}
 }
@@ -778,7 +766,7 @@ void IndexRootPage::debugBucket(Dbb *dbb
 	bdb->release(REL_HISTORY);
 }
 
-void IndexRootPage::redoIndexPage(Dbb* dbb, int32 pageNumber, int32 parentPage, int level, int32 priorPage, int32 nextPage, int length, const UCHAR *data, bool haveSuperNodes)
+void IndexRootPage::redoIndexPage(Dbb* dbb, int32 pageNumber, int level, 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_btree, NO_TRANSACTION);
@@ -786,9 +774,7 @@ void IndexRootPage::redoIndexPage(Dbb* d
 
 	IndexPage *indexPage = (IndexPage*) bdb->buffer;
 	indexPage->level = level;
-	indexPage->parentPage = parentPage;
 	indexPage->nextPage = nextPage;
-	indexPage->priorPage = priorPage;
 
 	if (haveSuperNodes)
 		{
@@ -801,38 +787,6 @@ void IndexRootPage::redoIndexPage(Dbb* d
 		memcpy(indexPage->nodes, data, length);
 		memset(indexPage->superNodes, 0, sizeof(indexPage->superNodes));
 		}
-
-	// If we have a parent page, propagate the first node upward
-
-	if (parentPage && indexPage->priorPage != 0)
-		{
-		IndexNode node(indexPage);
-		int number = node.getNumber();
-
-		if (number >= 0)
-			{
-			IndexKey indexKey(node.keyLength(), node.key);
-			Bdb *parentBdb = dbb->fetchPage(parentPage, PAGE_btree, Exclusive);
-			BDB_HISTORY(parentBdb);
-			IndexPage *parent = (IndexPage*) parentBdb->buffer;
-			
-			// Assertion disabled--for debug only.
-			// The parent page pointer is redundant and not always reliable.
-			// During an index split, lower level pages may be given a new
-			// parent page, but the parent pointers are not adjusted.
-			
-			//ASSERT(parent->level == indexPage->level + 1);
-			
-			if (level == 0)
-				indexKey.appendRecordNumber(number);
-				
-			parentBdb->mark(NO_TRANSACTION);
-			//AddNodeResult result = 
-			parent->addNode(dbb, &indexKey, pageNumber);
-			parentBdb->release(REL_HISTORY);
-			}
-		}
-
 	bdb->release(REL_HISTORY);
 }
 

=== modified file 'storage/falcon/IndexRootPage.h'
--- a/storage/falcon/IndexRootPage.h	2009-02-28 16:41:07 +0000
+++ b/storage/falcon/IndexRootPage.h	2009-03-02 18:36:32 +0000
@@ -44,13 +44,13 @@ public:
 	static void		deleteIndex (Dbb *dbb, int32 indexId, TransId transId);
 	static bool		deleteIndexEntry (Dbb *dbb, int32 indexId, IndexKey *key, int32 recordNumber, TransId transId);
 	static bool		splitIndexPage (Dbb *dbb, int32 indexId, Bdb *bdb, TransId transId,
-									AddNodeResult addResult, IndexKey *indexKey, int recordNumber);
+									AddNodeResult addResult, IndexKey *indexKey, int recordNumber, bool isRootPage);
 	static void		scanIndex (Dbb *dbb, int32 indexId, int32 rootPage, IndexKey *low, IndexKey *high, int searchFlags, TransId transId, Bitmap *bitmap);
 	static void		positionIndex(Dbb* dbb, int indexId, int32 rootPage, WalkIndex* walkIndex);
 	static void		repositionIndex(Dbb* dbb, int indexId, WalkIndex* walkIndex);
 	static Bdb*		findRoot (Dbb *dbb, int32 indexId, int32 rootPage, LockType lockType, TransId transId);
 	static Bdb*		findLeaf (Dbb *dbb, int32 indexId, int32 rootPage, IndexKey *key, LockType lockType, TransId transId);
-	static Bdb*		findInsertionLeaf (Dbb *dbb, int32 indexId, IndexKey *key, int32 recordNumber, TransId transId);
+	static Bdb*		findInsertionLeaf (Dbb *dbb, int32 indexId, IndexKey *key, int32 recordNumber, TransId transId, bool *isRootPage = NULL);
 	static bool		addIndexEntry (Dbb *dbb, int32 indexId, IndexKey *key, int32 recordNumber, TransId transId);
 	static int32	createIndex (Dbb *dbb, TransId transId);
 	static void		create (Dbb *dbb, TransId transId);
@@ -59,7 +59,7 @@ public:
 	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		redoIndexPage(Dbb* dbb, int32 pageNumber, int level, int32 next, int length, const UCHAR *data, bool haveSuperNodes);
 	static void		redoIndexDelete(Dbb* dbb, int indexId);
 	static void		redoCreateIndex(Dbb* dbb, int indexId, int pageNumber);
 };

=== modified file 'storage/falcon/PageInventoryPage.cpp'
--- a/storage/falcon/PageInventoryPage.cpp	2009-02-13 21:19:58 +0000
+++ b/storage/falcon/PageInventoryPage.cpp	2009-03-02 18:36:32 +0000
@@ -281,11 +281,9 @@ void PageInventoryPage::validateInventor
 								case PAGE_btree:
 									{
 									IndexPage *ipg = (IndexPage*) page;
-									validation->warning("orphan index page %d/%d, level %d, parent %d, prior %d, next %d", 
+									validation->warning("orphan index page %d/%d, level %d, next %d", 
 													pageNumber,  tableSpaceId,
 													ipg->level, 
-													ipg->parentPage, 
-													ipg->priorPage, 
 													ipg->nextPage);
 									}
 									break;

=== modified file 'storage/falcon/SRLIndexPage.cpp'
--- a/storage/falcon/SRLIndexPage.cpp	2009-02-28 16:41:07 +0000
+++ b/storage/falcon/SRLIndexPage.cpp	2009-03-02 18:36:32 +0000
@@ -41,7 +41,7 @@ SRLIndexPage::~SRLIndexPage()
 
 }
 
-void SRLIndexPage::append(Dbb *dbb, TransId transId, int idxVersion, int32 page, int32 lvl, int32 up, int32 left, int32 right, int length, const UCHAR *data)
+void SRLIndexPage::append(Dbb *dbb, TransId transId, int idxVersion, int32 page, int32 lvl, int32 right, int length, const UCHAR *data)
 {
 	START_RECORD(srlIndexPage, "SRLIndexPage::append");
 	
@@ -57,8 +57,6 @@ void SRLIndexPage::append(Dbb *dbb, Tran
 	putInt(idxVersion);	
 	putInt(page);
 	putInt(lvl);
-	putInt(up);
-	putInt(left);
 	putInt(right);
 	putInt(length);
 	putData(length, data);
@@ -78,15 +76,16 @@ void SRLIndexPage::read()
 
 	pageNumber = getInt();
 	level = getInt();
-	parent = getInt();
-	prior = getInt();
+	if (control->version < srlVersion19)
+		{
+		getInt(); // parent pointer
+		getInt(); // prior pointer
+		}
 	next = getInt();
 	length = getInt();
 	data = getData(length);
 
-	if (log->tracePage && (log->tracePage == pageNumber ||
-						   log->tracePage == prior ||
-						   log->tracePage == next))
+	if (log->tracePage && (log->tracePage == pageNumber || log->tracePage == next))
 		print();
 }
 
@@ -106,13 +105,13 @@ void SRLIndexPage::pass2()
 			switch (indexVersion)
 				{
 				case INDEX_VERSION_0:
-					Index2RootPage::redoIndexPage(log->getDbb(tableSpaceId), pageNumber, parent, level, prior, next, length, data);
+					Index2RootPage::redoIndexPage(log->getDbb(tableSpaceId), pageNumber, 0,  level, 0,  next, length, data);
 					break;
 				
 				case INDEX_VERSION_1:
 					{
 					bool haveSuperNodes = (control->version >=srlVersion14);
-					IndexRootPage::redoIndexPage(log->getDbb(tableSpaceId), pageNumber, parent, level, prior, next, length, data,
+					IndexRootPage::redoIndexPage(log->getDbb(tableSpaceId), pageNumber, level, next, length, data,
 						haveSuperNodes);
 					}
 					break;
@@ -128,7 +127,7 @@ void SRLIndexPage::pass2()
 
 void SRLIndexPage::print()
 {
-	logPrint("Index page %d/%d, level %d, parent %d, prior %d, next %d\n", pageNumber, tableSpaceId, level, parent, prior, next);
+	logPrint("Index page %d/%d, level %d, next %d\n", pageNumber, tableSpaceId, level, next);
 }
 
 void SRLIndexPage::redo()

=== modified file 'storage/falcon/SRLIndexPage.h'
--- a/storage/falcon/SRLIndexPage.h	2009-02-28 16:41:07 +0000
+++ b/storage/falcon/SRLIndexPage.h	2009-03-02 18:36:32 +0000
@@ -34,14 +34,12 @@ public:
 	virtual void pass2();
 	virtual void pass1();
 	virtual void read();
-	void append(Dbb *dbb, TransId transId, int idxVersion, int32 page, int32 lvl, int32 up, int32 left, int32 right, int length, const UCHAR *data);
+	void append(Dbb *dbb, TransId transId, int idxVersion, int32 page, int32 lvl, int32 up, int length, const UCHAR *data);
 	SRLIndexPage();
 	virtual ~SRLIndexPage();
 
 	int			indexVersion;
 	int32		pageNumber;
-	int32		parent;
-	int32		prior;
 	int32		next;
 	int32		level;
 	int32		length;

=== modified file 'storage/falcon/SRLVersion.h'
--- a/storage/falcon/SRLVersion.h	2009-02-28 16:41:07 +0000
+++ b/storage/falcon/SRLVersion.h	2009-03-02 18:36:32 +0000
@@ -45,7 +45,8 @@ static const int srlVersion15		= 15;	// 
 static const int srlVersion16		= 16;	// Added SRLInventoryPage January 26, 2009
 static const int srlVersion17		= 17;	// Log root page number in SRLCreateIndex
 static const int srlVersion18		= 18;	// Log tablespace list in SRLTableSpaces
-static const int srlCurrentVersion	= srlVersion18;
+static const int srlVersion19		= 19;	// Remove parent and prior pointers from SRLIndexPage
+static const int srlCurrentVersion	= srlVersion19;
 
 class SRLVersion : public SerialLogRecord  
 {


Attachment: [text/bzr-bundle] bzr/vvaintroub@mysql.com-20090302183632-3o6xoj1lgyebh0lf.bundle
Thread
bzr commit into mysql-6.0-falcon-team branch (vvaintroub:3043)Bug#42344 Bug#42903Vladislav Vaintroub2 Mar