1
/*-------------------------------------------------------------------------
4
* interface routines for the postgres rtree indexed access method.
6
* Portions Copyright (c) 1996-2005, PostgreSQL Global Development Group
7
* Portions Copyright (c) 1994, Regents of the University of California
11
* $PostgreSQL: pgsql/src/backend/access/rtree/rtree.c,v 1.85.4.1 2005-01-24 02:47:52 tgl Exp $
13
*-------------------------------------------------------------------------
18
#include "access/genam.h"
19
#include "access/heapam.h"
20
#include "access/rtree.h"
21
#include "access/xlogutils.h"
22
#include "catalog/index.h"
23
#include "commands/vacuum.h"
24
#include "executor/executor.h"
25
#include "miscadmin.h"
29
* XXX We assume that all datatypes indexable in rtrees are pass-by-reference.
30
* To fix this, you'd need to improve the IndexTupleGetDatum() macro, and
31
* do something with the various datum-pfreeing code. However, it's not that
32
* unreasonable an assumption in practice.
34
#define IndexTupleGetDatum(itup) \
35
PointerGetDatum(((char *) (itup)) + sizeof(IndexTupleData))
38
* Space-allocation macros. Note we count the item's line pointer in its size.
40
#define RTPageAvailSpace \
41
(BLCKSZ - (sizeof(PageHeaderData) - sizeof(ItemIdData)) \
42
- MAXALIGN(sizeof(RTreePageOpaqueData)))
43
#define IndexTupleTotalSize(itup) \
44
(MAXALIGN(IndexTupleSize(itup)) + sizeof(ItemIdData))
45
#define IndexTupleAttSize(itup) \
46
(IndexTupleSize(itup) - sizeof(IndexTupleData))
48
/* results of rtpicksplit() */
49
typedef struct SPLITVEC
51
OffsetNumber *spl_left;
54
OffsetNumber *spl_right;
59
/* for sorting tuples by cost, for picking split */
60
typedef struct SPLITCOST
62
OffsetNumber offset_number;
63
float cost_differential;
67
typedef struct RTSTATE
69
FmgrInfo unionFn; /* union function */
70
FmgrInfo sizeFn; /* size function */
71
FmgrInfo interFn; /* intersection function */
74
/* Working state for rtbuild and its callback */
81
/* non-export function prototypes */
82
static void rtbuildCallback(Relation index,
88
static InsertIndexResult rtdoinsert(Relation r, IndexTuple itup,
90
static void rttighten(Relation r, RTSTACK *stk, Datum datum, int att_size,
92
static InsertIndexResult rtdosplit(Relation r, Buffer buffer, RTSTACK *stack,
93
IndexTuple itup, RTSTATE *rtstate);
94
static void rtintinsert(Relation r, RTSTACK *stk, IndexTuple ltup,
95
IndexTuple rtup, RTSTATE *rtstate);
96
static void rtnewroot(Relation r, IndexTuple lt, IndexTuple rt);
97
static void rtpicksplit(Relation r, Page page, SPLITVEC *v, IndexTuple itup,
99
static void RTInitBuffer(Buffer b, uint32 f);
100
static OffsetNumber choose(Relation r, Page p, IndexTuple it,
102
static int nospace(Page p, IndexTuple it);
103
static void initRtstate(RTSTATE *rtstate, Relation index);
104
static int qsort_comp_splitcost(const void *a, const void *b);
108
* routine to build an index. Basically calls insert over and over
111
rtbuild(PG_FUNCTION_ARGS)
113
Relation heap = (Relation) PG_GETARG_POINTER(0);
114
Relation index = (Relation) PG_GETARG_POINTER(1);
115
IndexInfo *indexInfo = (IndexInfo *) PG_GETARG_POINTER(2);
117
RTBuildState buildstate;
120
/* no locking is needed */
122
initRtstate(&buildstate.rtState, index);
125
* We expect to be called exactly once for any index relation. If
126
* that's not the case, big trouble's what we have.
128
if (RelationGetNumberOfBlocks(index) != 0)
129
elog(ERROR, "index \"%s\" already contains data",
130
RelationGetRelationName(index));
132
/* initialize the root page */
133
buffer = ReadBuffer(index, P_NEW);
134
RTInitBuffer(buffer, F_LEAF);
137
/* build the index */
138
buildstate.indtuples = 0;
140
/* do the heap scan */
141
reltuples = IndexBuildHeapScan(heap, index, indexInfo,
142
rtbuildCallback, (void *) &buildstate);
144
/* okay, all heap tuples are indexed */
147
* Since we just counted the tuples in the heap, we update its stats
148
* in pg_class to guarantee that the planner takes advantage of the
149
* index we just created. But, only update statistics during normal
150
* index definitions, not for indices on system catalogs created
151
* during bootstrap processing. We must close the relations before
152
* updating statistics to guarantee that the relcache entries are
153
* flushed when we increment the command counter in UpdateStats(). But
154
* we do not release any locks on the relations; those will be held
155
* until end of transaction.
157
if (IsNormalProcessingMode())
159
Oid hrelid = RelationGetRelid(heap);
160
Oid irelid = RelationGetRelid(index);
162
heap_close(heap, NoLock);
164
UpdateStats(hrelid, reltuples);
165
UpdateStats(irelid, buildstate.indtuples);
172
* Per-tuple callback from IndexBuildHeapScan
175
rtbuildCallback(Relation index,
182
RTBuildState *buildstate = (RTBuildState *) state;
184
InsertIndexResult res;
186
/* form an index tuple and point it at the heap tuple */
187
itup = index_formtuple(RelationGetDescr(index), attdata, nulls);
188
itup->t_tid = htup->t_self;
190
/* rtree indexes don't index nulls, see notes in rtinsert */
191
if (IndexTupleHasNulls(itup))
198
* Since we already have the index relation locked, we call rtdoinsert
199
* directly. Normal access method calls dispatch through rtinsert,
200
* which locks the relation for write. This is the right thing to do
201
* if you're inserting single tups, but not when you're initializing
202
* the whole index at once.
204
res = rtdoinsert(index, itup, &buildstate->rtState);
209
buildstate->indtuples += 1;
215
* rtinsert -- wrapper for rtree tuple insertion.
217
* This is the public interface routine for tuple insertion in rtrees.
218
* It doesn't do any work; just locks the relation and passes the buck.
221
rtinsert(PG_FUNCTION_ARGS)
223
Relation r = (Relation) PG_GETARG_POINTER(0);
224
Datum *datum = (Datum *) PG_GETARG_POINTER(1);
225
char *nulls = (char *) PG_GETARG_POINTER(2);
226
ItemPointer ht_ctid = (ItemPointer) PG_GETARG_POINTER(3);
229
Relation heapRel = (Relation) PG_GETARG_POINTER(4);
230
bool checkUnique = PG_GETARG_BOOL(5);
232
InsertIndexResult res;
236
/* generate an index tuple */
237
itup = index_formtuple(RelationGetDescr(r), datum, nulls);
238
itup->t_tid = *ht_ctid;
241
* Currently, rtrees do not support indexing NULLs; considerable
242
* infrastructure work would have to be done to do anything reasonable
245
if (IndexTupleHasNulls(itup))
248
PG_RETURN_POINTER(NULL);
251
initRtstate(&rtState, r);
254
* Since rtree is not marked "amconcurrent" in pg_am, caller should
255
* have acquired exclusive lock on index relation. We need no locking
259
res = rtdoinsert(r, itup, &rtState);
261
PG_RETURN_POINTER(res);
264
static InsertIndexResult
265
rtdoinsert(Relation r, IndexTuple itup, RTSTATE *rtstate)
273
InsertIndexResult res;
274
RTreePageOpaque opaque;
278
buffer = InvalidBuffer;
283
/* let go of current buffer before getting next */
284
if (buffer != InvalidBuffer)
285
ReleaseBuffer(buffer);
287
/* get next buffer */
288
buffer = ReadBuffer(r, blk);
289
page = (Page) BufferGetPage(buffer);
291
opaque = (RTreePageOpaque) PageGetSpecialPointer(page);
292
if (!(opaque->flags & F_LEAF))
297
n = (RTSTACK *) palloc(sizeof(RTSTACK));
298
n->rts_parent = stack;
300
n->rts_child = choose(r, page, itup, rtstate);
303
iid = PageGetItemId(page, n->rts_child);
304
which = (IndexTuple) PageGetItem(page, iid);
305
blk = ItemPointerGetBlockNumber(&(which->t_tid));
307
} while (!(opaque->flags & F_LEAF));
309
if (nospace(page, itup))
311
/* need to do a split */
312
res = rtdosplit(r, buffer, stack, itup, rtstate);
314
WriteBuffer(buffer); /* don't forget to release buffer! */
318
/* add the item and write the buffer */
319
if (PageIsEmpty(page))
321
l = PageAddItem(page, (Item) itup, IndexTupleSize(itup),
327
l = PageAddItem(page, (Item) itup, IndexTupleSize(itup),
328
OffsetNumberNext(PageGetMaxOffsetNumber(page)),
331
if (l == InvalidOffsetNumber)
332
elog(ERROR, "failed to add index item to \"%s\"",
333
RelationGetRelationName(r));
337
datum = IndexTupleGetDatum(itup);
339
/* now expand the page boundary in the parent to include the new child */
340
rttighten(r, stack, datum, IndexTupleAttSize(itup), rtstate);
343
/* build and return an InsertIndexResult for this insertion */
344
res = (InsertIndexResult) palloc(sizeof(InsertIndexResultData));
345
ItemPointerSet(&(res->pointerData), blk, l);
351
rttighten(Relation r,
367
b = ReadBuffer(r, stk->rts_blk);
368
p = BufferGetPage(b);
370
oldud = IndexTupleGetDatum(PageGetItem(p,
371
PageGetItemId(p, stk->rts_child)));
373
FunctionCall2(&rtstate->sizeFn, oldud,
374
PointerGetDatum(&old_size));
376
datum = FunctionCall2(&rtstate->unionFn, oldud, datum);
378
FunctionCall2(&rtstate->sizeFn, datum,
379
PointerGetDatum(&newd_size));
382
* If newd_size == 0 we have degenerate rectangles, so we don't know
383
* if there was any change, so we have to assume there was.
385
if ((newd_size == 0) || (newd_size != old_size))
387
TupleDesc td = RelationGetDescr(r);
389
if (td->attrs[0]->attlen < 0)
392
* This is an internal page, so 'oldud' had better be a union
393
* (constant-length) key, too. (See comment below.)
395
Assert(VARSIZE(DatumGetPointer(datum)) ==
396
VARSIZE(DatumGetPointer(oldud)));
397
memmove(DatumGetPointer(oldud), DatumGetPointer(datum),
398
VARSIZE(DatumGetPointer(datum)));
402
memmove(DatumGetPointer(oldud), DatumGetPointer(datum),
408
* The user may be defining an index on variable-sized data (like
409
* polygons). If so, we need to get a constant-sized datum for
410
* insertion on the internal page. We do this by calling the
411
* union proc, which is required to return a rectangle.
413
tdatum = FunctionCall2(&rtstate->unionFn, datum, datum);
415
rttighten(r, stk->rts_parent, tdatum, att_size, rtstate);
416
pfree(DatumGetPointer(tdatum));
420
pfree(DatumGetPointer(datum));
424
* rtdosplit -- split a page in the tree.
426
* rtpicksplit does the interesting work of choosing the split.
427
* This routine just does the bit-pushing.
429
static InsertIndexResult
430
rtdosplit(Relation r,
447
OffsetNumber leftoff,
451
BlockNumber bufblock;
452
RTreePageOpaque opaque;
454
InsertIndexResult res;
457
OffsetNumber *spl_left,
461
OffsetNumber newitemoff;
463
p = (Page) BufferGetPage(buffer);
464
opaque = (RTreePageOpaque) PageGetSpecialPointer(p);
466
rtpicksplit(r, p, &v, itup, rtstate);
469
* The root of the tree is the first block in the relation. If we're
470
* about to split the root, we need to do some hocus-pocus to enforce
474
if (BufferGetBlockNumber(buffer) == P_ROOT)
476
leftbuf = ReadBuffer(r, P_NEW);
477
RTInitBuffer(leftbuf, opaque->flags);
478
lbknum = BufferGetBlockNumber(leftbuf);
479
left = (Page) BufferGetPage(leftbuf);
484
IncrBufferRefCount(buffer);
485
lbknum = BufferGetBlockNumber(buffer);
486
left = (Page) PageGetTempPage(p, sizeof(RTreePageOpaqueData));
489
rightbuf = ReadBuffer(r, P_NEW);
490
RTInitBuffer(rightbuf, opaque->flags);
491
rbknum = BufferGetBlockNumber(rightbuf);
492
right = (Page) BufferGetPage(rightbuf);
494
spl_left = v.spl_left;
495
spl_right = v.spl_right;
496
leftoff = rightoff = FirstOffsetNumber;
497
maxoff = PageGetMaxOffsetNumber(p);
498
newitemoff = OffsetNumberNext(maxoff);
500
/* build an InsertIndexResult for this insertion */
501
res = (InsertIndexResult) palloc(sizeof(InsertIndexResultData));
504
* spl_left contains a list of the offset numbers of the tuples that
505
* will go to the left page. For each offset number, get the tuple
506
* item, then add the item to the left page. Similarly for the right
511
for (n = 0; n < v.spl_nleft; n++)
518
itemid = PageGetItemId(p, i);
519
item = (IndexTuple) PageGetItem(p, itemid);
522
if (PageAddItem(left, (Item) item, IndexTupleSize(item),
523
leftoff, LP_USED) == InvalidOffsetNumber)
524
elog(ERROR, "failed to add index item to \"%s\"",
525
RelationGetRelationName(r));
526
leftoff = OffsetNumberNext(leftoff);
529
ItemPointerSet(&(res->pointerData), lbknum, leftoff);
531
spl_left++; /* advance in left split vector */
534
/* fill right node */
535
for (n = 0; n < v.spl_nright; n++)
542
itemid = PageGetItemId(p, i);
543
item = (IndexTuple) PageGetItem(p, itemid);
546
if (PageAddItem(right, (Item) item, IndexTupleSize(item),
547
rightoff, LP_USED) == InvalidOffsetNumber)
548
elog(ERROR, "failed to add index item to \"%s\"",
549
RelationGetRelationName(r));
550
rightoff = OffsetNumberNext(rightoff);
553
ItemPointerSet(&(res->pointerData), rbknum, rightoff);
555
spl_right++; /* advance in right split vector */
558
/* Make sure we consumed all of the split vectors, and release 'em */
559
Assert(*spl_left == InvalidOffsetNumber);
560
Assert(*spl_right == InvalidOffsetNumber);
564
if ((bufblock = BufferGetBlockNumber(buffer)) != P_ROOT)
565
PageRestoreTempPage(left, p);
566
WriteBuffer(leftbuf);
567
WriteBuffer(rightbuf);
570
* Okay, the page is split. We have three things left to do:
572
* 1) Adjust any active scans on this index to cope with changes we
573
* introduced in its structure by splitting this page.
575
* 2) "Tighten" the bounding box of the pointer to the left page in the
576
* parent node in the tree, if any. Since we moved a bunch of stuff
577
* off the left page, we expect it to get smaller. This happens in
578
* the internal insertion routine.
580
* 3) Insert a pointer to the right page in the parent. This may cause
581
* the parent to split. If it does, we need to repeat steps one and
582
* two for each split node in the tree.
585
/* adjust active scans */
586
rtadjscans(r, RTOP_SPLIT, bufblock, FirstOffsetNumber);
589
isnull = (char *) palloc(r->rd_rel->relnatts);
590
for (blank = 0; blank < r->rd_rel->relnatts; blank++)
593
ltup = (IndexTuple) index_formtuple(tupDesc,
594
&(v.spl_ldatum), isnull);
595
rtup = (IndexTuple) index_formtuple(tupDesc,
596
&(v.spl_rdatum), isnull);
598
pfree(DatumGetPointer(v.spl_ldatum));
599
pfree(DatumGetPointer(v.spl_rdatum));
601
/* set pointers to new child pages in the internal index tuples */
602
ItemPointerSet(&(ltup->t_tid), lbknum, 1);
603
ItemPointerSet(&(rtup->t_tid), rbknum, 1);
605
rtintinsert(r, stack, ltup, rtup, rtstate);
614
rtintinsert(Relation r,
626
InsertIndexResult res;
630
rtnewroot(r, ltup, rtup);
634
b = ReadBuffer(r, stk->rts_blk);
635
p = BufferGetPage(b);
636
old = (IndexTuple) PageGetItem(p, PageGetItemId(p, stk->rts_child));
639
* This is a hack. Right now, we force rtree internal keys to be
640
* constant size. To fix this, need delete the old key and add both
641
* left and right for the two new pages. The insertion of left may
642
* force a split if the new left key is bigger than the old key.
645
if (IndexTupleSize(old) != IndexTupleSize(ltup))
647
(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
648
errmsg("variable-length rtree keys are not supported")));
650
/* install pointer to left child */
651
memmove(old, ltup, IndexTupleSize(ltup));
653
if (nospace(p, rtup))
655
newdatum = IndexTupleGetDatum(ltup);
656
rttighten(r, stk->rts_parent, newdatum,
657
IndexTupleAttSize(ltup), rtstate);
658
res = rtdosplit(r, b, stk->rts_parent, rtup, rtstate);
659
WriteBuffer(b); /* don't forget to release buffer! -
665
if (PageAddItem(p, (Item) rtup, IndexTupleSize(rtup),
666
PageGetMaxOffsetNumber(p),
667
LP_USED) == InvalidOffsetNumber)
668
elog(ERROR, "failed to add index item to \"%s\"",
669
RelationGetRelationName(r));
671
ldatum = IndexTupleGetDatum(ltup);
672
rdatum = IndexTupleGetDatum(rtup);
673
newdatum = FunctionCall2(&rtstate->unionFn, ldatum, rdatum);
675
rttighten(r, stk->rts_parent, newdatum,
676
IndexTupleAttSize(rtup), rtstate);
678
pfree(DatumGetPointer(newdatum));
683
rtnewroot(Relation r, IndexTuple lt, IndexTuple rt)
688
b = ReadBuffer(r, P_ROOT);
690
p = BufferGetPage(b);
691
if (PageAddItem(p, (Item) lt, IndexTupleSize(lt),
693
LP_USED) == InvalidOffsetNumber)
694
elog(ERROR, "failed to add index item to \"%s\"",
695
RelationGetRelationName(r));
696
if (PageAddItem(p, (Item) rt, IndexTupleSize(rt),
697
OffsetNumberNext(FirstOffsetNumber),
698
LP_USED) == InvalidOffsetNumber)
699
elog(ERROR, "failed to add index item to \"%s\"",
700
RelationGetRelationName(r));
705
* Choose how to split an rtree page into two pages.
707
* We return two vectors of index item numbers, one for the items to be
708
* put on the left page, one for the items to be put on the right page.
709
* In addition, the item to be added (itup) is listed in the appropriate
710
* vector. It is represented by item number N+1 (N = # of items on page).
712
* Both vectors have a terminating sentinel value of InvalidOffsetNumber,
713
* but the sentinal value is no longer used, because the SPLITVEC
714
* vector also contains the length of each vector, and that information
715
* is now used to iterate over them in rtdosplit(). --kbb, 21 Sept 2001
717
* The bounding-box datums for the two new pages are also returned in *v.
719
* This is the quadratic-cost split algorithm Guttman describes in
720
* his paper. The reason we chose it is that you can implement this
721
* with less information about the data types on which you're operating.
723
* We must also deal with a consideration not found in Guttman's algorithm:
724
* variable-length data. In particular, the incoming item might be
725
* large enough that not just any split will work. In the worst case,
726
* our "split" may have to be the new item on one page and all the existing
727
* items on the other. Short of that, we have to take care that we do not
728
* make a split that leaves both pages too full for the new item.
731
rtpicksplit(Relation r,
761
OffsetNumber seed_1 = 0,
770
int total_num_tuples,
771
num_tuples_without_seeds,
772
max_after_split; /* in Guttman's lingo, (M - m) */
773
float diff; /* diff between cost of putting tuple left
775
SPLITCOST *cost_vector;
779
* First, make sure the new item is not so large that we can't
780
* possibly fit it on a page, even by itself. (It's sufficient to
781
* make this test here, since any oversize tuple must lead to a page
784
newitemsz = IndexTupleTotalSize(itup);
785
if (newitemsz > RTPageAvailSpace)
787
(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
788
errmsg("index row size %lu exceeds rtree maximum, %lu",
789
(unsigned long) newitemsz,
790
(unsigned long) RTPageAvailSpace)));
792
maxoff = PageGetMaxOffsetNumber(page);
793
newitemoff = OffsetNumberNext(maxoff); /* phony index for new
795
total_num_tuples = newitemoff;
796
num_tuples_without_seeds = total_num_tuples - 2;
797
max_after_split = total_num_tuples / 2; /* works for m = M/2 */
799
/* Make arrays big enough for worst case, including sentinel */
800
nbytes = (maxoff + 2) * sizeof(OffsetNumber);
801
v->spl_left = (OffsetNumber *) palloc(nbytes);
802
v->spl_right = (OffsetNumber *) palloc(nbytes);
807
for (i = FirstOffsetNumber; i < maxoff; i = OffsetNumberNext(i))
809
item_1 = (IndexTuple) PageGetItem(page, PageGetItemId(page, i));
810
datum_alpha = IndexTupleGetDatum(item_1);
811
item_1_sz = IndexTupleTotalSize(item_1);
813
for (j = OffsetNumberNext(i); j <= maxoff; j = OffsetNumberNext(j))
815
item_2 = (IndexTuple) PageGetItem(page, PageGetItemId(page, j));
816
datum_beta = IndexTupleGetDatum(item_2);
817
item_2_sz = IndexTupleTotalSize(item_2);
820
* Ignore seed pairs that don't leave room for the new item on
823
if (newitemsz + item_1_sz > RTPageAvailSpace &&
824
newitemsz + item_2_sz > RTPageAvailSpace)
827
/* compute the wasted space by unioning these guys */
828
union_d = FunctionCall2(&rtstate->unionFn,
829
datum_alpha, datum_beta);
830
FunctionCall2(&rtstate->sizeFn, union_d,
831
PointerGetDatum(&size_union));
832
inter_d = FunctionCall2(&rtstate->interFn,
833
datum_alpha, datum_beta);
836
* The interFn may return a NULL pointer (not an SQL null!) to
837
* indicate no intersection. sizeFn must cope with this.
839
FunctionCall2(&rtstate->sizeFn, inter_d,
840
PointerGetDatum(&size_inter));
841
size_waste = size_union - size_inter;
843
if (DatumGetPointer(union_d) != NULL)
844
pfree(DatumGetPointer(union_d));
845
if (DatumGetPointer(inter_d) != NULL)
846
pfree(DatumGetPointer(inter_d));
849
* are these a more promising split that what we've already
852
if (size_waste > waste || firsttime)
865
* There is no possible split except to put the new item on its
866
* own page. Since we still have to compute the union rectangles,
867
* we play dumb and run through the split algorithm anyway,
868
* setting seed_1 = first item on page and seed_2 = new item.
870
seed_1 = FirstOffsetNumber;
874
item_1 = (IndexTuple) PageGetItem(page, PageGetItemId(page, seed_1));
875
datum_alpha = IndexTupleGetDatum(item_1);
876
datum_l = FunctionCall2(&rtstate->unionFn, datum_alpha, datum_alpha);
877
FunctionCall2(&rtstate->sizeFn, datum_l, PointerGetDatum(&size_l));
878
left_avail_space = RTPageAvailSpace - IndexTupleTotalSize(item_1);
880
if (seed_2 == newitemoff)
883
/* Needn't leave room for new item in calculations below */
887
item_2 = (IndexTuple) PageGetItem(page, PageGetItemId(page, seed_2));
888
datum_beta = IndexTupleGetDatum(item_2);
889
datum_r = FunctionCall2(&rtstate->unionFn, datum_beta, datum_beta);
890
FunctionCall2(&rtstate->sizeFn, datum_r, PointerGetDatum(&size_r));
891
right_avail_space = RTPageAvailSpace - IndexTupleTotalSize(item_2);
894
* Now split up the regions between the two seeds.
896
* The cost_vector array will contain hints for determining where each
897
* tuple should go. Each record in the array will contain a boolean,
898
* choose_left, that indicates which node the tuple prefers to be on,
899
* and the absolute difference in cost between putting the tuple in
900
* its favored node and in the other node.
902
* Later, we will sort the cost_vector in descending order by cost
903
* difference, and consider the tuples in that order for placement.
904
* That way, the tuples that *really* want to be in one node or the
905
* other get to choose first, and the tuples that don't really care
908
* First, build the cost_vector array. The new index tuple will also be
909
* handled in this loop, and represented in the array, with
912
* In the case of variable size tuples it is possible that we only have
913
* the two seeds and no other tuples, in which case we don't do any of
914
* this cost_vector stuff.
917
/* to keep compiler quiet */
920
if (num_tuples_without_seeds > 0)
923
(SPLITCOST *) palloc(num_tuples_without_seeds * sizeof(SPLITCOST));
925
for (i = FirstOffsetNumber; i <= newitemoff; i = OffsetNumberNext(i))
927
/* Compute new union datums and sizes for both choices */
929
if ((i == seed_1) || (i == seed_2))
931
else if (i == newitemoff)
934
item_1 = (IndexTuple) PageGetItem(page, PageGetItemId(page, i));
936
datum_alpha = IndexTupleGetDatum(item_1);
937
union_dl = FunctionCall2(&rtstate->unionFn, datum_l, datum_alpha);
938
union_dr = FunctionCall2(&rtstate->unionFn, datum_r, datum_alpha);
939
FunctionCall2(&rtstate->sizeFn, union_dl,
940
PointerGetDatum(&size_alpha));
941
FunctionCall2(&rtstate->sizeFn, union_dr,
942
PointerGetDatum(&size_beta));
943
pfree(DatumGetPointer(union_dl));
944
pfree(DatumGetPointer(union_dr));
946
diff = (size_alpha - size_l) - (size_beta - size_r);
948
cost_vector[n].offset_number = i;
949
cost_vector[n].cost_differential = fabs(diff);
950
cost_vector[n].choose_left = (diff < 0);
956
* Sort the array. The function qsort_comp_splitcost is set up
957
* "backwards", to provided descending order.
959
qsort(cost_vector, num_tuples_without_seeds, sizeof(SPLITCOST),
960
&qsort_comp_splitcost);
964
* Now make the final decisions about where each tuple will go, and
965
* build the vectors to return in the SPLITVEC record.
967
* The cost_vector array contains (descriptions of) all the tuples, in
968
* the order that we want to consider them, so we we just iterate
969
* through it and place each tuple in left or right nodes, according
970
* to the criteria described below.
975
right = v->spl_right;
979
* Place the seeds first. left avail space, left union, right avail
980
* space, and right union have already been adjusted for the seeds.
989
for (n = 0; n < num_tuples_without_seeds; n++)
996
* We need to figure out which page needs the least enlargement in
997
* order to store the item.
1000
i = cost_vector[n].offset_number;
1002
/* Compute new union datums and sizes for both possible additions */
1003
if (i == newitemoff)
1006
/* Needn't leave room for new item anymore */
1010
item_1 = (IndexTuple) PageGetItem(page, PageGetItemId(page, i));
1011
item_1_sz = IndexTupleTotalSize(item_1);
1013
datum_alpha = IndexTupleGetDatum(item_1);
1014
union_dl = FunctionCall2(&rtstate->unionFn, datum_l, datum_alpha);
1015
union_dr = FunctionCall2(&rtstate->unionFn, datum_r, datum_alpha);
1016
FunctionCall2(&rtstate->sizeFn, union_dl,
1017
PointerGetDatum(&size_alpha));
1018
FunctionCall2(&rtstate->sizeFn, union_dr,
1019
PointerGetDatum(&size_beta));
1022
* We prefer the page that shows smaller enlargement of its union
1023
* area (Guttman's algorithm), but we must take care that at least
1024
* one page will still have room for the new item after this one
1027
* (We know that all the old items together can fit on one page, so
1028
* we need not worry about any other problem than failing to fit
1031
* Guttman's algorithm actually has two factors to consider (in
1032
* order): 1. if one node has so many tuples already assigned to
1033
* it that the other needs all the rest in order to satisfy the
1034
* condition that neither node has fewer than m tuples, then that
1035
* is decisive; 2. otherwise, choose the page that shows the
1036
* smaller enlargement of its union area.
1038
* I have chosen m = M/2, where M is the maximum number of tuples on
1039
* a page. (Actually, this is only strictly true for fixed size
1040
* tuples. For variable size tuples, there still might have to be
1041
* only one tuple on a page, if it is really big. But even with
1042
* variable size tuples we still try to get m as close as possible
1045
* The question of which page shows the smaller enlargement of its
1046
* union area has already been answered, and the answer stored in
1047
* the choose_left field of the SPLITCOST record.
1049
left_feasible = (left_avail_space >= item_1_sz &&
1050
((left_avail_space - item_1_sz) >= newitemsz ||
1051
right_avail_space >= newitemsz));
1052
right_feasible = (right_avail_space >= item_1_sz &&
1053
((right_avail_space - item_1_sz) >= newitemsz ||
1054
left_avail_space >= newitemsz));
1055
if (left_feasible && right_feasible)
1058
* Both feasible, use Guttman's algorithm. First check the m
1059
* condition described above, and if that doesn't apply,
1060
* choose the page with the smaller enlargement of its union
1063
if (v->spl_nleft > max_after_split)
1064
choose_left = false;
1065
else if (v->spl_nright > max_after_split)
1068
choose_left = cost_vector[n].choose_left;
1070
else if (left_feasible)
1072
else if (right_feasible)
1073
choose_left = false;
1076
elog(ERROR, "failed to find a workable rtree page split");
1077
choose_left = false; /* keep compiler quiet */
1082
pfree(DatumGetPointer(datum_l));
1083
pfree(DatumGetPointer(union_dr));
1085
size_l = size_alpha;
1086
left_avail_space -= item_1_sz;
1092
pfree(DatumGetPointer(datum_r));
1093
pfree(DatumGetPointer(union_dl));
1096
right_avail_space -= item_1_sz;
1102
if (num_tuples_without_seeds > 0)
1105
*left = *right = InvalidOffsetNumber; /* add ending sentinels */
1107
v->spl_ldatum = datum_l;
1108
v->spl_rdatum = datum_r;
1112
RTInitBuffer(Buffer b, uint32 f)
1114
RTreePageOpaque opaque;
1118
pageSize = BufferGetPageSize(b);
1120
page = BufferGetPage(b);
1122
PageInit(page, pageSize, sizeof(RTreePageOpaqueData));
1124
opaque = (RTreePageOpaque) PageGetSpecialPointer(page);
1129
choose(Relation r, Page p, IndexTuple it, RTSTATE *rtstate)
1131
OffsetNumber maxoff;
1141
id = IndexTupleGetDatum(it);
1142
maxoff = PageGetMaxOffsetNumber(p);
1146
for (i = FirstOffsetNumber; i <= maxoff; i = OffsetNumberNext(i))
1148
datum = IndexTupleGetDatum(PageGetItem(p, PageGetItemId(p, i)));
1149
FunctionCall2(&rtstate->sizeFn, datum,
1150
PointerGetDatum(&dsize));
1151
ud = FunctionCall2(&rtstate->unionFn, datum, id);
1152
FunctionCall2(&rtstate->sizeFn, ud,
1153
PointerGetDatum(&usize));
1154
pfree(DatumGetPointer(ud));
1155
if (which_grow < 0 || usize - dsize < which_grow)
1158
which_grow = usize - dsize;
1159
if (which_grow == 0)
1168
nospace(Page p, IndexTuple it)
1170
return PageGetFreeSpace(p) < IndexTupleSize(it);
1174
freestack(RTSTACK *s)
1187
* Bulk deletion of all index entries pointing to a set of heap tuples.
1188
* The set of target tuples is specified via a callback routine that tells
1189
* whether any given heap tuple (identified by ItemPointer) is being deleted.
1191
* Result: a palloc'd struct containing statistical info for VACUUM displays.
1194
rtbulkdelete(PG_FUNCTION_ARGS)
1196
Relation rel = (Relation) PG_GETARG_POINTER(0);
1197
IndexBulkDeleteCallback callback = (IndexBulkDeleteCallback) PG_GETARG_POINTER(1);
1198
void *callback_state = (void *) PG_GETARG_POINTER(2);
1199
IndexBulkDeleteResult *result;
1200
BlockNumber num_pages;
1201
double tuples_removed;
1202
double num_index_tuples;
1203
IndexScanDesc iscan;
1206
num_index_tuples = 0;
1209
* Since rtree is not marked "amconcurrent" in pg_am, caller should
1210
* have acquired exclusive lock on index relation. We need no locking
1215
* XXX generic implementation --- should be improved!
1218
/* walk through the entire index */
1219
iscan = index_beginscan(NULL, rel, SnapshotAny, 0, NULL);
1220
/* including killed tuples */
1221
iscan->ignore_killed_tuples = false;
1223
while (index_getnext_indexitem(iscan, ForwardScanDirection))
1225
vacuum_delay_point();
1227
if (callback(&iscan->xs_ctup.t_self, callback_state))
1229
ItemPointerData indextup = iscan->currentItemData;
1231
OffsetNumber offnum;
1235
blkno = ItemPointerGetBlockNumber(&indextup);
1236
offnum = ItemPointerGetOffsetNumber(&indextup);
1238
/* adjust any scans that will be affected by this deletion */
1239
/* (namely, my own scan) */
1240
rtadjscans(rel, RTOP_DEL, blkno, offnum);
1242
/* delete the index tuple */
1243
buf = ReadBuffer(rel, blkno);
1244
page = BufferGetPage(buf);
1246
PageIndexTupleDelete(page, offnum);
1250
tuples_removed += 1;
1253
num_index_tuples += 1;
1256
index_endscan(iscan);
1258
/* return statistics */
1259
num_pages = RelationGetNumberOfBlocks(rel);
1261
result = (IndexBulkDeleteResult *) palloc0(sizeof(IndexBulkDeleteResult));
1262
result->num_pages = num_pages;
1263
result->num_index_tuples = num_index_tuples;
1264
result->tuples_removed = tuples_removed;
1266
PG_RETURN_POINTER(result);
1271
initRtstate(RTSTATE *rtstate, Relation index)
1273
fmgr_info_copy(&rtstate->unionFn,
1274
index_getprocinfo(index, 1, RT_UNION_PROC),
1275
CurrentMemoryContext);
1276
fmgr_info_copy(&rtstate->sizeFn,
1277
index_getprocinfo(index, 1, RT_SIZE_PROC),
1278
CurrentMemoryContext);
1279
fmgr_info_copy(&rtstate->interFn,
1280
index_getprocinfo(index, 1, RT_INTER_PROC),
1281
CurrentMemoryContext);
1284
/* for sorting SPLITCOST records in descending order */
1286
qsort_comp_splitcost(const void *a, const void *b)
1289
((SPLITCOST *) a)->cost_differential -
1290
((SPLITCOST *) b)->cost_differential;
1307
OffsetNumber offnum,
1310
BlockNumber nblocks;
1313
BlockNumber itblkno;
1314
OffsetNumber itoffno;
1318
nblocks = RelationGetNumberOfBlocks(r);
1319
for (blkno = 0; blkno < nblocks; blkno++)
1321
buf = ReadBuffer(r, blkno);
1322
page = BufferGetPage(buf);
1323
po = (RTreePageOpaque) PageGetSpecialPointer(page);
1324
maxoff = PageGetMaxOffsetNumber(page);
1325
printf("Page %d maxoff %d <%s>\n", blkno, maxoff,
1326
(po->flags & F_LEAF ? "LEAF" : "INTERNAL"));
1328
if (PageIsEmpty(page))
1334
for (offnum = FirstOffsetNumber;
1336
offnum = OffsetNumberNext(offnum))
1338
itup = (IndexTuple) PageGetItem(page, PageGetItemId(page, offnum));
1339
itblkno = ItemPointerGetBlockNumber(&(itup->t_tid));
1340
itoffno = ItemPointerGetOffsetNumber(&(itup->t_tid));
1341
datum = IndexTupleGetDatum(itup);
1342
itkey = DatumGetCString(DirectFunctionCall1(box_out,
1344
printf("\t[%d] size %d heap <%d,%d> key:%s\n",
1345
offnum, IndexTupleSize(itup), itblkno, itoffno, itkey);
1352
#endif /* defined RTDEBUG */
1355
rtree_redo(XLogRecPtr lsn, XLogRecord *record)
1357
elog(PANIC, "rtree_redo: unimplemented");
1361
rtree_undo(XLogRecPtr lsn, XLogRecord *record)
1363
elog(PANIC, "rtree_undo: unimplemented");
1367
rtree_desc(char *buf, uint8 xl_info, char *rec)
added
removed
src/backend/access/rtree/rtree.c
