01_use_update_index_func_in_binaryheap.patch
application/octet-stream
Filename: 01_use_update_index_func_in_binaryheap.patch
Type: application/octet-stream
Part: 0
Patch
Format: unified
| File | + | − |
|---|---|---|
| src/backend/executor/nodeGatherMerge.c | 2 | 2 |
| src/backend/executor/nodeMergeAppend.c | 2 | 2 |
| src/backend/postmaster/pgarch.c | 1 | 1 |
| src/backend/replication/logical/reorderbuffer.c | 16 | 6 |
| src/backend/storage/buffer/bufmgr.c | 1 | 2 |
| src/bin/pg_dump/pg_backup_archiver.c | 1 | 2 |
| src/bin/pg_dump/pg_dump_sort.c | 1 | 1 |
| src/common/binaryheap.c | 24 | 128 |
| src/include/lib/binaryheap.h | 10 | 32 |
| src/include/replication/reorderbuffer.h | 3 | 0 |
diff --git a/src/backend/executor/nodeGatherMerge.c b/src/backend/executor/nodeGatherMerge.c
index ce19e0837a..7f1c743d0c 100644
--- a/src/backend/executor/nodeGatherMerge.c
+++ b/src/backend/executor/nodeGatherMerge.c
@@ -422,8 +422,8 @@ gather_merge_setup(GatherMergeState *gm_state)
/* Allocate the resources for the merge */
gm_state->gm_heap = binaryheap_allocate(nreaders + 1,
heap_compare_slots,
- false,
- gm_state);
+ gm_state,
+ NULL);
}
/*
diff --git a/src/backend/executor/nodeMergeAppend.c b/src/backend/executor/nodeMergeAppend.c
index 3efebd537f..7493421331 100644
--- a/src/backend/executor/nodeMergeAppend.c
+++ b/src/backend/executor/nodeMergeAppend.c
@@ -125,8 +125,8 @@ ExecInitMergeAppend(MergeAppend *node, EState *estate, int eflags)
mergestate->ms_nplans = nplans;
mergestate->ms_slots = (TupleTableSlot **) palloc0(sizeof(TupleTableSlot *) * nplans);
- mergestate->ms_heap = binaryheap_allocate(nplans, heap_compare_slots, false,
- mergestate);
+ mergestate->ms_heap = binaryheap_allocate(nplans, heap_compare_slots,
+ mergestate, NULL);
/*
* Miscellaneous initialization
diff --git a/src/backend/postmaster/pgarch.c b/src/backend/postmaster/pgarch.c
index 251f75e91d..fe2091498f 100644
--- a/src/backend/postmaster/pgarch.c
+++ b/src/backend/postmaster/pgarch.c
@@ -258,7 +258,7 @@ PgArchiverMain(char *startup_data, size_t startup_data_len)
/* Initialize our max-heap for prioritizing files to archive. */
arch_files->arch_heap = binaryheap_allocate(NUM_FILES_PER_DIRECTORY_SCAN,
- ready_file_comparator, false,
+ ready_file_comparator, NULL,
NULL);
/* Initialize our memory context. */
diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c
index 5cf28d4df4..17a44141c6 100644
--- a/src/backend/replication/logical/reorderbuffer.c
+++ b/src/backend/replication/logical/reorderbuffer.c
@@ -293,6 +293,7 @@ static void ReorderBufferSerializedPath(char *path, ReplicationSlot *slot,
static void ReorderBufferBuildMaxHeap(ReorderBuffer *rb);
static void ReorderBufferMaybeResetMaxHeap(ReorderBuffer *rb);
static int ReorderBufferTXNSizeCompare(Datum a, Datum b, void *arg);
+static void ReorderBufferTXNUpdateIndex(Datum d, int new_index);
static void ReorderBufferFreeSnap(ReorderBuffer *rb, Snapshot snap);
static Snapshot ReorderBufferCopySnap(ReorderBuffer *rb, Snapshot orig_snap,
@@ -399,7 +400,7 @@ ReorderBufferAllocate(void)
*/
buffer->txn_heap = binaryheap_allocate(MAX_HEAP_TXN_COUNT_THRESHOLD * 2,
ReorderBufferTXNSizeCompare,
- true, NULL);
+ NULL, ReorderBufferTXNUpdateIndex);
buffer->spillTxns = 0;
buffer->spillCount = 0;
@@ -1340,8 +1341,7 @@ ReorderBufferIterTXNInit(ReorderBuffer *rb, ReorderBufferTXN *txn,
/* allocate heap */
state->heap = binaryheap_allocate(state->nr_txns,
ReorderBufferIterCompare,
- false,
- state);
+ state, NULL);
/* Now that the state fields are initialized, it is safe to return it. */
*iter_state = state;
@@ -3277,7 +3277,7 @@ ReorderBufferChangeMemoryUpdate(ReorderBuffer *rb,
if ((txn->size - sz) == 0)
binaryheap_add(rb->txn_heap, PointerGetDatum(txn));
else
- binaryheap_update_up(rb->txn_heap, PointerGetDatum(txn));
+ binaryheap_update_up(rb->txn_heap, txn->heap_element_index);
}
}
else
@@ -3293,9 +3293,9 @@ ReorderBufferChangeMemoryUpdate(ReorderBuffer *rb,
if (ReorderBufferMaxHeapIsReady(rb))
{
if (txn->size == 0)
- binaryheap_remove_node_ptr(rb->txn_heap, PointerGetDatum(txn));
+ binaryheap_remove_node(rb->txn_heap, txn->heap_element_index);
else
- binaryheap_update_down(rb->txn_heap, PointerGetDatum(txn));
+ binaryheap_update_down(rb->txn_heap, txn->heap_element_index);
}
}
@@ -3552,6 +3552,16 @@ ReorderBufferSerializeReserve(ReorderBuffer *rb, Size sz)
}
}
+/*
+ * Callback function for updating the transaction's index in the max-heap.
+ */
+static void
+ReorderBufferTXNUpdateIndex(Datum d, int new_index)
+{
+ ReorderBufferTXN *txn = (ReorderBufferTXN *) DatumGetPointer(d);
+
+ txn->heap_element_index = new_index;
+}
/* Compare two transactions by size */
static int
diff --git a/src/backend/storage/buffer/bufmgr.c b/src/backend/storage/buffer/bufmgr.c
index 44836751b7..5c80ea4370 100644
--- a/src/backend/storage/buffer/bufmgr.c
+++ b/src/backend/storage/buffer/bufmgr.c
@@ -3014,8 +3014,7 @@ BufferSync(int flags)
*/
ts_heap = binaryheap_allocate(num_spaces,
ts_ckpt_progress_comparator,
- false,
- NULL);
+ NULL, NULL);
for (i = 0; i < num_spaces; i++)
{
diff --git a/src/bin/pg_dump/pg_backup_archiver.c b/src/bin/pg_dump/pg_backup_archiver.c
index 465e9ce777..aed7ca8786 100644
--- a/src/bin/pg_dump/pg_backup_archiver.c
+++ b/src/bin/pg_dump/pg_backup_archiver.c
@@ -4200,8 +4200,7 @@ restore_toc_entries_parallel(ArchiveHandle *AH, ParallelState *pstate,
/* Set up ready_heap with enough room for all known TocEntrys */
ready_heap = binaryheap_allocate(AH->tocCount,
TocEntrySizeCompareBinaryheap,
- false,
- NULL);
+ NULL, NULL);
/*
* The pending_list contains all items that we need to restore. Move all
diff --git a/src/bin/pg_dump/pg_dump_sort.c b/src/bin/pg_dump/pg_dump_sort.c
index 7362f7c961..9004704733 100644
--- a/src/bin/pg_dump/pg_dump_sort.c
+++ b/src/bin/pg_dump/pg_dump_sort.c
@@ -405,7 +405,7 @@ TopoSort(DumpableObject **objs,
return true;
/* Create workspace for the above-described heap */
- pendingHeap = binaryheap_allocate(numObjs, int_cmp, false, NULL);
+ pendingHeap = binaryheap_allocate(numObjs, int_cmp, NULL, NULL);
/*
* Scan the constraints, and for each item in the input, generate a count
diff --git a/src/common/binaryheap.c b/src/common/binaryheap.c
index c20ed50acc..f24835e7c4 100644
--- a/src/common/binaryheap.c
+++ b/src/common/binaryheap.c
@@ -25,27 +25,6 @@
#include "common/hashfn.h"
#include "lib/binaryheap.h"
-/*
- * Define parameters for hash table code generation. The interface is *also*
- * declared in binaryheap.h (to generate the types, which are externally
- * visible).
- */
-#define SH_PREFIX bh_nodeidx
-#define SH_ELEMENT_TYPE bh_nodeidx_entry
-#define SH_KEY_TYPE bh_node_type
-#define SH_KEY key
-#define SH_HASH_KEY(tb, key) \
- hash_bytes((const unsigned char *) &key, sizeof(bh_node_type))
-#define SH_EQUAL(tb, a, b) (memcmp(&a, &b, sizeof(bh_node_type)) == 0)
-#define SH_SCOPE extern
-#ifdef FRONTEND
-#define SH_RAW_ALLOCATOR pg_malloc0
-#endif
-#define SH_STORE_HASH
-#define SH_GET_HASH(tb, a) a->hash
-#define SH_DEFINE
-#include "lib/simplehash.h"
-
static void sift_down(binaryheap *heap, int node_off);
static void sift_up(binaryheap *heap, int node_off);
@@ -57,35 +36,25 @@ static void sift_up(binaryheap *heap, int node_off);
* function, which will be invoked with the additional argument specified by
* 'arg'.
*
- * If 'indexed' is true, we create a hash table to track each node's
- * index in the heap, enabling to perform some operations such as
- * binaryheap_remove_node_ptr() etc.
+ * The update_index callback function is optional. It it's set, the function
+ * is called whenever individual node moves to let the caller know the node's
+ * index in the bh_nodes.
*/
binaryheap *
-binaryheap_allocate(int num_nodes, binaryheap_comparator compare,
- bool indexed, void *arg)
+binaryheap_allocate(int num_nodes, binaryheap_comparator compare, void *arg,
+ binaryheap_update_index_func update_index)
{
binaryheap *heap;
heap = (binaryheap *) palloc(sizeof(binaryheap));
heap->bh_space = num_nodes;
heap->bh_compare = compare;
+ heap->bh_update_index = update_index;
heap->bh_arg = arg;
heap->bh_size = 0;
heap->bh_has_heap_property = true;
heap->bh_nodes = (bh_node_type *) palloc(sizeof(bh_node_type) * num_nodes);
- heap->bh_nodeidx = NULL;
-
- if (indexed)
- {
-#ifdef FRONTEND
- heap->bh_nodeidx = bh_nodeidx_create(num_nodes, NULL);
-#else
- heap->bh_nodeidx = bh_nodeidx_create(CurrentMemoryContext, num_nodes,
- NULL);
-#endif
- }
return heap;
}
@@ -101,9 +70,6 @@ binaryheap_reset(binaryheap *heap)
{
heap->bh_size = 0;
heap->bh_has_heap_property = true;
-
- if (binaryheap_indexed(heap))
- bh_nodeidx_reset(heap->bh_nodeidx);
}
/*
@@ -114,9 +80,6 @@ binaryheap_reset(binaryheap *heap)
void
binaryheap_free(binaryheap *heap)
{
- if (binaryheap_indexed(heap))
- bh_nodeidx_destroy(heap->bh_nodeidx);
-
pfree(heap->bh_nodes);
pfree(heap);
}
@@ -164,34 +127,15 @@ enlarge_node_array(binaryheap *heap)
*
* Return true if the node's index is already tracked.
*/
-static bool
+static void
set_node(binaryheap *heap, bh_node_type node, int index)
{
- bool found = false;
-
/* Set the node to the nodes array */
heap->bh_nodes[index] = node;
- if (binaryheap_indexed(heap))
- {
- bh_nodeidx_entry *ent;
-
- /* Keep track of the node index */
- ent = bh_nodeidx_insert(heap->bh_nodeidx, node, &found);
- ent->index = index;
- }
-
- return found;
-}
-
-/*
- * Remove the node's index from the hash table if the heap is indexed.
- */
-static inline void
-delete_nodeidx(binaryheap *heap, bh_node_type node)
-{
- if (binaryheap_indexed(heap))
- bh_nodeidx_delete(heap->bh_nodeidx, node);
+ /* Update the node index */
+ if (heap->bh_update_index != NULL)
+ heap->bh_update_index(node, index);
}
/*
@@ -203,21 +147,15 @@ delete_nodeidx(binaryheap *heap, bh_node_type node)
static void
replace_node(binaryheap *heap, int index, bh_node_type new_node)
{
- bool found PG_USED_FOR_ASSERTS_ONLY;
-
/* Quick return if not necessary to move */
if (heap->bh_nodes[index] == new_node)
return;
- /* Remove the overwritten node's index */
- delete_nodeidx(heap, heap->bh_nodes[index]);
-
/*
* Replace it with the given new node. This node's position must also be
* tracked as we assume to replace the node with the existing node.
*/
- found = set_node(heap, new_node, index);
- Assert(!binaryheap_indexed(heap) || found);
+ set_node(heap, new_node, index);
}
/*
@@ -310,7 +248,6 @@ binaryheap_remove_first(binaryheap *heap)
if (heap->bh_size == 1)
{
heap->bh_size--;
- delete_nodeidx(heap, result);
return result;
}
@@ -355,74 +292,33 @@ binaryheap_remove_node(binaryheap *heap, int n)
}
/*
- * binaryheap_remove_node_ptr
- *
- * Similar to binaryheap_remove_node() but removes the given node. The caller
- * must ensure that the given node is in the heap. O(log n) worst case.
+ * binaryheap_update_up
*
- * This function can be used only if the heap is indexed.
+ * Sift the nth (zero based) up after the node's key is updated. The caller
+ * must ensure that there are at least (n + 1) nodes in the heap. O(log n)
+ * worst case.
*/
void
-binaryheap_remove_node_ptr(binaryheap *heap, bh_node_type d)
+binaryheap_update_up(binaryheap *heap, int n)
{
- bh_nodeidx_entry *ent;
-
- Assert(!binaryheap_empty(heap) && heap->bh_has_heap_property);
- Assert(binaryheap_indexed(heap));
-
- ent = bh_nodeidx_lookup(heap->bh_nodeidx, d);
- Assert(ent);
-
- binaryheap_remove_node(heap, ent->index);
-}
-
-/*
- * Workhorse for binaryheap_update_up and binaryheap_update_down.
- */
-static void
-resift_node(binaryheap *heap, bh_node_type node, bool sift_dir_up)
-{
- bh_nodeidx_entry *ent;
-
Assert(!binaryheap_empty(heap) && heap->bh_has_heap_property);
- Assert(binaryheap_indexed(heap));
- ent = bh_nodeidx_lookup(heap->bh_nodeidx, node);
- Assert(ent);
- Assert(ent->index >= 0 && ent->index < heap->bh_size);
-
- if (sift_dir_up)
- sift_up(heap, ent->index);
- else
- sift_down(heap, ent->index);
-}
-
-/*
- * binaryheap_update_up
- *
- * Sift the given node up after the node's key is updated. The caller must
- * ensure that the given node is in the heap. O(log n) worst case.
- *
- * This function can be used only if the heap is indexed.
- */
-void
-binaryheap_update_up(binaryheap *heap, bh_node_type d)
-{
- resift_node(heap, d, true);
+ sift_up(heap, n);
}
/*
* binaryheap_update_down
*
- * Sift the given node down after the node's key is updated. The caller must
- * ensure that the given node is in the heap. O(log n) worst case.
- *
- * This function can be used only if the heap is indexed.
+ * Sift the nth (zero based) down after the node's key is updated. The caller
+ * must ensure that there are at least (n + 1) nodes in the heap. O(log n)
+ * worst case.
*/
void
-binaryheap_update_down(binaryheap *heap, bh_node_type d)
+binaryheap_update_down(binaryheap *heap, int n)
{
- resift_node(heap, d, false);
+ Assert(!binaryheap_empty(heap) && heap->bh_has_heap_property);
+
+ sift_up(heap, n);
}
/*
diff --git a/src/include/lib/binaryheap.h b/src/include/lib/binaryheap.h
index 4c1a1bb274..f3093a6ec9 100644
--- a/src/include/lib/binaryheap.h
+++ b/src/include/lib/binaryheap.h
@@ -30,27 +30,11 @@ typedef Datum bh_node_type;
typedef int (*binaryheap_comparator) (bh_node_type a, bh_node_type b, void *arg);
/*
- * Struct for a hash table element to store the node's index in the bh_nodes
- * array.
+ * This callback function is called whenever the node's position within
+ * the node array (i.e. bh_nodes) changes.
*/
-typedef struct bh_nodeidx_entry
-{
- bh_node_type key;
- int index; /* entry's index within the node array */
- char status; /* hash status */
- uint32 hash; /* hash values (cached) */
-} bh_nodeidx_entry;
-
-/* Define parameters necessary to generate the hash table interface. */
-#define SH_PREFIX bh_nodeidx
-#define SH_ELEMENT_TYPE bh_nodeidx_entry
-#define SH_KEY_TYPE bh_node_type
-#define SH_SCOPE extern
-#ifdef FRONTEND
-#define SH_RAW_ALLOCATOR pg_malloc0
-#endif
-#define SH_DECLARE
-#include "lib/simplehash.h"
+typedef void (*binaryheap_update_index_func) (bh_node_type d,
+ int new_element_index);
/*
* binaryheap
@@ -59,6 +43,7 @@ typedef struct bh_nodeidx_entry
* bh_space how many nodes can be stored in "nodes"
* bh_has_heap_property no unordered operations since last heap build
* bh_compare comparison function to define the heap property
+ * bh_update_index callback function called when updating the node's index
* bh_arg user data for comparison function
* bh_nodes variable-length array of "space" nodes
*/
@@ -68,20 +53,15 @@ typedef struct binaryheap
int bh_space;
bool bh_has_heap_property; /* debugging cross-check */
binaryheap_comparator bh_compare;
+ binaryheap_update_index_func bh_update_index;
void *bh_arg;
bh_node_type *bh_nodes;
-
- /*
- * If bh_nodeidx is not NULL, the bh_nodeidx is used to track of each
- * node's index in bh_nodes. This enables the caller to perform
- * binaryheap_remove_node_ptr(), binaryheap_update_up/down in O(log n).
- */
- bh_nodeidx_hash *bh_nodeidx;
} binaryheap;
extern binaryheap *binaryheap_allocate(int num_nodes,
binaryheap_comparator compare,
- bool indexed, void *arg);
+ void *arg,
+ binaryheap_update_index_func update_indx);
extern void binaryheap_reset(binaryheap *heap);
extern void binaryheap_free(binaryheap *heap);
extern void binaryheap_add_unordered(binaryheap *heap, bh_node_type d);
@@ -90,14 +70,12 @@ extern void binaryheap_add(binaryheap *heap, bh_node_type d);
extern bh_node_type binaryheap_first(binaryheap *heap);
extern bh_node_type binaryheap_remove_first(binaryheap *heap);
extern void binaryheap_remove_node(binaryheap *heap, int n);
-extern void binaryheap_remove_node_ptr(binaryheap *heap, bh_node_type d);
extern void binaryheap_replace_first(binaryheap *heap, bh_node_type d);
-extern void binaryheap_update_up(binaryheap *heap, bh_node_type d);
-extern void binaryheap_update_down(binaryheap *heap, bh_node_type d);
+extern void binaryheap_update_up(binaryheap *heap, int n);
+extern void binaryheap_update_down(binaryheap *heap, int n);
#define binaryheap_empty(h) ((h)->bh_size == 0)
#define binaryheap_size(h) ((h)->bh_size)
#define binaryheap_get_node(h, n) ((h)->bh_nodes[n])
-#define binaryheap_indexed(h) ((h)->bh_nodeidx != NULL)
#endif /* BINARYHEAP_H */
diff --git a/src/include/replication/reorderbuffer.h b/src/include/replication/reorderbuffer.h
index a5aec01c2f..202e976b2d 100644
--- a/src/include/replication/reorderbuffer.h
+++ b/src/include/replication/reorderbuffer.h
@@ -410,6 +410,9 @@ typedef struct ReorderBufferTXN
/* Size of top-transaction including sub-transactions. */
Size total_size;
+ /* Index of this transaction in the rb->txn_heap max-heap */
+ int heap_element_index;
+
/* If we have detected concurrent abort then ignore future changes. */
bool concurrent_abort;