GCC Code Coverage Report

Directory: src/
File: src/pg/parallel_backend.c
Date: 2026-10-23 09:17:32
Exec Total Coverage
Lines: 392 400 98.0%
Functions: 34 34 100.0%
Branches: 57 88 64.8%
Line Branch Exec Source
1 /*
2 * Copyright (c) 2026 Tiger Data, Inc.
3 * Licensed under the PostgreSQL License. See LICENSE for details.
4 *
5 * parallel_backend.c - PostgreSQL implementations of the build's back-end
6 * seams.
7 *
8 * The build driver is shared between the PG extension and the standalone
9 * engine; the parts that genuinely differ by back-end are expressed as
10 * same-named seam functions, with the PG versions here and the thread-based
11 * versions under src/standalone. It holds the PG implementations of all the
12 * build seams: DSM setup over shm_toc, ParallelContext launch/teardown, worker
13 * attach/detach, the heap parallel scan, and per-worker count reduction.
14 */
15
16 #include <postgres.h>
17
18 #include "vs_config.h"
19
20 #include <access/parallel.h>
21 #include <access/table.h>
22 #include <access/tableam.h>
23 #include <catalog/index.h>
24 #include <catalog/pg_operator_d.h>
25 #include <catalog/pg_type_d.h>
26 #include <executor/instrument.h>
27 #include <executor/tuptable.h>
28 #include <miscadmin.h>
29 #include <optimizer/plancat.h>
30 #include <pgstat.h>
31 #include <storage/buffile.h>
32 #include <storage/latch.h>
33 #include <storage/proc.h>
34 #include <storage/spin.h>
35 #include <tcop/tcopprot.h>
36 #include <utils/rel.h>
37 #include <utils/snapmgr.h>
38 #include <utils/tuplesort.h>
39 #include <utils/wait_event.h>
40
41 #include "build.h"
42 #include "index/index_build.h"
43 #include "index/parallel_build.h"
44 #include "pg/bufstorage.h"
45 #include "quant/matrix.h"
46 #include "quant/rabitq.h"
47 #include "support_pg.h"
48 #include "typeinfo.h"
49 #include "types/vec32.h"
50
51 /*
52 * PG-specific shared build state: the neutral PrismBuildShared plus the
53 * relation identity, query id, and the spinlock guarding its counters. A
54 * ParallelTableScanDesc is appended after it in the DSM segment. The base is
55 * the first member, so the PrismBuildShared * the workers look up out of the
56 * toc is recovered here as a PrismBuildSharedPg *.
57 */
58 typedef struct PrismBuildSharedPg
59 {
60 PrismBuildShared base;
61 Oid heaprelid;
62 Oid indexrelid;
63 int64 queryid;
64 /* DSM handle of the dedicated sample segment (sample-region seam). */
65 dsm_handle sample_handle;
66 /* DSM handle of the per-child subtree ring, created by the leader after
67 * root assignment (when the per-child sample counts that bound the slot
68 * size are known) and published before the ring barrier. */
69 dsm_handle subtree_ring_handle;
70 /* DSM handle of the exact centroid collection, created by the leader
71 * after the streaming tree write and published before the tree-ready
72 * barrier (exact-centroid seam). */
73 dsm_handle exact_centroids_handle;
74 slock_t mutex;
75 /* Striped locks guarding the shared leaf-refinement accumulator. */
76 slock_t accum_locks[PRISM_REFINE_LOCK_STRIPES];
77 } PrismBuildSharedPg;
78
79 #define ParallelTableScanFromVsShared(shared) \
80 ((ParallelTableScanDesc)((char *)(shared) + \
81 BUFFERALIGN(sizeof(PrismBuildSharedPg))))
82
83 /*
84 * Bridges PostgreSQL's heap-tuple callback to the back-end-neutral scan
85 * callback: skip nulls and hand the vector's data pointer (into the varlena,
86 * no copy) plus its tid to the shared logic.
87 *
88 * A halfvec column cannot be handed over no-copy -- the shared logic is
89 * float32 -- so it is widened into buf, reused across tuples. The callback
90 * consumes the pointer before returning, so one buffer per participant is
91 * enough.
92 */
93 typedef struct PrismPgScanAdapter
94 {
95 PrismBuildScanCb cb;
96 void *state;
97 Vec32Access input;
98 } PrismPgScanAdapter;
99
100 static void
101 230560 prism_pg_scan_adapter(
102 Relation index,
103 ItemPointer tid,
104 Datum *values,
105 bool *isnull,
106 bool tuple_is_alive,
107 void *adapter_state)
108 {
109 230560 PrismPgScanAdapter *a = (PrismPgScanAdapter *)adapter_state;
110
111 230560 (void)index;
112 230560 (void)tuple_is_alive;
113
114
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 230560 times.
 230560 if (isnull[0])
115 ✗ return;
116
117 230560 Vec32Ref vref = vec32_read(&a->input, values[0]);
118 230560 a->cb(a->state, *tid, vref.data);
119 }
120
121 /*
122 * Scan every vector via the shared parallel table scan, invoking cb per live
123 * tuple. The standalone back-end provides a same-named function that iterates
124 * its in-memory vector array (work-stealing) instead. allow_sync/progress
125 * are PostgreSQL table_index_build_scan flags, ignored in standalone; progress
126 * gates pg_stat_progress_create_index reporting, so only the leader sets it
127 * (otherwise every participant would inflate the tuples-scanned counter).
128 */
129 double
130 218 prism_build_scan(
131 Relation heap,
132 Relation index,
133 struct IndexInfo *indexInfo,
134 PrismBuildShared *shared,
135 bool allow_sync,
136 bool progress,
137 PrismBuildScanCb cb,
138 void *state)
139 {
140 218 Dimension dim = (Dimension)TupleDescAttr(index->rd_att, 0)->atttypmod;
141 /* Build-time path: resolved directly, since the per-backend cache reads a
142 * metadata page this build has not written yet. */
143 218 PrismPgScanAdapter actx = {
144 .cb = cb,
145 .state = state,
146 218 .input = vec32_access(
147 prism_index_type_info(index), dim, CurrentMemoryContext),
148 };
149
150 #if PG_VERSION_NUM >= 190000
151 TableScanDesc scan = table_beginscan_parallel(
152 heap, ParallelTableScanFromVsShared(shared), SO_NONE);
153 #else
154 218 TableScanDesc scan = table_beginscan_parallel(
155 heap, ParallelTableScanFromVsShared(shared));
156 #endif
157
158 218 return table_index_build_scan(
159 heap,
160 index,
161 indexInfo,
162 allow_sync,
163 progress,
164 prism_pg_scan_adapter,
165 &actx,
166 scan);
167 }
168
169 /*
170 * Relation lock modes a parallel worker uses for the heap and index. A
171 * concurrent build (CREATE INDEX CONCURRENTLY) must take weak locks: the
172 * leader holds only ShareUpdateExclusive on the heap, and acquiring a strong
173 * index lock in a worker logs it for hot-standby and assigns an XID, which is
174 * illegal in a parallel worker. Matches PostgreSQL's btree parallel build. The
175 * same pair is used to open the relations on attach and to close them on
176 * detach.
177 */
178 static void
179 170 worker_lockmodes(bool concurrent, LOCKMODE *heapmode, LOCKMODE *indexmode)
180 {
181 506 *heapmode = concurrent ? ShareUpdateExclusiveLock : ShareLock;
182 166 *indexmode = concurrent ? RowExclusiveLock : AccessExclusiveLock;
183 }
184
185 /* Per-worker build memory context (see prism_pbuild_worker_attach). */
186 static MemoryContext vs_pbuild_worker_ctx = NULL;
187 static MemoryContext vs_pbuild_worker_oldctx = NULL;
188
189 /*
190 * Join the parallel build: look up the shared state, open the heap and index,
191 * start per-worker instrumentation, and attach to the phase barrier. The
192 * standalone back-end provides a same-named function that takes the shared
193 * state and vectors directly and joins a thread barrier.
194 */
195 void
196 86 prism_pbuild_worker_attach(shm_toc *toc, PrismPBuildWorker *w)
197 {
198 86 PrismBuildShared *shared =
199 86 shm_toc_lookup(toc, PRISM_DSM_KEY_SHARED, false);
200 86 PrismBuildSharedPg *pg = (PrismBuildSharedPg *)shared;
201 86 Barrier *barrier = shm_toc_lookup(toc, PRISM_DSM_KEY_BARRIER, false);
202
203 86 char *sharedquery = shm_toc_lookup(toc, PRISM_DSM_KEY_QUERY_TEXT, true);
204 86 debug_query_string = sharedquery;
205 86 pgstat_report_activity(STATE_RUNNING, debug_query_string);
206 86 pgstat_report_query_id(pg->queryid, false);
207
208 86 LOCKMODE heapmode, indexmode;
209
2/2
✓ Branch 0 taken 84 times.
✓ Branch 1 taken 2 times.
 86 worker_lockmodes(shared->concurrent, &heapmode, &indexmode);
210
211 86 w->shared = shared;
212 86 w->barrier = barrier;
213 86 w->heapRel = table_open(pg->heaprelid, heapmode);
214 86 w->indexRel = index_open(pg->indexrelid, indexmode);
215 86 w->worker_id = ParallelWorkerNumber + 1;
216 86 w->dim = shared->dim;
217
218 /*
219 * All of this worker's build allocations (routing state, per-child
220 * subtree blobs, batch buffers) go into a named context so
221 * pg_backend_memory_contexts attributes them to the build and they are
222 * reclaimed together at detach. A file-static is safe for the same
223 * reason as vs_pbuild_snapshot: one build per worker, non-reentrant.
224 */
225 86 vs_pbuild_worker_ctx = AllocSetContextCreate(
226 CurrentMemoryContext, "vs worker build", ALLOCSET_DEFAULT_SIZES);
227 86 vs_pbuild_worker_oldctx = MemoryContextSwitchTo(vs_pbuild_worker_ctx);
228
229 86 InstrStartParallelQuery();
230
231 /*
232 * Attach to the dynamic phase barrier (before the first phase). The leader
233 * attaches before launching workers, so the party tracks all participants
234 * that actually start and stays in lockstep regardless of how many workers
235 * PostgreSQL launched.
236 */
237 86 BarrierAttach(barrier);
238 86 }
239
240 /*
241 * Leave the parallel build: report this worker's buffer/WAL usage back to the
242 * leader and close the relations. The standalone back-end's same-named
243 * function joins the thread and is otherwise a no-op.
244 */
245 void
246 84 prism_pbuild_worker_detach(shm_toc *toc, PrismPBuildWorker *w)
247 {
248 84 MemoryContextSwitchTo(vs_pbuild_worker_oldctx);
249 84 MemoryContextDelete(vs_pbuild_worker_ctx);
250 84 vs_pbuild_worker_ctx = NULL;
251 84 vs_pbuild_worker_oldctx = NULL;
252
253 84 BufferUsage *bufferusage =
254 84 shm_toc_lookup(toc, PRISM_DSM_KEY_BUFFER_USAGE, false);
255 84 WalUsage *walusage = shm_toc_lookup(toc, PRISM_DSM_KEY_WAL_USAGE, false);
256 84 InstrEndParallelQuery(
257 84 &bufferusage[ParallelWorkerNumber],
258 84 &walusage[ParallelWorkerNumber]);
259
260 84 LOCKMODE heapmode, indexmode;
261
2/2
✓ Branch 0 taken 82 times.
✓ Branch 1 taken 2 times.
 84 worker_lockmodes(w->shared->concurrent, &heapmode, &indexmode);
262 84 index_close(w->indexRel, indexmode);
263 84 table_close(w->heapRel, heapmode);
264 84 }
265
266 /*
267 * Page-backed routing storage seam (see parallel_build.h). PG workers are
268 * separate processes, so each opens its own VsStorage on the worker's index
269 * relation; the leader's storage pointer cannot cross the process boundary, so
270 * publish is a no-op here.
271 */
272 void
273 43 prism_pbuild_publish_storage(PrismBuildShared *shared, VsStorage *s)
274 {
275 43 (void)shared;
276 43 (void)s;
277 43 }
278
279 VsStorage *
280 84 prism_pbuild_worker_storage(PrismPBuildWorker *w)
281 {
282 /* No table relation needed (routing reads index pages only, no rerank). */
283 84 VsPgStorage *s = palloc(sizeof(VsPgStorage));
284 84 vs_pg_storage_init(s, w->indexRel, NULL, w->shared->metric);
285 84 s->build_mode = true; /* reads only; matches the leader's build storage */
286 84 return &s->base; /* base is the first member */
287 }
288
289 void
290 84 prism_pbuild_worker_storage_release(VsStorage *s)
291 {
292 /* The route helper releases every page it reads, so no buffer stays
293 * pinned; just free the wrapper (allocated in the worker's memory
294 * context). */
295 84 pfree(s);
296 84 }
297
298 /*
299 * Snapshot used to initialize the parallel heap scan for CREATE INDEX
300 * CONCURRENTLY (an MVCC snapshot; SnapshotAny/NULL for a normal build). It
301 * must stay registered for the whole parallel operation and is released at
302 * teardown. A file-static is safe: an index build is single-threaded and
303 * non-reentrant in the leader backend, and every parallel-build exit path runs
304 * prism_pbuild_teardown.
305 */
306 static Snapshot vs_pbuild_snapshot = NULL;
307
308 /*
309 * Tear the parallel context down and leave parallel mode. The standalone
310 * back-end provides a same-named function that joins its worker threads and
311 * frees the shared arena instead.
312 */
313 void
314 44 prism_pbuild_teardown(ParallelContext *pcxt)
315 {
316
2/2
✓ Branch 0 taken 1 times.
✓ Branch 1 taken 43 times.
 44 if (vs_pbuild_snapshot != NULL)
317 {
318 1 UnregisterSnapshot(vs_pbuild_snapshot);
319 1 vs_pbuild_snapshot = NULL;
320 }
321 44 DestroyParallelContext(pcxt);
322 44 ExitParallelMode();
323 44 }
324
325 /*
326 * Leader-only subtree blob store (see parallel_build.h): a BufFile temp
327 * file. Small blob sets never leave the kernel page cache; large ones spill
328 * to pgsql_tmp automatically, so the store adds no unbounded memory. Blobs
329 * are length-prefixed and read back strictly in append order.
330 */
331 struct PrismBlobStore
332 {
333 BufFile *file;
334 };
335
336 PrismBlobStore *
337 170 prism_pbuild_blobstore_begin(void)
338 {
339 170 PrismBlobStore *bs = palloc(sizeof(PrismBlobStore));
340 170 bs->file = BufFileCreateTemp(false);
341 170 return bs;
342 }
343
344 void
345 3771 prism_pbuild_blobstore_put(PrismBlobStore *bs, const void *blob, uint64_t size)
346 {
347 3771 BufFileWrite(bs->file, &size, sizeof(size));
348 3771 BufFileWrite(bs->file, blob, (size_t)size);
349 3771 }
350
351 void
352 170 prism_pbuild_blobstore_rewind(PrismBlobStore *bs)
353 {
354
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 170 times.
 170 if (BufFileSeek(bs->file, 0, 0, SEEK_SET) != 0)
355 ✗ ereport(ERROR,
356 (errcode_for_file_access(),
357 errmsg("could not rewind subtree blob store")));
358 170 }
359
360 uint64_t
361 3771 prism_pbuild_blobstore_get(PrismBlobStore *bs, void *buf, uint64_t max_size)
362 {
363 3771 uint64_t size;
364 3771 BufFileReadExact(bs->file, &size, sizeof(size));
365
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 3771 times.
 3771 if (size > max_size)
366 ✗ ereport(ERROR,
367 (errcode(ERRCODE_INTERNAL_ERROR),
368 errmsg("subtree blob larger than its slot (%llu > %llu)",
369 (unsigned long long)size,
370 (unsigned long long)max_size)));
371 3771 BufFileReadExact(bs->file, buf, (size_t)size);
372 3771 return size;
373 }
374
375 void
376 170 prism_pbuild_blobstore_end(PrismBlobStore *bs)
377 {
378 170 BufFileClose(bs->file);
379 170 pfree(bs);
380 170 }
381
382 /*
383 * Launch the worker participants and wait until they have all attached to the
384 * barrier (so the dynamic party reaches launched+1 before the leader advances
385 * the first phase). Returns false — after tearing the context down — if no
386 * workers started, so the caller falls back to a serial build. The standalone
387 * back-end provides a same-named function that spawns threads and joins them
388 * at the barrier instead.
389 */
390 bool
391 45 prism_pbuild_launch(
392 ParallelContext *pcxt, Barrier *barrier, PrismBuildShared *shared)
393 {
394 45 LaunchParallelWorkers(pcxt);
395
396
2/2
✓ Branch 0 taken 1 times.
✓ Branch 1 taken 44 times.
 45 if (pcxt->nworkers_launched == 0)
397 {
398 1 WaitForParallelWorkersToFinish(pcxt);
399 1 prism_pbuild_teardown(pcxt);
400 1 return false;
401 }
402
403 /*
404 * Workers attach to the barrier dynamically, so the party (1 leader + N
405 * launched) is not final until they all have; if the leader arrived first
406 * it could advance the phase alone and strand late workers.
407 * WaitForParallelWorkersToAttach surfaces a startup failure as an error
408 * rather than a hang; then poll until the live participant count is whole.
409 */
410 44 WaitForParallelWorkersToAttach(pcxt);
411
2/2
✓ Branch 2 taken 271 times.
✓ Branch 3 taken 44 times.
 315 while (BarrierParticipants(barrier) < pcxt->nworkers_launched + 1)
412 {
413
2/2
✓ Branch 0 taken 4 times.
✓ Branch 1 taken 267 times.
 271 CHECK_FOR_INTERRUPTS();
414 271 (void)WaitLatch(
415 MyLatch,
416 WL_LATCH_SET | WL_TIMEOUT | WL_EXIT_ON_PM_DEATH,
417 1L,
418 WAIT_EVENT_PARALLEL_CREATE_INDEX_SCAN);
419 271 ResetLatch(MyLatch);
420 }
421
422 /* The launch can fall short of the plan (the parallel-worker pool under
423 * max_parallel_workers is shared with concurrent queries). Narrow the
424 * participant count to the party that attached so the phases partition
425 * their work over participants that exist; the per-participant DSM
426 * regions keep their planned size and leave the tail slots unused. */
427 44 shared->nparticipants = pcxt->nworkers_launched + 1;
428 44 return true;
429 }
430
431 /*
432 * Allocate and populate the parallel build's shared state: the DSM segment and
433 * its regions (shared header, barrier, sample/centroid/assignment slots, the
434 * tree blob, the per-worker page queues, usage counters). Does not launch
435 * workers; the caller does. Coarse PG block — the standalone back-end provides
436 * a same-named function over a heap arena. Returns false (after tearing the
437 * parallel context down) if the DSM segment could not be created.
438 */
439 bool
440 45 prism_pbuild_setup_shared(
441 PrismPBuildLeader *lead,
442 Relation heap,
443 Relation index,
444 const PrismBuildConfig *config,
445 int nworkers)
446 {
447 45 Dimension dim = config->dim;
448 45 uint32_t nlist = config->nlist;
449 45 int nparticipants = nworkers + 1;
450 45 uint64_t rabitq_seed = VS_RABITQ_BUILD_SEED;
451 90 uint32_t fan_out = config->fan_out > 0 ? config->fan_out
452
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 45 times.
 45 : prism_auto_fan_out(0, nlist, 0);
453 45 uint32_t km_k = fan_out < nlist ? fan_out : nlist;
454
455 /*
456 * Size the k-means sample set. The samples live in one shared-memory
457 * region (total_samples * dim floats) that must stay resident for the
458 * whole tree build -- root k-means and every subtree -- so its size is the
459 * build's dominant memory cost. The ideal is ~256 samples per list, but at
460 * fine nlist that can dwarf available RAM (nlist=480k -> 123M samples ->
461 * ~360 GB), far beyond what a DSM segment can hold.
462 *
463 * Bound it by maintenance_work_mem: that is the build's memory budget and
464 * the knob operators already raise for large index builds. When the budget
465 * is smaller than the ideal the stride sampler simply draws a coarser (but
466 * still uniform) subsample to fit. estimate_rel_size() caps it to the rows
467 * that actually exist so small tables don't over-allocate; it is only a
468 * hint -- the budget is the hard bound, so an inaccurate estimate cannot
469 * over-commit shared memory.
470 */
471 45 uint64_t want_samples = (uint64_t)nlist * 256;
472
473 /* maintenance_work_mem is in kB; reserve it for the sample region. */
474 45 uint64_t mem_bytes = (uint64_t)maintenance_work_mem * UINT64CONST(1024);
475 45 uint64_t vec_nbytes = (uint64_t)dim * sizeof(float);
476
1/2
✓ Branch 0 taken 45 times.
✗ Branch 1 not taken.
 45 uint64_t budget = vec_nbytes > 0 ? mem_bytes / vec_nbytes : want_samples;
477 45 if (budget < 10000)
478 budget = 10000; /* k-means needs a workable minimum */
479
480 45 uint64_t total64 = Min(want_samples, budget);
481
482 45 BlockNumber est_pages;
483 45 double est_tuples;
484 45 double allvisfrac;
485 45 estimate_rel_size(heap, NULL, &est_pages, &est_tuples, &allvisfrac);
486
3/4
✓ Branch 0 taken 45 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 38 times.
✓ Branch 3 taken 7 times.
 45 if (est_tuples > 0.0 && (double)total64 > est_tuples)
487 38 total64 = (uint64_t)est_tuples;
488
489 /* total64 <= want_samples = nlist*256 <= 512M (nlist reloption max 2M), so
490 * it always fits a uint32. */
491 45 uint32_t total_samples = (uint32_t)Max(total64, UINT64CONST(1));
492
493
3/4
✓ Branch 0 taken 2 times.
✓ Branch 1 taken 43 times.
✓ Branch 2 taken 2 times.
✗ Branch 3 not taken.
 45 if (want_samples > budget && est_tuples > (double)budget)
494
1/2
✓ Branch 1 taken 2 times.
✗ Branch 2 not taken.
 2 elog(LOG,
495 VS_EXTENSION_NAME
496 ": k-means sample set limited to %u of the "
497 "ideal " UINT64_FORMAT " vectors by maintenance_work_mem "
498 "(%d kB); raise maintenance_work_mem for finer centroid "
499 "training on large tables",
500 total_samples,
501 want_samples,
502 maintenance_work_mem);
503
504 45 uint32_t max_per_worker = (total_samples + nparticipants - 1) /
505 nparticipants;
506
507 /*
508 * The refine decision itself is the leader's, made after clustering when
509 * the actual leaf count is known (see PrismBuildShared.refine); setup only
510 * publishes the gate inputs. The pass is single-shot by design: refine
511 * routes every row over the centroid PAGES, which it never rewrites (it
512 * updates the heads' encode references, which routing does not read), so
513 * the row-to-leaf assignment is a fixed point -- a second pass would
514 * rescan the whole table to recompute the exact same means.
515 */
516
517 45 EnterParallelMode();
518
519 45 ParallelContext *pcxt = CreateParallelContext(
520 VS_MODULE_NAME, "prism_parallel_build_main", nworkers);
521
522 /*
523 * The heap scan's snapshot. A normal build sees all tuples (SnapshotAny);
524 * CREATE INDEX CONCURRENTLY must use an MVCC snapshot so it indexes only
525 * tuples visible to it (heapam asserts SnapshotAny <-> a valid OldestXmin,
526 * so the concurrent path must not pass SnapshotAny). Register it for the
527 * duration — its serialized size also affects the DSM estimate below — and
528 * release it in prism_pbuild_teardown. Mirrors PostgreSQL's nbtsort.c.
529 */
530 90 Snapshot snapshot = config->concurrent
531 1 ? RegisterSnapshot(GetTransactionSnapshot())
532
2/2
✓ Branch 0 taken 1 times.
✓ Branch 1 taken 44 times.
 45 : SnapshotAny;
533
2/2
✓ Branch 0 taken 44 times.
✓ Branch 1 taken 1 times.
 45 vs_pbuild_snapshot = (snapshot != SnapshotAny) ? snapshot : NULL;
534 45 Size est_shared = add_size(
535 BUFFERALIGN(sizeof(PrismBuildSharedPg)),
536 table_parallelscan_estimate(heap, snapshot));
537
538 45 shm_toc_estimate_chunk(&pcxt->estimator, est_shared);
539 45 shm_toc_estimate_chunk(&pcxt->estimator, sizeof(Barrier));
540 /* K-means shared centroids + norms (root level, k=km_k) */
541 45 shm_toc_estimate_chunk(
542 &pcxt->estimator, prism_dsm_centroids_size(km_k, dim));
543 /* K-means per-worker accumulators */
544 45 shm_toc_estimate_chunk(
545 &pcxt->estimator,
546 prism_dsm_km_workers_size(nparticipants, km_k, dim));
547 /* Root assignments: per-worker uint32_t[max_per_worker] */
548 45 shm_toc_estimate_chunk(
549 &pcxt->estimator,
550 prism_dsm_root_assign_size(nparticipants, max_per_worker));
551 /* Shared coordinator for the cluster-keyed posting sort (sort seam). Sized
552 * for the planned participant count (upper bound on launched workers). */
553 45 shm_toc_estimate_chunk(
554 &pcxt->estimator, prism_pbuild_sort_shared_size(nparticipants));
555
556 /* Page-backed routing: the global mean, published by the leader before the
557 * tree-ready barrier. The posting-head base (leaf c's head = first_posting
558 * + c) is a scalar in PrismBuildShared, so no O(nlist) head array is
559 * shared.
560 */
561 45 shm_toc_estimate_chunk(&pcxt->estimator, (Size)vec_nbytes);
562
563 /* The leaf-refinement accumulator overlays the sample segment once the
564 * samples are dead (see the sample-region seam), so it needs no DSM chunk
565 * of its own; its tile capacity is bounded by the memory budget, by
566 * MaxAllocSize, and by the region it overlays. */
567
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 45 times.
 45 Size samp_sz = prism_dsm_samples_size(nparticipants, max_per_worker, dim);
568 45 uint64_t refine_cap_bytes =
569 45 Min((uint64_t)maintenance_work_mem * 1024, (uint64_t)MaxAllocSize);
570 45 if (refine_cap_bytes >
571 45 (uint64_t)samp_sz - offsetof(PrismDsmRefineAccum, sums))
572 refine_cap_bytes = (uint64_t)samp_sz -
573 offsetof(PrismDsmRefineAccum, sums);
574 45 uint32_t refine_tile_cap =
575
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 45 times.
 45 prism_refine_tile_leaves(nlist, dim, refine_cap_bytes);
576
577 45 shm_toc_estimate_chunk(
578 &pcxt->estimator, mul_size(sizeof(WalUsage), pcxt->nworkers));
579 45 shm_toc_estimate_chunk(
580 &pcxt->estimator, mul_size(sizeof(BufferUsage), pcxt->nworkers));
581
582 45 int querylen = 0;
583
1/2
✓ Branch 0 taken 45 times.
✗ Branch 1 not taken.
 45 if (debug_query_string)
584 {
585 45 querylen = strlen(debug_query_string);
586 45 shm_toc_estimate_chunk(&pcxt->estimator, querylen + 1);
587 }
588
589 /* nkeys: shared, barrier, centroids, km_workers, root_assign,
590 * sortshared, wal, buffer, global_mean + optionally query_text (samples
591 * and the subtree ring live in their own segments; the refine
592 * accumulator overlays the samples) */
593 45 int nkeys = 9;
594
1/2
✓ Branch 0 taken 45 times.
✗ Branch 1 not taken.
 45 if (debug_query_string)
595 45 nkeys++;
596 45 shm_toc_estimate_keys(&pcxt->estimator, nkeys);
597
598 /* Total bytes the leader is about to commit to the DSM segment (sum of all
599 * estimated chunks), for the planned-allocation introspection line. */
600 45 Size dsm_total = pcxt->estimator.space_for_chunks;
601
602 45 InitializeParallelDSM(pcxt);
603
604
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 45 times.
 45 if (pcxt->seg == NULL)
605 {
606 ✗ prism_pbuild_teardown(pcxt);
607 ✗ return false;
608 }
609
610 /* ---- Populate shared state ---- */
611 45 PrismBuildSharedPg *pg = shm_toc_allocate(pcxt->toc, est_shared);
612 45 PrismBuildShared *shared = &pg->base;
613 45 pg->heaprelid = RelationGetRelid(heap);
614 45 pg->indexrelid = RelationGetRelid(index);
615 45 pg->queryid = pgstat_get_my_query_id();
616 45 shared->concurrent = config->concurrent;
617 45 shared->dim = dim;
618 45 shared->metric = config->metric;
619 45 shared->nlist = nlist;
620 45 shared->fan_out = fan_out; /* resolved (auto if config 0) */
621 45 shared->subtree_slot_size = 0; /* leader sizes the ring post-assign */
622 45 shared->soar_lambda = config->soar_lambda;
623 45 shared->boundary_epsilon = config->boundary_epsilon;
624 45 shared->fastscan = config->fastscan;
625 45 shared->centroid_format = config->centroid_format;
626 45 shared->rabitq_seed = rabitq_seed;
627 45 shared->nparticipants = nparticipants;
628 45 shared->work_mem_kb = maintenance_work_mem;
629 45 shared->max_samples_per_worker = max_per_worker;
630 45 shared->km_max_iterations = 20;
631 45 shared->km_tolerance = 1e-4f;
632 45 shared->km_k = km_k;
633 45 shared->km_converged = false;
634 45 shared->refine_threshold = (uint32_t)prism_leaf_refine_threshold;
635 45 shared->refine = false; /* leader decides post-clustering */
636 45 shared->refine_tile_cap = refine_tile_cap;
637 /* Build routes for accuracy, not query speed (see PRISM_BUILD_CENTROID_*
638 * in posting_build.h): decouple from the query-tuned GUCs. */
639 45 shared->centroid_error_scale = PRISM_BUILD_CENTROID_ERROR_SCALE;
640 45 shared->centroid_beam_scale = PRISM_BUILD_CENTROID_BEAM_SCALE;
641 45 shared->fastscan_bits = prism_fastscan_bits;
642 45 SpinLockInit(&pg->mutex);
643
2/2
✓ Branch 0 taken 11520 times.
✓ Branch 1 taken 45 times.
 11565 for (int i = 0; i < PRISM_REFINE_LOCK_STRIPES; i++)
644 11520 SpinLockInit(&pg->accum_locks[i]);
645 45 shared->reltuples = 0.0;
646 45 shared->indtuples = 0.0;
647 45 shared->soar_dupes = 0.0;
648 /* The scan descriptor sits in the same chunk, right after the struct;
649 * est_shared sized it for both. The address is therefore inside the
650 * allocation while being past the end of pg->base, which is what the
651 * checker measures against. Mirrors nbtsort.c. */
652 /* coverity[OVERRUN] */
653 45 table_parallelscan_initialize(
654 heap, ParallelTableScanFromVsShared(shared), snapshot);
655 45 shm_toc_insert(pcxt->toc, PRISM_DSM_KEY_SHARED, shared);
656
657 /*
658 * Barrier for phase synchronization. It is a *dynamic* barrier (init with
659 * 0 parties): the leader and every launched worker join via BarrierAttach,
660 * so the party tracks the participants that actually start. A non-zero
661 * (static) party is wrong here — BarrierAttach forbids it
662 * (Assert(!static_party)), and a fixed planned count would deadlock when
663 * PostgreSQL launches fewer workers than planned. The leader attaches in
664 * do_parallel_build before launching workers.
665 */
666 45 Barrier *barrier = shm_toc_allocate(pcxt->toc, sizeof(Barrier));
667 45 BarrierInit(barrier, 0);
668 45 shm_toc_insert(pcxt->toc, PRISM_DSM_KEY_BARRIER, barrier);
669
670 /* Sample slots, in a dedicated DSM segment so the build's largest
671 * working set can be handed back (last detach destroys it) before the
672 * posting sort claims its own budget -- see the sample-region seam.
673 * Workers attach via the handle published in the shared state. */
674 45 dsm_segment *sample_seg = dsm_create(samp_sz, 0);
675 45 PrismDsmSamples *dsm_samples = dsm_segment_address(sample_seg);
676 /* Only the header + per-participant counts are read before being written;
677 * the sample data is filled by the sampling pass and read back bounded by
678 * those counts, so zeroing the (multi-GB) data region is wasted work. */
679 45 memset(dsm_samples,
680 0,
681 45 MAXALIGN(sizeof(PrismDsmSamples)) +
682 (size_t)nparticipants * sizeof(uint32_t));
683 45 dsm_samples->nparticipants = nparticipants;
684 45 dsm_samples->max_per_worker = max_per_worker;
685 45 dsm_samples->dim = dim;
686 45 pg->sample_handle = dsm_segment_handle(sample_seg);
687
688 /* Shared centroids + norms (root k-means, k=km_k) */
689 45 Size cent_sz = prism_dsm_centroids_size(km_k, dim);
690 45 char *centroids_base = shm_toc_allocate(pcxt->toc, cent_sz);
691 45 memset(centroids_base, 0, cent_sz);
692 45 shm_toc_insert(pcxt->toc, PRISM_DSM_KEY_CENTROIDS, centroids_base);
693 45 float *cents = prism_dsm_centroids(centroids_base);
694
695 /* Per-worker k-means accumulators */
696 45 Size km_sz = prism_dsm_km_workers_size(nparticipants, km_k, dim);
697 45 char *km_workers_base = shm_toc_allocate(pcxt->toc, km_sz);
698 45 memset(km_workers_base, 0, km_sz);
699 45 shm_toc_insert(pcxt->toc, PRISM_DSM_KEY_KM_WORKERS, km_workers_base);
700
701 /* Root assignment slots */
702 45 Size ra_sz = prism_dsm_root_assign_size(nparticipants, max_per_worker);
703 45 PrismDsmRootAssign *dsm_ra = shm_toc_allocate(pcxt->toc, ra_sz);
704 45 memset(dsm_ra, 0, ra_sz);
705 45 dsm_ra->nparticipants = nparticipants;
706 45 dsm_ra->max_per_worker = max_per_worker;
707 45 shm_toc_insert(pcxt->toc, PRISM_DSM_KEY_ROOT_ASSIGN, dsm_ra);
708
709 /* Shared coordinator for the cluster-keyed posting sort (sort seam). Sized
710 * for the planned participant count; the leader calls
711 * prism_pbuild_sort_shared_init with the actual launched count after
712 * launch.
713 */
714 45 Size sort_sz = prism_pbuild_sort_shared_size(nparticipants);
715 45 void *sortshared = shm_toc_allocate(pcxt->toc, sort_sz);
716 45 memset(sortshared, 0, sort_sz);
717 45 shm_toc_insert(pcxt->toc, PRISM_DSM_KEY_SORTSHARED, sortshared);
718
719 /* Page-backed routing region (filled by the leader before the tree-ready
720 * barrier): the global mean. The posting-head base is a scalar in
721 * PrismBuildShared, so there is no O(nlist) head array here. */
722 45 float *dsm_gmean = shm_toc_allocate(pcxt->toc, (Size)vec_nbytes);
723 45 memset(dsm_gmean, 0, (Size)vec_nbytes);
724 45 shm_toc_insert(pcxt->toc, PRISM_DSM_KEY_GLOBAL_MEAN, dsm_gmean);
725
726 45 WalUsage *walusage = shm_toc_allocate(
727 45 pcxt->toc, mul_size(sizeof(WalUsage), pcxt->nworkers));
728 45 memset(walusage, 0, mul_size(sizeof(WalUsage), pcxt->nworkers));
729 45 shm_toc_insert(pcxt->toc, PRISM_DSM_KEY_WAL_USAGE, walusage);
730
731 45 BufferUsage *bufferusage = shm_toc_allocate(
732 45 pcxt->toc, mul_size(sizeof(BufferUsage), pcxt->nworkers));
733 45 memset(bufferusage, 0, mul_size(sizeof(BufferUsage), pcxt->nworkers));
734 45 shm_toc_insert(pcxt->toc, PRISM_DSM_KEY_BUFFER_USAGE, bufferusage);
735
736
1/2
✓ Branch 0 taken 45 times.
✗ Branch 1 not taken.
 45 if (debug_query_string)
737 {
738 45 char *sq = shm_toc_allocate(pcxt->toc, querylen + 1);
739 45 memcpy(sq, debug_query_string, querylen + 1);
740 45 shm_toc_insert(pcxt->toc, PRISM_DSM_KEY_QUERY_TEXT, sq);
741 }
742
743 45 lead->pcxt = pcxt;
744 45 lead->shared = shared;
745 45 lead->barrier = barrier;
746 45 lead->dsm_samples = dsm_samples;
747 45 lead->sample_seg = sample_seg;
748 45 lead->centroids_base = centroids_base;
749 45 lead->cents = cents;
750 45 lead->km_workers_base = km_workers_base;
751 45 lead->dsm_ra = dsm_ra;
752 45 lead->walusage = walusage;
753 45 lead->bufferusage = bufferusage;
754 45 lead->nparticipants = nparticipants;
755 45 lead->km_k = km_k;
756 45 lead->max_per_worker = max_per_worker;
757 45 lead->dim = dim;
758 45 lead->nlist = nlist;
759 45 lead->rabitq_seed = rabitq_seed;
760 45 lead->fan_out = fan_out;
761 45 lead->dsm_total = dsm_total;
762 45 return true;
763 }
764
765 /*
766 * Accumulate one worker's tuple counts into the shared state under the lock.
767 * Back-end seam: only the spinlock is PG-specific (it lives in the derived
768 * struct); the standalone version locks a pthread mutex around the same adds.
769 */
770 void
771 84 prism_pbuild_worker_add_counts(
772 PrismBuildShared *shared,
773 double indtuples,
774 double soar_dupes,
775 double heap_tuples)
776 {
777 84 PrismBuildSharedPg *pg = (PrismBuildSharedPg *)shared;
778
779
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 84 times.
 84 SpinLockAcquire(&pg->mutex);
780 84 shared->indtuples += indtuples;
781 84 shared->soar_dupes += soar_dupes;
782 84 shared->reltuples += heap_tuples;
783 84 SpinLockRelease(&pg->mutex);
784 84 }
785
786 /*
787 * Sample-region seam (see parallel_build.h): the samples live in their own
788 * DSM segment; workers attach by the handle the leader published, and every
789 * participant detaches independently once the refine pass is done -- the
790 * last detach destroys the segment and returns the build's largest working
791 * set before the posting sort claims its budget.
792 */
793 PrismDsmSamples *
794 86 prism_pbuild_samples_attach(
795 shm_toc *toc, PrismBuildShared *shared, void **seg_out)
796 {
797 86 PrismBuildSharedPg *pg = (PrismBuildSharedPg *)shared;
798
799 86 (void)toc;
800 86 dsm_segment *seg = dsm_attach(pg->sample_handle);
801
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 86 times.
 86 if (seg == NULL)
802 ✗ ereport(ERROR,
803 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
804 errmsg("could not attach to prism sample segment")));
805 86 *seg_out = seg;
806 86 return (PrismDsmSamples *)dsm_segment_address(seg);
807 }
808
809 void
810 128 prism_pbuild_samples_release(PrismDsmSamples *samples, void *seg)
811 {
812 128 (void)samples;
813
1/2
✓ Branch 0 taken 128 times.
✗ Branch 1 not taken.
 128 if (seg != NULL)
814 128 dsm_detach((dsm_segment *)seg);
815 128 }
816
817 /*
818 * Subtree-ring seam: the ring of nparticipants subtree slots lives in its
819 * own DSM segment, created by the leader only after root assignment -- the
820 * per-child sample counts known then bound the slot far below the analytic
821 * worst case. The handle travels in shared; every participant detaches when
822 * the streaming pass is done, and the last detach frees the memory.
823 */
824 char *
825 20 prism_pbuild_subtree_ring_create(
826 PrismBuildShared *shared,
827 int nparticipants,
828 uint64_t slot_size,
829 void **seg_out)
830 {
831 20 PrismBuildSharedPg *pg = (PrismBuildSharedPg *)shared;
832
833 20 dsm_segment *seg = dsm_create(
834 prism_dsm_child_subtrees_size(nparticipants, slot_size), 0);
835 20 pg->subtree_ring_handle = dsm_segment_handle(seg);
836 20 shared->subtree_slot_size = slot_size;
837 20 *seg_out = seg;
838 20 return (char *)dsm_segment_address(seg);
839 }
840
841 char *
842 38 prism_pbuild_subtree_ring_attach(PrismBuildShared *shared, void **seg_out)
843 {
844 38 PrismBuildSharedPg *pg = (PrismBuildSharedPg *)shared;
845
846 38 dsm_segment *seg = dsm_attach(pg->subtree_ring_handle);
847
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 38 times.
 38 if (seg == NULL)
848 ✗ ereport(ERROR,
849 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
850 errmsg("could not attach to prism subtree segment")));
851 38 *seg_out = seg;
852 38 return (char *)dsm_segment_address(seg);
853 }
854
855 void
856 55 prism_pbuild_subtree_ring_release(void *seg)
857 {
858
1/2
✓ Branch 0 taken 55 times.
✗ Branch 1 not taken.
 55 if (seg != NULL)
859 55 dsm_detach((dsm_segment *)seg);
860 55 }
861
862 /*
863 * Exact-centroid seam: the exact centroid collection lives in its own DSM
864 * segment, created by the leader only after the streaming tree write (its
865 * size — a few MB — is known then). The handle travels in shared; every
866 * participant detaches when its page-backed routing is done, and the last
867 * detach frees the memory.
868 */
869 char *
870 43 prism_pbuild_exact_centroids_create(
871 PrismBuildShared *shared, uint64_t nbytes, void **seg_out)
872 {
873 43 PrismBuildSharedPg *pg = (PrismBuildSharedPg *)shared;
874
875 43 dsm_segment *seg = dsm_create(nbytes, 0);
876 43 pg->exact_centroids_handle = dsm_segment_handle(seg);
877 43 *seg_out = seg;
878 43 return (char *)dsm_segment_address(seg);
879 }
880
881 char *
882 84 prism_pbuild_exact_centroids_attach(PrismBuildShared *shared, void **seg_out)
883 {
884 84 PrismBuildSharedPg *pg = (PrismBuildSharedPg *)shared;
885
886 84 dsm_segment *seg = dsm_attach(pg->exact_centroids_handle);
887
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 84 times.
 84 if (seg == NULL)
888 ✗ ereport(ERROR,
889 (errcode(ERRCODE_OBJECT_NOT_IN_PREREQUISITE_STATE),
890 errmsg("could not attach to prism exact-centroid "
891 "segment")));
892 84 *seg_out = seg;
893 84 return (char *)dsm_segment_address(seg);
894 }
895
896 void
897 127 prism_pbuild_exact_centroids_release(void *seg)
898 {
899
1/2
✓ Branch 0 taken 127 times.
✗ Branch 1 not taken.
 127 if (seg != NULL)
900 127 dsm_detach((dsm_segment *)seg);
901 127 }
902
903 /*
904 * Re-initialize the parallel scan for the posting pass; the sampling pass
905 * consumed the first one. Back-end seam: the standalone version resets its
906 * work-stealing cursor.
907 */
908 void
909 45 prism_pbuild_rescan(Relation heap, PrismBuildShared *shared)
910 {
911 45 table_parallelscan_reinitialize(
912 heap, ParallelTableScanFromVsShared(shared));
913 45 }
914
915 /*
916 * Striped-lock seam for the shared leaf-refinement accumulator. Contention is
917 * low because rows spread across nleaves leaves into PRISM_REFINE_LOCK_STRIPES
918 * stripes.
919 */
920 void
921 12600 prism_pbuild_accum_lock(PrismBuildShared *shared, uint32_t stripe)
922 {
923 12600 PrismBuildSharedPg *pg = (PrismBuildSharedPg *)shared;
924
2/2
✓ Branch 1 taken 181 times.
✓ Branch 2 taken 12419 times.
 12600 SpinLockAcquire(&pg->accum_locks[stripe]);
925 12600 }
926
927 void
928 12600 prism_pbuild_accum_unlock(PrismBuildShared *shared, uint32_t stripe)
929 {
930 12600 PrismBuildSharedPg *pg = (PrismBuildSharedPg *)shared;
931 12600 SpinLockRelease(&pg->accum_locks[stripe]);
932 12600 }
933
934 /* ----------------------------------------------------------------
935 * Posting sort seam (PG back-end) — parallel tuplesort
936 *
937 * Each entry is sorted by a uint32 cluster id; the payload is a fixed-size
938 * opaque blob carried in a bytea column. Workers feed partial runs into the
939 * shared Sharedsort (in the build DSM); the leader merges and reads them back
940 * grouped by cluster. Memory is bounded by maintenance_work_mem (tuplesort
941 * spills past it). The standalone back-end provides the same-named seam over
942 * in-memory arrays.
943 * ---------------------------------------------------------------- */
944 struct PrismSorter
945 {
946 Tuplesortstate *ts;
947 TupleDesc tupdesc;
948 TupleTableSlot *slot;
949 SortCoordinate coord;
950 uint32_t entry_size;
951 bytea *payload; /* reusable scratch: VARHDRSZ + entry_size */
952 bool is_leader;
953 };
954
955 Size
956 90 prism_pbuild_sort_shared_size(int nparticipants)
957 {
958 90 return tuplesort_estimate_shared(nparticipants);
959 }
960
961 void
962 43 prism_pbuild_sort_shared_init(void *region, int nparticipants, void *seg)
963 {
964 43 tuplesort_initialize_shared(
965 (Sharedsort *)region, nparticipants, (dsm_segment *)seg);
966 43 }
967
968 PrismSorter *
969 278 prism_pbuild_sort_begin(
970 void *region,
971 void *seg,
972 int participant,
973 int nparticipants,
974 bool is_leader,
975 uint32_t entry_size,
976 int work_mem_kb)
977 {
978 278 (void)participant;
979 278 PrismSorter *s = palloc0(sizeof(PrismSorter));
980 278 s->entry_size = entry_size;
981 278 s->is_leader = is_leader;
982
983 278 s->tupdesc = CreateTemplateTupleDesc(2);
984 278 TupleDescInitEntry(s->tupdesc, 1, "cluster", INT4OID, -1, 0);
985 278 TupleDescInitEntry(s->tupdesc, 2, "payload", BYTEAOID, -1, 0);
986 #if PG_VERSION_NUM >= 190000
987 TupleDescFinalize(s->tupdesc);
988 #endif
989
990 /* region == NULL: a plain, non-parallel sort (the serial build) — no
991 * coordinate, no attach. Otherwise a parallel participant: a leader
992 * merging runs, or a worker producing one (which attaches to the shared
993 * fileset). */
994
2/2
✓ Branch 0 taken 127 times.
✓ Branch 1 taken 151 times.
 278 if (region != NULL)
995 {
996 127 s->coord = palloc0(sizeof(SortCoordinateData));
997 127 s->coord->sharedsort = (Sharedsort *)region;
998
2/2
✓ Branch 0 taken 43 times.
✓ Branch 1 taken 84 times.
 127 if (is_leader)
999 {
1000 43 s->coord->isWorker = false;
1001 43 s->coord->nParticipants = nparticipants;
1002 }
1003 else
1004 {
1005 84 s->coord->isWorker = true;
1006 84 s->coord->nParticipants = -1;
1007 }
1008 }
1009
1010 278 AttrNumber attNums[1] = {1};
1011 278 Oid sortOps[1] = {Int4LessOperator};
1012 278 Oid sortColls[1] = {InvalidOid};
1013 278 bool nullsF[1] = {false};
1014 278 s->ts = tuplesort_begin_heap(
1015 s->tupdesc,
1016 1,
1017 attNums,
1018 sortOps,
1019 sortColls,
1020 nullsF,
1021 work_mem_kb,
1022 s->coord,
1023 TUPLESORT_NONE);
1024
1025 /* Workers attach to the shared fileset; the leader holds it via the
1026 * backend's dsm reference and must not attach (per tuplesort.h). */
1027
2/2
✓ Branch 0 taken 84 times.
✓ Branch 1 taken 194 times.
 278 if (region != NULL && !is_leader)
1028 84 tuplesort_attach_shared((Sharedsort *)region, (dsm_segment *)seg);
1029
1030 278 s->slot = MakeSingleTupleTableSlot(s->tupdesc, &TTSOpsMinimalTuple);
1031 278 s->payload = (bytea *)palloc(VARHDRSZ + entry_size);
1032 278 SET_VARSIZE(s->payload, VARHDRSZ + entry_size);
1033 278 return s;
1034 }
1035
1036 void
1037 393420 prism_pbuild_sort_put(PrismSorter *s, uint32_t cluster, const void *entry)
1038 {
1039 393420 memcpy(VARDATA(s->payload), entry, s->entry_size);
1040 393420 ExecClearTuple(s->slot);
1041 393420 s->slot->tts_values[0] = Int32GetDatum((int32)cluster);
1042 393420 s->slot->tts_isnull[0] = false;
1043 393420 s->slot->tts_values[1] = PointerGetDatum(s->payload);
1044 393420 s->slot->tts_isnull[1] = false;
1045 393420 ExecStoreVirtualTuple(s->slot);
1046 393420 tuplesort_puttupleslot(s->ts, s->slot);
1047 393420 }
1048
1049 void
1050 278 prism_pbuild_sort_performsort(PrismSorter *s)
1051 {
1052 278 tuplesort_performsort(s->ts);
1053 278 }
1054
1055 bool
1056 393614 prism_pbuild_sort_getnext(
1057 PrismSorter *s, uint32_t *cluster, const void **entry)
1058 {
1059 393614 bool isnull;
1060
2/2
✓ Branch 1 taken 393420 times.
✓ Branch 2 taken 194 times.
 393614 if (!tuplesort_gettupleslot(s->ts, true, false, s->slot, NULL))
1061 return false;
1062 393420 *cluster = (uint32_t)DatumGetInt32(slot_getattr(s->slot, 1, &isnull));
1063 393420 bytea *pl = DatumGetByteaPP(slot_getattr(s->slot, 2, &isnull));
1064
2/2
✓ Branch 0 taken 391593 times.
✓ Branch 1 taken 1827 times.
 393420 *entry = VARDATA_ANY(pl);
1065 393420 return true;
1066 }
1067
1068 void
1069 278 prism_pbuild_sort_end(PrismSorter *s)
1070 {
1071 278 tuplesort_end(s->ts);
1072 278 ExecDropSingleTupleTableSlot(s->slot);
1073 278 FreeTupleDesc(s->tupdesc);
1074
2/2
✓ Branch 0 taken 127 times.
✓ Branch 1 taken 151 times.
 278 if (s->coord != NULL) /* NULL for a non-parallel (serial) sort */
1075 127 pfree(s->coord);
1076 278 pfree(s->payload);
1077 278 pfree(s);
1078 278 }
1079