Compare commits

..

21 Commits

Author SHA1 Message Date
Andrew Kane
5934aeb30d Skip more work 2023-10-05 10:37:57 -07:00
Andrew Kane
115c2bd91d Fixed alignment [skip ci] 2023-10-04 23:09:31 -07:00
Andrew Kane
97ac01773d Speed up HNSW index build 2023-10-04 19:43:25 -07:00
Andrew Kane
8d7abb6590 Revert "Fixed locking for index scans for HNSW - #256"
This reverts commit d032726976.
2023-09-26 23:00:14 -07:00
jeff-davis
b247b688a8 No need to MarkBufferDirty(); GenericXLogFinish() does that. (#265) 2023-09-15 13:14:10 -07:00
Andrew Kane
9672446a4c Updated order [skip ci] 2023-09-12 19:53:12 -07:00
Andrew Kane
334614b7f7 Added HnswFreeNeighbors function [skip ci] 2023-09-12 19:49:31 -07:00
Andrew Kane
643eacd9dc Improved variable name [skip ci] 2023-09-12 19:37:42 -07:00
Andrew Kane
bca50a03fa Use consistent variable name 2023-09-12 19:24:31 -07:00
Andrew Kane
d87833cacc Improved variable scoping [skip ci] 2023-09-12 19:16:55 -07:00
Andrew Kane
4c01073ac1 Improved variable scoping [skip ci] 2023-09-12 18:43:43 -07:00
Andrew Kane
6fed8f1e78 Improved types and scoping for k-means [skip ci] 2023-09-12 18:37:56 -07:00
Andrew Kane
611f5b1047 Improved variable scoping [skip ci] 2023-09-12 18:35:53 -07:00
Andrew Kane
e367155732 Improved types and scoping for k-means 2023-09-12 18:33:31 -07:00
Andrew Kane
466c556b1a Improved variable scoping [skip ci] 2023-09-12 18:24:46 -07:00
Andrew Kane
75e09265d6 Improved variable scoping [skip ci] 2023-09-12 18:14:20 -07:00
Andrew Kane
77c5070fb6 Improved variable scoping [skip ci] 2023-09-12 16:50:51 -07:00
Andrew Kane
1006fdf3f2 Improved variable scoping [skip ci] 2023-09-12 16:44:09 -07:00
Andrew Kane
4209c9b3af Improved variable scoping 2023-09-12 16:13:10 -07:00
Andrew Kane
ec0bb4e4ff Improved code 2023-09-12 15:43:28 -07:00
Andrew Kane
b164833933 Removed pinning for HNSW index scan 2023-09-11 12:12:28 -07:00
14 changed files with 217 additions and 140 deletions

View File

@@ -1,6 +1,6 @@
## 0.5.1 (unreleased)
- Fixed locking for index scans for HNSW
- Improved performance of index scans for IVFFlat after updates and deletes
## 0.5.0 (2023-08-28)

View File

@@ -110,11 +110,13 @@ typedef struct HnswCandidate
{
HnswElement element;
float distance;
bool closer;
} HnswCandidate;
typedef struct HnswNeighborArray
{
int length;
bool closerSet;
HnswCandidate *items;
} HnswNeighborArray;
@@ -218,7 +220,6 @@ typedef HnswNeighborTupleData * HnswNeighborTuple;
typedef struct HnswScanOpaqueData
{
bool first;
Buffer buf;
List *w;
MemoryContext tmpCtx;
@@ -259,7 +260,7 @@ typedef struct HnswVacuumState
/* Methods */
int HnswGetM(Relation index);
int HnswGetEfConstruction(Relation index);
FmgrInfo *HnswOptionalProcInfo(Relation rel, uint16 procnum);
FmgrInfo *HnswOptionalProcInfo(Relation index, uint16 procnum);
bool HnswNormValue(FmgrInfo *procinfo, Oid collation, Datum *value, Vector * result);
void HnswCommitBuffer(Buffer buf, GenericXLogState *state);
Buffer HnswNewBuffer(Relation index, ForkNumber forkNum);

View File

@@ -81,7 +81,6 @@ HnswBuildAppendPage(Relation index, Buffer *buf, Page *page, GenericXLogState **
HnswPageGetOpaque(*page)->nextblkno = BufferGetBlockNumber(newbuf);
/* Commit */
MarkBufferDirty(*buf);
GenericXLogFinish(*state);
UnlockReleaseBuffer(*buf);
@@ -179,7 +178,6 @@ CreateElementPages(HnswBuildState * buildstate)
insertPage = BufferGetBlockNumber(buf);
/* Commit */
MarkBufferDirty(buf);
GenericXLogFinish(state);
UnlockReleaseBuffer(buf);
@@ -227,7 +225,6 @@ CreateNeighborPages(HnswBuildState * buildstate)
elog(ERROR, "failed to add index item to \"%s\"", RelationGetRelationName(index));
/* Commit */
MarkBufferDirty(buf);
GenericXLogFinish(state);
UnlockReleaseBuffer(buf);
}

View File

@@ -202,8 +202,6 @@ WriteNewElementPages(Relation index, HnswElement e, int m, BlockNumber insertPag
HnswInsertAppendPage(index, &newbuf, &newpage, state, page);
/* Commit */
MarkBufferDirty(newbuf);
MarkBufferDirty(buf);
GenericXLogFinish(state);
/* Unlock previous buffer */
@@ -270,9 +268,6 @@ WriteNewElementPages(Relation index, HnswElement e, int m, BlockNumber insertPag
}
/* Commit */
MarkBufferDirty(buf);
if (nbuf != buf)
MarkBufferDirty(nbuf);
GenericXLogFinish(state);
UnlockReleaseBuffer(buf);
if (nbuf != buf)
@@ -391,7 +386,6 @@ HnswUpdateNeighborPages(Relation index, FmgrInfo *procinfo, Oid collation, HnswE
elog(ERROR, "failed to add index item to \"%s\"", RelationGetRelationName(index));
/* Commit */
MarkBufferDirty(buf);
GenericXLogFinish(state);
}
else
@@ -445,7 +439,6 @@ HnswAddDuplicate(Relation index, HnswElement element, HnswElement dup)
elog(ERROR, "failed to add index item to \"%s\"", RelationGetRelationName(index));
/* Commit */
MarkBufferDirty(buf);
GenericXLogFinish(state);
UnlockReleaseBuffer(buf);

View File

@@ -101,7 +101,6 @@ hnswbeginscan(Relation index, int nkeys, int norderbys)
scan = RelationGetIndexScan(index, nkeys, norderbys);
so = (HnswScanOpaque) palloc(sizeof(HnswScanOpaqueData));
so->buf = InvalidBuffer;
so->first = true;
so->tmpCtx = AllocSetContextCreate(CurrentMemoryContext,
"Hnsw scan temporary context",
@@ -114,12 +113,6 @@ hnswbeginscan(Relation index, int nkeys, int norderbys)
scan->opaque = so;
/*
* Get a shared lock. This allows vacuum to ensure no in-flight scans
* before marking tuples as deleted.
*/
LockPage(scan->indexRelation, HNSW_SCAN_LOCK, ShareLock);
return scan;
}
@@ -170,8 +163,17 @@ hnswgettuple(IndexScanDesc scan, ScanDirection dir)
/* Get scan value */
value = GetScanValue(scan);
/*
* Get a shared lock. This allows vacuum to ensure no in-flight scans
* before marking tuples as deleted.
*/
LockPage(scan->indexRelation, HNSW_SCAN_LOCK, ShareLock);
so->w = GetScanItems(scan, value);
/* Release shared lock */
UnlockPage(scan->indexRelation, HNSW_SCAN_LOCK, ShareLock);
so->first = false;
}
@@ -179,7 +181,6 @@ hnswgettuple(IndexScanDesc scan, ScanDirection dir)
{
HnswCandidate *hc = llast(so->w);
ItemPointer heaptid;
BlockNumber indexblkno;
/* Move to next element if no valid heap TIDs */
if (list_length(hc->element->heaptids) == 0)
@@ -189,7 +190,6 @@ hnswgettuple(IndexScanDesc scan, ScanDirection dir)
}
heaptid = llast(hc->element->heaptids);
indexblkno = hc->element->blkno;
hc->element->heaptids = list_delete_last(hc->element->heaptids);
@@ -201,17 +201,14 @@ hnswgettuple(IndexScanDesc scan, ScanDirection dir)
scan->xs_ctup.t_self = *heaptid;
#endif
/* Unpin buffer */
if (BufferIsValid(so->buf))
ReleaseBuffer(so->buf);
/*
* An index scan must maintain a pin on the index page holding the
* item last returned by amgettuple
* Typically, an index scan must maintain a pin on the index page
* holding the item last returned by amgettuple. However, this is not
* needed with the current vacuum strategy, which ensures scans do not
* visit tuples in danger of being marked as deleted.
*
* https://www.postgresql.org/docs/current/index-locking.html
*/
so->buf = ReadBuffer(scan->indexRelation, indexblkno);
scan->xs_recheckorderby = false;
return true;
@@ -229,13 +226,6 @@ hnswendscan(IndexScanDesc scan)
{
HnswScanOpaque so = (HnswScanOpaque) scan->opaque;
/* Release pin */
if (BufferIsValid(so->buf))
ReleaseBuffer(so->buf);
/* Release shared lock */
UnlockPage(scan->indexRelation, HNSW_SCAN_LOCK, ShareLock);
MemoryContextDelete(so->tmpCtx);
pfree(so);

View File

@@ -38,12 +38,12 @@ HnswGetEfConstruction(Relation index)
* Get proc
*/
FmgrInfo *
HnswOptionalProcInfo(Relation rel, uint16 procnum)
HnswOptionalProcInfo(Relation index, uint16 procnum)
{
if (!OidIsValid(index_getprocid(rel, 1, procnum)))
if (!OidIsValid(index_getprocid(index, 1, procnum)))
return NULL;
return index_getprocinfo(rel, 1, procnum);
return index_getprocinfo(index, 1, procnum);
}
/*
@@ -117,7 +117,6 @@ HnswInitRegisterPage(Relation index, Buffer *buf, Page *page, GenericXLogState *
void
HnswCommitBuffer(Buffer buf, GenericXLogState *state)
{
MarkBufferDirty(buf);
GenericXLogFinish(state);
UnlockReleaseBuffer(buf);
}
@@ -140,9 +139,21 @@ HnswInitNeighbors(HnswElement element, int m)
a = &element->neighbors[lc];
a->length = 0;
a->items = palloc(sizeof(HnswCandidate) * lm);
a->closerSet = false;
}
}
/*
* Free neighbors
*/
static void
HnswFreeNeighbors(HnswElement element)
{
for (int lc = 0; lc <= element->level; lc++)
pfree(element->neighbors[lc].items);
pfree(element->neighbors);
}
/*
* Allocate an element
*/
@@ -174,10 +185,8 @@ HnswInitElement(ItemPointer heaptid, int m, double ml, int maxLevel)
void
HnswFreeElement(HnswElement element)
{
HnswFreeNeighbors(element);
list_free_deep(element->heaptids);
for (int lc = 0; lc <= element->level; lc++)
pfree(element->neighbors[lc].items);
pfree(element->neighbors);
pfree(element->vec);
pfree(element);
}
@@ -740,11 +749,13 @@ CheckElementCloser(HnswCandidate * e, List *r, int lc, FmgrInfo *procinfo, Oid c
* Algorithm 4 from paper
*/
static List *
SelectNeighbors(List *c, int m, int lc, FmgrInfo *procinfo, Oid collation, HnswCandidate * *pruned)
SelectNeighbors(List *c, int m, int lc, FmgrInfo *procinfo, Oid collation, HnswElement e2, HnswCandidate * newCandidate, HnswCandidate * *pruned)
{
List *r = NIL;
List *w = list_copy(c);
pairingheap *wd;
bool mustCalculate = !e2->neighbors[lc].closerSet;
bool foundNew = false;
if (list_length(w) <= m)
return w;
@@ -755,18 +766,40 @@ SelectNeighbors(List *c, int m, int lc, FmgrInfo *procinfo, Oid collation, HnswC
{
/* Assumes w is already ordered desc */
HnswCandidate *e = llast(w);
bool closer;
w = list_delete_last(w);
closer = CheckElementCloser(e, r, lc, procinfo, collation);
/* Use previous state of r and wd to skip work when possible */
if (mustCalculate)
e->closer = CheckElementCloser(e, r, lc, procinfo, collation);
else if (foundNew)
{
/* If new or current candidate is not closer, no change in state */
if (newCandidate->closer && e->closer)
{
/* Only need to compare with new candidate */
float distance = HnswGetDistance(e->element, newCandidate->element, lc, procinfo, collation);
if (closer)
e->closer = e->distance < distance;
if (!e->closer)
mustCalculate = true;
}
}
else if (e == newCandidate)
{
e->closer = CheckElementCloser(e, r, lc, procinfo, collation);
foundNew = true;
}
if (e->closer)
r = lappend(r, e);
else
pairingheap_add(wd, &(CreatePairingHeapNode(e)->ph_node));
}
e2->neighbors[lc].closerSet = true;
/* Keep pruned connections */
while (!pairingheap_is_empty(wd) && list_length(r) < m)
r = lappend(r, ((HnswPairingHeapNode *) pairingheap_remove_first(wd))->inner);
@@ -901,7 +934,7 @@ HnswUpdateConnection(HnswElement element, HnswCandidate * hc, int m, int lc, int
c = lappend(c, &hc2);
list_sort(c, CompareCandidateDistances);
SelectNeighbors(c, m, lc, procinfo, collation, &pruned);
SelectNeighbors(c, m, lc, procinfo, collation, hc->element, &hc2, &pruned);
/* Should not happen */
if (pruned == NULL)
@@ -1000,7 +1033,7 @@ HnswInsertElement(HnswElement element, HnswElement entryPoint, Relation index, F
else
lw = w;
neighbors = SelectNeighbors(lw, lm, lc, procinfo, collation, NULL);
neighbors = SelectNeighbors(lw, lm, lc, procinfo, collation, element, NULL, NULL);
AddConnections(element, neighbors, lm, lc);

View File

@@ -128,10 +128,7 @@ RemoveHeapTids(HnswVacuumState * vacuumstate)
blkno = HnswPageGetOpaque(page)->nextblkno;
if (updated)
{
MarkBufferDirty(buf);
GenericXLogFinish(state);
}
else
GenericXLogAbort(state);
@@ -229,7 +226,6 @@ RepairGraphElement(HnswVacuumState * vacuumstate, HnswElement element, HnswEleme
elog(ERROR, "failed to add index item to \"%s\"", RelationGetRelationName(index));
/* Commit */
MarkBufferDirty(buf);
GenericXLogFinish(state);
UnlockReleaseBuffer(buf);
@@ -547,9 +543,6 @@ MarkDeleted(HnswVacuumState * vacuumstate)
elog(ERROR, "failed to add index item to \"%s\"", RelationGetRelationName(index));
/* Commit */
MarkBufferDirty(buf);
if (nbuf != buf)
MarkBufferDirty(nbuf);
GenericXLogFinish(state);
if (nbuf != buf)
UnlockReleaseBuffer(nbuf);

View File

@@ -506,29 +506,30 @@ CreateListPages(Relation index, VectorArray centers, int dimensions,
Buffer buf;
Page page;
GenericXLogState *state;
OffsetNumber offno;
Size itemsz;
Size listSize;
IvfflatList list;
itemsz = MAXALIGN(IVFFLAT_LIST_SIZE(dimensions));
list = palloc(itemsz);
listSize = MAXALIGN(IVFFLAT_LIST_SIZE(dimensions));
list = palloc(listSize);
buf = IvfflatNewBuffer(index, forkNum);
IvfflatInitRegisterPage(index, &buf, &page, &state);
for (int i = 0; i < lists; i++)
{
OffsetNumber offno;
/* Load list */
list->startPage = InvalidBlockNumber;
list->insertPage = InvalidBlockNumber;
memcpy(&list->center, VectorArrayGet(centers, i), VECTOR_SIZE(dimensions));
/* Ensure free space */
if (PageGetFreeSpace(page) < itemsz)
if (PageGetFreeSpace(page) < listSize)
IvfflatAppendPage(index, &buf, &page, &state, forkNum);
/* Add the item */
offno = PageAddItem(page, (Item) list, itemsz, InvalidOffsetNumber, false, false);
offno = PageAddItem(page, (Item) list, listSize, InvalidOffsetNumber, false, false);
if (offno == InvalidOffsetNumber)
elog(ERROR, "failed to add index item to \"%s\"", RelationGetRelationName(index));

View File

@@ -246,6 +246,8 @@ typedef struct IvfflatScanOpaqueData
int probes;
int dimensions;
bool first;
Buffer buf;
ItemPointerData heaptid;
/* Sorting */
Tuplesortstate *sortstate;
@@ -275,7 +277,7 @@ VectorArray VectorArrayInit(int maxlen, int dimensions);
void VectorArrayFree(VectorArray arr);
void PrintVectorArray(char *msg, VectorArray arr);
void IvfflatKmeans(Relation index, VectorArray samples, VectorArray centers);
FmgrInfo *IvfflatOptionalProcInfo(Relation rel, uint16 procnum);
FmgrInfo *IvfflatOptionalProcInfo(Relation index, uint16 procnum);
bool IvfflatNormValue(FmgrInfo *procinfo, Oid collation, Datum *value, Vector * result);
int IvfflatGetLists(Relation index);
void IvfflatGetMetaPageInfo(Relation index, int *lists, int *dimensions);

View File

@@ -11,36 +11,37 @@
* Find the list that minimizes the distance function
*/
static void
FindInsertPage(Relation rel, Datum *values, BlockNumber *insertPage, ListInfo * listInfo)
FindInsertPage(Relation index, Datum *values, BlockNumber *insertPage, ListInfo * listInfo)
{
Buffer cbuf;
Page cpage;
IvfflatList list;
double distance;
double minDistance = DBL_MAX;
BlockNumber nextblkno = IVFFLAT_HEAD_BLKNO;
FmgrInfo *procinfo;
Oid collation;
OffsetNumber offno;
OffsetNumber maxoffno;
/* Avoid compiler warning */
listInfo->blkno = nextblkno;
listInfo->offno = FirstOffsetNumber;
procinfo = index_getprocinfo(rel, 1, IVFFLAT_DISTANCE_PROC);
collation = rel->rd_indcollation[0];
procinfo = index_getprocinfo(index, 1, IVFFLAT_DISTANCE_PROC);
collation = index->rd_indcollation[0];
/* Search all list pages */
while (BlockNumberIsValid(nextblkno))
{
cbuf = ReadBuffer(rel, nextblkno);
Buffer cbuf;
Page cpage;
OffsetNumber maxoffno;
cbuf = ReadBuffer(index, nextblkno);
LockBuffer(cbuf, BUFFER_LOCK_SHARE);
cpage = BufferGetPage(cbuf);
maxoffno = PageGetMaxOffsetNumber(cpage);
for (offno = FirstOffsetNumber; offno <= maxoffno; offno = OffsetNumberNext(offno))
for (OffsetNumber offno = FirstOffsetNumber; offno <= maxoffno; offno = OffsetNumberNext(offno))
{
IvfflatList list;
double distance;
list = (IvfflatList) PageGetItem(cpage, PageGetItemId(cpage, offno));
distance = DatumGetFloat8(FunctionCall2Coll(procinfo, collation, values[0], PointerGetDatum(&list->center)));
@@ -63,7 +64,7 @@ FindInsertPage(Relation rel, Datum *values, BlockNumber *insertPage, ListInfo *
* Insert a tuple into the index
*/
static void
InsertTuple(Relation rel, Datum *values, bool *isnull, ItemPointer heap_tid, Relation heapRel)
InsertTuple(Relation index, Datum *values, bool *isnull, ItemPointer heap_tid, Relation heapRel)
{
IndexTuple itup;
Datum value;
@@ -80,20 +81,20 @@ InsertTuple(Relation rel, Datum *values, bool *isnull, ItemPointer heap_tid, Rel
value = PointerGetDatum(PG_DETOAST_DATUM(values[0]));
/* Normalize if needed */
normprocinfo = IvfflatOptionalProcInfo(rel, IVFFLAT_NORM_PROC);
normprocinfo = IvfflatOptionalProcInfo(index, IVFFLAT_NORM_PROC);
if (normprocinfo != NULL)
{
if (!IvfflatNormValue(normprocinfo, rel->rd_indcollation[0], &value, NULL))
if (!IvfflatNormValue(normprocinfo, index->rd_indcollation[0], &value, NULL))
return;
}
/* Find the insert page - sets the page and list info */
FindInsertPage(rel, values, &insertPage, &listInfo);
FindInsertPage(index, values, &insertPage, &listInfo);
Assert(BlockNumberIsValid(insertPage));
originalInsertPage = insertPage;
/* Form tuple */
itup = index_form_tuple(RelationGetDescr(rel), &value, isnull);
itup = index_form_tuple(RelationGetDescr(index), &value, isnull);
itup->t_tid = *heap_tid;
/* Get tuple size */
@@ -103,10 +104,10 @@ InsertTuple(Relation rel, Datum *values, bool *isnull, ItemPointer heap_tid, Rel
/* Find a page to insert the item */
for (;;)
{
buf = ReadBuffer(rel, insertPage);
buf = ReadBuffer(index, insertPage);
LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
state = GenericXLogStart(rel);
state = GenericXLogStart(index);
page = GenericXLogRegisterBuffer(state, buf, 0);
if (PageGetFreeSpace(page) >= itemsz)
@@ -126,9 +127,9 @@ InsertTuple(Relation rel, Datum *values, bool *isnull, ItemPointer heap_tid, Rel
Page newpage;
/* Add a new page */
LockRelationForExtension(rel, ExclusiveLock);
newbuf = IvfflatNewBuffer(rel, MAIN_FORKNUM);
UnlockRelationForExtension(rel, ExclusiveLock);
LockRelationForExtension(index, ExclusiveLock);
newbuf = IvfflatNewBuffer(index, MAIN_FORKNUM);
UnlockRelationForExtension(index, ExclusiveLock);
/* Init new page */
newpage = GenericXLogRegisterBuffer(state, newbuf, GENERIC_XLOG_FULL_IMAGE);
@@ -141,15 +142,13 @@ InsertTuple(Relation rel, Datum *values, bool *isnull, ItemPointer heap_tid, Rel
IvfflatPageGetOpaque(page)->nextblkno = insertPage;
/* Commit */
MarkBufferDirty(newbuf);
MarkBufferDirty(buf);
GenericXLogFinish(state);
/* Unlock previous buffer */
UnlockReleaseBuffer(buf);
/* Prepare new buffer */
state = GenericXLogStart(rel);
state = GenericXLogStart(index);
buf = newbuf;
page = GenericXLogRegisterBuffer(state, buf, 0);
break;
@@ -158,13 +157,13 @@ InsertTuple(Relation rel, Datum *values, bool *isnull, ItemPointer heap_tid, Rel
/* Add to next offset */
if (PageAddItem(page, (Item) itup, itemsz, InvalidOffsetNumber, false, false) == InvalidOffsetNumber)
elog(ERROR, "failed to add index item to \"%s\"", RelationGetRelationName(rel));
elog(ERROR, "failed to add index item to \"%s\"", RelationGetRelationName(index));
IvfflatCommitBuffer(buf, state);
/* Update the insert page */
if (insertPage != originalInsertPage)
IvfflatUpdateList(rel, listInfo, insertPage, originalInsertPage, InvalidBlockNumber, MAIN_FORKNUM);
IvfflatUpdateList(index, listInfo, insertPage, originalInsertPage, InvalidBlockNumber, MAIN_FORKNUM);
}
/*

View File

@@ -17,10 +17,6 @@ InitCenters(Relation index, VectorArray samples, VectorArray centers, float *low
FmgrInfo *procinfo;
Oid collation;
int64 j;
double distance;
double sum;
double choice;
Vector *vec;
float *weight = palloc(samples->length * sizeof(float));
int numCenters = centers->maxlen;
int numSamples = samples->length;
@@ -33,17 +29,21 @@ InitCenters(Relation index, VectorArray samples, VectorArray centers, float *low
centers->length++;
for (j = 0; j < numSamples; j++)
weight[j] = DBL_MAX;
weight[j] = FLT_MAX;
for (int i = 0; i < numCenters; i++)
{
double sum;
double choice;
CHECK_FOR_INTERRUPTS();
sum = 0.0;
for (j = 0; j < numSamples; j++)
{
vec = VectorArrayGet(samples, j);
Vector *vec = VectorArrayGet(samples, j);
double distance;
/* Only need to compute distance for new center */
/* TODO Use triangle inequality to reduce distance calculations */
@@ -112,7 +112,6 @@ CompareVectors(const void *a, const void *b)
static void
QuickCenters(Relation index, VectorArray samples, VectorArray centers)
{
Vector *vec;
int dimensions = centers->dim;
Oid collation = index->rd_indcollation[0];
FmgrInfo *normprocinfo = IvfflatOptionalProcInfo(index, IVFFLAT_KMEANS_NORM_PROC);
@@ -123,7 +122,7 @@ QuickCenters(Relation index, VectorArray samples, VectorArray centers)
qsort(samples->items, samples->length, VECTOR_SIZE(samples->dim), CompareVectors);
for (int i = 0; i < samples->length; i++)
{
vec = VectorArrayGet(samples, i);
Vector *vec = VectorArrayGet(samples, i);
if (i == 0 || CompareVectors(vec, VectorArrayGet(samples, i - 1)) != 0)
{
@@ -136,7 +135,7 @@ QuickCenters(Relation index, VectorArray samples, VectorArray centers)
/* Fill remaining with random data */
while (centers->length < centers->maxlen)
{
vec = VectorArrayGet(centers, centers->length);
Vector *vec = VectorArrayGet(centers, centers->length);
SET_VARSIZE(vec, VECTOR_SIZE(dimensions));
vec->dim = dimensions;
@@ -168,7 +167,6 @@ ElkanKmeans(Relation index, VectorArray samples, VectorArray centers)
Oid collation;
Vector *vec;
Vector *newCenter;
int iteration;
int64 j;
int64 k;
int dimensions = centers->dim;
@@ -182,14 +180,6 @@ ElkanKmeans(Relation index, VectorArray samples, VectorArray centers)
float *s;
float *halfcdist;
float *newcdist;
int changes;
double minDistance;
int closestCenter;
double distance;
bool rj;
bool rjreset;
double dxcx;
double dxc;
/* Calculate allocation sizes */
Size samplesSize = VECTOR_ARRAY_SIZE(samples->maxlen, samples->dim);
@@ -247,14 +237,14 @@ ElkanKmeans(Relation index, VectorArray samples, VectorArray centers)
/* Assign each x to its closest initial center c(x) = argmin d(x,c) */
for (j = 0; j < numSamples; j++)
{
minDistance = DBL_MAX;
closestCenter = 0;
float minDistance = FLT_MAX;
int closestCenter = 0;
/* Find closest center */
for (k = 0; k < numCenters; k++)
{
/* TODO Use Lemma 1 in k-means++ initialization */
distance = lowerBound[j * numCenters + k];
float distance = lowerBound[j * numCenters + k];
if (distance < minDistance)
{
@@ -268,13 +258,14 @@ ElkanKmeans(Relation index, VectorArray samples, VectorArray centers)
}
/* Give 500 iterations to converge */
for (iteration = 0; iteration < 500; iteration++)
for (int iteration = 0; iteration < 500; iteration++)
{
int changes = 0;
bool rjreset;
/* Can take a while, so ensure we can interrupt */
CHECK_FOR_INTERRUPTS();
changes = 0;
/* Step 1: For all centers, compute distance */
for (j = 0; j < numCenters; j++)
{
@@ -282,7 +273,8 @@ ElkanKmeans(Relation index, VectorArray samples, VectorArray centers)
for (k = j + 1; k < numCenters; k++)
{
distance = 0.5 * DatumGetFloat8(FunctionCall2Coll(procinfo, collation, PointerGetDatum(vec), PointerGetDatum(VectorArrayGet(centers, k))));
float distance = 0.5 * DatumGetFloat8(FunctionCall2Coll(procinfo, collation, PointerGetDatum(vec), PointerGetDatum(VectorArrayGet(centers, k))));
halfcdist[j * numCenters + k] = distance;
halfcdist[k * numCenters + j] = distance;
}
@@ -291,10 +283,12 @@ ElkanKmeans(Relation index, VectorArray samples, VectorArray centers)
/* For all centers c, compute s(c) */
for (j = 0; j < numCenters; j++)
{
minDistance = DBL_MAX;
float minDistance = FLT_MAX;
for (k = 0; k < numCenters; k++)
{
float distance;
if (j == k)
continue;
@@ -310,6 +304,8 @@ ElkanKmeans(Relation index, VectorArray samples, VectorArray centers)
for (j = 0; j < numSamples; j++)
{
bool rj;
/* Step 2: Identify all points x such that u(x) <= s(c(x)) */
if (upperBound[j] <= s[closestCenters[j]])
continue;
@@ -318,6 +314,8 @@ ElkanKmeans(Relation index, VectorArray samples, VectorArray centers)
for (k = 0; k < numCenters; k++)
{
float dxcx;
/* Step 3: For all remaining points x and centers c */
if (k == closestCenters[j])
continue;
@@ -347,7 +345,7 @@ ElkanKmeans(Relation index, VectorArray samples, VectorArray centers)
/* Step 3b */
if (dxcx > lowerBound[j * numCenters + k] || dxcx > halfcdist[closestCenters[j] * numCenters + k])
{
dxc = DatumGetFloat8(FunctionCall2Coll(procinfo, collation, PointerGetDatum(vec), PointerGetDatum(VectorArrayGet(centers, k))));
float dxc = DatumGetFloat8(FunctionCall2Coll(procinfo, collation, PointerGetDatum(vec), PointerGetDatum(VectorArrayGet(centers, k))));
/* d(x,c) calculated */
lowerBound[j * numCenters + k] = dxc;
@@ -361,7 +359,6 @@ ElkanKmeans(Relation index, VectorArray samples, VectorArray centers)
changes++;
}
}
}
}
@@ -378,6 +375,8 @@ ElkanKmeans(Relation index, VectorArray samples, VectorArray centers)
for (j = 0; j < numSamples; j++)
{
int closestCenter;
vec = VectorArrayGet(samples, j);
closestCenter = closestCenters[j];
@@ -426,7 +425,7 @@ ElkanKmeans(Relation index, VectorArray samples, VectorArray centers)
{
for (k = 0; k < numCenters; k++)
{
distance = lowerBound[j * numCenters + k] - newcdist[k];
float distance = lowerBound[j * numCenters + k] - newcdist[k];
if (distance < 0)
distance = 0;
@@ -442,7 +441,7 @@ ElkanKmeans(Relation index, VectorArray samples, VectorArray centers)
/* Step 7 */
for (j = 0; j < numCenters; j++)
memcpy(VectorArrayGet(centers, j), VectorArrayGet(newCenters, j), VECTOR_SIZE(dimensions));
VectorArraySet(centers, j, VectorArrayGet(newCenters, j));
if (changes == 0 && iteration != 0)
break;
@@ -465,9 +464,6 @@ static void
CheckCenters(Relation index, VectorArray centers)
{
FmgrInfo *normprocinfo;
Oid collation;
Vector *vec;
double norm;
if (centers->length != centers->maxlen)
elog(ERROR, "Not enough centers. Please report a bug.");
@@ -475,7 +471,7 @@ CheckCenters(Relation index, VectorArray centers)
/* Ensure no NaN or infinite values */
for (int i = 0; i < centers->length; i++)
{
vec = VectorArrayGet(centers, i);
Vector *vec = VectorArrayGet(centers, i);
for (int j = 0; j < vec->dim; j++)
{
@@ -501,11 +497,12 @@ CheckCenters(Relation index, VectorArray centers)
normprocinfo = IvfflatOptionalProcInfo(index, IVFFLAT_NORM_PROC);
if (normprocinfo != NULL)
{
collation = index->rd_indcollation[0];
Oid collation = index->rd_indcollation[0];
for (int i = 0; i < centers->length; i++)
{
norm = DatumGetFloat8(FunctionCall1Coll(normprocinfo, collation, PointerGetDatum(VectorArrayGet(centers, i))));
double norm = DatumGetFloat8(FunctionCall1Coll(normprocinfo, collation, PointerGetDatum(VectorArrayGet(centers, i))));
if (norm == 0)
elog(ERROR, "Zero norm detected. Please report a bug.");
}

View File

@@ -143,6 +143,10 @@ GetScanItems(IndexScanDesc scan, Datum value)
bool isnull;
ItemId itemid = PageGetItemId(page, offno);
/* Skip dead tuples */
if (scan->ignore_killed_tuples && ItemIdIsDead(itemid))
continue;
itup = (IndexTuple) PageGetItem(page, itemid);
datum = index_getattr(itup, 1, tupdesc, &isnull);
@@ -157,6 +161,8 @@ GetScanItems(IndexScanDesc scan, Datum value)
slot->tts_isnull[0] = false;
slot->tts_values[1] = PointerGetDatum(&itup->t_tid);
slot->tts_isnull[1] = false;
slot->tts_values[2] = Int32GetDatum((int) searchPage);
slot->tts_isnull[2] = false;
ExecStoreVirtualTuple(slot);
tuplesort_puttupleslot(so->sortstate, slot);
@@ -181,6 +187,55 @@ GetScanItems(IndexScanDesc scan, Datum value)
tuplesort_performsort(so->sortstate);
}
/*
* Mark prior tuple as dead
*/
static void
MarkPriorTupleDead(IndexScanDesc scan)
{
IvfflatScanOpaque so = (IvfflatScanOpaque) scan->opaque;
Buffer buf = so->buf;
Page page;
OffsetNumber maxoffno;
/* Safety check */
if (!BufferIsValid(so->buf) || !ItemPointerIsValid(&so->heaptid))
return;
/* Only a shared locked is needed for ItemIdMarkDead */
LockBuffer(buf, BUFFER_LOCK_SHARE);
page = BufferGetPage(buf);
maxoffno = PageGetMaxOffsetNumber(page);
for (OffsetNumber offno = FirstOffsetNumber; offno <= maxoffno; offno = OffsetNumberNext(offno))
{
ItemId itemid = PageGetItemId(page, offno);
IndexTuple itup = (IndexTuple) PageGetItem(page, itemid);
/*
* Find tuple. Since buffer has been pinned, tuple cannot have been
* vacuumed (and heap TID reused).
*/
if (ItemPointerEquals(&itup->t_tid, &so->heaptid))
{
/*
* Make sure tuple has not already been marked dead to avoid extra
* WAL if wal_log_hints or data checksums enabled
*/
if (!ItemIdIsDead(itemid))
{
ItemIdMarkDead(itemid);
MarkBufferDirtyHint(buf, true);
}
break;
}
}
/* Unlock buffer */
LockBuffer(buf, BUFFER_LOCK_UNLOCK);
}
/*
* Prepare for an index scan
*/
@@ -206,7 +261,9 @@ ivfflatbeginscan(Relation index, int nkeys, int norderbys)
probes = lists;
so = (IvfflatScanOpaque) palloc(offsetof(IvfflatScanOpaqueData, lists) + probes * sizeof(IvfflatScanList));
so->buf = InvalidBuffer;
so->first = true;
ItemPointerSetInvalid(&so->heaptid);
so->probes = probes;
so->dimensions = dimensions;
@@ -217,12 +274,13 @@ ivfflatbeginscan(Relation index, int nkeys, int norderbys)
/* Create tuple description for sorting */
#if PG_VERSION_NUM >= 120000
so->tupdesc = CreateTemplateTupleDesc(2);
so->tupdesc = CreateTemplateTupleDesc(3);
#else
so->tupdesc = CreateTemplateTupleDesc(2, false);
so->tupdesc = CreateTemplateTupleDesc(3, false);
#endif
TupleDescInitEntry(so->tupdesc, (AttrNumber) 1, "distance", FLOAT8OID, -1, 0);
TupleDescInitEntry(so->tupdesc, (AttrNumber) 2, "heaptid", TIDOID, -1, 0);
TupleDescInitEntry(so->tupdesc, (AttrNumber) 3, "indexblkno", INT4OID, -1, 0);
/* Prep sort */
so->sortstate = tuplesort_begin_heap(so->tupdesc, 1, attNums, sortOperators, sortCollations, nullsFirstFlags, work_mem, NULL, false);
@@ -254,6 +312,7 @@ ivfflatrescan(IndexScanDesc scan, ScanKey keys, int nkeys, ScanKey orderbys, int
#endif
so->first = true;
ItemPointerSetInvalid(&so->heaptid);
pairingheap_reset(so->listQueue);
if (keys && scan->numberOfKeys > 0)
@@ -288,10 +347,6 @@ ivfflatgettuple(IndexScanDesc scan, ScanDirection dir)
if (scan->orderByData == NULL)
elog(ERROR, "cannot scan ivfflat index without order");
/* Safety check */
if (!IsMVCCSnapshot(scan->xs_snapshot))
elog(ERROR, "non-MVCC snapshots are not supported with ivfflat");
if (scan->orderByData->sk_flags & SK_ISNULL)
value = PointerGetDatum(InitVector(so->dimensions));
else
@@ -315,10 +370,17 @@ ivfflatgettuple(IndexScanDesc scan, ScanDirection dir)
if (value != scan->orderByData->sk_argument)
pfree(DatumGetPointer(value));
}
else
{
/* Mark prior tuple as dead */
if (scan->kill_prior_tuple)
MarkPriorTupleDead(scan);
}
if (tuplesort_gettupleslot(so->sortstate, true, false, so->slot, NULL))
{
ItemPointer heaptid = (ItemPointer) DatumGetPointer(slot_getattr(so->slot, 2, &so->isnull));
BlockNumber indexblkno = DatumGetInt32(slot_getattr(so->slot, 3, &so->isnull));
#if PG_VERSION_NUM >= 120000
scan->xs_heaptid = *heaptid;
@@ -326,11 +388,20 @@ ivfflatgettuple(IndexScanDesc scan, ScanDirection dir)
scan->xs_ctup.t_self = *heaptid;
#endif
/* Keep track of info needed to mark tuple as dead */
so->heaptid = *heaptid;
/* Unpin buffer */
if (BufferIsValid(so->buf))
ReleaseBuffer(so->buf);
/*
* Since no pinning during sorting, requires MVCC-compliant snapshot.
* Also, cannot kill prior tuples, since heap TID may be recycled.
* An index scan must maintain a pin on the index page holding the
* item last returned by amgettuple
*
* https://www.postgresql.org/docs/current/index-locking.html
*/
so->buf = ReadBuffer(scan->indexRelation, indexblkno);
scan->xs_recheckorderby = false;
return true;
@@ -347,6 +418,10 @@ ivfflatendscan(IndexScanDesc scan)
{
IvfflatScanOpaque so = (IvfflatScanOpaque) scan->opaque;
/* Release pin */
if (BufferIsValid(so->buf))
ReleaseBuffer(so->buf);
pairingheap_free(so->listQueue);
tuplesort_end(so->sortstate);

View File

@@ -57,12 +57,12 @@ IvfflatGetLists(Relation index)
* Get proc
*/
FmgrInfo *
IvfflatOptionalProcInfo(Relation rel, uint16 procnum)
IvfflatOptionalProcInfo(Relation index, uint16 procnum)
{
if (!OidIsValid(index_getprocid(rel, 1, procnum)))
if (!OidIsValid(index_getprocid(index, 1, procnum)))
return NULL;
return index_getprocinfo(rel, 1, procnum);
return index_getprocinfo(index, 1, procnum);
}
/*
@@ -136,7 +136,6 @@ IvfflatInitRegisterPage(Relation index, Buffer *buf, Page *page, GenericXLogStat
void
IvfflatCommitBuffer(Buffer buf, GenericXLogState *state)
{
MarkBufferDirty(buf);
GenericXLogFinish(state);
UnlockReleaseBuffer(buf);
}
@@ -160,8 +159,6 @@ IvfflatAppendPage(Relation index, Buffer *buf, Page *page, GenericXLogState **st
IvfflatInitPage(newbuf, newpage);
/* Commit */
MarkBufferDirty(*buf);
MarkBufferDirty(newbuf);
GenericXLogFinish(*state);
/* Unlock */

View File

@@ -107,7 +107,6 @@ ivfflatbulkdelete(IndexVacuumInfo *info, IndexBulkDeleteResult *stats,
{
/* Delete tuples */
PageIndexMultiDelete(page, deletable, ndeletable);
MarkBufferDirty(buf);
GenericXLogFinish(state);
}
else