Compare commits

..

3 Commits

Author SHA1 Message Date
Andrew Kane
8fcf77f89a Added support for bigint attributes [skip ci] 2023-11-15 15:37:27 -08:00
Andrew Kane
08bd246529 Fixed vacuum test [skip ci] 2023-11-10 14:03:13 -08:00
Andrew Kane
f57f2b6821 Added support for inline filtering with HNSW 2023-11-10 13:28:48 -08:00
13 changed files with 519 additions and 597 deletions

View File

@@ -1,6 +1,6 @@
## 0.5.2 (unreleased) ## 0.6.0 (unreleased)
- Added support for on-disk parallel index builds for HNSW - Added support for inline filtering with HNSW
## 0.5.1 (2023-10-10) ## 0.5.1 (2023-10-10)

View File

@@ -315,6 +315,12 @@ Create an index on one [or more](https://www.postgresql.org/docs/current/indexes
CREATE INDEX ON items (category_id); CREATE INDEX ON items (category_id);
``` ```
Or a composite HNSW index for approximate search (added in 0.6.0)
```sql
CREATE INDEX ON items USING hnsw (embedding vector_l2_ops, category_id);
```
Or a [partial index](https://www.postgresql.org/docs/current/indexes-partial.html) on the vector column for approximate search Or a [partial index](https://www.postgresql.org/docs/current/indexes-partial.html) on the vector column for approximate search
```sql ```sql
@@ -678,16 +684,6 @@ SELECT extversion FROM pg_extension WHERE extname = 'vector';
## Upgrade Notes ## Upgrade Notes
### 0.6.0
If upgrading with Postgres < 13, remove this line from `sql/vector--0.5.1--0.6.0.sql`:
```sql
ALTER TYPE vector SET (STORAGE = external);
```
Then run `make install` and `ALTER EXTENSION vector UPDATE;`.
### 0.4.0 ### 0.4.0
If upgrading with Postgres < 13, remove this line from `sql/vector--0.3.2--0.4.0.sql`: If upgrading with Postgres < 13, remove this line from `sql/vector--0.3.2--0.4.0.sql`:
@@ -722,6 +718,7 @@ Thanks to:
- [k-means++: The Advantage of Careful Seeding](https://theory.stanford.edu/~sergei/papers/kMeansPP-soda.pdf) - [k-means++: The Advantage of Careful Seeding](https://theory.stanford.edu/~sergei/papers/kMeansPP-soda.pdf)
- [Concept Decompositions for Large Sparse Text Data using Clustering](https://www.cs.utexas.edu/users/inderjit/public_papers/concept_mlj.pdf) - [Concept Decompositions for Large Sparse Text Data using Clustering](https://www.cs.utexas.edu/users/inderjit/public_papers/concept_mlj.pdf)
- [Efficient and Robust Approximate Nearest Neighbor Search using Hierarchical Navigable Small World Graphs](https://arxiv.org/ftp/arxiv/papers/1603/1603.09320.pdf) - [Efficient and Robust Approximate Nearest Neighbor Search using Hierarchical Navigable Small World Graphs](https://arxiv.org/ftp/arxiv/papers/1603/1603.09320.pdf)
- [HQANN: Efficient and Robust Similarity Search for Hybrid Queries with Structured and Unstructured Constraints](https://arxiv.org/pdf/2207.07940.pdf)
## History ## History

View File

@@ -1,5 +1,18 @@
-- complain if script is sourced in psql, rather than via CREATE EXTENSION -- complain if script is sourced in psql, rather than via CREATE EXTENSION
\echo Use "ALTER EXTENSION vector UPDATE TO '0.6.0'" to load this file. \quit \echo Use "ALTER EXTENSION vector UPDATE TO '0.6.0'" to load this file. \quit
-- remove this single line for Postgres < 13 CREATE FUNCTION hnsw_attribute_distance(integer, integer) RETURNS float8
ALTER TYPE vector SET (STORAGE = external); AS 'MODULE_PATHNAME', 'hnsw_int4_attribute_distance' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
CREATE FUNCTION hnsw_attribute_distance(bigint, bigint) RETURNS float8
AS 'MODULE_PATHNAME', 'hnsw_int8_attribute_distance' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
CREATE OPERATOR CLASS vector_integer_ops
DEFAULT FOR TYPE integer USING hnsw AS
OPERATOR 2 = (integer, integer),
FUNCTION 3 hnsw_attribute_distance(integer, integer);
CREATE OPERATOR CLASS vector_bigint_ops
DEFAULT FOR TYPE bigint USING hnsw AS
OPERATOR 2 = (bigint, bigint),
FUNCTION 3 hnsw_attribute_distance(bigint, bigint);

View File

@@ -26,7 +26,7 @@ CREATE TYPE vector (
TYPMOD_IN = vector_typmod_in, TYPMOD_IN = vector_typmod_in,
RECEIVE = vector_recv, RECEIVE = vector_recv,
SEND = vector_send, SEND = vector_send,
STORAGE = external STORAGE = extended
); );
-- functions -- functions
@@ -290,3 +290,21 @@ CREATE OPERATOR CLASS vector_cosine_ops
OPERATOR 1 <=> (vector, vector) FOR ORDER BY float_ops, OPERATOR 1 <=> (vector, vector) FOR ORDER BY float_ops,
FUNCTION 1 vector_negative_inner_product(vector, vector), FUNCTION 1 vector_negative_inner_product(vector, vector),
FUNCTION 2 vector_norm(vector); FUNCTION 2 vector_norm(vector);
-- hnsw attributes
CREATE FUNCTION hnsw_attribute_distance(integer, integer) RETURNS float8
AS 'MODULE_PATHNAME', 'hnsw_int4_attribute_distance' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
CREATE FUNCTION hnsw_attribute_distance(bigint, bigint) RETURNS float8
AS 'MODULE_PATHNAME', 'hnsw_int8_attribute_distance' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
CREATE OPERATOR CLASS vector_integer_ops
DEFAULT FOR TYPE integer USING hnsw AS
OPERATOR 2 = (integer, integer),
FUNCTION 3 hnsw_attribute_distance(integer, integer);
CREATE OPERATOR CLASS vector_bigint_ops
DEFAULT FOR TYPE bigint USING hnsw AS
OPERATOR 2 = (bigint, bigint),
FUNCTION 3 hnsw_attribute_distance(bigint, bigint);

View File

@@ -14,7 +14,6 @@
#endif #endif
int hnsw_ef_search; int hnsw_ef_search;
bool hnsw_enable_parallel_build;
static relopt_kind hnsw_relopt_kind; static relopt_kind hnsw_relopt_kind;
/* /*
@@ -40,11 +39,6 @@ HnswInit(void)
DefineCustomIntVariable("hnsw.ef_search", "Sets the size of the dynamic candidate list for search", DefineCustomIntVariable("hnsw.ef_search", "Sets the size of the dynamic candidate list for search",
"Valid range is 1..1000.", &hnsw_ef_search, "Valid range is 1..1000.", &hnsw_ef_search,
HNSW_DEFAULT_EF_SEARCH, HNSW_MIN_EF_SEARCH, HNSW_MAX_EF_SEARCH, PGC_USERSET, 0, NULL, NULL, NULL); HNSW_DEFAULT_EF_SEARCH, HNSW_MIN_EF_SEARCH, HNSW_MAX_EF_SEARCH, PGC_USERSET, 0, NULL, NULL, NULL);
/* Behind a variable for now since can be slower than building in memory */
DefineCustomBoolVariable("hnsw.enable_parallel_build", "Enables or disables building indexes in parallel",
NULL, &hnsw_enable_parallel_build,
false, PGC_USERSET, 0, NULL, NULL, NULL);
} }
/* /*
@@ -173,7 +167,7 @@ hnswhandler(PG_FUNCTION_ARGS)
IndexAmRoutine *amroutine = makeNode(IndexAmRoutine); IndexAmRoutine *amroutine = makeNode(IndexAmRoutine);
amroutine->amstrategies = 0; amroutine->amstrategies = 0;
amroutine->amsupport = 2; amroutine->amsupport = 3;
#if PG_VERSION_NUM >= 130000 #if PG_VERSION_NUM >= 130000
amroutine->amoptsprocnum = 0; amroutine->amoptsprocnum = 0;
#endif #endif
@@ -181,7 +175,7 @@ hnswhandler(PG_FUNCTION_ARGS)
amroutine->amcanorderbyop = true; amroutine->amcanorderbyop = true;
amroutine->amcanbackward = false; /* can change direction mid-scan */ amroutine->amcanbackward = false; /* can change direction mid-scan */
amroutine->amcanunique = false; amroutine->amcanunique = false;
amroutine->amcanmulticol = false; amroutine->amcanmulticol = true;
amroutine->amoptionalkey = true; amroutine->amoptionalkey = true;
amroutine->amsearcharray = false; amroutine->amsearcharray = false;
amroutine->amsearchnulls = false; amroutine->amsearchnulls = false;
@@ -228,3 +222,31 @@ hnswhandler(PG_FUNCTION_ARGS)
PG_RETURN_POINTER(amroutine); PG_RETURN_POINTER(amroutine);
} }
/*
* Get the distance between two int4 attributes
*/
PGDLLEXPORT PG_FUNCTION_INFO_V1(hnsw_int4_attribute_distance);
Datum
hnsw_int4_attribute_distance(PG_FUNCTION_ARGS)
{
int32 a = PG_GETARG_INT32(0);
int32 b = PG_GETARG_INT32(1);
double distance = ((double) a) - ((double) b);
PG_RETURN_FLOAT8(distance);
}
/*
* Get the distance between two int8 attributes
*/
PGDLLEXPORT PG_FUNCTION_INFO_V1(hnsw_int8_attribute_distance);
Datum
hnsw_int8_attribute_distance(PG_FUNCTION_ARGS)
{
int64 a = PG_GETARG_INT64(0);
int64 b = PG_GETARG_INT64(1);
double distance = ((double) a) - ((double) b);
PG_RETURN_FLOAT8(distance);
}

View File

@@ -4,7 +4,6 @@
#include "postgres.h" #include "postgres.h"
#include "access/generic_xlog.h" #include "access/generic_xlog.h"
#include "access/parallel.h"
#include "access/reloptions.h" #include "access/reloptions.h"
#include "nodes/execnodes.h" #include "nodes/execnodes.h"
#include "port.h" /* for random() */ #include "port.h" /* for random() */
@@ -15,15 +14,12 @@
#error "Requires PostgreSQL 11+" #error "Requires PostgreSQL 11+"
#endif #endif
#if PG_VERSION_NUM < 120000
#include "access/relscan.h"
#endif
#define HNSW_MAX_DIM 2000 #define HNSW_MAX_DIM 2000
/* Support functions */ /* Support functions */
#define HNSW_DISTANCE_PROC 1 #define HNSW_DISTANCE_PROC 1
#define HNSW_NORM_PROC 2 #define HNSW_NORM_PROC 2
#define HNSW_ATTRIBUTE_DISTANCE_PROC 3
#define HNSW_VERSION 1 #define HNSW_VERSION 1
#define HNSW_MAGIC_NUMBER 0xA953A953 #define HNSW_MAGIC_NUMBER 0xA953A953
@@ -95,7 +91,6 @@
/* Variables */ /* Variables */
extern int hnsw_ef_search; extern int hnsw_ef_search;
extern bool hnsw_enable_parallel_build;
typedef struct HnswNeighborArray HnswNeighborArray; typedef struct HnswNeighborArray HnswNeighborArray;
@@ -110,6 +105,7 @@ typedef struct HnswElementData
OffsetNumber neighborOffno; OffsetNumber neighborOffno;
BlockNumber neighborPage; BlockNumber neighborPage;
Datum value; Datum value;
IndexTuple itup;
} HnswElementData; } HnswElementData;
typedef HnswElementData * HnswElement; typedef HnswElementData * HnswElement;
@@ -142,49 +138,6 @@ typedef struct HnswOptions
int efConstruction; /* size of dynamic candidate list */ int efConstruction; /* size of dynamic candidate list */
} HnswOptions; } HnswOptions;
typedef struct HnswSpool
{
Relation heap;
Relation index;
} HnswSpool;
typedef struct HnswShared
{
/* Immutable state */
Oid heaprelid;
Oid indexrelid;
bool isconcurrent;
int scantuplesortstates;
/* Worker progress */
ConditionVariable workersdonecv;
/* Mutex for mutable state */
slock_t mutex;
/* Mutable state */
int nparticipantsdone;
double reltuples;
double indtuples;
#if PG_VERSION_NUM < 120000
ParallelHeapScanDescData heapdesc; /* must come last */
#endif
} HnswShared;
#if PG_VERSION_NUM >= 120000
#define ParallelTableScanFromHnswShared(shared) \
(ParallelTableScanDesc) ((char *) (shared) + BUFFERALIGN(sizeof(HnswShared)))
#endif
typedef struct HnswLeader
{
ParallelContext *pcxt;
int nparticipanttuplesorts;
HnswShared *hnswshared;
Snapshot snapshot;
} HnswLeader;
typedef struct HnswBuildState typedef struct HnswBuildState
{ {
/* Info */ /* Info */
@@ -203,9 +156,9 @@ typedef struct HnswBuildState
double reltuples; double reltuples;
/* Support functions */ /* Support functions */
FmgrInfo *procinfo; FmgrInfo **procinfos;
FmgrInfo *normprocinfo; FmgrInfo *normprocinfo;
Oid collation; Oid *collations;
/* Variables */ /* Variables */
List *elements; List *elements;
@@ -214,14 +167,11 @@ typedef struct HnswBuildState
int maxLevel; int maxLevel;
long memoryLeft; long memoryLeft;
bool flushed; bool flushed;
bool useIndexTuple;
Vector *normvec; Vector *normvec;
/* Memory */ /* Memory */
MemoryContext tmpCtx; MemoryContext tmpCtx;
/* Parallel builds */
HnswLeader *hnswleader;
HnswShared *hnswshared;
} HnswBuildState; } HnswBuildState;
typedef struct HnswMetaPageData typedef struct HnswMetaPageData
@@ -279,9 +229,9 @@ typedef struct HnswScanOpaqueData
MemoryContext tmpCtx; MemoryContext tmpCtx;
/* Support functions */ /* Support functions */
FmgrInfo *procinfo; FmgrInfo **procinfos;
FmgrInfo *normprocinfo; FmgrInfo *normprocinfo;
Oid collation; Oid *collations;
} HnswScanOpaqueData; } HnswScanOpaqueData;
typedef HnswScanOpaqueData * HnswScanOpaque; typedef HnswScanOpaqueData * HnswScanOpaque;
@@ -299,8 +249,8 @@ typedef struct HnswVacuumState
int efConstruction; int efConstruction;
/* Support functions */ /* Support functions */
FmgrInfo *procinfo; FmgrInfo **procinfos;
Oid collation; Oid *collations;
/* Variables */ /* Variables */
HTAB *deleted; HTAB *deleted;
@@ -322,27 +272,27 @@ Buffer HnswNewBuffer(Relation index, ForkNumber forkNum);
void HnswInitPage(Buffer buf, Page page); void HnswInitPage(Buffer buf, Page page);
void HnswInitRegisterPage(Relation index, Buffer *buf, Page *page, GenericXLogState **state); void HnswInitRegisterPage(Relation index, Buffer *buf, Page *page, GenericXLogState **state);
void HnswInit(void); void HnswInit(void);
List *HnswSearchLayer(Datum q, List *ep, int ef, int lc, Relation index, FmgrInfo *procinfo, Oid collation, int m, bool inserting, HnswElement skipElement); List *HnswSearchLayer(Datum q, IndexTuple qtup, ScanKeyData *keyData, List *ep, int ef, int lc, Relation index, FmgrInfo **procinfos, Oid *collations, int m, bool loadVec, HnswElement skipElement, bool inMemory);
HnswElement HnswGetEntryPoint(Relation index); HnswElement HnswGetEntryPoint(Relation index);
void HnswGetMetaPageInfo(Relation index, int *m, HnswElement * entryPoint); void HnswGetMetaPageInfo(Relation index, int *m, HnswElement * entryPoint);
HnswElement HnswInitElement(ItemPointer tid, int m, double ml, int maxLevel); HnswElement HnswInitElement(ItemPointer tid, int m, double ml, int maxLevel);
void HnswFreeElement(HnswElement element); void HnswFreeElement(HnswElement element);
HnswElement HnswInitElementFromBlock(BlockNumber blkno, OffsetNumber offno); HnswElement HnswInitElementFromBlock(BlockNumber blkno, OffsetNumber offno);
void HnswInsertElement(HnswElement element, HnswElement entryPoint, Relation index, FmgrInfo *procinfo, Oid collation, int m, int efConstruction, bool existing); void HnswInsertElement(HnswElement element, HnswElement entryPoint, Relation index, FmgrInfo **procinfos, Oid *collations, int m, int efConstruction, bool existing, bool inMemory);
HnswElement HnswFindDuplicate(HnswElement e); HnswElement HnswFindDuplicate(HnswElement e, Relation index);
HnswCandidate *HnswEntryCandidate(HnswElement em, Datum q, Relation rel, FmgrInfo *procinfo, Oid collation, bool loadVec); HnswCandidate *HnswEntryCandidate(HnswElement em, Datum q, IndexTuple qtup, ScanKeyData *keyData, Relation rel, FmgrInfo **procinfos, Oid *collations, bool loadVec, bool inMemory);
void HnswUpdateMetaPage(Relation index, int updateEntry, HnswElement entryPoint, BlockNumber insertPage, ForkNumber forkNum); void HnswUpdateMetaPage(Relation index, int updateEntry, HnswElement entryPoint, BlockNumber insertPage, ForkNumber forkNum);
void HnswSetNeighborTuple(HnswNeighborTuple ntup, HnswElement e, int m); void HnswSetNeighborTuple(HnswNeighborTuple ntup, HnswElement e, int m);
void HnswAddHeapTid(HnswElement element, ItemPointer heaptid); void HnswAddHeapTid(HnswElement element, ItemPointer heaptid);
void HnswInitNeighbors(HnswElement element, int m); void HnswInitNeighbors(HnswElement element, int m);
bool HnswInsertTuple(Relation index, Datum *values, bool *isnull, ItemPointer heap_tid, Relation heapRel); bool HnswInsertTuple(Relation index, Datum *values, bool *isnull, ItemPointer heap_tid, Relation heapRel);
void HnswUpdateNeighborPages(Relation index, FmgrInfo *procinfo, Oid collation, HnswElement e, int m, bool checkExisting); void HnswUpdateNeighborPages(Relation index, FmgrInfo **procinfos, Oid *collations, HnswElement e, int m, bool checkExisting);
void HnswLoadElementFromTuple(HnswElement element, HnswElementTuple etup, bool loadHeaptids, bool loadVec); void HnswLoadElementFromTuple(HnswElement element, HnswElementTuple etup, bool loadHeaptids, bool loadVec, Relation index);
void HnswLoadElement(HnswElement element, float *distance, Datum *q, Relation index, FmgrInfo *procinfo, Oid collation, bool loadVec); void HnswLoadElement(HnswElement element, float *distance, Datum *q, IndexTuple qtup, ScanKeyData *keyData, Relation index, FmgrInfo **procinfos, Oid *collations, bool loadVec);
void HnswSetElementTuple(HnswElementTuple etup, HnswElement element); void HnswSetElementTuple(HnswElementTuple etup, HnswElement element, bool useIndexTuple);
void HnswUpdateConnection(HnswElement element, HnswCandidate * hc, int m, int lc, int *updateIdx, Relation index, FmgrInfo *procinfo, Oid collation); void HnswUpdateConnection(HnswElement element, HnswCandidate * hc, int m, int lc, int *updateIdx, Relation index, FmgrInfo **procinfos, Oid *collations, bool inMemory);
void HnswLoadNeighbors(HnswElement element, Relation index, int m); void HnswLoadNeighbors(HnswElement element, Relation index, int m);
PGDLLEXPORT void HnswParallelBuildMain(dsm_segment *seg, shm_toc *toc); void HnswElementSetData(HnswElement element, Relation index, Datum value, Datum *values, bool *isnull);
/* Index access methods */ /* Index access methods */
IndexBuildResult *hnswbuild(Relation heap, Relation index, IndexInfo *indexInfo); IndexBuildResult *hnswbuild(Relation heap, Relation index, IndexInfo *indexInfo);
@@ -359,5 +309,6 @@ IndexScanDesc hnswbeginscan(Relation index, int nkeys, int norderbys);
void hnswrescan(IndexScanDesc scan, ScanKey keys, int nkeys, ScanKey orderbys, int norderbys); void hnswrescan(IndexScanDesc scan, ScanKey keys, int nkeys, ScanKey orderbys, int norderbys);
bool hnswgettuple(IndexScanDesc scan, ScanDirection dir); bool hnswgettuple(IndexScanDesc scan, ScanDirection dir);
void hnswendscan(IndexScanDesc scan); void hnswendscan(IndexScanDesc scan);
FmgrInfo **HnswInitProcinfos(Relation index);
#endif #endif

View File

@@ -2,15 +2,12 @@
#include <math.h> #include <math.h>
#include "access/parallel.h"
#include "access/xact.h"
#include "catalog/index.h" #include "catalog/index.h"
#include "hnsw.h" #include "hnsw.h"
#include "miscadmin.h" #include "miscadmin.h"
#include "lib/pairingheap.h" #include "lib/pairingheap.h"
#include "nodes/pg_list.h" #include "nodes/pg_list.h"
#include "storage/bufmgr.h" #include "storage/bufmgr.h"
#include "tcop/tcopprot.h"
#include "utils/datum.h" #include "utils/datum.h"
#include "utils/memutils.h" #include "utils/memutils.h"
@@ -39,23 +36,6 @@
#define UpdateProgress(index, val) ((void)val) #define UpdateProgress(index, val) ((void)val)
#endif #endif
#if PG_VERSION_NUM >= 140000
#include "utils/backend_status.h"
#include "utils/wait_event.h"
#endif
#if PG_VERSION_NUM >= 120000
#include "access/table.h"
#include "optimizer/optimizer.h"
#else
#include "access/heapam.h"
#include "optimizer/planner.h"
#include "pgstat.h"
#endif
#define PARALLEL_KEY_HNSW_SHARED UINT64CONST(0xA000000000000001)
#define PARALLEL_KEY_QUERY_TEXT UINT64CONST(0xA000000000000002)
/* /*
* Create the metapage * Create the metapage
*/ */
@@ -126,6 +106,7 @@ CreateElementPages(HnswBuildState * buildstate)
{ {
Relation index = buildstate->index; Relation index = buildstate->index;
ForkNumber forkNum = buildstate->forkNum; ForkNumber forkNum = buildstate->forkNum;
bool useIndexTuple = buildstate->useIndexTuple;
Size etupAllocSize; Size etupAllocSize;
Size maxSize; Size maxSize;
HnswElementTuple etup; HnswElementTuple etup;
@@ -161,7 +142,7 @@ CreateElementPages(HnswBuildState * buildstate)
MemSet(etup, 0, etupAllocSize); MemSet(etup, 0, etupAllocSize);
/* Calculate sizes */ /* Calculate sizes */
etupSize = HNSW_ELEMENT_TUPLE_SIZE(VARSIZE_ANY(DatumGetPointer(element->value))); etupSize = HNSW_ELEMENT_TUPLE_SIZE(useIndexTuple ? IndexTupleSize(element->itup) : VARSIZE_ANY(DatumGetPointer(element->value)));
ntupSize = HNSW_NEIGHBOR_TUPLE_SIZE(element->level, buildstate->m); ntupSize = HNSW_NEIGHBOR_TUPLE_SIZE(element->level, buildstate->m);
combinedSize = etupSize + ntupSize + sizeof(ItemIdData); combinedSize = etupSize + ntupSize + sizeof(ItemIdData);
@@ -169,7 +150,7 @@ CreateElementPages(HnswBuildState * buildstate)
if (etupSize > etupAllocSize) if (etupSize > etupAllocSize)
elog(ERROR, "index tuple too large"); elog(ERROR, "index tuple too large");
HnswSetElementTuple(etup, element); HnswSetElementTuple(etup, element, useIndexTuple);
/* Keep element and neighbors on the same page if possible */ /* Keep element and neighbors on the same page if possible */
if (PageGetFreeSpace(page) < etupSize || (combinedSize <= maxSize && PageGetFreeSpace(page) < combinedSize)) if (PageGetFreeSpace(page) < etupSize || (combinedSize <= maxSize && PageGetFreeSpace(page) < combinedSize))
@@ -293,13 +274,14 @@ FlushPages(HnswBuildState * buildstate)
* Insert tuple * Insert tuple
*/ */
static bool static bool
InsertTuple(Relation index, Datum *values, HnswElement element, HnswBuildState * buildstate, HnswElement * dup, MemoryContext outerCtx) InsertTuple(Relation index, Datum *values, bool *isnull, HnswElement element, HnswBuildState * buildstate, HnswElement * dup, MemoryContext outerCtx)
{ {
FmgrInfo *procinfo = buildstate->procinfo; FmgrInfo **procinfos = buildstate->procinfos;
Oid collation = buildstate->collation; Oid *collations = buildstate->collations;
HnswElement entryPoint = buildstate->entryPoint; HnswElement entryPoint = buildstate->entryPoint;
int efConstruction = buildstate->efConstruction; int efConstruction = buildstate->efConstruction;
int m = buildstate->m; int m = buildstate->m;
bool inMemory = true;
MemoryContext oldCtx; MemoryContext oldCtx;
/* Detoast once for all calls */ /* Detoast once for all calls */
@@ -308,20 +290,20 @@ InsertTuple(Relation index, Datum *values, HnswElement element, HnswBuildState *
/* Normalize if needed */ /* Normalize if needed */
if (buildstate->normprocinfo != NULL) if (buildstate->normprocinfo != NULL)
{ {
if (!HnswNormValue(buildstate->normprocinfo, collation, &value, buildstate->normvec)) if (!HnswNormValue(buildstate->normprocinfo, collations[0], &value, buildstate->normvec))
return false; return false;
} }
/* Copy value to element so accessible outside of memory context */ /* Copy value to element so accessible outside of memory context */
oldCtx = MemoryContextSwitchTo(outerCtx); oldCtx = MemoryContextSwitchTo(outerCtx);
element->value = datumCopy(value, false, -1); HnswElementSetData(element, index, value, values, isnull);
MemoryContextSwitchTo(oldCtx); MemoryContextSwitchTo(oldCtx);
/* Insert element in graph */ /* Insert element in graph */
HnswInsertElement(element, entryPoint, NULL, procinfo, collation, m, efConstruction, false); HnswInsertElement(element, entryPoint, index, procinfos, collations, m, efConstruction, false, inMemory);
/* Look for duplicate */ /* Look for duplicate */
*dup = HnswFindDuplicate(element); *dup = HnswFindDuplicate(element, index);
/* Update neighbors if needed */ /* Update neighbors if needed */
if (*dup == NULL) if (*dup == NULL)
@@ -332,7 +314,7 @@ InsertTuple(Relation index, Datum *values, HnswElement element, HnswBuildState *
HnswNeighborArray *neighbors = &element->neighbors[lc]; HnswNeighborArray *neighbors = &element->neighbors[lc];
for (int i = 0; i < neighbors->length; i++) for (int i = 0; i < neighbors->length; i++)
HnswUpdateConnection(element, &neighbors->items[i], lm, lc, NULL, NULL, procinfo, collation); HnswUpdateConnection(element, &neighbors->items[i], lm, lc, NULL, index, procinfos, collations, inMemory);
} }
} }
@@ -356,7 +338,7 @@ HnswElementMemory(HnswElement e, int m)
elementSize += sizeof(HnswNeighborArray) * (e->level + 1); elementSize += sizeof(HnswNeighborArray) * (e->level + 1);
elementSize += sizeof(HnswCandidate) * (m * (e->level + 2)); elementSize += sizeof(HnswCandidate) * (m * (e->level + 2));
elementSize += sizeof(ItemPointerData); elementSize += sizeof(ItemPointerData);
elementSize += VARSIZE_ANY(DatumGetPointer(e->value)); elementSize += IndexTupleSize(e->itup);
return elementSize; return elementSize;
} }
@@ -396,18 +378,7 @@ BuildCallback(Relation index, CALLBACK_ITEM_POINTER, Datum *values,
oldCtx = MemoryContextSwitchTo(buildstate->tmpCtx); oldCtx = MemoryContextSwitchTo(buildstate->tmpCtx);
if (HnswInsertTuple(buildstate->index, values, isnull, tid, buildstate->heap)) if (HnswInsertTuple(buildstate->index, values, isnull, tid, buildstate->heap))
{ UpdateProgress(PROGRESS_CREATEIDX_TUPLES_DONE, ++buildstate->indtuples);
if (buildstate->hnswshared)
{
HnswShared *hnswshared = buildstate->hnswshared;
SpinLockAcquire(&hnswshared->mutex);
UpdateProgress(PROGRESS_CREATEIDX_TUPLES_DONE, ++hnswshared->indtuples);
SpinLockRelease(&hnswshared->mutex);
}
else
UpdateProgress(PROGRESS_CREATEIDX_TUPLES_DONE, ++buildstate->indtuples);
}
/* Reset memory context */ /* Reset memory context */
MemoryContextSwitchTo(oldCtx); MemoryContextSwitchTo(oldCtx);
@@ -423,7 +394,7 @@ BuildCallback(Relation index, CALLBACK_ITEM_POINTER, Datum *values,
oldCtx = MemoryContextSwitchTo(buildstate->tmpCtx); oldCtx = MemoryContextSwitchTo(buildstate->tmpCtx);
/* Insert tuple */ /* Insert tuple */
inserted = InsertTuple(index, values, element, buildstate, &dup, oldCtx); inserted = InsertTuple(index, values, isnull, element, buildstate, &dup, oldCtx);
/* Reset memory context */ /* Reset memory context */
MemoryContextSwitchTo(oldCtx); MemoryContextSwitchTo(oldCtx);
@@ -461,6 +432,19 @@ InitBuildState(HnswBuildState * buildstate, Relation heap, Relation index, Index
buildstate->efConstruction = HnswGetEfConstruction(index); buildstate->efConstruction = HnswGetEfConstruction(index);
buildstate->dimensions = TupleDescAttr(index->rd_att, 0)->atttypmod; buildstate->dimensions = TupleDescAttr(index->rd_att, 0)->atttypmod;
/* TODO See if needed */
if (IndexRelationGetNumberOfKeyAttributes(index) > 2)
elog(ERROR, "index cannot have more than two columns");
if (!OidIsValid(index_getprocid(index, 1, HNSW_DISTANCE_PROC)))
elog(ERROR, "first column must be a vector");
for (int i = 1; i < IndexRelationGetNumberOfKeyAttributes(index); i++)
{
if (!OidIsValid(index_getprocid(index, i + 1, HNSW_ATTRIBUTE_DISTANCE_PROC)))
elog(ERROR, "column %d cannot be a vector", i + 1);
}
/* Require column to have dimensions to be indexed */ /* Require column to have dimensions to be indexed */
if (buildstate->dimensions < 0) if (buildstate->dimensions < 0)
elog(ERROR, "column does not have dimensions"); elog(ERROR, "column does not have dimensions");
@@ -475,9 +459,9 @@ InitBuildState(HnswBuildState * buildstate, Relation heap, Relation index, Index
buildstate->indtuples = 0; buildstate->indtuples = 0;
/* Get support functions */ /* Get support functions */
buildstate->procinfo = index_getprocinfo(index, 1, HNSW_DISTANCE_PROC); buildstate->procinfos = HnswInitProcinfos(index);
buildstate->normprocinfo = HnswOptionalProcInfo(index, HNSW_NORM_PROC); buildstate->normprocinfo = HnswOptionalProcInfo(index, HNSW_NORM_PROC);
buildstate->collation = index->rd_indcollation[0]; buildstate->collations = index->rd_indcollation;
buildstate->elements = NIL; buildstate->elements = NIL;
buildstate->entryPoint = NULL; buildstate->entryPoint = NULL;
@@ -485,6 +469,7 @@ InitBuildState(HnswBuildState * buildstate, Relation heap, Relation index, Index
buildstate->maxLevel = HnswGetMaxLevel(buildstate->m); buildstate->maxLevel = HnswGetMaxLevel(buildstate->m);
buildstate->memoryLeft = maintenance_work_mem * 1024L; buildstate->memoryLeft = maintenance_work_mem * 1024L;
buildstate->flushed = false; buildstate->flushed = false;
buildstate->useIndexTuple = IndexRelationGetNumberOfAttributes(index) > 1;
/* Reuse for each tuple */ /* Reuse for each tuple */
buildstate->normvec = InitVector(buildstate->dimensions); buildstate->normvec = InitVector(buildstate->dimensions);
@@ -492,9 +477,6 @@ InitBuildState(HnswBuildState * buildstate, Relation heap, Relation index, Index
buildstate->tmpCtx = AllocSetContextCreate(CurrentMemoryContext, buildstate->tmpCtx = AllocSetContextCreate(CurrentMemoryContext,
"Hnsw build temporary context", "Hnsw build temporary context",
ALLOCSET_DEFAULT_SIZES); ALLOCSET_DEFAULT_SIZES);
buildstate->hnswleader = NULL;
buildstate->hnswshared = NULL;
} }
/* /*
@@ -503,378 +485,26 @@ InitBuildState(HnswBuildState * buildstate, Relation heap, Relation index, Index
static void static void
FreeBuildState(HnswBuildState * buildstate) FreeBuildState(HnswBuildState * buildstate)
{ {
pfree(buildstate->procinfos);
pfree(buildstate->normvec); pfree(buildstate->normvec);
MemoryContextDelete(buildstate->tmpCtx); MemoryContextDelete(buildstate->tmpCtx);
} }
/*
* Within leader, wait for end of heap scan
*/
static double
ParallelHeapScan(HnswBuildState * buildstate)
{
HnswShared *hnswshared = buildstate->hnswleader->hnswshared;
int nparticipanttuplesorts;
double reltuples;
nparticipanttuplesorts = buildstate->hnswleader->nparticipanttuplesorts;
for (;;)
{
SpinLockAcquire(&hnswshared->mutex);
if (hnswshared->nparticipantsdone == nparticipanttuplesorts)
{
buildstate->indtuples = hnswshared->indtuples;
reltuples = hnswshared->reltuples;
SpinLockRelease(&hnswshared->mutex);
break;
}
SpinLockRelease(&hnswshared->mutex);
ConditionVariableSleep(&hnswshared->workersdonecv,
WAIT_EVENT_PARALLEL_CREATE_INDEX_SCAN);
}
ConditionVariableCancelSleep();
return reltuples;
}
/*
* Perform a worker's portion of a parallel insert
*/
static void
HnswParallelScanAndInsert(HnswSpool * hnswspool, HnswShared * hnswshared, bool progress)
{
HnswBuildState buildstate;
#if PG_VERSION_NUM >= 120000
TableScanDesc scan;
#else
HeapScanDesc scan;
#endif
double reltuples;
IndexInfo *indexInfo;
/* Join parallel scan */
indexInfo = BuildIndexInfo(hnswspool->index);
indexInfo->ii_Concurrent = hnswshared->isconcurrent;
InitBuildState(&buildstate, hnswspool->heap, hnswspool->index, indexInfo, MAIN_FORKNUM);
/* TODO Support in-memory builds */
buildstate.memoryLeft = 0;
buildstate.flushed = true;
buildstate.hnswshared = hnswshared;
#if PG_VERSION_NUM >= 120000
scan = table_beginscan_parallel(hnswspool->heap,
ParallelTableScanFromHnswShared(hnswshared));
reltuples = table_index_build_scan(hnswspool->heap, hnswspool->index, indexInfo,
true, progress, BuildCallback,
(void *) &buildstate, scan);
#else
scan = heap_beginscan_parallel(hnswspool->heap, &hnswshared->heapdesc);
reltuples = IndexBuildHeapScan(hnswspool->heap, hnswspool->index, indexInfo,
true, BuildCallback,
(void *) &buildstate, scan);
#endif
/* Record statistics */
SpinLockAcquire(&hnswshared->mutex);
hnswshared->nparticipantsdone++;
hnswshared->reltuples += reltuples;
SpinLockRelease(&hnswshared->mutex);
/* Log statistics */
if (progress)
ereport(DEBUG1, (errmsg("leader processed " INT64_FORMAT " tuples", (int64) reltuples)));
else
ereport(DEBUG1, (errmsg("worker processed " INT64_FORMAT " tuples", (int64) reltuples)));
/* Notify leader */
ConditionVariableSignal(&hnswshared->workersdonecv);
FreeBuildState(&buildstate);
}
/*
* Perform work within a launched parallel process
*/
void
HnswParallelBuildMain(dsm_segment *seg, shm_toc *toc)
{
char *sharedquery;
HnswSpool *hnswspool;
HnswShared *hnswshared;
Relation heapRel;
Relation indexRel;
LOCKMODE heapLockmode;
LOCKMODE indexLockmode;
/* Set debug_query_string for individual workers first */
sharedquery = shm_toc_lookup(toc, PARALLEL_KEY_QUERY_TEXT, true);
debug_query_string = sharedquery;
/* Report the query string from leader */
pgstat_report_activity(STATE_RUNNING, debug_query_string);
/* Look up shared state */
hnswshared = shm_toc_lookup(toc, PARALLEL_KEY_HNSW_SHARED, false);
/* Open relations using lock modes known to be obtained by index.c */
if (!hnswshared->isconcurrent)
{
heapLockmode = ShareLock;
indexLockmode = AccessExclusiveLock;
}
else
{
heapLockmode = ShareUpdateExclusiveLock;
indexLockmode = RowExclusiveLock;
}
/* Open relations within worker */
#if PG_VERSION_NUM >= 120000
heapRel = table_open(hnswshared->heaprelid, heapLockmode);
#else
heapRel = heap_open(hnswshared->heaprelid, heapLockmode);
#endif
indexRel = index_open(hnswshared->indexrelid, indexLockmode);
/* Initialize worker's own spool */
hnswspool = (HnswSpool *) palloc0(sizeof(HnswSpool));
hnswspool->heap = heapRel;
hnswspool->index = indexRel;
/* Perform inserts */
HnswParallelScanAndInsert(hnswspool, hnswshared, false);
/* Close relations within worker */
index_close(indexRel, indexLockmode);
#if PG_VERSION_NUM >= 120000
table_close(heapRel, heapLockmode);
#else
heap_close(heapRel, heapLockmode);
#endif
}
/*
* End parallel build
*/
static void
HnswEndParallel(HnswLeader * hnswleader)
{
/* Shutdown worker processes */
WaitForParallelWorkersToFinish(hnswleader->pcxt);
/* Free last reference to MVCC snapshot, if one was used */
if (IsMVCCSnapshot(hnswleader->snapshot))
UnregisterSnapshot(hnswleader->snapshot);
DestroyParallelContext(hnswleader->pcxt);
ExitParallelMode();
}
/*
* Return size of shared memory required for parallel index build
*/
static Size
ParallelEstimateShared(Relation heap, Snapshot snapshot)
{
#if PG_VERSION_NUM >= 120000
return add_size(BUFFERALIGN(sizeof(HnswShared)), table_parallelscan_estimate(heap, snapshot));
#else
if (!IsMVCCSnapshot(snapshot))
{
Assert(snapshot == SnapshotAny);
return sizeof(HnswShared);
}
return add_size(offsetof(HnswShared, heapdesc) +
offsetof(ParallelHeapScanDescData, phs_snapshot_data),
EstimateSnapshotSpace(snapshot));
#endif
}
/*
* Within leader, participate as a parallel worker
*/
static void
HnswLeaderParticipateAsWorker(HnswBuildState * buildstate)
{
HnswLeader *hnswleader = buildstate->hnswleader;
HnswSpool *leaderworker;
/* Allocate memory and initialize private spool */
leaderworker = (HnswSpool *) palloc0(sizeof(HnswSpool));
leaderworker->heap = buildstate->heap;
leaderworker->index = buildstate->index;
/* Perform work common to all participants */
HnswParallelScanAndInsert(leaderworker, hnswleader->hnswshared, true);
}
/*
* Begin parallel build
*/
static void
HnswBeginParallel(HnswBuildState * buildstate, bool isconcurrent, int request)
{
ParallelContext *pcxt;
int scantuplesortstates;
Snapshot snapshot;
Size esthnswshared;
HnswShared *hnswshared;
HnswLeader *hnswleader = (HnswLeader *) palloc0(sizeof(HnswLeader));
bool leaderparticipates = true;
int querylen;
#ifdef DISABLE_LEADER_PARTICIPATION
leaderparticipates = false;
#endif
/* Enter parallel mode and create context */
EnterParallelMode();
Assert(request > 0);
#if PG_VERSION_NUM >= 120000
pcxt = CreateParallelContext("vector", "HnswParallelBuildMain", request);
#else
pcxt = CreateParallelContext("vector", "HnswParallelBuildMain", request, true);
#endif
scantuplesortstates = leaderparticipates ? request + 1 : request;
/* Get snapshot for table scan */
if (!isconcurrent)
snapshot = SnapshotAny;
else
snapshot = RegisterSnapshot(GetTransactionSnapshot());
/* Estimate size of workspaces */
esthnswshared = ParallelEstimateShared(buildstate->heap, snapshot);
shm_toc_estimate_chunk(&pcxt->estimator, esthnswshared);
shm_toc_estimate_keys(&pcxt->estimator, 1);
/* Finally, estimate PARALLEL_KEY_QUERY_TEXT space */
if (debug_query_string)
{
querylen = strlen(debug_query_string);
shm_toc_estimate_chunk(&pcxt->estimator, querylen + 1);
shm_toc_estimate_keys(&pcxt->estimator, 1);
}
else
querylen = 0; /* keep compiler quiet */
/* Everyone's had a chance to ask for space, so now create the DSM */
InitializeParallelDSM(pcxt);
/* If no DSM segment was available, back out (do serial build) */
if (pcxt->seg == NULL)
{
if (IsMVCCSnapshot(snapshot))
UnregisterSnapshot(snapshot);
DestroyParallelContext(pcxt);
ExitParallelMode();
return;
}
/* Store shared build state, for which we reserved space */
hnswshared = (HnswShared *) shm_toc_allocate(pcxt->toc, esthnswshared);
/* Initialize immutable state */
hnswshared->heaprelid = RelationGetRelid(buildstate->heap);
hnswshared->indexrelid = RelationGetRelid(buildstate->index);
hnswshared->isconcurrent = isconcurrent;
hnswshared->scantuplesortstates = scantuplesortstates;
ConditionVariableInit(&hnswshared->workersdonecv);
SpinLockInit(&hnswshared->mutex);
/* Initialize mutable state */
hnswshared->nparticipantsdone = 0;
hnswshared->reltuples = 0;
hnswshared->indtuples = 0;
#if PG_VERSION_NUM >= 120000
table_parallelscan_initialize(buildstate->heap,
ParallelTableScanFromHnswShared(hnswshared),
snapshot);
#else
heap_parallelscan_initialize(&hnswshared->heapdesc, buildstate->heap, snapshot);
#endif
shm_toc_insert(pcxt->toc, PARALLEL_KEY_HNSW_SHARED, hnswshared);
/* Store query string for workers */
if (debug_query_string)
{
char *sharedquery;
sharedquery = (char *) shm_toc_allocate(pcxt->toc, querylen + 1);
memcpy(sharedquery, debug_query_string, querylen + 1);
shm_toc_insert(pcxt->toc, PARALLEL_KEY_QUERY_TEXT, sharedquery);
}
/* Launch workers, saving status for leader/caller */
LaunchParallelWorkers(pcxt);
hnswleader->pcxt = pcxt;
hnswleader->nparticipanttuplesorts = pcxt->nworkers_launched;
if (leaderparticipates)
hnswleader->nparticipanttuplesorts++;
hnswleader->hnswshared = hnswshared;
hnswleader->snapshot = snapshot;
/* If no workers were successfully launched, back out (do serial build) */
if (pcxt->nworkers_launched == 0)
{
HnswEndParallel(hnswleader);
return;
}
/* Log participants */
ereport(DEBUG1, (errmsg("using %d parallel workers", pcxt->nworkers_launched)));
/* Save leader state now that it's clear build will be parallel */
buildstate->hnswleader = hnswleader;
/* Join heap scan ourselves */
if (leaderparticipates)
HnswLeaderParticipateAsWorker(buildstate);
/* Wait for all launched workers */
WaitForParallelWorkersToAttach(pcxt);
}
/* /*
* Build graph * Build graph
*/ */
static void static void
BuildGraph(HnswBuildState * buildstate, ForkNumber forkNum) BuildGraph(HnswBuildState * buildstate, ForkNumber forkNum)
{ {
int parallel_workers = 0;
UpdateProgress(PROGRESS_CREATEIDX_SUBPHASE, PROGRESS_HNSW_PHASE_LOAD); UpdateProgress(PROGRESS_CREATEIDX_SUBPHASE, PROGRESS_HNSW_PHASE_LOAD);
/* Calculate parallel workers */
if (hnsw_enable_parallel_build)
parallel_workers = plan_create_index_workers(RelationGetRelid(buildstate->heap), RelationGetRelid(buildstate->index));
/* Attempt to launch parallel worker scan when required */
if (parallel_workers > 0)
{
/* TODO Support in-memory builds */
FlushPages(buildstate);
HnswBeginParallel(buildstate, buildstate->indexInfo->ii_Concurrent, parallel_workers);
}
/* Add tuples to sort */
if (buildstate->hnswleader)
buildstate->reltuples = ParallelHeapScan(buildstate);
else
{
#if PG_VERSION_NUM >= 120000 #if PG_VERSION_NUM >= 120000
buildstate->reltuples = table_index_build_scan(buildstate->heap, buildstate->index, buildstate->indexInfo, buildstate->reltuples = table_index_build_scan(buildstate->heap, buildstate->index, buildstate->indexInfo,
true, true, BuildCallback, (void *) buildstate, NULL); true, true, BuildCallback, (void *) buildstate, NULL);
#else #else
buildstate->reltuples = IndexBuildHeapScan(buildstate->heap, buildstate->index, buildstate->indexInfo, buildstate->reltuples = IndexBuildHeapScan(buildstate->heap, buildstate->index, buildstate->indexInfo,
true, BuildCallback, (void *) buildstate, NULL); true, BuildCallback, (void *) buildstate, NULL);
#endif #endif
}
/* End parallel build */
if (buildstate->hnswleader)
HnswEndParallel(buildstate->hnswleader);
} }
/* /*

View File

@@ -129,9 +129,10 @@ WriteNewElementPages(Relation index, HnswElement e, int m, BlockNumber insertPag
OffsetNumber freeOffno = InvalidOffsetNumber; OffsetNumber freeOffno = InvalidOffsetNumber;
OffsetNumber freeNeighborOffno = InvalidOffsetNumber; OffsetNumber freeNeighborOffno = InvalidOffsetNumber;
BlockNumber newInsertPage = InvalidBlockNumber; BlockNumber newInsertPage = InvalidBlockNumber;
bool useIndexTuple = IndexRelationGetNumberOfAttributes(index) > 1;
/* Calculate sizes */ /* Calculate sizes */
etupSize = HNSW_ELEMENT_TUPLE_SIZE(VARSIZE_ANY(DatumGetPointer(e->value))); etupSize = HNSW_ELEMENT_TUPLE_SIZE(useIndexTuple ? IndexTupleSize(e->itup) : VARSIZE_ANY(DatumGetPointer(e->value)));
ntupSize = HNSW_NEIGHBOR_TUPLE_SIZE(e->level, m); ntupSize = HNSW_NEIGHBOR_TUPLE_SIZE(e->level, m);
combinedSize = etupSize + ntupSize + sizeof(ItemIdData); combinedSize = etupSize + ntupSize + sizeof(ItemIdData);
maxSize = HNSW_MAX_SIZE; maxSize = HNSW_MAX_SIZE;
@@ -139,7 +140,7 @@ WriteNewElementPages(Relation index, HnswElement e, int m, BlockNumber insertPag
/* Prepare element tuple */ /* Prepare element tuple */
etup = palloc0(etupSize); etup = palloc0(etupSize);
HnswSetElementTuple(etup, e); HnswSetElementTuple(etup, e, useIndexTuple);
/* Prepare neighbor tuple */ /* Prepare neighbor tuple */
ntup = palloc0(ntupSize); ntup = palloc0(ntupSize);
@@ -301,7 +302,7 @@ ConnectionExists(HnswElement e, HnswNeighborTuple ntup, int startIdx, int lm)
* Update neighbors * Update neighbors
*/ */
void void
HnswUpdateNeighborPages(Relation index, FmgrInfo *procinfo, Oid collation, HnswElement e, int m, bool checkExisting) HnswUpdateNeighborPages(Relation index, FmgrInfo **procinfos, Oid *collations, HnswElement e, int m, bool checkExisting)
{ {
for (int lc = e->level; lc >= 0; lc--) for (int lc = e->level; lc >= 0; lc--)
{ {
@@ -333,7 +334,7 @@ HnswUpdateNeighborPages(Relation index, FmgrInfo *procinfo, Oid collation, HnswE
*/ */
/* Select neighbors */ /* Select neighbors */
HnswUpdateConnection(e, hc, lm, lc, &idx, index, procinfo, collation); HnswUpdateConnection(e, hc, lm, lc, &idx, index, procinfos, collations, false);
/* New element was not selected as a neighbor */ /* New element was not selected as a neighbor */
if (idx == -1) if (idx == -1)
@@ -451,7 +452,7 @@ HnswAddDuplicate(Relation index, HnswElement element, HnswElement dup)
* Write changes to disk * Write changes to disk
*/ */
static void static void
WriteElement(Relation index, FmgrInfo *procinfo, Oid collation, HnswElement element, int m, int efConstruction, HnswElement dup, HnswElement entryPoint) WriteElement(Relation index, FmgrInfo **procinfos, Oid *collations, HnswElement element, int m, int efConstruction, HnswElement dup, HnswElement entryPoint)
{ {
BlockNumber newInsertPage = InvalidBlockNumber; BlockNumber newInsertPage = InvalidBlockNumber;
@@ -470,7 +471,7 @@ WriteElement(Relation index, FmgrInfo *procinfo, Oid collation, HnswElement elem
HnswUpdateMetaPage(index, 0, NULL, newInsertPage, MAIN_FORKNUM); HnswUpdateMetaPage(index, 0, NULL, newInsertPage, MAIN_FORKNUM);
/* Update neighbors */ /* Update neighbors */
HnswUpdateNeighborPages(index, procinfo, collation, element, m, false); HnswUpdateNeighborPages(index, procinfos, collations, element, m, false);
/* Update metapage if needed */ /* Update metapage if needed */
if (entryPoint == NULL || element->level > entryPoint->level) if (entryPoint == NULL || element->level > entryPoint->level)
@@ -489,8 +490,8 @@ HnswInsertTuple(Relation index, Datum *values, bool *isnull, ItemPointer heap_ti
HnswElement element; HnswElement element;
int m; int m;
int efConstruction = HnswGetEfConstruction(index); int efConstruction = HnswGetEfConstruction(index);
FmgrInfo *procinfo = index_getprocinfo(index, 1, HNSW_DISTANCE_PROC); FmgrInfo **procinfos = HnswInitProcinfos(index);
Oid collation = index->rd_indcollation[0]; Oid *collations = index->rd_indcollation;
HnswElement dup; HnswElement dup;
LOCKMODE lockmode = ShareLock; LOCKMODE lockmode = ShareLock;
@@ -501,7 +502,7 @@ HnswInsertTuple(Relation index, Datum *values, bool *isnull, ItemPointer heap_ti
normprocinfo = HnswOptionalProcInfo(index, HNSW_NORM_PROC); normprocinfo = HnswOptionalProcInfo(index, HNSW_NORM_PROC);
if (normprocinfo != NULL) if (normprocinfo != NULL)
{ {
if (!HnswNormValue(normprocinfo, collation, &value, NULL)) if (!HnswNormValue(normprocinfo, collations[0], &value, NULL))
return false; return false;
} }
@@ -517,7 +518,7 @@ HnswInsertTuple(Relation index, Datum *values, bool *isnull, ItemPointer heap_ti
/* Create an element */ /* Create an element */
element = HnswInitElement(heap_tid, m, HnswGetMl(m), HnswGetMaxLevel(m)); element = HnswInitElement(heap_tid, m, HnswGetMl(m), HnswGetMaxLevel(m));
element->value = value; HnswElementSetData(element, index, value, values, isnull);
/* Prevent concurrent inserts when likely updating entry point */ /* Prevent concurrent inserts when likely updating entry point */
if (entryPoint == NULL || element->level > entryPoint->level) if (entryPoint == NULL || element->level > entryPoint->level)
@@ -534,13 +535,13 @@ HnswInsertTuple(Relation index, Datum *values, bool *isnull, ItemPointer heap_ti
} }
/* Insert element in graph */ /* Insert element in graph */
HnswInsertElement(element, entryPoint, index, procinfo, collation, m, efConstruction, false); HnswInsertElement(element, entryPoint, index, procinfos, collations, m, efConstruction, false, false);
/* Look for duplicate */ /* Look for duplicate */
dup = HnswFindDuplicate(element); dup = HnswFindDuplicate(element, index);
/* Write to disk */ /* Write to disk */
WriteElement(index, procinfo, collation, element, m, efConstruction, dup, entryPoint); WriteElement(index, procinfos, collations, element, m, efConstruction, dup, entryPoint);
/* Release lock */ /* Release lock */
UnlockPage(index, HNSW_UPDATE_LOCK, lockmode); UnlockPage(index, HNSW_UPDATE_LOCK, lockmode);

View File

@@ -15,12 +15,13 @@ GetScanItems(IndexScanDesc scan, Datum q)
{ {
HnswScanOpaque so = (HnswScanOpaque) scan->opaque; HnswScanOpaque so = (HnswScanOpaque) scan->opaque;
Relation index = scan->indexRelation; Relation index = scan->indexRelation;
FmgrInfo *procinfo = so->procinfo; FmgrInfo **procinfos = so->procinfos;
Oid collation = so->collation; Oid *collations = so->collations;
List *ep; List *ep;
List *w; List *w;
int m; int m;
HnswElement entryPoint; HnswElement entryPoint;
ScanKeyData *keyData = scan->keyData;
/* Get m and entry point */ /* Get m and entry point */
HnswGetMetaPageInfo(index, &m, &entryPoint); HnswGetMetaPageInfo(index, &m, &entryPoint);
@@ -28,15 +29,15 @@ GetScanItems(IndexScanDesc scan, Datum q)
if (entryPoint == NULL) if (entryPoint == NULL)
return NIL; return NIL;
ep = list_make1(HnswEntryCandidate(entryPoint, q, index, procinfo, collation, false)); ep = list_make1(HnswEntryCandidate(entryPoint, q, NULL, keyData, index, procinfos, collations, false, false));
for (int lc = entryPoint->level; lc >= 1; lc--) for (int lc = entryPoint->level; lc >= 1; lc--)
{ {
w = HnswSearchLayer(q, ep, 1, lc, index, procinfo, collation, m, false, NULL); w = HnswSearchLayer(q, NULL, keyData, ep, 1, lc, index, procinfos, collations, m, false, NULL, false);
ep = w; ep = w;
} }
return HnswSearchLayer(q, ep, hnsw_ef_search, 0, index, procinfo, collation, m, false, NULL); return HnswSearchLayer(q, NULL, keyData, ep, hnsw_ef_search, 0, index, procinfos, collations, m, false, NULL, false);
} }
/* /*
@@ -83,7 +84,7 @@ GetScanValue(IndexScanDesc scan)
/* Fine if normalization fails */ /* Fine if normalization fails */
if (so->normprocinfo != NULL) if (so->normprocinfo != NULL)
HnswNormValue(so->normprocinfo, so->collation, &value, NULL); HnswNormValue(so->normprocinfo, so->collations[0], &value, NULL);
} }
return value; return value;
@@ -107,9 +108,9 @@ hnswbeginscan(Relation index, int nkeys, int norderbys)
ALLOCSET_DEFAULT_SIZES); ALLOCSET_DEFAULT_SIZES);
/* Set support functions */ /* Set support functions */
so->procinfo = index_getprocinfo(index, 1, HNSW_DISTANCE_PROC); so->procinfos = HnswInitProcinfos(index);
so->normprocinfo = HnswOptionalProcInfo(index, HNSW_NORM_PROC); so->normprocinfo = HnswOptionalProcInfo(index, HNSW_NORM_PROC);
so->collation = index->rd_indcollation[0]; so->collations = index->rd_indcollation;
scan->opaque = so; scan->opaque = so;
@@ -206,6 +207,9 @@ hnswgettuple(IndexScanDesc scan, ScanDirection dir)
scan->xs_ctup.t_self = *heaptid; scan->xs_ctup.t_self = *heaptid;
#endif #endif
/* TODO Check during scan */
scan->xs_recheck = scan->numberOfKeys > 0;
scan->xs_recheckorderby = false; scan->xs_recheckorderby = false;
return true; return true;
} }
@@ -222,6 +226,7 @@ hnswendscan(IndexScanDesc scan)
{ {
HnswScanOpaque so = (HnswScanOpaque) scan->opaque; HnswScanOpaque so = (HnswScanOpaque) scan->opaque;
pfree(so->procinfos);
MemoryContextDelete(so->tmpCtx); MemoryContextDelete(so->tmpCtx);
pfree(so); pfree(so);

View File

@@ -7,6 +7,10 @@
#include "utils/datum.h" #include "utils/datum.h"
#include "vector.h" #include "vector.h"
#if PG_VERSION_NUM < 130000
#define TYPSTORAGE_PLAIN 'p'
#endif
/* /*
* Get the max number of connections in an upper layer for each element in the index * Get the max number of connections in an upper layer for each element in the index
*/ */
@@ -47,6 +51,22 @@ HnswOptionalProcInfo(Relation index, uint16 procnum)
return index_getprocinfo(index, 1, procnum); return index_getprocinfo(index, 1, procnum);
} }
/*
* Init procs
*/
FmgrInfo **
HnswInitProcinfos(Relation index)
{
int keyAttributes = IndexRelationGetNumberOfKeyAttributes(index);
FmgrInfo **procinfos = palloc(keyAttributes * sizeof(FmgrInfo *));
procinfos[0] = index_getprocinfo(index, 1, HNSW_DISTANCE_PROC);
for (int i = 1; i < keyAttributes; i++)
procinfos[i] = index_getprocinfo(index, i + 1, HNSW_ATTRIBUTE_DISTANCE_PROC);
return procinfos;
}
/* /*
* Divide by the norm * Divide by the norm
* *
@@ -174,6 +194,7 @@ HnswInitElement(ItemPointer heaptid, int m, double ml, int maxLevel)
element->level = level; element->level = level;
element->deleted = 0; element->deleted = 0;
element->itup = NULL;
HnswInitNeighbors(element, m); HnswInitNeighbors(element, m);
@@ -188,8 +209,8 @@ HnswFreeElement(HnswElement element)
{ {
HnswFreeNeighbors(element); HnswFreeNeighbors(element);
list_free_deep(element->heaptids); list_free_deep(element->heaptids);
if (DatumGetPointer(element->value)) if (element->itup)
pfree(DatumGetPointer(element->value)); pfree(element->itup);
pfree(element); pfree(element);
} }
@@ -217,6 +238,7 @@ HnswInitElementFromBlock(BlockNumber blkno, OffsetNumber offno)
element->offno = offno; element->offno = offno;
element->neighbors = NULL; element->neighbors = NULL;
element->value = PointerGetDatum(NULL); element->value = PointerGetDatum(NULL);
element->itup = NULL;
return element; return element;
} }
@@ -314,7 +336,7 @@ HnswUpdateMetaPage(Relation index, int updateEntry, HnswElement entryPoint, Bloc
* Set element tuple, except for neighbor info * Set element tuple, except for neighbor info
*/ */
void void
HnswSetElementTuple(HnswElementTuple etup, HnswElement element) HnswSetElementTuple(HnswElementTuple etup, HnswElement element, bool useIndexTuple)
{ {
etup->type = HNSW_ELEMENT_TUPLE_TYPE; etup->type = HNSW_ELEMENT_TUPLE_TYPE;
etup->level = element->level; etup->level = element->level;
@@ -326,7 +348,11 @@ HnswSetElementTuple(HnswElementTuple etup, HnswElement element)
else else
ItemPointerSetInvalid(&etup->heaptids[i]); ItemPointerSetInvalid(&etup->heaptids[i]);
} }
memcpy(&etup->data, DatumGetPointer(element->value), VARSIZE_ANY(DatumGetPointer(element->value)));
if (useIndexTuple)
memcpy(&etup->data, element->itup, IndexTupleSize(element->itup));
else
memcpy(&etup->data, DatumGetPointer(element->value), VARSIZE_ANY(DatumGetPointer(element->value)));
} }
/* /*
@@ -427,7 +453,7 @@ HnswLoadNeighbors(HnswElement element, Relation index, int m)
* Load an element from a tuple * Load an element from a tuple
*/ */
void void
HnswLoadElementFromTuple(HnswElement element, HnswElementTuple etup, bool loadHeaptids, bool loadVec) HnswLoadElementFromTuple(HnswElement element, HnswElementTuple etup, bool loadHeaptids, bool loadVec, Relation index)
{ {
element->level = etup->level; element->level = etup->level;
element->deleted = etup->deleted; element->deleted = etup->deleted;
@@ -448,14 +474,150 @@ HnswLoadElementFromTuple(HnswElement element, HnswElementTuple etup, bool loadHe
} }
if (loadVec) if (loadVec)
element->value = datumCopy(PointerGetDatum(&etup->data), false, -1); {
if (IndexRelationGetNumberOfAttributes(index) > 1)
{
TupleDesc tupdesc = RelationGetDescr(index);
bool unused;
element->itup = CopyIndexTuple((IndexTuple) &etup->data);
element->value = index_getattr(element->itup, 1, tupdesc, &unused);
}
else
{
Vector *vec = palloc(VARSIZE_ANY(&etup->data));
memcpy(vec, &etup->data, VARSIZE_ANY(&etup->data));
element->value = PointerGetDatum(vec);
}
}
}
/*
* Get the tuple descriptor
*/
static TupleDesc
HnswTupleDesc(Relation index)
{
TupleDesc tupdesc = CreateTupleDescCopyConstr(RelationGetDescr(index));
/* Prevent compression */
TupleDescAttr(tupdesc, 0)->attstorage = TYPSTORAGE_PLAIN;
return tupdesc;
}
/*
* Set element data
*/
void
HnswElementSetData(HnswElement element, Relation index, Datum value, Datum *values, bool *isnull)
{
/* TODO Create once per index build */
TupleDesc tupdesc = HnswTupleDesc(index);
bool unused;
Datum tmp;
tmp = values[0];
values[0] = value;
element->itup = index_form_tuple(tupdesc, values, isnull);
values[0] = tmp;
element->value = index_getattr(element->itup, 1, tupdesc, &unused);
FreeTupleDesc(tupdesc);
}
/*
* Get the attribute distance
*/
static inline double
AttributeDistance(double e)
{
/* TODO Better bias */
/* must be >> max(w * g) + 1 / log10(2) */
double bias = 4.32;
return e > 0 ? bias - 1.0 / log10(e + 1) : 0;
}
/*
* Get the distance
*/
static double
GetDistance(IndexTuple itup, Datum vec, Datum q, IndexTuple qtup, ScanKeyData *keyData, Relation index, FmgrInfo **procinfos, Oid *collations)
{
double g = DatumGetFloat8(FunctionCall2Coll(procinfos[0], collations[0], q, vec));
if (IndexRelationGetNumberOfKeyAttributes(index) > 1)
{
double w = 0.25;
double e = 0.0;
TupleDesc tupdesc = RelationGetDescr(index);
if (keyData)
{
/* TODO need to pass length of key data */
int keyCount = 1;
for (int i = 0; i < keyCount; i++)
{
ScanKey key = &keyData[i];
bool isnull;
Datum value = index_getattr(itup, key->sk_attno, tupdesc, &isnull);
bool attnull = key->sk_flags & SK_ISNULL;
if (isnull || attnull)
{
if (isnull != attnull)
e += 1000;
}
else if (!DatumGetBool(FunctionCall2Coll(&key->sk_func, key->sk_collation, value, key->sk_argument)))
{
double ei = fabs(DatumGetFloat8(FunctionCall2Coll(procinfos[key->sk_attno - 1], collations[key->sk_attno - 1], value, key->sk_argument)));
if (ei > 0)
e += ei;
else
/* Distance is zero for inequality */
e += 1000;
}
}
return w * g + AttributeDistance(e);
}
else if (qtup)
{
int keyCount = IndexRelationGetNumberOfKeyAttributes(index) - 1;
for (int i = 0; i < keyCount; i++)
{
bool isnull;
bool attnull;
Datum value = index_getattr(itup, i + 2, tupdesc, &isnull);
Datum value2 = index_getattr(qtup, i + 2, tupdesc, &attnull);
if (isnull || attnull)
{
if (isnull != attnull)
e += 1000;
}
else
e += fabs(DatumGetFloat8(FunctionCall2Coll(procinfos[i + 1], collations[i + 1], value, value2)));
}
return w * g + AttributeDistance(e);
}
}
return g;
} }
/* /*
* Load an element and optionally get its distance from q * Load an element and optionally get its distance from q
*/ */
void void
HnswLoadElement(HnswElement element, float *distance, Datum *q, Relation index, FmgrInfo *procinfo, Oid collation, bool loadVec) HnswLoadElement(HnswElement element, float *distance, Datum *q, IndexTuple qtup, ScanKeyData *keyData, Relation index, FmgrInfo **procinfos, Oid *collations, bool loadVec)
{ {
Buffer buf; Buffer buf;
Page page; Page page;
@@ -471,11 +633,27 @@ HnswLoadElement(HnswElement element, float *distance, Datum *q, Relation index,
Assert(HnswIsElementTuple(etup)); Assert(HnswIsElementTuple(etup));
/* Load element */ /* Load element */
HnswLoadElementFromTuple(element, etup, true, loadVec); HnswLoadElementFromTuple(element, etup, true, loadVec, index);
/* Calculate distance */ /* Calculate distance */
if (distance != NULL) if (distance != NULL)
*distance = (float) DatumGetFloat8(FunctionCall2Coll(procinfo, collation, *q, PointerGetDatum(&etup->data))); {
IndexTuple itup = NULL;
Datum value;
if (IndexRelationGetNumberOfAttributes(index) > 1)
{
TupleDesc tupdesc = RelationGetDescr(index);
bool unused;
itup = (IndexTuple) &etup->data;
value = index_getattr(itup, 1, tupdesc, &unused);
}
else
value = PointerGetDatum(&etup->data);
*distance = GetDistance(itup, value, *q, qtup, keyData, index, procinfos, collations);
}
UnlockReleaseBuffer(buf); UnlockReleaseBuffer(buf);
} }
@@ -484,24 +662,24 @@ HnswLoadElement(HnswElement element, float *distance, Datum *q, Relation index,
* Get the distance for a candidate * Get the distance for a candidate
*/ */
static float static float
GetCandidateDistance(HnswCandidate * hc, Datum q, FmgrInfo *procinfo, Oid collation) GetCandidateDistance(HnswCandidate * hc, Datum q, IndexTuple qtup, ScanKeyData *keyData, Relation index, FmgrInfo **procinfos, Oid *collations)
{ {
return DatumGetFloat8(FunctionCall2Coll(procinfo, collation, q, hc->element->value)); return GetDistance(hc->element->itup, hc->element->value, q, qtup, keyData, index, procinfos, collations);
} }
/* /*
* Create a candidate for the entry point * Create a candidate for the entry point
*/ */
HnswCandidate * HnswCandidate *
HnswEntryCandidate(HnswElement entryPoint, Datum q, Relation index, FmgrInfo *procinfo, Oid collation, bool loadVec) HnswEntryCandidate(HnswElement entryPoint, Datum q, IndexTuple qtup, ScanKeyData *keyData, Relation index, FmgrInfo **procinfos, Oid *collations, bool loadVec, bool inMemory)
{ {
HnswCandidate *hc = palloc(sizeof(HnswCandidate)); HnswCandidate *hc = palloc(sizeof(HnswCandidate));
hc->element = entryPoint; hc->element = entryPoint;
if (index == NULL) if (inMemory)
hc->distance = GetCandidateDistance(hc, q, procinfo, collation); hc->distance = GetCandidateDistance(hc, q, qtup, keyData, index, procinfos, collations);
else else
HnswLoadElement(hc->element, &hc->distance, &q, index, procinfo, collation, loadVec); HnswLoadElement(hc->element, &hc->distance, &q, qtup, keyData, index, procinfos, collations, loadVec);
return hc; return hc;
} }
@@ -551,9 +729,9 @@ CreatePairingHeapNode(HnswCandidate * c)
* Add to visited * Add to visited
*/ */
static inline void static inline void
AddToVisited(HTAB *v, HnswCandidate * hc, Relation index, bool *found) AddToVisited(HTAB *v, HnswCandidate * hc, bool inMemory, bool *found)
{ {
if (index == NULL) if (inMemory)
hash_search(v, &hc->element, HASH_ENTER, found); hash_search(v, &hc->element, HASH_ENTER, found);
else else
{ {
@@ -568,7 +746,7 @@ AddToVisited(HTAB *v, HnswCandidate * hc, Relation index, bool *found)
* Algorithm 2 from paper * Algorithm 2 from paper
*/ */
List * List *
HnswSearchLayer(Datum q, List *ep, int ef, int lc, Relation index, FmgrInfo *procinfo, Oid collation, int m, bool inserting, HnswElement skipElement) HnswSearchLayer(Datum q, IndexTuple qtup, ScanKeyData *keyData, List *ep, int ef, int lc, Relation index, FmgrInfo **procinfos, Oid *collations, int m, bool loadVec, HnswElement skipElement, bool inMemory)
{ {
ListCell *lc2; ListCell *lc2;
@@ -580,7 +758,7 @@ HnswSearchLayer(Datum q, List *ep, int ef, int lc, Relation index, FmgrInfo *pro
HTAB *v; HTAB *v;
/* Create hash table */ /* Create hash table */
if (index == NULL) if (inMemory)
{ {
hash_ctl.keysize = sizeof(HnswElement *); hash_ctl.keysize = sizeof(HnswElement *);
hash_ctl.entrysize = sizeof(HnswElement *); hash_ctl.entrysize = sizeof(HnswElement *);
@@ -599,7 +777,7 @@ HnswSearchLayer(Datum q, List *ep, int ef, int lc, Relation index, FmgrInfo *pro
{ {
HnswCandidate *hc = (HnswCandidate *) lfirst(lc2); HnswCandidate *hc = (HnswCandidate *) lfirst(lc2);
AddToVisited(v, hc, index, NULL); AddToVisited(v, hc, inMemory, NULL);
pairingheap_add(C, &(CreatePairingHeapNode(hc)->ph_node)); pairingheap_add(C, &(CreatePairingHeapNode(hc)->ph_node));
pairingheap_add(W, &(CreatePairingHeapNode(hc)->ph_node)); pairingheap_add(W, &(CreatePairingHeapNode(hc)->ph_node));
@@ -633,7 +811,7 @@ HnswSearchLayer(Datum q, List *ep, int ef, int lc, Relation index, FmgrInfo *pro
HnswCandidate *e = &neighborhood->items[i]; HnswCandidate *e = &neighborhood->items[i];
bool visited; bool visited;
AddToVisited(v, e, index, &visited); AddToVisited(v, e, inMemory, &visited);
if (!visited) if (!visited)
{ {
@@ -641,10 +819,10 @@ HnswSearchLayer(Datum q, List *ep, int ef, int lc, Relation index, FmgrInfo *pro
f = ((HnswPairingHeapNode *) pairingheap_first(W))->inner; f = ((HnswPairingHeapNode *) pairingheap_first(W))->inner;
if (index == NULL) if (inMemory)
eDistance = GetCandidateDistance(e, q, procinfo, collation); eDistance = GetCandidateDistance(e, q, qtup, keyData, index, procinfos, collations);
else else
HnswLoadElement(e->element, &eDistance, &q, index, procinfo, collation, inserting); HnswLoadElement(e->element, &eDistance, &q, qtup, keyData, index, procinfos, collations, loadVec);
Assert(!e->element->deleted); Assert(!e->element->deleted);
@@ -724,7 +902,7 @@ CompareCandidateDistances(const void *a, const void *b)
* Calculate the distance between elements * Calculate the distance between elements
*/ */
static float static float
HnswGetDistance(HnswElement a, HnswElement b, int lc, FmgrInfo *procinfo, Oid collation) HnswGetCachedDistance(HnswElement a, HnswElement b, int lc, Relation index, FmgrInfo **procinfos, Oid *collations)
{ {
/* Look for cached distance */ /* Look for cached distance */
if (a->neighbors != NULL) if (a->neighbors != NULL)
@@ -749,21 +927,21 @@ HnswGetDistance(HnswElement a, HnswElement b, int lc, FmgrInfo *procinfo, Oid co
} }
} }
return DatumGetFloat8(FunctionCall2Coll(procinfo, collation, a->value, b->value)); return GetDistance(a->itup, a->value, b->value, b->itup, NULL, index, procinfos, collations);
} }
/* /*
* Check if an element is closer to q than any element from R * Check if an element is closer to q than any element from R
*/ */
static bool static bool
CheckElementCloser(HnswCandidate * e, List *r, int lc, FmgrInfo *procinfo, Oid collation) CheckElementCloser(HnswCandidate * e, List *r, int lc, Relation index, FmgrInfo **procinfos, Oid *collations)
{ {
ListCell *lc2; ListCell *lc2;
foreach(lc2, r) foreach(lc2, r)
{ {
HnswCandidate *ri = lfirst(lc2); HnswCandidate *ri = lfirst(lc2);
float distance = HnswGetDistance(e->element, ri->element, lc, procinfo, collation); float distance = HnswGetCachedDistance(e->element, ri->element, lc, index, procinfos, collations);
if (distance <= e->distance) if (distance <= e->distance)
return false; return false;
@@ -776,7 +954,7 @@ CheckElementCloser(HnswCandidate * e, List *r, int lc, FmgrInfo *procinfo, Oid c
* Algorithm 4 from paper * Algorithm 4 from paper
*/ */
static List * static List *
SelectNeighbors(List *c, int m, int lc, FmgrInfo *procinfo, Oid collation, HnswElement e2, HnswCandidate * newCandidate, HnswCandidate * *pruned, bool sortCandidates) SelectNeighbors(List *c, int m, int lc, Relation index, FmgrInfo **procinfos, Oid *collations, HnswElement e2, HnswCandidate * newCandidate, HnswCandidate * *pruned, bool sortCandidates)
{ {
List *r = NIL; List *r = NIL;
List *w = list_copy(c); List *w = list_copy(c);
@@ -803,7 +981,7 @@ SelectNeighbors(List *c, int m, int lc, FmgrInfo *procinfo, Oid collation, HnswE
/* Use previous state of r and wd to skip work when possible */ /* Use previous state of r and wd to skip work when possible */
if (mustCalculate) if (mustCalculate)
e->closer = CheckElementCloser(e, r, lc, procinfo, collation); e->closer = CheckElementCloser(e, r, lc, index, procinfos, collations);
else if (list_length(added) > 0) else if (list_length(added) > 0)
{ {
/* /*
@@ -812,7 +990,7 @@ SelectNeighbors(List *c, int m, int lc, FmgrInfo *procinfo, Oid collation, HnswE
*/ */
if (e->closer) if (e->closer)
{ {
e->closer = CheckElementCloser(e, added, lc, procinfo, collation); e->closer = CheckElementCloser(e, added, lc, index, procinfos, collations);
if (!e->closer) if (!e->closer)
removedAny = true; removedAny = true;
@@ -825,7 +1003,7 @@ SelectNeighbors(List *c, int m, int lc, FmgrInfo *procinfo, Oid collation, HnswE
*/ */
if (removedAny) if (removedAny)
{ {
e->closer = CheckElementCloser(e, r, lc, procinfo, collation); e->closer = CheckElementCloser(e, r, lc, index, procinfos, collations);
if (e->closer) if (e->closer)
added = lappend(added, e); added = lappend(added, e);
} }
@@ -833,7 +1011,7 @@ SelectNeighbors(List *c, int m, int lc, FmgrInfo *procinfo, Oid collation, HnswE
} }
else if (e == newCandidate) else if (e == newCandidate)
{ {
e->closer = CheckElementCloser(e, r, lc, procinfo, collation); e->closer = CheckElementCloser(e, r, lc, index, procinfos, collations);
if (e->closer) if (e->closer)
added = lappend(added, e); added = lappend(added, e);
} }
@@ -867,10 +1045,14 @@ SelectNeighbors(List *c, int m, int lc, FmgrInfo *procinfo, Oid collation, HnswE
* Find duplicate element * Find duplicate element
*/ */
HnswElement HnswElement
HnswFindDuplicate(HnswElement e) HnswFindDuplicate(HnswElement e, Relation index)
{ {
HnswNeighborArray *neighbors = &e->neighbors[0]; HnswNeighborArray *neighbors = &e->neighbors[0];
/* TODO Implement */
if (IndexRelationGetNumberOfAttributes(index) > 1)
return NULL;
for (int i = 0; i < neighbors->length; i++) for (int i = 0; i < neighbors->length; i++)
{ {
HnswCandidate *neighbor = &neighbors->items[i]; HnswCandidate *neighbor = &neighbors->items[i];
@@ -904,7 +1086,7 @@ AddConnections(HnswElement element, List *neighbors, int m, int lc)
* Update connections * Update connections
*/ */
void void
HnswUpdateConnection(HnswElement element, HnswCandidate * hc, int m, int lc, int *updateIdx, Relation index, FmgrInfo *procinfo, Oid collation) HnswUpdateConnection(HnswElement element, HnswCandidate * hc, int m, int lc, int *updateIdx, Relation index, FmgrInfo **procinfos, Oid *collations, bool inMemory)
{ {
HnswNeighborArray *currentNeighbors = &hc->element->neighbors[lc]; HnswNeighborArray *currentNeighbors = &hc->element->neighbors[lc];
@@ -927,18 +1109,20 @@ HnswUpdateConnection(HnswElement element, HnswCandidate * hc, int m, int lc, int
HnswCandidate *pruned = NULL; HnswCandidate *pruned = NULL;
/* Load elements on insert */ /* Load elements on insert */
if (index != NULL) if (!inMemory)
{ {
Datum q = hc->element->value; Datum q = hc->element->value;
IndexTuple qtup = hc->element->itup;
ScanKeyData *keyData = NULL;
for (int i = 0; i < currentNeighbors->length; i++) for (int i = 0; i < currentNeighbors->length; i++)
{ {
HnswCandidate *hc3 = &currentNeighbors->items[i]; HnswCandidate *hc3 = &currentNeighbors->items[i];
if (DatumGetPointer(hc3->element->value) == NULL) if (DatumGetPointer(hc3->element->value) == NULL)
HnswLoadElement(hc3->element, &hc3->distance, &q, index, procinfo, collation, true); HnswLoadElement(hc3->element, &hc3->distance, &q, qtup, keyData, index, procinfos, collations, true);
else else
hc3->distance = GetCandidateDistance(hc3, q, procinfo, collation); hc3->distance = GetCandidateDistance(hc3, q, qtup, keyData, index, procinfos, collations);
/* Prune element if being deleted */ /* Prune element if being deleted */
if (list_length(hc3->element->heaptids) == 0) if (list_length(hc3->element->heaptids) == 0)
@@ -958,7 +1142,7 @@ HnswUpdateConnection(HnswElement element, HnswCandidate * hc, int m, int lc, int
c = lappend(c, &currentNeighbors->items[i]); c = lappend(c, &currentNeighbors->items[i]);
c = lappend(c, &hc2); c = lappend(c, &hc2);
SelectNeighbors(c, m, lc, procinfo, collation, hc->element, &hc2, &pruned, true); SelectNeighbors(c, m, lc, index, procinfos, collations, hc->element, &hc2, &pruned, true);
/* Should not happen */ /* Should not happen */
if (pruned == NULL) if (pruned == NULL)
@@ -1010,13 +1194,15 @@ RemoveElements(List *w, HnswElement skipElement)
* Algorithm 1 from paper * Algorithm 1 from paper
*/ */
void void
HnswInsertElement(HnswElement element, HnswElement entryPoint, Relation index, FmgrInfo *procinfo, Oid collation, int m, int efConstruction, bool existing) HnswInsertElement(HnswElement element, HnswElement entryPoint, Relation index, FmgrInfo **procinfos, Oid *collations, int m, int efConstruction, bool existing, bool inMemory)
{ {
List *ep; List *ep;
List *w; List *w;
int level = element->level; int level = element->level;
int entryLevel; int entryLevel;
Datum q = element->value; Datum q = element->value;
IndexTuple qtup = element->itup;
ScanKeyData *keyData = NULL;
HnswElement skipElement = existing ? element : NULL; HnswElement skipElement = existing ? element : NULL;
/* No neighbors if no entry point */ /* No neighbors if no entry point */
@@ -1024,13 +1210,13 @@ HnswInsertElement(HnswElement element, HnswElement entryPoint, Relation index, F
return; return;
/* Get entry point and level */ /* Get entry point and level */
ep = list_make1(HnswEntryCandidate(entryPoint, q, index, procinfo, collation, true)); ep = list_make1(HnswEntryCandidate(entryPoint, q, qtup, keyData, index, procinfos, collations, true, inMemory));
entryLevel = entryPoint->level; entryLevel = entryPoint->level;
/* 1st phase: greedy search to insert level */ /* 1st phase: greedy search to insert level */
for (int lc = entryLevel; lc >= level + 1; lc--) for (int lc = entryLevel; lc >= level + 1; lc--)
{ {
w = HnswSearchLayer(q, ep, 1, lc, index, procinfo, collation, m, true, skipElement); w = HnswSearchLayer(q, qtup, keyData, ep, 1, lc, index, procinfos, collations, m, true, skipElement, inMemory);
ep = w; ep = w;
} }
@@ -1048,11 +1234,11 @@ HnswInsertElement(HnswElement element, HnswElement entryPoint, Relation index, F
List *neighbors; List *neighbors;
List *lw; List *lw;
w = HnswSearchLayer(q, ep, efConstruction, lc, index, procinfo, collation, m, true, skipElement); w = HnswSearchLayer(q, qtup, keyData, ep, efConstruction, lc, index, procinfos, collations, m, true, skipElement, inMemory);
/* Elements being deleted or skipped can help with search */ /* Elements being deleted or skipped can help with search */
/* but should be removed before selecting neighbors */ /* but should be removed before selecting neighbors */
if (index != NULL) if (!inMemory)
lw = RemoveElements(w, skipElement); lw = RemoveElements(w, skipElement);
else else
lw = w; lw = w;
@@ -1062,7 +1248,7 @@ HnswInsertElement(HnswElement element, HnswElement entryPoint, Relation index, F
* sortCandidates to true for in-memory builds to enable closer * sortCandidates to true for in-memory builds to enable closer
* caching, but there does not seem to be a difference in performance. * caching, but there does not seem to be a difference in performance.
*/ */
neighbors = SelectNeighbors(lw, lm, lc, procinfo, collation, element, NULL, NULL, false); neighbors = SelectNeighbors(lw, lm, lc, index, procinfos, collations, element, NULL, NULL, false);
AddConnections(element, neighbors, lm, lc); AddConnections(element, neighbors, lm, lc);

View File

@@ -195,8 +195,8 @@ RepairGraphElement(HnswVacuumState * vacuumstate, HnswElement element, HnswEleme
GenericXLogState *state; GenericXLogState *state;
int m = vacuumstate->m; int m = vacuumstate->m;
int efConstruction = vacuumstate->efConstruction; int efConstruction = vacuumstate->efConstruction;
FmgrInfo *procinfo = vacuumstate->procinfo; FmgrInfo **procinfos = vacuumstate->procinfos;
Oid collation = vacuumstate->collation; Oid *collations = vacuumstate->collations;
BufferAccessStrategy bas = vacuumstate->bas; BufferAccessStrategy bas = vacuumstate->bas;
HnswNeighborTuple ntup = vacuumstate->ntup; HnswNeighborTuple ntup = vacuumstate->ntup;
Size ntupSize = HNSW_NEIGHBOR_TUPLE_SIZE(element->level, m); Size ntupSize = HNSW_NEIGHBOR_TUPLE_SIZE(element->level, m);
@@ -210,7 +210,7 @@ RepairGraphElement(HnswVacuumState * vacuumstate, HnswElement element, HnswEleme
element->heaptids = NIL; element->heaptids = NIL;
/* Add element to graph, skipping itself */ /* Add element to graph, skipping itself */
HnswInsertElement(element, entryPoint, index, procinfo, collation, m, efConstruction, true); HnswInsertElement(element, entryPoint, index, procinfos, collations, m, efConstruction, true, false);
/* Update neighbor tuple */ /* Update neighbor tuple */
/* Do this before getting page to minimize locking */ /* Do this before getting page to minimize locking */
@@ -231,7 +231,7 @@ RepairGraphElement(HnswVacuumState * vacuumstate, HnswElement element, HnswEleme
UnlockReleaseBuffer(buf); UnlockReleaseBuffer(buf);
/* Update neighbors */ /* Update neighbors */
HnswUpdateNeighborPages(index, procinfo, collation, element, m, true); HnswUpdateNeighborPages(index, procinfos, collations, element, m, true);
} }
/* /*
@@ -258,7 +258,7 @@ RepairGraphEntryPoint(HnswVacuumState * vacuumstate)
LockPage(index, HNSW_UPDATE_LOCK, ShareLock); LockPage(index, HNSW_UPDATE_LOCK, ShareLock);
/* Load element */ /* Load element */
HnswLoadElement(highestPoint, NULL, NULL, index, vacuumstate->procinfo, vacuumstate->collation, true); HnswLoadElement(highestPoint, NULL, NULL, NULL, NULL, index, vacuumstate->procinfos, vacuumstate->collations, true);
/* Repair if needed */ /* Repair if needed */
if (NeedsUpdated(vacuumstate, highestPoint)) if (NeedsUpdated(vacuumstate, highestPoint))
@@ -296,7 +296,7 @@ RepairGraphEntryPoint(HnswVacuumState * vacuumstate)
* is outdated, this can remove connections at higher levels in * is outdated, this can remove connections at higher levels in
* the graph until they are repaired, but this should be fine. * the graph until they are repaired, but this should be fine.
*/ */
HnswLoadElement(entryPoint, NULL, NULL, index, vacuumstate->procinfo, vacuumstate->collation, true); HnswLoadElement(entryPoint, NULL, NULL, NULL, NULL, index, vacuumstate->procinfos, vacuumstate->collations, true);
if (NeedsUpdated(vacuumstate, entryPoint)) if (NeedsUpdated(vacuumstate, entryPoint))
{ {
@@ -372,7 +372,7 @@ RepairGraph(HnswVacuumState * vacuumstate)
/* Create an element */ /* Create an element */
element = HnswInitElementFromBlock(blkno, offno); element = HnswInitElementFromBlock(blkno, offno);
HnswLoadElementFromTuple(element, etup, false, true); HnswLoadElementFromTuple(element, etup, false, true, index);
elements = lappend(elements, element); elements = lappend(elements, element);
} }
@@ -442,6 +442,7 @@ MarkDeleted(HnswVacuumState * vacuumstate)
BlockNumber insertPage = InvalidBlockNumber; BlockNumber insertPage = InvalidBlockNumber;
Relation index = vacuumstate->index; Relation index = vacuumstate->index;
BufferAccessStrategy bas = vacuumstate->bas; BufferAccessStrategy bas = vacuumstate->bas;
bool useIndexTuple = IndexRelationGetNumberOfAttributes(index);
/* /*
* Wait for index scans to complete. Scans before this point may contain * Wait for index scans to complete. Scans before this point may contain
@@ -530,7 +531,18 @@ MarkDeleted(HnswVacuumState * vacuumstate)
/* Overwrite element */ /* Overwrite element */
etup->deleted = 1; etup->deleted = 1;
MemSet(&etup->data, 0, VARSIZE_ANY(&etup->data)); if (useIndexTuple)
{
IndexTuple itup = (IndexTuple) &etup->data;
MemSet(itup, 0, IndexTupleSize(itup));
}
else
{
Vector *vec = (Vector *) (&etup->data);
MemSet(vec, 0, VARSIZE_ANY(vec));
}
/* Overwrite neighbors */ /* Overwrite neighbors */
for (int i = 0; i < ntup->count; i++) for (int i = 0; i < ntup->count; i++)
@@ -586,8 +598,8 @@ InitVacuumState(HnswVacuumState * vacuumstate, IndexVacuumInfo *info, IndexBulkD
vacuumstate->callback_state = callback_state; vacuumstate->callback_state = callback_state;
vacuumstate->efConstruction = HnswGetEfConstruction(index); vacuumstate->efConstruction = HnswGetEfConstruction(index);
vacuumstate->bas = GetAccessStrategy(BAS_BULKREAD); vacuumstate->bas = GetAccessStrategy(BAS_BULKREAD);
vacuumstate->procinfo = index_getprocinfo(index, 1, HNSW_DISTANCE_PROC); vacuumstate->procinfos = HnswInitProcinfos(index);
vacuumstate->collation = index->rd_indcollation[0]; vacuumstate->collations = index->rd_indcollation;
vacuumstate->ntup = palloc0(BLCKSZ); vacuumstate->ntup = palloc0(BLCKSZ);
vacuumstate->tmpCtx = AllocSetContextCreate(CurrentMemoryContext, vacuumstate->tmpCtx = AllocSetContextCreate(CurrentMemoryContext,
"Hnsw vacuum temporary context", "Hnsw vacuum temporary context",
@@ -611,6 +623,7 @@ FreeVacuumState(HnswVacuumState * vacuumstate)
{ {
hash_destroy(vacuumstate->deleted); hash_destroy(vacuumstate->deleted);
FreeAccessStrategy(vacuumstate->bas); FreeAccessStrategy(vacuumstate->bas);
pfree(vacuumstate->procinfos);
pfree(vacuumstate->ntup); pfree(vacuumstate->ntup);
MemoryContextDelete(vacuumstate->tmpCtx); MemoryContextDelete(vacuumstate->tmpCtx);
} }

View File

@@ -83,32 +83,11 @@ for my $i (0 .. $#operators)
push(@expected, $res); push(@expected, $res);
} }
# Build index serially # Add index
$node->safe_psql("postgres", qq( $node->safe_psql("postgres", "CREATE INDEX ON tst USING hnsw (v $opclass);");
SET max_parallel_maintenance_workers = 0;
CREATE INDEX idx ON tst USING hnsw (v $opclass);
));
# Test approximate results
my $min = $operator eq "<#>" ? 0.80 : 0.99; my $min = $operator eq "<#>" ? 0.80 : 0.99;
test_recall($min, $operator); test_recall($min, $operator);
$node->safe_psql("postgres", "DROP INDEX idx;");
# Build index in parallel
my ($ret, $stdout, $stderr) = $node->psql("postgres", qq(
SET client_min_messages = DEBUG;
SET min_parallel_table_scan_size = 1;
SET hnsw.enable_parallel_build = on;
CREATE INDEX idx ON tst USING hnsw (v $opclass);
));
is($ret, 0, $stderr);
like($stderr, qr/using \d+ parallel workers/);
# Test approximate results
test_recall($min, $operator);
$node->safe_psql("postgres", "DROP INDEX idx;");
} }
done_testing(); done_testing();

View File

@@ -0,0 +1,107 @@
use strict;
use warnings;
use PostgresNode;
use TestLib;
use Test::More;
my $node;
my @queries = ();
my @cs = ();
my @expected;
my $limit = 20;
my $dim = 3;
my $array_sql = join(",", ('random()') x $dim);
my $nc = 50;
sub test_recall
{
my ($min, $operator) = @_;
my $correct = 0;
my $total = 0;
my $explain = $node->safe_psql("postgres", qq(
SET enable_seqscan = off;
EXPLAIN ANALYZE SELECT i FROM tst WHERE c = $cs[0] ORDER BY v $operator '$queries[0]' LIMIT $limit;
));
like($explain, qr/Index Cond/);
for my $i (0 .. $#queries)
{
my $actual = $node->safe_psql("postgres", qq(
SET enable_seqscan = off;
SELECT i FROM tst WHERE c = $cs[$i] ORDER BY v $operator '$queries[$i]' LIMIT $limit;
));
my @actual_ids = split("\n", $actual);
my %actual_set = map { $_ => 1 } @actual_ids;
my @expected_ids = split("\n", $expected[$i]);
foreach (@expected_ids)
{
if (exists($actual_set{$_}))
{
$correct++;
}
$total++;
}
}
cmp_ok($correct / $total, ">=", $min, $operator);
}
# Initialize node
$node = get_new_node('node');
$node->init;
$node->start;
# Create table
$node->safe_psql("postgres", "CREATE EXTENSION vector;");
$node->safe_psql("postgres", "CREATE TABLE tst (i int4, v vector($dim), c int8);");
$node->safe_psql("postgres",
"INSERT INTO tst SELECT i, ARRAY[$array_sql], i % $nc FROM generate_series(1, 20000) i;"
);
# Generate queries
for (1 .. 20)
{
my @r = ();
for (1 .. $dim)
{
push(@r, rand());
}
push(@queries, "[" . join(",", @r) . "]");
push(@cs, int(rand() * $nc));
}
# Get exact results
@expected = ();
for my $i (0 .. $#queries)
{
my $res = $node->safe_psql("postgres", "SELECT i FROM tst WHERE c = $cs[$i] ORDER BY v <-> '$queries[$i]' LIMIT $limit;");
push(@expected, $res);
}
# Add index
$node->safe_psql("postgres", "CREATE INDEX ON tst USING hnsw (v vector_l2_ops, c);");
# Test recall
test_recall(0.99, '<->');
# Test vacuum
$node->safe_psql("postgres", "DELETE FROM tst WHERE c > 5;");
$node->safe_psql("postgres", "VACUUM tst;");
# Test columns
my ($ret, $stdout, $stderr) = $node->psql("postgres", "CREATE INDEX ON tst USING hnsw (c);");
like($stderr, qr/first column must be a vector/);
($ret, $stdout, $stderr) = $node->psql("postgres", "CREATE INDEX ON tst USING hnsw (c, v vector_l2_ops);");
like($stderr, qr/first column must be a vector/);
($ret, $stdout, $stderr) = $node->psql("postgres", "CREATE INDEX ON tst USING hnsw (v vector_l2_ops, c, c);");
like($stderr, qr/index cannot have more than two columns/);
($ret, $stdout, $stderr) = $node->psql("postgres", "CREATE INDEX ON tst USING hnsw (v vector_l2_ops, v vector_l2_ops);");
like($stderr, qr/column 2 cannot be a vector/);
done_testing();