Compare commits

..

4 Commits

Author SHA1 Message Date
Andrew Kane
f90d52d562 Added test for explicit zeros [skip ci] 2023-11-06 14:29:52 -08:00
Andrew Kane
22a1b06924 Added todo [skip ci] 2023-11-05 21:26:28 -08:00
Andrew Kane
b17b9c1ca2 Updated version [skip ci] 2023-11-05 18:21:09 -08:00
Andrew Kane
492ae1225c Added support for sparse vectors 2023-11-05 18:12:19 -08:00
20 changed files with 1353 additions and 898 deletions

View File

@@ -1,9 +1,6 @@
## 0.5.2 (unreleased)
## 0.6.0 (unreleased)
- Improved performance of HNSW
- Added support for on-disk parallel index builds for HNSW
- Reduced WAL generation for HNSW index builds
- Fixed `invalid memory alloc request size` error with HNSW index build
- Added support for sparse vectors
## 0.5.1 (2023-10-10)

View File

@@ -3,8 +3,8 @@ EXTVERSION = 0.5.1
MODULE_big = vector
DATA = $(wildcard sql/*--*.sql)
OBJS = src/hnsw.o src/hnswbuild.o src/hnswinsert.o src/hnswscan.o src/hnswutils.o src/hnswvacuum.o src/ivfbuild.o src/ivfflat.o src/ivfinsert.o src/ivfkmeans.o src/ivfscan.o src/ivfutils.o src/ivfvacuum.o src/vector.o
HEADERS = src/vector.h
OBJS = src/hnsw.o src/hnswbuild.o src/hnswinsert.o src/hnswscan.o src/hnswutils.o src/hnswvacuum.o src/ivfbuild.o src/ivfflat.o src/ivfinsert.o src/ivfkmeans.o src/ivfscan.o src/ivfutils.o src/ivfvacuum.o src/svector.o src/vector.o
HEADERS = src/svector.h src/vector.h
TESTS = $(wildcard test/sql/*.sql)
REGRESS = $(patsubst test/sql/%.sql,%,$(TESTS))

View File

@@ -1,8 +1,8 @@
EXTENSION = vector
EXTVERSION = 0.5.1
OBJS = src\hnsw.obj src\hnswbuild.obj src\hnswinsert.obj src\hnswscan.obj src\hnswutils.obj src\hnswvacuum.obj src\ivfbuild.obj src\ivfflat.obj src\ivfinsert.obj src\ivfkmeans.obj src\ivfscan.obj src\ivfutils.obj src\ivfvacuum.obj src\vector.obj
HEADERS = src\vector.h
OBJS = src\hnsw.obj src\hnswbuild.obj src\hnswinsert.obj src\hnswscan.obj src\hnswutils.obj src\hnswvacuum.obj src\ivfbuild.obj src\ivfflat.obj src\ivfinsert.obj src\ivfkmeans.obj src\ivfscan.obj src\ivfutils.obj src\ivfvacuum.obj src\svector.obj src\vector.obj
HEADERS = src\svector.h 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=$(EXTENSION)

View File

@@ -10,7 +10,7 @@ Store your vectors with the rest of your data. Supports:
Plus [ACID](https://en.wikipedia.org/wiki/ACID) compliance, point-in-time recovery, JOINs, and all of the other [great features](https://www.postgresql.org/about/) of Postgres
[![Build Status](https://github.com/pgvector/pgvector/actions/workflows/build.yml/badge.svg)](https://github.com/pgvector/pgvector/actions)
[![Build Status](https://github.com/pgvector/pgvector/workflows/build/badge.svg?branch=master)](https://github.com/pgvector/pgvector/actions)
## Installation
@@ -269,8 +269,6 @@ Specify HNSW parameters
CREATE INDEX ON items USING hnsw (embedding vector_l2_ops) WITH (m = 16, ef_construction = 64);
```
A higher value of `ef_construction` provides better recall at the cost of index build time / insert speed.
### Query Options
Specify the size of the dynamic candidate list for search (40 by default)
@@ -371,6 +369,26 @@ To speed up queries with an IVFFlat index, increase the number of inverted lists
CREATE INDEX ON items USING ivfflat (embedding vector_l2_ops) WITH (lists = 1000);
```
## Sparse Vectors
Create a sparse vector column with 10 dimensions
```sql
CREATE TABLE items (id bigserial PRIMARY KEY, embedding svector(10));
```
Insert vectors
```sql
INSERT INTO items (embedding) VALUES ('(0,1),(1,2),(2,3)|10|'), ('(0,4),(1,5),(4,6)|10|');
```
Get the nearest neighbors by L2 distance
```sql
SELECT * FROM items ORDER BY embedding <-> '(0,3),(1,1),(2,2)|10|' LIMIT 5;
```
## Languages
Use pgvector from any language with a Postgres client. You can even generate and store vectors in one language and query them in another.
@@ -379,7 +397,7 @@ Language | Libraries / Examples
--- | ---
C | [pgvector-c](https://github.com/pgvector/pgvector-c)
C++ | [pgvector-cpp](https://github.com/pgvector/pgvector-cpp)
C#, F#, Visual Basic | [pgvector-dotnet](https://github.com/pgvector/pgvector-dotnet)
C# | [pgvector-dotnet](https://github.com/pgvector/pgvector-dotnet)
Crystal | [pgvector-crystal](https://github.com/pgvector/pgvector-crystal)
Dart | [pgvector-dart](https://github.com/pgvector/pgvector-dart)
Elixir | [pgvector-elixir](https://github.com/pgvector/pgvector-elixir)
@@ -388,10 +406,8 @@ Haskell | [pgvector-haskell](https://github.com/pgvector/pgvector-haskell)
Java, Kotlin, Groovy, Scala | [pgvector-java](https://github.com/pgvector/pgvector-java)
JavaScript, TypeScript | [pgvector-node](https://github.com/pgvector/pgvector-node)
Julia | [pgvector-julia](https://github.com/pgvector/pgvector-julia)
Lisp | [pgvector-lisp](https://github.com/pgvector/pgvector-lisp)
Lua | [pgvector-lua](https://github.com/pgvector/pgvector-lua)
Nim | [pgvector-nim](https://github.com/pgvector/pgvector-nim)
OCaml | [pgvector-ocaml](https://github.com/pgvector/pgvector-ocaml)
Perl | [pgvector-perl](https://github.com/pgvector/pgvector-perl)
PHP | [pgvector-php](https://github.com/pgvector/pgvector-php)
Python | [pgvector-python](https://github.com/pgvector/pgvector-python)
@@ -464,14 +480,6 @@ and query with:
SELECT * FROM items ORDER BY embedding::vector(3) <-> '[3,1,2]' LIMIT 5;
```
#### Do indexes need to fit into memory?
No, but like other index types, youll likely see better performance if they do. You can get the size of an index with:
```sql
SELECT pg_size_pretty(pg_relation_size('index_name'));
```
## Troubleshooting
#### Why isnt a query using an index?

View File

@@ -0,0 +1,79 @@
-- 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
CREATE TYPE svector;
CREATE FUNCTION svector_in(cstring, oid, integer) RETURNS svector
AS 'MODULE_PATHNAME' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
CREATE FUNCTION svector_out(svector) RETURNS cstring
AS 'MODULE_PATHNAME' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
CREATE FUNCTION svector_typmod_in(cstring[]) RETURNS integer
AS 'MODULE_PATHNAME' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
CREATE FUNCTION svector_recv(internal, oid, integer) RETURNS svector
AS 'MODULE_PATHNAME' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
CREATE FUNCTION svector_send(svector) RETURNS bytea
AS 'MODULE_PATHNAME' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
CREATE TYPE svector (
INPUT = svector_in,
OUTPUT = svector_out,
TYPMOD_IN = svector_typmod_in,
RECEIVE = svector_recv,
SEND = svector_send,
STORAGE = external
);
CREATE FUNCTION l2_distance(svector, svector) RETURNS float8
AS 'MODULE_PATHNAME', 'svector_l2_distance' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
CREATE FUNCTION inner_product(svector, svector) RETURNS float8
AS 'MODULE_PATHNAME', 'svector_inner_product' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
CREATE FUNCTION cosine_distance(svector, svector) RETURNS float8
AS 'MODULE_PATHNAME', 'svector_cosine_distance' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
CREATE FUNCTION jaccard_distance(svector, svector) RETURNS float8
AS 'MODULE_PATHNAME', 'svector_jaccard_distance' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
CREATE FUNCTION svector_l2_squared_distance(svector, svector) RETURNS float8
AS 'MODULE_PATHNAME' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
CREATE FUNCTION svector_negative_inner_product(svector, svector) RETURNS float8
AS 'MODULE_PATHNAME' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
CREATE FUNCTION svector(svector, integer, boolean) RETURNS svector
AS 'MODULE_PATHNAME' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
CREATE FUNCTION vector_to_svector(vector, integer, boolean) RETURNS svector
AS 'MODULE_PATHNAME' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
CREATE FUNCTION svector_to_vector(svector, integer, boolean) RETURNS vector
AS 'MODULE_PATHNAME' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
CREATE CAST (svector AS svector)
WITH FUNCTION svector(svector, integer, boolean) AS IMPLICIT;
CREATE CAST (svector AS vector)
WITH FUNCTION svector_to_vector(svector, integer, boolean) AS IMPLICIT;
CREATE CAST (vector AS svector)
WITH FUNCTION vector_to_svector(vector, integer, boolean) AS IMPLICIT;
CREATE OPERATOR <-> (
LEFTARG = svector, RIGHTARG = svector, PROCEDURE = l2_distance,
COMMUTATOR = '<->'
);
CREATE OPERATOR <#> (
LEFTARG = svector, RIGHTARG = svector, PROCEDURE = svector_negative_inner_product,
COMMUTATOR = '<#>'
);
CREATE OPERATOR <=> (
LEFTARG = svector, RIGHTARG = svector, PROCEDURE = cosine_distance,
COMMUTATOR = '<=>'
);

View File

@@ -290,3 +290,92 @@ CREATE OPERATOR CLASS vector_cosine_ops
OPERATOR 1 <=> (vector, vector) FOR ORDER BY float_ops,
FUNCTION 1 vector_negative_inner_product(vector, vector),
FUNCTION 2 vector_norm(vector);
--- svector type
CREATE TYPE svector;
CREATE FUNCTION svector_in(cstring, oid, integer) RETURNS svector
AS 'MODULE_PATHNAME' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
CREATE FUNCTION svector_out(svector) RETURNS cstring
AS 'MODULE_PATHNAME' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
CREATE FUNCTION svector_typmod_in(cstring[]) RETURNS integer
AS 'MODULE_PATHNAME' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
CREATE FUNCTION svector_recv(internal, oid, integer) RETURNS svector
AS 'MODULE_PATHNAME' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
CREATE FUNCTION svector_send(svector) RETURNS bytea
AS 'MODULE_PATHNAME' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
CREATE TYPE svector (
INPUT = svector_in,
OUTPUT = svector_out,
TYPMOD_IN = svector_typmod_in,
RECEIVE = svector_recv,
SEND = svector_send,
STORAGE = external
);
-- svector functions
CREATE FUNCTION l2_distance(svector, svector) RETURNS float8
AS 'MODULE_PATHNAME', 'svector_l2_distance' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
CREATE FUNCTION inner_product(svector, svector) RETURNS float8
AS 'MODULE_PATHNAME', 'svector_inner_product' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
CREATE FUNCTION cosine_distance(svector, svector) RETURNS float8
AS 'MODULE_PATHNAME', 'svector_cosine_distance' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
CREATE FUNCTION jaccard_distance(svector, svector) RETURNS float8
AS 'MODULE_PATHNAME', 'svector_jaccard_distance' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
-- svector private functions
CREATE FUNCTION svector_l2_squared_distance(svector, svector) RETURNS float8
AS 'MODULE_PATHNAME' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
CREATE FUNCTION svector_negative_inner_product(svector, svector) RETURNS float8
AS 'MODULE_PATHNAME' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
-- svector cast functions
CREATE FUNCTION svector(svector, integer, boolean) RETURNS svector
AS 'MODULE_PATHNAME' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
CREATE FUNCTION vector_to_svector(vector, integer, boolean) RETURNS svector
AS 'MODULE_PATHNAME' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
CREATE FUNCTION svector_to_vector(svector, integer, boolean) RETURNS vector
AS 'MODULE_PATHNAME' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
-- svector casts
CREATE CAST (svector AS svector)
WITH FUNCTION svector(svector, integer, boolean) AS IMPLICIT;
CREATE CAST (svector AS vector)
WITH FUNCTION svector_to_vector(svector, integer, boolean) AS IMPLICIT;
CREATE CAST (vector AS svector)
WITH FUNCTION vector_to_svector(vector, integer, boolean) AS IMPLICIT;
-- svector operators
CREATE OPERATOR <-> (
LEFTARG = svector, RIGHTARG = svector, PROCEDURE = l2_distance,
COMMUTATOR = '<->'
);
CREATE OPERATOR <#> (
LEFTARG = svector, RIGHTARG = svector, PROCEDURE = svector_negative_inner_product,
COMMUTATOR = '<#>'
);
CREATE OPERATOR <=> (
LEFTARG = svector, RIGHTARG = svector, PROCEDURE = cosine_distance,
COMMUTATOR = '<=>'
);

View File

@@ -14,7 +14,6 @@
#endif
int hnsw_ef_search;
bool hnsw_enable_parallel_build;
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",
"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);
/* 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);
}
/*

View File

@@ -4,7 +4,6 @@
#include "postgres.h"
#include "access/generic_xlog.h"
#include "access/parallel.h"
#include "access/reloptions.h"
#include "nodes/execnodes.h"
#include "port.h" /* for random() */
@@ -15,10 +14,6 @@
#error "Requires PostgreSQL 11+"
#endif
#if PG_VERSION_NUM < 120000
#include "access/relscan.h"
#endif
#define HNSW_MAX_DIM 2000
/* Support functions */
@@ -64,7 +59,7 @@
#define HNSW_MAX_SIZE (BLCKSZ - MAXALIGN(SizeOfPageHeaderData) - MAXALIGN(sizeof(HnswPageOpaqueData)) - sizeof(ItemIdData))
#define HNSW_ELEMENT_TUPLE_SIZE(size) MAXALIGN(offsetof(HnswElementTupleData, data) + (size))
#define HNSW_ELEMENT_TUPLE_SIZE(_dim) MAXALIGN(offsetof(HnswElementTupleData, vec) + VECTOR_SIZE(_dim))
#define HNSW_NEIGHBOR_TUPLE_SIZE(level, m) MAXALIGN(offsetof(HnswNeighborTupleData, indextids) + ((level) + 2) * (m) * sizeof(ItemPointerData))
#define HnswPageGetOpaque(page) ((HnswPageOpaque) PageGetSpecialPointer(page))
@@ -95,7 +90,6 @@
/* Variables */
extern int hnsw_ef_search;
extern bool hnsw_enable_parallel_build;
typedef struct HnswNeighborArray HnswNeighborArray;
@@ -104,13 +98,12 @@ typedef struct HnswElementData
List *heaptids;
uint8 level;
uint8 deleted;
uint32 hash;
HnswNeighborArray *neighbors;
BlockNumber blkno;
OffsetNumber offno;
OffsetNumber neighborOffno;
BlockNumber neighborPage;
Datum value;
Vector *vec;
} HnswElementData;
typedef HnswElementData * HnswElement;
@@ -143,49 +136,6 @@ typedef struct HnswOptions
int efConstruction; /* size of dynamic candidate list */
} 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
{
/* Info */
@@ -213,16 +163,12 @@ typedef struct HnswBuildState
HnswElement entryPoint;
double ml;
int maxLevel;
long memoryLeft;
double maxInMemoryElements;
bool flushed;
Vector *normvec;
/* Memory */
MemoryContext tmpCtx;
/* Parallel builds */
HnswLeader *hnswleader;
HnswShared *hnswshared;
} HnswBuildState;
typedef struct HnswMetaPageData
@@ -258,7 +204,7 @@ typedef struct HnswElementTupleData
ItemPointerData heaptids[HNSW_HEAPTIDS];
ItemPointerData neighbortid;
uint16 unused2;
Vector data;
Vector vec;
} HnswElementTupleData;
typedef HnswElementTupleData * HnswElementTuple;
@@ -304,7 +250,7 @@ typedef struct HnswVacuumState
Oid collation;
/* Variables */
struct tidhash_hash *deleted;
HTAB *deleted;
BufferAccessStrategy bas;
HnswNeighborTuple ntup;
HnswElementData highestPoint;
@@ -318,8 +264,10 @@ int HnswGetM(Relation index);
int HnswGetEfConstruction(Relation index);
FmgrInfo *HnswOptionalProcInfo(Relation index, uint16 procnum);
bool HnswNormValue(FmgrInfo *procinfo, Oid collation, Datum *value, Vector * result);
void HnswCommitBuffer(Buffer buf, GenericXLogState *state);
Buffer HnswNewBuffer(Relation index, ForkNumber forkNum);
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, int m, bool inserting, HnswElement skipElement);
HnswElement HnswGetEntryPoint(Relation index);
@@ -330,18 +278,17 @@ HnswElement HnswInitElementFromBlock(BlockNumber blkno, OffsetNumber offno);
void HnswInsertElement(HnswElement element, HnswElement entryPoint, Relation index, FmgrInfo *procinfo, Oid collation, int m, int efConstruction, bool existing);
HnswElement HnswFindDuplicate(HnswElement e);
HnswCandidate *HnswEntryCandidate(HnswElement em, Datum q, Relation rel, FmgrInfo *procinfo, Oid collation, bool loadVec);
void HnswUpdateMetaPage(Relation index, int updateEntry, HnswElement entryPoint, BlockNumber insertPage, ForkNumber forkNum, bool building);
void HnswUpdateMetaPage(Relation index, int updateEntry, HnswElement entryPoint, BlockNumber insertPage, ForkNumber forkNum);
void HnswSetNeighborTuple(HnswNeighborTuple ntup, HnswElement e, int m);
void HnswAddHeapTid(HnswElement element, ItemPointer heaptid);
void HnswInitNeighbors(HnswElement element, int m);
bool HnswInsertTuple(Relation index, Datum *values, bool *isnull, ItemPointer heap_tid, Relation heapRel, bool building);
void HnswUpdateNeighborPages(Relation index, FmgrInfo *procinfo, Oid collation, HnswElement e, int m, bool checkExisting, bool building);
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 HnswLoadElementFromTuple(HnswElement element, HnswElementTuple etup, bool loadHeaptids, bool loadVec);
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, int m);
PGDLLEXPORT void HnswParallelBuildMain(dsm_segment *seg, shm_toc *toc);
/* Index access methods */
IndexBuildResult *hnswbuild(Relation heap, Relation index, IndexInfo *indexInfo);
@@ -359,31 +306,4 @@ void hnswrescan(IndexScanDesc scan, ScanKey keys, int nkeys, ScanKey orderbys,
bool hnswgettuple(IndexScanDesc scan, ScanDirection dir);
void hnswendscan(IndexScanDesc scan);
/* Hash tables */
typedef struct TidHashEntry
{
ItemPointerData tid;
char status;
} TidHashEntry;
#define SH_PREFIX tidhash
#define SH_ELEMENT_TYPE TidHashEntry
#define SH_KEY_TYPE ItemPointerData
#define SH_SCOPE extern
#define SH_DECLARE
#include "lib/simplehash.h"
typedef struct PointerHashEntry
{
uintptr_t ptr;
char status;
} PointerHashEntry;
#define SH_PREFIX pointerhash
#define SH_ELEMENT_TYPE PointerHashEntry
#define SH_KEY_TYPE uintptr_t
#define SH_SCOPE extern
#define SH_DECLARE
#include "lib/simplehash.h"
#endif

View File

@@ -2,16 +2,12 @@
#include <math.h>
#include "access/parallel.h"
#include "access/xact.h"
#include "catalog/index.h"
#include "hnsw.h"
#include "miscadmin.h"
#include "lib/pairingheap.h"
#include "nodes/pg_list.h"
#include "storage/bufmgr.h"
#include "tcop/tcopprot.h"
#include "utils/datum.h"
#include "utils/memutils.h"
#if PG_VERSION_NUM >= 140000
@@ -39,25 +35,6 @@
#define UpdateProgress(index, val) ((void)val)
#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)
#define LIST_MAX_LENGTH ((1 << 26) - 1)
/*
* Create the metapage
*/
@@ -68,11 +45,11 @@ CreateMetaPage(HnswBuildState * buildstate)
ForkNumber forkNum = buildstate->forkNum;
Buffer buf;
Page page;
GenericXLogState *state;
HnswMetaPage metap;
buf = HnswNewBuffer(index, forkNum);
page = BufferGetPage(buf);
HnswInitPage(buf, page);
HnswInitRegisterPage(index, &buf, &page, &state);
/* Set metapage data */
metap = HnswPageGetMeta(page);
@@ -88,14 +65,14 @@ CreateMetaPage(HnswBuildState * buildstate)
((PageHeader) page)->pd_lower =
((char *) metap + sizeof(HnswMetaPageData)) - (char *) page;
UnlockReleaseBuffer(buf);
HnswCommitBuffer(buf, state);
}
/*
* Add a new page
*/
static void
HnswBuildAppendPage(Relation index, Buffer *buf, Page *page, ForkNumber forkNum)
HnswBuildAppendPage(Relation index, Buffer *buf, Page *page, GenericXLogState **state, ForkNumber forkNum)
{
/* Add a new page */
Buffer newbuf = HnswNewBuffer(index, forkNum);
@@ -104,6 +81,7 @@ HnswBuildAppendPage(Relation index, Buffer *buf, Page *page, ForkNumber forkNum)
HnswPageGetOpaque(*page)->nextblkno = BufferGetBlockNumber(newbuf);
/* Commit */
GenericXLogFinish(*state);
UnlockReleaseBuffer(*buf);
/* Can take a while, so ensure we can interrupt */
@@ -114,7 +92,8 @@ HnswBuildAppendPage(Relation index, Buffer *buf, Page *page, ForkNumber forkNum)
/* Prepare new page */
*buf = newbuf;
*page = BufferGetPage(*buf);
*state = GenericXLogStart(index);
*page = GenericXLogRegisterBuffer(*state, *buf, GENERIC_XLOG_FULL_IMAGE);
HnswInitPage(*buf, *page);
}
@@ -126,52 +105,46 @@ CreateElementPages(HnswBuildState * buildstate)
{
Relation index = buildstate->index;
ForkNumber forkNum = buildstate->forkNum;
Size etupAllocSize;
int dimensions = buildstate->dimensions;
Size etupSize;
Size maxSize;
HnswElementTuple etup;
HnswNeighborTuple ntup;
BlockNumber insertPage;
Buffer buf;
Page page;
GenericXLogState *state;
ListCell *lc;
/* Calculate sizes */
etupAllocSize = BLCKSZ;
maxSize = HNSW_MAX_SIZE;
etupSize = HNSW_ELEMENT_TUPLE_SIZE(dimensions);
/* Allocate once */
etup = palloc0(etupAllocSize);
etup = palloc0(etupSize);
ntup = palloc0(BLCKSZ);
/* Prepare first page */
buf = HnswNewBuffer(index, forkNum);
page = BufferGetPage(buf);
state = GenericXLogStart(index);
page = GenericXLogRegisterBuffer(state, buf, GENERIC_XLOG_FULL_IMAGE);
HnswInitPage(buf, page);
foreach(lc, buildstate->elements)
{
HnswElement element = lfirst(lc);
Size etupSize;
Size ntupSize;
Size combinedSize;
/* Zero memory for each element */
MemSet(etup, 0, etupAllocSize);
HnswSetElementTuple(etup, element);
/* Calculate sizes */
etupSize = HNSW_ELEMENT_TUPLE_SIZE(VARSIZE_ANY(DatumGetPointer(element->value)));
ntupSize = HNSW_NEIGHBOR_TUPLE_SIZE(element->level, buildstate->m);
combinedSize = etupSize + ntupSize + sizeof(ItemIdData);
/* Initial size check */
if (etupSize > etupAllocSize)
elog(ERROR, "index tuple too large");
HnswSetElementTuple(etup, element);
/* Keep element and neighbors on the same page if possible */
if (PageGetFreeSpace(page) < etupSize || (combinedSize <= maxSize && PageGetFreeSpace(page) < combinedSize))
HnswBuildAppendPage(index, &buf, &page, forkNum);
HnswBuildAppendPage(index, &buf, &page, &state, forkNum);
/* Calculate offsets */
element->blkno = BufferGetBlockNumber(buf);
@@ -195,7 +168,7 @@ CreateElementPages(HnswBuildState * buildstate)
/* Add new page if needed */
if (PageGetFreeSpace(page) < ntupSize)
HnswBuildAppendPage(index, &buf, &page, forkNum);
HnswBuildAppendPage(index, &buf, &page, &state, forkNum);
/* Add placeholder for neighbors */
if (PageAddItem(page, (Item) ntup, ntupSize, InvalidOffsetNumber, false, false) != element->neighborOffno)
@@ -205,9 +178,10 @@ CreateElementPages(HnswBuildState * buildstate)
insertPage = BufferGetBlockNumber(buf);
/* Commit */
GenericXLogFinish(state);
UnlockReleaseBuffer(buf);
HnswUpdateMetaPage(index, HNSW_UPDATE_ENTRY_ALWAYS, buildstate->entryPoint, insertPage, forkNum, true);
HnswUpdateMetaPage(index, HNSW_UPDATE_ENTRY_ALWAYS, buildstate->entryPoint, insertPage, forkNum);
pfree(etup);
pfree(ntup);
@@ -233,6 +207,7 @@ CreateNeighborPages(HnswBuildState * buildstate)
HnswElement e = lfirst(lc);
Buffer buf;
Page page;
GenericXLogState *state;
Size ntupSize = HNSW_NEIGHBOR_TUPLE_SIZE(e->level, m);
/* Can take a while, so ensure we can interrupt */
@@ -241,7 +216,8 @@ CreateNeighborPages(HnswBuildState * buildstate)
buf = ReadBufferExtended(index, forkNum, e->neighborPage, RBM_NORMAL, NULL);
LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
page = BufferGetPage(buf);
state = GenericXLogStart(index);
page = GenericXLogRegisterBuffer(state, buf, 0);
HnswSetNeighborTuple(ntup, e, m);
@@ -249,6 +225,7 @@ CreateNeighborPages(HnswBuildState * buildstate)
elog(ERROR, "failed to add index item to \"%s\"", RelationGetRelationName(index));
/* Commit */
GenericXLogFinish(state);
UnlockReleaseBuffer(buf);
}
@@ -287,14 +264,13 @@ FlushPages(HnswBuildState * buildstate)
* Insert tuple
*/
static bool
InsertTuple(Relation index, Datum *values, HnswElement element, HnswBuildState * buildstate, HnswElement * dup, MemoryContext outerCtx)
InsertTuple(Relation index, Datum *values, HnswElement element, HnswBuildState * buildstate, HnswElement * dup)
{
FmgrInfo *procinfo = buildstate->procinfo;
Oid collation = buildstate->collation;
HnswElement entryPoint = buildstate->entryPoint;
int efConstruction = buildstate->efConstruction;
int m = buildstate->m;
MemoryContext oldCtx;
/* Detoast once for all calls */
Datum value = PointerGetDatum(PG_DETOAST_DATUM(values[0]));
@@ -307,9 +283,7 @@ InsertTuple(Relation index, Datum *values, HnswElement element, HnswBuildState *
}
/* Copy value to element so accessible outside of memory context */
oldCtx = MemoryContextSwitchTo(outerCtx);
element->value = datumCopy(value, false, -1);
MemoryContextSwitchTo(oldCtx);
memcpy(element->vec, DatumGetVector(value), VECTOR_SIZE(buildstate->dimensions));
/* Insert element in graph */
HnswInsertElement(element, entryPoint, NULL, procinfo, collation, m, efConstruction, false);
@@ -339,21 +313,6 @@ InsertTuple(Relation index, Datum *values, HnswElement element, HnswBuildState *
return *dup == NULL;
}
/*
* Get the memory used by an element
*/
static long
HnswElementMemory(HnswElement e, int m)
{
long elementSize = sizeof(HnswElementData);
elementSize += sizeof(HnswNeighborArray) * (e->level + 1);
elementSize += sizeof(HnswCandidate) * (m * (e->level + 2));
elementSize += sizeof(ItemPointerData);
elementSize += VARSIZE_ANY(DatumGetPointer(e->value));
return elementSize;
}
/*
* Callback for table_index_build_scan
*/
@@ -375,34 +334,22 @@ BuildCallback(Relation index, CALLBACK_ITEM_POINTER, Datum *values,
if (isnull[0])
return;
if (buildstate->flushed || buildstate->memoryLeft <= 0 || list_length(buildstate->elements) == LIST_MAX_LENGTH)
if (buildstate->indtuples >= buildstate->maxInMemoryElements)
{
if (!buildstate->flushed)
{
if (buildstate->memoryLeft <= 0)
ereport(NOTICE,
(errmsg("hnsw graph no longer fits into maintenance_work_mem after " INT64_FORMAT " tuples", (int64) buildstate->indtuples),
errdetail("Building will take significantly more time."),
errhint("Increase maintenance_work_mem to speed up builds.")));
ereport(NOTICE,
(errmsg("hnsw graph no longer fits into maintenance_work_mem after " INT64_FORMAT " tuples", (int64) buildstate->indtuples),
errdetail("Building will take significantly more time."),
errhint("Increase maintenance_work_mem to speed up builds.")));
FlushPages(buildstate);
}
oldCtx = MemoryContextSwitchTo(buildstate->tmpCtx);
if (HnswInsertTuple(buildstate->index, values, isnull, tid, buildstate->heap, true))
{
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);
}
if (HnswInsertTuple(buildstate->index, values, isnull, tid, buildstate->heap))
UpdateProgress(PROGRESS_CREATEIDX_TUPLES_DONE, ++buildstate->indtuples);
/* Reset memory context */
MemoryContextSwitchTo(oldCtx);
@@ -413,12 +360,13 @@ BuildCallback(Relation index, CALLBACK_ITEM_POINTER, Datum *values,
/* Allocate necessary memory outside of memory context */
element = HnswInitElement(tid, buildstate->m, buildstate->ml, buildstate->maxLevel);
element->vec = palloc(VECTOR_SIZE(buildstate->dimensions));
/* Use memory context since detoast can allocate */
oldCtx = MemoryContextSwitchTo(buildstate->tmpCtx);
/* Insert tuple */
inserted = InsertTuple(index, values, element, buildstate, &dup, oldCtx);
inserted = InsertTuple(index, values, element, buildstate, &dup);
/* Reset memory context */
MemoryContextSwitchTo(oldCtx);
@@ -426,21 +374,31 @@ BuildCallback(Relation index, CALLBACK_ITEM_POINTER, Datum *values,
/* Add outside memory context */
if (dup != NULL)
{
HnswAddHeapTid(dup, tid);
buildstate->memoryLeft -= sizeof(ItemPointerData);
}
/* Add to buildstate or free */
if (inserted)
{
buildstate->elements = lappend(buildstate->elements, element);
buildstate->memoryLeft -= HnswElementMemory(element, buildstate->m);
}
else
HnswFreeElement(element);
}
/*
* Get the max number of elements that fit into maintenance_work_mem
*/
static double
HnswGetMaxInMemoryElements(int m, double ml, int dimensions)
{
Size elementSize = sizeof(HnswElementData);
double avgLevel = -log(0.5) * ml;
elementSize += sizeof(HnswNeighborArray) * (avgLevel + 1);
elementSize += sizeof(HnswCandidate) * (m * (avgLevel + 2));
elementSize += sizeof(ItemPointerData);
elementSize += VECTOR_SIZE(dimensions);
return (maintenance_work_mem * 1024L) / elementSize;
}
/*
* Initialize the build state
*/
@@ -478,7 +436,7 @@ InitBuildState(HnswBuildState * buildstate, Relation heap, Relation index, Index
buildstate->entryPoint = NULL;
buildstate->ml = HnswGetMl(buildstate->m);
buildstate->maxLevel = HnswGetMaxLevel(buildstate->m);
buildstate->memoryLeft = maintenance_work_mem * 1024L;
buildstate->maxInMemoryElements = HnswGetMaxInMemoryElements(buildstate->m, buildstate->ml, buildstate->dimensions);
buildstate->flushed = false;
/* Reuse for each tuple */
@@ -487,9 +445,6 @@ InitBuildState(HnswBuildState * buildstate, Relation heap, Relation index, Index
buildstate->tmpCtx = AllocSetContextCreate(CurrentMemoryContext,
"Hnsw build temporary context",
ALLOCSET_DEFAULT_SIZES);
buildstate->hnswleader = NULL;
buildstate->hnswshared = NULL;
}
/*
@@ -502,392 +457,23 @@ FreeBuildState(HnswBuildState * buildstate)
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
*/
static void
BuildGraph(HnswBuildState * buildstate, ForkNumber forkNum)
{
int parallel_workers = 0;
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
buildstate->reltuples = table_index_build_scan(buildstate->heap, buildstate->index, buildstate->indexInfo,
true, true, BuildCallback, (void *) buildstate, NULL);
buildstate->reltuples = table_index_build_scan(buildstate->heap, buildstate->index, buildstate->indexInfo,
true, true, BuildCallback, (void *) buildstate, NULL);
#else
buildstate->reltuples = IndexBuildHeapScan(buildstate->heap, buildstate->index, buildstate->indexInfo,
true, BuildCallback, (void *) buildstate, NULL);
buildstate->reltuples = IndexBuildHeapScan(buildstate->heap, buildstate->index, buildstate->indexInfo,
true, BuildCallback, (void *) buildstate, NULL);
#endif
}
/* End parallel build */
if (buildstate->hnswleader)
HnswEndParallel(buildstate->hnswleader);
}
#if PG_VERSION_NUM < 110008
void
log_newpage_range(Relation rel, ForkNumber forkNum, BlockNumber startblk, BlockNumber endblk, bool page_std)
{
for (BlockNumber blkno = startblk; blkno < endblk; blkno++)
{
Buffer buf = ReadBufferExtended(rel, forkNum, blkno, RBM_NORMAL, NULL);
LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
MarkBufferDirty(buf);
log_newpage_buffer(buf, page_std);
UnlockReleaseBuffer(buf);
}
}
#endif
/*
* Build the index
*/
@@ -903,9 +489,6 @@ BuildIndex(Relation heap, Relation index, IndexInfo *indexInfo,
if (!buildstate->flushed)
FlushPages(buildstate);
if (RelationNeedsWAL(index))
log_newpage_range(index, forkNum, 0, RelationGetNumberOfBlocks(index), true);
FreeBuildState(buildstate);
}

View File

@@ -92,7 +92,7 @@ HnswFreeOffset(Relation index, Buffer buf, Page page, HnswElement element, Size
* Add a new page
*/
static void
HnswInsertAppendPage(Relation index, Buffer *nbuf, Page *npage, GenericXLogState *state, Page page, bool building)
HnswInsertAppendPage(Relation index, Buffer *nbuf, Page *npage, GenericXLogState *state, Page page)
{
/* Add a new page */
LockRelationForExtension(index, ExclusiveLock);
@@ -100,11 +100,7 @@ HnswInsertAppendPage(Relation index, Buffer *nbuf, Page *npage, GenericXLogState
UnlockRelationForExtension(index, ExclusiveLock);
/* Init new page */
if (building)
*npage = BufferGetPage(*nbuf);
else
*npage = GenericXLogRegisterBuffer(state, *nbuf, GENERIC_XLOG_FULL_IMAGE);
*npage = GenericXLogRegisterBuffer(state, *nbuf, GENERIC_XLOG_FULL_IMAGE);
HnswInitPage(*nbuf, *npage);
/* Update previous buffer */
@@ -115,7 +111,7 @@ HnswInsertAppendPage(Relation index, Buffer *nbuf, Page *npage, GenericXLogState
* Add to element and neighbor pages
*/
static void
WriteNewElementPages(Relation index, HnswElement e, int m, BlockNumber insertPage, BlockNumber *updatedInsertPage, bool building)
WriteNewElementPages(Relation index, HnswElement e, int m, BlockNumber insertPage, BlockNumber *updatedInsertPage)
{
Buffer buf;
Page page;
@@ -127,6 +123,7 @@ WriteNewElementPages(Relation index, HnswElement e, int m, BlockNumber insertPag
Size minCombinedSize;
HnswElementTuple etup;
BlockNumber currentPage = insertPage;
int dimensions = e->vec->dim;
HnswNeighborTuple ntup;
Buffer nbuf;
Page npage;
@@ -135,7 +132,7 @@ WriteNewElementPages(Relation index, HnswElement e, int m, BlockNumber insertPag
BlockNumber newInsertPage = InvalidBlockNumber;
/* Calculate sizes */
etupSize = HNSW_ELEMENT_TUPLE_SIZE(VARSIZE_ANY(DatumGetPointer(e->value)));
etupSize = HNSW_ELEMENT_TUPLE_SIZE(dimensions);
ntupSize = HNSW_NEIGHBOR_TUPLE_SIZE(e->level, m);
combinedSize = etupSize + ntupSize + sizeof(ItemIdData);
maxSize = HNSW_MAX_SIZE;
@@ -155,16 +152,8 @@ WriteNewElementPages(Relation index, HnswElement e, int m, BlockNumber insertPag
buf = ReadBuffer(index, currentPage);
LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
if (building)
{
state = NULL;
page = BufferGetPage(buf);
}
else
{
state = GenericXLogStart(index);
page = GenericXLogRegisterBuffer(state, buf, 0);
}
state = GenericXLogStart(index);
page = GenericXLogRegisterBuffer(state, buf, 0);
/* Keep track of first page where element at level 0 can fit */
if (!BlockNumberIsValid(newInsertPage) && PageGetFreeSpace(page) >= minCombinedSize)
@@ -184,12 +173,7 @@ WriteNewElementPages(Relation index, HnswElement e, int m, BlockNumber insertPag
if (HnswFreeOffset(index, buf, page, e, ntupSize, &nbuf, &npage, &freeOffno, &freeNeighborOffno, &newInsertPage))
{
if (nbuf != buf)
{
if (building)
npage = BufferGetPage(nbuf);
else
npage = GenericXLogRegisterBuffer(state, nbuf, 0);
}
npage = GenericXLogRegisterBuffer(state, nbuf, 0);
break;
}
@@ -198,7 +182,7 @@ WriteNewElementPages(Relation index, HnswElement e, int m, BlockNumber insertPag
/* Skip if both tuples can fit on the same page */
if (combinedSize > maxSize && PageGetFreeSpace(page) >= etupSize && !BlockNumberIsValid(HnswPageGetOpaque(page)->nextblkno))
{
HnswInsertAppendPage(index, &nbuf, &npage, state, page, building);
HnswInsertAppendPage(index, &nbuf, &npage, state, page);
break;
}
@@ -207,8 +191,7 @@ WriteNewElementPages(Relation index, HnswElement e, int m, BlockNumber insertPag
if (BlockNumberIsValid(currentPage))
{
/* Move to next page */
if (!building)
GenericXLogAbort(state);
GenericXLogAbort(state);
UnlockReleaseBuffer(buf);
}
else
@@ -216,31 +199,22 @@ WriteNewElementPages(Relation index, HnswElement e, int m, BlockNumber insertPag
Buffer newbuf;
Page newpage;
HnswInsertAppendPage(index, &newbuf, &newpage, state, page, building);
HnswInsertAppendPage(index, &newbuf, &newpage, state, page);
/* Commit */
if (!building)
GenericXLogFinish(state);
GenericXLogFinish(state);
/* Unlock previous buffer */
UnlockReleaseBuffer(buf);
/* Prepare new buffer */
state = GenericXLogStart(index);
buf = newbuf;
if (building)
{
state = NULL;
page = BufferGetPage(buf);
}
else
{
state = GenericXLogStart(index);
page = GenericXLogRegisterBuffer(state, buf, 0);
}
page = GenericXLogRegisterBuffer(state, buf, 0);
/* Create new page for neighbors if needed */
if (PageGetFreeSpace(page) < combinedSize)
HnswInsertAppendPage(index, &nbuf, &npage, state, page, building);
HnswInsertAppendPage(index, &nbuf, &npage, state, page);
else
{
nbuf = buf;
@@ -294,8 +268,7 @@ WriteNewElementPages(Relation index, HnswElement e, int m, BlockNumber insertPag
}
/* Commit */
if (!building)
GenericXLogFinish(state);
GenericXLogFinish(state);
UnlockReleaseBuffer(buf);
if (nbuf != buf)
UnlockReleaseBuffer(nbuf);
@@ -329,7 +302,7 @@ ConnectionExists(HnswElement e, HnswNeighborTuple ntup, int startIdx, int lm)
* Update neighbors
*/
void
HnswUpdateNeighborPages(Relation index, FmgrInfo *procinfo, Oid collation, HnswElement e, int m, bool checkExisting, bool building)
HnswUpdateNeighborPages(Relation index, FmgrInfo *procinfo, Oid collation, HnswElement e, int m, bool checkExisting)
{
for (int lc = e->level; lc >= 0; lc--)
{
@@ -370,16 +343,8 @@ HnswUpdateNeighborPages(Relation index, FmgrInfo *procinfo, Oid collation, HnswE
/* Register page */
buf = ReadBuffer(index, hc->element->neighborPage);
LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
if (building)
{
state = NULL;
page = BufferGetPage(buf);
}
else
{
state = GenericXLogStart(index);
page = GenericXLogRegisterBuffer(state, buf, 0);
}
state = GenericXLogStart(index);
page = GenericXLogRegisterBuffer(state, buf, 0);
/* Get tuple */
itemid = PageGetItemId(page, offno);
@@ -421,10 +386,9 @@ HnswUpdateNeighborPages(Relation index, FmgrInfo *procinfo, Oid collation, HnswE
elog(ERROR, "failed to add index item to \"%s\"", RelationGetRelationName(index));
/* Commit */
if (!building)
GenericXLogFinish(state);
GenericXLogFinish(state);
}
else if (!building)
else
GenericXLogAbort(state);
UnlockReleaseBuffer(buf);
@@ -436,34 +400,23 @@ HnswUpdateNeighborPages(Relation index, FmgrInfo *procinfo, Oid collation, HnswE
* Add a heap TID to an existing element
*/
static bool
HnswAddDuplicate(Relation index, HnswElement element, HnswElement dup, bool building)
HnswAddDuplicate(Relation index, HnswElement element, HnswElement dup)
{
Buffer buf;
Page page;
GenericXLogState *state;
ItemId itemid;
Size etupSize = HNSW_ELEMENT_TUPLE_SIZE(dup->vec->dim);
HnswElementTuple etup;
Size etupSize;
int i;
/* Read page */
buf = ReadBuffer(index, dup->blkno);
LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
if (building)
{
state = NULL;
page = BufferGetPage(buf);
}
else
{
state = GenericXLogStart(index);
page = GenericXLogRegisterBuffer(state, buf, 0);
}
state = GenericXLogStart(index);
page = GenericXLogRegisterBuffer(state, buf, 0);
/* Find space */
itemid = PageGetItemId(page, dup->offno);
etup = (HnswElementTuple) PageGetItem(page, itemid);
etupSize = ItemIdGetLength(itemid);
etup = (HnswElementTuple) PageGetItem(page, PageGetItemId(page, dup->offno));
for (i = 0; i < HNSW_HEAPTIDS; i++)
{
if (!ItemPointerIsValid(&etup->heaptids[i]))
@@ -473,8 +426,7 @@ HnswAddDuplicate(Relation index, HnswElement element, HnswElement dup, bool buil
/* Either being deleted or we lost our chance to another backend */
if (i == 0 || i == HNSW_HEAPTIDS)
{
if (!building)
GenericXLogAbort(state);
GenericXLogAbort(state);
UnlockReleaseBuffer(buf);
return false;
}
@@ -487,8 +439,7 @@ HnswAddDuplicate(Relation index, HnswElement element, HnswElement dup, bool buil
elog(ERROR, "failed to add index item to \"%s\"", RelationGetRelationName(index));
/* Commit */
if (!building)
GenericXLogFinish(state);
GenericXLogFinish(state);
UnlockReleaseBuffer(buf);
return true;
@@ -498,37 +449,37 @@ HnswAddDuplicate(Relation index, HnswElement element, HnswElement dup, bool buil
* Write changes to disk
*/
static void
WriteElement(Relation index, FmgrInfo *procinfo, Oid collation, HnswElement element, int m, int efConstruction, HnswElement dup, HnswElement entryPoint, bool building)
WriteElement(Relation index, FmgrInfo *procinfo, Oid collation, HnswElement element, int m, int efConstruction, HnswElement dup, HnswElement entryPoint)
{
BlockNumber newInsertPage = InvalidBlockNumber;
/* Try to add to existing page */
if (dup != NULL)
{
if (HnswAddDuplicate(index, element, dup, building))
if (HnswAddDuplicate(index, element, dup))
return;
}
/* Write element and neighbor tuples */
WriteNewElementPages(index, element, m, GetInsertPage(index), &newInsertPage, building);
WriteNewElementPages(index, element, m, GetInsertPage(index), &newInsertPage);
/* Update insert page if needed */
if (BlockNumberIsValid(newInsertPage))
HnswUpdateMetaPage(index, 0, NULL, newInsertPage, MAIN_FORKNUM, building);
HnswUpdateMetaPage(index, 0, NULL, newInsertPage, MAIN_FORKNUM);
/* Update neighbors */
HnswUpdateNeighborPages(index, procinfo, collation, element, m, false, building);
HnswUpdateNeighborPages(index, procinfo, collation, element, m, false);
/* Update metapage if needed */
if (entryPoint == NULL || element->level > entryPoint->level)
HnswUpdateMetaPage(index, HNSW_UPDATE_ENTRY_GREATER, element, InvalidBlockNumber, MAIN_FORKNUM, building);
HnswUpdateMetaPage(index, HNSW_UPDATE_ENTRY_GREATER, element, InvalidBlockNumber, MAIN_FORKNUM);
}
/*
* Insert a tuple into the index
*/
bool
HnswInsertTuple(Relation index, Datum *values, bool *isnull, ItemPointer heap_tid, Relation heapRel, bool building)
HnswInsertTuple(Relation index, Datum *values, bool *isnull, ItemPointer heap_tid, Relation heapRel)
{
Datum value;
FmgrInfo *normprocinfo;
@@ -564,7 +515,7 @@ HnswInsertTuple(Relation index, Datum *values, bool *isnull, ItemPointer heap_ti
/* Create an element */
element = HnswInitElement(heap_tid, m, HnswGetMl(m), HnswGetMaxLevel(m));
element->value = value;
element->vec = DatumGetVector(value);
/* Prevent concurrent inserts when likely updating entry point */
if (entryPoint == NULL || element->level > entryPoint->level)
@@ -587,7 +538,7 @@ HnswInsertTuple(Relation index, Datum *values, bool *isnull, ItemPointer heap_ti
dup = HnswFindDuplicate(element);
/* Write to disk */
WriteElement(index, procinfo, collation, element, m, efConstruction, dup, entryPoint, building);
WriteElement(index, procinfo, collation, element, m, efConstruction, dup, entryPoint);
/* Release lock */
UnlockPage(index, HNSW_UPDATE_LOCK, lockmode);
@@ -621,7 +572,7 @@ hnswinsert(Relation index, Datum *values, bool *isnull, ItemPointer heap_tid,
oldCtx = MemoryContextSwitchTo(insertCtx);
/* Insert tuple */
HnswInsertTuple(index, values, isnull, heap_tid, heap, false);
HnswInsertTuple(index, values, isnull, heap_tid, heap);
/* Delete memory context */
MemoryContextSwitchTo(oldCtx);

View File

@@ -4,92 +4,8 @@
#include "hnsw.h"
#include "storage/bufmgr.h"
#include "utils/datum.h"
#include "vector.h"
#if PG_VERSION_NUM >= 130000
#include "common/hashfn.h"
#else
#include "utils/hashutils.h"
#endif
#if PG_VERSION_NUM < 170000
static inline uint64
murmurhash64(uint64 data)
{
uint64 h = data;
h ^= h >> 33;
h *= 0xff51afd7ed558ccd;
h ^= h >> 33;
h *= 0xc4ceb9fe1a85ec53;
h ^= h >> 33;
return h;
}
#endif
/* TID hash table */
static uint32
hash_tid(ItemPointerData tid)
{
union
{
uint64 i;
ItemPointerData tid;
} x;
/* Initialize unused bytes */
x.i = 0;
x.tid = tid;
return murmurhash64(x.i);
}
#define SH_PREFIX tidhash
#define SH_ELEMENT_TYPE TidHashEntry
#define SH_KEY_TYPE ItemPointerData
#define SH_KEY tid
#define SH_HASH_KEY(tb, key) hash_tid(key)
#define SH_EQUAL(tb, a, b) ItemPointerEquals(&a, &b)
#define SH_SCOPE extern
#define SH_DEFINE
#include "lib/simplehash.h"
/* Needed to include simplehash.h twice */
#if PG_VERSION_NUM < 120000
#undef SH_EQUAL
#define sh_log2 pointerhash_sh_log2
#define sh_pow2 pointerhash_sh_pow2
#endif
/* Pointer hash table */
static uint32
hash_pointer(uintptr_t ptr)
{
#if SIZEOF_VOID_P == 8
return murmurhash64((uint64) ptr);
#else
return murmurhash32((uint32) ptr);
#endif
}
#define SH_PREFIX pointerhash
#define SH_ELEMENT_TYPE PointerHashEntry
#define SH_KEY_TYPE uintptr_t
#define SH_KEY ptr
#define SH_HASH_KEY(tb, key) hash_pointer(key)
#define SH_EQUAL(tb, a, b) (a == b)
#define SH_SCOPE extern
#define SH_DEFINE
#include "lib/simplehash.h"
typedef union
{
pointerhash_hash *pointers;
tidhash_hash *tids;
} visited_hash;
/*
* Get the max number of connections in an upper layer for each element in the index
*/
@@ -184,6 +100,27 @@ HnswInitPage(Buffer buf, Page page)
HnswPageGetOpaque(page)->page_id = HNSW_PAGE_ID;
}
/*
* Init and register page
*/
void
HnswInitRegisterPage(Relation index, Buffer *buf, Page *page, GenericXLogState **state)
{
*state = GenericXLogStart(index);
*page = GenericXLogRegisterBuffer(*state, *buf, GENERIC_XLOG_FULL_IMAGE);
HnswInitPage(*buf, *page);
}
/*
* Commit buffer
*/
void
HnswCommitBuffer(Buffer buf, GenericXLogState *state)
{
GenericXLogFinish(state);
UnlockReleaseBuffer(buf);
}
/*
* Allocate neighbors
*/
@@ -239,8 +176,6 @@ HnswInitElement(ItemPointer heaptid, int m, double ml, int maxLevel)
HnswInitNeighbors(element, m);
element->value = PointerGetDatum(NULL);
return element;
}
@@ -252,8 +187,7 @@ HnswFreeElement(HnswElement element)
{
HnswFreeNeighbors(element);
list_free_deep(element->heaptids);
if (DatumGetPointer(element->value))
pfree(DatumGetPointer(element->value));
pfree(element->vec);
pfree(element);
}
@@ -280,7 +214,7 @@ HnswInitElementFromBlock(BlockNumber blkno, OffsetNumber offno)
element->blkno = blkno;
element->offno = offno;
element->neighbors = NULL;
element->value = PointerGetDatum(NULL);
element->vec = NULL;
return element;
}
@@ -358,7 +292,7 @@ HnswUpdateMetaPageInfo(Page page, int updateEntry, HnswElement entryPoint, Block
* Update the metapage
*/
void
HnswUpdateMetaPage(Relation index, int updateEntry, HnswElement entryPoint, BlockNumber insertPage, ForkNumber forkNum, bool building)
HnswUpdateMetaPage(Relation index, int updateEntry, HnswElement entryPoint, BlockNumber insertPage, ForkNumber forkNum)
{
Buffer buf;
Page page;
@@ -366,22 +300,12 @@ HnswUpdateMetaPage(Relation index, int updateEntry, HnswElement entryPoint, Bloc
buf = ReadBufferExtended(index, forkNum, HNSW_METAPAGE_BLKNO, RBM_NORMAL, NULL);
LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
if (building)
{
state = NULL;
page = BufferGetPage(buf);
}
else
{
state = GenericXLogStart(index);
page = GenericXLogRegisterBuffer(state, buf, 0);
}
state = GenericXLogStart(index);
page = GenericXLogRegisterBuffer(state, buf, 0);
HnswUpdateMetaPageInfo(page, updateEntry, entryPoint, insertPage);
if (!building)
GenericXLogFinish(state);
UnlockReleaseBuffer(buf);
HnswCommitBuffer(buf, state);
}
/*
@@ -400,7 +324,7 @@ HnswSetElementTuple(HnswElementTuple etup, HnswElement element)
else
ItemPointerSetInvalid(&etup->heaptids[i]);
}
memcpy(&etup->data, DatumGetPointer(element->value), VARSIZE_ANY(DatumGetPointer(element->value)));
memcpy(&etup->vec, element->vec, VECTOR_SIZE(element->vec->dim));
}
/*
@@ -522,7 +446,10 @@ HnswLoadElementFromTuple(HnswElement element, HnswElementTuple etup, bool loadHe
}
if (loadVec)
element->value = datumCopy(PointerGetDatum(&etup->data), false, -1);
{
element->vec = palloc(VECTOR_SIZE(etup->vec.dim));
memcpy(element->vec, &etup->vec, VECTOR_SIZE(etup->vec.dim));
}
}
/*
@@ -549,7 +476,7 @@ HnswLoadElement(HnswElement element, float *distance, Datum *q, Relation index,
/* Calculate distance */
if (distance != NULL)
*distance = (float) DatumGetFloat8(FunctionCall2Coll(procinfo, collation, *q, PointerGetDatum(&etup->data)));
*distance = (float) DatumGetFloat8(FunctionCall2Coll(procinfo, collation, *q, PointerGetDatum(&etup->vec)));
UnlockReleaseBuffer(buf);
}
@@ -560,7 +487,7 @@ HnswLoadElement(HnswElement element, float *distance, Datum *q, Relation index,
static float
GetCandidateDistance(HnswCandidate * hc, Datum q, FmgrInfo *procinfo, Oid collation)
{
return DatumGetFloat8(FunctionCall2Coll(procinfo, collation, q, hc->element->value));
return DatumGetFloat8(FunctionCall2Coll(procinfo, collation, q, PointerGetDatum(hc->element->vec)));
}
/*
@@ -625,22 +552,16 @@ CreatePairingHeapNode(HnswCandidate * c)
* Add to visited
*/
static inline void
AddToVisited(visited_hash v, HnswCandidate * hc, Relation index, bool *found)
AddToVisited(HTAB *v, HnswCandidate * hc, Relation index, bool *found)
{
if (index == NULL)
{
#if PG_VERSION_NUM >= 130000
pointerhash_insert_hash(v.pointers, (uintptr_t) hc->element, hc->element->hash, found);
#else
pointerhash_insert(v.pointers, (uintptr_t) hc->element, found);
#endif
}
hash_search(v, &hc->element, HASH_ENTER, found);
else
{
ItemPointerData indextid;
ItemPointerSet(&indextid, hc->element->blkno, hc->element->offno);
tidhash_insert(v.tids, indextid, found);
hash_search(v, &indextid, HASH_ENTER, found);
}
}
@@ -656,21 +577,30 @@ 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;
visited_hash v;
HASHCTL hash_ctl;
HTAB *v;
/* Create hash table */
if (index == NULL)
v.pointers = pointerhash_create(CurrentMemoryContext, ef * m * 2, NULL);
{
hash_ctl.keysize = sizeof(HnswElement *);
hash_ctl.entrysize = sizeof(HnswElement *);
}
else
v.tids = tidhash_create(CurrentMemoryContext, ef * m * 2, NULL);
{
hash_ctl.keysize = sizeof(ItemPointerData);
hash_ctl.entrysize = sizeof(ItemPointerData);
}
hash_ctl.hcxt = CurrentMemoryContext;
v = hash_create("hnsw visited", 256, &hash_ctl, HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
/* Add entry points to v, C, and W */
foreach(lc2, ep)
{
HnswCandidate *hc = (HnswCandidate *) lfirst(lc2);
bool found;
AddToVisited(v, hc, index, &found);
AddToVisited(v, hc, index, NULL);
pairingheap_add(C, &(CreatePairingHeapNode(hc)->ph_node));
pairingheap_add(W, &(CreatePairingHeapNode(hc)->ph_node));
@@ -820,7 +750,7 @@ HnswGetDistance(HnswElement a, HnswElement b, int lc, FmgrInfo *procinfo, Oid co
}
}
return DatumGetFloat8(FunctionCall2Coll(procinfo, collation, a->value, b->value));
return DatumGetFloat8(FunctionCall2Coll(procinfo, collation, PointerGetDatum(a->vec), PointerGetDatum(b->vec)));
}
/*
@@ -947,7 +877,7 @@ HnswFindDuplicate(HnswElement e)
HnswCandidate *neighbor = &neighbors->items[i];
/* Exit early since ordered by distance */
if (!datumIsEqual(e->value, neighbor->element->value, false, -1))
if (vector_cmp_internal(e->vec, neighbor->element->vec) != 0)
break;
/* Check for space */
@@ -1000,13 +930,13 @@ HnswUpdateConnection(HnswElement element, HnswCandidate * hc, int m, int lc, int
/* Load elements on insert */
if (index != NULL)
{
Datum q = hc->element->value;
Datum q = PointerGetDatum(hc->element->vec);
for (int i = 0; i < currentNeighbors->length; i++)
{
HnswCandidate *hc3 = &currentNeighbors->items[i];
if (DatumGetPointer(hc3->element->value) == NULL)
if (hc3->element->vec == NULL)
HnswLoadElement(hc3->element, &hc3->distance, &q, index, procinfo, collation, true);
else
hc3->distance = GetCandidateDistance(hc3, q, procinfo, collation);
@@ -1087,15 +1017,9 @@ HnswInsertElement(HnswElement element, HnswElement entryPoint, Relation index, F
List *w;
int level = element->level;
int entryLevel;
Datum q = element->value;
Datum q = PointerGetDatum(element->vec);
HnswElement skipElement = existing ? element : NULL;
#if PG_VERSION_NUM >= 130000
/* Precompute hash */
if (index == NULL)
element->hash = hash_pointer((uintptr_t) element);
#endif
/* No neighbors if no entry point */
if (entryPoint == NULL)
return;

View File

@@ -12,9 +12,12 @@
* Check if deleted list contains an index TID
*/
static bool
DeletedContains(tidhash_hash * deleted, ItemPointer indextid)
DeletedContains(HTAB *deleted, ItemPointer indextid)
{
return tidhash_lookup(deleted, *indextid) != NULL;
bool found;
hash_search(deleted, indextid, HASH_FIND, &found);
return found;
}
/*
@@ -59,8 +62,7 @@ RemoveHeapTids(HnswVacuumState * vacuumstate)
/* Iterate over nodes */
for (offno = FirstOffsetNumber; offno <= maxoffno; offno = OffsetNumberNext(offno))
{
ItemId itemid = PageGetItemId(page, offno);
HnswElementTuple etup = (HnswElementTuple) PageGetItem(page, itemid);
HnswElementTuple etup = (HnswElementTuple) PageGetItem(page, PageGetItemId(page, offno));
int idx = 0;
bool itemUpdated = false;
@@ -91,7 +93,7 @@ RemoveHeapTids(HnswVacuumState * vacuumstate)
if (itemUpdated)
{
Size etupSize = ItemIdGetLength(itemid);
Size etupSize = HNSW_ELEMENT_TUPLE_SIZE(etup->vec.dim);
/* Mark rest as invalid */
for (int i = idx; i < HNSW_HEAPTIDS; i++)
@@ -107,13 +109,11 @@ RemoveHeapTids(HnswVacuumState * vacuumstate)
if (!ItemPointerIsValid(&etup->heaptids[0]))
{
ItemPointerData ip;
bool found;
/* Add to deleted list */
ItemPointerSet(&ip, blkno, offno);
tidhash_insert(vacuumstate->deleted, ip, &found);
Assert(!found);
(void) hash_search(vacuumstate->deleted, &ip, HASH_ENTER, NULL);
}
else if (etup->level > highestLevel && !(entryPoint != NULL && blkno == entryPoint->blkno && offno == entryPoint->offno))
{
@@ -230,7 +230,7 @@ RepairGraphElement(HnswVacuumState * vacuumstate, HnswElement element, HnswEleme
UnlockReleaseBuffer(buf);
/* Update neighbors */
HnswUpdateNeighborPages(index, procinfo, collation, element, m, true, false);
HnswUpdateNeighborPages(index, procinfo, collation, element, m, true);
}
/*
@@ -286,7 +286,7 @@ RepairGraphEntryPoint(HnswVacuumState * vacuumstate)
* point is outdated and empty, the entry point will be empty
* until an element is repaired.
*/
HnswUpdateMetaPage(index, HNSW_UPDATE_ENTRY_ALWAYS, highestPoint, InvalidBlockNumber, MAIN_FORKNUM, false);
HnswUpdateMetaPage(index, HNSW_UPDATE_ENTRY_ALWAYS, highestPoint, InvalidBlockNumber, MAIN_FORKNUM);
}
else
{
@@ -419,7 +419,7 @@ RepairGraph(HnswVacuumState * vacuumstate)
* was replaced and highest point was outdated.
*/
if (entryPoint == NULL || element->level > entryPoint->level)
HnswUpdateMetaPage(index, HNSW_UPDATE_ENTRY_GREATER, element, InvalidBlockNumber, MAIN_FORKNUM, false);
HnswUpdateMetaPage(index, HNSW_UPDATE_ENTRY_GREATER, element, InvalidBlockNumber, MAIN_FORKNUM);
/* Release lock */
UnlockPage(index, HNSW_UPDATE_LOCK, lockmode);
@@ -477,8 +477,7 @@ MarkDeleted(HnswVacuumState * vacuumstate)
/* Update element and neighbors together */
for (offno = FirstOffsetNumber; offno <= maxoffno; offno = OffsetNumberNext(offno))
{
ItemId itemid = PageGetItemId(page, offno);
HnswElementTuple etup = (HnswElementTuple) PageGetItem(page, itemid);
HnswElementTuple etup = (HnswElementTuple) PageGetItem(page, PageGetItemId(page, offno));
HnswNeighborTuple ntup;
Size etupSize;
Size ntupSize;
@@ -506,7 +505,7 @@ MarkDeleted(HnswVacuumState * vacuumstate)
continue;
/* Calculate sizes */
etupSize = ItemIdGetLength(itemid);
etupSize = HNSW_ELEMENT_TUPLE_SIZE(etup->vec.dim);
ntupSize = HNSW_NEIGHBOR_TUPLE_SIZE(etup->level, vacuumstate->m);
/* Get neighbor page */
@@ -529,7 +528,7 @@ MarkDeleted(HnswVacuumState * vacuumstate)
/* Overwrite element */
etup->deleted = 1;
MemSet(&etup->data, 0, VARSIZE_ANY(&etup->data));
MemSet(&etup->vec.x, 0, etup->vec.dim * sizeof(float));
/* Overwrite neighbors */
for (int i = 0; i < ntup->count; i++)
@@ -564,7 +563,7 @@ MarkDeleted(HnswVacuumState * vacuumstate)
}
/* Update insert page last, after everything has been marked as deleted */
HnswUpdateMetaPage(index, 0, NULL, insertPage, MAIN_FORKNUM, false);
HnswUpdateMetaPage(index, 0, NULL, insertPage, MAIN_FORKNUM);
}
/*
@@ -574,6 +573,7 @@ static void
InitVacuumState(HnswVacuumState * vacuumstate, IndexVacuumInfo *info, IndexBulkDeleteResult *stats, IndexBulkDeleteCallback callback, void *callback_state)
{
Relation index = info->index;
HASHCTL hash_ctl;
if (stats == NULL)
stats = (IndexBulkDeleteResult *) palloc0(sizeof(IndexBulkDeleteResult));
@@ -595,7 +595,10 @@ InitVacuumState(HnswVacuumState * vacuumstate, IndexVacuumInfo *info, IndexBulkD
HnswGetMetaPageInfo(index, &vacuumstate->m, NULL);
/* Create hash table */
vacuumstate->deleted = tidhash_create(CurrentMemoryContext, 256, NULL);
hash_ctl.keysize = sizeof(ItemPointerData);
hash_ctl.entrysize = sizeof(ItemPointerData);
hash_ctl.hcxt = CurrentMemoryContext;
vacuumstate->deleted = hash_create("hnswbulkdelete indextids", 256, &hash_ctl, HASH_ELEM | HASH_BLOBS | HASH_CONTEXT);
}
/*
@@ -604,7 +607,7 @@ InitVacuumState(HnswVacuumState * vacuumstate, IndexVacuumInfo *info, IndexBulkD
static void
FreeVacuumState(HnswVacuumState * vacuumstate)
{
tidhash_destroy(vacuumstate->deleted);
hash_destroy(vacuumstate->deleted);
FreeAccessStrategy(vacuumstate->bas);
pfree(vacuumstate->ntup);
MemoryContextDelete(vacuumstate->tmpCtx);

705
src/svector.c Normal file
View File

@@ -0,0 +1,705 @@
#include "postgres.h"
#include <math.h>
#include "fmgr.h"
#include "libpq/pqformat.h"
#include "svector.h"
#include "utils/array.h"
#include "vector.h"
#if PG_VERSION_NUM >= 120000
#include "common/shortest_dec.h"
#include "utils/float.h"
#else
#include <float.h>
#include "utils/builtins.h"
#endif
/*
* Ensure same dimensions
*/
static inline void
CheckDims(SVector * a, SVector * b)
{
if (a->dim != b->dim)
ereport(ERROR,
(errcode(ERRCODE_DATA_EXCEPTION),
errmsg("different svector dimensions %d and %d", a->dim, b->dim)));
}
/*
* Ensure expected dimensions
*/
static inline void
CheckExpectedDim(int32 typmod, int dim)
{
if (typmod != -1 && typmod != dim)
ereport(ERROR,
(errcode(ERRCODE_DATA_EXCEPTION),
errmsg("expected %d dimensions, not %d", typmod, dim)));
}
/*
* Ensure valid dimensions
*/
static inline void
CheckDim(int dim)
{
if (dim < 1)
ereport(ERROR,
(errcode(ERRCODE_DATA_EXCEPTION),
errmsg("svector must have at least 1 dimension")));
if (dim > SVECTOR_MAX_DIM)
ereport(ERROR,
(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
errmsg("svector cannot have more than %d dimensions", SVECTOR_MAX_DIM)));
}
/*
* Ensure valid nnz
*/
static inline void
CheckNnz(int nnz, int dim)
{
if (nnz < 0)
ereport(ERROR,
(errcode(ERRCODE_DATA_EXCEPTION),
errmsg("svector must have at least one element")));
if (nnz > dim)
ereport(ERROR,
(errcode(ERRCODE_PROGRAM_LIMIT_EXCEEDED),
errmsg("svector cannot have more elements than dimensions")));
}
/*
* Ensure valid index
*/
static inline void
CheckIndex(int32 *indices, int i, int dim)
{
int32 index = indices[i];
if (index < 0)
ereport(ERROR,
(errcode(ERRCODE_DATA_EXCEPTION),
errmsg("index must not be negative")));
if (index >= dim)
ereport(ERROR,
(errcode(ERRCODE_DATA_EXCEPTION),
errmsg("index must be less than dimensions")));
if (i > 0)
{
if (index < indices[i - 1])
ereport(ERROR,
(errcode(ERRCODE_DATA_EXCEPTION),
errmsg("indexes must be in ascending order")));
if (index == indices[i - 1])
ereport(ERROR,
(errcode(ERRCODE_DATA_EXCEPTION),
errmsg("indexes must not contain duplicates")));
}
}
/*
* Ensure finite element
*/
static inline void
CheckElement(float value)
{
if (isnan(value))
ereport(ERROR,
(errcode(ERRCODE_DATA_EXCEPTION),
errmsg("NaN not allowed in svector")));
if (isinf(value))
ereport(ERROR,
(errcode(ERRCODE_DATA_EXCEPTION),
errmsg("infinite value not allowed in svector")));
}
/*
* Allocate and initialize a new sparse vector
*/
SVector *
InitSVector(int dim, int nnz)
{
SVector *result;
int size;
size = SVECTOR_SIZE(nnz);
result = (SVector *) palloc0(size);
SET_VARSIZE(result, size);
result->dim = dim;
result->nnz = nnz;
return result;
}
/*
* Convert textual representation to internal representation
*/
PGDLLEXPORT PG_FUNCTION_INFO_V1(svector_in);
Datum
svector_in(PG_FUNCTION_ARGS)
{
char *str = PG_GETARG_CSTRING(0);
int32 typmod = PG_GETARG_INT32(2);
int dim;
char *pt;
SVector *result;
float *rvalues;
char *lit = pstrdup(str);
int n;
int32 *indices;
float *values;
int index;
float value;
int maxNnz;
int nnz = 0;
/* TODO Improve code and checks after deciding on format */
maxNnz = 1;
pt = str;
while (*pt != '\0')
{
if (*pt == ',')
maxNnz++;
pt++;
}
maxNnz /= 2;
indices = palloc(maxNnz * sizeof(int32));
values = palloc(maxNnz * sizeof(float));
while (sscanf(str, "(%d,%f)%n", &index, &value, &n) == 2)
{
/* TODO Better error */
if (nnz == maxNnz)
ereport(ERROR,
(errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
errmsg("ran out of buffer: \"%s\"", lit)));
/* TODO Decide whether to store zero values */
indices[nnz] = index;
values[nnz] = value;
nnz++;
str += n;
if (*str == ',')
str++;
else if (*str == '|')
break;
else
ereport(ERROR,
(errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
errmsg("malformed svector literal: \"%s\"", lit)));
}
if (sscanf(str, "|%d|%n", &dim, &n) != 1)
ereport(ERROR,
(errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
errmsg("malformed svector literal: \"%s\"", lit)));
str += n;
if (*str != '\0')
ereport(ERROR,
(errcode(ERRCODE_INVALID_TEXT_REPRESENTATION),
errmsg("malformed svector literal: \"%s\"", lit),
errdetail("Junk after closing pipe.")));
pfree(lit);
CheckDim(dim);
CheckExpectedDim(typmod, dim);
result = InitSVector(dim, nnz);
rvalues = SVECTOR_VALUES(result);
for (int i = 0; i < nnz; i++)
{
result->indices[i] = indices[i];
rvalues[i] = values[i];
CheckIndex(result->indices, i, dim);
CheckElement(rvalues[i]);
}
PG_RETURN_POINTER(result);
}
/*
* Convert internal representation to textual representation
*/
PGDLLEXPORT PG_FUNCTION_INFO_V1(svector_out);
Datum
svector_out(PG_FUNCTION_ARGS)
{
SVector *svector = PG_GETARG_SVECTOR_P(0);
float *values = SVECTOR_VALUES(svector);
char *buf;
char *ptr;
int n;
/* TODO Improve code after deciding on format */
#if PG_VERSION_NUM < 120000
int ndig = FLT_DIG + extra_float_digits;
if (ndig < 1)
ndig = 1;
#define FLOAT_SHORTEST_DECIMAL_LEN (ndig + 10)
#endif
/* TODO Move */
#define APPEND_CHAR(ptr, ch) (*(ptr)++ = (ch))
/* TODO Improve */
buf = (char *) palloc((FLOAT_SHORTEST_DECIMAL_LEN + 20) * svector->nnz + 20);
ptr = buf;
for (int i = 0; i < svector->nnz; i++)
{
if (i > 0)
APPEND_CHAR(ptr, ',');
n = sprintf(ptr, "(%d,", svector->indices[i]);
ptr += n;
#if PG_VERSION_NUM >= 120000
n = float_to_shortest_decimal_bufn(values[i], ptr);
#else
n = sprintf(ptr, "%.*g", ndig, values[i]);
#endif
ptr += n;
APPEND_CHAR(ptr, ')');
}
n = sprintf(ptr, "|%d|", svector->dim);
ptr += n;
APPEND_CHAR(ptr, '\0');
PG_FREE_IF_COPY(svector, 0);
PG_RETURN_CSTRING(buf);
}
/*
* Convert type modifier
*/
PGDLLEXPORT PG_FUNCTION_INFO_V1(svector_typmod_in);
Datum
svector_typmod_in(PG_FUNCTION_ARGS)
{
ArrayType *ta = PG_GETARG_ARRAYTYPE_P(0);
int32 *tl;
int n;
tl = ArrayGetIntegerTypmods(ta, &n);
if (n != 1)
ereport(ERROR,
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
errmsg("invalid type modifier")));
if (*tl < 1)
ereport(ERROR,
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
errmsg("dimensions for type svector must be at least 1")));
if (*tl > SVECTOR_MAX_DIM)
ereport(ERROR,
(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
errmsg("dimensions for type svector cannot exceed %d", SVECTOR_MAX_DIM)));
PG_RETURN_INT32(*tl);
}
/*
* Convert external binary representation to internal representation
*/
PGDLLEXPORT PG_FUNCTION_INFO_V1(svector_recv);
Datum
svector_recv(PG_FUNCTION_ARGS)
{
StringInfo buf = (StringInfo) PG_GETARG_POINTER(0);
int32 typmod = PG_GETARG_INT32(2);
SVector *result;
int32 dim;
int32 nnz;
int32 unused;
float *values;
dim = pq_getmsgint(buf, sizeof(int32));
nnz = pq_getmsgint(buf, sizeof(int32));
unused = pq_getmsgint(buf, sizeof(int32));
CheckDim(dim);
CheckNnz(nnz, dim);
CheckExpectedDim(typmod, dim);
if (unused != 0)
ereport(ERROR,
(errcode(ERRCODE_DATA_EXCEPTION),
errmsg("expected unused to be 0, not %d", unused)));
result = InitSVector(dim, nnz);
values = SVECTOR_VALUES(result);
for (int i = 0; i < nnz; i++)
{
result->indices[i] = pq_getmsgint(buf, sizeof(int32));
CheckIndex(result->indices, i, dim);
}
for (int i = 0; i < nnz; i++)
{
values[i] = pq_getmsgfloat4(buf);
CheckElement(values[i]);
}
PG_RETURN_POINTER(result);
}
/*
* Convert internal representation to the external binary representation
*/
PGDLLEXPORT PG_FUNCTION_INFO_V1(svector_send);
Datum
svector_send(PG_FUNCTION_ARGS)
{
SVector *svec = PG_GETARG_SVECTOR_P(0);
float *values = SVECTOR_VALUES(svec);
StringInfoData buf;
pq_begintypsend(&buf);
pq_sendint(&buf, svec->dim, sizeof(int32));
pq_sendint(&buf, svec->nnz, sizeof(int32));
pq_sendint(&buf, svec->unused, sizeof(int32));
for (int i = 0; i < svec->nnz; i++)
pq_sendint(&buf, svec->indices[i], sizeof(int32));
for (int i = 0; i < svec->nnz; i++)
pq_sendfloat4(&buf, values[i]);
PG_RETURN_BYTEA_P(pq_endtypsend(&buf));
}
/*
* Convert sparse vector to sparse vector
* This is needed to check the type modifier
*/
PGDLLEXPORT PG_FUNCTION_INFO_V1(svector);
Datum
svector(PG_FUNCTION_ARGS)
{
SVector *svec = PG_GETARG_SVECTOR_P(0);
int32 typmod = PG_GETARG_INT32(1);
CheckExpectedDim(typmod, svec->dim);
PG_RETURN_POINTER(svec);
}
/*
* Convert dense vector to sparse vector
*/
PGDLLEXPORT PG_FUNCTION_INFO_V1(vector_to_svector);
Datum
vector_to_svector(PG_FUNCTION_ARGS)
{
Vector *vec = PG_GETARG_VECTOR_P(0);
int32 typmod = PG_GETARG_INT32(1);
SVector *result;
int dim = vec->dim;
int nnz = 0;
float *values;
int j = 0;
CheckDim(dim);
CheckExpectedDim(typmod, dim);
for (int i = 0; i < dim; i++)
{
if (vec->x[i] != 0)
nnz++;
}
result = InitSVector(dim, nnz);
values = SVECTOR_VALUES(result);
for (int i = 0; i < dim; i++)
{
if (vec->x[i] != 0)
{
/* Safety check */
if (j == nnz)
elog(ERROR, "safety check failed");
result->indices[j] = i;
values[j] = vec->x[i];
j++;
}
}
PG_RETURN_POINTER(result);
}
/*
* Get the L2 squared distance between sparse vectors
*/
static double
l2_distance_squared_internal(SVector * a, SVector * b)
{
float *ax = SVECTOR_VALUES(a);
float *bx = SVECTOR_VALUES(b);
double distance = 0.0;
int bpos = 0;
for (int i = 0; i < a->nnz; i++)
{
int ai = a->indices[i];
int bi = -1;
for (int j = bpos; j < b->nnz; j++)
{
bi = b->indices[j];
if (ai == bi)
{
double diff = ax[i] - bx[j];
distance += diff * diff;
}
else if (ai > bi)
distance += bx[j] * bx[j];
/* Update start for next iteration */
if (ai >= bi)
bpos = j + 1;
/* Found or passed it */
if (bi >= ai)
break;
}
if (ai != bi)
distance += ax[i] * ax[i];
}
for (int j = bpos; j < b->nnz; j++)
distance += bx[j] * bx[j];
return distance;
}
/*
* Get the L2 distance between sparse vectors
*/
PGDLLEXPORT PG_FUNCTION_INFO_V1(svector_l2_distance);
Datum
svector_l2_distance(PG_FUNCTION_ARGS)
{
SVector *a = PG_GETARG_SVECTOR_P(0);
SVector *b = PG_GETARG_SVECTOR_P(1);
CheckDims(a, b);
PG_RETURN_FLOAT8(sqrt(l2_distance_squared_internal(a, b)));
}
/*
* Get the L2 squared distance between sparse vectors
* This saves a sqrt calculation
*/
PGDLLEXPORT PG_FUNCTION_INFO_V1(svector_l2_squared_distance);
Datum
svector_l2_squared_distance(PG_FUNCTION_ARGS)
{
SVector *a = PG_GETARG_SVECTOR_P(0);
SVector *b = PG_GETARG_SVECTOR_P(1);
CheckDims(a, b);
PG_RETURN_FLOAT8(l2_distance_squared_internal(a, b));
}
/*
* Get the inner product of two sparse vectors
*/
static double
inner_product_internal(SVector * a, SVector * b)
{
float *ax = SVECTOR_VALUES(a);
float *bx = SVECTOR_VALUES(b);
double distance = 0.0;
int bpos = 0;
for (int i = 0; i < a->nnz; i++)
{
int ai = a->indices[i];
for (int j = bpos; j < b->nnz; j++)
{
int bi = b->indices[j];
/* Only update when the same index */
if (ai == bi)
distance += ax[i] * bx[j];
/* Update start for next iteration */
if (ai >= bi)
bpos = j + 1;
/* Found or passed it */
if (bi >= ai)
break;
}
}
return distance;
}
/*
* Get the inner product of two sparse vectors
*/
PGDLLEXPORT PG_FUNCTION_INFO_V1(svector_inner_product);
Datum
svector_inner_product(PG_FUNCTION_ARGS)
{
SVector *a = PG_GETARG_SVECTOR_P(0);
SVector *b = PG_GETARG_SVECTOR_P(1);
CheckDims(a, b);
PG_RETURN_FLOAT8(inner_product_internal(a, b));
}
/*
* Get the negative inner product of two sparse vectors
*/
PGDLLEXPORT PG_FUNCTION_INFO_V1(svector_negative_inner_product);
Datum
svector_negative_inner_product(PG_FUNCTION_ARGS)
{
SVector *a = PG_GETARG_SVECTOR_P(0);
SVector *b = PG_GETARG_SVECTOR_P(1);
CheckDims(a, b);
PG_RETURN_FLOAT8(-inner_product_internal(a, b));
}
/*
* Get the cosine distance between two sparse vectors
*/
PGDLLEXPORT PG_FUNCTION_INFO_V1(svector_cosine_distance);
Datum
svector_cosine_distance(PG_FUNCTION_ARGS)
{
SVector *a = PG_GETARG_SVECTOR_P(0);
SVector *b = PG_GETARG_SVECTOR_P(1);
float *ax = SVECTOR_VALUES(a);
float *bx = SVECTOR_VALUES(b);
float norma = 0.0;
float normb = 0.0;
double similarity;
CheckDims(a, b);
similarity = inner_product_internal(a, b);
/* Auto-vectorized */
for (int i = 0; i < a->nnz; i++)
norma += ax[i] * ax[i];
/* Auto-vectorized */
for (int i = 0; i < b->nnz; i++)
normb += bx[i] * bx[i];
/* Use sqrt(a * b) over sqrt(a) * sqrt(b) */
similarity /= sqrt((double) norma * (double) normb);
#ifdef _MSC_VER
/* /fp:fast may not propagate NaN */
if (isnan(similarity))
PG_RETURN_FLOAT8(NAN);
#endif
/* Keep in range */
if (similarity > 1)
similarity = 1.0;
else if (similarity < -1)
similarity = -1.0;
PG_RETURN_FLOAT8(1.0 - similarity);
}
/*
* Get the weighted Jaccard distance between two sparse vectors
*/
PGDLLEXPORT PG_FUNCTION_INFO_V1(svector_jaccard_distance);
Datum
svector_jaccard_distance(PG_FUNCTION_ARGS)
{
SVector *a = PG_GETARG_SVECTOR_P(0);
SVector *b = PG_GETARG_SVECTOR_P(1);
float *ax = SVECTOR_VALUES(a);
float *bx = SVECTOR_VALUES(b);
double num = 0.0;
double denom = 0.0;
int bpos = 0;
CheckDims(a, b);
/*
* Weighted Jaccard distance is not defined for vectors with negative
* values. Could check and return NaN if minimal impact on performance.
*/
for (int i = 0; i < a->nnz; i++)
{
int ai = a->indices[i];
int bi = -1;
for (int j = bpos; j < b->nnz; j++)
{
bi = b->indices[j];
if (ai == bi)
{
num += ax[i] < bx[j] ? ax[i] : bx[j];
denom += ax[i] > bx[j] ? ax[i] : bx[j];
}
else if (ai > bi)
denom += bx[j];
/* Update start for next iteration */
if (ai >= bi)
bpos = j + 1;
/* Found or passed it */
if (bi >= ai)
break;
}
if (ai != bi)
denom += ax[i];
}
for (int j = bpos; j < b->nnz; j++)
denom += bx[j];
if (denom > 0)
PG_RETURN_FLOAT8(1.0 - (num / denom));
else
PG_RETURN_FLOAT8(NAN);
}

23
src/svector.h Normal file
View File

@@ -0,0 +1,23 @@
#ifndef SVECTOR_H
#define SVECTOR_H
#define SVECTOR_MAX_DIM 100000
#define SVECTOR_SIZE(_nnz) (offsetof(SVector, indices) + (_nnz) * sizeof(int32) + (_nnz * sizeof(float)))
#define SVECTOR_VALUES(x) ((float *) (((char *) (x)) + offsetof(SVector, indices) + (x)->nnz * sizeof(int32)))
#define DatumGetSVector(x) ((SVector *) PG_DETOAST_DATUM(x))
#define PG_GETARG_SVECTOR_P(x) DatumGetSVector(PG_GETARG_DATUM(x))
#define PG_RETURN_SVECTOR_P(x) PG_RETURN_POINTER(x)
typedef struct SVector
{
int32 vl_len_; /* varlena header (do not touch directly!) */
int32 dim; /* number of dimensions */
int32 nnz;
int32 unused;
int32 indices[FLEXIBLE_ARRAY_MEMBER];
} SVector;
SVector *InitSVector(int dim, int nnz);
#endif

View File

@@ -9,6 +9,7 @@
#include "lib/stringinfo.h"
#include "libpq/pqformat.h"
#include "port.h" /* for strtof() */
#include "svector.h"
#include "utils/array.h"
#include "utils/builtins.h"
#include "utils/lsyscache.h"
@@ -724,7 +725,7 @@ vector_spherical_distance(PG_FUNCTION_ARGS)
}
/*
* Get the L1 distance between two vectors
* Get the L1 distance between vectors
*/
PGDLLEXPORT PG_FUNCTION_INFO_V1(l1_distance);
Datum
@@ -903,8 +904,8 @@ PGDLLEXPORT PG_FUNCTION_INFO_V1(vector_lt);
Datum
vector_lt(PG_FUNCTION_ARGS)
{
Vector *a = PG_GETARG_VECTOR_P(0);
Vector *b = PG_GETARG_VECTOR_P(1);
Vector *a = (Vector *) PG_GETARG_VECTOR_P(0);
Vector *b = (Vector *) PG_GETARG_VECTOR_P(1);
PG_RETURN_BOOL(vector_cmp_internal(a, b) < 0);
}
@@ -916,8 +917,8 @@ PGDLLEXPORT PG_FUNCTION_INFO_V1(vector_le);
Datum
vector_le(PG_FUNCTION_ARGS)
{
Vector *a = PG_GETARG_VECTOR_P(0);
Vector *b = PG_GETARG_VECTOR_P(1);
Vector *a = (Vector *) PG_GETARG_VECTOR_P(0);
Vector *b = (Vector *) PG_GETARG_VECTOR_P(1);
PG_RETURN_BOOL(vector_cmp_internal(a, b) <= 0);
}
@@ -929,8 +930,8 @@ PGDLLEXPORT PG_FUNCTION_INFO_V1(vector_eq);
Datum
vector_eq(PG_FUNCTION_ARGS)
{
Vector *a = PG_GETARG_VECTOR_P(0);
Vector *b = PG_GETARG_VECTOR_P(1);
Vector *a = (Vector *) PG_GETARG_VECTOR_P(0);
Vector *b = (Vector *) PG_GETARG_VECTOR_P(1);
PG_RETURN_BOOL(vector_cmp_internal(a, b) == 0);
}
@@ -942,8 +943,8 @@ PGDLLEXPORT PG_FUNCTION_INFO_V1(vector_ne);
Datum
vector_ne(PG_FUNCTION_ARGS)
{
Vector *a = PG_GETARG_VECTOR_P(0);
Vector *b = PG_GETARG_VECTOR_P(1);
Vector *a = (Vector *) PG_GETARG_VECTOR_P(0);
Vector *b = (Vector *) PG_GETARG_VECTOR_P(1);
PG_RETURN_BOOL(vector_cmp_internal(a, b) != 0);
}
@@ -955,8 +956,8 @@ PGDLLEXPORT PG_FUNCTION_INFO_V1(vector_ge);
Datum
vector_ge(PG_FUNCTION_ARGS)
{
Vector *a = PG_GETARG_VECTOR_P(0);
Vector *b = PG_GETARG_VECTOR_P(1);
Vector *a = (Vector *) PG_GETARG_VECTOR_P(0);
Vector *b = (Vector *) PG_GETARG_VECTOR_P(1);
PG_RETURN_BOOL(vector_cmp_internal(a, b) >= 0);
}
@@ -968,8 +969,8 @@ PGDLLEXPORT PG_FUNCTION_INFO_V1(vector_gt);
Datum
vector_gt(PG_FUNCTION_ARGS)
{
Vector *a = PG_GETARG_VECTOR_P(0);
Vector *b = PG_GETARG_VECTOR_P(1);
Vector *a = (Vector *) PG_GETARG_VECTOR_P(0);
Vector *b = (Vector *) PG_GETARG_VECTOR_P(1);
PG_RETURN_BOOL(vector_cmp_internal(a, b) > 0);
}
@@ -981,8 +982,8 @@ PGDLLEXPORT PG_FUNCTION_INFO_V1(vector_cmp);
Datum
vector_cmp(PG_FUNCTION_ARGS)
{
Vector *a = PG_GETARG_VECTOR_P(0);
Vector *b = PG_GETARG_VECTOR_P(1);
Vector *a = (Vector *) PG_GETARG_VECTOR_P(0);
Vector *b = (Vector *) PG_GETARG_VECTOR_P(1);
PG_RETURN_INT32(vector_cmp_internal(a, b));
}
@@ -1151,3 +1152,26 @@ vector_avg(PG_FUNCTION_ARGS)
PG_RETURN_POINTER(result);
}
/*
* Convert sparse vector to dense vector
*/
PGDLLEXPORT PG_FUNCTION_INFO_V1(svector_to_vector);
Datum
svector_to_vector(PG_FUNCTION_ARGS)
{
SVector *svec = PG_GETARG_SVECTOR_P(0);
int32 typmod = PG_GETARG_INT32(1);
Vector *result;
int dim = svec->dim;
float *values = SVECTOR_VALUES(svec);
CheckDim(dim);
CheckExpectedDim(typmod, dim);
result = InitVector(dim);
for (int i = 0; i < svec->nnz; i++)
result->x[svec->indices[i]] = values[i];
PG_RETURN_POINTER(result);
}

View File

@@ -54,85 +54,85 @@ SELECT vector_norm('[3e37,4e37]')::real;
5e+37
(1 row)
SELECT l2_distance('[0,0]', '[3,4]');
SELECT l2_distance('[0,0]'::vector, '[3,4]');
l2_distance
-------------
5
(1 row)
SELECT l2_distance('[0,0]', '[0,1]');
SELECT l2_distance('[0,0]'::vector, '[0,1]');
l2_distance
-------------
1
(1 row)
SELECT l2_distance('[1,2]', '[3]');
SELECT l2_distance('[1,2]'::vector, '[3]');
ERROR: different vector dimensions 2 and 1
SELECT l2_distance('[3e38]', '[-3e38]');
SELECT l2_distance('[3e38]'::vector, '[-3e38]');
l2_distance
-------------
Infinity
(1 row)
SELECT inner_product('[1,2]', '[3,4]');
SELECT inner_product('[1,2]'::vector, '[3,4]');
inner_product
---------------
11
(1 row)
SELECT inner_product('[1,2]', '[3]');
SELECT inner_product('[1,2]'::vector, '[3]');
ERROR: different vector dimensions 2 and 1
SELECT inner_product('[3e38]', '[3e38]');
SELECT inner_product('[3e38]'::vector, '[3e38]');
inner_product
---------------
Infinity
(1 row)
SELECT cosine_distance('[1,2]', '[2,4]');
SELECT cosine_distance('[1,2]'::vector, '[2,4]');
cosine_distance
-----------------
0
(1 row)
SELECT cosine_distance('[1,2]', '[0,0]');
SELECT cosine_distance('[1,2]'::vector, '[0,0]');
cosine_distance
-----------------
NaN
(1 row)
SELECT cosine_distance('[1,1]', '[1,1]');
SELECT cosine_distance('[1,1]'::vector, '[1,1]');
cosine_distance
-----------------
0
(1 row)
SELECT cosine_distance('[1,0]', '[0,2]');
SELECT cosine_distance('[1,0]'::vector, '[0,2]');
cosine_distance
-----------------
1
(1 row)
SELECT cosine_distance('[1,1]', '[-1,-1]');
SELECT cosine_distance('[1,1]'::vector, '[-1,-1]');
cosine_distance
-----------------
2
(1 row)
SELECT cosine_distance('[1,2]', '[3]');
SELECT cosine_distance('[1,2]'::vector, '[3]');
ERROR: different vector dimensions 2 and 1
SELECT cosine_distance('[1,1]', '[1.1,1.1]');
SELECT cosine_distance('[1,1]'::vector, '[1.1,1.1]');
cosine_distance
-----------------
0
(1 row)
SELECT cosine_distance('[1,1]', '[-1.1,-1.1]');
SELECT cosine_distance('[1,1]'::vector, '[-1.1,-1.1]');
cosine_distance
-----------------
2
(1 row)
SELECT cosine_distance('[3e38]', '[3e38]');
SELECT cosine_distance('[3e38]'::vector, '[3e38]');
cosine_distance
-----------------
NaN

140
test/expected/svector.out Normal file
View File

@@ -0,0 +1,140 @@
SELECT '(0,1.5),(2,3.5)|5|'::svector;
svector
--------------------
(0,1.5),(2,3.5)|5|
(1 row)
SELECT '(0,1.5),(2,3.5)|5|'::svector::vector;
vector
-----------------
[1.5,0,3.5,0,0]
(1 row)
SELECT '(0,1.5),(2,3.5)|5|'::svector::vector(5);
vector
-----------------
[1.5,0,3.5,0,0]
(1 row)
SELECT '(0,1.5),(2,3.5)|5|'::svector::vector(4);
ERROR: expected 4 dimensions, not 5
SELECT '[0,1.5,0,3.5,0]'::vector::svector;
svector
--------------------
(1,1.5),(3,3.5)|5|
(1 row)
SELECT '(0,0),(1,1),(2,0)|3|'::svector;
svector
----------------------
(0,0),(1,1),(2,0)|3|
(1 row)
SELECT '|5|'::svector;
svector
---------
|5|
(1 row)
SELECT '|-1|'::svector;
ERROR: svector must have at least 1 dimension
LINE 1: SELECT '|-1|'::svector;
^
SELECT '|100001|'::svector;
ERROR: svector cannot have more than 100000 dimensions
LINE 1: SELECT '|100001|'::svector;
^
SELECT '|16001|'::svector::vector;
ERROR: vector cannot have more than 16000 dimensions
SELECT '(-1,1)|1|'::svector;
ERROR: index must not be negative
LINE 1: SELECT '(-1,1)|1|'::svector;
^
SELECT '(1,1)|1|'::svector;
ERROR: index must be less than dimensions
LINE 1: SELECT '(1,1)|1|'::svector;
^
SELECT '|1|'::svector(2);
ERROR: expected 2 dimensions, not 1
SELECT l2_distance('|2|'::svector, '(0,3),(1,4)|2|');
l2_distance
-------------
5
(1 row)
SELECT l2_distance('|2|'::svector, '(1,1)|2|');
l2_distance
-------------
1
(1 row)
SELECT '|2|'::svector <-> '(0,3),(1,4)|2|';
?column?
----------
5
(1 row)
SELECT inner_product('(0,1),(1,2)|2|'::svector, '(0,2),(1,4)|2|');
inner_product
---------------
10
(1 row)
SELECT svector_negative_inner_product('(0,1),(1,2)|2|', '(0,2),(1,4)|2|');
svector_negative_inner_product
--------------------------------
-10
(1 row)
SELECT cosine_distance('(0,1),(1,2)|2|'::svector, '(0,2),(1,4)|2|');
cosine_distance
-----------------
0
(1 row)
SELECT cosine_distance('(0,1),(1,2)|2|'::svector, '|2|');
cosine_distance
-----------------
NaN
(1 row)
SELECT cosine_distance('(0,1),(1,1)|2|'::svector, '(0,-1),(1,-1)|2|');
cosine_distance
-----------------
2
(1 row)
SELECT cosine_distance('(0,1)|2|'::svector, '(1,2)|2|');
cosine_distance
-----------------
1
(1 row)
SELECT cosine_distance('|1|'::svector, '|1|');
cosine_distance
-----------------
NaN
(1 row)
SELECT cosine_distance('(0,1)|2|'::svector, '(0,1)|3|');
ERROR: different svector dimensions 2 and 3
SELECT jaccard_distance('(0,1)|2|', '(0,1)|2|');
jaccard_distance
------------------
0
(1 row)
SELECT jaccard_distance('(0,1)|2|', '(1,1)|2|');
jaccard_distance
------------------
1
(1 row)
SELECT jaccard_distance('|1|', '|1|');
jaccard_distance
------------------
NaN
(1 row)
SELECT jaccard_distance('(0,1)|2|', '(0,1)|3|');
ERROR: different svector dimensions 2 and 3

View File

@@ -13,24 +13,24 @@ SELECT vector_norm('[3,4]');
SELECT vector_norm('[0,1]');
SELECT vector_norm('[3e37,4e37]')::real;
SELECT l2_distance('[0,0]', '[3,4]');
SELECT l2_distance('[0,0]', '[0,1]');
SELECT l2_distance('[1,2]', '[3]');
SELECT l2_distance('[3e38]', '[-3e38]');
SELECT l2_distance('[0,0]'::vector, '[3,4]');
SELECT l2_distance('[0,0]'::vector, '[0,1]');
SELECT l2_distance('[1,2]'::vector, '[3]');
SELECT l2_distance('[3e38]'::vector, '[-3e38]');
SELECT inner_product('[1,2]', '[3,4]');
SELECT inner_product('[1,2]', '[3]');
SELECT inner_product('[3e38]', '[3e38]');
SELECT inner_product('[1,2]'::vector, '[3,4]');
SELECT inner_product('[1,2]'::vector, '[3]');
SELECT inner_product('[3e38]'::vector, '[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]');
SELECT cosine_distance('[1,1]', '[-1.1,-1.1]');
SELECT cosine_distance('[3e38]', '[3e38]');
SELECT cosine_distance('[1,2]'::vector, '[2,4]');
SELECT cosine_distance('[1,2]'::vector, '[0,0]');
SELECT cosine_distance('[1,1]'::vector, '[1,1]');
SELECT cosine_distance('[1,0]'::vector, '[0,2]');
SELECT cosine_distance('[1,1]'::vector, '[-1,-1]');
SELECT cosine_distance('[1,2]'::vector, '[3]');
SELECT cosine_distance('[1,1]'::vector, '[1.1,1.1]');
SELECT cosine_distance('[1,1]'::vector, '[-1.1,-1.1]');
SELECT cosine_distance('[3e38]'::vector, '[3e38]');
SELECT l1_distance('[0,0]', '[3,4]');
SELECT l1_distance('[0,0]', '[0,1]');

36
test/sql/svector.sql Normal file
View File

@@ -0,0 +1,36 @@
SELECT '(0,1.5),(2,3.5)|5|'::svector;
SELECT '(0,1.5),(2,3.5)|5|'::svector::vector;
SELECT '(0,1.5),(2,3.5)|5|'::svector::vector(5);
SELECT '(0,1.5),(2,3.5)|5|'::svector::vector(4);
SELECT '[0,1.5,0,3.5,0]'::vector::svector;
SELECT '(0,0),(1,1),(2,0)|3|'::svector;
SELECT '|5|'::svector;
SELECT '|-1|'::svector;
SELECT '|100001|'::svector;
SELECT '|16001|'::svector::vector;
SELECT '(-1,1)|1|'::svector;
SELECT '(1,1)|1|'::svector;
SELECT '|1|'::svector(2);
SELECT l2_distance('|2|'::svector, '(0,3),(1,4)|2|');
SELECT l2_distance('|2|'::svector, '(1,1)|2|');
SELECT '|2|'::svector <-> '(0,3),(1,4)|2|';
SELECT inner_product('(0,1),(1,2)|2|'::svector, '(0,2),(1,4)|2|');
SELECT svector_negative_inner_product('(0,1),(1,2)|2|', '(0,2),(1,4)|2|');
SELECT cosine_distance('(0,1),(1,2)|2|'::svector, '(0,2),(1,4)|2|');
SELECT cosine_distance('(0,1),(1,2)|2|'::svector, '|2|');
SELECT cosine_distance('(0,1),(1,1)|2|'::svector, '(0,-1),(1,-1)|2|');
SELECT cosine_distance('(0,1)|2|'::svector, '(1,2)|2|');
SELECT cosine_distance('|1|'::svector, '|1|');
SELECT cosine_distance('(0,1)|2|'::svector, '(0,1)|3|');
SELECT jaccard_distance('(0,1)|2|', '(0,1)|2|');
SELECT jaccard_distance('(0,1)|2|', '(1,1)|2|');
SELECT jaccard_distance('|1|', '|1|');
SELECT jaccard_distance('(0,1)|2|', '(0,1)|3|');

View File

@@ -83,32 +83,11 @@ for my $i (0 .. $#operators)
push(@expected, $res);
}
# Build index serially
$node->safe_psql("postgres", qq(
SET max_parallel_maintenance_workers = 0;
CREATE INDEX idx ON tst USING hnsw (v $opclass);
));
# Add index
$node->safe_psql("postgres", "CREATE INDEX ON tst USING hnsw (v $opclass);");
# Test approximate results
my $min = $operator eq "<#>" ? 0.80 : 0.99;
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();