From c0a4ec9672a0c7da58d980452c232009aecbc5b0 Mon Sep 17 00:00:00 2001 From: Masahiko Sawada Date: Fri, 26 Jan 2024 11:20:23 +0900 Subject: [PATCH v3 2/3] Add functions for updating keys and removing nodes to binaryheap. Previously, binaryheap didn't support key updates and removing nodes in an efficient way. For example, in order to remove a node from the binaryheap, the caller has to pass the node's position within the array that the binaryheap internally has. This operation can be done in O(log n) but searching for the key's position is O(n). This commit adds a hash table to binaryheap to track of positions of each nodes in the binaryheap. That way, by using newly added functions such as binaryheap_update_up() etc., both updating a key and removing a node can node can be done in O(1) in an average and O(log n) in worst case. This is known as the indexed priority queue. Author: Reviewed-by: Discussion: https://postgr.es/m/ Backpatch-through: --- src/backend/executor/nodeGatherMerge.c | 1 + src/backend/executor/nodeMergeAppend.c | 2 +- src/backend/postmaster/pgarch.c | 3 +- .../replication/logical/reorderbuffer.c | 1 + src/backend/storage/buffer/bufmgr.c | 1 + src/bin/pg_dump/pg_backup_archiver.c | 1 + src/bin/pg_dump/pg_dump_sort.c | 2 +- src/common/binaryheap.c | 190 +++++++++++++++++- src/include/lib/binaryheap.h | 36 +++- 9 files changed, 223 insertions(+), 14 deletions(-) diff --git a/src/backend/executor/nodeGatherMerge.c b/src/backend/executor/nodeGatherMerge.c index 2d552f4224..250f226d5f 100644 --- a/src/backend/executor/nodeGatherMerge.c +++ b/src/backend/executor/nodeGatherMerge.c @@ -427,6 +427,7 @@ 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); } diff --git a/src/backend/executor/nodeMergeAppend.c b/src/backend/executor/nodeMergeAppend.c index 0817868452..1980794cb7 100644 --- a/src/backend/executor/nodeMergeAppend.c +++ b/src/backend/executor/nodeMergeAppend.c @@ -125,7 +125,7 @@ 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, + mergestate->ms_heap = binaryheap_allocate(nplans, heap_compare_slots, false, mergestate); /* diff --git a/src/backend/postmaster/pgarch.c b/src/backend/postmaster/pgarch.c index 67693b0580..f3ec0a8918 100644 --- a/src/backend/postmaster/pgarch.c +++ b/src/backend/postmaster/pgarch.c @@ -250,7 +250,8 @@ PgArchiverMain(void) /* Initialize our max-heap for prioritizing files to archive. */ arch_files->arch_heap = binaryheap_allocate(NUM_FILES_PER_DIRECTORY_SCAN, - ready_file_comparator, NULL); + ready_file_comparator, false, + NULL); /* Load the archive_library. */ LoadArchiveLibrary(); diff --git a/src/backend/replication/logical/reorderbuffer.c b/src/backend/replication/logical/reorderbuffer.c index bbf0966182..c390d96ac3 100644 --- a/src/backend/replication/logical/reorderbuffer.c +++ b/src/backend/replication/logical/reorderbuffer.c @@ -1295,6 +1295,7 @@ ReorderBufferIterTXNInit(ReorderBuffer *rb, ReorderBufferTXN *txn, /* allocate heap */ state->heap = binaryheap_allocate(state->nr_txns, ReorderBufferIterCompare, + false, state); /* Now that the state fields are initialized, it is safe to return it. */ diff --git a/src/backend/storage/buffer/bufmgr.c b/src/backend/storage/buffer/bufmgr.c index eb1ec3b86d..183b91394c 100644 --- a/src/backend/storage/buffer/bufmgr.c +++ b/src/backend/storage/buffer/bufmgr.c @@ -2725,6 +2725,7 @@ BufferSync(int flags) */ ts_heap = binaryheap_allocate(num_spaces, ts_ckpt_progress_comparator, + false, 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 256d1e35a4..a044a684c8 100644 --- a/src/bin/pg_dump/pg_backup_archiver.c +++ b/src/bin/pg_dump/pg_backup_archiver.c @@ -4032,6 +4032,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); /* diff --git a/src/bin/pg_dump/pg_dump_sort.c b/src/bin/pg_dump/pg_dump_sort.c index f358dd22b9..63b1c3570d 100644 --- a/src/bin/pg_dump/pg_dump_sort.c +++ b/src/bin/pg_dump/pg_dump_sort.c @@ -404,7 +404,7 @@ TopoSort(DumpableObject **objs, return true; /* Create workspace for the above-described heap */ - pendingHeap = binaryheap_allocate(numObjs, int_cmp, NULL); + pendingHeap = binaryheap_allocate(numObjs, int_cmp, false, 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 bc43aca093..a5bb3b148d 100644 --- a/src/common/binaryheap.c +++ b/src/common/binaryheap.c @@ -22,8 +22,28 @@ #ifdef FRONTEND #include "common/logging.h" #endif +#include "common/hashfn.h" #include "lib/binaryheap.h" +/* + * Define parameters for hash table code generation. The interface is *also*" + * declared in binaryheaph.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_DEFINE +#include "lib/simplehash.h" + static void sift_down(binaryheap *heap, int node_off); static void sift_up(binaryheap *heap, int node_off); @@ -36,7 +56,8 @@ static void sift_up(binaryheap *heap, int node_off); * argument specified by 'arg'. */ binaryheap * -binaryheap_allocate(int capacity, binaryheap_comparator compare, void *arg) +binaryheap_allocate(int capacity, binaryheap_comparator compare, + bool indexed, void *arg) { binaryheap *heap; @@ -49,6 +70,17 @@ binaryheap_allocate(int capacity, binaryheap_comparator compare, void *arg) heap->bh_has_heap_property = true; heap->bh_nodes = (bh_node_type *) palloc(sizeof(bh_node_type) * capacity); + heap->bh_indexed = indexed; + if (heap->bh_indexed) + { +#ifdef FRONTEND + heap->bh_nodeidx = bh_nodeidx_create(capacity, NULL); +#else + heap->bh_nodeidx = bh_nodeidx_create(CurrentMemoryContext, capacity, + NULL); +#endif + } + return heap; } @@ -63,6 +95,9 @@ binaryheap_reset(binaryheap *heap) { heap->bh_size = 0; heap->bh_has_heap_property = true; + + if (heap->bh_indexed) + bh_nodeidx_reset(heap->bh_nodeidx); } /* @@ -73,6 +108,8 @@ binaryheap_reset(binaryheap *heap) void binaryheap_free(binaryheap *heap) { + if (heap->bh_indexed) + bh_nodeidx_destroy(heap->bh_nodeidx); pfree(heap); } @@ -117,6 +154,44 @@ bh_enlarge_node_array(binaryheap *heap) sizeof(bh_node_type) * heap->bh_space); } +/* + * Set the given node at the 'idx' and updates its position accordingly. + */ +static void +bh_set_node(binaryheap *heap, bh_node_type d, int idx) +{ + bh_nodeidx_entry *ent; + bool found; + + /* Set the node to the nodes array */ + heap->bh_nodes[idx] = d; + + if (heap->bh_indexed) + { + /* Remember its index in the nodes array */ + ent = bh_nodeidx_insert(heap->bh_nodeidx, d, &found); + ent->idx = idx; + } +} + +/* + * Replace the node at 'idx' with the given node 'replaced_by'. Also + * update their positions accordingly. + */ +static void +bh_replace_node(binaryheap *heap, int idx, bh_node_type replaced_by) +{ + bh_node_type node = heap->bh_nodes[idx]; + + /* Remove overwritten node's index */ + if (heap->bh_indexed) + (void) bh_nodeidx_delete(heap->bh_nodeidx, node); + + /* Replace it with the given new node */ + if (idx < heap->bh_size) + bh_set_node(heap, replaced_by, idx); +} + /* * binaryheap_add_unordered * @@ -131,7 +206,7 @@ binaryheap_add_unordered(binaryheap *heap, bh_node_type d) bh_enlarge_node_array(heap); heap->bh_has_heap_property = false; - heap->bh_nodes[heap->bh_size] = d; + bh_set_node(heap, d, heap->bh_size); heap->bh_size++; } @@ -162,7 +237,7 @@ binaryheap_add(binaryheap *heap, bh_node_type d) { bh_enlarge_node_array(heap); - heap->bh_nodes[heap->bh_size] = d; + bh_set_node(heap, d, heap->bh_size); heap->bh_size++; sift_up(heap, heap->bh_size - 1); } @@ -203,6 +278,10 @@ binaryheap_remove_first(binaryheap *heap) if (heap->bh_size == 1) { heap->bh_size--; + + if (heap->bh_indexed) + bh_nodeidx_delete(heap->bh_nodeidx, result); + return result; } @@ -210,7 +289,7 @@ binaryheap_remove_first(binaryheap *heap) * Remove the last node, placing it in the vacated root entry, and sift * the new root node down to its correct position. */ - heap->bh_nodes[0] = heap->bh_nodes[--heap->bh_size]; + bh_replace_node(heap, 0, heap->bh_nodes[--heap->bh_size]); sift_down(heap, 0); return result; @@ -236,7 +315,7 @@ binaryheap_remove_node(binaryheap *heap, int n) heap->bh_arg); /* remove the last node, placing it in the vacated entry */ - heap->bh_nodes[n] = heap->bh_nodes[heap->bh_size]; + bh_replace_node(heap, n, heap->bh_nodes[heap->bh_size]); /* sift as needed to preserve the heap property */ if (cmp > 0) @@ -245,6 +324,97 @@ binaryheap_remove_node(binaryheap *heap, int n) sift_down(heap, 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. + * + * This function can be used only if bh_indexed is true. + */ +void +binaryheap_remove_node_ptr(binaryheap *heap, bh_node_type d) +{ + bh_nodeidx_entry *ent; + + Assert(!binaryheap_empty(heap) && heap->bh_has_heap_property); + Assert(heap->bh_indexed); + + ent = bh_nodeidx_lookup(heap->bh_nodeidx, d); + Assert(ent); + + binaryheap_remove_node(heap, ent->idx); +} + +/* + * binaryheap_remove_node_ptr_unordered + * + * Remove the given datum from binaryheap in O(1) without preserving the heap property. + * To obtain a valid heap, one must call binaryheap_build() afterwards. + * + * This function can be used only if bh_indexed is true. + */ +void +binaryheap_remove_node_ptr_unordered(binaryheap *heap, bh_node_type d) +{ + bh_nodeidx_entry *ent; + + Assert(!binaryheap_empty(heap)); + Assert(heap->bh_indexed); + + ent = bh_nodeidx_lookup(heap->bh_nodeidx, d); + Assert(ent); + + heap->bh_has_heap_property = false; + bh_replace_node(heap, ent->idx, heap->bh_nodes[--heap->bh_size]); +} + +/* + * 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 bh_indexed is true. + */ +void +binaryheap_update_up(binaryheap *heap, bh_node_type d) +{ + bh_nodeidx_entry *ent; + + Assert(!binaryheap_empty(heap) && heap->bh_has_heap_property); + Assert(heap->bh_indexed); + + ent = bh_nodeidx_lookup(heap->bh_nodeidx, d); + Assert(ent); + Assert(ent->idx >= 0 && ent->idx < heap->bh_size); + + sift_up(heap, ent->idx); +} + +/* + * 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 bh_indexed is true. + */ +void +binaryheap_update_down(binaryheap *heap, bh_node_type d) +{ + bh_nodeidx_entry *ent; + + Assert(!binaryheap_empty(heap) && heap->bh_has_heap_property); + Assert(heap->bh_indexed); + + ent = bh_nodeidx_lookup(heap->bh_nodeidx, d); + Assert(ent); + Assert(ent->idx >= 0 && ent->idx < heap->bh_size); + + sift_down(heap, ent->idx); +} + /* * binaryheap_replace_first * @@ -257,7 +427,7 @@ binaryheap_replace_first(binaryheap *heap, bh_node_type d) { Assert(!binaryheap_empty(heap) && heap->bh_has_heap_property); - heap->bh_nodes[0] = d; + bh_replace_node(heap, 0, d); if (heap->bh_size > 1) sift_down(heap, 0); @@ -299,11 +469,11 @@ sift_up(binaryheap *heap, int node_off) * Otherwise, swap the parent value with the hole, and go on to check * the node's new parent. */ - heap->bh_nodes[node_off] = parent_val; + bh_set_node(heap, parent_val, node_off); node_off = parent_off; } /* Re-fill the hole */ - heap->bh_nodes[node_off] = node_val; + bh_set_node(heap, node_val, node_off); } /* @@ -358,9 +528,9 @@ sift_down(binaryheap *heap, int node_off) * Otherwise, swap the hole with the child that violates the heap * property; then go on to check its children. */ - heap->bh_nodes[node_off] = heap->bh_nodes[swap_off]; + bh_set_node(heap, heap->bh_nodes[swap_off], node_off); node_off = swap_off; } /* Re-fill the hole */ - heap->bh_nodes[node_off] = node_val; + bh_set_node(heap, node_val, node_off); } diff --git a/src/include/lib/binaryheap.h b/src/include/lib/binaryheap.h index 1439f20803..0f6a02573b 100644 --- a/src/include/lib/binaryheap.h +++ b/src/include/lib/binaryheap.h @@ -29,6 +29,28 @@ 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. + */ +typedef struct bh_nodeidx_entry +{ + bh_node_type key; + char status; + int idx; +} 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" + /* * binaryheap * @@ -47,11 +69,19 @@ typedef struct binaryheap binaryheap_comparator bh_compare; void *bh_arg; bh_node_type *bh_nodes; + + /* + * If bh_indexed is true, 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). + */ + bool bh_indexed; + bh_nodeidx_hash *bh_nodeidx; } binaryheap; extern binaryheap *binaryheap_allocate(int capacity, binaryheap_comparator compare, - void *arg); + bool indexed, void *arg); extern void binaryheap_reset(binaryheap *heap); extern void binaryheap_free(binaryheap *heap); extern void binaryheap_add_unordered(binaryheap *heap, bh_node_type d); @@ -60,7 +90,11 @@ 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_remove_node_ptr_unordered(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); #define binaryheap_empty(h) ((h)->bh_size == 0) #define binaryheap_size(h) ((h)->bh_size) -- 2.39.3