Skip to content

Commit 3626a0b

Browse files
author
Artur Zakirov
committed
Added macroses to handle generic XLOG
1 parent 26f88a0 commit 3626a0b

File tree

6 files changed

+122
-85
lines changed

6 files changed

+122
-85
lines changed

‎rum.h

Lines changed: 19 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -296,6 +296,21 @@ typedef struct
296296
#define RumPageGetIndexes(page) \
297297
((RumDataLeafItemIndex *)(RumDataPageGetData(page) + RumDataPageSize))
298298

299+
/*
300+
* Macros to handle generic XLOG
301+
*/
302+
#define RumGenericXLogStart(index, isbuild) \
303+
(!(isbuild) ? GenericXLogStart(index) : NULL)
304+
305+
#define RumGenericXLogRegisterBuffer(state, buffer, flags, isbuild) \
306+
(!(isbuild) ? GenericXLogRegisterBuffer(state, buffer, flags) : \
307+
BufferGetPage(buffer))
308+
309+
#define RumGenericXLogFinish(state, isbuild) \
310+
(!(isbuild) ? GenericXLogFinish(state) : 0)
311+
312+
#define RumGenericXLogAbort(state, isbuild) \
313+
(!(isbuild) ? GenericXLogAbort(state) : (void) 0)
299314

300315
/*
301316
* Storage type for RUM's reloptions
@@ -377,6 +392,8 @@ typedef struct RumConfig
377392
Oid addInfoTypeOid;
378393
} RumConfig;
379394

395+
396+
380397
/* rumutil.c */
381398
extern bytea *rumoptions(Datum reloptions, bool validate);
382399
extern Datum rumhandler(PG_FUNCTION_ARGS);
@@ -401,7 +418,8 @@ extern Datum rumtuple_get_key(RumState * rumstate, IndexTuple tuple,
401418
RumNullCategory * category);
402419

403420
extern void rumGetStats(Relation index, GinStatsData *stats);
404-
extern void rumUpdateStats(Relation index, const GinStatsData *stats);
421+
extern void rumUpdateStats(Relation index, const GinStatsData *stats,
422+
bool isBuild);
405423

406424
/* ruminsert.c */
407425
extern IndexBuildResult *rumbuild(Relation heap, Relation index,

‎rumbtree.c

Lines changed: 20 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -385,11 +385,12 @@ rumInsertValue(Relation index, RumBtree btree, RumBtreeStack * stack,
385385

386386
if (btree->isEnoughSpace(btree, stack->buffer, stack->off))
387387
{
388-
state = GenericXLogStart(index);
389-
page = GenericXLogRegisterBuffer(state, stack->buffer, 0);
388+
state = RumGenericXLogStart(index, btree->rumstate->isBuild);
389+
page = RumGenericXLogRegisterBuffer(state, stack->buffer, 0,
390+
btree->rumstate->isBuild);
390391

391392
btree->placeToPage(btree, page, stack->off);
392-
GenericXLogFinish(state);
393+
RumGenericXLogFinish(state, btree->rumstate->isBuild);
393394

394395
LockBuffer(stack->buffer, RUM_UNLOCK);
395396
freeRumBtreeStack(stack);
@@ -416,11 +417,13 @@ rumInsertValue(Relation index, RumBtree btree, RumBtreeStack * stack,
416417
{
417418
Buffer lbuffer;
418419

419-
state = GenericXLogStart(index);
420+
state = RumGenericXLogStart(index, btree->rumstate->isBuild);
420421

421-
page = GenericXLogRegisterBuffer(state, stack->buffer, 0);
422-
rpage = GenericXLogRegisterBuffer(state, rbuffer,
423-
GENERIC_XLOG_FULL_IMAGE);
422+
page = RumGenericXLogRegisterBuffer(state, stack->buffer, 0,
423+
btree->rumstate->isBuild);
424+
rpage = RumGenericXLogRegisterBuffer(state, rbuffer,
425+
GENERIC_XLOG_FULL_IMAGE,
426+
btree->rumstate->isBuild);
424427

425428
/*
426429
* newlpage is a pointer to memory page, it doesn't associate
@@ -434,8 +437,9 @@ rumInsertValue(Relation index, RumBtree btree, RumBtreeStack * stack,
434437
* pointer on root to left and right page
435438
*/
436439
lbuffer = RumNewBuffer(btree->index);
437-
lpage = GenericXLogRegisterBuffer(state, lbuffer,
438-
GENERIC_XLOG_FULL_IMAGE);
440+
lpage = RumGenericXLogRegisterBuffer(state, lbuffer,
441+
GENERIC_XLOG_FULL_IMAGE,
442+
btree->rumstate->isBuild);
439443

440444
RumPageGetOpaque(rpage)->rightlink = InvalidBlockNumber;
441445
RumPageGetOpaque(newlpage)->leftlink = InvalidBlockNumber;
@@ -448,7 +452,7 @@ rumInsertValue(Relation index, RumBtree btree, RumBtreeStack * stack,
448452
btree->fillRoot(btree, stack->buffer, lbuffer, rbuffer,
449453
page, lpage, rpage);
450454

451-
GenericXLogFinish(state);
455+
RumGenericXLogFinish(state, btree->rumstate->isBuild);
452456

453457
UnlockReleaseBuffer(rbuffer);
454458
UnlockReleaseBuffer(lbuffer);
@@ -471,10 +475,12 @@ rumInsertValue(Relation index, RumBtree btree, RumBtreeStack * stack,
471475
{
472476
/* split non-root page */
473477

474-
state = GenericXLogStart(index);
478+
state = RumGenericXLogStart(index, btree->rumstate->isBuild);
475479

476-
lpage = GenericXLogRegisterBuffer(state, stack->buffer, 0);
477-
rpage = GenericXLogRegisterBuffer(state, rbuffer, 0);
480+
lpage = RumGenericXLogRegisterBuffer(state, stack->buffer, 0,
481+
btree->rumstate->isBuild);
482+
rpage = RumGenericXLogRegisterBuffer(state, rbuffer, 0,
483+
btree->rumstate->isBuild);
478484

479485
/*
480486
* newlpage is a pointer to memory page, it doesn't associate
@@ -490,7 +496,7 @@ rumInsertValue(Relation index, RumBtree btree, RumBtreeStack * stack,
490496

491497
PageRestoreTempPage(newlpage, lpage);
492498

493-
GenericXLogFinish(state);
499+
RumGenericXLogFinish(state, btree->rumstate->isBuild);
494500

495501
UnlockReleaseBuffer(rbuffer);
496502
}

‎rumfast.c

Lines changed: 44 additions & 39 deletions
Original file line numberDiff line numberDiff line change
@@ -47,33 +47,33 @@ typedef struct KeyArray
4747
*
4848
* Returns amount of free space left on the page.
4949
*/
50-
static int32
51-
writeListPage(Relation index, Buffer buffer,
52-
IndexTuple *tuples, int32 ntuples, BlockNumber rightlink)
50+
static uint32
51+
writeListPage(RumState *rumstate, Buffer buffer,
52+
IndexTuple *tuples, uint32 ntuples, BlockNumber rightlink)
5353
{
5454
Page page;
55-
int32 i,
55+
uint32 i,
5656
freesize;
5757
OffsetNumber l,
5858
off;
5959
GenericXLogState *state;
6060

61-
state = GenericXLogStart(index);
61+
state = RumGenericXLogStart(rumstate->index, rumstate->isBuild);
6262

63-
page = GenericXLogRegisterBuffer(state, buffer, 0);
63+
page = RumGenericXLogRegisterBuffer(state, buffer, 0, rumstate->isBuild);
6464
RumInitPage(page, RUM_LIST, BufferGetPageSize(buffer));
6565

6666
off = FirstOffsetNumber;
6767

6868
for (i = 0; i < ntuples; i++)
6969
{
70-
int this_size = IndexTupleSize(tuples[i]);
70+
Size this_size = IndexTupleSize(tuples[i]);
7171

7272
l = PageAddItem(page, (Item) tuples[i], this_size, off, false, false);
7373

7474
if (l == InvalidOffsetNumber)
7575
elog(ERROR, "failed to add item to index page in \"%s\"",
76-
RelationGetRelationName(index));
76+
RelationGetRelationName(rumstate->index));
7777

7878
off++;
7979
}
@@ -97,22 +97,22 @@ writeListPage(Relation index, Buffer buffer,
9797

9898
/* get free space before releasing buffer */
9999
freesize = PageGetExactFreeSpace(page);
100-
GenericXLogFinish(state);
100+
RumGenericXLogFinish(state, rumstate->isBuild);
101101
UnlockReleaseBuffer(buffer);
102102

103103
return freesize;
104104
}
105105

106106
static void
107-
makeSublist(Relation index, IndexTuple *tuples, int32 ntuples,
107+
makeSublist(RumState *rumstate, IndexTuple *tuples, uint32 ntuples,
108108
RumMetaPageData * res)
109109
{
110110
Buffer curBuffer = InvalidBuffer;
111111
Buffer prevBuffer = InvalidBuffer;
112-
int i,
113-
size = 0,
112+
uint32 i,
113+
startTuple = 0;
114+
uint64 size = 0,
114115
tupsize;
115-
int startTuple = 0;
116116

117117
Assert(ntuples > 0);
118118

@@ -123,12 +123,12 @@ makeSublist(Relation index, IndexTuple *tuples, int32 ntuples,
123123
{
124124
if (curBuffer == InvalidBuffer)
125125
{
126-
curBuffer = RumNewBuffer(index);
126+
curBuffer = RumNewBuffer(rumstate->index);
127127

128128
if (prevBuffer != InvalidBuffer)
129129
{
130130
res->nPendingPages++;
131-
writeListPage(index, prevBuffer,
131+
writeListPage(rumstate, prevBuffer,
132132
tuples + startTuple,
133133
i - startTuple,
134134
BufferGetBlockNumber(curBuffer));
@@ -161,7 +161,7 @@ makeSublist(Relation index, IndexTuple *tuples, int32 ntuples,
161161
* Write last page
162162
*/
163163
res->tail = BufferGetBlockNumber(curBuffer);
164-
res->tailFreeSize = writeListPage(index, curBuffer,
164+
res->tailFreeSize = writeListPage(rumstate, curBuffer,
165165
tuples + startTuple,
166166
ntuples - startTuple,
167167
InvalidBlockNumber);
@@ -193,7 +193,7 @@ rumHeapTupleFastInsert(RumState * rumstate, RumTupleCollector * collector)
193193
if (collector->ntuples == 0)
194194
return;
195195

196-
state = GenericXLogStart(rumstate->index);
196+
state = RumGenericXLogStart(rumstate->index, rumstate->isBuild);
197197
metabuffer = ReadBuffer(index, RUM_METAPAGE_BLKNO);
198198

199199
if (collector->sumsize + collector->ntuples * sizeof(ItemIdData) > RumListPageSize)
@@ -230,13 +230,14 @@ rumHeapTupleFastInsert(RumState * rumstate, RumTupleCollector * collector)
230230
RumMetaPageData sublist;
231231

232232
memset(&sublist, 0, sizeof(RumMetaPageData));
233-
makeSublist(index, collector->tuples, collector->ntuples, &sublist);
233+
makeSublist(rumstate, collector->tuples, collector->ntuples, &sublist);
234234

235235
/*
236236
* metapage was unlocked, see above
237237
*/
238238
LockBuffer(metabuffer, RUM_EXCLUSIVE);
239-
metapage = GenericXLogRegisterBuffer(state, metabuffer, 0);
239+
metapage = RumGenericXLogRegisterBuffer(state, metabuffer, 0,
240+
rumstate->isBuild);
240241
metadata = RumPageGetMeta(metapage);
241242

242243
if (metadata->head == InvalidBlockNumber)
@@ -259,7 +260,8 @@ rumHeapTupleFastInsert(RumState * rumstate, RumTupleCollector * collector)
259260

260261
buffer = ReadBuffer(index, metadata->tail);
261262
LockBuffer(buffer, RUM_EXCLUSIVE);
262-
page = GenericXLogRegisterBuffer(state, buffer, 0);
263+
page = RumGenericXLogRegisterBuffer(state, buffer, 0,
264+
rumstate->isBuild);
263265

264266
Assert(RumPageGetOpaque(page)->rightlink == InvalidBlockNumber);
265267

@@ -279,15 +281,16 @@ rumHeapTupleFastInsert(RumState * rumstate, RumTupleCollector * collector)
279281
*/
280282
OffsetNumber l,
281283
off;
282-
int i,
283-
tupsize;
284+
uint32 i;
285+
Size tupsize;
284286

285-
metapage = GenericXLogRegisterBuffer(state, metabuffer, 0);
287+
metapage = RumGenericXLogRegisterBuffer(state, metabuffer, 0,
288+
rumstate->isBuild);
286289
metadata = RumPageGetMeta(metapage);
287290

288291
buffer = ReadBuffer(index, metadata->tail);
289292
LockBuffer(buffer, RUM_EXCLUSIVE);
290-
page = GenericXLogRegisterBuffer(state, buffer, 0);
293+
page = RumGenericXLogRegisterBuffer(state, buffer, 0, rumstate->isBuild);
291294

292295
off = (PageIsEmpty(page)) ? FirstOffsetNumber :
293296
OffsetNumberNext(PageGetMaxOffsetNumber(page));
@@ -306,7 +309,7 @@ rumHeapTupleFastInsert(RumState * rumstate, RumTupleCollector * collector)
306309

307310
if (l == InvalidOffsetNumber)
308311
{
309-
GenericXLogAbort(state);
312+
RumGenericXLogAbort(state, rumstate->isBuild);
310313
elog(ERROR, "failed to add item to index page in \"%s\"",
311314
RelationGetRelationName(index));
312315
}
@@ -329,7 +332,7 @@ rumHeapTupleFastInsert(RumState * rumstate, RumTupleCollector * collector)
329332
if (metadata->nPendingPages * RUM_PAGE_FREESIZE > work_mem * 1024L)
330333
needCleanup = true;
331334

332-
GenericXLogFinish(state);
335+
RumGenericXLogFinish(state, rumstate->isBuild);
333336

334337
if (buffer != InvalidBuffer)
335338
UnlockReleaseBuffer(buffer);
@@ -349,7 +352,7 @@ RumFastFormTuple(RumState * rumstate,
349352
Datum datums[3];
350353
bool isnull[3];
351354
IndexTuple itup;
352-
uint32 newsize;
355+
Size newsize;
353356

354357
/* Build the basic tuple: optional column number, plus key datum */
355358

@@ -380,7 +383,7 @@ RumFastFormTuple(RumState * rumstate,
380383

381384
if (category != RUM_CAT_NORM_KEY)
382385
{
383-
uint32 minsize;
386+
Size minsize;
384387

385388
Assert(IndexTupleHasNulls(itup));
386389
minsize = IndexInfoFindDataOffset(itup->t_info) +
@@ -497,17 +500,18 @@ rumHeapTupleFastCollect(RumState * rumstate,
497500
* (if so, we can just abandon our own efforts)
498501
*/
499502
static bool
500-
shiftList(Relation index, Buffer metabuffer, BlockNumber newHead,
503+
shiftList(RumState *rumstate, Buffer metabuffer, BlockNumber newHead,
501504
IndexBulkDeleteResult *stats)
502505
{
503506
Page metapage;
504507
RumMetaPageData *metadata;
505508
BlockNumber blknoToDelete;
506509
GenericXLogState *metastate;
507510

508-
metastate = GenericXLogStart(index);
509-
metapage = GenericXLogRegisterBuffer(metastate, metabuffer,
510-
GENERIC_XLOG_FULL_IMAGE);
511+
metastate = RumGenericXLogStart(rumstate->index, rumstate->isBuild);
512+
metapage = RumGenericXLogRegisterBuffer(metastate, metabuffer,
513+
GENERIC_XLOG_FULL_IMAGE,
514+
rumstate->isBuild);
511515
metadata = RumPageGetMeta(metapage);
512516
blknoToDelete = metadata->head;
513517

@@ -522,7 +526,7 @@ shiftList(Relation index, Buffer metabuffer, BlockNumber newHead,
522526

523527
while (nDeleted < RUM_NDELETE_AT_ONCE && blknoToDelete != newHead)
524528
{
525-
buffers[nDeleted] = ReadBuffer(index, blknoToDelete);
529+
buffers[nDeleted] = ReadBuffer(rumstate->index, blknoToDelete);
526530
LockBuffer(buffers[nDeleted], RUM_EXCLUSIVE);
527531

528532
page = BufferGetPage(buffers[nDeleted]);
@@ -531,7 +535,7 @@ shiftList(Relation index, Buffer metabuffer, BlockNumber newHead,
531535

532536
if (RumPageIsDeleted(page))
533537
{
534-
GenericXLogAbort(metastate);
538+
RumGenericXLogAbort(metastate, rumstate->isBuild);
535539
/* concurrent cleanup process is detected */
536540
for (i = 0; i < nDeleted; i++)
537541
UnlockReleaseBuffer(buffers[i]);
@@ -565,18 +569,19 @@ shiftList(Relation index, Buffer metabuffer, BlockNumber newHead,
565569

566570
for (i = 0; i < nDeleted; i++)
567571
{
568-
state = GenericXLogStart(index);
569-
page = GenericXLogRegisterBuffer(state, buffers[i], 0);
572+
state = RumGenericXLogStart(rumstate->index, rumstate->isBuild);
573+
page = RumGenericXLogRegisterBuffer(state, buffers[i], 0,
574+
rumstate->isBuild);
570575

571576
RumPageGetOpaque(page)->flags = RUM_DELETED;
572-
GenericXLogFinish(state);
577+
RumGenericXLogFinish(state, rumstate->isBuild);
573578
}
574579

575580
for (i = 0; i < nDeleted; i++)
576581
UnlockReleaseBuffer(buffers[i]);
577582
} while (blknoToDelete != newHead);
578583

579-
GenericXLogFinish(metastate);
584+
RumGenericXLogFinish(metastate, rumstate->isBuild);
580585

581586
return false;
582587
}
@@ -882,7 +887,7 @@ rumInsertCleanup(RumState * rumstate,
882887
* remove read pages from pending list, at this point all content
883888
* of read pages is in regular structure
884889
*/
885-
if (shiftList(index, metabuffer, blkno, stats))
890+
if (shiftList(rumstate, metabuffer, blkno, stats))
886891
{
887892
/* another cleanup process is running concurrently */
888893
LockBuffer(metabuffer, RUM_UNLOCK);

0 commit comments

Comments
 (0)