Compare commits

..

1 Commits

Author SHA1 Message Date
Andrew Kane
90aaf2102b Added support for async I/O [skip ci] 2026-04-14 22:47:53 -07:00
14 changed files with 159 additions and 286 deletions

View File

@@ -97,7 +97,7 @@ jobs:
with: with:
postgres-version: ${{ matrix.postgres }} postgres-version: ${{ matrix.postgres }}
- run: | - run: |
call "C:\Program Files\Microsoft Visual Studio\${{ matrix.os == 'windows-2025' && 18 || 2022 }}\Enterprise\VC\Auxiliary\Build\vcvars64.bat" && ^ call "C:\Program Files\Microsoft Visual Studio\2022\Enterprise\VC\Auxiliary\Build\vcvars64.bat" && ^
nmake /NOLOGO /F Makefile.win && ^ nmake /NOLOGO /F Makefile.win && ^
nmake /NOLOGO /F Makefile.win install && ^ nmake /NOLOGO /F Makefile.win install && ^
nmake /NOLOGO /F Makefile.win installcheck ${{ matrix.postgres != 17 && 'PG_REGRESS=$(PGROOT)\bin\pg_regress' || '' }} && ^ nmake /NOLOGO /F Makefile.win installcheck ${{ matrix.postgres != 17 && 'PG_REGRESS=$(PGROOT)\bin\pg_regress' || '' }} && ^

View File

@@ -1,8 +1,3 @@
## 0.8.3 (unreleased)
- Fixed possible index corruption with HNSW vacuuming
- Fixed performance regression with Hamming distance and Jaccard distance with Postgres 18
## 0.8.2 (2026-02-25) ## 0.8.2 (2026-02-25)
- Fixed buffer overflow with parallel HNSW index build - [more info](https://github.com/pgvector/pgvector/issues/959) - Fixed buffer overflow with parallel HNSW index build - [more info](https://github.com/pgvector/pgvector/issues/959)

View File

@@ -11,8 +11,6 @@ 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 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
Have a lot of vectors? Use [quantization](#scaling) to scale
[![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/actions/workflows/build.yml/badge.svg)](https://github.com/pgvector/pgvector/actions)
## Installation ## Installation
@@ -316,8 +314,6 @@ For a large number of workers, you may need to increase `max_parallel_workers` (
The [index options](#index-options) also have a significant impact on build time (use the defaults unless seeing low recall) The [index options](#index-options) also have a significant impact on build time (use the defaults unless seeing low recall)
Use [binary quantization](#binary-quantization) for faster build times at scale
### Indexing Progress ### Indexing Progress
Check [indexing progress](https://www.postgresql.org/docs/current/progress-reporting.html#CREATE-INDEX-PROGRESS-REPORTING) Check [indexing progress](https://www.postgresql.org/docs/current/progress-reporting.html#CREATE-INDEX-PROGRESS-REPORTING)
@@ -447,7 +443,13 @@ Exact indexes work well for conditions that match a low percentage of rows. Othe
CREATE INDEX ON items USING hnsw (embedding vector_l2_ops); CREATE INDEX ON items USING hnsw (embedding vector_l2_ops);
``` ```
With approximate indexes, filtering is applied *after* the index is scanned. If a condition matches 10% of rows, with HNSW and the default `hnsw.ef_search` of 40, only 4 rows will match on average. For more rows, enable [iterative index scans](#iterative-index-scans), which will automatically scan more of the index when needed. With approximate indexes, filtering is applied *after* the index is scanned. If a condition matches 10% of rows, with HNSW and the default `hnsw.ef_search` of 40, only 4 rows will match on average. For more rows, increase `hnsw.ef_search`.
```sql
SET hnsw.ef_search = 200;
```
Starting with 0.8.0, you can enable [iterative index scans](#iterative-index-scans), which will automatically scan more of the index when needed.
```sql ```sql
SET hnsw.iterative_scan = strict_order; SET hnsw.iterative_scan = strict_order;
@@ -671,10 +673,6 @@ SHOW shared_buffers;
Be sure to restart Postgres for changes to take effect. Be sure to restart Postgres for changes to take effect.
### Storing
Use the `halfvec` type instead of `vector` for a smaller working set.
### Loading ### Loading
Use `COPY` for bulk loading data ([example](https://github.com/pgvector/pgvector-python/blob/master/examples/loading/example.py)). Use `COPY` for bulk loading data ([example](https://github.com/pgvector/pgvector-python/blob/master/examples/loading/example.py)).
@@ -689,8 +687,6 @@ Add any indexes *after* loading the initial data for best performance.
See index build time for [HNSW](#index-build-time) and [IVFFlat](#index-build-time-1). See index build time for [HNSW](#index-build-time) and [IVFFlat](#index-build-time-1).
Use [binary quantization](#binary-quantization) for smaller indexes and faster build times at scale.
In production environments, create indexes concurrently to avoid blocking writes. In production environments, create indexes concurrently to avoid blocking writes.
```sql ```sql
@@ -721,8 +717,6 @@ SELECT * FROM items ORDER BY embedding <#> '[3,1,2]' LIMIT 5;
#### Approximate Search #### Approximate Search
Use [binary quantization](#binary-quantization) with re-ranking to keep indexes in-memory at scale.
To speed up queries with an IVFFlat index, increase the number of inverted lists (at the expense of recall). To speed up queries with an IVFFlat index, increase the number of inverted lists (at the expense of recall).
```sql ```sql
@@ -738,20 +732,21 @@ REINDEX INDEX CONCURRENTLY index_name;
VACUUM table_name; VACUUM table_name;
``` ```
## Scaling
For a smaller working set:
1. Use the `halfvec` type instead of `vector` for tables
2. Use [binary quantization](#binary-quantization) for indexes (with re-ranking for search)
Scale vertically by increasing memory, CPU, and storage on a single instance. Use existing tools to [tune parameters](#tuning) and [monitor performance](#monitoring).
Scale horizontally with [replicas](https://www.postgresql.org/docs/current/hot-standby.html), or use [Citus](https://github.com/citusdata/citus), [PgDog](https://github.com/pgdogdev/pgdog), or another approach for sharding ([example](https://github.com/pgvector/pgvector-python/blob/master/examples/citus/example.py)).
## Monitoring ## Monitoring
Use existing tools like [pg_stat_statements](https://www.postgresql.org/docs/current/pgstatstatements.html) or [PgHero](https://github.com/ankane/pghero) to monitor performance. Monitor performance with [pg_stat_statements](https://www.postgresql.org/docs/current/pgstatstatements.html) (be sure to add it to `shared_preload_libraries`).
```sql
CREATE EXTENSION pg_stat_statements;
```
Get the most time-consuming queries with:
```sql
SELECT query, calls, ROUND((total_plan_time + total_exec_time) / calls) AS avg_time_ms,
ROUND((total_plan_time + total_exec_time) / 60000) AS total_time_min
FROM pg_stat_statements ORDER BY total_plan_time + total_exec_time DESC LIMIT 20;
```
Monitor recall by comparing results from approximate search with exact search. Monitor recall by comparing results from approximate search with exact search.
@@ -762,6 +757,14 @@ SELECT ...
COMMIT; COMMIT;
``` ```
## Scaling
Scale pgvector the same way you scale Postgres.
Scale vertically by increasing memory, CPU, and storage on a single instance. Use existing tools to [tune parameters](#tuning) and [monitor performance](#monitoring).
Scale horizontally with [replicas](https://www.postgresql.org/docs/current/hot-standby.html), or use [Citus](https://github.com/citusdata/citus) or another approach for sharding ([example](https://github.com/pgvector/pgvector-python/blob/master/examples/citus/example.py)).
## Languages ## 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. Use pgvector from any language with a Postgres client. You can even generate and store vectors in one language and query them in another.
@@ -875,8 +878,6 @@ No, but like other index types, youll likely see better performance if they d
SELECT pg_size_pretty(pg_relation_size('index_name')); SELECT pg_size_pretty(pg_relation_size('index_name'));
``` ```
Use [half-precision indexing](#half-precision-indexing) or [binary quantization](#binary-quantization) for smaller indexes.
## Troubleshooting ## Troubleshooting
#### Why isnt a query using an index? #### Why isnt a query using an index?
@@ -1329,7 +1330,7 @@ make clean && PG_CFLAGS="-DUSE_ASSERT_CHECKING" make && make install
To enable benchmarking: To enable benchmarking:
```sh ```sh
make clean && PG_CFLAGS="-DHNSW_BENCH -DIVFFLAT_BENCH" make && make install make clean && PG_CFLAGS="-DIVFFLAT_BENCH" make && make install
``` ```
To show memory usage: To show memory usage:

View File

@@ -31,12 +31,10 @@
#define BIT_TARGET_CLONES #define BIT_TARGET_CLONES
#endif #endif
/* Use built-ins when possible for Postgres < 19 for inlining */ /* Use built-ins when possible for inlining */
#if PG_VERSION_NUM >= 190000 #if defined(HAVE__BUILTIN_POPCOUNT) && defined(HAVE_LONG_INT_64)
#define popcount64(x) pg_popcount64(x)
#elif defined(HAVE__BUILTIN_POPCOUNT) && (defined(HAVE_LONG_INT_64) || SIZEOF_LONG == 8)
#define popcount64(x) __builtin_popcountl(x) #define popcount64(x) __builtin_popcountl(x)
#elif defined(HAVE__BUILTIN_POPCOUNT) && (defined(HAVE_LONG_LONG_INT_64) || SIZEOF_LONG_LONG == 8) #elif defined(HAVE__BUILTIN_POPCOUNT) && defined(HAVE_LONG_LONG_INT_64)
#define popcount64(x) __builtin_popcountll(x) #define popcount64(x) __builtin_popcountll(x)
#elif !defined(_MSC_VER) #elif !defined(_MSC_VER)
/* Fails to resolve with MSVC */ /* Fails to resolve with MSVC */

View File

@@ -290,11 +290,7 @@ hnswhandler(PG_FUNCTION_ARGS)
.amcanparallel = false, .amcanparallel = false,
.amcanbuildparallel = true, .amcanbuildparallel = true,
.amcaninclude = false, .amcaninclude = false,
#if PG_VERSION_NUM >= 170000
.amusemaintenanceworkmem = true,
#else
.amusemaintenanceworkmem = false, .amusemaintenanceworkmem = false,
#endif
.amsummarizing = false, .amsummarizing = false,
.amparallelvacuumoptions = VACUUM_OPTION_PARALLEL_BULKDEL, .amparallelvacuumoptions = VACUUM_OPTION_PARALLEL_BULKDEL,
.amkeytype = InvalidOid, .amkeytype = InvalidOid,
@@ -355,11 +351,7 @@ hnswhandler(PG_FUNCTION_ARGS)
amroutine->amcanbuildparallel = true; amroutine->amcanbuildparallel = true;
#endif #endif
amroutine->amcaninclude = false; amroutine->amcaninclude = false;
#if PG_VERSION_NUM >= 170000 amroutine->amusemaintenanceworkmem = false; /* not used during VACUUM */
amroutine->amusemaintenanceworkmem = true;
#else
amroutine->amusemaintenanceworkmem = false;
#endif
#if PG_VERSION_NUM >= 160000 #if PG_VERSION_NUM >= 160000
amroutine->amsummarizing = false; amroutine->amsummarizing = false;
#endif #endif

View File

@@ -18,20 +18,10 @@
#include "utils/sampling.h" #include "utils/sampling.h"
#include "vector.h" #include "vector.h"
#ifdef HNSW_BENCH
#include "portability/instr_time.h"
#endif
#if PG_VERSION_NUM >= 190000 #if PG_VERSION_NUM >= 190000
typedef Pointer Item; typedef Pointer Item;
#endif #endif
#if PG_VERSION_NUM >= 170000
#define HnswTidStore TidStore
#else
#define HnswTidStore tidhash_hash
#endif
#define HNSW_MAX_DIM 2000 #define HNSW_MAX_DIM 2000
#define HNSW_MAX_NNZ 1000 #define HNSW_MAX_NNZ 1000
@@ -88,21 +78,6 @@ typedef Pointer Item;
#define HnswPageGetOpaque(page) ((HnswPageOpaque) PageGetSpecialPointer(page)) #define HnswPageGetOpaque(page) ((HnswPageOpaque) PageGetSpecialPointer(page))
#define HnswPageGetMeta(page) ((HnswMetaPageData *) PageGetContents(page)) #define HnswPageGetMeta(page) ((HnswMetaPageData *) PageGetContents(page))
#ifdef HNSW_BENCH
#define HnswBench(name, code) \
do { \
instr_time start; \
instr_time duration; \
INSTR_TIME_SET_CURRENT(start); \
(code); \
INSTR_TIME_SET_CURRENT(duration); \
INSTR_TIME_SUBTRACT(duration, start); \
elog(INFO, "%s: %.3f ms", name, INSTR_TIME_GET_MILLISEC(duration)); \
} while (0)
#else
#define HnswBench(name, code) (code)
#endif
#if PG_VERSION_NUM >= 150000 #if PG_VERSION_NUM >= 150000
#define RandomDouble() pg_prng_double(&pg_global_prng_state) #define RandomDouble() pg_prng_double(&pg_global_prng_state)
#define SeedRandom(seed) pg_prng_seed(&pg_global_prng_state, seed) #define SeedRandom(seed) pg_prng_seed(&pg_global_prng_state, seed)
@@ -397,6 +372,13 @@ typedef union
ItemPointerData indextid; ItemPointerData indextid;
} HnswUnvisited; } HnswUnvisited;
typedef struct HnswReadStreamData
{
HnswUnvisited *unvisited;
int unvisitedLength;
int visited;
} HnswReadStreamData;
typedef struct HnswScanOpaqueData typedef struct HnswScanOpaqueData
{ {
const HnswTypeInfo *typeInfo; const HnswTypeInfo *typeInfo;
@@ -433,11 +415,10 @@ typedef struct HnswVacuumState
HnswSupport support; HnswSupport support;
/* Variables */ /* Variables */
struct HnswTidStore *deleted; struct tidhash_hash *deleted;
BufferAccessStrategy bas; BufferAccessStrategy bas;
HnswNeighborTuple ntup; HnswNeighborTuple ntup;
HnswElementData highestPoint; HnswElementData highestPoint;
HnswElementData fallbackPoint;
/* Memory */ /* Memory */
MemoryContext tmpCtx; MemoryContext tmpCtx;
@@ -453,13 +434,13 @@ bool HnswCheckNorm(HnswSupport * support, Datum value);
Buffer HnswNewBuffer(Relation index, ForkNumber forkNum); Buffer HnswNewBuffer(Relation index, ForkNumber forkNum);
void HnswInitPage(Buffer buf, Page page); void HnswInitPage(Buffer buf, Page page);
void HnswInit(void); void HnswInit(void);
List *HnswSearchLayer(char *base, HnswQuery * q, List *ep, int ef, int lc, Relation index, HnswSupport * support, int m, bool inserting, HnswElement skipElement, visited_hash * v, pairingheap **discarded, bool initVisited, int64 *tuples); List *HnswSearchLayer(char *base, HnswQuery * q, List *ep, int ef, int lc, Relation index, HnswSupport * support, int m, bool inserting, HnswElement skipElement, visited_hash * v, pairingheap **discarded, bool initVisited, int64 *tuples, bool maintenance);
HnswElement HnswGetEntryPoint(Relation index); HnswElement HnswGetEntryPoint(Relation index);
void HnswGetMetaPageInfo(Relation index, int *m, HnswElement * entryPoint); void HnswGetMetaPageInfo(Relation index, int *m, HnswElement * entryPoint);
void *HnswAlloc(HnswAllocator * allocator, Size size); void *HnswAlloc(HnswAllocator * allocator, Size size);
HnswElement HnswInitElement(char *base, ItemPointer tid, int m, double ml, int maxLevel, HnswAllocator * alloc); HnswElement HnswInitElement(char *base, ItemPointer tid, int m, double ml, int maxLevel, HnswAllocator * alloc);
HnswElement HnswInitElementFromBlock(BlockNumber blkno, OffsetNumber offno); HnswElement HnswInitElementFromBlock(BlockNumber blkno, OffsetNumber offno);
void HnswFindElementNeighbors(char *base, HnswElement element, HnswElement entryPoint, Relation index, HnswSupport * support, int m, int efConstruction, bool existing); void HnswFindElementNeighbors(char *base, HnswElement element, HnswElement entryPoint, Relation index, HnswSupport * support, int m, int efConstruction, bool existing, bool maintenance);
HnswSearchCandidate *HnswEntryCandidate(char *base, HnswElement entryPoint, HnswQuery * q, Relation index, HnswSupport * support, bool loadVec); HnswSearchCandidate *HnswEntryCandidate(char *base, HnswElement entryPoint, HnswQuery * q, Relation index, HnswSupport * support, 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, bool building);
void HnswSetNeighborTuple(char *base, HnswNeighborTuple ntup, HnswElement e, int m); void HnswSetNeighborTuple(char *base, HnswNeighborTuple ntup, HnswElement e, int m);

View File

@@ -470,7 +470,7 @@ InsertTupleInMemory(HnswBuildState * buildstate, HnswElement element)
} }
/* Find neighbors for element */ /* Find neighbors for element */
HnswFindElementNeighbors(base, element, entryPoint, NULL, support, m, efConstruction, false); HnswFindElementNeighbors(base, element, entryPoint, NULL, support, m, efConstruction, false, true);
/* Update graph in memory */ /* Update graph in memory */
UpdateGraphInMemory(support, element, m, entryPoint, buildstate); UpdateGraphInMemory(support, element, m, entryPoint, buildstate);

View File

@@ -731,7 +731,7 @@ HnswInsertTupleOnDisk(Relation index, HnswSupport * support, Datum value, ItemPo
} }
/* Find neighbors for element */ /* Find neighbors for element */
HnswFindElementNeighbors(base, element, entryPoint, index, support, m, efConstruction, false); HnswFindElementNeighbors(base, element, entryPoint, index, support, m, efConstruction, false, building);
/* Update graph on disk */ /* Update graph on disk */
UpdateGraphOnDisk(index, support, element, m, entryPoint, building); UpdateGraphOnDisk(index, support, element, m, entryPoint, building);

View File

@@ -48,11 +48,11 @@ GetScanItems(IndexScanDesc scan, Datum value)
for (int lc = entryPoint->level; lc >= 1; lc--) for (int lc = entryPoint->level; lc >= 1; lc--)
{ {
w = HnswSearchLayer(base, q, ep, 1, lc, index, support, m, false, NULL, NULL, NULL, true, NULL); w = HnswSearchLayer(base, q, ep, 1, lc, index, support, m, false, NULL, NULL, NULL, true, NULL, false);
ep = w; ep = w;
} }
return HnswSearchLayer(base, q, ep, hnsw_ef_search, 0, index, support, m, false, NULL, &so->v, hnsw_iterative_scan != HNSW_ITERATIVE_SCAN_OFF ? &so->discarded : NULL, true, &so->tuples); return HnswSearchLayer(base, q, ep, hnsw_ef_search, 0, index, support, m, false, NULL, &so->v, hnsw_iterative_scan != HNSW_ITERATIVE_SCAN_OFF ? &so->discarded : NULL, true, &so->tuples, false);
} }
/* /*
@@ -83,7 +83,7 @@ ResumeScanItems(IndexScanDesc scan)
ep = lappend(ep, sc); ep = lappend(ep, sc);
} }
return HnswSearchLayer(base, &so->q, ep, batch_size, 0, index, &so->support, so->m, false, NULL, &so->v, &so->discarded, false, &so->tuples); return HnswSearchLayer(base, &so->q, ep, batch_size, 0, index, &so->support, so->m, false, NULL, &so->v, &so->discarded, false, &so->tuples, false);
} }
/* /*

View File

@@ -21,6 +21,10 @@
#include "varatt.h" #include "varatt.h"
#endif #endif
#if PG_VERSION_NUM >= 190000
#include "storage/read_stream.h"
#endif
#if PG_VERSION_NUM < 170000 #if PG_VERSION_NUM < 170000
static inline uint64 static inline uint64
murmurhash64(uint64 data) murmurhash64(uint64 data)
@@ -531,14 +535,12 @@ HnswGetDistance(Datum a, Datum b, HnswSupport * support)
* Load an element and optionally get its distance from q * Load an element and optionally get its distance from q
*/ */
static void static void
HnswLoadElementImpl(BlockNumber blkno, OffsetNumber offno, double *distance, HnswQuery * q, Relation index, HnswSupport * support, bool loadVec, double *maxDistance, HnswElement * element) HnswLoadElementImpl(Buffer buf, OffsetNumber offno, double *distance, HnswQuery * q, Relation index, HnswSupport * support, bool loadVec, double *maxDistance, HnswElement * element)
{ {
Buffer buf;
Page page; Page page;
HnswElementTuple etup; HnswElementTuple etup;
/* Read vector */ /* Read vector */
buf = ReadBuffer(index, blkno);
LockBuffer(buf, BUFFER_LOCK_SHARE); LockBuffer(buf, BUFFER_LOCK_SHARE);
page = BufferGetPage(buf); page = BufferGetPage(buf);
@@ -546,9 +548,6 @@ HnswLoadElementImpl(BlockNumber blkno, OffsetNumber offno, double *distance, Hns
Assert(HnswIsElementTuple(etup)); Assert(HnswIsElementTuple(etup));
if (unlikely(etup->deleted))
elog(ERROR, "cannot load deleted element");
/* Calculate distance */ /* Calculate distance */
if (distance != NULL) if (distance != NULL)
{ {
@@ -562,7 +561,7 @@ HnswLoadElementImpl(BlockNumber blkno, OffsetNumber offno, double *distance, Hns
if (distance == NULL || maxDistance == NULL || *distance < *maxDistance) if (distance == NULL || maxDistance == NULL || *distance < *maxDistance)
{ {
if (*element == NULL) if (*element == NULL)
*element = HnswInitElementFromBlock(blkno, offno); *element = HnswInitElementFromBlock(BufferGetBlockNumber(buf), offno);
HnswLoadElementFromTuple(*element, etup, true, loadVec); HnswLoadElementFromTuple(*element, etup, true, loadVec);
} }
@@ -576,7 +575,9 @@ HnswLoadElementImpl(BlockNumber blkno, OffsetNumber offno, double *distance, Hns
void void
HnswLoadElement(HnswElement element, double *distance, HnswQuery * q, Relation index, HnswSupport * support, bool loadVec, double *maxDistance) HnswLoadElement(HnswElement element, double *distance, HnswQuery * q, Relation index, HnswSupport * support, bool loadVec, double *maxDistance)
{ {
HnswLoadElementImpl(element->blkno, element->offno, distance, q, index, support, loadVec, maxDistance, &element); Buffer buf = ReadBuffer(index, element->blkno);
HnswLoadElementImpl(buf, element->offno, distance, q, index, support, loadVec, maxDistance, &element);
} }
/* /*
@@ -816,11 +817,31 @@ HnswLoadUnvisitedFromDisk(HnswElement element, HnswUnvisited * unvisited, int *u
} }
} }
#if PG_VERSION_NUM >= 190000
/*
* Get next block number for read stream
*/
static BlockNumber
HnswReadStreamNextBlock(ReadStream *stream, void *callback_private_data, void *per_buffer_data)
{
HnswReadStreamData *streamData = callback_private_data;
OffsetNumber *offno = per_buffer_data;
HnswUnvisited *uv;
if (streamData->visited == streamData->unvisitedLength)
return InvalidBlockNumber;
uv = &streamData->unvisited[streamData->visited++];
*offno = ItemPointerGetOffsetNumber(&uv->indextid);
return ItemPointerGetBlockNumber(&uv->indextid);
}
#endif
/* /*
* Algorithm 2 from paper * Algorithm 2 from paper
*/ */
List * List *
HnswSearchLayer(char *base, HnswQuery * q, List *ep, int ef, int lc, Relation index, HnswSupport * support, int m, bool inserting, HnswElement skipElement, visited_hash * v, pairingheap **discarded, bool initVisited, int64 *tuples) HnswSearchLayer(char *base, HnswQuery * q, List *ep, int ef, int lc, Relation index, HnswSupport * support, int m, bool inserting, HnswElement skipElement, visited_hash * v, pairingheap **discarded, bool initVisited, int64 *tuples, bool maintenance)
{ {
List *w = NIL; List *w = NIL;
pairingheap *C = pairingheap_allocate(CompareNearestCandidates, NULL); pairingheap *C = pairingheap_allocate(CompareNearestCandidates, NULL);
@@ -835,6 +856,21 @@ HnswSearchLayer(char *base, HnswQuery * q, List *ep, int ef, int lc, Relation in
int unvisitedLength; int unvisitedLength;
bool inMemory = index == NULL; bool inMemory = index == NULL;
#if PG_VERSION_NUM >= 190000
HnswReadStreamData streamData;
ReadStream *stream = NULL;
if (!inMemory)
{
int flags = READ_STREAM_USE_BATCHING;
if (maintenance)
flags |= READ_STREAM_MAINTENANCE;
stream = read_stream_begin_relation(flags, NULL, index, MAIN_FORKNUM, HnswReadStreamNextBlock, &streamData, sizeof(OffsetNumber));
}
#endif
if (v == NULL) if (v == NULL)
{ {
v = &vh; v = &vh;
@@ -897,13 +933,23 @@ HnswSearchLayer(char *base, HnswQuery * q, List *ep, int ef, int lc, Relation in
if (inMemory) if (inMemory)
HnswLoadUnvisitedFromMemory(base, cElement, unvisited, &unvisitedLength, v, lc, localNeighborhood, neighborhoodSize); HnswLoadUnvisitedFromMemory(base, cElement, unvisited, &unvisitedLength, v, lc, localNeighborhood, neighborhoodSize);
else else
{
HnswLoadUnvisitedFromDisk(cElement, unvisited, &unvisitedLength, v, index, m, lm, lc); HnswLoadUnvisitedFromDisk(cElement, unvisited, &unvisitedLength, v, index, m, lm, lc);
#if PG_VERSION_NUM >= 190000
read_stream_resume(stream);
streamData.unvisited = unvisited;
streamData.unvisitedLength = unvisitedLength;
streamData.visited = 0;
#endif
}
/* OK to count elements instead of tuples */ /* OK to count elements instead of tuples */
if (tuples != NULL) if (tuples != NULL)
(*tuples) += unvisitedLength; (*tuples) += unvisitedLength;
for (int i = 0; i < unvisitedLength; i++) for (int i = 0;; i++)
{ {
HnswElement eElement; HnswElement eElement;
HnswSearchCandidate *e; HnswSearchCandidate *e;
@@ -914,18 +960,40 @@ HnswSearchLayer(char *base, HnswQuery * q, List *ep, int ef, int lc, Relation in
if (inMemory) if (inMemory)
{ {
if (i == unvisitedLength)
break;
eElement = unvisited[i].element; eElement = unvisited[i].element;
eDistance = GetElementDistance(base, eElement, q, support); eDistance = GetElementDistance(base, eElement, q, support);
} }
else else
{ {
ItemPointer indextid = &unvisited[i].indextid; Buffer buf;
BlockNumber blkno = ItemPointerGetBlockNumber(indextid); OffsetNumber offno;
OffsetNumber offno = ItemPointerGetOffsetNumber(indextid);
#if PG_VERSION_NUM >= 190000
void *offnoPtr;
buf = read_stream_next_buffer(stream, &offnoPtr);
if (!BufferIsValid(buf))
break;
offno = *((OffsetNumber *) offnoPtr);
#else
ItemPointer indextid;
if (i == unvisitedLength)
break;
indextid = &unvisited[i].indextid;
buf = ReadBuffer(index, ItemPointerGetBlockNumber(indextid));
offno = ItemPointerGetOffsetNumber(indextid);
#endif
/* Avoid any allocations if not adding */ /* Avoid any allocations if not adding */
eElement = NULL; eElement = NULL;
HnswLoadElementImpl(blkno, offno, &eDistance, q, index, support, inserting, alwaysAdd || discarded != NULL ? NULL : &f->distance, &eElement); HnswLoadElementImpl(buf, offno, &eDistance, q, index, support, inserting, alwaysAdd || discarded != NULL ? NULL : &f->distance, &eElement);
if (eElement == NULL) if (eElement == NULL)
continue; continue;
@@ -981,6 +1049,11 @@ HnswSearchLayer(char *base, HnswQuery * q, List *ep, int ef, int lc, Relation in
w = lappend(w, sc); w = lappend(w, sc);
} }
#if PG_VERSION_NUM >= 190000
if (!inMemory)
read_stream_end(stream);
#endif
return w; return w;
} }
@@ -1276,7 +1349,7 @@ PrecomputeHash(char *base, HnswElement element)
* Algorithm 1 from paper * Algorithm 1 from paper
*/ */
void void
HnswFindElementNeighbors(char *base, HnswElement element, HnswElement entryPoint, Relation index, HnswSupport * support, int m, int efConstruction, bool existing) HnswFindElementNeighbors(char *base, HnswElement element, HnswElement entryPoint, Relation index, HnswSupport * support, int m, int efConstruction, bool existing, bool maintenance)
{ {
List *ep; List *ep;
List *w; List *w;
@@ -1303,7 +1376,7 @@ HnswFindElementNeighbors(char *base, HnswElement element, HnswElement entryPoint
/* 1st phase: greedy search to insert level */ /* 1st phase: greedy search to insert level */
for (int lc = entryLevel; lc >= level + 1; lc--) for (int lc = entryLevel; lc >= level + 1; lc--)
{ {
w = HnswSearchLayer(base, &q, ep, 1, lc, index, support, m, true, skipElement, NULL, NULL, true, NULL); w = HnswSearchLayer(base, &q, ep, 1, lc, index, support, m, true, skipElement, NULL, NULL, true, NULL, maintenance);
ep = w; ep = w;
} }
@@ -1322,7 +1395,7 @@ HnswFindElementNeighbors(char *base, HnswElement element, HnswElement entryPoint
List *lw = NIL; List *lw = NIL;
ListCell *lc2; ListCell *lc2;
w = HnswSearchLayer(base, &q, ep, efConstruction, lc, index, support, m, true, skipElement, NULL, NULL, true, NULL); w = HnswSearchLayer(base, &q, ep, efConstruction, lc, index, support, m, true, skipElement, NULL, NULL, true, NULL, maintenance);
/* Convert search candidates to candidates */ /* Convert search candidates to candidates */
foreach(lc2, w) foreach(lc2, w)

View File

@@ -10,12 +10,6 @@
#include "utils/memutils.h" #include "utils/memutils.h"
#include "utils/rel.h" #include "utils/rel.h"
#if PG_VERSION_NUM >= 170000
#include "access/tidstore.h"
#include "miscadmin.h"
#include "postmaster/autovacuum.h"
#endif
#if PG_VERSION_NUM >= 160000 #if PG_VERSION_NUM >= 160000
#include "varatt.h" #include "varatt.h"
#endif #endif
@@ -28,13 +22,9 @@
* Check if deleted list contains an index TID * Check if deleted list contains an index TID
*/ */
static bool static bool
DeletedContains(HnswTidStore * deleted, ItemPointer indextid) DeletedContains(tidhash_hash * deleted, ItemPointer indextid)
{ {
#if PG_VERSION_NUM >= 170000
return TidStoreIsMember(deleted, indextid);
#else
return tidhash_lookup(deleted, *indextid) != NULL; return tidhash_lookup(deleted, *indextid) != NULL;
#endif
} }
/* /*
@@ -47,20 +37,17 @@ RemoveHeapTids(HnswVacuumState * vacuumstate)
{ {
BlockNumber blkno = HNSW_HEAD_BLKNO; BlockNumber blkno = HNSW_HEAD_BLKNO;
HnswElement highestPoint = &vacuumstate->highestPoint; HnswElement highestPoint = &vacuumstate->highestPoint;
HnswElement fallbackPoint = &vacuumstate->fallbackPoint;
Relation index = vacuumstate->index; Relation index = vacuumstate->index;
BufferAccessStrategy bas = vacuumstate->bas; BufferAccessStrategy bas = vacuumstate->bas;
HnswElement entryPoint = HnswGetEntryPoint(vacuumstate->index);
IndexBulkDeleteResult *stats = vacuumstate->stats; IndexBulkDeleteResult *stats = vacuumstate->stats;
/* Store separately since HnswElement level is uint8 */ /* Store separately since highestPoint.level is uint8 */
int highestLevel = -1; int highestLevel = -1;
int fallbackLevel = -1;
/* Initialize highest point and fallback point */ /* Initialize highest point */
highestPoint->blkno = InvalidBlockNumber; highestPoint->blkno = InvalidBlockNumber;
highestPoint->offno = InvalidOffsetNumber; highestPoint->offno = InvalidOffsetNumber;
fallbackPoint->blkno = InvalidBlockNumber;
fallbackPoint->offno = InvalidOffsetNumber;
while (BlockNumberIsValid(blkno)) while (BlockNumberIsValid(blkno))
{ {
@@ -70,10 +57,6 @@ RemoveHeapTids(HnswVacuumState * vacuumstate)
OffsetNumber offno; OffsetNumber offno;
OffsetNumber maxoffno; OffsetNumber maxoffno;
bool updated = false; bool updated = false;
#if PG_VERSION_NUM >= 170000
OffsetNumber deletedoffs[MaxOffsetNumber];
int ndeletedoffs = 0;
#endif
vacuum_delay_point(); vacuum_delay_point();
@@ -127,9 +110,6 @@ RemoveHeapTids(HnswVacuumState * vacuumstate)
if (!ItemPointerIsValid(&etup->heaptids[0])) if (!ItemPointerIsValid(&etup->heaptids[0]))
{ {
#if PG_VERSION_NUM >= 170000
deletedoffs[ndeletedoffs++] = offno;
#else
ItemPointerData ip; ItemPointerData ip;
bool found; bool found;
@@ -138,39 +118,17 @@ RemoveHeapTids(HnswVacuumState * vacuumstate)
tidhash_insert(vacuumstate->deleted, ip, &found); tidhash_insert(vacuumstate->deleted, ip, &found);
Assert(!found); Assert(!found);
#endif
} }
else if (etup->level > highestLevel) else if (etup->level > highestLevel && !(entryPoint != NULL && blkno == entryPoint->blkno && offno == entryPoint->offno))
{ {
if (BlockNumberIsValid(highestPoint->blkno)) /* Keep track of highest non-entry point */
{
/* Current highest point becomes fallback */
fallbackPoint->blkno = highestPoint->blkno;
fallbackPoint->offno = highestPoint->offno;
fallbackPoint->level = highestPoint->level;
fallbackLevel = highestLevel;
}
/* Keep track of highest point */
highestPoint->blkno = blkno; highestPoint->blkno = blkno;
highestPoint->offno = offno; highestPoint->offno = offno;
highestPoint->level = etup->level; highestPoint->level = etup->level;
highestLevel = etup->level; highestLevel = etup->level;
} }
else if (etup->level > fallbackLevel)
{
/* Keep track of second highest point */
fallbackPoint->blkno = blkno;
fallbackPoint->offno = offno;
fallbackPoint->level = etup->level;
fallbackLevel = etup->level;
}
} }
#if PG_VERSION_NUM >= 170000
TidStoreSetBlockOffsets(vacuumstate->deleted, blkno, deletedoffs, ndeletedoffs);
#endif
blkno = HnswPageGetOpaque(page)->nextblkno; blkno = HnswPageGetOpaque(page)->nextblkno;
if (updated) if (updated)
@@ -180,10 +138,6 @@ RemoveHeapTids(HnswVacuumState * vacuumstate)
UnlockReleaseBuffer(buf); UnlockReleaseBuffer(buf);
} }
#ifdef HNSW_MEMORY
elog(INFO, "memory: %zu KB", MemoryContextMemAllocated(CurrentMemoryContext, true) / 1024);
#endif
} }
/* /*
@@ -264,7 +218,7 @@ RepairGraphElement(HnswVacuumState * vacuumstate, HnswElement element, HnswEleme
element->heaptidsLength = 0; element->heaptidsLength = 0;
/* Find neighbors for element, skipping itself */ /* Find neighbors for element, skipping itself */
HnswFindElementNeighbors(base, element, entryPoint, index, support, m, efConstruction, true); HnswFindElementNeighbors(base, element, entryPoint, index, support, m, efConstruction, true, true);
/* Zero memory for each element */ /* Zero memory for each element */
MemSet(ntup, 0, HNSW_TUPLE_ALLOC_SIZE); MemSet(ntup, 0, HNSW_TUPLE_ALLOC_SIZE);
@@ -315,27 +269,12 @@ RepairGraphEntryPoint(HnswVacuumState * vacuumstate)
/* Get a shared lock */ /* Get a shared lock */
LockPage(index, HNSW_UPDATE_LOCK, ShareLock); LockPage(index, HNSW_UPDATE_LOCK, ShareLock);
/* Get latest entry point */ /* Load element */
entryPoint = HnswGetEntryPoint(index); HnswLoadElement(highestPoint, NULL, NULL, index, support, true, NULL);
/* Use fallback point if highest point is entry point */ /* Repair if needed */
if (entryPoint != NULL && entryPoint->blkno == highestPoint->blkno && entryPoint->offno == highestPoint->offno) if (NeedsUpdated(vacuumstate, highestPoint))
{ RepairGraphElement(vacuumstate, highestPoint, HnswGetEntryPoint(index));
highestPoint = &vacuumstate->fallbackPoint;
if (!BlockNumberIsValid(highestPoint->blkno))
highestPoint = NULL;
}
if (highestPoint != NULL)
{
/* Load element */
HnswLoadElement(highestPoint, NULL, NULL, index, support, true, NULL);
/* Repair if needed */
if (NeedsUpdated(vacuumstate, highestPoint))
RepairGraphElement(vacuumstate, highestPoint, entryPoint);
}
/* Release lock */ /* Release lock */
UnlockPage(index, HNSW_UPDATE_LOCK, ShareLock); UnlockPage(index, HNSW_UPDATE_LOCK, ShareLock);
@@ -502,99 +441,6 @@ RepairGraph(HnswVacuumState * vacuumstate)
/* Reset memory context */ /* Reset memory context */
MemoryContextSwitchTo(oldCtx); MemoryContextSwitchTo(oldCtx);
MemoryContextReset(vacuumstate->tmpCtx); MemoryContextReset(vacuumstate->tmpCtx);
#ifdef HNSW_VACUUM_PROGRESS
if (!BlockNumberIsValid(blkno) || ((blkno - HNSW_HEAD_BLKNO) % 1000 == 0 && blkno != HNSW_HEAD_BLKNO))
{
BlockNumber totalBlocks = RelationGetNumberOfBlocks(index);
BlockNumber currentBlocks = BlockNumberIsValid(blkno) ? blkno : totalBlocks;
elog(INFO, "hnsw vacuum progress: %.1f%%", 100.0 * currentBlocks / totalBlocks);
}
#endif
}
}
/*
* Confirm graph was repaired
*/
static void
ConfirmRepaired(HnswVacuumState * vacuumstate)
{
BlockNumber blkno = HNSW_HEAD_BLKNO;
Relation index = vacuumstate->index;
BufferAccessStrategy bas = vacuumstate->bas;
while (BlockNumberIsValid(blkno))
{
Buffer buf;
Page page;
OffsetNumber offno;
OffsetNumber maxoffno;
vacuum_delay_point();
buf = ReadBufferExtended(index, MAIN_FORKNUM, blkno, RBM_NORMAL, bas);
LockBuffer(buf, BUFFER_LOCK_SHARE);
page = BufferGetPage(buf);
maxoffno = PageGetMaxOffsetNumber(page);
/* Iterate over nodes */
for (offno = FirstOffsetNumber; offno <= maxoffno; offno = OffsetNumberNext(offno))
{
HnswElementTuple etup = (HnswElementTuple) PageGetItem(page, PageGetItemId(page, offno));
HnswNeighborTuple ntup;
Buffer nbuf;
Page npage;
BlockNumber neighborPage;
OffsetNumber neighborOffno;
/* Skip neighbor tuples */
if (!HnswIsElementTuple(etup))
continue;
/* Skip if being deleted */
if (!ItemPointerIsValid(&etup->heaptids[0]))
continue;
/* Get neighbor page */
neighborPage = ItemPointerGetBlockNumber(&etup->neighbortid);
neighborOffno = ItemPointerGetOffsetNumber(&etup->neighbortid);
if (neighborPage == blkno)
{
nbuf = buf;
npage = page;
}
else
{
nbuf = ReadBufferExtended(index, MAIN_FORKNUM, neighborPage, RBM_NORMAL, bas);
LockBuffer(nbuf, BUFFER_LOCK_SHARE);
npage = BufferGetPage(nbuf);
}
ntup = (HnswNeighborTuple) PageGetItem(npage, PageGetItemId(npage, neighborOffno));
/* Check neighbors */
for (int i = 0; i < ntup->count; i++)
{
ItemPointer indextid = &ntup->indextids[i];
if (!ItemPointerIsValid(indextid))
continue;
/* Check if in deleted list */
if (DeletedContains(vacuumstate->deleted, indextid))
elog(ERROR, "hnsw graph not repaired");
}
if (nbuf != buf)
UnlockReleaseBuffer(nbuf);
}
blkno = HnswPageGetOpaque(page)->nextblkno;
UnlockReleaseBuffer(buf);
} }
} }
@@ -761,11 +607,7 @@ InitVacuumState(HnswVacuumState * vacuumstate, IndexVacuumInfo *info, IndexBulkD
HnswGetMetaPageInfo(index, &vacuumstate->m, NULL); HnswGetMetaPageInfo(index, &vacuumstate->m, NULL);
/* Create hash table */ /* Create hash table */
#if PG_VERSION_NUM >= 170000
vacuumstate->deleted = TidStoreCreateLocal((AmAutoVacuumWorkerProcess() && autovacuum_work_mem != -1) ? autovacuum_work_mem : maintenance_work_mem, true);
#else
vacuumstate->deleted = tidhash_create(CurrentMemoryContext, 256, NULL); vacuumstate->deleted = tidhash_create(CurrentMemoryContext, 256, NULL);
#endif
} }
/* /*
@@ -774,11 +616,7 @@ InitVacuumState(HnswVacuumState * vacuumstate, IndexVacuumInfo *info, IndexBulkD
static void static void
FreeVacuumState(HnswVacuumState * vacuumstate) FreeVacuumState(HnswVacuumState * vacuumstate)
{ {
#if PG_VERSION_NUM >= 170000
TidStoreDestroy(vacuumstate->deleted);
#else
tidhash_destroy(vacuumstate->deleted); tidhash_destroy(vacuumstate->deleted);
#endif
FreeAccessStrategy(vacuumstate->bas); FreeAccessStrategy(vacuumstate->bas);
pfree(vacuumstate->ntup); pfree(vacuumstate->ntup);
MemoryContextDelete(vacuumstate->tmpCtx); MemoryContextDelete(vacuumstate->tmpCtx);
@@ -796,16 +634,13 @@ hnswbulkdelete(IndexVacuumInfo *info, IndexBulkDeleteResult *stats,
InitVacuumState(&vacuumstate, info, stats, callback, callback_state); InitVacuumState(&vacuumstate, info, stats, callback, callback_state);
/* Pass 1: Remove heap TIDs */ /* Pass 1: Remove heap TIDs */
HnswBench("RemoveHeapTids", RemoveHeapTids(&vacuumstate)); RemoveHeapTids(&vacuumstate);
/* Pass 2: Repair graph */ /* Pass 2: Repair graph */
HnswBench("RepairGraph", RepairGraph(&vacuumstate)); RepairGraph(&vacuumstate);
/* Pass 3: Confirm repaired */ /* Pass 3: Mark as deleted */
HnswBench("ConfirmRepaired", ConfirmRepaired(&vacuumstate)); MarkDeleted(&vacuumstate);
/* Pass 4: Mark as deleted */
HnswBench("MarkDeleted", MarkDeleted(&vacuumstate));
FreeVacuumState(&vacuumstate); FreeVacuumState(&vacuumstate);

View File

@@ -145,9 +145,8 @@ SampleRows(IvfflatBuildState * buildstate)
{ {
BlockNumber targblock = BlockSampler_Next(&buildstate->bs); BlockNumber targblock = BlockSampler_Next(&buildstate->bs);
/* Set anyvisible to false like table_index_build_scan */
table_index_build_range_scan(buildstate->heap, buildstate->index, buildstate->indexInfo, table_index_build_range_scan(buildstate->heap, buildstate->index, buildstate->indexInfo,
false, false, false, targblock, 1, SampleCallback, (void *) buildstate, NULL); false, true, false, targblock, 1, SampleCallback, (void *) buildstate, NULL);
} }
/* Normalize if needed */ /* Normalize if needed */
@@ -399,7 +398,6 @@ InitBuildState(IvfflatBuildState * buildstate, Relation heap, Relation index, In
buildstate->slot = MakeSingleTupleTableSlot(buildstate->sortdesc, &TTSOpsVirtual); buildstate->slot = MakeSingleTupleTableSlot(buildstate->sortdesc, &TTSOpsVirtual);
/* TODO Ensure within maintenance_work_mem */
buildstate->centers = VectorArrayInit(buildstate->lists, buildstate->dimensions, buildstate->typeInfo->itemSize(buildstate->dimensions)); buildstate->centers = VectorArrayInit(buildstate->lists, buildstate->dimensions, buildstate->typeInfo->itemSize(buildstate->dimensions));
buildstate->listInfo = palloc(sizeof(ListInfo) * buildstate->lists); buildstate->listInfo = palloc(sizeof(ListInfo) * buildstate->lists);

View File

@@ -182,10 +182,10 @@ sparsevec_isspace(char ch)
static int static int
CompareIndices(const void *a, const void *b) CompareIndices(const void *a, const void *b)
{ {
if (((const SparseInputElement *) a)->index < ((const SparseInputElement *) b)->index) if (((SparseInputElement *) a)->index < ((SparseInputElement *) b)->index)
return -1; return -1;
if (((const SparseInputElement *) a)->index > ((const SparseInputElement *) b)->index) if (((SparseInputElement *) a)->index > ((SparseInputElement *) b)->index)
return 1; return 1;
return 0; return 0;

View File

@@ -40,7 +40,7 @@
#endif #endif
#if PG_VERSION_NUM >= 180000 #if PG_VERSION_NUM >= 180000
PG_MODULE_MAGIC_EXT(.name = "vector", .version = "0.8.2"); PG_MODULE_MAGIC_EXT(.name = "vector",.version = "0.8.2");
#else #else
PG_MODULE_MAGIC; PG_MODULE_MAGIC;
#endif #endif