mirror of
https://github.com/pgvector/pgvector.git
synced 2026-07-22 03:57:34 +08:00
Compare commits
32 Commits
kill-prior
...
non-mvcc-s
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
cb91de3332 | ||
|
|
b164833933 | ||
|
|
30fb4dd602 | ||
|
|
d032726976 | ||
|
|
8fa9001474 | ||
|
|
3431acef94 | ||
|
|
41bdf24cb7 | ||
|
|
3469a0e74c | ||
|
|
0fe43ca675 | ||
|
|
bbbe1db72b | ||
|
|
bab5fea9e7 | ||
|
|
b1f9519689 | ||
|
|
4571fccc60 | ||
|
|
db747e5aa0 | ||
|
|
2179414c05 | ||
|
|
8426ee61d2 | ||
|
|
c98c4e13aa | ||
|
|
04312f6638 | ||
|
|
72ea3c1210 | ||
|
|
b0801b8833 | ||
|
|
d05d6ee83d | ||
|
|
4022bb66a0 | ||
|
|
034d4acaea | ||
|
|
01f58e470a | ||
|
|
dbef8d1ad1 | ||
|
|
5c005cf57c | ||
|
|
5665a11a05 | ||
|
|
6b2e215447 | ||
|
|
0d86191eaf | ||
|
|
cf9f7aeea9 | ||
|
|
0b0e542ce6 | ||
|
|
a4590d2d9d |
@@ -1,3 +1,8 @@
|
||||
## 0.5.1 (unreleased)
|
||||
|
||||
- Improved performance of index scans for IVFFlat after updates and deletes
|
||||
- Fixed locking for index scans for HNSW
|
||||
|
||||
## 0.5.0 (2023-08-28)
|
||||
|
||||
- Added HNSW index type
|
||||
|
||||
2
Makefile
2
Makefile
@@ -8,7 +8,7 @@ HEADERS = src/vector.h
|
||||
|
||||
TESTS = $(wildcard test/sql/*.sql)
|
||||
REGRESS = $(patsubst test/sql/%.sql,%,$(TESTS))
|
||||
REGRESS_OPTS = --inputdir=test --load-extension=vector
|
||||
REGRESS_OPTS = --inputdir=test --load-extension=$(EXTENSION)
|
||||
|
||||
OPTFLAGS = -march=native
|
||||
|
||||
|
||||
@@ -5,7 +5,7 @@ OBJS = src\hnsw.obj src\hnswbuild.obj src\hnswinsert.obj src\hnswscan.obj src\hn
|
||||
HEADERS = src\vector.h
|
||||
|
||||
REGRESS = btree cast copy functions input ivfflat_cosine ivfflat_ip ivfflat_l2 ivfflat_options ivfflat_unlogged
|
||||
REGRESS_OPTS = --inputdir=test --load-extension=vector
|
||||
REGRESS_OPTS = --inputdir=test --load-extension=$(EXTENSION)
|
||||
|
||||
# For /arch flags
|
||||
# https://learn.microsoft.com/en-us/cpp/build/reference/arch-minimum-cpu-architecture
|
||||
|
||||
@@ -595,12 +595,18 @@ pgvector is available on [these providers](https://github.com/pgvector/pgvector/
|
||||
|
||||
## Upgrading
|
||||
|
||||
Install the latest version and run:
|
||||
Install the latest version. Then in each database you want to upgrade, run:
|
||||
|
||||
```sql
|
||||
ALTER EXTENSION vector UPDATE;
|
||||
```
|
||||
|
||||
You can check the version in the current database with:
|
||||
|
||||
```sql
|
||||
SELECT extversion FROM pg_extension WHERE extname = 'vector';
|
||||
```
|
||||
|
||||
## Upgrade Notes
|
||||
|
||||
### 0.4.0
|
||||
|
||||
@@ -91,7 +91,7 @@ hnswcostestimate(PlannerInfo *root, IndexPath *path, double loop_count,
|
||||
MemSet(&costs, 0, sizeof(costs));
|
||||
|
||||
index = index_open(path->indexinfo->indexoid, NoLock);
|
||||
m = HnswGetM(index);
|
||||
HnswGetMetaPageInfo(index, &m, NULL);
|
||||
index_close(index, NoLock);
|
||||
|
||||
/* Approximate entry level */
|
||||
@@ -196,7 +196,7 @@ hnswhandler(PG_FUNCTION_ARGS)
|
||||
amroutine->aminsert = hnswinsert;
|
||||
amroutine->ambulkdelete = hnswbulkdelete;
|
||||
amroutine->amvacuumcleanup = hnswvacuumcleanup;
|
||||
amroutine->amcanreturn = NULL; /* tuple not included in heapsort */
|
||||
amroutine->amcanreturn = NULL;
|
||||
amroutine->amcostestimate = hnswcostestimate;
|
||||
amroutine->amoptions = hnswoptions;
|
||||
amroutine->amproperty = NULL; /* TODO AMPROP_DISTANCE_ORDERABLE */
|
||||
|
||||
@@ -218,10 +218,6 @@ typedef HnswNeighborTupleData * HnswNeighborTuple;
|
||||
typedef struct HnswScanOpaqueData
|
||||
{
|
||||
bool first;
|
||||
Buffer buf;
|
||||
ItemPointerData heaptid;
|
||||
OffsetNumber offno;
|
||||
int removedCount;
|
||||
List *w;
|
||||
MemoryContext tmpCtx;
|
||||
|
||||
@@ -269,8 +265,9 @@ Buffer HnswNewBuffer(Relation index, ForkNumber forkNum);
|
||||
void HnswInitPage(Buffer buf, Page page);
|
||||
void HnswInitRegisterPage(Relation index, Buffer *buf, Page *page, GenericXLogState **state);
|
||||
void HnswInit(void);
|
||||
List *HnswSearchLayer(Datum q, List *ep, int ef, int lc, Relation index, FmgrInfo *procinfo, Oid collation, bool inserting, HnswElement skipElement);
|
||||
List *HnswSearchLayer(Datum q, List *ep, int ef, int lc, Relation index, FmgrInfo *procinfo, Oid collation, int m, bool inserting, HnswElement skipElement);
|
||||
HnswElement HnswGetEntryPoint(Relation index);
|
||||
void HnswGetMetaPageInfo(Relation index, int *m, HnswElement * entryPoint);
|
||||
HnswElement HnswInitElement(ItemPointer tid, int m, double ml, int maxLevel);
|
||||
void HnswFreeElement(HnswElement element);
|
||||
HnswElement HnswInitElementFromBlock(BlockNumber blkno, OffsetNumber offno);
|
||||
@@ -287,7 +284,7 @@ void HnswLoadElementFromTuple(HnswElement element, HnswElementTuple etup, bool
|
||||
void HnswLoadElement(HnswElement element, float *distance, Datum *q, Relation index, FmgrInfo *procinfo, Oid collation, bool loadVec);
|
||||
void HnswSetElementTuple(HnswElementTuple etup, HnswElement element);
|
||||
void HnswUpdateConnection(HnswElement element, HnswCandidate * hc, int m, int lc, int *updateIdx, Relation index, FmgrInfo *procinfo, Oid collation);
|
||||
void HnswLoadNeighbors(HnswElement element, Relation index);
|
||||
void HnswLoadNeighbors(HnswElement element, Relation index, int m);
|
||||
|
||||
/* Index access methods */
|
||||
IndexBuildResult *hnswbuild(Relation heap, Relation index, IndexInfo *indexInfo);
|
||||
|
||||
@@ -329,7 +329,7 @@ HnswUpdateNeighborPages(Relation index, FmgrInfo *procinfo, Oid collation, HnswE
|
||||
|
||||
/* Get latest neighbors since they may have changed */
|
||||
/* Do not lock yet since selecting neighbors can take time */
|
||||
HnswLoadNeighbors(hc->element, index);
|
||||
HnswLoadNeighbors(hc->element, index, m);
|
||||
|
||||
/*
|
||||
* Could improve performance for vacuuming by checking neighbors
|
||||
@@ -492,9 +492,8 @@ HnswInsertTuple(Relation index, Datum *values, bool *isnull, ItemPointer heap_ti
|
||||
FmgrInfo *normprocinfo;
|
||||
HnswElement entryPoint;
|
||||
HnswElement element;
|
||||
int m = HnswGetM(index);
|
||||
int m;
|
||||
int efConstruction = HnswGetEfConstruction(index);
|
||||
double ml = HnswGetMl(m);
|
||||
FmgrInfo *procinfo = index_getprocinfo(index, 1, HNSW_DISTANCE_PROC);
|
||||
Oid collation = index->rd_indcollation[0];
|
||||
HnswElement dup;
|
||||
@@ -511,10 +510,6 @@ HnswInsertTuple(Relation index, Datum *values, bool *isnull, ItemPointer heap_ti
|
||||
return false;
|
||||
}
|
||||
|
||||
/* Create an element */
|
||||
element = HnswInitElement(heap_tid, m, ml, HnswGetMaxLevel(m));
|
||||
element->vec = DatumGetVector(value);
|
||||
|
||||
/*
|
||||
* Get a shared lock. This allows vacuum to ensure no in-flight inserts
|
||||
* before repairing graph. Use a page lock so it does not interfere with
|
||||
@@ -522,8 +517,12 @@ HnswInsertTuple(Relation index, Datum *values, bool *isnull, ItemPointer heap_ti
|
||||
*/
|
||||
LockPage(index, HNSW_UPDATE_LOCK, lockmode);
|
||||
|
||||
/* Get entry point */
|
||||
entryPoint = HnswGetEntryPoint(index);
|
||||
/* Get m and entry point */
|
||||
HnswGetMetaPageInfo(index, &m, &entryPoint);
|
||||
|
||||
/* Create an element */
|
||||
element = HnswInitElement(heap_tid, m, HnswGetMl(m), HnswGetMaxLevel(m));
|
||||
element->vec = DatumGetVector(value);
|
||||
|
||||
/* Prevent concurrent inserts when likely updating entry point */
|
||||
if (entryPoint == NULL || element->level > entryPoint->level)
|
||||
|
||||
162
src/hnswscan.c
162
src/hnswscan.c
@@ -19,7 +19,11 @@ GetScanItems(IndexScanDesc scan, Datum q)
|
||||
Oid collation = so->collation;
|
||||
List *ep;
|
||||
List *w;
|
||||
HnswElement entryPoint = HnswGetEntryPoint(index);
|
||||
int m;
|
||||
HnswElement entryPoint;
|
||||
|
||||
/* Get m and entry point */
|
||||
HnswGetMetaPageInfo(index, &m, &entryPoint);
|
||||
|
||||
if (entryPoint == NULL)
|
||||
return NIL;
|
||||
@@ -28,11 +32,11 @@ GetScanItems(IndexScanDesc scan, Datum q)
|
||||
|
||||
for (int lc = entryPoint->level; lc >= 1; lc--)
|
||||
{
|
||||
w = HnswSearchLayer(q, ep, 1, lc, index, procinfo, collation, false, NULL);
|
||||
w = HnswSearchLayer(q, ep, 1, lc, index, procinfo, collation, m, false, NULL);
|
||||
ep = w;
|
||||
}
|
||||
|
||||
return HnswSearchLayer(q, ep, hnsw_ef_search, 0, index, procinfo, collation, false, NULL);
|
||||
return HnswSearchLayer(q, ep, hnsw_ef_search, 0, index, procinfo, collation, m, false, NULL);
|
||||
}
|
||||
|
||||
/*
|
||||
@@ -59,72 +63,30 @@ GetDimensions(Relation index)
|
||||
}
|
||||
|
||||
/*
|
||||
* Remove deleted heap TID
|
||||
* Get scan value
|
||||
*/
|
||||
static void
|
||||
RemoveHeapTid(IndexScanDesc scan)
|
||||
static Datum
|
||||
GetScanValue(IndexScanDesc scan)
|
||||
{
|
||||
HnswScanOpaque so = (HnswScanOpaque) scan->opaque;
|
||||
Relation index = scan->indexRelation;
|
||||
Buffer buf = so->buf;
|
||||
Page page;
|
||||
GenericXLogState *state;
|
||||
ItemId itemid;
|
||||
HnswElementTuple etup;
|
||||
Size etupSize;
|
||||
int idx = -1;
|
||||
Datum value;
|
||||
|
||||
/* Safety check */
|
||||
if (!BufferIsValid(buf) || !OffsetNumberIsValid(so->offno) || !ItemPointerIsValid(&so->heaptid))
|
||||
return;
|
||||
|
||||
/* Use WAL rather than hint */
|
||||
LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
|
||||
state = GenericXLogStart(index);
|
||||
page = GenericXLogRegisterBuffer(state, buf, 0);
|
||||
itemid = PageGetItemId(page, so->offno);
|
||||
etup = (HnswElementTuple) PageGetItem(page, itemid);
|
||||
etupSize = ItemIdGetLength(itemid);
|
||||
|
||||
Assert(HnswIsElementTuple(etup));
|
||||
|
||||
/* Find index */
|
||||
for (int i = 0; i < HNSW_HEAPTIDS; i++)
|
||||
{
|
||||
if (!ItemPointerIsValid(&etup->heaptids[i]))
|
||||
break;
|
||||
|
||||
if (ItemPointerEquals(&etup->heaptids[i], &so->heaptid))
|
||||
{
|
||||
idx = i;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
if (idx == -1)
|
||||
GenericXLogAbort(state);
|
||||
if (scan->orderByData->sk_flags & SK_ISNULL)
|
||||
value = PointerGetDatum(InitVector(GetDimensions(scan->indexRelation)));
|
||||
else
|
||||
{
|
||||
/* Move pointers forward */
|
||||
for (int i = idx; i < HNSW_HEAPTIDS; i++)
|
||||
{
|
||||
if (i + 1 == HNSW_HEAPTIDS || !ItemPointerIsValid(&etup->heaptids[i + 1]))
|
||||
ItemPointerSetInvalid(&etup->heaptids[i]);
|
||||
else
|
||||
ItemPointerCopy(&etup->heaptids[i + 1], &etup->heaptids[i]);
|
||||
}
|
||||
value = scan->orderByData->sk_argument;
|
||||
|
||||
/* Overwrite tuple */
|
||||
if (!PageIndexTupleOverwrite(page, so->offno, (Item) etup, etupSize))
|
||||
elog(ERROR, "failed to add index item to \"%s\"", RelationGetRelationName(index));
|
||||
/* Value should not be compressed or toasted */
|
||||
Assert(!VARATT_IS_COMPRESSED(DatumGetPointer(value)));
|
||||
Assert(!VARATT_IS_EXTENDED(DatumGetPointer(value)));
|
||||
|
||||
/* Commit */
|
||||
MarkBufferDirty(buf);
|
||||
GenericXLogFinish(state);
|
||||
/* Fine if normalization fails */
|
||||
if (so->normprocinfo != NULL)
|
||||
HnswNormValue(so->normprocinfo, so->collation, &value, NULL);
|
||||
}
|
||||
|
||||
/* Unlock buffer */
|
||||
LockBuffer(buf, BUFFER_LOCK_UNLOCK);
|
||||
return value;
|
||||
}
|
||||
|
||||
/*
|
||||
@@ -139,10 +101,6 @@ hnswbeginscan(Relation index, int nkeys, int norderbys)
|
||||
scan = RelationGetIndexScan(index, nkeys, norderbys);
|
||||
|
||||
so = (HnswScanOpaque) palloc(sizeof(HnswScanOpaqueData));
|
||||
so->buf = InvalidBuffer;
|
||||
ItemPointerSetInvalid(&so->heaptid);
|
||||
so->offno = InvalidOffsetNumber;
|
||||
so->removedCount = 0;
|
||||
so->first = true;
|
||||
so->tmpCtx = AllocSetContextCreate(CurrentMemoryContext,
|
||||
"Hnsw scan temporary context",
|
||||
@@ -155,6 +113,12 @@ 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;
|
||||
}
|
||||
|
||||
@@ -167,7 +131,6 @@ hnswrescan(IndexScanDesc scan, ScanKey keys, int nkeys, ScanKey orderbys, int no
|
||||
HnswScanOpaque so = (HnswScanOpaque) scan->opaque;
|
||||
|
||||
so->first = true;
|
||||
ItemPointerSetInvalid(&so->heaptid);
|
||||
MemoryContextReset(so->tmpCtx);
|
||||
|
||||
if (keys && scan->numberOfKeys > 0)
|
||||
@@ -203,90 +166,46 @@ hnswgettuple(IndexScanDesc scan, ScanDirection dir)
|
||||
if (scan->orderByData == NULL)
|
||||
elog(ERROR, "cannot scan hnsw index without order");
|
||||
|
||||
if (scan->orderByData->sk_flags & SK_ISNULL)
|
||||
value = PointerGetDatum(InitVector(GetDimensions(scan->indexRelation)));
|
||||
else
|
||||
{
|
||||
value = scan->orderByData->sk_argument;
|
||||
|
||||
/* Value should not be compressed or toasted */
|
||||
Assert(!VARATT_IS_COMPRESSED(DatumGetPointer(value)));
|
||||
Assert(!VARATT_IS_EXTENDED(DatumGetPointer(value)));
|
||||
|
||||
/* Fine if normalization fails */
|
||||
if (so->normprocinfo != NULL)
|
||||
HnswNormValue(so->normprocinfo, so->collation, &value, NULL);
|
||||
}
|
||||
|
||||
/*
|
||||
* 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);
|
||||
/* Get scan value */
|
||||
value = GetScanValue(scan);
|
||||
|
||||
so->w = GetScanItems(scan, value);
|
||||
|
||||
/* Release shared lock */
|
||||
UnlockPage(scan->indexRelation, HNSW_SCAN_LOCK, ShareLock);
|
||||
|
||||
so->first = false;
|
||||
}
|
||||
else
|
||||
{
|
||||
/*
|
||||
* Remove dead tuples. kill_prior_tuple will only be true if not in
|
||||
* recovery. Limit the number removed per scan for performance.
|
||||
*/
|
||||
if (scan->kill_prior_tuple && so->removedCount < 3)
|
||||
{
|
||||
RemoveHeapTid(scan);
|
||||
so->removedCount++;
|
||||
}
|
||||
}
|
||||
|
||||
while (list_length(so->w) > 0)
|
||||
{
|
||||
HnswCandidate *hc = llast(so->w);
|
||||
ItemPointer tid;
|
||||
BlockNumber indexblkno;
|
||||
OffsetNumber indexoffno;
|
||||
ItemPointer heaptid;
|
||||
|
||||
/* Move to next element if no valid heap tids */
|
||||
/* Move to next element if no valid heap TIDs */
|
||||
if (list_length(hc->element->heaptids) == 0)
|
||||
{
|
||||
so->w = list_delete_last(so->w);
|
||||
continue;
|
||||
}
|
||||
|
||||
tid = llast(hc->element->heaptids);
|
||||
indexblkno = hc->element->blkno;
|
||||
indexoffno = hc->element->offno;
|
||||
heaptid = llast(hc->element->heaptids);
|
||||
|
||||
hc->element->heaptids = list_delete_last(hc->element->heaptids);
|
||||
|
||||
MemoryContextSwitchTo(oldCtx);
|
||||
|
||||
#if PG_VERSION_NUM >= 120000
|
||||
scan->xs_heaptid = *tid;
|
||||
scan->xs_heaptid = *heaptid;
|
||||
#else
|
||||
scan->xs_ctup.t_self = *tid;
|
||||
scan->xs_ctup.t_self = *heaptid;
|
||||
#endif
|
||||
|
||||
/* Keep track of info needed to remove dead tuples */
|
||||
so->heaptid = *tid;
|
||||
so->offno = indexoffno;
|
||||
|
||||
/* 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;
|
||||
@@ -304,9 +223,8 @@ 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);
|
||||
|
||||
|
||||
@@ -210,25 +210,43 @@ HnswInitElementFromBlock(BlockNumber blkno, OffsetNumber offno)
|
||||
}
|
||||
|
||||
/*
|
||||
* Get the entry point
|
||||
* Get the metapage info
|
||||
*/
|
||||
HnswElement
|
||||
HnswGetEntryPoint(Relation index)
|
||||
void
|
||||
HnswGetMetaPageInfo(Relation index, int *m, HnswElement * entryPoint)
|
||||
{
|
||||
Buffer buf;
|
||||
Page page;
|
||||
HnswMetaPage metap;
|
||||
HnswElement entryPoint = NULL;
|
||||
|
||||
buf = ReadBuffer(index, HNSW_METAPAGE_BLKNO);
|
||||
LockBuffer(buf, BUFFER_LOCK_SHARE);
|
||||
page = BufferGetPage(buf);
|
||||
metap = HnswPageGetMeta(page);
|
||||
|
||||
if (BlockNumberIsValid(metap->entryBlkno))
|
||||
entryPoint = HnswInitElementFromBlock(metap->entryBlkno, metap->entryOffno);
|
||||
if (m != NULL)
|
||||
*m = metap->m;
|
||||
|
||||
if (entryPoint != NULL)
|
||||
{
|
||||
if (BlockNumberIsValid(metap->entryBlkno))
|
||||
*entryPoint = HnswInitElementFromBlock(metap->entryBlkno, metap->entryOffno);
|
||||
else
|
||||
*entryPoint = NULL;
|
||||
}
|
||||
|
||||
UnlockReleaseBuffer(buf);
|
||||
}
|
||||
|
||||
/*
|
||||
* Get the entry point
|
||||
*/
|
||||
HnswElement
|
||||
HnswGetEntryPoint(Relation index)
|
||||
{
|
||||
HnswElement entryPoint;
|
||||
|
||||
HnswGetMetaPageInfo(index, NULL, &entryPoint);
|
||||
|
||||
return entryPoint;
|
||||
}
|
||||
@@ -337,10 +355,9 @@ HnswSetNeighborTuple(HnswNeighborTuple ntup, HnswElement e, int m)
|
||||
* Load neighbors from page
|
||||
*/
|
||||
static void
|
||||
LoadNeighborsFromPage(HnswElement element, Relation index, Page page)
|
||||
LoadNeighborsFromPage(HnswElement element, Relation index, Page page, int m)
|
||||
{
|
||||
HnswNeighborTuple ntup = (HnswNeighborTuple) PageGetItem(page, PageGetItemId(page, element->neighborOffno));
|
||||
int m = HnswGetM(index);
|
||||
int neighborCount = (element->level + 2) * m;
|
||||
|
||||
Assert(HnswIsNeighborTuple(ntup));
|
||||
@@ -381,7 +398,7 @@ LoadNeighborsFromPage(HnswElement element, Relation index, Page page)
|
||||
* Load neighbors
|
||||
*/
|
||||
void
|
||||
HnswLoadNeighbors(HnswElement element, Relation index)
|
||||
HnswLoadNeighbors(HnswElement element, Relation index, int m)
|
||||
{
|
||||
Buffer buf;
|
||||
Page page;
|
||||
@@ -390,7 +407,7 @@ HnswLoadNeighbors(HnswElement element, Relation index)
|
||||
LockBuffer(buf, BUFFER_LOCK_SHARE);
|
||||
page = BufferGetPage(buf);
|
||||
|
||||
LoadNeighborsFromPage(element, index, page);
|
||||
LoadNeighborsFromPage(element, index, page, m);
|
||||
|
||||
UnlockReleaseBuffer(buf);
|
||||
}
|
||||
@@ -543,7 +560,7 @@ AddToVisited(HTAB *v, HnswCandidate * hc, Relation index, bool *found)
|
||||
* Algorithm 2 from paper
|
||||
*/
|
||||
List *
|
||||
HnswSearchLayer(Datum q, List *ep, int ef, int lc, Relation index, FmgrInfo *procinfo, Oid collation, bool inserting, HnswElement skipElement)
|
||||
HnswSearchLayer(Datum q, List *ep, int ef, int lc, Relation index, FmgrInfo *procinfo, Oid collation, int m, bool inserting, HnswElement skipElement)
|
||||
{
|
||||
ListCell *lc2;
|
||||
|
||||
@@ -551,8 +568,6 @@ HnswSearchLayer(Datum q, List *ep, int ef, int lc, Relation index, FmgrInfo *pro
|
||||
pairingheap *C = pairingheap_allocate(CompareNearestCandidates, NULL);
|
||||
pairingheap *W = pairingheap_allocate(CompareFurthestCandidates, NULL);
|
||||
int wlen = 0;
|
||||
uint64 dead = 0;
|
||||
uint64 maxAdditional = skipElement == NULL ? ef : PG_UINT64_MAX;
|
||||
HASHCTL hash_ctl;
|
||||
HTAB *v;
|
||||
|
||||
@@ -581,14 +596,13 @@ HnswSearchLayer(Datum q, List *ep, int ef, int lc, Relation index, FmgrInfo *pro
|
||||
pairingheap_add(C, &(CreatePairingHeapNode(hc)->ph_node));
|
||||
pairingheap_add(W, &(CreatePairingHeapNode(hc)->ph_node));
|
||||
|
||||
/* Do not count certain number of dead elements towards ef */
|
||||
if (list_length(hc->element->heaptids) == 0)
|
||||
{
|
||||
if ((++dead) <= maxAdditional)
|
||||
continue;
|
||||
}
|
||||
|
||||
wlen++;
|
||||
/*
|
||||
* Do not count elements being deleted towards ef when vacuuming. It
|
||||
* would be ideal to do this for inserts as well, but this could
|
||||
* affect insert performance.
|
||||
*/
|
||||
if (skipElement == NULL || list_length(hc->element->heaptids) != 0)
|
||||
wlen++;
|
||||
}
|
||||
|
||||
while (!pairingheap_is_empty(C))
|
||||
@@ -601,7 +615,7 @@ HnswSearchLayer(Datum q, List *ep, int ef, int lc, Relation index, FmgrInfo *pro
|
||||
break;
|
||||
|
||||
if (c->element->neighbors == NULL)
|
||||
HnswLoadNeighbors(c->element, index);
|
||||
HnswLoadNeighbors(c->element, index, m);
|
||||
|
||||
/* Get the neighborhood at layer lc */
|
||||
neighborhood = &c->element->neighbors[lc];
|
||||
@@ -641,18 +655,19 @@ HnswSearchLayer(Datum q, List *ep, int ef, int lc, Relation index, FmgrInfo *pro
|
||||
pairingheap_add(C, &(CreatePairingHeapNode(ec)->ph_node));
|
||||
pairingheap_add(W, &(CreatePairingHeapNode(ec)->ph_node));
|
||||
|
||||
/* Do not count certain number of dead elements towards ef */
|
||||
if (list_length(e->element->heaptids) == 0)
|
||||
/*
|
||||
* Do not count elements being deleted towards ef when
|
||||
* vacuuming. It would be ideal to do this for inserts as
|
||||
* well, but this could affect insert performance.
|
||||
*/
|
||||
if (skipElement == NULL || list_length(e->element->heaptids) != 0)
|
||||
{
|
||||
if ((++dead) <= maxAdditional)
|
||||
continue;
|
||||
wlen++;
|
||||
|
||||
/* No need to decrement wlen */
|
||||
if (wlen > ef)
|
||||
pairingheap_remove_first(W);
|
||||
}
|
||||
|
||||
wlen++;
|
||||
|
||||
/* No need to decrement wlen */
|
||||
if (wlen > ef)
|
||||
pairingheap_remove_first(W);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -958,7 +973,7 @@ HnswInsertElement(HnswElement element, HnswElement entryPoint, Relation index, F
|
||||
/* 1st phase: greedy search to insert level */
|
||||
for (int lc = entryLevel; lc >= level + 1; lc--)
|
||||
{
|
||||
w = HnswSearchLayer(q, ep, 1, lc, index, procinfo, collation, true, skipElement);
|
||||
w = HnswSearchLayer(q, ep, 1, lc, index, procinfo, collation, m, true, skipElement);
|
||||
ep = w;
|
||||
}
|
||||
|
||||
@@ -976,7 +991,7 @@ HnswInsertElement(HnswElement element, HnswElement entryPoint, Relation index, F
|
||||
List *neighbors;
|
||||
List *lw;
|
||||
|
||||
w = HnswSearchLayer(q, ep, efConstruction, lc, index, procinfo, collation, true, skipElement);
|
||||
w = HnswSearchLayer(q, ep, efConstruction, lc, index, procinfo, collation, m, true, skipElement);
|
||||
|
||||
/* Elements being deleted or skipped can help with search */
|
||||
/* but should be removed before selecting neighbors */
|
||||
|
||||
@@ -330,7 +330,10 @@ RepairGraph(HnswVacuumState * vacuumstate)
|
||||
BufferAccessStrategy bas = vacuumstate->bas;
|
||||
BlockNumber blkno = HNSW_HEAD_BLKNO;
|
||||
|
||||
/* Wait for inserts to complete */
|
||||
/*
|
||||
* Wait for inserts to complete. Inserts before this point may have
|
||||
* neighbors about to be deleted. Inserts after this point will not.
|
||||
*/
|
||||
LockPage(index, HNSW_UPDATE_LOCK, ExclusiveLock);
|
||||
UnlockPage(index, HNSW_UPDATE_LOCK, ExclusiveLock);
|
||||
|
||||
@@ -443,7 +446,11 @@ MarkDeleted(HnswVacuumState * vacuumstate)
|
||||
Relation index = vacuumstate->index;
|
||||
BufferAccessStrategy bas = vacuumstate->bas;
|
||||
|
||||
/* Wait for selects to complete */
|
||||
/*
|
||||
* Wait for index scans to complete. Scans before this point may contain
|
||||
* tuples about to be deleted. Scans after this point will not, since the
|
||||
* graph has been repaired.
|
||||
*/
|
||||
LockPage(index, HNSW_SCAN_LOCK, ExclusiveLock);
|
||||
UnlockPage(index, HNSW_SCAN_LOCK, ExclusiveLock);
|
||||
|
||||
@@ -582,7 +589,6 @@ InitVacuumState(HnswVacuumState * vacuumstate, IndexVacuumInfo *info, IndexBulkD
|
||||
vacuumstate->stats = stats;
|
||||
vacuumstate->callback = callback;
|
||||
vacuumstate->callback_state = callback_state;
|
||||
vacuumstate->m = HnswGetM(index);
|
||||
vacuumstate->efConstruction = HnswGetEfConstruction(index);
|
||||
vacuumstate->bas = GetAccessStrategy(BAS_BULKREAD);
|
||||
vacuumstate->procinfo = index_getprocinfo(index, 1, HNSW_DISTANCE_PROC);
|
||||
@@ -592,6 +598,9 @@ InitVacuumState(HnswVacuumState * vacuumstate, IndexVacuumInfo *info, IndexBulkD
|
||||
"Hnsw vacuum temporary context",
|
||||
ALLOCSET_DEFAULT_SIZES);
|
||||
|
||||
/* Get m from metapage */
|
||||
HnswGetMetaPageInfo(index, &vacuumstate->m, NULL);
|
||||
|
||||
/* Create hash table */
|
||||
hash_ctl.keysize = sizeof(ItemPointerData);
|
||||
hash_ctl.entrysize = sizeof(ItemPointerData);
|
||||
|
||||
@@ -71,7 +71,7 @@ ivfflatcostestimate(PlannerInfo *root, IndexPath *path, double loop_count,
|
||||
int lists;
|
||||
double ratio;
|
||||
double spc_seq_page_cost;
|
||||
Relation indexRel;
|
||||
Relation index;
|
||||
#if PG_VERSION_NUM < 120000
|
||||
List *qinfos;
|
||||
#endif
|
||||
@@ -89,9 +89,9 @@ ivfflatcostestimate(PlannerInfo *root, IndexPath *path, double loop_count,
|
||||
|
||||
MemSet(&costs, 0, sizeof(costs));
|
||||
|
||||
indexRel = index_open(path->indexinfo->indexoid, NoLock);
|
||||
lists = IvfflatGetLists(indexRel);
|
||||
index_close(indexRel, NoLock);
|
||||
index = index_open(path->indexinfo->indexoid, NoLock);
|
||||
IvfflatGetMetaPageInfo(index, &lists, NULL);
|
||||
index_close(index, NoLock);
|
||||
|
||||
/* Get the ratio of lists that we need to visit */
|
||||
ratio = ((double) ivfflat_probes) / lists;
|
||||
|
||||
@@ -40,6 +40,9 @@
|
||||
#define IVFFLAT_METAPAGE_BLKNO 0
|
||||
#define IVFFLAT_HEAD_BLKNO 1 /* first list page */
|
||||
|
||||
/* Must correspond to page numbers since page lock is used */
|
||||
#define IVFFLAT_SCAN_LOCK 0
|
||||
|
||||
/* IVFFlat parameters */
|
||||
#define IVFFLAT_DEFAULT_LISTS 100
|
||||
#define IVFFLAT_MIN_LISTS 1
|
||||
@@ -244,8 +247,11 @@ typedef struct IvfflatScanList
|
||||
typedef struct IvfflatScanOpaqueData
|
||||
{
|
||||
int probes;
|
||||
int dimensions;
|
||||
bool first;
|
||||
bool hasLock;
|
||||
Buffer buf;
|
||||
ItemPointerData heaptid;
|
||||
|
||||
/* Sorting */
|
||||
Tuplesortstate *sortstate;
|
||||
@@ -278,6 +284,7 @@ void IvfflatKmeans(Relation index, VectorArray samples, VectorArray centers);
|
||||
FmgrInfo *IvfflatOptionalProcInfo(Relation rel, uint16 procnum);
|
||||
bool IvfflatNormValue(FmgrInfo *procinfo, Oid collation, Datum *value, Vector * result);
|
||||
int IvfflatGetLists(Relation index);
|
||||
void IvfflatGetMetaPageInfo(Relation index, int *lists, int *dimensions);
|
||||
void IvfflatUpdateList(Relation index, ListInfo listInfo, BlockNumber insertPage, BlockNumber originalInsertPage, BlockNumber startPage, ForkNumber forkNum);
|
||||
void IvfflatCommitBuffer(Buffer buf, GenericXLogState *state);
|
||||
void IvfflatAppendPage(Relation index, Buffer *buf, Page *page, GenericXLogState **state, ForkNumber forkNum);
|
||||
|
||||
137
src/ivfscan.c
137
src/ivfscan.c
@@ -9,6 +9,7 @@
|
||||
#include "miscadmin.h"
|
||||
#include "pgstat.h"
|
||||
#include "storage/bufmgr.h"
|
||||
#include "storage/lmgr.h"
|
||||
|
||||
/*
|
||||
* Compare list distances
|
||||
@@ -31,36 +32,36 @@ CompareLists(const pairingheap_node *a, const pairingheap_node *b, void *arg)
|
||||
static void
|
||||
GetScanLists(IndexScanDesc scan, Datum value)
|
||||
{
|
||||
Buffer cbuf;
|
||||
Page cpage;
|
||||
IvfflatList list;
|
||||
OffsetNumber offno;
|
||||
OffsetNumber maxoffno;
|
||||
IvfflatScanOpaque so = (IvfflatScanOpaque) scan->opaque;
|
||||
BlockNumber nextblkno = IVFFLAT_HEAD_BLKNO;
|
||||
int listCount = 0;
|
||||
IvfflatScanOpaque so = (IvfflatScanOpaque) scan->opaque;
|
||||
double distance;
|
||||
IvfflatScanList *scanlist;
|
||||
double maxDistance = DBL_MAX;
|
||||
|
||||
/* Search all list pages */
|
||||
while (BlockNumberIsValid(nextblkno))
|
||||
{
|
||||
Buffer cbuf;
|
||||
Page cpage;
|
||||
OffsetNumber maxoffno;
|
||||
|
||||
cbuf = ReadBuffer(scan->indexRelation, 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))
|
||||
{
|
||||
list = (IvfflatList) PageGetItem(cpage, PageGetItemId(cpage, offno));
|
||||
IvfflatList list = (IvfflatList) PageGetItem(cpage, PageGetItemId(cpage, offno));
|
||||
double distance;
|
||||
|
||||
/* Use procinfo from the index instead of scan key for performance */
|
||||
distance = DatumGetFloat8(FunctionCall2Coll(so->procinfo, so->collation, PointerGetDatum(&list->center), value));
|
||||
|
||||
if (listCount < so->probes)
|
||||
{
|
||||
IvfflatScanList *scanlist;
|
||||
|
||||
scanlist = &so->lists[listCount];
|
||||
scanlist->startPage = list->startPage;
|
||||
scanlist->distance = distance;
|
||||
@@ -75,6 +76,8 @@ GetScanLists(IndexScanDesc scan, Datum value)
|
||||
}
|
||||
else if (distance < maxDistance)
|
||||
{
|
||||
IvfflatScanList *scanlist;
|
||||
|
||||
/* Remove */
|
||||
scanlist = (IvfflatScanList *) pairingheap_remove_first(so->listQueue);
|
||||
|
||||
@@ -101,14 +104,6 @@ static void
|
||||
GetScanItems(IndexScanDesc scan, Datum value)
|
||||
{
|
||||
IvfflatScanOpaque so = (IvfflatScanOpaque) scan->opaque;
|
||||
Buffer buf;
|
||||
Page page;
|
||||
IndexTuple itup;
|
||||
BlockNumber searchPage;
|
||||
OffsetNumber offno;
|
||||
OffsetNumber maxoffno;
|
||||
Datum datum;
|
||||
bool isnull;
|
||||
TupleDesc tupdesc = RelationGetDescr(scan->indexRelation);
|
||||
double tuples = 0;
|
||||
|
||||
@@ -128,19 +123,32 @@ GetScanItems(IndexScanDesc scan, Datum value)
|
||||
/* Search closest probes lists */
|
||||
while (!pairingheap_is_empty(so->listQueue))
|
||||
{
|
||||
searchPage = ((IvfflatScanList *) pairingheap_remove_first(so->listQueue))->startPage;
|
||||
BlockNumber searchPage = ((IvfflatScanList *) pairingheap_remove_first(so->listQueue))->startPage;
|
||||
|
||||
/* Search all entry pages for list */
|
||||
while (BlockNumberIsValid(searchPage))
|
||||
{
|
||||
Buffer buf;
|
||||
Page page;
|
||||
OffsetNumber maxoffno;
|
||||
|
||||
buf = ReadBufferExtended(scan->indexRelation, MAIN_FORKNUM, searchPage, RBM_NORMAL, bas);
|
||||
LockBuffer(buf, BUFFER_LOCK_SHARE);
|
||||
page = BufferGetPage(buf);
|
||||
maxoffno = PageGetMaxOffsetNumber(page);
|
||||
|
||||
for (offno = FirstOffsetNumber; offno <= maxoffno; offno = OffsetNumberNext(offno))
|
||||
for (OffsetNumber offno = FirstOffsetNumber; offno <= maxoffno; offno = OffsetNumberNext(offno))
|
||||
{
|
||||
itup = (IndexTuple) PageGetItem(page, PageGetItemId(page, offno));
|
||||
IndexTuple itup;
|
||||
Datum datum;
|
||||
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);
|
||||
|
||||
/*
|
||||
@@ -181,26 +189,52 @@ GetScanItems(IndexScanDesc scan, Datum value)
|
||||
}
|
||||
|
||||
/*
|
||||
* Get dimensions from metapage
|
||||
* Mark prior tuple as dead
|
||||
*/
|
||||
static int
|
||||
GetDimensions(Relation index)
|
||||
static void
|
||||
MarkPriorTupleDead(IndexScanDesc scan)
|
||||
{
|
||||
Buffer buf;
|
||||
IvfflatScanOpaque so = (IvfflatScanOpaque) scan->opaque;
|
||||
Buffer buf = so->buf;
|
||||
Page page;
|
||||
IvfflatMetaPage metap;
|
||||
int dimensions;
|
||||
OffsetNumber maxoffno;
|
||||
|
||||
buf = ReadBuffer(index, IVFFLAT_METAPAGE_BLKNO);
|
||||
/* 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);
|
||||
metap = IvfflatPageGetMeta(page);
|
||||
maxoffno = PageGetMaxOffsetNumber(page);
|
||||
|
||||
dimensions = metap->dimensions;
|
||||
for (OffsetNumber offno = FirstOffsetNumber; offno <= maxoffno; offno = OffsetNumberNext(offno))
|
||||
{
|
||||
ItemId itemid = PageGetItemId(page, offno);
|
||||
IndexTuple itup = (IndexTuple) PageGetItem(page, itemid);
|
||||
|
||||
UnlockReleaseBuffer(buf);
|
||||
/*
|
||||
* 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);
|
||||
}
|
||||
|
||||
return dimensions;
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
/* Unlock buffer */
|
||||
LockBuffer(buf, BUFFER_LOCK_UNLOCK);
|
||||
}
|
||||
|
||||
/*
|
||||
@@ -212,6 +246,7 @@ ivfflatbeginscan(Relation index, int nkeys, int norderbys)
|
||||
IndexScanDesc scan;
|
||||
IvfflatScanOpaque so;
|
||||
int lists;
|
||||
int dimensions;
|
||||
AttrNumber attNums[] = {1};
|
||||
Oid sortOperators[] = {Float8LessOperator};
|
||||
Oid sortCollations[] = {InvalidOid};
|
||||
@@ -219,7 +254,9 @@ ivfflatbeginscan(Relation index, int nkeys, int norderbys)
|
||||
int probes = ivfflat_probes;
|
||||
|
||||
scan = RelationGetIndexScan(index, nkeys, norderbys);
|
||||
lists = IvfflatGetLists(scan->indexRelation);
|
||||
|
||||
/* Get lists and dimensions from metapage */
|
||||
IvfflatGetMetaPageInfo(index, &lists, &dimensions);
|
||||
|
||||
if (probes > lists)
|
||||
probes = lists;
|
||||
@@ -227,7 +264,10 @@ ivfflatbeginscan(Relation index, int nkeys, int norderbys)
|
||||
so = (IvfflatScanOpaque) palloc(offsetof(IvfflatScanOpaqueData, lists) + probes * sizeof(IvfflatScanList));
|
||||
so->buf = InvalidBuffer;
|
||||
so->first = true;
|
||||
so->hasLock = false;
|
||||
ItemPointerSetInvalid(&so->heaptid);
|
||||
so->probes = probes;
|
||||
so->dimensions = dimensions;
|
||||
|
||||
/* Set support functions */
|
||||
so->procinfo = index_getprocinfo(index, 1, IVFFLAT_DISTANCE_PROC);
|
||||
@@ -241,7 +281,7 @@ ivfflatbeginscan(Relation index, int nkeys, int norderbys)
|
||||
so->tupdesc = CreateTemplateTupleDesc(3, false);
|
||||
#endif
|
||||
TupleDescInitEntry(so->tupdesc, (AttrNumber) 1, "distance", FLOAT8OID, -1, 0);
|
||||
TupleDescInitEntry(so->tupdesc, (AttrNumber) 2, "tid", TIDOID, -1, 0);
|
||||
TupleDescInitEntry(so->tupdesc, (AttrNumber) 2, "heaptid", TIDOID, -1, 0);
|
||||
TupleDescInitEntry(so->tupdesc, (AttrNumber) 3, "indexblkno", INT4OID, -1, 0);
|
||||
|
||||
/* Prep sort */
|
||||
@@ -274,6 +314,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)
|
||||
@@ -308,8 +349,15 @@ ivfflatgettuple(IndexScanDesc scan, ScanDirection dir)
|
||||
if (scan->orderByData == NULL)
|
||||
elog(ERROR, "cannot scan ivfflat index without order");
|
||||
|
||||
/* Get a shared lock for non-MVCC snapshots */
|
||||
if (!so->hasLock && !IsMVCCSnapshot(scan->xs_snapshot))
|
||||
{
|
||||
so->hasLock = true;
|
||||
LockPage(scan->indexRelation, IVFFLAT_SCAN_LOCK, ShareLock);
|
||||
}
|
||||
|
||||
if (scan->orderByData->sk_flags & SK_ISNULL)
|
||||
value = PointerGetDatum(InitVector(GetDimensions(scan->indexRelation)));
|
||||
value = PointerGetDatum(InitVector(so->dimensions));
|
||||
else
|
||||
{
|
||||
value = scan->orderByData->sk_argument;
|
||||
@@ -331,18 +379,27 @@ 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 tid = (ItemPointer) DatumGetPointer(slot_getattr(so->slot, 2, &so->isnull));
|
||||
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 = *tid;
|
||||
scan->xs_heaptid = *heaptid;
|
||||
#else
|
||||
scan->xs_ctup.t_self = *tid;
|
||||
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);
|
||||
@@ -374,6 +431,10 @@ ivfflatendscan(IndexScanDesc scan)
|
||||
if (BufferIsValid(so->buf))
|
||||
ReleaseBuffer(so->buf);
|
||||
|
||||
/* Release lock */
|
||||
if (so->hasLock)
|
||||
UnlockPage(scan->indexRelation, IVFFLAT_SCAN_LOCK, ShareLock);
|
||||
|
||||
pairingheap_free(so->listQueue);
|
||||
tuplesort_end(so->sortstate);
|
||||
|
||||
|
||||
@@ -172,6 +172,29 @@ IvfflatAppendPage(Relation index, Buffer *buf, Page *page, GenericXLogState **st
|
||||
*buf = newbuf;
|
||||
}
|
||||
|
||||
/*
|
||||
* Get the metapage info
|
||||
*/
|
||||
void
|
||||
IvfflatGetMetaPageInfo(Relation index, int *lists, int *dimensions)
|
||||
{
|
||||
Buffer buf;
|
||||
Page page;
|
||||
IvfflatMetaPage metap;
|
||||
|
||||
buf = ReadBuffer(index, IVFFLAT_METAPAGE_BLKNO);
|
||||
LockBuffer(buf, BUFFER_LOCK_SHARE);
|
||||
page = BufferGetPage(buf);
|
||||
metap = IvfflatPageGetMeta(page);
|
||||
|
||||
*lists = metap->lists;
|
||||
|
||||
if (dimensions != NULL)
|
||||
*dimensions = metap->dimensions;
|
||||
|
||||
UnlockReleaseBuffer(buf);
|
||||
}
|
||||
|
||||
/*
|
||||
* Update the start or insert page of a list
|
||||
*/
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
#include "commands/vacuum.h"
|
||||
#include "ivfflat.h"
|
||||
#include "storage/bufmgr.h"
|
||||
#include "storage/lmgr.h"
|
||||
|
||||
/*
|
||||
* Bulk delete tuples from the index
|
||||
@@ -65,14 +66,10 @@ ivfflatbulkdelete(IndexVacuumInfo *info, IndexBulkDeleteResult *stats,
|
||||
|
||||
vacuum_delay_point();
|
||||
|
||||
buf = ReadBufferExtended(index, MAIN_FORKNUM, searchPage, RBM_NORMAL, bas);
|
||||
/* Ensure no in-flight index scans for non-MVCC snapshots */
|
||||
LockPage(index, IVFFLAT_SCAN_LOCK, ExclusiveLock);
|
||||
|
||||
/*
|
||||
* ambulkdelete cannot delete entries from pages that are
|
||||
* pinned by other backends
|
||||
*
|
||||
* https://www.postgresql.org/docs/current/index-locking.html
|
||||
*/
|
||||
buf = ReadBufferExtended(index, MAIN_FORKNUM, searchPage, RBM_NORMAL, bas);
|
||||
LockBufferForCleanup(buf);
|
||||
|
||||
state = GenericXLogStart(index);
|
||||
@@ -114,6 +111,8 @@ ivfflatbulkdelete(IndexVacuumInfo *info, IndexBulkDeleteResult *stats,
|
||||
GenericXLogAbort(state);
|
||||
|
||||
UnlockReleaseBuffer(buf);
|
||||
|
||||
UnlockPage(index, IVFFLAT_SCAN_LOCK, ExclusiveLock);
|
||||
}
|
||||
|
||||
/*
|
||||
|
||||
@@ -695,6 +695,8 @@ vector_spherical_distance(PG_FUNCTION_ARGS)
|
||||
{
|
||||
Vector *a = PG_GETARG_VECTOR_P(0);
|
||||
Vector *b = PG_GETARG_VECTOR_P(1);
|
||||
float *ax = a->x;
|
||||
float *bx = b->x;
|
||||
float dp = 0.0;
|
||||
double distance;
|
||||
|
||||
@@ -702,7 +704,7 @@ vector_spherical_distance(PG_FUNCTION_ARGS)
|
||||
|
||||
/* Auto-vectorized */
|
||||
for (int i = 0; i < a->dim; i++)
|
||||
dp += a->x[i] * b->x[i];
|
||||
dp += ax[i] * bx[i];
|
||||
|
||||
distance = (double) dp;
|
||||
|
||||
|
||||
@@ -106,6 +106,12 @@ SELECT cosine_distance('[1,1]', '[1,1]');
|
||||
0
|
||||
(1 row)
|
||||
|
||||
SELECT cosine_distance('[1,0]', '[0,2]');
|
||||
cosine_distance
|
||||
-----------------
|
||||
1
|
||||
(1 row)
|
||||
|
||||
SELECT cosine_distance('[1,1]', '[-1,-1]');
|
||||
cosine_distance
|
||||
-----------------
|
||||
|
||||
@@ -25,6 +25,7 @@ SELECT inner_product('[3e38]', '[3e38]');
|
||||
SELECT cosine_distance('[1,2]', '[2,4]');
|
||||
SELECT cosine_distance('[1,2]', '[0,0]');
|
||||
SELECT cosine_distance('[1,1]', '[1,1]');
|
||||
SELECT cosine_distance('[1,0]', '[0,2]');
|
||||
SELECT cosine_distance('[1,1]', '[-1,-1]');
|
||||
SELECT cosine_distance('[1,2]', '[3]');
|
||||
SELECT cosine_distance('[1,1]', '[1.1,1.1]');
|
||||
|
||||
@@ -19,8 +19,6 @@ sub test_index_replay
|
||||
|
||||
# Wait for replica to catch up
|
||||
my $applname = $node_replica->name;
|
||||
|
||||
my $server_version_num = $node_primary->safe_psql("postgres", "SHOW server_version_num");
|
||||
my $caughtup_query = "SELECT pg_current_wal_lsn() <= replay_lsn FROM pg_stat_replication WHERE application_name = '$applname';";
|
||||
$node_primary->poll_query_until('postgres', $caughtup_query)
|
||||
or die "Timed out while waiting for replica 1 to catch up";
|
||||
|
||||
@@ -19,8 +19,6 @@ sub test_index_replay
|
||||
|
||||
# Wait for replica to catch up
|
||||
my $applname = $node_replica->name;
|
||||
|
||||
my $server_version_num = $node_primary->safe_psql("postgres", "SHOW server_version_num");
|
||||
my $caughtup_query = "SELECT pg_current_wal_lsn() <= replay_lsn FROM pg_stat_replication WHERE application_name = '$applname';";
|
||||
$node_primary->poll_query_until('postgres', $caughtup_query)
|
||||
or die "Timed out while waiting for replica 1 to catch up";
|
||||
@@ -38,9 +36,8 @@ sub test_index_replay
|
||||
);
|
||||
|
||||
# Run test queries and compare their result
|
||||
# Query replica first since index scan on primary can generate WAL removing tuples
|
||||
my $replica_result = $node_replica->safe_psql("postgres", $queries);
|
||||
my $primary_result = $node_primary->safe_psql("postgres", $queries);
|
||||
my $replica_result = $node_replica->safe_psql("postgres", $queries);
|
||||
|
||||
is($primary_result, $replica_result, "$test_name: query result matches");
|
||||
return;
|
||||
|
||||
@@ -23,27 +23,25 @@ sub insert_vectors
|
||||
|
||||
sub test_duplicates
|
||||
{
|
||||
my ($exp) = @_;
|
||||
|
||||
my $res = $node->safe_psql("postgres", qq(
|
||||
SET enable_seqscan = off;
|
||||
SET hnsw.ef_search = 1;
|
||||
SELECT COUNT(*) FROM (SELECT * FROM tst ORDER BY v <-> '[1,1,1]') t;
|
||||
));
|
||||
is($res, $exp);
|
||||
is($res, 10);
|
||||
}
|
||||
|
||||
# Test duplicates with build
|
||||
insert_vectors();
|
||||
$node->safe_psql("postgres", "CREATE INDEX idx ON tst USING hnsw (v vector_l2_ops);");
|
||||
test_duplicates(10);
|
||||
test_duplicates();
|
||||
|
||||
# Reset
|
||||
$node->safe_psql("postgres", "TRUNCATE tst;");
|
||||
|
||||
# Test duplicates with inserts
|
||||
insert_vectors();
|
||||
test_duplicates(10);
|
||||
test_duplicates();
|
||||
|
||||
# Test fallback path for inserts
|
||||
$node->pgbench(
|
||||
@@ -57,15 +55,4 @@ $node->pgbench(
|
||||
}
|
||||
);
|
||||
|
||||
# Reset
|
||||
$node->safe_psql("postgres", "TRUNCATE tst;");
|
||||
|
||||
# Test deletes with index scan
|
||||
$node->safe_psql("postgres", "INSERT INTO tst SELECT '[1,1,1]' FROM generate_series(1, 10) i;");
|
||||
$node->safe_psql("postgres", "DELETE FROM tst WHERE ctid IN (SELECT ctid FROM tst ORDER BY random() LIMIT 5);");
|
||||
for (1 .. 3)
|
||||
{
|
||||
test_duplicates(5);
|
||||
}
|
||||
|
||||
done_testing();
|
||||
|
||||
43
test/t/018_ivfflat_deletes.pl
Normal file
43
test/t/018_ivfflat_deletes.pl
Normal file
@@ -0,0 +1,43 @@
|
||||
use strict;
|
||||
use warnings;
|
||||
use PostgresNode;
|
||||
use TestLib;
|
||||
use Test::More;
|
||||
|
||||
my $dim = 3;
|
||||
|
||||
my $array_sql = join(",", ('random()') x $dim);
|
||||
|
||||
# Initialize node
|
||||
my $node = get_new_node('node');
|
||||
$node->init;
|
||||
$node->start;
|
||||
|
||||
# Create table and index
|
||||
$node->safe_psql("postgres", "CREATE EXTENSION vector;");
|
||||
$node->safe_psql("postgres", "CREATE TABLE tst (i serial, v vector($dim));");
|
||||
$node->safe_psql("postgres",
|
||||
"INSERT INTO tst (v) SELECT ARRAY[$array_sql] FROM generate_series(1, 10000) i;"
|
||||
);
|
||||
$node->safe_psql("postgres", "CREATE INDEX ON tst USING ivfflat (v vector_l2_ops);");
|
||||
|
||||
# Delete data
|
||||
$node->safe_psql("postgres", "DELETE FROM tst WHERE i % 100 != 0;");
|
||||
|
||||
my $exp = $node->safe_psql("postgres", qq(
|
||||
SET enable_indexscan = off;
|
||||
SELECT i FROM tst ORDER BY v <-> '[0,0,0]';
|
||||
));
|
||||
|
||||
# Run twice to make sure correct tuples marked as dead
|
||||
for (1 .. 2)
|
||||
{
|
||||
my $res = $node->safe_psql("postgres", qq(
|
||||
SET enable_seqscan = off;
|
||||
SET ivfflat.probes = 100;
|
||||
SELECT i FROM tst ORDER BY v <-> '[0,0,0]';
|
||||
));
|
||||
is($res, $exp);
|
||||
}
|
||||
|
||||
done_testing();
|
||||
@@ -1,4 +1,4 @@
|
||||
comment = 'vector data type and ivfflat access method'
|
||||
comment = 'vector data type and ivfflat and hnsw access methods'
|
||||
default_version = '0.5.0'
|
||||
module_pathname = '$libdir/vector'
|
||||
relocatable = true
|
||||
|
||||
Reference in New Issue
Block a user