On 12/15/2014 11:59 PM, Jeff Janes wrote:
On Mon, Dec 15, 2014 at 5:08 AM, Heikki Linnakangas <hlinnakan...@vmware.com
wrote:
Here's a new version of the patch. It now uses the same pairing heap code
that I posted in the other thread ("advance local xmin more aggressivley",
http://www.postgresql.org/message-id/5488acf0.8050...@vmware.com). The
pairingheap_remove() function is unused in this patch, but it is needed by
that other patch.
Under enable-cassert, this tries to call pairingheap_empty, but that
function doesn't exist.
I looked in the other patch and didn't find it defined there, either.
Ah, I renamed pairingheap_empty to pairingheap_is_empty at the last
minute, and missed the Asserts. Here's a corrected version.
- Heikki
diff --git a/src/backend/access/gist/gistget.c b/src/backend/access/gist/gistget.c
index 7a8692b..e5eb6f6 100644
--- a/src/backend/access/gist/gistget.c
+++ b/src/backend/access/gist/gistget.c
@@ -18,6 +18,7 @@
#include "access/relscan.h"
#include "miscadmin.h"
#include "pgstat.h"
+#include "lib/pairingheap.h"
#include "utils/builtins.h"
#include "utils/memutils.h"
#include "utils/rel.h"
@@ -243,8 +244,6 @@ gistScanPage(IndexScanDesc scan, GISTSearchItem *pageItem, double *myDistances,
GISTPageOpaque opaque;
OffsetNumber maxoff;
OffsetNumber i;
- GISTSearchTreeItem *tmpItem = so->tmpTreeItem;
- bool isNew;
MemoryContext oldcxt;
Assert(!GISTSearchItemIsHeap(*pageItem));
@@ -275,18 +274,15 @@ gistScanPage(IndexScanDesc scan, GISTSearchItem *pageItem, double *myDistances,
oldcxt = MemoryContextSwitchTo(so->queueCxt);
/* Create new GISTSearchItem for the right sibling index page */
- item = palloc(sizeof(GISTSearchItem));
- item->next = NULL;
+ item = palloc(SizeOfGISTSearchItem(scan->numberOfOrderBys));
item->blkno = opaque->rightlink;
item->data.parentlsn = pageItem->data.parentlsn;
/* Insert it into the queue using same distances as for this page */
- tmpItem->head = item;
- tmpItem->lastHeap = NULL;
- memcpy(tmpItem->distances, myDistances,
+ memcpy(item->distances, myDistances,
sizeof(double) * scan->numberOfOrderBys);
- (void) rb_insert(so->queue, (RBNode *) tmpItem, &isNew);
+ pairingheap_add(so->queue, &item->phNode);
MemoryContextSwitchTo(oldcxt);
}
@@ -348,8 +344,7 @@ gistScanPage(IndexScanDesc scan, GISTSearchItem *pageItem, double *myDistances,
oldcxt = MemoryContextSwitchTo(so->queueCxt);
/* Create new GISTSearchItem for this item */
- item = palloc(sizeof(GISTSearchItem));
- item->next = NULL;
+ item = palloc(SizeOfGISTSearchItem(scan->numberOfOrderBys));
if (GistPageIsLeaf(page))
{
@@ -372,12 +367,10 @@ gistScanPage(IndexScanDesc scan, GISTSearchItem *pageItem, double *myDistances,
}
/* Insert it into the queue using new distance data */
- tmpItem->head = item;
- tmpItem->lastHeap = GISTSearchItemIsHeap(*item) ? item : NULL;
- memcpy(tmpItem->distances, so->distances,
+ memcpy(item->distances, so->distances,
sizeof(double) * scan->numberOfOrderBys);
- (void) rb_insert(so->queue, (RBNode *) tmpItem, &isNew);
+ pairingheap_add(so->queue, &item->phNode);
MemoryContextSwitchTo(oldcxt);
}
@@ -390,44 +383,24 @@ gistScanPage(IndexScanDesc scan, GISTSearchItem *pageItem, double *myDistances,
* Extract next item (in order) from search queue
*
* Returns a GISTSearchItem or NULL. Caller must pfree item when done with it.
- *
- * NOTE: on successful return, so->curTreeItem is the GISTSearchTreeItem that
- * contained the result item. Callers can use so->curTreeItem->distances as
- * the distances value for the item.
*/
static GISTSearchItem *
getNextGISTSearchItem(GISTScanOpaque so)
{
- for (;;)
- {
- GISTSearchItem *item;
-
- /* Update curTreeItem if we don't have one */
- if (so->curTreeItem == NULL)
- {
- so->curTreeItem = (GISTSearchTreeItem *) rb_leftmost(so->queue);
- /* Done when tree is empty */
- if (so->curTreeItem == NULL)
- break;
- }
+ GISTSearchItem *item;
- item = so->curTreeItem->head;
- if (item != NULL)
- {
- /* Delink item from chain */
- so->curTreeItem->head = item->next;
- if (item == so->curTreeItem->lastHeap)
- so->curTreeItem->lastHeap = NULL;
- /* Return item; caller is responsible to pfree it */
- return item;
- }
-
- /* curTreeItem is exhausted, so remove it from rbtree */
- rb_delete(so->queue, (RBNode *) so->curTreeItem);
- so->curTreeItem = NULL;
+ if (!pairingheap_is_empty(so->queue))
+ {
+ item = (GISTSearchItem *) pairingheap_remove_first(so->queue);
+ }
+ else
+ {
+ /* Done when both heaps are empty */
+ item = NULL;
}
- return NULL;
+ /* Return item; caller is responsible to pfree it */
+ return item;
}
/*
@@ -458,7 +431,7 @@ getNextNearest(IndexScanDesc scan)
/* visit an index page, extract its items into queue */
CHECK_FOR_INTERRUPTS();
- gistScanPage(scan, item, so->curTreeItem->distances, NULL, NULL);
+ gistScanPage(scan, item, item->distances, NULL, NULL);
}
pfree(item);
@@ -491,7 +464,6 @@ gistgettuple(PG_FUNCTION_ARGS)
pgstat_count_index_scan(scan->indexRelation);
so->firstCall = false;
- so->curTreeItem = NULL;
so->curPageData = so->nPageData = 0;
fakeItem.blkno = GIST_ROOT_BLKNO;
@@ -534,7 +506,7 @@ gistgettuple(PG_FUNCTION_ARGS)
* this page, we fall out of the inner "do" and loop around to
* return them.
*/
- gistScanPage(scan, item, so->curTreeItem->distances, NULL, NULL);
+ gistScanPage(scan, item, item->distances, NULL, NULL);
pfree(item);
} while (so->nPageData == 0);
@@ -560,7 +532,6 @@ gistgetbitmap(PG_FUNCTION_ARGS)
pgstat_count_index_scan(scan->indexRelation);
/* Begin the scan by processing the root page */
- so->curTreeItem = NULL;
so->curPageData = so->nPageData = 0;
fakeItem.blkno = GIST_ROOT_BLKNO;
@@ -580,7 +551,7 @@ gistgetbitmap(PG_FUNCTION_ARGS)
CHECK_FOR_INTERRUPTS();
- gistScanPage(scan, item, so->curTreeItem->distances, tbm, &ntids);
+ gistScanPage(scan, item, item->distances, tbm, &ntids);
pfree(item);
}
diff --git a/src/backend/access/gist/gistscan.c b/src/backend/access/gist/gistscan.c
index 8360b16..eff02c4 100644
--- a/src/backend/access/gist/gistscan.c
+++ b/src/backend/access/gist/gistscan.c
@@ -22,14 +22,13 @@
/*
- * RBTree support functions for the GISTSearchTreeItem queue
+ * Pairing heap comparison function for the GISTSearchItem queue
*/
-
static int
-GISTSearchTreeItemComparator(const RBNode *a, const RBNode *b, void *arg)
+pairingheap_GISTSearchItem_cmp(const pairingheap_node *a, const pairingheap_node *b, void *arg)
{
- const GISTSearchTreeItem *sa = (const GISTSearchTreeItem *) a;
- const GISTSearchTreeItem *sb = (const GISTSearchTreeItem *) b;
+ const GISTSearchItem *sa = (const GISTSearchItem *) a;
+ const GISTSearchItem *sb = (const GISTSearchItem *) b;
IndexScanDesc scan = (IndexScanDesc) arg;
int i;
@@ -37,59 +36,16 @@ GISTSearchTreeItemComparator(const RBNode *a, const RBNode *b, void *arg)
for (i = 0; i < scan->numberOfOrderBys; i++)
{
if (sa->distances[i] != sb->distances[i])
- return (sa->distances[i] > sb->distances[i]) ? 1 : -1;
- }
-
- return 0;
-}
-
-static void
-GISTSearchTreeItemCombiner(RBNode *existing, const RBNode *newrb, void *arg)
-{
- GISTSearchTreeItem *scurrent = (GISTSearchTreeItem *) existing;
- const GISTSearchTreeItem *snew = (const GISTSearchTreeItem *) newrb;
- GISTSearchItem *newitem = snew->head;
-
- /* snew should have just one item in its chain */
- Assert(newitem && newitem->next == NULL);
-
- /*
- * If new item is heap tuple, it goes to front of chain; otherwise insert
- * it before the first index-page item, so that index pages are visited in
- * LIFO order, ensuring depth-first search of index pages. See comments
- * in gist_private.h.
- */
- if (GISTSearchItemIsHeap(*newitem))
- {
- newitem->next = scurrent->head;
- scurrent->head = newitem;
- if (scurrent->lastHeap == NULL)
- scurrent->lastHeap = newitem;
+ return (sa->distances[i] < sb->distances[i]) ? 1 : -1;
}
- else if (scurrent->lastHeap == NULL)
- {
- newitem->next = scurrent->head;
- scurrent->head = newitem;
- }
- else
- {
- newitem->next = scurrent->lastHeap->next;
- scurrent->lastHeap->next = newitem;
- }
-}
-static RBNode *
-GISTSearchTreeItemAllocator(void *arg)
-{
- IndexScanDesc scan = (IndexScanDesc) arg;
-
- return palloc(GSTIHDRSZ + sizeof(double) * scan->numberOfOrderBys);
-}
+ /* Heap items go before inner pages, to ensure a depth-first search */
+ if (GISTSearchItemIsHeap(*sa) && !GISTSearchItemIsHeap(*sb))
+ return -1;
+ if (!GISTSearchItemIsHeap(*sa) && GISTSearchItemIsHeap(*sb))
+ return 1;
-static void
-GISTSearchTreeItemDeleter(RBNode *rb, void *arg)
-{
- pfree(rb);
+ return 0;
}
@@ -127,7 +83,6 @@ gistbeginscan(PG_FUNCTION_ARGS)
so->queueCxt = giststate->scanCxt; /* see gistrescan */
/* workspaces with size dependent on numberOfOrderBys: */
- so->tmpTreeItem = palloc(GSTIHDRSZ + sizeof(double) * scan->numberOfOrderBys);
so->distances = palloc(sizeof(double) * scan->numberOfOrderBys);
so->qual_ok = true; /* in case there are zero keys */
@@ -188,15 +143,9 @@ gistrescan(PG_FUNCTION_ARGS)
/* create new, empty RBTree for search queue */
oldCxt = MemoryContextSwitchTo(so->queueCxt);
- so->queue = rb_create(GSTIHDRSZ + sizeof(double) * scan->numberOfOrderBys,
- GISTSearchTreeItemComparator,
- GISTSearchTreeItemCombiner,
- GISTSearchTreeItemAllocator,
- GISTSearchTreeItemDeleter,
- scan);
+ so->queue = pairingheap_allocate(pairingheap_GISTSearchItem_cmp, scan);
MemoryContextSwitchTo(oldCxt);
- so->curTreeItem = NULL;
so->firstCall = true;
/* Update scan key, if a new one is given */
diff --git a/src/backend/lib/Makefile b/src/backend/lib/Makefile
index 327a1bc..b24ece6 100644
--- a/src/backend/lib/Makefile
+++ b/src/backend/lib/Makefile
@@ -12,6 +12,6 @@ subdir = src/backend/lib
top_builddir = ../../..
include $(top_builddir)/src/Makefile.global
-OBJS = ilist.o binaryheap.o stringinfo.o
+OBJS = ilist.o binaryheap.o pairingheap.o stringinfo.o
include $(top_srcdir)/src/backend/common.mk
diff --git a/src/backend/lib/pairingheap.c b/src/backend/lib/pairingheap.c
new file mode 100644
index 0000000..a7e8901
--- /dev/null
+++ b/src/backend/lib/pairingheap.c
@@ -0,0 +1,235 @@
+/*-------------------------------------------------------------------------
+ *
+ * pairingheap.c
+ * A Pairing Heap implementation
+ *
+ * Portions Copyright (c) 2012-2014, PostgreSQL Global Development Group
+ *
+ * IDENTIFICATION
+ * src/backend/lib/pairingheap.c
+ *
+ *-------------------------------------------------------------------------
+ */
+
+#include "postgres.h"
+
+#include "lib/pairingheap.h"
+
+static pairingheap_node *merge(pairingheap *heap, pairingheap_node *a, pairingheap_node *b);
+static pairingheap_node *merge_children(pairingheap *heap, pairingheap_node *children);
+
+/*
+ * pairingheap_allocate
+ *
+ * Returns a pointer to a newly-allocated heap, with the heap property
+ * defined by the given comparator function, which will be invoked with the
+ * additional argument specified by 'arg'.
+ */
+pairingheap *
+pairingheap_allocate(pairingheap_comparator compare, void *arg)
+{
+ pairingheap *heap;
+
+ heap = (pairingheap *) palloc(sizeof(pairingheap));
+ heap->ph_compare = compare;
+ heap->ph_arg = arg;
+
+ heap->ph_root = NULL;
+
+ return heap;
+}
+
+/*
+ * pairingheap_free
+ *
+ * Releases memory used by the given pairingheap.
+ *
+ * Note: The items in the heap are not released!
+ */
+void
+pairingheap_free(pairingheap *heap)
+{
+ pfree(heap);
+}
+
+
+/* A helper function to merge two subheaps into one. */
+static pairingheap_node *
+merge(pairingheap *heap, pairingheap_node *a, pairingheap_node *b)
+{
+ if (a == NULL)
+ return b;
+ if (b == NULL)
+ return a;
+
+ /* Put the larger of the items as a child of the smaller one */
+ if (heap->ph_compare(a, b, heap->ph_arg) < 0)
+ {
+ pairingheap_node *tmp;
+
+ tmp = a;
+ a = b;
+ b = tmp;
+ }
+
+ if (a->first_child)
+ a->first_child->prev_or_parent = b;
+ b->prev_or_parent = a;
+ b->next_sibling = a->first_child;
+ a->first_child = b;
+ return a;
+}
+
+/*
+ * pairingheap_add
+ *
+ * Adds the given datum to the heap in O(1) time.
+ */
+void
+pairingheap_add(pairingheap *heap, pairingheap_node *d)
+{
+ d->first_child = NULL;
+
+ /* Link the new item as a new tree */
+ heap->ph_root = merge(heap, heap->ph_root, d);
+}
+
+/*
+ * pairingheap_first
+ *
+ * Returns a pointer to the first (root, topmost) node in the heap
+ * without modifying the heap. The caller must ensure that this
+ * routine is not used on an empty heap. Always O(1).
+ */
+pairingheap_node *
+pairingheap_first(pairingheap *heap)
+{
+ Assert(!pairingheap_is_empty(heap));
+ return heap->ph_root;
+}
+
+/*
+ * pairingheap_remove_first
+ *
+ * Removes the first (root, topmost) node in the heap and returns a
+ * pointer to it after rebalancing the heap. The caller must ensure
+ * that this routine is not used on an empty heap. O(log n) amortized.
+ */
+pairingheap_node *
+pairingheap_remove_first(pairingheap *heap)
+{
+ pairingheap_node *result;
+ pairingheap_node *children;
+
+ Assert(!pairingheap_is_empty(heap));
+
+ /* Remove the smallest root. */
+ result = heap->ph_root;
+ children = result->first_child;
+
+ heap->ph_root = merge_children(heap, children);
+
+ return result;
+}
+
+/*
+ * Merge a list of subheaps into a single heap.
+ *
+ * This implements the basic two-pass merging strategy, first forming
+ * pairs from left to right, and then merging the pairs.
+ */
+static pairingheap_node *
+merge_children(pairingheap *heap, pairingheap_node *children)
+{
+ pairingheap_node *item, *next;
+ pairingheap_node *pairs;
+ pairingheap_node *newroot;
+
+ if (children == NULL || children->next_sibling == NULL)
+ return children;
+
+ /* Walk the remaining subheaps from left to right, merging in pairs */
+ next = children;
+ pairs = NULL;
+ for (;;)
+ {
+ item = next;
+ if (item == NULL)
+ break;
+ if (item->next_sibling == NULL)
+ {
+ /* last odd item at the end of list */
+ item->next_sibling = pairs;
+ pairs = item;
+ break;
+ }
+ else
+ {
+ next = item->next_sibling->next_sibling;
+
+ item = merge(heap, item, item->next_sibling);
+ item->next_sibling = pairs;
+ pairs = item;
+ }
+ }
+
+ /*
+ * Form a single (sub)heap from the pairs.
+ */
+ newroot = pairs;
+ next = pairs->next_sibling;
+ while (next)
+ {
+ item = next;
+ next = item->next_sibling;
+
+ newroot = merge(heap, newroot, item);
+ }
+
+ return newroot;
+}
+
+/*
+ * Remove 'item' from the heap. O(log n) amortized.
+ */
+void
+pairingheap_remove(pairingheap *heap, pairingheap_node *item)
+{
+ pairingheap_node *children;
+ pairingheap_node *replacement;
+ pairingheap_node *next_sibling;
+ pairingheap_node **prev_ptr;
+
+ if (item == heap->ph_root)
+ {
+ (void) pairingheap_remove_first(heap);
+ return;
+ }
+
+ children = item->first_child;
+ next_sibling = item->next_sibling;
+
+ if (item->prev_or_parent->first_child == item)
+ prev_ptr = &item->prev_or_parent->first_child;
+ else
+ prev_ptr = &item->prev_or_parent->next_sibling;
+ Assert(*prev_ptr == item);
+
+ /* Form a new heap of the children */
+ replacement = merge_children(heap, children);
+
+ if (replacement == NULL)
+ {
+ *prev_ptr = next_sibling;
+ if (next_sibling)
+ next_sibling->prev_or_parent = item->prev_or_parent;
+ }
+ else
+ {
+ replacement->prev_or_parent = item->prev_or_parent;
+ replacement->next_sibling = item->next_sibling;
+ *prev_ptr = replacement;
+ if (next_sibling)
+ next_sibling->prev_or_parent = replacement;
+ }
+}
diff --git a/src/include/access/gist_private.h b/src/include/access/gist_private.h
index 2cbc918..07bc607 100644
--- a/src/include/access/gist_private.h
+++ b/src/include/access/gist_private.h
@@ -18,9 +18,9 @@
#include "access/itup.h"
#include "access/xlogreader.h"
#include "fmgr.h"
+#include "lib/pairingheap.h"
#include "storage/bufmgr.h"
#include "storage/buffile.h"
-#include "utils/rbtree.h"
#include "utils/hsearch.h"
/*
@@ -123,7 +123,7 @@ typedef struct GISTSearchHeapItem
/* Unvisited item, either index page or heap tuple */
typedef struct GISTSearchItem
{
- struct GISTSearchItem *next; /* list link */
+ pairingheap_node phNode;
BlockNumber blkno; /* index page number, or InvalidBlockNumber */
union
{
@@ -131,24 +131,12 @@ typedef struct GISTSearchItem
/* we must store parentlsn to detect whether a split occurred */
GISTSearchHeapItem heap; /* heap info, if heap tuple */
} data;
+ double distances[1]; /* array with numberOfOrderBys entries */
} GISTSearchItem;
#define GISTSearchItemIsHeap(item) ((item).blkno == InvalidBlockNumber)
-/*
- * Within a GISTSearchTreeItem's chain, heap items always appear before
- * index-page items, since we want to visit heap items first. lastHeap points
- * to the last heap item in the chain, or is NULL if there are none.
- */
-typedef struct GISTSearchTreeItem
-{
- RBNode rbnode; /* this is an RBTree item */
- GISTSearchItem *head; /* first chain member */
- GISTSearchItem *lastHeap; /* last heap-tuple member, if any */
- double distances[1]; /* array with numberOfOrderBys entries */
-} GISTSearchTreeItem;
-
-#define GSTIHDRSZ offsetof(GISTSearchTreeItem, distances)
+#define SizeOfGISTSearchItem(n_distances) (offsetof(GISTSearchItem, distances) + sizeof(double) * (n_distances))
/*
* GISTScanOpaqueData: private state for a scan of a GiST index
@@ -156,15 +144,12 @@ typedef struct GISTSearchTreeItem
typedef struct GISTScanOpaqueData
{
GISTSTATE *giststate; /* index information, see above */
- RBTree *queue; /* queue of unvisited items */
+ pairingheap *queue; /* queue of unvisited items */
MemoryContext queueCxt; /* context holding the queue */
bool qual_ok; /* false if qual can never be satisfied */
bool firstCall; /* true until first gistgettuple call */
- GISTSearchTreeItem *curTreeItem; /* current queue item, if any */
-
/* pre-allocated workspace arrays */
- GISTSearchTreeItem *tmpTreeItem; /* workspace to pass to rb_insert */
double *distances; /* output area for gistindex_keytest */
/* In a non-ordered search, returnable heap items are stored here: */
diff --git a/src/include/lib/pairingheap.h b/src/include/lib/pairingheap.h
new file mode 100644
index 0000000..e78196d
--- /dev/null
+++ b/src/include/lib/pairingheap.h
@@ -0,0 +1,67 @@
+/*
+ * pairingheap.h
+ *
+ * A Pairing Heap implementation
+ *
+ * Portions Copyright (c) 2012-2014, PostgreSQL Global Development Group
+ *
+ * src/include/lib/pairingheap.h
+ */
+
+#ifndef PAIRINGHEAP_H
+#define PAIRINGHEAP_H
+
+/*
+ * This represents an element stored in the heap. Embed this in a larger
+ * struct containing the actual data you're storing.
+ */
+typedef struct pairingheap_node
+{
+ struct pairingheap_node *first_child;
+ struct pairingheap_node *next_sibling;
+ struct pairingheap_node *prev_or_parent;
+} pairingheap_node;
+
+/*
+ * Return the containing struct of 'type' where 'membername' is the
+ * pairingheap_node pointed at by 'ptr'.
+ *
+ * This is used to convert a pairingheap_node * back to its containing struct.
+ */
+#define pairingheap_container(type, membername, ptr) \
+ (AssertVariableIsOfTypeMacro(ptr, pairingheap_node *), \
+ AssertVariableIsOfTypeMacro(((type *) NULL)->membername, pairingheap_node), \
+ ((type *) ((char *) (ptr) - offsetof(type, membername))))
+
+/*
+ * For a max-heap, the comparator must return <0 iff a < b, 0 iff a == b,
+ * and >0 iff a > b. For a min-heap, the conditions are reversed.
+ */
+typedef int (*pairingheap_comparator) (const pairingheap_node *a, const pairingheap_node *b, void *arg);
+
+/*
+ * A pairing heap.
+ */
+typedef struct pairingheap
+{
+ pairingheap_comparator ph_compare; /* comparison function */
+ void *ph_arg; /* opaque argument to ph_compare */
+ pairingheap_node *ph_root; /* current root of the heap */
+} pairingheap;
+
+extern pairingheap *pairingheap_allocate(pairingheap_comparator compare,
+ void *arg);
+extern void pairingheap_free(pairingheap *heap);
+extern void pairingheap_add(pairingheap *heap, pairingheap_node *d);
+extern pairingheap_node *pairingheap_first(pairingheap *heap);
+extern pairingheap_node *pairingheap_remove_first(pairingheap *heap);
+extern void pairingheap_remove(pairingheap *heap, pairingheap_node *d);
+
+#define pairingheap_reset(h) ((h)->ph_root = NULL)
+
+#define pairingheap_is_empty(h) ((h)->ph_root == NULL)
+
+/* Returns true if the heap contains exactly one item */
+#define pairingheap_is_singular(h) ((h)->ph_root && (h)->ph_root->first_child == NULL)
+
+#endif /* PAIRINGHEAP_H */
--
Sent via pgsql-hackers mailing list (pgsql-hackers@postgresql.org)
To make changes to your subscription:
http://www.postgresql.org/mailpref/pgsql-hackers
