diff --git a/src/backend/access/gin/ginvacuum.c b/src/backend/access/gin/ginvacuum.c index 3104bc12b6..3d5f6a4ca5 100644 --- a/src/backend/access/gin/ginvacuum.c +++ b/src/backend/access/gin/ginvacuum.c @@ -316,10 +316,8 @@ ginScanToDelete(GinVacuumState *gvs, BlockNumber blkno, bool isRoot, /* * Scan through posting tree, delete empty tuples from leaf pages. - * Also, this function collects empty subtrees (with all empty leafs). - * For parents of these subtrees CleanUp lock is taken, then we call - * ScanToDelete. This is done for every inner page, which points to - * empty subtree. + * If there are empty pages root CleanUp lock is taken, then we call + * ScanToDelete. */ static bool ginVacuumPostingTreeLeaves(GinVacuumState *gvs, BlockNumber blkno, bool isRoot) @@ -356,13 +354,11 @@ ginVacuumPostingTreeLeaves(GinVacuumState *gvs, BlockNumber blkno, bool isRoot) { OffsetNumber i; bool hasEmptyChild = false; - bool hasNonEmptyChild = false; OffsetNumber maxoff = GinPageGetOpaque(page)->maxoff; BlockNumber *children = palloc(sizeof(BlockNumber) * (maxoff + 1)); /* - * Read all children BlockNumbers. Not sure it is safe if there are - * many concurrent vacuums. + * Read all children BlockNumbers. */ for (i = FirstOffsetNumber; i <= maxoff; i++) @@ -372,26 +368,38 @@ ginVacuumPostingTreeLeaves(GinVacuumState *gvs, BlockNumber blkno, bool isRoot) children[i] = PostingItemGetBlockNumber(pitem); } - UnlockReleaseBuffer(buffer); + LockBuffer(buffer, GIN_UNLOCK); for (i = FirstOffsetNumber; i <= maxoff; i++) { if (ginVacuumPostingTreeLeaves(gvs, children[i], false)) hasEmptyChild = true; - else - hasNonEmptyChild = true; } + LockBuffer(buffer, GIN_SHARE); + /* There might be concurrent splits of downlinks*/ + if ((GinPageGetOpaque(page)->maxoff) != maxoff) + { + OffsetNumber maxoff = GinPageGetOpaque(page)->maxoff; + for (i = FirstOffsetNumber; i <= maxoff; i++) + { + PostingItem *pitem = GinDataPageGetPostingItem(page, i); + if (ginVacuumPostingTreeLeaves(gvs, PostingItemGetBlockNumber(pitem), false)) + hasEmptyChild = true; + } + } + UnlockReleaseBuffer(buffer); + pfree(children); vacuum_delay_point(); /* - * All subtree is empty - just return true to indicate that parent - * must do a cleanup, unless we are ROOT and there is way to go upper. + * We signal upper that there are empty subtrees, root call will clean + * them up, after taking neccesary lock. */ - if (hasEmptyChild && !hasNonEmptyChild && !isRoot) + if (hasEmptyChild && !isRoot) return true; if (hasEmptyChild)