From f0b18f225e78acfc2ffed3af008ad2a5f645eb15 Mon Sep 17 00:00:00 2001 From: Pavel Borisov Date: Thu, 25 Mar 2021 23:32:58 +0400 Subject: [PATCH v14] Covering SP-GiST index - support for INCLUDE columns Adding INCLUDE columns for SPGiST index is intended to increase the speed of queries by making scans index-only likewise in btree and GiST index. These columns are added only to leaf tuples and they are not used in index tree search but they can be fetched during index scan. The other point of INCLUDE columns is to overcome SP-GiST limitation of being single-column in principle. I.e. in certain cases a single covering SP-GiST index can replace several separate ones with less disk space and shared buffers consumption, faster, update etc. Also, any data types without SP-GiST supported opclasses can be included. Discussion: https://www.postgresql.org/message-id/flat/CALT9ZEFi-vMp4faht9f9Junb1nO3NOSjhpxTmbm1UGLMsLqiEQ@mail.gmail.com --- doc/src/sgml/indices.sgml | 4 +- doc/src/sgml/ref/create_index.sgml | 4 +- doc/src/sgml/spgist.sgml | 8 + src/backend/access/common/indextuple.c | 46 +++-- src/backend/access/spgist/README | 17 ++ src/backend/access/spgist/spgdoinsert.c | 174 ++++++++++------ src/backend/access/spgist/spginsert.c | 5 +- src/backend/access/spgist/spgscan.c | 91 +++++++-- src/backend/access/spgist/spgutils.c | 191 +++++++++++++++--- src/backend/access/spgist/spgvacuum.c | 25 ++- src/backend/access/spgist/spgxlog.c | 6 +- src/include/access/itup.h | 3 + src/include/access/spgist_private.h | 128 ++++++++---- src/test/regress/expected/amutils.out | 2 +- src/test/regress/expected/index_including.out | 3 +- src/test/regress/parallel_schedule | 2 +- src/test/regress/serial_schedule | 1 + src/test/regress/sql/index_including.sql | 2 +- 18 files changed, 524 insertions(+), 188 deletions(-) diff --git a/doc/src/sgml/indices.sgml b/doc/src/sgml/indices.sgml index 623962d1d8..7a6bb03bef 100644 --- a/doc/src/sgml/indices.sgml +++ b/doc/src/sgml/indices.sgml @@ -1208,8 +1208,8 @@ CREATE UNIQUE INDEX tab_x_y ON tab(x) INCLUDE (y); likely to not need to access the heap. If the heap tuple must be visited anyway, it costs nothing more to get the column's value from there. Other restrictions are that expressions are not currently supported as - included columns, and that only B-tree and GiST indexes currently support - included columns. + included columns, and that only B-tree, GiST and SP-GiST indexes currently + support included columns. diff --git a/doc/src/sgml/ref/create_index.sgml b/doc/src/sgml/ref/create_index.sgml index 682af4bbe4..cb74c998b1 100644 --- a/doc/src/sgml/ref/create_index.sgml +++ b/doc/src/sgml/ref/create_index.sgml @@ -187,8 +187,8 @@ CREATE [ UNIQUE ] INDEX [ CONCURRENTLY ] [ [ IF NOT EXISTS ] - Currently, the B-tree and the GiST index access methods support this - feature. In B-tree and the GiST indexes, the values of columns listed + Currently, the B-tree, GiST and SP-GiST index access methods support + this feature. In these indexes, the values of columns listed in the INCLUDE clause are included in leaf tuples which correspond to heap tuples, but are not included in upper-level index entries used for tree navigation. diff --git a/doc/src/sgml/spgist.sgml b/doc/src/sgml/spgist.sgml index ea88ae45e5..148b5629b7 100644 --- a/doc/src/sgml/spgist.sgml +++ b/doc/src/sgml/spgist.sgml @@ -214,6 +214,14 @@ inner tuples that are passed through to reach the leaf level. + + In case when SP-GiST index is created with + INCLUDE clause i.e. covering index, leaf tuples also + contain data from included columns. This data is stored uncompressed and can have + data types without any SP-GiST operator class. + + + Inner tuples are more complex, since they are branching points in the search tree. Each inner tuple contains a set of one or more diff --git a/src/backend/access/common/indextuple.c b/src/backend/access/common/indextuple.c index 1f6b7b77d4..f4819fbed0 100644 --- a/src/backend/access/common/indextuple.c +++ b/src/backend/access/common/indextuple.c @@ -422,35 +422,55 @@ nocache_index_getattr(IndexTuple tup, return fetchatt(TupleDescAttr(tupleDesc, attnum), tp + off); } +/* Convert index tuple into Datum/isnull arrays */ +void +index_deform_tuple(IndexTuple tup, TupleDesc tupleDescriptor, + Datum *values, bool *isnull) +{ + bits8 *bp; /* ptr to null bitmap in tuple */ + char *tp; /* ptr to tuple data */ + + bp = (bits8 *) ((char *) tup + sizeof(IndexTupleData)); + tp = (char *) tup + IndexInfoFindDataOffset(tup->t_info); + + index_deform_anyheader_tuple((char *) tup, tupleDescriptor, + values, isnull, + bp, tp, + (bool) IndexTupleHasNulls(tup)); +} + /* - * Convert an index tuple into Datum/isnull arrays. + * Convert an index-like tuples but with arbitrary header length into + * Datum/isnull arrays. * * The caller must allocate sufficient storage for the output arrays. * (INDEX_MAX_KEYS entries should be enough.) * - * This is nearly the same as heap_deform_tuple(), but for IndexTuples. - * One difference is that the tuple should never have any missing columns. + * This is nearly the same as heap_deform_tuple(), but for IndexTuples and + * SpGistLeafTuples. One difference is that the tuple should never have any + * missing columns. + * + * Callers are expected to provide pointer to null bitmap, MAXALIGN-ed + * pointer to tuple data and hasnulls bit got from the header. */ void -index_deform_tuple(IndexTuple tup, TupleDesc tupleDescriptor, - Datum *values, bool *isnull) +index_deform_anyheader_tuple(char *tup, TupleDesc tupleDescriptor, + Datum *values, bool *isnull, + bits8 *bp, char *tp, bool hasnulls) { - int hasnulls = IndexTupleHasNulls(tup); int natts = tupleDescriptor->natts; /* number of atts to extract */ int attnum; - char *tp; /* ptr to tuple data */ - int off; /* offset in tuple data */ - bits8 *bp; /* ptr to null bitmap in tuple */ + int off = 0; /* offset in tuple data */ bool slow = false; /* can we use/set attcacheoff? */ /* Assert to protect callers who allocate fixed-size arrays */ Assert(natts <= INDEX_MAX_KEYS); - /* XXX "knows" t_bits are just after fixed tuple header! */ - bp = (bits8 *) ((char *) tup + sizeof(IndexTupleData)); + /* If has hulls, null mask should be present in a tuple */ + Assert(hasnulls == false || ((char *) bp < tp)); - tp = (char *) tup + IndexInfoFindDataOffset(tup->t_info); - off = 0; + /* Tuple data should start from MAXALIGN */ + Assert(tp == (char *) MAXALIGN(tp)); for (attnum = 0; attnum < natts; attnum++) { diff --git a/src/backend/access/spgist/README b/src/backend/access/spgist/README index b55b073832..ab5ff3d434 100644 --- a/src/backend/access/spgist/README +++ b/src/backend/access/spgist/README @@ -76,6 +76,19 @@ Leaf tuple consists of: ItemPointer to the heap + nextOffset number of next leaf tuple in a chain on a leaf page + optional nullmask + optional INCLUDE columns values + +Leaf tuple layout changed since PostgreSQL version 14 to support INCLUDE +columns but in a way that doesn't change the header and the key value +placement for a tuple without INCLUDE columns. So indexes created earlier +remain fully supported. + +Nullmask is added only if there are INCLUDE columns and nulls in a tuple. +It is added without alignment and 64-bit architectures has a good chance +to fit alignment gap before first value which makes its storage free of +charge. NULLS HANDLING @@ -90,6 +103,10 @@ Insertions and searches in the nulls tree do not use any of the opclass-supplied functions, but just use hardwired logic comparable to AllTheSame cases in the normal tree. +For INCLUDE attributes nulls are handled in ordinary per leaf-tuple way i.e. +if null mask presence bit in a header is set, nullmask is added to tuple. +If there is nullmask it covers all attributes, key attribute as well. It is +redundant but allows to use index tuple code to operate SpGistLeafTuple. INSERTION ALGORITHM diff --git a/src/backend/access/spgist/spgdoinsert.c b/src/backend/access/spgist/spgdoinsert.c index 7bd269fd2a..01a0ee0921 100644 --- a/src/backend/access/spgist/spgdoinsert.c +++ b/src/backend/access/spgist/spgdoinsert.c @@ -22,7 +22,7 @@ #include "miscadmin.h" #include "storage/bufmgr.h" #include "utils/rel.h" - +#include "access/htup_details.h" /* * SPPageDesc tracks all info about a page we are inserting into. In some @@ -220,7 +220,7 @@ addLeafTuple(Relation index, SpGistState *state, SpGistLeafTuple leafTuple, SpGistBlockIsRoot(current->blkno)) { /* Tuple is not part of a chain */ - leafTuple->nextOffset = InvalidOffsetNumber; + SGLT_SET_OFFSET(leafTuple, InvalidOffsetNumber); current->offnum = SpGistPageAddNewItem(state, current->page, (Item) leafTuple, leafTuple->size, NULL, false); @@ -253,7 +253,7 @@ addLeafTuple(Relation index, SpGistState *state, SpGistLeafTuple leafTuple, PageGetItemId(current->page, current->offnum)); if (head->tupstate == SPGIST_LIVE) { - leafTuple->nextOffset = head->nextOffset; + SGLT_SET_OFFSET(leafTuple, SGLT_GET_OFFSET(head)); offnum = SpGistPageAddNewItem(state, current->page, (Item) leafTuple, leafTuple->size, NULL, false); @@ -264,14 +264,14 @@ addLeafTuple(Relation index, SpGistState *state, SpGistLeafTuple leafTuple, */ head = (SpGistLeafTuple) PageGetItem(current->page, PageGetItemId(current->page, current->offnum)); - head->nextOffset = offnum; + SGLT_SET_OFFSET(head, offnum); xlrec.offnumLeaf = offnum; xlrec.offnumHeadLeaf = current->offnum; } else if (head->tupstate == SPGIST_DEAD) { - leafTuple->nextOffset = InvalidOffsetNumber; + SGLT_SET_OFFSET(leafTuple, InvalidOffsetNumber); PageIndexTupleDelete(current->page, current->offnum); if (PageAddItem(current->page, (Item) leafTuple, leafTuple->size, @@ -362,13 +362,13 @@ checkSplitConditions(Relation index, SpGistState *state, { /* We could see a DEAD tuple as first/only chain item */ Assert(i == current->offnum); - Assert(it->nextOffset == InvalidOffsetNumber); + Assert(SGLT_GET_OFFSET(it) == InvalidOffsetNumber); /* Don't count it in result, because it won't go to other page */ } else elog(ERROR, "unexpected SPGiST tuple state: %d", it->tupstate); - i = it->nextOffset; + i = SGLT_GET_OFFSET(it); } *nToSplit = n; @@ -437,7 +437,7 @@ moveLeafs(Relation index, SpGistState *state, { /* We could see a DEAD tuple as first/only chain item */ Assert(i == current->offnum); - Assert(it->nextOffset == InvalidOffsetNumber); + Assert(SGLT_GET_OFFSET(it) == InvalidOffsetNumber); /* We don't want to move it, so don't count it in size */ toDelete[nDelete] = i; nDelete++; @@ -446,7 +446,7 @@ moveLeafs(Relation index, SpGistState *state, else elog(ERROR, "unexpected SPGiST tuple state: %d", it->tupstate); - i = it->nextOffset; + i = SGLT_GET_OFFSET(it); } /* Find a leaf page that will hold them */ @@ -475,7 +475,7 @@ moveLeafs(Relation index, SpGistState *state, * don't care). We're modifying the tuple on the source page * here, but it's okay since we're about to delete it. */ - it->nextOffset = r; + SGLT_SET_OFFSET(it, r); r = SpGistPageAddNewItem(state, npage, (Item) it, it->size, &startOffset, false); @@ -490,7 +490,7 @@ moveLeafs(Relation index, SpGistState *state, } /* add the new tuple as well */ - newLeafTuple->nextOffset = r; + SGLT_SET_OFFSET(newLeafTuple, r); r = SpGistPageAddNewItem(state, npage, (Item) newLeafTuple, newLeafTuple->size, &startOffset, false); @@ -690,13 +690,13 @@ doPickSplit(Relation index, SpGistState *state, *nodes; Buffer newInnerBuffer, newLeafBuffer; - ItemPointerData *heapPtrs; uint8 *leafPageSelect; int *leafSizes; OffsetNumber *toDelete; OffsetNumber *toInsert; OffsetNumber redirectTuplePos = InvalidOffsetNumber; OffsetNumber startOffsets[2]; + SpGistLeafTuple *oldLeafs; SpGistLeafTuple *newLeafs; int spaceToDelete; int currentFreeSpace; @@ -709,6 +709,8 @@ doPickSplit(Relation index, SpGistState *state, int nToDelete, nToInsert, maxToInclude; + Datum leafDatums[INDEX_MAX_KEYS]; + bool leafIsnulls[INDEX_MAX_KEYS]; in.level = level; @@ -718,12 +720,11 @@ doPickSplit(Relation index, SpGistState *state, max = PageGetMaxOffsetNumber(current->page); n = max + 1; in.datums = (Datum *) palloc(sizeof(Datum) * n); - heapPtrs = (ItemPointerData *) palloc(sizeof(ItemPointerData) * n); toDelete = (OffsetNumber *) palloc(sizeof(OffsetNumber) * n); toInsert = (OffsetNumber *) palloc(sizeof(OffsetNumber) * n); + oldLeafs = (SpGistLeafTuple *) palloc(sizeof(SpGistLeafTuple) * n); newLeafs = (SpGistLeafTuple *) palloc(sizeof(SpGistLeafTuple) * n); leafPageSelect = (uint8 *) palloc(sizeof(uint8) * n); - STORE_STATE(state, xlrec.stateSrc); /* @@ -739,6 +740,7 @@ doPickSplit(Relation index, SpGistState *state, * we are just computing a pointer that isn't going to get dereferenced. * So it's not worth guarding the calls with isNulls checks. */ + nToInsert = 0; nToDelete = 0; spaceToDelete = 0; @@ -758,7 +760,7 @@ doPickSplit(Relation index, SpGistState *state, if (it->tupstate == SPGIST_LIVE) { in.datums[nToInsert] = SGLTDATUM(it, state); - heapPtrs[nToInsert] = it->heapPtr; + oldLeafs[nToInsert] = it; nToInsert++; toDelete[nToDelete] = i; nToDelete++; @@ -783,7 +785,7 @@ doPickSplit(Relation index, SpGistState *state, if (it->tupstate == SPGIST_LIVE) { in.datums[nToInsert] = SGLTDATUM(it, state); - heapPtrs[nToInsert] = it->heapPtr; + oldLeafs[nToInsert] = it; nToInsert++; toDelete[nToDelete] = i; nToDelete++; @@ -795,7 +797,7 @@ doPickSplit(Relation index, SpGistState *state, { /* We could see a DEAD tuple as first/only chain item */ Assert(i == current->offnum); - Assert(it->nextOffset == InvalidOffsetNumber); + Assert(SGLT_GET_OFFSET(it) == InvalidOffsetNumber); toDelete[nToDelete] = i; nToDelete++; /* replacing it with redirect will save no space */ @@ -803,7 +805,7 @@ doPickSplit(Relation index, SpGistState *state, else elog(ERROR, "unexpected SPGiST tuple state: %d", it->tupstate); - i = it->nextOffset; + i = SGLT_GET_OFFSET(it); } } in.nTuples = nToInsert; @@ -815,7 +817,7 @@ doPickSplit(Relation index, SpGistState *state, * for the picksplit function. So don't increment nToInsert yet. */ in.datums[in.nTuples] = SGLTDATUM(newLeafTuple, state); - heapPtrs[in.nTuples] = newLeafTuple->heapPtr; + oldLeafs[in.nTuples] = newLeafTuple; in.nTuples++; memset(&out, 0, sizeof(out)); @@ -837,9 +839,17 @@ doPickSplit(Relation index, SpGistState *state, totalLeafSizes = 0; for (i = 0; i < in.nTuples; i++) { - newLeafs[i] = spgFormLeafTuple(state, heapPtrs + i, - out.leafTupleDatums[i], - false); + spgDeformLeafTuple(oldLeafs[i], state->tupleDescriptor, + leafDatums, + leafIsnulls, + isNulls); + + leafDatums[spgKeyColumn] = (Datum) out.leafTupleDatums[i]; + leafIsnulls[spgKeyColumn] = false; + + newLeafs[i] = spgFormLeafTuple(state, &oldLeafs[i]->heapPtr, + leafDatums, + leafIsnulls); totalLeafSizes += newLeafs[i]->size + sizeof(ItemIdData); } } @@ -860,9 +870,20 @@ doPickSplit(Relation index, SpGistState *state, totalLeafSizes = 0; for (i = 0; i < in.nTuples; i++) { - newLeafs[i] = spgFormLeafTuple(state, heapPtrs + i, - (Datum) 0, - true); + spgDeformLeafTuple(oldLeafs[i], state->tupleDescriptor, + leafDatums, + leafIsnulls, + isNulls); + + /* + * Nulls tree can contain only null key values. + */ + leafDatums[spgKeyColumn] = (Datum) 0; + leafIsnulls[spgKeyColumn] = true; + + newLeafs[i] = spgFormLeafTuple(state, &oldLeafs[i]->heapPtr, + leafDatums, + leafIsnulls); totalLeafSizes += newLeafs[i]->size + sizeof(ItemIdData); } } @@ -1196,10 +1217,10 @@ doPickSplit(Relation index, SpGistState *state, if (ItemPointerIsValid(&nodes[n]->t_tid)) { Assert(ItemPointerGetBlockNumber(&nodes[n]->t_tid) == leafBlock); - it->nextOffset = ItemPointerGetOffsetNumber(&nodes[n]->t_tid); + SGLT_SET_OFFSET(it, ItemPointerGetOffsetNumber(&nodes[n]->t_tid)); } else - it->nextOffset = InvalidOffsetNumber; + SGLT_SET_OFFSET(it, InvalidOffsetNumber); /* Insert it on page */ newoffset = SpGistPageAddNewItem(state, BufferGetPage(leafBuffer), @@ -1889,67 +1910,87 @@ spgSplitNodeAction(Relation index, SpGistState *state, */ bool spgdoinsert(Relation index, SpGistState *state, - ItemPointer heapPtr, Datum datum, bool isnull) + ItemPointer heapPtr, Datum *datum, bool *isnull) { int level = 0; - Datum leafDatum; + Datum *leafDatum; int leafSize; SPPageDesc current, parent; FmgrInfo *procinfo = NULL; + int i; /* * Look up FmgrInfo of the user-defined choose function once, to save * cycles in the loop below. */ - if (!isnull) + if (!isnull[spgKeyColumn]) procinfo = index_getprocinfo(index, 1, SPGIST_CHOOSE_PROC); /* * Prepare the leaf datum to insert. - * + */ + + leafDatum = (Datum *) palloc0(sizeof(Datum) * (IndexRelationGetNumberOfAttributes(index))); + + /* * If an optional "compress" method is provided, then call it to form the - * leaf datum from the input datum. Otherwise store the input datum as - * is. Since we don't use index_form_tuple in this AM, we have to make - * sure value to be inserted is not toasted; FormIndexDatum doesn't - * guarantee that. But we assume the "compress" method to return an - * untoasted value. + * leaf key column datum of a leaf from the input datum. Otherwise, store + * the input datum as is. Since we don't use index_form_tuple in this AM, + * we have to make sure value to be inserted is not toasted; + * FormIndexDatum doesn't guarantee that. But we assume the "compress" + * method to return an untoasted value. */ - if (!isnull) + if (!isnull[spgKeyColumn]) { if (OidIsValid(index_getprocid(index, 1, SPGIST_COMPRESS_PROC))) { FmgrInfo *compressProcinfo = NULL; compressProcinfo = index_getprocinfo(index, 1, SPGIST_COMPRESS_PROC); - leafDatum = FunctionCall1Coll(compressProcinfo, - index->rd_indcollation[0], - datum); + leafDatum[spgKeyColumn] = FunctionCall1Coll(compressProcinfo, + index->rd_indcollation[0], + datum[spgKeyColumn]); } else { Assert(state->attLeafType.type == state->attType.type); if (state->attType.attlen == -1) - leafDatum = PointerGetDatum(PG_DETOAST_DATUM(datum)); + leafDatum[spgKeyColumn] = PointerGetDatum(PG_DETOAST_DATUM(datum[spgKeyColumn])); else - leafDatum = datum; + leafDatum[spgKeyColumn] = datum[spgKeyColumn]; } } else - leafDatum = (Datum) 0; + leafDatum[spgKeyColumn] = (Datum) 0; + + for (i = spgFirstIncludeColumn; i < IndexRelationGetNumberOfAttributes(index); i++) + { + if (!isnull[i]) + { + if (TupleDescAttr(state->tupleDescriptor, i)->attlen == -1) + leafDatum[i] = PointerGetDatum(PG_DETOAST_DATUM(datum[i])); + else + leafDatum[i] = datum[i]; + } + else + leafDatum[i] = (Datum) 0; + } + /* - * Compute space needed for a leaf tuple containing the given datum. + * Compute space needed on a page for a leaf tuple containing the given + * datum. * * If it isn't gonna fit, and the opclass can't reduce the datum size by * suffixing, bail out now rather than getting into an endless loop. */ - if (!isnull) - leafSize = SGLTHDRSZ + sizeof(ItemIdData) + - SpGistGetTypeSize(&state->attLeafType, leafDatum); - else - leafSize = SGDTSIZE + sizeof(ItemIdData); + leafSize = MAXALIGN(sizeof(SpGistLeafTupleData) + + spgNullmaskSize(state->tupleDescriptor->natts, isnull)) + + heap_compute_data_size(state->tupleDescriptor, leafDatum, isnull); + leafSize = leafSize < SGDTSIZE ? SGDTSIZE : MAXALIGN(leafSize); + leafSize += sizeof(ItemIdData); if (leafSize > SPGIST_PAGE_CAPACITY && !state->config.longValuesOK) ereport(ERROR, @@ -1961,7 +2002,7 @@ spgdoinsert(Relation index, SpGistState *state, errhint("Values larger than a buffer page cannot be indexed."))); /* Initialize "current" to the appropriate root page */ - current.blkno = isnull ? SPGIST_NULL_BLKNO : SPGIST_ROOT_BLKNO; + current.blkno = isnull[spgKeyColumn] ? SPGIST_NULL_BLKNO : SPGIST_ROOT_BLKNO; current.buffer = InvalidBuffer; current.page = NULL; current.offnum = FirstOffsetNumber; @@ -1995,7 +2036,7 @@ spgdoinsert(Relation index, SpGistState *state, */ current.buffer = SpGistGetBuffer(index, - GBUF_LEAF | (isnull ? GBUF_NULLS : 0), + GBUF_LEAF | (isnull[spgKeyColumn] ? GBUF_NULLS : 0), Min(leafSize, SPGIST_PAGE_CAPACITY), &isNew); current.blkno = BufferGetBlockNumber(current.buffer); @@ -2037,7 +2078,7 @@ spgdoinsert(Relation index, SpGistState *state, current.page = BufferGetPage(current.buffer); /* should not arrive at a page of the wrong type */ - if (isnull ? !SpGistPageStoresNulls(current.page) : + if (isnull[spgKeyColumn] ? !SpGistPageStoresNulls(current.page) : SpGistPageStoresNulls(current.page)) elog(ERROR, "SPGiST index page %u has wrong nulls flag", current.blkno); @@ -2054,7 +2095,7 @@ spgdoinsert(Relation index, SpGistState *state, { /* it fits on page, so insert it and we're done */ addLeafTuple(index, state, leafTuple, - ¤t, &parent, isnull, isNew); + ¤t, &parent, isnull[spgKeyColumn], isNew); break; } else if ((sizeToSplit = @@ -2068,14 +2109,14 @@ spgdoinsert(Relation index, SpGistState *state, * chain to another leaf page rather than splitting it. */ Assert(!isNew); - moveLeafs(index, state, ¤t, &parent, leafTuple, isnull); + moveLeafs(index, state, ¤t, &parent, leafTuple, isnull[spgKeyColumn]); break; /* we're done */ } else { /* picksplit */ if (doPickSplit(index, state, ¤t, &parent, - leafTuple, level, isnull, isNew)) + leafTuple, level, isnull[spgKeyColumn], isNew)) break; /* doPickSplit installed new tuples */ /* leaf tuple will not be inserted yet */ @@ -2110,8 +2151,8 @@ spgdoinsert(Relation index, SpGistState *state, innerTuple = (SpGistInnerTuple) PageGetItem(current.page, PageGetItemId(current.page, current.offnum)); - in.datum = datum; - in.leafDatum = leafDatum; + in.datum = datum[spgKeyColumn]; + in.leafDatum = leafDatum[spgKeyColumn]; in.level = level; in.allTheSame = innerTuple->allTheSame; in.hasPrefix = (innerTuple->prefixSize > 0); @@ -2121,7 +2162,7 @@ spgdoinsert(Relation index, SpGistState *state, memset(&out, 0, sizeof(out)); - if (!isnull) + if (!isnull[spgKeyColumn]) { /* use user-defined choose method */ FunctionCall2Coll(procinfo, @@ -2158,11 +2199,18 @@ spgdoinsert(Relation index, SpGistState *state, /* Adjust level as per opclass request */ level += out.result.matchNode.levelAdd; /* Replace leafDatum and recompute leafSize */ - if (!isnull) + if (!isnull[spgKeyColumn]) { - leafDatum = out.result.matchNode.restDatum; - leafSize = SGLTHDRSZ + sizeof(ItemIdData) + - SpGistGetTypeSize(&state->attLeafType, leafDatum); + leafDatum[spgKeyColumn] = out.result.matchNode.restDatum; + leafSize = MAXALIGN(sizeof(SpGistLeafTupleData) + + spgNullmaskSize(state->tupleDescriptor->natts, isnull)) + + heap_compute_data_size(state->tupleDescriptor, leafDatum, isnull); + + /* + * check for leafSize < SGDTSIZE is not needed for + * non-null datum + */ + leafSize = MAXALIGN(leafSize) + sizeof(ItemIdData); } /* @@ -2227,6 +2275,6 @@ spgdoinsert(Relation index, SpGistState *state, SpGistSetLastUsedPage(index, parent.buffer); UnlockReleaseBuffer(parent.buffer); } - + pfree(leafDatum); return true; } diff --git a/src/backend/access/spgist/spginsert.c b/src/backend/access/spgist/spginsert.c index 0ca621450e..eeba97328e 100644 --- a/src/backend/access/spgist/spginsert.c +++ b/src/backend/access/spgist/spginsert.c @@ -55,8 +55,7 @@ spgistBuildCallback(Relation index, ItemPointer tid, Datum *values, * lock on some buffer. So we need to be willing to retry. We can flush * any temp data when retrying. */ - while (!spgdoinsert(index, &buildstate->spgstate, tid, - *values, *isnull)) + while (!spgdoinsert(index, &buildstate->spgstate, tid, values, isnull)) { MemoryContextReset(buildstate->tmpCtx); } @@ -227,7 +226,7 @@ spginsert(Relation index, Datum *values, bool *isnull, * to avoid cumulative memory consumption. That means we also have to * redo initSpGistState(), but it's cheap enough not to matter. */ - while (!spgdoinsert(index, &spgstate, ht_ctid, *values, *isnull)) + while (!spgdoinsert(index, &spgstate, ht_ctid, values, isnull)) { MemoryContextReset(insertCtx); initSpGistState(&spgstate, index); diff --git a/src/backend/access/spgist/spgscan.c b/src/backend/access/spgist/spgscan.c index 20e67c3f7d..0ef1bcafd1 100644 --- a/src/backend/access/spgist/spgscan.c +++ b/src/backend/access/spgist/spgscan.c @@ -28,7 +28,8 @@ typedef void (*storeRes_func) (SpGistScanOpaque so, ItemPointer heapPtr, Datum leafValue, bool isNull, bool recheck, - bool recheckDistances, double *distances); + bool recheckDistances, double *distances, + SpGistLeafTuple leafTuple); /* * Pairing heap comparison function for the SpGistSearchItem queue. @@ -88,6 +89,9 @@ spgFreeSearchItem(SpGistScanOpaque so, SpGistSearchItem *item) if (item->traversalValue) pfree(item->traversalValue); + if (item->isLeaf && item->leafTuple) + pfree(item->leafTuple); + pfree(item); } @@ -134,6 +138,8 @@ spgAddStartItem(SpGistScanOpaque so, bool isnull) startEntry->recheck = false; startEntry->recheckDistances = false; + startEntry->leafTuple = NULL; + spgAddSearchItemToQueue(so, startEntry); } @@ -438,14 +444,28 @@ spgendscan(IndexScanDesc scan) * Leaf SpGistSearchItem constructor, called in queue context */ static SpGistSearchItem * -spgNewHeapItem(SpGistScanOpaque so, int level, ItemPointer heapPtr, +spgNewHeapItem(SpGistScanOpaque so, int level, SpGistLeafTuple leafTuple, Datum leafValue, bool recheck, bool recheckDistances, bool isnull, double *distances) { SpGistSearchItem *item = spgAllocSearchItem(so, isnull, distances); + /* + * If there are INCLUDE attributes search item in the queue should contain + * them. + */ + if (so->state.tupleDescriptor->natts > 1) + { + item->leafTuple = palloc(leafTuple->size); + memcpy(item->leafTuple, leafTuple, leafTuple->size); + } + else + { + item->leafTuple = NULL; + } + item->level = level; - item->heapPtr = *heapPtr; + item->heapPtr = leafTuple->heapPtr; /* copy value to queue cxt out of tmp cxt */ item->value = isnull ? (Datum) 0 : datumCopy(leafValue, so->state.attLeafType.attbyval, @@ -503,6 +523,8 @@ spgLeafTest(SpGistScanOpaque so, SpGistSearchItem *item, in.returnData = so->want_itup; in.leafDatum = SGLTDATUM(leafTuple, &so->state); + + out.leafValue = (Datum) 0; out.recheck = false; out.distances = NULL; @@ -528,7 +550,7 @@ spgLeafTest(SpGistScanOpaque so, SpGistSearchItem *item, /* the scan is ordered -> add the item to the queue */ MemoryContext oldCxt = MemoryContextSwitchTo(so->traversalCxt); SpGistSearchItem *heapItem = spgNewHeapItem(so, item->level, - &leafTuple->heapPtr, + leafTuple, leafValue, recheck, recheckDistances, @@ -543,8 +565,10 @@ spgLeafTest(SpGistScanOpaque so, SpGistSearchItem *item, { /* non-ordered scan, so report the item right away */ Assert(!recheckDistances); + storeRes(so, &leafTuple->heapPtr, leafValue, isnull, - recheck, false, NULL); + recheck, false, NULL, leafTuple); + *reportedSome = true; } } @@ -736,7 +760,7 @@ spgTestLeafTuple(SpGistScanOpaque so, /* dead tuple should be first in chain */ Assert(offset == ItemPointerGetOffsetNumber(&item->heapPtr)); /* No live entries on this page */ - Assert(leafTuple->nextOffset == InvalidOffsetNumber); + Assert(SGLT_GET_OFFSET(leafTuple) == InvalidOffsetNumber); return SpGistBreakOffsetNumber; } } @@ -750,7 +774,7 @@ spgTestLeafTuple(SpGistScanOpaque so, spgLeafTest(so, item, leafTuple, isnull, reportedSome, storeRes); - return leafTuple->nextOffset; + return SGLT_GET_OFFSET(leafTuple); } /* @@ -782,8 +806,8 @@ redirect: { /* We store heap items in the queue only in case of ordered search */ Assert(so->numberOfNonNullOrderBys > 0); - storeRes(so, &item->heapPtr, item->value, item->isNull, - item->recheck, item->recheckDistances, item->distances); + storeRes(so, &item->heapPtr, item->value, item->isNull, item->recheck, + item->recheckDistances, item->distances, item->leafTuple); reportedSome = true; } else @@ -877,7 +901,7 @@ redirect: static void storeBitmap(SpGistScanOpaque so, ItemPointer heapPtr, Datum leafValue, bool isnull, bool recheck, bool recheckDistances, - double *distances) + double *distances, SpGistLeafTuple leafTuple) { Assert(!recheckDistances && !distances); tbm_add_tuples(so->tbm, heapPtr, 1, recheck); @@ -904,7 +928,7 @@ spggetbitmap(IndexScanDesc scan, TIDBitmap *tbm) static void storeGettuple(SpGistScanOpaque so, ItemPointer heapPtr, Datum leafValue, bool isnull, bool recheck, bool recheckDistances, - double *nonNullDistances) + double *nonNullDistances, SpGistLeafTuple leafTuple) { Assert(so->nPtrs < MaxIndexTuplesPerPage); so->heapPtrs[so->nPtrs] = *heapPtr; @@ -949,9 +973,44 @@ storeGettuple(SpGistScanOpaque so, ItemPointer heapPtr, * Reconstruct index data. We have to copy the datum out of the temp * context anyway, so we may as well create the tuple here. */ - so->reconTups[so->nPtrs] = heap_form_tuple(so->indexTupDesc, - &leafValue, - &isnull); + if (so->state.tupleDescriptor->natts > 1) + { + /* + * A general case of one key attribute and several INCLUDE + * attributes + */ + Datum *leafDatums; + bool *leafIsnulls; + + leafDatums = (Datum *) palloc(sizeof(Datum) * (so->state.tupleDescriptor->natts)); + leafIsnulls = (bool *) palloc(sizeof(bool) * (so->state.tupleDescriptor->natts)); + + spgDeformLeafTuple(leafTuple, so->state.tupleDescriptor, leafDatums, leafIsnulls, isnull); + + /* + * Override key value extracted from LeafTuple in case we've + * reconstructed it already + */ + leafDatums[spgKeyColumn] = leafValue; + leafIsnulls[spgKeyColumn] = isnull; + + so->reconTups[so->nPtrs] = heap_form_tuple(so->indexTupDesc, + leafDatums, + leafIsnulls); + pfree(leafDatums); + pfree(leafIsnulls); + } + else + { + /* + * A particular case of no include attributes works exactly like + * more general case above but don't make redundant allocations + * and rewrites when there is only one attribute. + */ + so->reconTups[so->nPtrs] = heap_form_tuple(so->indexTupDesc, + &leafValue, + &isnull); + } } so->nPtrs++; } @@ -1019,6 +1078,10 @@ spgcanreturn(Relation index, int attno) { SpGistCache *cache; + /* INCLUDE attributes can always be fetched for index-only scans */ + if (attno > 1) + return true; + /* We can do it if the opclass config function says so */ cache = spgGetCache(index); diff --git a/src/backend/access/spgist/spgutils.c b/src/backend/access/spgist/spgutils.c index d8b1815061..e0284d50eb 100644 --- a/src/backend/access/spgist/spgutils.c +++ b/src/backend/access/spgist/spgutils.c @@ -31,7 +31,11 @@ #include "utils/index_selfuncs.h" #include "utils/lsyscache.h" #include "utils/syscache.h" - +#include "access/itup.h" +#include "access/detoast.h" +#include "access/toast_internals.h" +#include "access/heaptoast.h" +#include "utils/expandeddatum.h" /* * SP-GiST handler function: return IndexAmRoutine with access method parameters @@ -57,7 +61,7 @@ spghandler(PG_FUNCTION_ARGS) amroutine->amclusterable = false; amroutine->ampredlocks = false; amroutine->amcanparallel = false; - amroutine->amcaninclude = false; + amroutine->amcaninclude = true; amroutine->amusemaintenanceworkmem = false; amroutine->amparallelvacuumoptions = VACUUM_OPTION_PARALLEL_BULKDEL | VACUUM_OPTION_PARALLEL_COND_CLEANUP; @@ -105,6 +109,8 @@ SpGistCache * spgGetCache(Relation index) { SpGistCache *cache; + int i; + int natts = IndexRelationGetNumberOfAttributes(index); if (index->rd_amcache == NULL) { @@ -113,18 +119,42 @@ spgGetCache(Relation index) FmgrInfo *procinfo; Buffer metabuffer; SpGistMetaPageData *metadata; + TupleDesc tmpTupleDescriptor; + + /* + * SPGiST should have one key column and can also have INCLUDE columns + */ + if (IndexRelationGetNumberOfKeyAttributes(index) != 1) + ereport(ERROR, + (errcode(ERRCODE_INVALID_PARAMETER_VALUE), + errmsg("SPGiST index can have only one key column"))); + if (natts >= INDEX_MAX_KEYS) + ereport(ERROR, + (errcode(ERRCODE_TOO_MANY_COLUMNS), + errmsg("number of index columns (%d) exceeds limit (%d)", + natts, INDEX_MAX_KEYS))); + + /* Form a tuple descriptor for all columns */ + tmpTupleDescriptor = CreateTemplateTupleDesc(natts); + + for (i = spgKeyColumn; i < natts; i++) + TupleDescInitEntry(tmpTupleDescriptor, i + 1, NULL, + TupleDescAttr(index->rd_att, i)->atttypid, -1, 0); cache = MemoryContextAllocZero(index->rd_indexcxt, - sizeof(SpGistCache)); + offsetof(struct SpGistCache, tupleDescriptor) + + TupleDescSize(tmpTupleDescriptor)); - /* SPGiST doesn't support multi-column indexes */ - Assert(index->rd_att->natts == 1); + memcpy(&cache->tupleDescriptor, tmpTupleDescriptor, + TupleDescSize(tmpTupleDescriptor)); + pfree(tmpTupleDescriptor); /* - * Get the actual data type of the indexed column from the index - * tupdesc. We pass this to the opclass config function so that - * polymorphic opclasses are possible. + * Get the actual data type of the key column from the index tupdesc. + * We pass this to the opclass config function so that polymorphic + * opclasses are possible. */ + atttype = TupleDescAttr(index->rd_att, 0)->atttypid; /* Call the config function to get config info for the opclass */ @@ -196,6 +226,7 @@ initSpGistState(SpGistState *state, Relation index) state->attLeafType = cache->attLeafType; state->attPrefixType = cache->attPrefixType; state->attLabelType = cache->attLabelType; + state->tupleDescriptor = &cache->tupleDescriptor; /* Make workspace for constructing dead tuples */ state->deadTupleStorage = palloc0(SGDTSIZE); @@ -205,6 +236,7 @@ initSpGistState(SpGistState *state, Relation index) /* Assume we're not in an index build (spgbuild will override) */ state->isBuild = false; + } /* @@ -604,8 +636,8 @@ spgoptions(Datum reloptions, bool validate) /* * Get the space needed to store a non-null datum of the indicated type. - * Note the result is already rounded up to a MAXALIGN boundary. - * Also, we follow the SPGiST convention that pass-by-val types are + * Note the result is not maxaligned and this should be done by the caller if + * needed. Also, we follow the SPGiST convention that pass-by-val types are * just stored in their Datum representation (compare memcpyDatum). */ unsigned int @@ -620,7 +652,7 @@ SpGistGetTypeSize(SpGistTypeDesc *att, Datum datum) else size = VARSIZE_ANY(datum); - return MAXALIGN(size); + return size; } /* @@ -642,36 +674,85 @@ memcpyDatum(void *target, SpGistTypeDesc *att, Datum datum) } } +Size +spgNullmaskSize(int natts, bool *isnull) +{ + int i; + + Assert(natts > 0); + Assert(natts <= INDEX_MAX_KEYS); + + /* + * If there is only a key attribute (natts == 1), nullmask will not be + * inserted (even inside the tuples, which have NULL key value). This + * ensures compatibility with the previous versions tuple layout. + */ + if (natts == 1) + return 0; + + for (i = spgKeyColumn; i < natts; i++) + { + /* + * If there is at least one null, then nullmask will be sized to + * contain a key attribute and all INCLUDE attributes. + */ + if (isnull[i]) + return ((natts - 1) / 8) + 1; + } + return 0; +} + /* - * Construct a leaf tuple containing the given heap TID and datum value + * Construct a leaf tuple containing the given heap TID, datums and isnulls arrays. + * Nullmask apply only to INCLUDE attribute and is placed just after header if + * there is at least one NULL among INCLUDE attributes. It doesn't need alignment. + * Then all attributes data follow starting from MAXALIGN. */ SpGistLeafTuple spgFormLeafTuple(SpGistState *state, ItemPointer heapPtr, - Datum datum, bool isnull) + Datum *datum, bool *isnull) { SpGistLeafTuple tup; - unsigned int size; - - /* compute space needed (note result is already maxaligned) */ - size = SGLTHDRSZ; - if (!isnull) - size += SpGistGetTypeSize(&state->attLeafType, datum); + TupleDesc tupleDescriptor = state->tupleDescriptor; + uint16 tupmask = 0; + Size data_size = heap_compute_data_size(tupleDescriptor, datum, isnull); + Size nullmask_size = spgNullmaskSize(tupleDescriptor->natts, isnull); + Size hoff = MAXALIGN(sizeof(SpGistLeafTupleData) + nullmask_size); + Size size = MAXALIGN(hoff + data_size); + char *tp; /* ptr to tuple data */ /* - * Ensure that we can replace the tuple with a dead tuple later. This - * test is unnecessary when !isnull, but let's be safe. + * Ensure that we can replace the tuple with a dead tuple later. This test + * is unnecessary when !isnull[spgKeyColumn], but let's be safe. */ - if (size < SGDTSIZE) - size = SGDTSIZE; + size = size < SGDTSIZE ? SGDTSIZE : MAXALIGN(size); - /* OK, form the tuple */ tup = (SpGistLeafTuple) palloc0(size); - tup->size = size; - tup->nextOffset = InvalidOffsetNumber; + SGLT_SET_OFFSET(tup, InvalidOffsetNumber); tup->heapPtr = *heapPtr; - if (!isnull) - memcpyDatum(SGLTDATAPTR(tup), &state->attLeafType, datum); + tp = (char *) tup + hoff; + + if (nullmask_size) + { + bits8 *bp; /* ptr to null bitmap in tuple */ + + /* Set nullmask presence bit in SpGistLeafTuple header if needed */ + SGLT_SET_CONTAINSNULLMASK(tup, true); + /* Fill nullmask and data part of a tuple */ + bp = (bits8 *) ((char *) tup + sizeof(SpGistLeafTupleData)); + heap_fill_tuple(tupleDescriptor, datum, isnull, tp, data_size, &tupmask, bp); + } + else if (tupleDescriptor->natts > 1 || isnull[0] == false) + + /* + * Prevent filling nullmask in the tuple in case there should be no + * nullmask. + */ + heap_fill_tuple(tupleDescriptor, datum, isnull, tp, data_size, &tupmask, + (bits8 *) NULL); + + /* Single-column tuple with NULL value doesn't need filling data portion */ return tup; } @@ -689,10 +770,10 @@ spgFormNodeTuple(SpGistState *state, Datum label, bool isnull) unsigned int size; unsigned short infomask = 0; - /* compute space needed (note result is already maxaligned) */ + /* compute space needed */ size = SGNTHDRSZ; if (!isnull) - size += SpGistGetTypeSize(&state->attLabelType, label); + size += MAXALIGN(SpGistGetTypeSize(&state->attLabelType, label)); /* * Here we make sure that the size will fit in the field reserved for it @@ -736,7 +817,7 @@ spgFormInnerTuple(SpGistState *state, bool hasPrefix, Datum prefix, /* Compute size needed */ if (hasPrefix) - prefixSize = SpGistGetTypeSize(&state->attPrefixType, prefix); + prefixSize = MAXALIGN(SpGistGetTypeSize(&state->attPrefixType, prefix)); else prefixSize = 0; @@ -815,7 +896,7 @@ spgFormDeadTuple(SpGistState *state, int tupstate, tuple->tupstate = tupstate; tuple->size = SGDTSIZE; - tuple->nextOffset = InvalidOffsetNumber; + tuple->t_info = InvalidOffsetNumber; if (tupstate == SPGIST_REDIRECT) { @@ -1047,3 +1128,49 @@ spgproperty(Oid index_oid, int attno, return true; } + +/* + * Convert an SpGist tuple into palloc'd Datum/isnull arrays. + */ +void +spgDeformLeafTuple(SpGistLeafTuple tup, TupleDesc tupleDescriptor, + Datum *values, bool *isnull, bool keyColumnIsNull) +{ + bool hasNullsMask = SGLT_GET_CONTAINSNULLMASK(tup); + char *tp; /* ptr to tuple data */ + bits8 *bp; /* ptr to null bitmap in tuple */ + + if (keyColumnIsNull && tupleDescriptor->natts == 1) + { + /* + * Trivial case: there is only key attribute and we're in a nulls + * tree. hasNullsMask bit in a tuple header should not be set for + * single attribute case even if it has NULL value (for compatibility + * with pre-v14 SpGist tuple format) We should not call + * index_deform_anyheader_tuple() in this trivial case as it expects + * nullmask in a tuple present in this case. + */ + Assert(hasNullsMask == 0); + + isnull[spgKeyColumn] = true; + values[spgKeyColumn] = (Datum) 0; + return; + } + else if (hasNullsMask) + tp = (char *) tup + MAXALIGN(sizeof(SpGistLeafTupleData) + + ((tupleDescriptor->natts - 1) / 8 + 1)); + else + tp = (char *) tup + MAXALIGN(sizeof(SpGistLeafTupleData)); + + bp = (bits8 *) ((char *) tup + sizeof(SpGistLeafTupleData)); + + index_deform_anyheader_tuple((char *) tup, tupleDescriptor, + values, isnull, + bp, tp, hasNullsMask); + + /* + * Key column isnull value from a tuple should be consistent with + * keyColumnIsNull got from the caller + */ + Assert(keyColumnIsNull == isnull[spgKeyColumn]); +} diff --git a/src/backend/access/spgist/spgvacuum.c b/src/backend/access/spgist/spgvacuum.c index a9ffca5183..684efa7cd6 100644 --- a/src/backend/access/spgist/spgvacuum.c +++ b/src/backend/access/spgist/spgvacuum.c @@ -168,23 +168,28 @@ vacuumLeafPage(spgBulkDeleteState *bds, Relation index, Buffer buffer, } /* Form predecessor map, too */ - if (lt->nextOffset != InvalidOffsetNumber) + if (SGLT_GET_OFFSET(lt) != InvalidOffsetNumber) { /* paranoia about corrupted chain links */ - if (lt->nextOffset < FirstOffsetNumber || - lt->nextOffset > max || - predecessor[lt->nextOffset] != InvalidOffsetNumber) + if (SGLT_GET_OFFSET(lt) < FirstOffsetNumber || + SGLT_GET_OFFSET(lt) > max || + predecessor[SGLT_GET_OFFSET(lt)] != InvalidOffsetNumber) elog(ERROR, "inconsistent tuple chain links in page %u of index \"%s\"", BufferGetBlockNumber(buffer), RelationGetRelationName(index)); - predecessor[lt->nextOffset] = i; + predecessor[SGLT_GET_OFFSET(lt)] = i; } } else if (lt->tupstate == SPGIST_REDIRECT) { SpGistDeadTuple dt = (SpGistDeadTuple) lt; - Assert(dt->nextOffset == InvalidOffsetNumber); + /* + * Dead tuple nextOffset is allowed to have any values of two + * highest bits in case it is inherited from SpGistLeafTuple where + * these bits have their own meaning. + */ + Assert(SGLT_GET_OFFSET(dt) == InvalidOffsetNumber); Assert(ItemPointerIsValid(&dt->pointer)); /* @@ -201,7 +206,7 @@ vacuumLeafPage(spgBulkDeleteState *bds, Relation index, Buffer buffer, } else { - Assert(lt->nextOffset == InvalidOffsetNumber); + Assert(SGLT_GET_OFFSET(lt) == InvalidOffsetNumber); } } @@ -250,7 +255,7 @@ vacuumLeafPage(spgBulkDeleteState *bds, Relation index, Buffer buffer, prevLive = deletable[i] ? InvalidOffsetNumber : i; /* scan down the chain ... */ - j = head->nextOffset; + j = SGLT_GET_OFFSET(head); while (j != InvalidOffsetNumber) { SpGistLeafTuple lt; @@ -301,7 +306,7 @@ vacuumLeafPage(spgBulkDeleteState *bds, Relation index, Buffer buffer, interveningDeletable = false; } - j = lt->nextOffset; + j = SGLT_GET_OFFSET(lt); } if (prevLive == InvalidOffsetNumber) @@ -366,7 +371,7 @@ vacuumLeafPage(spgBulkDeleteState *bds, Relation index, Buffer buffer, lt = (SpGistLeafTuple) PageGetItem(page, PageGetItemId(page, chainSrc[i])); Assert(lt->tupstate == SPGIST_LIVE); - lt->nextOffset = chainDest[i]; + SGLT_SET_OFFSET(lt, chainDest[i]); } MarkBufferDirty(buffer); diff --git a/src/backend/access/spgist/spgxlog.c b/src/backend/access/spgist/spgxlog.c index d40c7b5877..a1d8686907 100644 --- a/src/backend/access/spgist/spgxlog.c +++ b/src/backend/access/spgist/spgxlog.c @@ -122,8 +122,8 @@ spgRedoAddLeaf(XLogReaderState *record) head = (SpGistLeafTuple) PageGetItem(page, PageGetItemId(page, xldata->offnumHeadLeaf)); - Assert(head->nextOffset == leafTupleHdr.nextOffset); - head->nextOffset = xldata->offnumLeaf; + Assert(SGLT_GET_OFFSET(head) == SGLT_GET_OFFSET(&leafTupleHdr)); + SGLT_SET_OFFSET(head, xldata->offnumLeaf); } } else @@ -822,7 +822,7 @@ spgRedoVacuumLeaf(XLogReaderState *record) lt = (SpGistLeafTuple) PageGetItem(page, PageGetItemId(page, chainSrc[i])); Assert(lt->tupstate == SPGIST_LIVE); - lt->nextOffset = chainDest[i]; + SGLT_SET_OFFSET(lt, chainDest[i]); } PageSetLSN(page, lsn); diff --git a/src/include/access/itup.h b/src/include/access/itup.h index b6813707d0..09d947a996 100644 --- a/src/include/access/itup.h +++ b/src/include/access/itup.h @@ -154,6 +154,9 @@ extern Datum nocache_index_getattr(IndexTuple tup, int attnum, TupleDesc tupleDesc); extern void index_deform_tuple(IndexTuple tup, TupleDesc tupleDescriptor, Datum *values, bool *isnull); +extern void index_deform_anyheader_tuple(char *tup, TupleDesc tupleDescriptor, + Datum *values, bool *isnull, + bits8 *bp, char *tp, bool hasnulls); extern IndexTuple CopyIndexTuple(IndexTuple source); extern IndexTuple index_truncate_tuple(TupleDesc sourceDescriptor, IndexTuple source, int leavenatts); diff --git a/src/include/access/spgist_private.h b/src/include/access/spgist_private.h index a81bab24ea..530f8cccd8 100644 --- a/src/include/access/spgist_private.h +++ b/src/include/access/spgist_private.h @@ -22,13 +22,15 @@ #include "utils/geo_decls.h" #include "utils/relcache.h" - typedef struct SpGistOptions { int32 varlena_header_; /* varlena header (do not touch directly!) */ int fillfactor; /* page fill factor in percent (0..100) */ } SpGistOptions; +#define spgKeyColumn 0 +#define spgFirstIncludeColumn 1 + #define SpGistGetFillFactor(relation) \ (AssertMacro(relation->rd_rel->relkind == RELKIND_INDEX && \ relation->rd_rel->relam == SPGIST_AM_OID), \ @@ -146,28 +148,9 @@ typedef struct SpGistState TransactionId myXid; /* XID to use when creating a redirect tuple */ bool isBuild; /* true if doing index build */ + TupleDesc tupleDescriptor; /* tuple descriptor */ } SpGistState; -typedef struct SpGistSearchItem -{ - pairingheap_node phNode; /* pairing heap node */ - Datum value; /* value reconstructed from parent or - * leafValue if heaptuple */ - void *traversalValue; /* opclass-specific traverse value */ - int level; /* level of items on this page */ - ItemPointerData heapPtr; /* heap info, if heap tuple */ - bool isNull; /* SearchItem is NULL item */ - bool isLeaf; /* SearchItem is heap item */ - bool recheck; /* qual recheck is needed */ - bool recheckDistances; /* distance recheck is needed */ - - /* array with numberOfOrderBys entries */ - double distances[FLEXIBLE_ARRAY_MEMBER]; -} SpGistSearchItem; - -#define SizeOfSpGistSearchItem(n_distances) \ - (offsetof(SpGistSearchItem, distances) + sizeof(double) * (n_distances)) - /* * Private state of an index scan */ @@ -243,9 +226,9 @@ typedef struct SpGistCache SpGistTypeDesc attLabelType; /* type of node label values */ SpGistLUPCache lastUsedPages; /* local storage of last-used info */ + TupleDescData tupleDescriptor; /* descriptor for leaf tuples */ } SpGistCache; - /* * SPGiST tuple types. Note: inner, leaf, and dead tuple structs * must have the same tupstate field in the same position! Real inner and @@ -305,8 +288,8 @@ typedef SpGistInnerTupleData *SpGistInnerTuple; * SPGiST node tuple: one node within an inner tuple * * Node tuples use the same header as ordinary Postgres IndexTuples, but - * we do not use a null bitmap, because we know there is only one column - * so the INDEX_NULL_MASK bit suffices. Also, pass-by-value datums are + * we do not use a null bitmap, because we know there is only one key column + * so the INDEX_NULL_MASK bit suffices. Also, pass-by-value datums are * stored as a full Datum, the same convention as for inner tuple prefixes * and leaf tuple datums. */ @@ -322,23 +305,21 @@ typedef SpGistNodeTupleData *SpGistNodeTuple; PointerGetDatum(SGNTDATAPTR(x))) /* - * SPGiST leaf tuple: carries a datum and a heap tuple TID + * SPGiST leaf tuple: carries a heap tuple TID and columns datums and + * nullmasks. * - * In the simplest case, the datum is the same as the indexed value; but + * In the simplest case, the key datum is the same as the indexed value; but * it could also be a suffix or some other sort of delta that permits * reconstruction given knowledge of the prefix path traversed to get here. + * Datums of INCLUDE columns are stored without modification. * * The size field is wider than could possibly be needed for an on-disk leaf * tuple, but this allows us to form leaf tuples even when the datum is too * wide to be stored immediately, and it costs nothing because of alignment * considerations. * - * Normally, nextOffset links to the next tuple belonging to the same parent - * node (which must be on the same page). But when the root page is a leaf - * page, we don't chain its tuples, so nextOffset is always 0 on the root. - * * size must be a multiple of MAXALIGN; also, it must be at least SGDTSIZE - * so that the tuple can be converted to REDIRECT status later. (This + * so that the tuple can be converted to REDIRECT status later. (This * restriction only adds bytes for the null-datum case, otherwise alignment * restrictions force it anyway.) * @@ -346,23 +327,65 @@ typedef SpGistNodeTupleData *SpGistNodeTuple; * however, the SGDTSIZE limit ensures that's there's a Datum word there * anyway, so SGLTDATUM can be applied safely as long as you don't do * anything with the result. + * + * Normally, nextOffset inside t_info links to the next tuple belonging to + * the same parent node (which must be on the same page). But when the root + * page is a leaf page, we don't chain its tuples, so nextOffset is always 0 + * on the root. Minimum space to store SpGistLeafTuple plus ItemIdData on a + * page is 16 bytes, so 15 lower bits for nextOffset is enough to store tuple + * number in a chain on a page even if a page size is 64Kb. + * + * The highest bit in t_info is to store per-tuple information is there nulls + * mask exist for the case there are INCLUDE attributes. If there are no + * INCLUDE columns this bit is set to 0 and nullmask is not added even if it + * is an empty tuple with NULL key value. + * + * Datums for all columns are stored in ordinary index-tuple-like way starting + * from MAXALIGN boundary. Nullmask with size (number of columns)/8 + * bytes is put without alignment just after the ending of tuple header. + * On 64-bit architecture nullmask has a good chance to fit into the alignment + * gap between the header and the first datum, thus making its storage free + * of charge. */ + typedef struct SpGistLeafTupleData { unsigned int tupstate:2, /* LIVE/REDIRECT/DEAD/PLACEHOLDER */ size:30; /* large enough for any palloc'able value */ - OffsetNumber nextOffset; /* next tuple in chain, or InvalidOffsetNumber */ + + /* --------------- + * t_info is laid out in the following fashion: + * + * 15th (high) bit: values has nulls + * 14-0 bit: nextOffset i.e. number of next tuple in chain on a page, + * or InvalidOffsetNumber + * --------------- + */ + unsigned short t_info; /* nextOffset for linking tuples in a chain on + * a leaf page, and additional info */ ItemPointerData heapPtr; /* TID of represented heap tuple */ - /* leaf datum follows */ + /* nullmask follows if there are nulls among attributes */ + /* attributes data follow starting from MAXALIGN */ } SpGistLeafTupleData; typedef SpGistLeafTupleData *SpGistLeafTuple; #define SGLTHDRSZ MAXALIGN(sizeof(SpGistLeafTupleData)) #define SGLTDATAPTR(x) (((char *) (x)) + SGLTHDRSZ) -#define SGLTDATUM(x, s) ((s)->attLeafType.attbyval ? \ - *(Datum *) SGLTDATAPTR(x) : \ - PointerGetDatum(SGLTDATAPTR(x))) +#define SGLTDATUM(x, s) fetch_att(SGLTDATAPTR(x), (s)->attLeafType.attbyval, \ + (s)->attLeafType.attlen) +/* + * Macros to access nextOffset and bit fields inside t_info independently. + */ +#define SGLT_GET_OFFSET(spgLeafTuple) ( (spgLeafTuple)->t_info & 0x3FFF ) +#define SGLT_GET_CONTAINSNULLMASK(spgLeafTuple) \ + ( (bool)((spgLeafTuple)->t_info >> 15) ) +#define SGLT_SET_OFFSET(spgLeafTuple, offsetNumber) \ + ( (spgLeafTuple)->t_info = \ + ((spgLeafTuple)->t_info & 0xC000) | ((offsetNumber) & 0x3FFF) ) +#define SGLT_SET_CONTAINSNULLMASK(spgLeafTuple, is_null) \ + ( (spgLeafTuple)->t_info = \ + ((uint16)(bool)(is_null) << 15) | ((spgLeafTuple)->t_info & 0x3FFF) ) /* * SPGiST dead tuple: declaration for examining non-live tuples @@ -372,14 +395,14 @@ typedef SpGistLeafTupleData *SpGistLeafTuple; * Also, the pointer field must be in the same place as a leaf tuple's heapPtr * field, to satisfy some Asserts that we make when replacing a leaf tuple * with a dead tuple. - * We don't use nextOffset, but it's needed to align the pointer field. + * We don't use t_info, but it's needed to align the pointer field. * pointer and xid are only valid when tupstate = REDIRECT. */ typedef struct SpGistDeadTupleData { unsigned int tupstate:2, /* LIVE/REDIRECT/DEAD/PLACEHOLDER */ size:30; - OffsetNumber nextOffset; /* not used in dead tuples */ + unsigned short t_info; /* not used in dead tuples */ ItemPointerData pointer; /* redirection inside index */ TransactionId xid; /* ID of xact that inserted this tuple */ } SpGistDeadTupleData; @@ -394,7 +417,6 @@ typedef SpGistDeadTupleData *SpGistDeadTuple; * size plus sizeof(ItemIdData) (for the line pointer). This works correctly * so long as tuple sizes are always maxaligned. */ - /* Page capacity after allowing for fixed header and special space */ #define SPGIST_PAGE_CAPACITY \ MAXALIGN_DOWN(BLCKSZ - \ @@ -410,6 +432,27 @@ typedef SpGistDeadTupleData *SpGistDeadTuple; Min(SpGistPageGetOpaque(p)->nPlaceholder, n) * \ (SGDTSIZE + sizeof(ItemIdData))) + +typedef struct SpGistSearchItem +{ + pairingheap_node phNode; /* pairing heap node */ + Datum value; /* value reconstructed from parent or + * leafValue if heaptuple */ + void *traversalValue; /* opclass-specific traverse value */ + int level; /* level of items on this page */ + ItemPointerData heapPtr; /* heap info, if heap tuple */ + bool isNull; /* SearchItem is NULL item */ + bool isLeaf; /* SearchItem is heap item */ + bool recheck; /* qual recheck is needed */ + bool recheckDistances; /* distance recheck is needed */ + SpGistLeafTuple leafTuple; + /* array with numberOfOrderBys entries */ + double distances[FLEXIBLE_ARRAY_MEMBER]; +} SpGistSearchItem; + +#define SizeOfSpGistSearchItem(n_distances) \ + (offsetof(SpGistSearchItem, distances) + sizeof(double) * (n_distances)) + /* * XLOG stuff */ @@ -456,9 +499,10 @@ extern void SpGistInitPage(Page page, uint16 f); extern void SpGistInitBuffer(Buffer b, uint16 f); extern void SpGistInitMetapage(Page page); extern unsigned int SpGistGetTypeSize(SpGistTypeDesc *att, Datum datum); +extern Size spgNullmaskSize(int natts, bool *isnull); extern SpGistLeafTuple spgFormLeafTuple(SpGistState *state, ItemPointer heapPtr, - Datum datum, bool isnull); + Datum *datum, bool *isnull); extern SpGistNodeTuple spgFormNodeTuple(SpGistState *state, Datum label, bool isnull); extern SpGistInnerTuple spgFormInnerTuple(SpGistState *state, @@ -466,6 +510,8 @@ extern SpGistInnerTuple spgFormInnerTuple(SpGistState *state, int nNodes, SpGistNodeTuple *nodes); extern SpGistDeadTuple spgFormDeadTuple(SpGistState *state, int tupstate, BlockNumber blkno, OffsetNumber offnum); +extern void spgDeformLeafTuple(SpGistLeafTuple tup, TupleDesc tupleDescriptor, + Datum *datum, bool *isnull, bool keyIsNull); extern Datum *spgExtractNodeLabels(SpGistState *state, SpGistInnerTuple innerTuple); extern OffsetNumber SpGistPageAddNewItem(SpGistState *state, Page page, @@ -484,7 +530,7 @@ extern void spgPageIndexMultiDelete(SpGistState *state, Page page, int firststate, int reststate, BlockNumber blkno, OffsetNumber offnum); extern bool spgdoinsert(Relation index, SpGistState *state, - ItemPointer heapPtr, Datum datum, bool isnull); + ItemPointer heapPtr, Datum *datum, bool *isnull); /* spgproc.c */ extern double *spg_key_orderbys_distances(Datum key, bool isLeaf, diff --git a/src/test/regress/expected/amutils.out b/src/test/regress/expected/amutils.out index d92a6d12c6..7ab6113c61 100644 --- a/src/test/regress/expected/amutils.out +++ b/src/test/regress/expected/amutils.out @@ -171,7 +171,7 @@ select amname, prop, pg_indexam_has_property(a.oid, prop) as p spgist | can_unique | f spgist | can_multi_col | f spgist | can_exclude | t - spgist | can_include | f + spgist | can_include | t spgist | bogus | (36 rows) diff --git a/src/test/regress/expected/index_including.out b/src/test/regress/expected/index_including.out index 8e5d53e712..86510687c7 100644 --- a/src/test/regress/expected/index_including.out +++ b/src/test/regress/expected/index_including.out @@ -349,14 +349,13 @@ SELECT indexdef FROM pg_indexes WHERE tablename = 'tbl' ORDER BY indexname; DROP TABLE tbl; /* - * 7. Check various AMs. All but btree and gist must fail. + * 7. Check various AMs. All but btree, gist and spgist must fail. */ CREATE TABLE tbl (c1 int,c2 int, c3 box, c4 box); CREATE INDEX on tbl USING brin(c1, c2) INCLUDE (c3, c4); ERROR: access method "brin" does not support included columns CREATE INDEX on tbl USING gist(c3) INCLUDE (c1, c4); CREATE INDEX on tbl USING spgist(c3) INCLUDE (c4); -ERROR: access method "spgist" does not support included columns CREATE INDEX on tbl USING gin(c1, c2) INCLUDE (c3, c4); ERROR: access method "gin" does not support included columns CREATE INDEX on tbl USING hash(c1, c2) INCLUDE (c3, c4); diff --git a/src/test/regress/parallel_schedule b/src/test/regress/parallel_schedule index b3605db88c..5fa57d4201 100644 --- a/src/test/regress/parallel_schedule +++ b/src/test/regress/parallel_schedule @@ -52,7 +52,7 @@ test: copy copyselect copydml insert insert_conflict # ---------- test: create_misc create_operator create_procedure # These depend on create_misc and create_operator -test: create_index create_index_spgist create_view index_including index_including_gist +test: create_index create_index_spgist create_view index_including index_including_gist index_including_spgist # ---------- # Another group of parallel tests diff --git a/src/test/regress/serial_schedule b/src/test/regress/serial_schedule index 15ec7548e3..d89d21187b 100644 --- a/src/test/regress/serial_schedule +++ b/src/test/regress/serial_schedule @@ -68,6 +68,7 @@ test: create_index_spgist test: create_view test: index_including test: index_including_gist +test: index_including_spgist test: create_aggregate test: create_function_3 test: create_cast diff --git a/src/test/regress/sql/index_including.sql b/src/test/regress/sql/index_including.sql index 7e517483ad..44b340053b 100644 --- a/src/test/regress/sql/index_including.sql +++ b/src/test/regress/sql/index_including.sql @@ -182,7 +182,7 @@ SELECT indexdef FROM pg_indexes WHERE tablename = 'tbl' ORDER BY indexname; DROP TABLE tbl; /* - * 7. Check various AMs. All but btree and gist must fail. + * 7. Check various AMs. All but btree, gist and spgist must fail. */ CREATE TABLE tbl (c1 int,c2 int, c3 box, c4 box); CREATE INDEX on tbl USING brin(c1, c2) INCLUDE (c3, c4); -- 2.28.0