Compare commits

..

9 Commits

Author SHA1 Message Date
Andrew Kane
a7b6be73e1 Use DSA 2024-01-16 21:59:11 -08:00
Andrew Kane
d801a843f4 Removed HnswPtrSetNull to avoid setting relptr_off directly 2024-01-16 17:08:13 -08:00
Andrew Kane
1458c7bb2a Improved code [skip ci] 2024-01-16 14:03:28 -08:00
Andrew Kane
cad48d9203 Improved locking 2024-01-16 13:34:55 -08:00
Heikki Linnakangas
719b4b7436 Use LWLocks instead of SpinLocks (#410)
Spinlocks should be held only for a few instructions, for multiple
reasons:

- You have to be very careful not to elog() out while holding a
  spinlock, because there is no mechanism to release the spinlock on
  error.

- Waiters can waste a lot of cycles spinning if the lock is
  contended. I you wait on a spinlock for too long, the PostgreSQL
  implementation will actually PANIC, see s_lock_stuck().

The flushLock is particularly problematic. It is held in exclusive
mode, which means it holds a spinlock, over the call to
FlushPages(). FlushPages() performs lots of I/O so it can take a very
long time (>= minutes), and can also easily error out for various
reasons.

allocatorLock would perhaps be OK as a spinlocks, but even that feels
a bit heavy, so I converted that to an LWLock, too.

entryLock is usually held for a very short time, in shared mode, so
that would be fine as a spinlock. However, in the rare case that the
entry point is updated, it's held for a very long time. An LWLock used
in shared mode is about as fast a spinlock, that path is pretty
heavily optimized.

I think we have some problems with the per-element spinlocks too. In
HnswUpdateNeighborPagesInMemory(), it's held over a call to
HnswUpdateConnection(), but HnswUpdateConnection() can error out at
least in case of an out-of-memory error (it uses lappend(), which
calls palloc()). It also calls the distance function, and I don't
think they are guaranteed to be ereport-free either. However, I didn't
address that in this PR, it needs a bit more thinking.
2024-01-16 13:25:03 -08:00
Andrew Kane
fa0acbf62d Fixed CI 2024-01-15 19:55:46 -08:00
Andrew Kane
1612b84069 Fixed error on Windows [skip ci] 2024-01-15 19:33:16 -08:00
Andrew Kane
2f9371516d Leave space for other objects in shared memory 2024-01-15 19:17:50 -08:00
Andrew Kane
9d3e4e74df Added support for in-memory parallel index builds for HNSW 2024-01-15 15:07:31 -08:00
26 changed files with 496 additions and 616 deletions

View File

@@ -20,6 +20,8 @@ jobs:
os: ubuntu-20.04
- postgres: 12
os: ubuntu-20.04
- postgres: 11
os: ubuntu-20.04
steps:
- uses: actions/checkout@v4
- uses: ankane/setup-postgres@v1
@@ -81,10 +83,10 @@ jobs:
if: ${{ !startsWith(github.ref_name, 'mac') && !startsWith(github.ref_name, 'windows') }}
runs-on: ubuntu-latest
container:
image: debian:12
image: debian:11
options: --platform linux/386
steps:
- run: apt-get update && apt-get install -y build-essential git libipc-run-perl postgresql-15 postgresql-server-dev-15 sudo
- run: apt-get update && apt-get install -y build-essential git libipc-run-perl postgresql-13 postgresql-server-dev-13 sudo
- run: service postgresql start
- run: |
git clone https://github.com/${{ github.repository }}.git pgvector

View File

@@ -1,17 +1,10 @@
## 0.6.0 (unreleased)
## 0.5.2 (unreleased)
If upgrading with Postgres < 13, see [this note](https://github.com/pgvector/pgvector#060).
- Changed storage for vector from `extended` to `external`
- Improved performance of HNSW
- Added support for parallel index builds for HNSW
- Added `hamming_distance` function
- Added validation for GUC parameters
- Reduced memory usage for HNSW index builds
- Reduced WAL generation for HNSW index builds
- Fixed error with logical replication
- Fixed `invalid memory alloc request size` error with HNSW index build
- Dropped support for Postgres 11
## 0.5.1 (2023-10-10)

View File

@@ -12,7 +12,7 @@
"prereqs": {
"runtime": {
"requires": {
"PostgreSQL": "12.0.0"
"PostgreSQL": "11.0.0"
}
}
},

View File

@@ -304,16 +304,6 @@ SELECT ...
COMMIT;
```
### Index Build Time
Speed up index creation on large tables by increasing the number of parallel workers (2 by default)
```sql
SET max_parallel_maintenance_workers = 7; -- plus leader
```
For a large number of workers, you may also need to increase `max_parallel_workers` (8 by default)
### Indexing Progress
Check [indexing progress](https://www.postgresql.org/docs/current/progress-reporting.html#CREATE-INDEX-PROGRESS-REPORTING) with Postgres 12+
@@ -582,7 +572,7 @@ sum(vector) → vector | sum | 0.5.0
If your machine has multiple Postgres installations, specify the path to [pg_config](https://www.postgresql.org/docs/current/app-pgconfig.html) with:
```sh
export PG_CONFIG=/Library/PostgreSQL/16/bin/pg_config
export PG_CONFIG=/Applications/Postgres.app/Contents/Versions/latest/bin/pg_config
```
Then re-run the installation instructions (run `make clean` before `make` if needed). If `sudo` is needed for `make install`, use:
@@ -591,14 +581,6 @@ Then re-run the installation instructions (run `make clean` before `make` if nee
sudo --preserve-env=PG_CONFIG make install
```
A few common paths on Mac are:
- EDB installer - `/Library/PostgreSQL/16/bin/pg_config`
- Homebrew (arm64) - `/opt/homebrew/opt/postgresql@16/bin/pg_config`
- Homebrew (x86-64) - `/usr/local/opt/postgresql@16/bin/pg_config`
Note: Replace `16` with your Postgres server version
### Missing Header
If compilation fails with `fatal error: postgres.h: No such file or directory`, make sure Postgres development files are installed on the server.
@@ -611,10 +593,6 @@ sudo apt install postgresql-server-dev-16
Note: Replace `16` with your Postgres server version
### Missing SDK
If compilation fails and the output includes `warning: no such sysroot directory` on Mac, reinstall Xcode Command Line Tools.
### Windows
Support for Windows is currently experimental. Ensure [C++ support in Visual Studio](https://learn.microsoft.com/en-us/cpp/build/building-on-the-command-line?view=msvc-170#download-and-install-the-tools) is installed, and run:
@@ -729,16 +707,6 @@ SELECT extversion FROM pg_extension WHERE extname = 'vector';
## Upgrade Notes
### 0.6.0 [unreleased]
If upgrading with Postgres < 13, remove this line from `sql/vector--0.5.1--0.6.0.sql`:
```sql
ALTER TYPE vector SET (STORAGE = external);
```
Then run `make install` and `ALTER EXTENSION vector UPDATE;`.
### 0.4.0
If upgrading with Postgres < 13, remove this line from `sql/vector--0.3.2--0.4.0.sql`:
@@ -813,25 +781,7 @@ make prove_installcheck PROVE_TESTS=test/t/001_wal.pl # TAP test
To enable benchmarking:
```sh
make clean && PG_CFLAGS="-DIVFFLAT_BENCH" make && make install
```
To show memory usage:
```sh
make clean && PG_CFLAGS="-DHNSW_MEMORY -DIVFFLAT_MEMORY" make && make install
```
To enable assertions:
```sh
make clean && PG_CFLAGS="-DUSE_ASSERT_CHECKING" make && make install
```
To get k-means metrics:
```sh
make clean && PG_CFLAGS="-DIVFFLAT_KMEANS_DEBUG" make && make install
make clean && PG_CFLAGS=-DIVFFLAT_BENCH make && make install
```
Resources for contributors

View File

@@ -1,8 +0,0 @@
-- 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
-- remove this single line for Postgres < 13
ALTER TYPE vector SET (STORAGE = external);
CREATE FUNCTION hamming_distance(bytea, bytea) RETURNS float8
AS 'MODULE_PATHNAME' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;

View File

@@ -26,7 +26,7 @@ CREATE TYPE vector (
TYPMOD_IN = vector_typmod_in,
RECEIVE = vector_recv,
SEND = vector_send,
STORAGE = external
STORAGE = extended
);
-- functions
@@ -290,8 +290,3 @@ 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);
-- bytea functions
CREATE FUNCTION hamming_distance(bytea, bytea) RETURNS float8
AS 'MODULE_PATHNAME' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;

View File

@@ -4,15 +4,13 @@
#include <math.h>
#include "access/amapi.h"
#include "access/reloptions.h"
#include "commands/progress.h"
#include "commands/vacuum.h"
#include "hnsw.h"
#include "utils/guc.h"
#include "utils/selfuncs.h"
#if PG_VERSION_NUM < 150000
#define MarkGUCPrefixReserved(x) EmitWarningsOnPlaceholders(x)
#if PG_VERSION_NUM >= 120000
#include "commands/progress.h"
#endif
int hnsw_ef_search;
@@ -72,13 +70,12 @@ 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);
MarkGUCPrefixReserved("hnsw");
}
/*
* Get the name of index build phase
*/
#if PG_VERSION_NUM >= 120000
static char *
hnswbuildphasename(int64 phasenum)
{
@@ -92,6 +89,7 @@ hnswbuildphasename(int64 phasenum)
return NULL;
}
}
#endif
/*
* Estimate the cost of an index scan
@@ -106,6 +104,9 @@ hnswcostestimate(PlannerInfo *root, IndexPath *path, double loop_count,
int m;
int entryLevel;
Relation index;
#if PG_VERSION_NUM < 120000
List *qinfos;
#endif
/* Never use index without order */
if (path->indexorderbys == NULL)
@@ -131,7 +132,12 @@ hnswcostestimate(PlannerInfo *root, IndexPath *path, double loop_count,
/* Account for number of tuples (or entry level), m, and ef_search */
costs.numIndexTuples = (entryLevel + 2) * m;
#if PG_VERSION_NUM >= 120000
genericcostestimate(root, path, loop_count, &costs);
#else
qinfos = deconstruct_indexquals(path);
genericcostestimate(root, path, loop_count, qinfos, &costs);
#endif
/* Use total cost since most work happens before first tuple is returned */
*indexStartupCost = costs.indexTotalCost;
@@ -225,7 +231,9 @@ hnswhandler(PG_FUNCTION_ARGS)
amroutine->amcostestimate = hnswcostestimate;
amroutine->amoptions = hnswoptions;
amroutine->amproperty = NULL; /* TODO AMPROP_DISTANCE_ORDERABLE */
#if PG_VERSION_NUM >= 120000
amroutine->ambuildphasename = hnswbuildphasename;
#endif
amroutine->amvalidate = hnswvalidate;
#if PG_VERSION_NUM >= 140000
amroutine->amadjustmembers = NULL;

View File

@@ -3,17 +3,21 @@
#include "postgres.h"
#include "access/genam.h"
#include "access/generic_xlog.h"
#include "access/parallel.h"
#include "lib/pairingheap.h"
#include "access/reloptions.h"
#include "nodes/execnodes.h"
#include "port.h" /* for random() */
#include "utils/relptr.h"
#include "utils/sampling.h"
#include "vector.h"
#if PG_VERSION_NUM < 110000
#error "Requires PostgreSQL 11+"
#endif
#if PG_VERSION_NUM < 120000
#error "Requires PostgreSQL 12+"
#include "access/relscan.h"
#endif
#define HNSW_MAX_DIM 2000
@@ -110,6 +114,8 @@
/* Variables */
extern int hnsw_ef_search;
/* This is initialized when the module is loaded */
extern int hnsw_lock_tranche_id;
typedef struct HnswElementData HnswElementData;
@@ -194,12 +200,20 @@ typedef struct HnswGraph
bool flushed;
} HnswGraph;
typedef struct HnswSpool
{
Relation heap;
Relation index;
} HnswSpool;
typedef struct HnswShared
{
/* Immutable state */
Oid heaprelid;
Oid indexrelid;
bool isconcurrent;
int scantuplesortstates;
dsa_pointer dsaptr;
/* Worker progress */
ConditionVariable workersdonecv;
@@ -211,10 +225,16 @@ typedef struct HnswShared
int nparticipantsdone;
double reltuples;
HnswGraph graphData;
#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
{
@@ -373,18 +393,18 @@ void HnswGetMetaPageInfo(Relation index, int *m, HnswElement * entryPoint);
void *HnswAlloc(HnswAllocator * allocator, Size size);
HnswElement HnswInitElement(char *base, ItemPointer tid, int m, double ml, int maxLevel, HnswAllocator * alloc);
HnswElement HnswInitElementFromBlock(BlockNumber blkno, OffsetNumber offno);
void HnswFindElementNeighbors(char *base, HnswElement element, HnswElement entryPoint, Relation index, FmgrInfo *procinfo, Oid collation, int m, int efConstruction, bool existing);
void HnswInsertElement(char *base, HnswElement element, HnswElement entryPoint, Relation index, FmgrInfo *procinfo, Oid collation, int m, int efConstruction, bool existing);
HnswCandidate *HnswEntryCandidate(char *base, 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 HnswSetNeighborTuple(char *base, HnswNeighborTuple ntup, HnswElement e, int m);
void HnswAddHeapTid(HnswElement element, ItemPointer heaptid);
void HnswInitNeighbors(char *base, HnswElement element, int m, HnswAllocator * alloc);
bool HnswInsertTupleOnDisk(Relation index, Datum value, Datum *values, bool *isnull, ItemPointer heap_tid, Relation heapRel, bool building);
void HnswUpdateNeighborsOnDisk(Relation index, FmgrInfo *procinfo, Oid collation, HnswElement e, int m, bool checkExisting, bool building);
void HnswUpdateNeighborPages(Relation index, FmgrInfo *procinfo, Oid collation, HnswElement e, int m, bool checkExisting, bool building);
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(char *base, HnswElementTuple etup, HnswElement element);
void HnswUpdateConnection(char *base, HnswElement element, HnswCandidate * hc, int lm, int lc, int *updateIdx, Relation index, FmgrInfo *procinfo, Oid collation);
void HnswUpdateConnection(char *base, 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);

View File

@@ -1,53 +1,14 @@
/*
* The HNSW build happens in two phases:
*
* 1. In-memory phase
*
* In this first phase, the graph is held completely in memory. When the graph
* is fully built, or we run out of memory reserved for the build (determined
* by maintenance_work_mem), we materialize the graph to disk (see
* FlushPages()), and switch to the on-disk phase.
*
* In a parallel build, a large contiguous chunk of shared memory is allocated
* to hold the graph. Each worker process has its own HnswBuildState struct in
* private memory, which contains information that doesn't change throughout
* the build, and pointers to the shared structs in shared memory. The shared
* memory area is mapped to a different address in each worker process, and
* 'HnswBuildState.hnswarea' points to the beginning of the shared area in the
* worker process's address space. All pointers used in the graph are
* "relative pointers", stored as an offset from 'hnswarea'.
*
* Each element is protected by an LWLock. It must be held when reading or
* modifying the element's neighbors or 'heaptids'.
*
* In a non-parallel build, the graph is held in backend-private memory. All
* the elements are allocated in a dedicated memory context, 'graphCtx', and
* the pointers used in the graph are regular pointers.
*
* 2. On-disk phase
*
* In the on-disk phase, the index is built by inserting each vector to the
* index one by one, just like on INSERT. The only difference is that we don't
* WAL-log the individual inserts. If the graph fit completely in memory and
* was fully built in the in-memory phase, the on-disk phase is skipped.
*
* After we have finished building the graph, we perform one more scan through
* the index and write all the pages to the WAL.
*/
#include "postgres.h"
#include <math.h>
#include "access/parallel.h"
#include "access/table.h"
#include "access/tableam.h"
#include "access/xact.h"
#include "access/xloginsert.h"
#include "catalog/index.h"
#include "commands/progress.h"
#include "hnsw.h"
#include "miscadmin.h"
#include "optimizer/optimizer.h"
#include "lib/pairingheap.h"
#include "nodes/pg_list.h"
#include "storage/bufmgr.h"
#include "tcop/tcopprot.h"
#include "utils/datum.h"
@@ -55,23 +16,43 @@
#if PG_VERSION_NUM >= 140000
#include "utils/backend_progress.h"
#else
#elif PG_VERSION_NUM >= 120000
#include "pgstat.h"
#endif
#if PG_VERSION_NUM >= 120000
#include "access/tableam.h"
#include "commands/progress.h"
#else
#define PROGRESS_CREATEIDX_TUPLES_DONE 0
#endif
#if PG_VERSION_NUM >= 130000
#define CALLBACK_ITEM_POINTER ItemPointer tid
#else
#define CALLBACK_ITEM_POINTER HeapTuple hup
#endif
#if PG_VERSION_NUM >= 120000
#define UpdateProgress(index, val) pgstat_progress_update_param(index, val)
#else
#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_HNSW_AREA UINT64CONST(0xA000000000000002)
#define PARALLEL_KEY_QUERY_TEXT UINT64CONST(0xA000000000000003)
@@ -314,7 +295,7 @@ FlushPages(HnswBuildState * buildstate)
* Add a heap TID to an existing element
*/
static bool
AddDuplicateInMemory(HnswElement element, HnswElement dup)
HnswAddDuplicateInMemory(HnswElement element, HnswElement dup)
{
LWLockAcquire(&dup->lock, LW_EXCLUSIVE);
@@ -335,15 +316,15 @@ AddDuplicateInMemory(HnswElement element, HnswElement dup)
* Find duplicate element
*/
static bool
FindDuplicateInMemory(char *base, HnswElement element)
HnswFindDuplicateInMemory(char *base, HnswElement element)
{
HnswNeighborArray *neighbors = HnswGetNeighbors(base, element, 0);
Datum value = HnswGetValue(base, element);
for (int i = 0; i < neighbors->length; i++)
{
HnswCandidate *neighbor = &neighbors->items[i];
HnswElement neighborElement = HnswPtrAccess(base, neighbor->element);
Datum value = HnswGetValue(base, element);
Datum neighborValue = HnswGetValue(base, neighborElement);
/* Exit early since ordered by distance */
@@ -351,7 +332,7 @@ FindDuplicateInMemory(char *base, HnswElement element)
return false;
/* Check for space */
if (AddDuplicateInMemory(element, neighborElement))
if (HnswAddDuplicateInMemory(element, neighborElement))
return true;
}
@@ -359,10 +340,10 @@ FindDuplicateInMemory(char *base, HnswElement element)
}
/*
* Add to element list
* Add to element and neighbor pages
*/
static void
AddElementInMemory(char *base, HnswGraph * graph, HnswElement element)
WriteNewElementPagesInMemory(char *base, HnswGraph * graph, HnswElement element)
{
SpinLockAcquire(&graph->lock);
element->next = graph->head;
@@ -374,7 +355,7 @@ AddElementInMemory(char *base, HnswGraph * graph, HnswElement element)
* Update neighbors
*/
static void
UpdateNeighborsInMemory(char *base, FmgrInfo *procinfo, Oid collation, HnswElement e, int m)
HnswUpdateNeighborPagesInMemory(char *base, FmgrInfo *procinfo, Oid collation, HnswElement e, int m)
{
for (int lc = e->level; lc >= 0; lc--)
{
@@ -398,82 +379,45 @@ UpdateNeighborsInMemory(char *base, FmgrInfo *procinfo, Oid collation, HnswEleme
}
/*
* Update graph in memory
* Write changes in memory
*/
static void
UpdateGraphInMemory(FmgrInfo *procinfo, Oid collation, HnswElement element, int m, int efConstruction, HnswElement entryPoint, HnswBuildState * buildstate)
WriteElementInMemory(Relation index, FmgrInfo *procinfo, Oid collation, HnswElement element, int m, int efConstruction, HnswElement entryPoint, HnswBuildState * buildstate, HnswGraph * graph, bool updateEntryPoint)
{
HnswGraph *graph = buildstate->graph;
char *base = buildstate->hnswarea;
/* Look for duplicate */
if (FindDuplicateInMemory(base, element))
/* Try to add to existing page */
if (HnswFindDuplicateInMemory(base, element))
return;
/* Add element */
AddElementInMemory(base, graph, element);
/* Write element and neighbor tuples */
WriteNewElementPagesInMemory(base, graph, element);
/* Update neighbors */
UpdateNeighborsInMemory(base, procinfo, collation, element, m);
HnswUpdateNeighborPagesInMemory(base, procinfo, collation, element, m);
/* Update entry point if needed (already have lock) */
if (entryPoint == NULL || element->level > entryPoint->level)
if (updateEntryPoint)
HnswPtrStore(base, graph->entryPoint, element);
}
/*
* Insert tuple in memory
*/
static void
InsertTupleInMemory(HnswBuildState * buildstate, HnswElement element)
{
FmgrInfo *procinfo = buildstate->procinfo;
Oid collation = buildstate->collation;
HnswGraph *graph = buildstate->graph;
HnswElement entryPoint;
LWLock *entryLock = &graph->entryLock;
int efConstruction = buildstate->efConstruction;
int m = buildstate->m;
char *base = buildstate->hnswarea;
/* Get entry point */
LWLockAcquire(entryLock, LW_SHARED);
entryPoint = HnswPtrAccess(base, graph->entryPoint);
/* Prevent concurrent inserts when likely updating entry point */
if (entryPoint == NULL || element->level > entryPoint->level)
{
/* Release shared lock */
LWLockRelease(entryLock);
/* Get exclusive lock */
LWLockAcquire(entryLock, LW_EXCLUSIVE);
/* Get latest entry point after lock is acquired */
entryPoint = HnswPtrAccess(base, graph->entryPoint);
}
/* Find neighbors for element */
HnswFindElementNeighbors(base, element, entryPoint, NULL, procinfo, collation, m, efConstruction, false);
/* Update graph in memory */
UpdateGraphInMemory(procinfo, collation, element, m, efConstruction, entryPoint, buildstate);
/* Release entry lock */
LWLockRelease(entryLock);
}
/*
* Insert tuple
*/
static bool
InsertTuple(Relation index, Datum *values, bool *isnull, ItemPointer heaptid, HnswBuildState * buildstate)
{
FmgrInfo *procinfo = buildstate->procinfo;
Oid collation = buildstate->collation;
HnswGraph *graph = buildstate->graph;
HnswElement entryPoint;
int efConstruction = buildstate->efConstruction;
int m = buildstate->m;
HnswElement element;
HnswAllocator *allocator = &buildstate->allocator;
Size valueSize;
Pointer valuePtr;
bool updateEntryPoint;
LWLock *flushLock = &graph->flushLock;
char *base = buildstate->hnswarea;
@@ -483,7 +427,7 @@ InsertTuple(Relation index, Datum *values, bool *isnull, ItemPointer heaptid, Hn
/* Normalize if needed */
if (buildstate->normprocinfo != NULL)
{
if (!HnswNormValue(buildstate->normprocinfo, buildstate->collation, &value, buildstate->normvec))
if (!HnswNormValue(buildstate->normprocinfo, collation, &value, buildstate->normvec))
return false;
}
@@ -493,7 +437,6 @@ InsertTuple(Relation index, Datum *values, bool *isnull, ItemPointer heaptid, Hn
/* Ensure graph not flushed when inserting */
LWLockAcquire(flushLock, LW_SHARED);
/* Are we in the on-disk phase? */
if (graph->flushed)
{
LWLockRelease(flushLock);
@@ -501,16 +444,10 @@ InsertTuple(Relation index, Datum *values, bool *isnull, ItemPointer heaptid, Hn
return HnswInsertTupleOnDisk(index, value, values, isnull, heaptid, buildstate->heap, true);
}
/*
* In a parallel build, the HnswElement is allocated from the shared
* memory area, so we need to coordinate with other processes.
*/
/* Get lock for allocator */
LWLockAcquire(&graph->allocatorLock, LW_EXCLUSIVE);
/*
* Check that we have enough memory available for the new element now that
* we have the allocator lock, and flush pages if needed.
*/
/* Flush pages if needed */
if (graph->memoryUsed >= graph->memoryTotal)
{
LWLockRelease(&graph->allocatorLock);
@@ -533,26 +470,38 @@ InsertTuple(Relation index, Datum *values, bool *isnull, ItemPointer heaptid, Hn
return HnswInsertTupleOnDisk(index, value, values, isnull, heaptid, buildstate->heap, true);
}
/* Ok, we can proceed to allocate the element */
/* Create an element */
element = HnswInitElement(base, heaptid, buildstate->m, buildstate->ml, buildstate->maxLevel, allocator);
valuePtr = HnswAlloc(allocator, valueSize);
/*
* We have now allocated the space needed for the element, so we don't
* need the allocator lock anymore. Release it and initialize the rest of
* the element.
*/
/* Release allocator lock */
LWLockRelease(&graph->allocatorLock);
/* Copy the datum */
/* Copy datum */
memcpy(valuePtr, DatumGetPointer(value), valueSize);
HnswPtrStore(base, element->value, valuePtr);
/* Create a lock for the element */
/* Create element lock */
LWLockInitialize(&element->lock, hnsw_lock_tranche_id);
/* Insert tuple */
InsertTupleInMemory(buildstate, element);
/* Get entry point */
LWLockAcquire(&graph->entryLock, LW_EXCLUSIVE);
entryPoint = HnswPtrAccess(base, graph->entryPoint);
updateEntryPoint = entryPoint == NULL || element->level > entryPoint->level;
/* Release lock if not updating entry point */
if (!updateEntryPoint)
LWLockRelease(&graph->entryLock);
/* Insert element in graph */
HnswInsertElement(base, element, entryPoint, NULL, procinfo, collation, m, efConstruction, false);
/* Write to memory */
WriteElementInMemory(index, procinfo, collation, element, m, efConstruction, entryPoint, buildstate, graph, updateEntryPoint);
/* Release lock if needed */
if (updateEntryPoint)
LWLockRelease(&graph->entryLock);
/* Release flush lock */
LWLockRelease(flushLock);
@@ -761,25 +710,36 @@ ParallelHeapScan(HnswBuildState * buildstate)
* Perform a worker's portion of a parallel insert
*/
static void
HnswParallelScanAndInsert(Relation heapRel, Relation indexRel, HnswShared * hnswshared, char *hnswarea, bool progress)
HnswParallelScanAndInsert(HnswSpool * hnswspool, HnswShared * hnswshared, char *hnswarea, bool progress)
{
HnswBuildState buildstate;
#if PG_VERSION_NUM >= 120000
TableScanDesc scan;
#else
HeapScanDesc scan;
#endif
double reltuples;
IndexInfo *indexInfo;
/* Join parallel scan */
indexInfo = BuildIndexInfo(indexRel);
indexInfo = BuildIndexInfo(hnswspool->index);
indexInfo->ii_Concurrent = hnswshared->isconcurrent;
InitBuildState(&buildstate, heapRel, indexRel, indexInfo, MAIN_FORKNUM);
InitBuildState(&buildstate, hnswspool->heap, hnswspool->index, indexInfo, MAIN_FORKNUM);
buildstate.graph = &hnswshared->graphData;
buildstate.hnswarea = hnswarea;
InitAllocator(&buildstate.allocator, &HnswSharedMemoryAlloc, &buildstate);
scan = table_beginscan_parallel(heapRel,
#if PG_VERSION_NUM >= 120000
scan = table_beginscan_parallel(hnswspool->heap,
ParallelTableScanFromHnswShared(hnswshared));
reltuples = table_index_build_scan(heapRel, indexRel, indexInfo,
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);
@@ -806,7 +766,10 @@ void
HnswParallelBuildMain(dsm_segment *seg, shm_toc *toc)
{
char *sharedquery;
HnswSpool *hnswspool;
HnswShared *hnswshared;
dsa_area *dsa;
void *dsaspace;
char *hnswarea;
Relation heapRel;
Relation indexRel;
@@ -836,17 +799,32 @@ HnswParallelBuildMain(dsm_segment *seg, shm_toc *toc)
}
/* 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);
hnswarea = shm_toc_lookup(toc, PARALLEL_KEY_HNSW_AREA, false);
/* Initialize worker's own spool */
hnswspool = (HnswSpool *) palloc0(sizeof(HnswSpool));
hnswspool->heap = heapRel;
hnswspool->index = indexRel;
dsaspace = shm_toc_lookup(toc, PARALLEL_KEY_HNSW_AREA, false);
dsa = dsa_attach_in_place(dsaspace, seg);
hnswarea = dsa_get_address(dsa, hnswshared->dsaptr);
/* Perform inserts */
HnswParallelScanAndInsert(heapRel, indexRel, hnswshared, hnswarea, false);
HnswParallelScanAndInsert(hnswspool, hnswshared, hnswarea, false);
/* Close relations within worker */
index_close(indexRel, indexLockmode);
#if PG_VERSION_NUM >= 120000
table_close(heapRel, heapLockmode);
#else
heap_close(heapRel, heapLockmode);
#endif
}
/*
@@ -871,7 +849,19 @@ HnswEndParallel(HnswLeader * hnswleader)
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
}
/*
@@ -881,9 +871,15 @@ 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(buildstate->heap, buildstate->index, hnswleader->hnswshared, hnswleader->hnswarea, true);
HnswParallelScanAndInsert(leaderworker, hnswleader->hnswshared, hnswleader->hnswarea, true);
}
/*
@@ -893,11 +889,14 @@ static void
HnswBeginParallel(HnswBuildState * buildstate, bool isconcurrent, int request)
{
ParallelContext *pcxt;
int scantuplesortstates;
Snapshot snapshot;
Size esthnswshared;
Size esthnswarea;
Size estother;
Size estdsaspace;
HnswShared *hnswshared;
void *dsaspace;
dsa_area *dsa;
char *hnswarea;
HnswLeader *hnswleader = (HnswLeader *) palloc0(sizeof(HnswLeader));
bool leaderparticipates = true;
@@ -910,7 +909,13 @@ HnswBeginParallel(HnswBuildState * buildstate, bool isconcurrent, int request)
/* 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)
@@ -922,15 +927,9 @@ HnswBeginParallel(HnswBuildState * buildstate, bool isconcurrent, int request)
esthnswshared = ParallelEstimateShared(buildstate->heap, snapshot);
shm_toc_estimate_chunk(&pcxt->estimator, esthnswshared);
/* Leave space for other objects in shared memory */
/* Docker has a default limit of 64 MB for shm_size */
/* which happens to be the default value of maintenance_work_mem */
esthnswarea = maintenance_work_mem * 1024L;
estother = 3 * 1024 * 1024;
if (esthnswarea > estother)
esthnswarea -= estother;
shm_toc_estimate_chunk(&pcxt->estimator, esthnswarea);
/* Start with a minimal DSA so InitializeParallelDSM does not fail */
estdsaspace = dsa_minimum_size();
shm_toc_estimate_chunk(&pcxt->estimator, estdsaspace);
shm_toc_estimate_keys(&pcxt->estimator, 2);
/* Finally, estimate PARALLEL_KEY_QUERY_TEXT space */
@@ -962,21 +961,42 @@ HnswBeginParallel(HnswBuildState * buildstate, bool isconcurrent, int request)
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;
#if PG_VERSION_NUM >= 120000
table_parallelscan_initialize(buildstate->heap,
ParallelTableScanFromHnswShared(hnswshared),
snapshot);
#else
heap_parallelscan_initialize(&hnswshared->heapdesc, buildstate->heap, snapshot);
#endif
hnswarea = (char *) shm_toc_allocate(pcxt->toc, esthnswarea);
dsaspace = shm_toc_allocate(pcxt->toc, estdsaspace);
dsa = dsa_create_in_place(dsaspace, estdsaspace, hnsw_lock_tranche_id, pcxt->seg);
esthnswarea = maintenance_work_mem * 1024L;
hnswshared->dsaptr = dsa_allocate_extended(dsa, esthnswarea, DSA_ALLOC_HUGE | DSA_ALLOC_NO_OOM);
/* If not enough shared memory, back out (do serial build) */
if (!DsaPointerIsValid(hnswshared->dsaptr))
{
if (IsMVCCSnapshot(snapshot))
UnregisterSnapshot(snapshot);
DestroyParallelContext(pcxt);
ExitParallelMode();
return;
}
hnswarea = dsa_get_address(dsa, hnswshared->dsaptr);
/* Report less than allocated so never fails */
InitGraph(&hnswshared->graphData, hnswarea, esthnswarea - 1024 * 1024);
shm_toc_insert(pcxt->toc, PARALLEL_KEY_HNSW_SHARED, hnswshared);
shm_toc_insert(pcxt->toc, PARALLEL_KEY_HNSW_AREA, hnswarea);
shm_toc_insert(pcxt->toc, PARALLEL_KEY_HNSW_AREA, dsaspace);
/* Store query string for workers */
if (debug_query_string)
@@ -1064,8 +1084,15 @@ BuildGraph(HnswBuildState * buildstate, ForkNumber forkNum)
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);
#else
buildstate->reltuples = IndexBuildHeapScan(buildstate->heap, buildstate->index, buildstate->indexInfo,
true, BuildCallback, (void *) buildstate, NULL);
#endif
}
buildstate->indtuples = buildstate->graph->indtuples;
}
@@ -1079,6 +1106,22 @@ BuildGraph(HnswBuildState * buildstate, ForkNumber forkNum)
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
*/

View File

@@ -2,7 +2,6 @@
#include <math.h>
#include "access/generic_xlog.h"
#include "hnsw.h"
#include "storage/bufmgr.h"
#include "storage/lmgr.h"
@@ -117,7 +116,7 @@ HnswInsertAppendPage(Relation index, Buffer *nbuf, Page *npage, GenericXLogState
* Add to element and neighbor pages
*/
static void
AddElementOnDisk(Relation index, HnswElement e, int m, BlockNumber insertPage, BlockNumber *updatedInsertPage, bool building)
WriteNewElementPages(Relation index, HnswElement e, int m, BlockNumber insertPage, BlockNumber *updatedInsertPage, bool building)
{
Buffer buf;
Page page;
@@ -340,7 +339,7 @@ ConnectionExists(HnswElement e, HnswNeighborTuple ntup, int startIdx, int lm)
* Update neighbors
*/
void
HnswUpdateNeighborsOnDisk(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, bool building)
{
char *base = NULL;
@@ -452,7 +451,7 @@ HnswUpdateNeighborsOnDisk(Relation index, FmgrInfo *procinfo, Oid collation, Hns
* Add a heap TID to an existing element
*/
static bool
AddDuplicateOnDisk(Relation index, HnswElement element, HnswElement dup, bool building)
HnswAddDuplicate(Relation index, HnswElement element, HnswElement dup, bool building)
{
Buffer buf;
Page page;
@@ -516,23 +515,23 @@ AddDuplicateOnDisk(Relation index, HnswElement element, HnswElement dup, bool bu
* Find duplicate element
*/
static bool
FindDuplicateOnDisk(Relation index, HnswElement element, bool building)
HnswFindDuplicate(Relation index, HnswElement element, bool building)
{
char *base = NULL;
HnswNeighborArray *neighbors = HnswGetNeighbors(base, element, 0);
Datum value = HnswGetValue(base, element);
for (int i = 0; i < neighbors->length; i++)
{
HnswCandidate *neighbor = &neighbors->items[i];
HnswElement neighborElement = HnswPtrAccess(base, neighbor->element);
Datum value = HnswGetValue(base, element);
Datum neighborValue = HnswGetValue(base, neighborElement);
/* Exit early since ordered by distance */
if (!datumIsEqual(value, neighborValue, false, -1))
return false;
if (AddDuplicateOnDisk(index, element, neighborElement, building))
if (HnswAddDuplicate(index, element, neighborElement, building))
return true;
}
@@ -540,26 +539,26 @@ FindDuplicateOnDisk(Relation index, HnswElement element, bool building)
}
/*
* Update graph on disk
* Write changes to disk
*/
static void
UpdateGraphOnDisk(Relation index, FmgrInfo *procinfo, Oid collation, HnswElement element, int m, int efConstruction, HnswElement entryPoint, bool building)
WriteElement(Relation index, FmgrInfo *procinfo, Oid collation, HnswElement element, int m, int efConstruction, HnswElement entryPoint, bool building)
{
BlockNumber newInsertPage = InvalidBlockNumber;
/* Look for duplicate */
if (FindDuplicateOnDisk(index, element, building))
if (HnswFindDuplicate(index, element, building))
return;
/* Add element */
AddElementOnDisk(index, element, m, GetInsertPage(index), &newInsertPage, building);
/* Write element and neighbor tuples */
WriteNewElementPages(index, element, m, GetInsertPage(index), &newInsertPage, building);
/* Update insert page if needed */
if (BlockNumberIsValid(newInsertPage))
HnswUpdateMetaPage(index, 0, NULL, newInsertPage, MAIN_FORKNUM, building);
/* Update neighbors */
HnswUpdateNeighborsOnDisk(index, procinfo, collation, element, m, false, building);
HnswUpdateNeighborPages(index, procinfo, collation, element, m, false, building);
/* Update entry point if needed */
if (entryPoint == NULL || element->level > entryPoint->level)
@@ -609,11 +608,11 @@ HnswInsertTupleOnDisk(Relation index, Datum value, Datum *values, bool *isnull,
entryPoint = HnswGetEntryPoint(index);
}
/* Find neighbors for element */
HnswFindElementNeighbors(base, element, entryPoint, index, procinfo, collation, m, efConstruction, false);
/* Insert element in graph */
HnswInsertElement(base, element, entryPoint, index, procinfo, collation, m, efConstruction, false);
/* Update graph on disk */
UpdateGraphOnDisk(index, procinfo, collation, element, m, efConstruction, entryPoint, building);
/* Write to disk */
WriteElement(index, procinfo, collation, element, m, efConstruction, entryPoint, building);
/* Release lock */
UnlockPage(index, HNSW_UPDATE_LOCK, lockmode);

View File

@@ -201,7 +201,12 @@ hnswgettuple(IndexScanDesc scan, ScanDirection dir)
MemoryContextSwitchTo(oldCtx);
#if PG_VERSION_NUM >= 120000
scan->xs_heaptid = *heaptid;
#else
scan->xs_ctup.t_self = *heaptid;
#endif
scan->xs_recheckorderby = false;
return true;
}

View File

@@ -2,12 +2,9 @@
#include <math.h>
#include "access/generic_xlog.h"
#include "hnsw.h"
#include "lib/pairingheap.h"
#include "storage/bufmgr.h"
#include "utils/datum.h"
#include "utils/rel.h"
#include "vector.h"
#if PG_VERSION_NUM >= 130000
@@ -59,6 +56,13 @@ hash_tid(ItemPointerData tid)
#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)
@@ -80,6 +84,15 @@ hash_pointer(uintptr_t ptr)
#define SH_DEFINE
#include "lib/simplehash.h"
/* Needed to include simplehash.h again */
#if PG_VERSION_NUM < 120000
#undef SH_EQUAL
#undef sh_log2
#undef sh_pow2
#define sh_log2 offsethash_sh_log2
#define sh_pow2 offsethash_sh_pow2
#endif
/* Offset hash table */
static uint32
hash_offset(Size offset)
@@ -643,25 +656,11 @@ CreatePairingHeapNode(HnswCandidate * c)
return node;
}
/*
* Init visited
*/
static inline void
InitVisited(char *base, visited_hash * v, Relation index, int ef, int m)
{
if (index != NULL)
v->tids = tidhash_create(CurrentMemoryContext, ef * m * 2, NULL);
else if (base != NULL)
v->offsets = offsethash_create(CurrentMemoryContext, ef * m * 2, NULL);
else
v->pointers = pointerhash_create(CurrentMemoryContext, ef * m * 2, NULL);
}
/*
* Add to visited
*/
static inline void
AddToVisited(char *base, visited_hash * v, HnswCandidate * hc, Relation index, bool *found)
AddToVisited(char *base, visited_hash v, HnswCandidate * hc, Relation index, bool *found)
{
if (index != NULL)
{
@@ -669,16 +668,16 @@ AddToVisited(char *base, visited_hash * v, HnswCandidate * hc, Relation index, b
ItemPointerData indextid;
ItemPointerSet(&indextid, element->blkno, element->offno);
tidhash_insert(v->tids, indextid, found);
tidhash_insert(v.tids, indextid, found);
}
else if (base != NULL)
{
#if PG_VERSION_NUM >= 130000
HnswElement element = HnswPtrAccess(base, hc->element);
offsethash_insert_hash(v->offsets, HnswPtrOffset(hc->element), element->hash, found);
offsethash_insert_hash(v.offsets, HnswPtrOffset(hc->element), element->hash, found);
#else
offsethash_insert(v->offsets, HnswPtrOffset(hc->element), found);
offsethash_insert(v.offsets, HnswPtrOffset(hc->element), found);
#endif
}
else
@@ -686,9 +685,9 @@ AddToVisited(char *base, visited_hash * v, HnswCandidate * hc, Relation index, b
#if PG_VERSION_NUM >= 130000
HnswElement element = HnswPtrAccess(base, hc->element);
pointerhash_insert_hash(v->pointers, (uintptr_t) HnswPtrPointer(hc->element), element->hash, found);
pointerhash_insert_hash(v.pointers, (uintptr_t) HnswPtrPointer(hc->element), element->hash, found);
#else
pointerhash_insert(v->pointers, (uintptr_t) HnswPtrPointer(hc->element), found);
pointerhash_insert(v.pointers, (uintptr_t) HnswPtrPointer(hc->element), found);
#endif
}
}
@@ -726,7 +725,13 @@ HnswSearchLayer(char *base, Datum q, List *ep, int ef, int lc, Relation index, F
HnswNeighborArray *neighborhoodData = NULL;
Size neighborhoodSize;
InitVisited(base, &v, index, ef, m);
/* Create hash table */
if (index != NULL)
v.tids = tidhash_create(CurrentMemoryContext, ef * m * 2, NULL);
else if (base != NULL)
v.offsets = offsethash_create(CurrentMemoryContext, ef * m * 2, NULL);
else
v.pointers = pointerhash_create(CurrentMemoryContext, ef * m * 2, NULL);
/* Create local memory for neighborhood if needed */
if (index == NULL)
@@ -741,7 +746,7 @@ HnswSearchLayer(char *base, Datum q, List *ep, int ef, int lc, Relation index, F
HnswCandidate *hc = (HnswCandidate *) lfirst(lc2);
bool found;
AddToVisited(base, &v, hc, index, &found);
AddToVisited(base, v, hc, index, &found);
pairingheap_add(C, &(CreatePairingHeapNode(hc)->ph_node));
pairingheap_add(W, &(CreatePairingHeapNode(hc)->ph_node));
@@ -776,7 +781,7 @@ HnswSearchLayer(char *base, Datum q, List *ep, int ef, int lc, Relation index, F
/* Copy neighborhood to local memory if needed */
if (index == NULL)
{
LWLockAcquire(&cElement->lock, LW_SHARED);
LWLockAcquire(&cElement->lock, LW_EXCLUSIVE);
memcpy(neighborhoodData, neighborhood, neighborhoodSize);
LWLockRelease(&cElement->lock);
neighborhood = neighborhoodData;
@@ -787,7 +792,7 @@ HnswSearchLayer(char *base, Datum q, List *ep, int ef, int lc, Relation index, F
HnswCandidate *e = &neighborhood->items[i];
bool visited;
AddToVisited(base, &v, e, index, &visited);
AddToVisited(base, v, e, index, &visited);
if (!visited)
{
@@ -958,6 +963,7 @@ CheckElementCloser(char *base, HnswCandidate * e, List *r, int lc, FmgrInfo *pro
{
HnswCandidate *ri = lfirst(lc2);
HnswElement riElement = HnswPtrAccess(base, ri->element);
float distance = HnswGetDistance(base, eElement, riElement, lc, procinfo, collation);
if (distance <= e->distance)
@@ -971,7 +977,7 @@ CheckElementCloser(char *base, HnswCandidate * e, List *r, int lc, FmgrInfo *pro
* Algorithm 4 from paper
*/
static List *
SelectNeighbors(char *base, List *c, int lm, int lc, FmgrInfo *procinfo, Oid collation, HnswElement e2, HnswCandidate * newCandidate, HnswCandidate * *pruned, bool sortCandidates)
SelectNeighbors(char *base, List *c, int m, int lc, FmgrInfo *procinfo, Oid collation, HnswElement e2, HnswCandidate * newCandidate, HnswCandidate * *pruned, bool sortCandidates)
{
List *r = NIL;
List *w = list_copy(c);
@@ -981,7 +987,7 @@ SelectNeighbors(char *base, List *c, int lm, int lc, FmgrInfo *procinfo, Oid col
List *added = NIL;
bool removedAny = false;
if (list_length(w) <= lm)
if (list_length(w) <= m)
return w;
wd = pairingheap_allocate(CompareNearestCandidates, NULL);
@@ -995,7 +1001,7 @@ SelectNeighbors(char *base, List *c, int lm, int lc, FmgrInfo *procinfo, Oid col
list_sort(w, CompareCandidateDistancesOffset);
}
while (list_length(w) > 0 && list_length(r) < lm)
while (list_length(w) > 0 && list_length(r) < m)
{
/* Assumes w is already ordered desc */
HnswCandidate *e = llast(w);
@@ -1049,7 +1055,7 @@ SelectNeighbors(char *base, List *c, int lm, int lc, FmgrInfo *procinfo, Oid col
neighbors->closerSet = sortCandidates;
/* Keep pruned connections */
while (!pairingheap_is_empty(wd) && list_length(r) < lm)
while (!pairingheap_is_empty(wd) && list_length(r) < m)
r = lappend(r, ((HnswPairingHeapNode *) pairingheap_remove_first(wd))->inner);
/* Return pruned for update connections */
@@ -1068,7 +1074,7 @@ SelectNeighbors(char *base, List *c, int lm, int lc, FmgrInfo *procinfo, Oid col
* Add connections
*/
static void
AddConnections(char *base, HnswElement element, List *neighbors, int lc)
AddConnections(char *base, HnswElement element, List *neighbors, int m, int lc)
{
ListCell *lc2;
HnswNeighborArray *a = HnswGetNeighbors(base, element, lc);
@@ -1081,7 +1087,7 @@ AddConnections(char *base, HnswElement element, List *neighbors, int lc)
* Update connections
*/
void
HnswUpdateConnection(char *base, HnswElement element, HnswCandidate * hc, int lm, int lc, int *updateIdx, Relation index, FmgrInfo *procinfo, Oid collation)
HnswUpdateConnection(char *base, HnswElement element, HnswCandidate * hc, int m, int lc, int *updateIdx, Relation index, FmgrInfo *procinfo, Oid collation)
{
HnswElement hce = HnswPtrAccess(base, hc->element);
HnswNeighborArray *currentNeighbors = HnswGetNeighbors(base, hce, lc);
@@ -1090,7 +1096,7 @@ HnswUpdateConnection(char *base, HnswElement element, HnswCandidate * hc, int lm
HnswPtrStore(base, hc2.element, element);
hc2.distance = hc->distance;
if (currentNeighbors->length < lm)
if (currentNeighbors->length < m)
{
currentNeighbors->items[currentNeighbors->length++] = hc2;
@@ -1136,7 +1142,7 @@ HnswUpdateConnection(char *base, HnswElement element, HnswCandidate * hc, int lm
c = lappend(c, &currentNeighbors->items[i]);
c = lappend(c, &hc2);
SelectNeighbors(base, c, lm, lc, procinfo, collation, hce, &hc2, &pruned, true);
SelectNeighbors(base, c, m, lc, procinfo, collation, hce, &hc2, &pruned, true);
/* Should not happen */
if (pruned == NULL)
@@ -1210,7 +1216,7 @@ PrecomputeHash(char *base, HnswElement element)
* Algorithm 1 from paper
*/
void
HnswFindElementNeighbors(char *base, HnswElement element, HnswElement entryPoint, Relation index, FmgrInfo *procinfo, Oid collation, int m, int efConstruction, bool existing)
HnswInsertElement(char *base, HnswElement element, HnswElement entryPoint, Relation index, FmgrInfo *procinfo, Oid collation, int m, int efConstruction, bool existing)
{
List *ep;
List *w;
@@ -1270,7 +1276,7 @@ HnswFindElementNeighbors(char *base, HnswElement element, HnswElement entryPoint
*/
neighbors = SelectNeighbors(base, lw, lm, lc, procinfo, collation, element, NULL, NULL, false);
AddConnections(base, element, neighbors, lc);
AddConnections(base, element, neighbors, lm, lc);
ep = w;
}

View File

@@ -2,7 +2,6 @@
#include <math.h>
#include "access/generic_xlog.h"
#include "commands/vacuum.h"
#include "hnsw.h"
#include "storage/bufmgr.h"
@@ -200,22 +199,21 @@ RepairGraphElement(HnswVacuumState * vacuumstate, HnswElement element, HnswEleme
BufferAccessStrategy bas = vacuumstate->bas;
HnswNeighborTuple ntup = vacuumstate->ntup;
Size ntupSize = HNSW_NEIGHBOR_TUPLE_SIZE(element->level, m);
char *base = NULL;
/* Skip if element is entry point */
if (entryPoint != NULL && element->blkno == entryPoint->blkno && element->offno == entryPoint->offno)
return;
/* Init fields */
HnswInitNeighbors(base, element, m, NULL);
HnswInitNeighbors(NULL, element, m, NULL);
element->heaptidsLength = 0;
/* Find neighbors for element, skipping itself */
HnswFindElementNeighbors(base, element, entryPoint, index, procinfo, collation, m, efConstruction, true);
/* Add element to graph, skipping itself */
HnswInsertElement(NULL, element, entryPoint, index, procinfo, collation, m, efConstruction, true);
/* Update neighbor tuple */
/* Do this before getting page to minimize locking */
HnswSetNeighborTuple(base, ntup, element, m);
HnswSetNeighborTuple(NULL, ntup, element, m);
/* Get neighbor page */
buf = ReadBufferExtended(index, MAIN_FORKNUM, element->neighborPage, RBM_NORMAL, bas);
@@ -232,7 +230,7 @@ RepairGraphElement(HnswVacuumState * vacuumstate, HnswElement element, HnswEleme
UnlockReleaseBuffer(buf);
/* Update neighbors */
HnswUpdateNeighborsOnDisk(index, procinfo, collation, element, m, true, false);
HnswUpdateNeighborPages(index, procinfo, collation, element, m, true, false);
}
/*

View File

@@ -2,40 +2,58 @@
#include <float.h>
#include "access/table.h"
#include "access/tableam.h"
#include "access/parallel.h"
#include "access/xact.h"
#include "catalog/index.h"
#include "catalog/pg_operator_d.h"
#include "catalog/pg_type_d.h"
#include "commands/progress.h"
#include "ivfflat.h"
#include "miscadmin.h"
#include "optimizer/optimizer.h"
#include "storage/bufmgr.h"
#include "tcop/tcopprot.h"
#include "utils/memutils.h"
#if PG_VERSION_NUM >= 140000
#include "utils/backend_progress.h"
#else
#elif PG_VERSION_NUM >= 120000
#include "pgstat.h"
#endif
#if PG_VERSION_NUM >= 120000
#include "access/tableam.h"
#include "commands/progress.h"
#else
#define PROGRESS_CREATEIDX_SUBPHASE 0
#define PROGRESS_CREATEIDX_TUPLES_TOTAL 0
#define PROGRESS_CREATEIDX_TUPLES_DONE 0
#endif
#if PG_VERSION_NUM >= 130000
#define CALLBACK_ITEM_POINTER ItemPointer tid
#else
#define CALLBACK_ITEM_POINTER HeapTuple hup
#endif
#if PG_VERSION_NUM >= 120000
#define UpdateProgress(index, val) pgstat_progress_update_param(index, val)
#else
#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_IVFFLAT_SHARED UINT64CONST(0xA000000000000001)
#define PARALLEL_KEY_TUPLESORT UINT64CONST(0xA000000000000002)
#define PARALLEL_KEY_IVFFLAT_CENTERS UINT64CONST(0xA000000000000003)
@@ -132,8 +150,13 @@ SampleRows(IvfflatBuildState * buildstate)
{
BlockNumber targblock = BlockSampler_Next(&buildstate->bs);
#if PG_VERSION_NUM >= 120000
table_index_build_range_scan(buildstate->heap, buildstate->index, buildstate->indexInfo,
false, true, false, targblock, 1, SampleCallback, (void *) buildstate, NULL);
#else
IndexBuildHeapRangeScan(buildstate->heap, buildstate->index, buildstate->indexInfo,
false, true, targblock, 1, SampleCallback, (void *) buildstate, NULL);
#endif
}
}
@@ -259,7 +282,11 @@ InsertTuples(Relation index, IvfflatBuildState * buildstate, ForkNumber forkNum)
IndexTuple itup = NULL; /* silence compiler warning */
int64 inserted = 0;
#if PG_VERSION_NUM >= 120000
TupleTableSlot *slot = MakeSingleTupleTableSlot(buildstate->tupdesc, &TTSOpsMinimalTuple);
#else
TupleTableSlot *slot = MakeSingleTupleTableSlot(buildstate->tupdesc);
#endif
TupleDesc tupdesc = RelationGetDescr(index);
UpdateProgress(PROGRESS_CREATEIDX_SUBPHASE, PROGRESS_IVFFLAT_PHASE_LOAD);
@@ -348,12 +375,20 @@ InitBuildState(IvfflatBuildState * buildstate, Relation heap, Relation index, In
elog(ERROR, "dimensions must be greater than one for this opclass");
/* Create tuple description for sorting */
#if PG_VERSION_NUM >= 120000
buildstate->tupdesc = CreateTemplateTupleDesc(3);
#else
buildstate->tupdesc = CreateTemplateTupleDesc(3, false);
#endif
TupleDescInitEntry(buildstate->tupdesc, (AttrNumber) 1, "list", INT4OID, -1, 0);
TupleDescInitEntry(buildstate->tupdesc, (AttrNumber) 2, "tid", TIDOID, -1, 0);
TupleDescInitEntry(buildstate->tupdesc, (AttrNumber) 3, "vector", RelationGetDescr(index)->attrs[0].atttypid, -1, 0);
#if PG_VERSION_NUM >= 120000
buildstate->slot = MakeSingleTupleTableSlot(buildstate->tupdesc, &TTSOpsVirtual);
#else
buildstate->slot = MakeSingleTupleTableSlot(buildstate->tupdesc);
#endif
buildstate->centers = VectorArrayInit(buildstate->lists, buildstate->dimensions);
buildstate->listInfo = palloc(sizeof(ListInfo) * buildstate->lists);
@@ -596,7 +631,11 @@ IvfflatParallelScanAndSort(IvfflatSpool * ivfspool, IvfflatShared * ivfshared, S
{
SortCoordinate coordinate;
IvfflatBuildState buildstate;
#if PG_VERSION_NUM >= 120000
TableScanDesc scan;
#else
HeapScanDesc scan;
#endif
double reltuples;
IndexInfo *indexInfo;
@@ -620,11 +659,18 @@ IvfflatParallelScanAndSort(IvfflatSpool * ivfspool, IvfflatShared * ivfshared, S
buildstate.centers->length = buildstate.centers->maxlen;
ivfspool->sortstate = tuplesort_begin_heap(buildstate.tupdesc, 1, attNums, sortOperators, sortCollations, nullsFirstFlags, sortmem, coordinate, false);
buildstate.sortstate = ivfspool->sortstate;
#if PG_VERSION_NUM >= 120000
scan = table_beginscan_parallel(ivfspool->heap,
ParallelTableScanFromIvfflatShared(ivfshared));
reltuples = table_index_build_scan(ivfspool->heap, ivfspool->index, indexInfo,
true, progress, BuildCallback,
(void *) &buildstate, scan);
#else
scan = heap_beginscan_parallel(ivfspool->heap, &ivfshared->heapdesc);
reltuples = IndexBuildHeapScan(ivfspool->heap, ivfspool->index, indexInfo,
true, BuildCallback,
(void *) &buildstate, scan);
#endif
/* Execute this worker's part of the sort */
tuplesort_performsort(ivfspool->sortstate);
@@ -694,7 +740,11 @@ IvfflatParallelBuildMain(dsm_segment *seg, shm_toc *toc)
}
/* Open relations within worker */
#if PG_VERSION_NUM >= 120000
heapRel = table_open(ivfshared->heaprelid, heapLockmode);
#else
heapRel = heap_open(ivfshared->heaprelid, heapLockmode);
#endif
indexRel = index_open(ivfshared->indexrelid, indexLockmode);
/* Initialize worker's own spool */
@@ -714,7 +764,11 @@ IvfflatParallelBuildMain(dsm_segment *seg, shm_toc *toc)
/* Close relations within worker */
index_close(indexRel, indexLockmode);
#if PG_VERSION_NUM >= 120000
table_close(heapRel, heapLockmode);
#else
heap_close(heapRel, heapLockmode);
#endif
}
/*
@@ -739,7 +793,19 @@ IvfflatEndParallel(IvfflatLeader * ivfleader)
static Size
ParallelEstimateShared(Relation heap, Snapshot snapshot)
{
#if PG_VERSION_NUM >= 120000
return add_size(BUFFERALIGN(sizeof(IvfflatShared)), table_parallelscan_estimate(heap, snapshot));
#else
if (!IsMVCCSnapshot(snapshot))
{
Assert(snapshot == SnapshotAny);
return sizeof(IvfflatShared);
}
return add_size(offsetof(IvfflatShared, heapdesc) +
offsetof(ParallelHeapScanDescData, phs_snapshot_data),
EstimateSnapshotSpace(snapshot));
#endif
}
/*
@@ -790,7 +856,11 @@ IvfflatBeginParallel(IvfflatBuildState * buildstate, bool isconcurrent, int requ
/* Enter parallel mode and create context */
EnterParallelMode();
Assert(request > 0);
#if PG_VERSION_NUM >= 120000
pcxt = CreateParallelContext("vector", "IvfflatParallelBuildMain", request);
#else
pcxt = CreateParallelContext("vector", "IvfflatParallelBuildMain", request, true);
#endif
scantuplesortstates = leaderparticipates ? request + 1 : request;
@@ -848,9 +918,13 @@ IvfflatBeginParallel(IvfflatBuildState * buildstate, bool isconcurrent, int requ
#ifdef IVFFLAT_KMEANS_DEBUG
ivfshared->inertia = 0;
#endif
#if PG_VERSION_NUM >= 120000
table_parallelscan_initialize(buildstate->heap,
ParallelTableScanFromIvfflatShared(ivfshared),
snapshot);
#else
heap_parallelscan_initialize(&ivfshared->heapdesc, buildstate->heap, snapshot);
#endif
/* Store shared tuplesort-private state, for which we reserved space */
sharedsort = (Sharedsort *) shm_toc_allocate(pcxt->toc, estsort);
@@ -949,8 +1023,15 @@ AssignTuples(IvfflatBuildState * buildstate)
if (buildstate->ivfleader)
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);
#else
buildstate->reltuples = IndexBuildHeapScan(buildstate->heap, buildstate->index, buildstate->indexInfo,
true, BuildCallback, (void *) buildstate, NULL);
#endif
}
#ifdef IVFFLAT_KMEANS_DEBUG
PrintKmeansMetrics(buildstate);

View File

@@ -3,16 +3,14 @@
#include <float.h>
#include "access/amapi.h"
#include "access/reloptions.h"
#include "commands/progress.h"
#include "commands/vacuum.h"
#include "ivfflat.h"
#include "utils/guc.h"
#include "utils/selfuncs.h"
#include "utils/spccache.h"
#if PG_VERSION_NUM < 150000
#define MarkGUCPrefixReserved(x) EmitWarningsOnPlaceholders(x)
#if PG_VERSION_NUM >= 120000
#include "commands/progress.h"
#endif
int ivfflat_probes;
@@ -35,13 +33,12 @@ IvfflatInit(void)
DefineCustomIntVariable("ivfflat.probes", "Sets the number of probes",
"Valid range is 1..lists.", &ivfflat_probes,
IVFFLAT_DEFAULT_PROBES, IVFFLAT_MIN_LISTS, IVFFLAT_MAX_LISTS, PGC_USERSET, 0, NULL, NULL, NULL);
MarkGUCPrefixReserved("ivfflat");
}
/*
* Get the name of index build phase
*/
#if PG_VERSION_NUM >= 120000
static char *
ivfflatbuildphasename(int64 phasenum)
{
@@ -59,6 +56,7 @@ ivfflatbuildphasename(int64 phasenum)
return NULL;
}
}
#endif
/*
* Estimate the cost of an index scan
@@ -74,6 +72,9 @@ ivfflatcostestimate(PlannerInfo *root, IndexPath *path, double loop_count,
double ratio;
double spc_seq_page_cost;
Relation index;
#if PG_VERSION_NUM < 120000
List *qinfos;
#endif
/* Never use index without order */
if (path->indexorderbys == NULL)
@@ -104,7 +105,12 @@ ivfflatcostestimate(PlannerInfo *root, IndexPath *path, double loop_count,
*/
costs.numIndexTuples = path->indexinfo->tuples * ratio;
#if PG_VERSION_NUM >= 120000
genericcostestimate(root, path, loop_count, &costs);
#else
qinfos = deconstruct_indexquals(path);
genericcostestimate(root, path, loop_count, qinfos, &costs);
#endif
get_tablespace_page_costs(path->indexinfo->reltablespace, NULL, &spc_seq_page_cost);
@@ -221,7 +227,9 @@ ivfflathandler(PG_FUNCTION_ARGS)
amroutine->amcostestimate = ivfflatcostestimate;
amroutine->amoptions = ivfflatoptions;
amroutine->amproperty = NULL; /* TODO AMPROP_DISTANCE_ORDERABLE */
#if PG_VERSION_NUM >= 120000
amroutine->ambuildphasename = ivfflatbuildphasename;
#endif
amroutine->amvalidate = ivfflatvalidate;
#if PG_VERSION_NUM >= 140000
amroutine->amadjustmembers = NULL;

View File

@@ -3,10 +3,9 @@
#include "postgres.h"
#include "access/genam.h"
#include "access/generic_xlog.h"
#include "access/parallel.h"
#include "lib/pairingheap.h"
#include "access/reloptions.h"
#include "nodes/execnodes.h"
#include "port.h" /* for random() */
#include "utils/sampling.h"
@@ -17,6 +16,10 @@
#include "common/pg_prng.h"
#endif
#if PG_VERSION_NUM < 120000
#include "access/relscan.h"
#endif
#ifdef IVFFLAT_BENCH
#include "portability/instr_time.h"
#endif
@@ -132,10 +135,16 @@ typedef struct IvfflatShared
#ifdef IVFFLAT_KMEANS_DEBUG
double inertia;
#endif
#if PG_VERSION_NUM < 120000
ParallelHeapScanDescData heapdesc; /* must come last */
#endif
} IvfflatShared;
#if PG_VERSION_NUM >= 120000
#define ParallelTableScanFromIvfflatShared(shared) \
(ParallelTableScanDesc) ((char *) (shared) + BUFFERALIGN(sizeof(IvfflatShared)))
#endif
typedef struct IvfflatLeader
{

View File

@@ -2,7 +2,6 @@
#include <float.h>
#include "access/generic_xlog.h"
#include "ivfflat.h"
#include "storage/bufmgr.h"
#include "storage/lmgr.h"

View File

@@ -5,7 +5,6 @@
#include "access/relscan.h"
#include "catalog/pg_operator_d.h"
#include "catalog/pg_type_d.h"
#include "lib/pairingheap.h"
#include "ivfflat.h"
#include "miscadmin.h"
#include "pgstat.h"
@@ -106,7 +105,12 @@ GetScanItems(IndexScanDesc scan, Datum value)
IvfflatScanOpaque so = (IvfflatScanOpaque) scan->opaque;
TupleDesc tupdesc = RelationGetDescr(scan->indexRelation);
double tuples = 0;
#if PG_VERSION_NUM >= 120000
TupleTableSlot *slot = MakeSingleTupleTableSlot(so->tupdesc, &TTSOpsVirtual);
#else
TupleTableSlot *slot = MakeSingleTupleTableSlot(so->tupdesc);
#endif
/*
* Reuse same set of shared buffers for scan
@@ -212,14 +216,22 @@ ivfflatbeginscan(Relation index, int nkeys, int norderbys)
so->collation = index->rd_indcollation[0];
/* Create tuple description for sorting */
#if PG_VERSION_NUM >= 120000
so->tupdesc = CreateTemplateTupleDesc(2);
#else
so->tupdesc = CreateTemplateTupleDesc(2, false);
#endif
TupleDescInitEntry(so->tupdesc, (AttrNumber) 1, "distance", FLOAT8OID, -1, 0);
TupleDescInitEntry(so->tupdesc, (AttrNumber) 2, "heaptid", TIDOID, -1, 0);
/* Prep sort */
so->sortstate = tuplesort_begin_heap(so->tupdesc, 1, attNums, sortOperators, sortCollations, nullsFirstFlags, work_mem, NULL, false);
#if PG_VERSION_NUM >= 120000
so->slot = MakeSingleTupleTableSlot(so->tupdesc, &TTSOpsMinimalTuple);
#else
so->slot = MakeSingleTupleTableSlot(so->tupdesc);
#endif
so->listQueue = pairingheap_allocate(CompareLists, scan);
@@ -309,7 +321,12 @@ ivfflatgettuple(IndexScanDesc scan, ScanDirection dir)
{
ItemPointer heaptid = (ItemPointer) DatumGetPointer(slot_getattr(so->slot, 2, &so->isnull));
#if PG_VERSION_NUM >= 120000
scan->xs_heaptid = *heaptid;
#else
scan->xs_ctup.t_self = *heaptid;
#endif
scan->xs_recheckorderby = false;
return true;
}

View File

@@ -1,6 +1,5 @@
#include "postgres.h"
#include "access/generic_xlog.h"
#include "ivfflat.h"
#include "storage/bufmgr.h"
#include "vector.h"

View File

@@ -1,6 +1,5 @@
#include "postgres.h"
#include "access/generic_xlog.h"
#include "commands/vacuum.h"
#include "ivfflat.h"
#include "storage/bufmgr.h"

View File

@@ -3,17 +3,14 @@
#include <math.h>
#include "catalog/pg_type.h"
#include "common/shortest_dec.h"
#include "fmgr.h"
#include "hnsw.h"
#include "ivfflat.h"
#include "lib/stringinfo.h"
#include "libpq/pqformat.h"
#include "port.h" /* for strtof() */
#include "port/pg_bitutils.h"
#include "utils/array.h"
#include "utils/builtins.h"
#include "utils/float.h"
#include "utils/lsyscache.h"
#include "utils/numeric.h"
#include "vector.h"
@@ -22,6 +19,13 @@
#include "varatt.h"
#endif
#if PG_VERSION_NUM >= 120000
#include "common/shortest_dec.h"
#include "utils/float.h"
#else
#include <float.h>
#endif
#if PG_VERSION_NUM < 130000
#define TYPALIGN_DOUBLE 'd'
#define TYPALIGN_INT 'i'
@@ -173,15 +177,14 @@ PGDLLEXPORT PG_FUNCTION_INFO_V1(vector_in);
Datum
vector_in(PG_FUNCTION_ARGS)
{
char *lit = PG_GETARG_CSTRING(0);
char *str = PG_GETARG_CSTRING(0);
int32 typmod = PG_GETARG_INT32(2);
float x[VECTOR_MAX_DIM];
int dim = 0;
char *pt;
char *stringEnd;
Vector *result;
char *litcopy = pstrdup(lit);
char *str = litcopy;
char *lit = pstrdup(str);
while (vector_isspace(*str))
str++;
@@ -265,7 +268,7 @@ vector_in(PG_FUNCTION_ARGS)
(errcode(ERRCODE_DATA_EXCEPTION),
errmsg("vector must have at least 1 dimension")));
pfree(litcopy);
pfree(lit);
CheckExpectedDim(typmod, dim);
@@ -289,6 +292,15 @@ vector_out(PG_FUNCTION_ARGS)
char *ptr;
int n;
#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
/*
* Need:
*
@@ -312,7 +324,11 @@ vector_out(PG_FUNCTION_ARGS)
ptr++;
}
#if PG_VERSION_NUM >= 120000
n = float_to_shortest_decimal_bufn(vector->x[i], ptr);
#else
n = sprintf(ptr, "%.*g", ndig, vector->x[i]);
#endif
ptr += n;
}
*ptr = ']';
@@ -1135,43 +1151,3 @@ vector_avg(PG_FUNCTION_ARGS)
PG_RETURN_POINTER(result);
}
/*
* Ensure same number of bytes
*/
static inline void
CheckByteLengths(uint32 aLen, uint32 bLen)
{
if (aLen != bLen)
ereport(ERROR,
(errcode(ERRCODE_DATA_EXCEPTION),
errmsg("different byte lengths %u and %u", aLen, bLen)));
}
/*
* Get the Hamming distance between two binary strings
*/
PGDLLEXPORT PG_FUNCTION_INFO_V1(hamming_distance);
Datum
hamming_distance(PG_FUNCTION_ARGS)
{
bytea *a = PG_GETARG_BYTEA_PP(0);
bytea *b = PG_GETARG_BYTEA_PP(1);
char *ax = VARDATA_ANY(a);
char *bx = VARDATA_ANY(b);
uint32 aLen = VARSIZE_ANY_EXHDR(a);
uint32 bLen = VARSIZE_ANY_EXHDR(b);
uint64 distance = 0;
CheckByteLengths(aLen, bLen);
for (uint32 i = 0; i < aLen; i++)
{
unsigned char diff = (unsigned char) (ax[i] ^ bx[i]);
distance += pg_number_of_ones[diff];
}
/* TODO Decide on return type */
PG_RETURN_FLOAT8((double) distance);
}

View File

@@ -158,32 +158,6 @@ SELECT l1_distance('[3e38]', '[-3e38]');
Infinity
(1 row)
SELECT hamming_distance('\xFFFF', '\xFFFF');
hamming_distance
------------------
0
(1 row)
SELECT hamming_distance('\xFFFF', '\xFFFE');
hamming_distance
------------------
1
(1 row)
SELECT hamming_distance('\xFFFF', '\xFFFC');
hamming_distance
------------------
2
(1 row)
SELECT hamming_distance('\xFFFF', '\x0000');
hamming_distance
------------------
16
(1 row)
SELECT hamming_distance('\xFFFF', '\x00');
ERROR: different byte lengths 2 and 1
SELECT avg(v) FROM unnest(ARRAY['[1,2,3]'::vector, '[3,5,7]']) v;
avg
-----------

View File

@@ -37,12 +37,6 @@ SELECT l1_distance('[0,0]', '[0,1]');
SELECT l1_distance('[1,2]', '[3]');
SELECT l1_distance('[3e38]', '[-3e38]');
SELECT hamming_distance('\xFFFF', '\xFFFF');
SELECT hamming_distance('\xFFFF', '\xFFFE');
SELECT hamming_distance('\xFFFF', '\xFFFC');
SELECT hamming_distance('\xFFFF', '\x0000');
SELECT hamming_distance('\xFFFF', '\x00');
SELECT avg(v) FROM unnest(ARRAY['[1,2,3]'::vector, '[3,5,7]']) v;
SELECT avg(v) FROM unnest(ARRAY['[1,2,3]'::vector, '[3,5,7]', NULL]) v;
SELECT avg(v) FROM unnest(ARRAY[]::vector[]) v;

View File

@@ -1,114 +0,0 @@
use strict;
use warnings;
use PostgresNode;
use TestLib;
use Test::More;
my $dim = 3;
my $nc = 50;
my $limit = 20;
my $array_sql = join(",", ('random()') x $dim);
# Initialize node
my $node = get_new_node('node');
$node->init;
$node->start;
# Create table and index
$node->safe_psql("postgres", "CREATE EXTENSION vector;");
$node->safe_psql("postgres", "CREATE TABLE tst (i int4, v vector($dim), c int4, t text);");
$node->safe_psql("postgres",
"INSERT INTO tst SELECT i, ARRAY[$array_sql], i % $nc, 'test ' || i FROM generate_series(1, 10000) i;"
);
$node->safe_psql("postgres", "CREATE INDEX idx ON tst USING hnsw (v vector_l2_ops);");
$node->safe_psql("postgres", "ANALYZE tst;");
# Generate query
my @r = ();
for (1 .. $dim)
{
push(@r, rand());
}
my $query = "[" . join(",", @r) . "]";
my $c = int(rand() * $nc);
# Test attribute filtering
my $explain = $node->safe_psql("postgres", qq(
EXPLAIN ANALYZE SELECT i FROM tst WHERE c = $c ORDER BY v <-> '$query' LIMIT $limit;
));
# TODO Do not use index
like($explain, qr/Index Scan using idx/);
# Test attribute filtering with few rows removed
$explain = $node->safe_psql("postgres", qq(
EXPLAIN ANALYZE SELECT i FROM tst WHERE c != $c ORDER BY v <-> '$query' LIMIT $limit;
));
like($explain, qr/Index Scan using idx/);
# Test attribute filtering with few rows removed comparison
$explain = $node->safe_psql("postgres", qq(
EXPLAIN ANALYZE SELECT i FROM tst WHERE c >= 1 ORDER BY v <-> '$query' LIMIT $limit;
));
like($explain, qr/Index Scan using idx/);
# Test attribute filtering with many rows removed comparison
$explain = $node->safe_psql("postgres", qq(
EXPLAIN ANALYZE SELECT i FROM tst WHERE c < 1 ORDER BY v <-> '$query' LIMIT $limit;
));
# TODO Do not use index
like($explain, qr/Index Scan using idx/);
# Test attribute filtering with few rows removed like
$explain = $node->safe_psql("postgres", qq(
EXPLAIN ANALYZE SELECT i FROM tst WHERE t LIKE '%%test%%' ORDER BY v <-> '$query' LIMIT $limit;
));
like($explain, qr/Index Scan using idx/);
# Test attribute filtering with many rows removed like
$explain = $node->safe_psql("postgres", qq(
EXPLAIN ANALYZE SELECT i FROM tst WHERE t LIKE '%%other%%' ORDER BY v <-> '$query' LIMIT $limit;
));
like($explain, qr/Seq Scan/);
# Test distance filtering
$explain = $node->safe_psql("postgres", qq(
EXPLAIN ANALYZE SELECT i FROM tst WHERE v <-> '$query' < 1 ORDER BY v <-> '$query' LIMIT $limit;
));
like($explain, qr/Index Scan using idx/);
# Test distance filtering greater than distance
$explain = $node->safe_psql("postgres", qq(
EXPLAIN ANALYZE SELECT i FROM tst WHERE v <-> '$query' > 1 ORDER BY v <-> '$query' LIMIT $limit;
));
# TODO Do not use index
like($explain, qr/Index Scan using idx/);
# Test distance filtering without order
$explain = $node->safe_psql("postgres", qq(
EXPLAIN ANALYZE SELECT i FROM tst WHERE v <-> '$query' < 1;
));
like($explain, qr/Seq Scan/);
# Test distance filtering without limit
$explain = $node->safe_psql("postgres", qq(
EXPLAIN ANALYZE SELECT i FROM tst WHERE v <-> '$query' < 1 ORDER BY v <-> '$query';
));
like($explain, qr/Seq Scan/);
# Test attribute index
$node->safe_psql("postgres", "CREATE INDEX attribute_idx ON tst (c);");
$explain = $node->safe_psql("postgres", qq(
EXPLAIN ANALYZE SELECT i FROM tst WHERE c = $c ORDER BY v <-> '$query' LIMIT $limit;
));
# TODO Use attribute index
like($explain, qr/Index Scan using idx/);
# Test partial index
$node->safe_psql("postgres", "CREATE INDEX partial_idx ON tst USING hnsw (v vector_l2_ops) WHERE (c = $c);");
$explain = $node->safe_psql("postgres", qq(
EXPLAIN ANALYZE SELECT i FROM tst WHERE c = $c ORDER BY v <-> '$query' LIMIT $limit;
));
like($explain, qr/Index Scan using partial_idx/);
done_testing();

View File

@@ -0,0 +1,43 @@
use strict;
use warnings;
use PostgresNode;
use TestLib;
use Test::More;
my $dim = 3;
my $array_sql = join(",", ('random()') x $dim);
# Initialize node
my $node = get_new_node('node');
$node->init;
$node->start;
# Create table and index
$node->safe_psql("postgres", "CREATE EXTENSION vector;");
$node->safe_psql("postgres", "CREATE TABLE tst (i serial, v vector($dim));");
$node->safe_psql("postgres",
"INSERT INTO tst (v) SELECT ARRAY[$array_sql] FROM generate_series(1, 10000) i;"
);
$node->safe_psql("postgres", "CREATE INDEX ON tst USING ivfflat (v vector_l2_ops);");
# Delete data
$node->safe_psql("postgres", "DELETE FROM tst WHERE i % 100 != 0;");
my $exp = $node->safe_psql("postgres", qq(
SET enable_indexscan = off;
SELECT i FROM tst ORDER BY v <-> '[0,0,0]';
));
# Run twice to make sure correct tuples marked as dead
for (1 .. 2)
{
my $res = $node->safe_psql("postgres", qq(
SET enable_seqscan = off;
SET ivfflat.probes = 100;
SELECT i FROM tst ORDER BY v <-> '[0,0,0]';
));
is($res, $exp);
}
done_testing();

View File

@@ -1,116 +0,0 @@
use strict;
use warnings;
use PostgresNode;
use TestLib;
use Test::More;
my $dim = 3;
my $nc = 50;
my $limit = 20;
my $array_sql = join(",", ('random()') x $dim);
# Initialize node
my $node = get_new_node('node');
$node->init;
$node->start;
# Create table and index
$node->safe_psql("postgres", "CREATE EXTENSION vector;");
$node->safe_psql("postgres", "CREATE TABLE tst (i int4, v vector($dim), c int4, t text);");
$node->safe_psql("postgres",
"INSERT INTO tst SELECT i, ARRAY[$array_sql], i % $nc, 'test ' || i FROM generate_series(1, 10000) i;"
);
$node->safe_psql("postgres", "CREATE INDEX idx ON tst USING ivfflat (v vector_l2_ops) WITH (lists = 100);");
$node->safe_psql("postgres", "ANALYZE tst;");
# Generate query
my @r = ();
for (1 .. $dim)
{
push(@r, rand());
}
my $query = "[" . join(",", @r) . "]";
my $c = int(rand() * $nc);
# Test attribute filtering
my $explain = $node->safe_psql("postgres", qq(
EXPLAIN ANALYZE SELECT i FROM tst WHERE c = $c ORDER BY v <-> '$query' LIMIT $limit;
));
# TODO Do not use index
like($explain, qr/Index Scan using idx/);
# Test attribute filtering with few rows removed
$explain = $node->safe_psql("postgres", qq(
EXPLAIN ANALYZE SELECT i FROM tst WHERE c != $c ORDER BY v <-> '$query' LIMIT $limit;
));
like($explain, qr/Index Scan using idx/);
# Test attribute filtering with few rows removed comparison
$explain = $node->safe_psql("postgres", qq(
EXPLAIN ANALYZE SELECT i FROM tst WHERE c >= 1 ORDER BY v <-> '$query' LIMIT $limit;
));
like($explain, qr/Index Scan using idx/);
# Test attribute filtering with many rows removed comparison
$explain = $node->safe_psql("postgres", qq(
EXPLAIN ANALYZE SELECT i FROM tst WHERE c < 1 ORDER BY v <-> '$query' LIMIT $limit;
));
# TODO Do not use index
like($explain, qr/Index Scan using idx/);
# Test attribute filtering with few rows removed like
$explain = $node->safe_psql("postgres", qq(
EXPLAIN ANALYZE SELECT i FROM tst WHERE t LIKE '%%test%%' ORDER BY v <-> '$query' LIMIT $limit;
));
like($explain, qr/Index Scan using idx/);
# Test attribute filtering with many rows removed like
$explain = $node->safe_psql("postgres", qq(
EXPLAIN ANALYZE SELECT i FROM tst WHERE t LIKE '%%other%%' ORDER BY v <-> '$query' LIMIT $limit;
));
like($explain, qr/Seq Scan/);
# Test distance filtering
$explain = $node->safe_psql("postgres", qq(
EXPLAIN ANALYZE SELECT i FROM tst WHERE v <-> '$query' < 1 ORDER BY v <-> '$query' LIMIT $limit;
));
like($explain, qr/Index Scan using idx/);
# Test distance filtering greater than distance
$explain = $node->safe_psql("postgres", qq(
EXPLAIN ANALYZE SELECT i FROM tst WHERE v <-> '$query' > 1 ORDER BY v <-> '$query' LIMIT $limit;
));
# TODO Do not use index
like($explain, qr/Index Scan using idx/);
# Test distance filtering without order
$explain = $node->safe_psql("postgres", qq(
EXPLAIN ANALYZE SELECT i FROM tst WHERE v <-> '$query' < 1;
));
like($explain, qr/Seq Scan/);
# Test distance filtering without limit
$explain = $node->safe_psql("postgres", qq(
EXPLAIN ANALYZE SELECT i FROM tst WHERE v <-> '$query' < 1 ORDER BY v <-> '$query';
));
# TODO Do not use index
like($explain, qr/Index Scan using idx/);
# Test attribute index
$node->safe_psql("postgres", "CREATE INDEX attribute_idx ON tst (c);");
$explain = $node->safe_psql("postgres", qq(
EXPLAIN ANALYZE SELECT i FROM tst WHERE c = $c ORDER BY v <-> '$query' LIMIT $limit;
));
# TODO Use attribute index
like($explain, qr/Index Scan using idx/);
# Test partial index
$node->safe_psql("postgres", "CREATE INDEX partial_idx ON tst USING ivfflat (v vector_l2_ops) WITH (lists = 5) WHERE (c = $c);");
$explain = $node->safe_psql("postgres", qq(
EXPLAIN ANALYZE SELECT i FROM tst WHERE c = $c ORDER BY v <-> '$query' LIMIT $limit;
));
# TODO Use partial index
like($explain, qr/Index Scan using idx/);
done_testing();