Init leader

This reverts commit ef1209eaf4.

.github/workflows/build.yml

@@ -77,5 +77,24 @@ jobs:
           nmake /NOLOGO /F Makefile.win clean && ^
           nmake /NOLOGO /F Makefile.win uninstall
         shell: cmd
-      - if: ${{ failure() }}
-        run: cat regression.diffs
+  i386:
+    if: ${{ !startsWith(github.ref_name, 'mac') && !startsWith(github.ref_name, 'windows') }}
+    runs-on: ubuntu-latest
+    container:
+      image: debian:11
+      options: --platform linux/386
+    steps:
+      - 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
+          cd pgvector
+          git fetch origin ${{ github.ref }}
+          git reset --hard FETCH_HEAD
+          make
+          make install
+          chown -R postgres .
+          sudo -u postgres make installcheck
+          sudo -u postgres make prove_installcheck
+        env:
+          PG_CFLAGS: -Wall -Wextra -Werror -Wno-unused-parameter -Wno-sign-compare

CHANGELOG.md

@@ -1,14 +1,13 @@
-## 0.5.0 (unreleased)
+## 0.5.0 (2023-08-28)
 
 - Added HNSW index type
-- Added support for parallel index builds
+- Added support for parallel index builds for IVFFlat
 - Added `l1_distance` function
-- Added `normalize_l2` function
 - Added element-wise multiplication for vectors
 - Added `sum` aggregate
 - Improved performance of distance functions
 - Fixed out of range results for cosine distance
-- Fixed results for NULL and NaN distances
+- Fixed results for NULL and NaN distances for IVFFlat
 
 ## 0.4.4 (2023-06-12)
 

META.json

@@ -2,7 +2,7 @@
 	"name": "vector",
 	"abstract": "Open-source vector similarity search for Postgres",
 	"description": "Supports L2 distance, inner product, and cosine distance",
-	"version": "0.4.4",
+	"version": "0.5.0",
 	"maintainer": [
 		"Andrew Kane <andrew@ankane.org>"
 	],
@@ -20,7 +20,7 @@
 		"vector": {
 			"file": "sql/vector.sql",
 			"docfile": "README.md",
-			"version": "0.4.4",
+			"version": "0.5.0",
 			"abstract": "Open-source vector similarity search for Postgres"
 		}
 	},

Makefile

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

Makefile.win

@@ -1,7 +1,8 @@
 EXTENSION = vector
-EXTVERSION = 0.4.4
+EXTVERSION = 0.5.0
 
 OBJS = src\hnsw.obj src\hnswbuild.obj src\hnswinsert.obj src\hnswscan.obj src\hnswutils.obj src\hnswvacuum.obj src\ivfbuild.obj src\ivfflat.obj src\ivfinsert.obj src\ivfkmeans.obj src\ivfscan.obj src\ivfutils.obj src\ivfvacuum.obj src\vector.obj
+HEADERS = src\vector.h
 
 REGRESS = btree cast copy functions input ivfflat_cosine ivfflat_ip ivfflat_l2 ivfflat_options ivfflat_unlogged
 REGRESS_OPTS = --inputdir=test --load-extension=vector
@@ -54,6 +55,8 @@ install:
 	copy $(SHLIB) "$(PKGLIBDIR)"
 	copy $(EXTENSION).control "$(SHAREDIR)\extension"
 	copy sql\$(EXTENSION)--*.sql "$(SHAREDIR)\extension"
+	mkdir "$(INCLUDEDIR_SERVER)\extension\$(EXTENSION)"
+	copy $(HEADERS) "$(INCLUDEDIR_SERVER)\extension\$(EXTENSION)"
 
 installcheck:
 	"$(BINDIR)\pg_regress" --bindir="$(BINDIR)" $(REGRESS_OPTS) $(REGRESS)
@@ -61,7 +64,9 @@ installcheck:
 uninstall:
 	del /f "$(PKGLIBDIR)\$(SHLIB)"
 	del /f "$(SHAREDIR)\extension\$(EXTENSION).control"
-	del /f "$(SHAREDIR)\extension\vector--*.sql"
+	del /f "$(SHAREDIR)\extension\$(EXTENSION)--*.sql"
+	del /f "$(INCLUDEDIR_SERVER)\extension\$(EXTENSION)\*.h"
+	rmdir "$(INCLUDEDIR_SERVER)\extension\$(EXTENSION)"
 
 clean:
 	del /f $(SHLIB) $(EXTENSION).lib $(EXTENSION).exp

README.md

@@ -2,7 +2,7 @@
 
 Open-source vector similarity search for Postgres
 
-Store all of your application data in one place. Supports:
+Store your vectors with the rest of your data. Supports:
 
 - exact and approximate nearest neighbor search
 - L2 distance, inner product, and cosine distance
@@ -18,7 +18,7 @@ Compile and install the extension (supports Postgres 11+)
 
 ```sh
 cd /tmp
-git clone --branch v0.4.4 https://github.com/pgvector/pgvector.git
+git clone --branch v0.5.0 https://github.com/pgvector/pgvector.git
 cd pgvector
 make
 make install # may need sudo
@@ -157,7 +157,16 @@ SELECT category_id, AVG(embedding) FROM items GROUP BY category_id;
 
 By default, pgvector performs exact nearest neighbor search, which provides perfect recall.
 
-You can add an index to use approximate nearest neighbor search, which trades some recall for performance. Unlike typical indexes, you will see different results for queries after adding an approximate index.
+You can add an index to use approximate nearest neighbor search, which trades some recall for speed. Unlike typical indexes, you will see different results for queries after adding an approximate index.
+
+Supported index types are:
+
+- [IVFFlat](#ivfflat)
+- [HNSW](#hnsw) - added in 0.5.0
+
+## IVFFlat
+
+An IVFFlat index divides vectors into lists, and then searches a subset of those lists that are closest to the query vector. It has faster build times and uses less memory than HNSW, but has lower query performance (in terms of speed-recall tradeoff).
 
 Three keys to achieving good recall are:
 
@@ -206,7 +215,63 @@ SELECT ...
 COMMIT;
 ```
 
-### Indexing Progress
+## HNSW
+
+An HNSW index creates a multilayer graph. It has slower build times and uses more memory than IVFFlat, but has better query performance (in terms of speed-recall tradeoff). There’s no training step like IVFFlat, so the index can be created without any data in the table.
+
+Add an index for each distance function you want to use.
+
+L2 distance
+
+```sql
+CREATE INDEX ON items USING hnsw (embedding vector_l2_ops);
+```
+
+Inner product
+
+```sql
+CREATE INDEX ON items USING hnsw (embedding vector_ip_ops);
+```
+
+Cosine distance
+
+```sql
+CREATE INDEX ON items USING hnsw (embedding vector_cosine_ops);
+```
+
+Vectors with up to 2,000 dimensions can be indexed.
+
+### Index Options
+
+Specify HNSW parameters
+
+- `m` - the max number of connections per layer (16 by default)
+- `ef_construction` - the size of the dynamic candidate list for constructing the graph (64 by default)
+
+```sql
+CREATE INDEX ON items USING hnsw (embedding vector_l2_ops) WITH (m = 16, ef_construction = 64);
+```
+
+### Query Options
+
+Specify the size of the dynamic candidate list for search (40 by default)
+
+```sql
+SET hnsw.ef_search = 100;
+```
+
+A higher value provides better recall at the cost of speed.
+
+Use `SET LOCAL` inside a transaction to set it for a single query
+
+```sql
+BEGIN;
+SET LOCAL hnsw.ef_search = 100;
+SELECT ...
+COMMIT;
+```
+
+## Indexing Progress
 
 Check [indexing progress](https://www.postgresql.org/docs/current/progress-reporting.html#CREATE-INDEX-PROGRESS-REPORTING) with Postgres 12+
 
@@ -217,8 +282,8 @@ SELECT phase, tuples_done, tuples_total FROM pg_stat_progress_create_index;
 The phases are:
 
 1. `initializing`
-2. `performing k-means`
-3. `sorting tuples`
+2. `performing k-means` - IVFFlat only
+3. `assigning tuples` - IVFFlat only
 4. `loading tuples`
 
 Note: `tuples_done` and `tuples_total` are only populated during the `loading tuples` phase
@@ -283,7 +348,7 @@ SELECT * FROM items ORDER BY embedding <#> '[3,1,2]' LIMIT 5;
 
 ### Approximate Search
 
-To speed up queries with an 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
 CREATE INDEX ON items USING ivfflat (embedding vector_l2_ops) WITH (lists = 1000);
@@ -359,7 +424,7 @@ or choose to store vectors inline:
 ALTER TABLE items ALTER COLUMN embedding SET STORAGE PLAIN;
 ```
 
-#### Why are there less results for a query after adding an index?
+#### Why are there less results for a query after adding an IVFFlat index?
 
 The index was likely created with too little data for the number of lists. Drop the index until the table has more data.
 
@@ -375,33 +440,32 @@ Each vector takes `4 * dimensions + 8` bytes of storage. Each element is a singl
 
 ### Vector Operators
 
-Operator | Description
---- | ---
-\+ | element-wise addition
-\- | element-wise subtraction
-\* | element-wise multiplication [unreleased]
-<-> | Euclidean distance
-<#> | negative inner product
-<=> | cosine distance
+Operator | Description | Added
+--- | --- | ---
+\+ | element-wise addition |
+\- | element-wise subtraction |
+\* | element-wise multiplication | 0.5.0
+<-> | Euclidean distance |
+<#> | negative inner product |
+<=> | cosine distance |
 
 ### Vector Functions
 
-Function | Description
---- | ---
-cosine_distance(vector, vector) → double precision | cosine distance
-inner_product(vector, vector) → double precision | inner product
-l2_distance(vector, vector) → double precision | Euclidean distance
-l1_distance(vector, vector) → double precision | taxicab distance [unreleased]
-normalize_l2(vector) → vector | normalize with Euclidean norm [unreleased]
-vector_dims(vector) → integer | number of dimensions
-vector_norm(vector) → double precision | Euclidean norm
+Function | Description | Added
+--- | --- | ---
+cosine_distance(vector, vector) → double precision | cosine distance |
+inner_product(vector, vector) → double precision | inner product |
+l2_distance(vector, vector) → double precision | Euclidean distance |
+l1_distance(vector, vector) → double precision | taxicab distance | 0.5.0
+vector_dims(vector) → integer | number of dimensions |
+vector_norm(vector) → double precision | Euclidean norm |
 
 ### Aggregate Functions
 
-Function | Description
---- | ---
-avg(vector) → vector | arithmetic mean
-sum(vector) → vector | sum [unreleased]
+Function | Description | Added
+--- | --- | ---
+avg(vector) → vector | average |
+sum(vector) → vector | sum | 0.5.0
 
 ## Installation Notes
 
@@ -445,7 +509,7 @@ Then use `nmake` to build:
 
 ```cmd
 set "PGROOT=C:\Program Files\PostgreSQL\15"
-git clone --branch v0.4.4 https://github.com/pgvector/pgvector.git
+git clone --branch v0.5.0 https://github.com/pgvector/pgvector.git
 cd pgvector
 nmake /F Makefile.win
 nmake /F Makefile.win install
@@ -466,9 +530,8 @@ This adds pgvector to the [Postgres image](https://hub.docker.com/_/postgres) (r
 You can also build the image manually:
 
 ```sh
-git clone --branch v0.4.4 https://github.com/pgvector/pgvector.git
+git clone --branch v0.5.0 https://github.com/pgvector/pgvector.git
 cd pgvector
-git cherry-pick 237a6df
 docker build --build-arg PG_MAJOR=15 -t myuser/pgvector .
 ```
 

sql/vector--0.4.4--0.5.0.sql

@@ -4,9 +4,6 @@
 CREATE FUNCTION l1_distance(vector, vector) RETURNS float8
 	AS 'MODULE_PATHNAME' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
 
-CREATE FUNCTION normalize_l2(vector) RETURNS vector
-	AS 'MODULE_PATHNAME' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
-
 CREATE FUNCTION vector_mul(vector, vector) RETURNS vector
 	AS 'MODULE_PATHNAME' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
 

sql/vector.sql

@@ -49,9 +49,6 @@ CREATE FUNCTION vector_dims(vector) RETURNS integer
 CREATE FUNCTION vector_norm(vector) RETURNS float8
 	AS 'MODULE_PATHNAME' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
 
-CREATE FUNCTION normalize_l2(vector) RETURNS vector
-	AS 'MODULE_PATHNAME' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
-
 CREATE FUNCTION vector_add(vector, vector) RETURNS vector
 	AS 'MODULE_PATHNAME' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
 

src/hnsw.c

@@ -37,7 +37,7 @@ HnswInit(void)
 		);
 
 	DefineCustomIntVariable("hnsw.ef_search", "Sets the size of the dynamic candidate list for search",
-							"Valid range is 10..1000.", &hnsw_ef_search,
+							"Valid range is 1..1000.", &hnsw_ef_search,
 							HNSW_DEFAULT_EF_SEARCH, HNSW_MIN_EF_SEARCH, HNSW_MAX_EF_SEARCH, PGC_USERSET, 0, NULL, NULL, NULL);
 }
 

src/hnsw.h

@@ -4,6 +4,7 @@
 #include "postgres.h"
 
 #include "access/generic_xlog.h"
+#include "access/parallel.h"
 #include "access/reloptions.h"
 #include "nodes/execnodes.h"
 #include "port.h"				/* for random() */
@@ -14,6 +15,10 @@
 #error "Requires PostgreSQL 11+"
 #endif
 
+#if PG_VERSION_NUM < 120000
+#include "access/relscan.h"
+#endif
+
 #define HNSW_MAX_DIM 2000
 
 /* Support functions */
@@ -22,28 +27,37 @@
 
 #define HNSW_VERSION	1
 #define HNSW_MAGIC_NUMBER 0xA953A953
-#define HNSW_PAGE_ID	0xFF85
+#define HNSW_PAGE_ID	0xFF90
 
 /* Preserved page numbers */
 #define HNSW_METAPAGE_BLKNO	0
 #define HNSW_HEAD_BLKNO		1	/* first element page */
 
+/* Must correspond to page numbers since page lock is used */
+#define HNSW_UPDATE_LOCK 	0
+#define HNSW_SCAN_LOCK		1
+
+/* HNSW parameters */
 #define HNSW_DEFAULT_M	16
-#define HNSW_MIN_M	4
+#define HNSW_MIN_M	2
 #define HNSW_MAX_M		100
-#define HNSW_DEFAULT_EF_CONSTRUCTION	40
-#define HNSW_MIN_EF_CONSTRUCTION	10
+#define HNSW_DEFAULT_EF_CONSTRUCTION	64
+#define HNSW_MIN_EF_CONSTRUCTION	4
 #define HNSW_MAX_EF_CONSTRUCTION		1000
 #define HNSW_DEFAULT_EF_SEARCH	40
-#define HNSW_MIN_EF_SEARCH		10
+#define HNSW_MIN_EF_SEARCH		1
 #define HNSW_MAX_EF_SEARCH		1000
 
+/* Tuple types */
 #define HNSW_ELEMENT_TUPLE_TYPE  1
 #define HNSW_NEIGHBOR_TUPLE_TYPE 2
 
 /* Make graph robust against non-HOT updates */
 #define HNSW_HEAPTIDS 10
 
+#define HNSW_UPDATE_ENTRY_GREATER 1
+#define HNSW_UPDATE_ENTRY_ALWAYS 2
+
 /* Build phases */
 /* PROGRESS_CREATEIDX_SUBPHASE_INITIALIZE is 1 */
 #define PROGRESS_HNSW_PHASE_LOAD		2
@@ -68,10 +82,13 @@
 #define HnswIsElementTuple(tup) ((tup)->type == HNSW_ELEMENT_TUPLE_TYPE)
 #define HnswIsNeighborTuple(tup) ((tup)->type == HNSW_NEIGHBOR_TUPLE_TYPE)
 
-#define HnswGetLayerM(m, layer) (layer == 0 ? m * 2 : m)
+/* 2 * M connections for ground layer */
+#define HnswGetLayerM(m, layer) (layer == 0 ? (m) * 2 : (m))
+
+/* Optimal ML from paper */
 #define HnswGetMl(m) (1 / log(m))
 
-/* Ensure fits in uint8 */
+/* Ensure fits on page and in uint8 */
 #define HnswGetMaxLevel(m) Min(((BLCKSZ - MAXALIGN(SizeOfPageHeaderData) - MAXALIGN(sizeof(HnswPageOpaqueData)) - offsetof(HnswNeighborTupleData, indextids) - sizeof(ItemIdData)) / (sizeof(ItemPointerData)) / m) - 2, 255)
 
 /* Variables */
@@ -106,13 +123,6 @@ typedef struct HnswNeighborArray
 	HnswCandidate *items;
 }			HnswNeighborArray;
 
-typedef struct HnswUpdate
-{
-	HnswCandidate hc;
-	int			level;
-	int			index;
-}			HnswUpdate;
-
 typedef struct HnswPairingHeapNode
 {
 	pairingheap_node ph_node;
@@ -127,6 +137,49 @@ typedef struct HnswOptions
 	int			efConstruction; /* size of dynamic candidate list */
 }			HnswOptions;
 
+typedef struct HnswSpool
+{
+	Relation	heap;
+	Relation	index;
+}			HnswSpool;
+
+typedef struct HnswShared
+{
+	/* Immutable state */
+	Oid			heaprelid;
+	Oid			indexrelid;
+	bool		isconcurrent;
+	int			scantuplesortstates;
+
+	/* Worker progress */
+	ConditionVariable workersdonecv;
+
+	/* Mutex for mutable state */
+	slock_t		mutex;
+
+	/* Mutable state */
+	int			nparticipantsdone;
+	double		reltuples;
+	double		indtuples;
+
+#if PG_VERSION_NUM < 120000
+	ParallelHeapScanDescData heapdesc;	/* must come last */
+#endif
+}			HnswShared;
+
+#if PG_VERSION_NUM >= 120000
+#define ParallelTableScanFromHnswShared(shared) \
+	(ParallelTableScanDesc) ((char *) (shared) + BUFFERALIGN(sizeof(HnswShared)))
+#endif
+
+typedef struct HnswLeader
+{
+	ParallelContext *pcxt;
+	int			nparticipanttuplesorts;
+	HnswShared *hnswshared;
+	Snapshot	snapshot;
+}			HnswLeader;
+
 typedef struct HnswBuildState
 {
 	/* Info */
@@ -160,6 +213,9 @@ typedef struct HnswBuildState
 
 	/* Memory */
 	MemoryContext tmpCtx;
+
+	/* Parallel builds */
+	HnswLeader *hnswleader;
 }			HnswBuildState;
 
 typedef struct HnswMetaPageData
@@ -261,19 +317,26 @@ Buffer		HnswNewBuffer(Relation index, ForkNumber forkNum);
 void		HnswInitPage(Buffer buf, Page page);
 void		HnswInitRegisterPage(Relation index, Buffer *buf, Page *page, GenericXLogState **state);
 void		HnswInit(void);
-List	   *HnswSearchLayer(Datum q, List *ep, int ef, int lc, Relation index, FmgrInfo *procinfo, Oid collation, bool inserting, BlockNumber *skipPage, OffsetNumber *skipOffno);
+List	   *HnswSearchLayer(Datum q, List *ep, int ef, int lc, Relation index, FmgrInfo *procinfo, Oid collation, bool inserting, HnswElement skipElement);
 HnswElement HnswGetEntryPoint(Relation index);
 HnswElement HnswInitElement(ItemPointer tid, int m, double ml, int maxLevel);
 void		HnswFreeElement(HnswElement element);
-HnswElement HnswInsertElement(HnswElement element, HnswElement entryPoint, Relation index, FmgrInfo *procinfo, Oid collation, int m, int efConstruction, List **updates, bool vacuuming);
-HnswCandidate *HnswEntryCandidate(HnswElement em, Datum q, Relation rel, FmgrInfo *procinfo, Oid collation, bool loadvec);
-void		HnswUpdateMetaPage(Relation index, bool updateEntry, HnswElement entryPoint, BlockNumber insertPage, ForkNumber forkNum);
+HnswElement HnswInitElementFromBlock(BlockNumber blkno, OffsetNumber offno);
+void		HnswInsertElement(HnswElement element, HnswElement entryPoint, Relation index, FmgrInfo *procinfo, Oid collation, int m, int efConstruction, bool existing);
+HnswElement HnswFindDuplicate(HnswElement e);
+HnswCandidate *HnswEntryCandidate(HnswElement em, Datum q, Relation rel, FmgrInfo *procinfo, Oid collation, bool loadVec);
+void		HnswUpdateMetaPage(Relation index, int updateEntry, HnswElement entryPoint, BlockNumber insertPage, ForkNumber forkNum);
 void		HnswSetNeighborTuple(HnswNeighborTuple ntup, HnswElement e, int m);
 void		HnswAddHeapTid(HnswElement element, ItemPointer heaptid);
 void		HnswInitNeighbors(HnswElement element, int m);
 bool		HnswInsertTuple(Relation index, Datum *values, bool *isnull, ItemPointer heap_tid, Relation heapRel);
+void		HnswUpdateNeighborPages(Relation index, FmgrInfo *procinfo, Oid collation, HnswElement e, int m, bool checkExisting);
+void		HnswLoadElementFromTuple(HnswElement element, HnswElementTuple etup, bool loadHeaptids, bool loadVec);
 void		HnswLoadElement(HnswElement element, float *distance, Datum *q, Relation index, FmgrInfo *procinfo, Oid collation, bool loadVec);
 void		HnswSetElementTuple(HnswElementTuple etup, HnswElement element);
+void		HnswUpdateConnection(HnswElement element, HnswCandidate * hc, int m, int lc, int *updateIdx, Relation index, FmgrInfo *procinfo, Oid collation);
+void		HnswLoadNeighbors(HnswElement element, Relation index);
+PGDLLEXPORT void HnswParallelBuildMain(dsm_segment *seg, shm_toc *toc);
 
 /* Index access methods */
 IndexBuildResult *hnswbuild(Relation heap, Relation index, IndexInfo *indexInfo);

src/hnswbuild.c

@@ -2,12 +2,15 @@
 
 #include <math.h>
 
+#include "access/parallel.h"
+#include "access/xact.h"
 #include "catalog/index.h"
 #include "hnsw.h"
 #include "miscadmin.h"
 #include "lib/pairingheap.h"
 #include "nodes/pg_list.h"
 #include "storage/bufmgr.h"
+#include "tcop/tcopprot.h"
 #include "utils/memutils.h"
 
 #if PG_VERSION_NUM >= 140000
@@ -35,6 +38,23 @@
 #define UpdateProgress(index, val) ((void)val)
 #endif
 
+#if PG_VERSION_NUM >= 140000
+#include "utils/backend_status.h"
+#include "utils/wait_event.h"
+#endif
+
+#if PG_VERSION_NUM >= 120000
+#include "access/table.h"
+#include "optimizer/optimizer.h"
+#else
+#include "access/heapam.h"
+#include "optimizer/planner.h"
+#include "pgstat.h"
+#endif
+
+#define PARALLEL_KEY_HNSW_SHARED		UINT64CONST(0xA000000000000001)
+#define PARALLEL_KEY_QUERY_TEXT			UINT64CONST(0xA000000000000002)
+
 /*
  * Create the metapage
  */
@@ -60,6 +80,7 @@ CreateMetaPage(HnswBuildState * buildstate)
 	metap->efConstruction = buildstate->efConstruction;
 	metap->entryBlkno = InvalidBlockNumber;
 	metap->entryOffno = InvalidOffsetNumber;
+	metap->entryLevel = -1;
 	metap->insertPage = InvalidBlockNumber;
 	((PageHeader) page)->pd_lower =
 		((char *) metap + sizeof(HnswMetaPageData)) - (char *) page;
@@ -182,7 +203,7 @@ CreateElementPages(HnswBuildState * buildstate)
 	GenericXLogFinish(state);
 	UnlockReleaseBuffer(buf);
 
-	HnswUpdateMetaPage(index, true, buildstate->entryPoint, insertPage, forkNum);
+	HnswUpdateMetaPage(index, HNSW_UPDATE_ENTRY_ALWAYS, buildstate->entryPoint, insertPage, forkNum);
 
 	pfree(etup);
 	pfree(ntup);
@@ -288,7 +309,23 @@ InsertTuple(Relation index, Datum *values, HnswElement element, HnswBuildState *
 	memcpy(element->vec, DatumGetVector(value), VECTOR_SIZE(buildstate->dimensions));
 
 	/* Insert element in graph */
-	*dup = HnswInsertElement(element, entryPoint, NULL, procinfo, collation, m, efConstruction, NULL, false);
+	HnswInsertElement(element, entryPoint, NULL, procinfo, collation, m, efConstruction, false);
+
+	/* Look for duplicate */
+	*dup = HnswFindDuplicate(element);
+
+	/* Update neighbors if needed */
+	if (*dup == NULL)
+	{
+		for (int lc = element->level; lc >= 0; lc--)
+		{
+			int			lm = HnswGetLayerM(m, lc);
+			HnswNeighborArray *neighbors = &element->neighbors[lc];
+
+			for (int i = 0; i < neighbors->length; i++)
+				HnswUpdateConnection(element, &neighbors->items[i], lm, lc, NULL, NULL, procinfo, collation);
+		}
+	}
 
 	/* Update entry point if needed */
 	if (*dup == NULL && (entryPoint == NULL || element->level > entryPoint->level))
@@ -334,6 +371,7 @@ BuildCallback(Relation index, CALLBACK_ITEM_POINTER, Datum *values,
 
 		oldCtx = MemoryContextSwitchTo(buildstate->tmpCtx);
 
+		/* TODO Fix progress for parallel builds */
 		if (HnswInsertTuple(buildstate->index, values, isnull, tid, buildstate->heap))
 			UpdateProgress(PROGRESS_CREATEIDX_TUPLES_DONE, ++buildstate->indtuples);
 
@@ -407,6 +445,9 @@ InitBuildState(HnswBuildState * buildstate, Relation heap, Relation index, Index
 	if (buildstate->dimensions > HNSW_MAX_DIM)
 		elog(ERROR, "column cannot have more than %d dimensions for hnsw index", HNSW_MAX_DIM);
 
+	if (buildstate->efConstruction < 2 * buildstate->m)
+		elog(ERROR, "ef_construction must be greater than or equal to 2 * m");
+
 	buildstate->reltuples = 0;
 	buildstate->indtuples = 0;
 
@@ -428,6 +469,8 @@ InitBuildState(HnswBuildState * buildstate, Relation heap, Relation index, Index
 	buildstate->tmpCtx = AllocSetContextCreate(CurrentMemoryContext,
 											   "Hnsw build temporary context",
 											   ALLOCSET_DEFAULT_SIZES);
+
+	buildstate->hnswleader = NULL;
 }
 
 /*
@@ -440,21 +483,373 @@ FreeBuildState(HnswBuildState * buildstate)
 	MemoryContextDelete(buildstate->tmpCtx);
 }
 
+/*
+ * Within leader, wait for end of heap scan
+ */
+static double
+ParallelHeapScan(HnswBuildState * buildstate)
+{
+	HnswShared *hnswshared = buildstate->hnswleader->hnswshared;
+	int			nparticipanttuplesorts;
+	double		reltuples;
+
+	nparticipanttuplesorts = buildstate->hnswleader->nparticipanttuplesorts;
+	for (;;)
+	{
+		SpinLockAcquire(&hnswshared->mutex);
+		if (hnswshared->nparticipantsdone == nparticipanttuplesorts)
+		{
+			buildstate->indtuples = hnswshared->indtuples;
+			reltuples = hnswshared->reltuples;
+			SpinLockRelease(&hnswshared->mutex);
+			break;
+		}
+		SpinLockRelease(&hnswshared->mutex);
+
+		ConditionVariableSleep(&hnswshared->workersdonecv,
+							   WAIT_EVENT_PARALLEL_CREATE_INDEX_SCAN);
+	}
+
+	ConditionVariableCancelSleep();
+
+	return reltuples;
+}
+
+/*
+ * Perform a worker's portion of a parallel insert
+ */
+static void
+HnswParallelScanAndInsert(HnswSpool * hnswspool, HnswShared * hnswshared, bool progress)
+{
+	HnswBuildState buildstate;
+#if PG_VERSION_NUM >= 120000
+	TableScanDesc scan;
+#else
+	HeapScanDesc scan;
+#endif
+	double		reltuples;
+	IndexInfo  *indexInfo;
+
+	/* Join parallel scan */
+	indexInfo = BuildIndexInfo(hnswspool->index);
+	indexInfo->ii_Concurrent = hnswshared->isconcurrent;
+	InitBuildState(&buildstate, hnswspool->heap, hnswspool->index, indexInfo, MAIN_FORKNUM);
+	/* TODO Support in-memory builds */
+	buildstate.maxInMemoryElements = 0;
+	buildstate.flushed = true;
+#if PG_VERSION_NUM >= 120000
+	scan = table_beginscan_parallel(hnswspool->heap,
+									ParallelTableScanFromHnswShared(hnswshared));
+	reltuples = table_index_build_scan(hnswspool->heap, hnswspool->index, indexInfo,
+									   true, progress, BuildCallback,
+									   (void *) &buildstate, scan);
+#else
+	scan = heap_beginscan_parallel(hnswspool->heap, &hnswshared->heapdesc);
+	reltuples = IndexBuildHeapScan(hnswspool->heap, hnswspool->index, indexInfo,
+								   true, BuildCallback,
+								   (void *) &buildstate, scan);
+#endif
+
+	/* Record statistics */
+	SpinLockAcquire(&hnswshared->mutex);
+	hnswshared->nparticipantsdone++;
+	hnswshared->reltuples += reltuples;
+	hnswshared->indtuples += buildstate.indtuples;
+	SpinLockRelease(&hnswshared->mutex);
+
+	/* Log statistics */
+	if (progress)
+		ereport(DEBUG1, (errmsg("leader processed " INT64_FORMAT " tuples", (int64) reltuples)));
+	else
+		ereport(DEBUG1, (errmsg("worker processed " INT64_FORMAT " tuples", (int64) reltuples)));
+
+	/* Notify leader */
+	ConditionVariableSignal(&hnswshared->workersdonecv);
+
+	FreeBuildState(&buildstate);
+}
+
+/*
+ * Perform work within a launched parallel process
+ */
+void
+HnswParallelBuildMain(dsm_segment *seg, shm_toc *toc)
+{
+	char	   *sharedquery;
+	HnswSpool  *hnswspool;
+	HnswShared *hnswshared;
+	Relation	heapRel;
+	Relation	indexRel;
+	LOCKMODE	heapLockmode;
+	LOCKMODE	indexLockmode;
+
+	/* Set debug_query_string for individual workers first */
+	sharedquery = shm_toc_lookup(toc, PARALLEL_KEY_QUERY_TEXT, true);
+	debug_query_string = sharedquery;
+
+	/* Report the query string from leader */
+	pgstat_report_activity(STATE_RUNNING, debug_query_string);
+
+	/* Look up shared state */
+	hnswshared = shm_toc_lookup(toc, PARALLEL_KEY_HNSW_SHARED, false);
+
+	/* Open relations using lock modes known to be obtained by index.c */
+	if (!hnswshared->isconcurrent)
+	{
+		heapLockmode = ShareLock;
+		indexLockmode = AccessExclusiveLock;
+	}
+	else
+	{
+		heapLockmode = ShareUpdateExclusiveLock;
+		indexLockmode = RowExclusiveLock;
+	}
+
+	/* Open relations within worker */
+#if PG_VERSION_NUM >= 120000
+	heapRel = table_open(hnswshared->heaprelid, heapLockmode);
+#else
+	heapRel = heap_open(hnswshared->heaprelid, heapLockmode);
+#endif
+	indexRel = index_open(hnswshared->indexrelid, indexLockmode);
+
+	/* Initialize worker's own spool */
+	hnswspool = (HnswSpool *) palloc0(sizeof(HnswSpool));
+	hnswspool->heap = heapRel;
+	hnswspool->index = indexRel;
+
+	/* Perform inserts */
+	HnswParallelScanAndInsert(hnswspool, hnswshared, false);
+
+	/* Close relations within worker */
+	index_close(indexRel, indexLockmode);
+#if PG_VERSION_NUM >= 120000
+	table_close(heapRel, heapLockmode);
+#else
+	heap_close(heapRel, heapLockmode);
+#endif
+}
+
+/*
+ * End parallel build
+ */
+static void
+HnswEndParallel(HnswLeader * hnswleader)
+{
+	/* Shutdown worker processes */
+	WaitForParallelWorkersToFinish(hnswleader->pcxt);
+
+	/* Free last reference to MVCC snapshot, if one was used */
+	if (IsMVCCSnapshot(hnswleader->snapshot))
+		UnregisterSnapshot(hnswleader->snapshot);
+	DestroyParallelContext(hnswleader->pcxt);
+	ExitParallelMode();
+}
+
+/*
+ * Return size of shared memory required for parallel index build
+ */
+static Size
+ParallelEstimateShared(Relation heap, Snapshot snapshot)
+{
+#if PG_VERSION_NUM >= 120000
+	return add_size(BUFFERALIGN(sizeof(HnswShared)), table_parallelscan_estimate(heap, snapshot));
+#else
+	if (!IsMVCCSnapshot(snapshot))
+	{
+		Assert(snapshot == SnapshotAny);
+		return sizeof(HnswShared);
+	}
+
+	return add_size(offsetof(HnswShared, heapdesc) +
+					offsetof(ParallelHeapScanDescData, phs_snapshot_data),
+					EstimateSnapshotSpace(snapshot));
+#endif
+}
+
+/*
+ * Within leader, participate as a parallel worker
+ */
+static void
+HnswLeaderParticipateAsWorker(HnswBuildState * buildstate)
+{
+	HnswLeader *hnswleader = buildstate->hnswleader;
+	HnswSpool  *leaderworker;
+
+	/* Allocate memory and initialize private spool */
+	leaderworker = (HnswSpool *) palloc0(sizeof(HnswSpool));
+	leaderworker->heap = buildstate->heap;
+	leaderworker->index = buildstate->index;
+
+	/* Perform work common to all participants */
+	HnswParallelScanAndInsert(leaderworker, hnswleader->hnswshared, true);
+}
+
+/*
+ * Begin parallel build
+ */
+static void
+HnswBeginParallel(HnswBuildState * buildstate, bool isconcurrent, int request)
+{
+	ParallelContext *pcxt;
+	int			scantuplesortstates;
+	Snapshot	snapshot;
+	Size		esthnswshared;
+	HnswShared *hnswshared;
+	HnswLeader *hnswleader = (HnswLeader *) palloc0(sizeof(HnswLeader));
+	bool		leaderparticipates = true;
+	int			querylen;
+
+#ifdef DISABLE_LEADER_PARTICIPATION
+	leaderparticipates = false;
+#endif
+
+	/* Enter parallel mode and create context */
+	EnterParallelMode();
+	Assert(request > 0);
+#if PG_VERSION_NUM >= 120000
+	pcxt = CreateParallelContext("vector", "HnswParallelBuildMain", request);
+#else
+	pcxt = CreateParallelContext("vector", "HnswParallelBuildMain", request, true);
+#endif
+
+	scantuplesortstates = leaderparticipates ? request + 1 : request;
+
+	/* Get snapshot for table scan */
+	if (!isconcurrent)
+		snapshot = SnapshotAny;
+	else
+		snapshot = RegisterSnapshot(GetTransactionSnapshot());
+
+	/* Estimate size of workspaces */
+	esthnswshared = ParallelEstimateShared(buildstate->heap, snapshot);
+	shm_toc_estimate_chunk(&pcxt->estimator, esthnswshared);
+	shm_toc_estimate_keys(&pcxt->estimator, 1);
+
+	/* Finally, estimate PARALLEL_KEY_QUERY_TEXT space */
+	if (debug_query_string)
+	{
+		querylen = strlen(debug_query_string);
+		shm_toc_estimate_chunk(&pcxt->estimator, querylen + 1);
+		shm_toc_estimate_keys(&pcxt->estimator, 1);
+	}
+	else
+		querylen = 0;			/* keep compiler quiet */
+
+	/* Everyone's had a chance to ask for space, so now create the DSM */
+	InitializeParallelDSM(pcxt);
+
+	/* If no DSM segment was available, back out (do serial build) */
+	if (pcxt->seg == NULL)
+	{
+		if (IsMVCCSnapshot(snapshot))
+			UnregisterSnapshot(snapshot);
+		DestroyParallelContext(pcxt);
+		ExitParallelMode();
+		return;
+	}
+
+	/* Store shared build state, for which we reserved space */
+	hnswshared = (HnswShared *) shm_toc_allocate(pcxt->toc, esthnswshared);
+	/* Initialize immutable state */
+	hnswshared->heaprelid = RelationGetRelid(buildstate->heap);
+	hnswshared->indexrelid = RelationGetRelid(buildstate->index);
+	hnswshared->isconcurrent = isconcurrent;
+	hnswshared->scantuplesortstates = scantuplesortstates;
+	ConditionVariableInit(&hnswshared->workersdonecv);
+	SpinLockInit(&hnswshared->mutex);
+	/* Initialize mutable state */
+	hnswshared->nparticipantsdone = 0;
+	hnswshared->reltuples = 0;
+	hnswshared->indtuples = 0;
+#if PG_VERSION_NUM >= 120000
+	table_parallelscan_initialize(buildstate->heap,
+								  ParallelTableScanFromHnswShared(hnswshared),
+								  snapshot);
+#else
+	heap_parallelscan_initialize(&hnswshared->heapdesc, buildstate->heap, snapshot);
+#endif
+
+	shm_toc_insert(pcxt->toc, PARALLEL_KEY_HNSW_SHARED, hnswshared);
+
+	/* Store query string for workers */
+	if (debug_query_string)
+	{
+		char	   *sharedquery;
+
+		sharedquery = (char *) shm_toc_allocate(pcxt->toc, querylen + 1);
+		memcpy(sharedquery, debug_query_string, querylen + 1);
+		shm_toc_insert(pcxt->toc, PARALLEL_KEY_QUERY_TEXT, sharedquery);
+	}
+
+	/* Launch workers, saving status for leader/caller */
+	LaunchParallelWorkers(pcxt);
+	hnswleader->pcxt = pcxt;
+	hnswleader->nparticipanttuplesorts = pcxt->nworkers_launched;
+	if (leaderparticipates)
+		hnswleader->nparticipanttuplesorts++;
+	hnswleader->hnswshared = hnswshared;
+	hnswleader->snapshot = snapshot;
+
+	/* If no workers were successfully launched, back out (do serial build) */
+	if (pcxt->nworkers_launched == 0)
+	{
+		HnswEndParallel(hnswleader);
+		return;
+	}
+
+	/* Log participants */
+	ereport(DEBUG1, (errmsg("using %d parallel workers", pcxt->nworkers_launched)));
+
+	/* Save leader state now that it's clear build will be parallel */
+	buildstate->hnswleader = hnswleader;
+
+	/* Join heap scan ourselves */
+	if (leaderparticipates)
+		HnswLeaderParticipateAsWorker(buildstate);
+
+	/* Wait for all launched workers */
+	WaitForParallelWorkersToAttach(pcxt);
+}
+
 /*
  * Build graph
  */
 static void
 BuildGraph(HnswBuildState * buildstate, ForkNumber forkNum)
 {
+	int			parallel_workers = 0;
+
 	UpdateProgress(PROGRESS_CREATEIDX_SUBPHASE, PROGRESS_HNSW_PHASE_LOAD);
 
+	/* Calculate parallel workers */
+	parallel_workers = plan_create_index_workers(RelationGetRelid(buildstate->heap), RelationGetRelid(buildstate->index));
+
+	/* Attempt to launch parallel worker scan when required */
+	if (parallel_workers > 0)
+	{
+		/* TODO Support in-memory builds */
+		FlushPages(buildstate);
+		HnswBeginParallel(buildstate, buildstate->indexInfo->ii_Concurrent, parallel_workers);
+	}
+
+	/* Add tuples to sort */
+	if (buildstate->hnswleader)
+		buildstate->reltuples = ParallelHeapScan(buildstate);
+	else
+	{
 #if PG_VERSION_NUM >= 120000
-	buildstate->reltuples = table_index_build_scan(buildstate->heap, buildstate->index, buildstate->indexInfo,
-												   true, true, BuildCallback, (void *) buildstate, NULL);
+		buildstate->reltuples = table_index_build_scan(buildstate->heap, buildstate->index, buildstate->indexInfo,
+													   true, true, BuildCallback, (void *) buildstate, NULL);
 #else
-	buildstate->reltuples = IndexBuildHeapScan(buildstate->heap, buildstate->index, buildstate->indexInfo,
-											   true, BuildCallback, (void *) buildstate, NULL);
+		buildstate->reltuples = IndexBuildHeapScan(buildstate->heap, buildstate->index, buildstate->indexInfo,
+												   true, BuildCallback, (void *) buildstate, NULL);
 #endif
+	}
+
+	/* End parallel build */
+	if (buildstate->hnswleader)
+		HnswEndParallel(buildstate->hnswleader);
 }
 
 /*

src/hnswinsert.c

@@ -34,7 +34,7 @@ GetInsertPage(Relation index)
  * Check for a free offset
  */
 static bool
-HnswFreeOffset(Relation index, Buffer buf, Page page, HnswElement element, Size ntupSize, Buffer *nbuf, Page *npage, OffsetNumber *freeOffno, OffsetNumber *freeNeighborOffno, BlockNumber *firstFreePage)
+HnswFreeOffset(Relation index, Buffer buf, Page page, HnswElement element, Size ntupSize, Buffer *nbuf, Page *npage, OffsetNumber *freeOffno, OffsetNumber *freeNeighborOffno, BlockNumber *newInsertPage)
 {
 	OffsetNumber offno;
 	OffsetNumber maxoffno = PageGetMaxOffsetNumber(page);
@@ -49,14 +49,15 @@ HnswFreeOffset(Relation index, Buffer buf, Page page, HnswElement element, Size
 
 		if (etup->deleted)
 		{
+			BlockNumber elementPage = BufferGetBlockNumber(buf);
 			BlockNumber neighborPage = ItemPointerGetBlockNumber(&etup->neighbortid);
 			OffsetNumber neighborOffno = ItemPointerGetOffsetNumber(&etup->neighbortid);
 			ItemId		itemid;
 
-			if (!BlockNumberIsValid(*firstFreePage))
-				*firstFreePage = neighborPage;
+			if (!BlockNumberIsValid(*newInsertPage))
+				*newInsertPage = elementPage;
 
-			if (neighborPage == BufferGetBlockNumber(buf))
+			if (neighborPage == elementPage)
 			{
 				*nbuf = buf;
 				*npage = page;
@@ -110,7 +111,7 @@ HnswInsertAppendPage(Relation index, Buffer *nbuf, Page *npage, GenericXLogState
  * Add to element and neighbor pages
  */
 static void
-WriteNewElementPages(Relation index, HnswElement e, int m)
+WriteNewElementPages(Relation index, HnswElement e, int m, BlockNumber insertPage, BlockNumber *updatedInsertPage)
 {
 	Buffer		buf;
 	Page		page;
@@ -118,21 +119,24 @@ WriteNewElementPages(Relation index, HnswElement e, int m)
 	Size		etupSize;
 	Size		ntupSize;
 	Size		combinedSize;
+	Size		maxSize;
+	Size		minCombinedSize;
 	HnswElementTuple etup;
-	BlockNumber insertPage = GetInsertPage(index);
-	BlockNumber originalInsertPage = insertPage;
+	BlockNumber currentPage = insertPage;
 	int			dimensions = e->vec->dim;
 	HnswNeighborTuple ntup;
 	Buffer		nbuf;
 	Page		npage;
 	OffsetNumber freeOffno = InvalidOffsetNumber;
 	OffsetNumber freeNeighborOffno = InvalidOffsetNumber;
-	BlockNumber firstFreePage = InvalidBlockNumber;
+	BlockNumber newInsertPage = InvalidBlockNumber;
 
 	/* Calculate sizes */
 	etupSize = HNSW_ELEMENT_TUPLE_SIZE(dimensions);
 	ntupSize = HNSW_NEIGHBOR_TUPLE_SIZE(e->level, m);
 	combinedSize = etupSize + ntupSize + sizeof(ItemIdData);
+	maxSize = BLCKSZ - MAXALIGN(SizeOfPageHeaderData) - MAXALIGN(sizeof(HnswPageOpaqueData));
+	minCombinedSize = etupSize + HNSW_NEIGHBOR_TUPLE_SIZE(0, m) + sizeof(ItemIdData);
 
 	/* Prepare element tuple */
 	etup = palloc0(etupSize);
@@ -142,16 +146,22 @@ WriteNewElementPages(Relation index, HnswElement e, int m)
 	ntup = palloc0(ntupSize);
 	HnswSetNeighborTuple(ntup, e, m);
 
-	/* Find a page to insert the item */
+	/* Find a page (or two if needed) to insert the tuples */
 	for (;;)
 	{
-		buf = ReadBuffer(index, insertPage);
+		buf = ReadBuffer(index, currentPage);
 		LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
 
 		state = GenericXLogStart(index);
 		page = GenericXLogRegisterBuffer(state, buf, 0);
 
-		/* Space for both */
+		/* Keep track of first page where element at level 0 can fit */
+		if (!BlockNumberIsValid(newInsertPage) && PageGetFreeSpace(page) >= minCombinedSize)
+			newInsertPage = currentPage;
+
+		/* First, try the fastest path */
+		/* Space for both tuples on the current page */
+		/* This can split existing tuples in rare cases */
 		if (PageGetFreeSpace(page) >= combinedSize)
 		{
 			nbuf = buf;
@@ -159,15 +169,8 @@ WriteNewElementPages(Relation index, HnswElement e, int m)
 			break;
 		}
 
-		/* Space for element but not neighbors and last page */
-		if (PageGetFreeSpace(page) >= etupSize && !BlockNumberIsValid(HnswPageGetOpaque(page)->nextblkno))
-		{
-			HnswInsertAppendPage(index, &nbuf, &npage, state, page);
-			break;
-		}
-
-		/* Space from deleted item */
-		if (HnswFreeOffset(index, buf, page, e, ntupSize, &nbuf, &npage, &freeOffno, &freeNeighborOffno, &firstFreePage))
+		/* Next, try space from a deleted element */
+		if (HnswFreeOffset(index, buf, page, e, ntupSize, &nbuf, &npage, &freeOffno, &freeNeighborOffno, &newInsertPage))
 		{
 			if (nbuf != buf)
 				npage = GenericXLogRegisterBuffer(state, nbuf, 0);
@@ -175,9 +178,17 @@ WriteNewElementPages(Relation index, HnswElement e, int m)
 			break;
 		}
 
-		insertPage = HnswPageGetOpaque(page)->nextblkno;
+		/* Finally, try space for element only if last page */
+		/* Skip if both tuples can fit on the same page */
+		if (combinedSize > maxSize && PageGetFreeSpace(page) >= etupSize && !BlockNumberIsValid(HnswPageGetOpaque(page)->nextblkno))
+		{
+			HnswInsertAppendPage(index, &nbuf, &npage, state, page);
+			break;
+		}
 
-		if (BlockNumberIsValid(insertPage))
+		currentPage = HnswPageGetOpaque(page)->nextblkno;
+
+		if (BlockNumberIsValid(currentPage))
 		{
 			/* Move to next page */
 			GenericXLogAbort(state);
@@ -219,7 +230,10 @@ WriteNewElementPages(Relation index, HnswElement e, int m)
 	e->blkno = BufferGetBlockNumber(buf);
 	e->neighborPage = BufferGetBlockNumber(nbuf);
 
-	insertPage = e->neighborPage;
+	/* Added tuple to new page if newInsertPage is not set */
+	/* So can set to neighbor page instead of element page */
+	if (!BlockNumberIsValid(newInsertPage))
+		newInsertPage = e->neighborPage;
 
 	if (OffsetNumberIsValid(freeOffno))
 	{
@@ -265,75 +279,126 @@ WriteNewElementPages(Relation index, HnswElement e, int m)
 		UnlockReleaseBuffer(nbuf);
 
 	/* Update the insert page */
-	if (insertPage != originalInsertPage && (!OffsetNumberIsValid(freeOffno) || firstFreePage == insertPage))
-		HnswUpdateMetaPage(index, false, NULL, insertPage, MAIN_FORKNUM);
+	if (BlockNumberIsValid(newInsertPage) && newInsertPage != insertPage)
+		*updatedInsertPage = newInsertPage;
 }
 
 /*
- * Calculate index for update
+ * Check if connection already exists
  */
-static int
-HnswGetIndex(HnswUpdate * update, int m)
+static bool
+ConnectionExists(HnswElement e, HnswNeighborTuple ntup, int startIdx, int lm)
 {
-	return (update->hc.element->level - update->level) * m + update->index;
+	for (int i = 0; i < lm; i++)
+	{
+		ItemPointer indextid = &ntup->indextids[startIdx + i];
+
+		if (!ItemPointerIsValid(indextid))
+			break;
+
+		if (ItemPointerGetBlockNumber(indextid) == e->blkno && ItemPointerGetOffsetNumber(indextid) == e->offno)
+			return true;
+	}
+
+	return false;
 }
 
 /*
  * Update neighbors
  */
-static void
-UpdateNeighborPages(Relation index, HnswElement e, int m, List *updates)
+void
+HnswUpdateNeighborPages(Relation index, FmgrInfo *procinfo, Oid collation, HnswElement e, int m, bool checkExisting)
 {
-	ListCell   *lc;
-
-	/* Could update multiple at once for same element */
-	/* but should only happen a low percent of time, so keep simple for now */
-	foreach(lc, updates)
+	for (int lc = e->level; lc >= 0; lc--)
 	{
-		Buffer		buf;
-		Page		page;
-		GenericXLogState *state;
-		HnswUpdate *update = lfirst(lc);
-		ItemId		itemid;
-		HnswNeighborTuple ntup;
-		Size		ntupSize;
-		int			idx;
-		OffsetNumber offno = update->hc.element->neighborOffno;
+		int			lm = HnswGetLayerM(m, lc);
+		HnswNeighborArray *neighbors = &e->neighbors[lc];
 
-		/* Register page */
-		buf = ReadBuffer(index, update->hc.element->neighborPage);
-		LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
-		state = GenericXLogStart(index);
-		page = GenericXLogRegisterBuffer(state, buf, 0);
-
-		/* Get tuple */
-		itemid = PageGetItemId(page, offno);
-		ntup = (HnswNeighborTuple) PageGetItem(page, itemid);
-		ntupSize = ItemIdGetLength(itemid);
-
-		/* Calculate index */
-		idx = HnswGetIndex(update, m);
-
-		/* Make robust to issues */
-		if (idx < ntup->count)
+		for (int i = 0; i < neighbors->length; i++)
 		{
-			ItemPointer indextid = &ntup->indextids[idx];
+			HnswCandidate *hc = &neighbors->items[i];
+			Buffer		buf;
+			Page		page;
+			GenericXLogState *state;
+			ItemId		itemid;
+			HnswNeighborTuple ntup;
+			Size		ntupSize;
+			int			idx = -1;
+			int			startIdx;
+			OffsetNumber offno = hc->element->neighborOffno;
 
-			/* Update neighbor */
-			ItemPointerSet(indextid, e->blkno, e->offno);
+			/* Get latest neighbors since they may have changed */
+			/* Do not lock yet since selecting neighbors can take time */
+			HnswLoadNeighbors(hc->element, index);
 
-			/* Overwrite tuple */
-			if (!PageIndexTupleOverwrite(page, offno, (Item) ntup, ntupSize))
-				elog(ERROR, "failed to add index item to \"%s\"", RelationGetRelationName(index));
+			/*
+			 * Could improve performance for vacuuming by checking neighbors
+			 * against list of elements being deleted to find index. It's
+			 * important to exclude already deleted elements for this since
+			 * they can be replaced at any time.
+			 */
 
-			/* Commit */
-			MarkBufferDirty(buf);
-			GenericXLogFinish(state);
+			/* Select neighbors */
+			HnswUpdateConnection(e, hc, lm, lc, &idx, index, procinfo, collation);
+
+			/* New element was not selected as a neighbor */
+			if (idx == -1)
+				continue;
+
+			/* Register page */
+			buf = ReadBuffer(index, hc->element->neighborPage);
+			LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
+			state = GenericXLogStart(index);
+			page = GenericXLogRegisterBuffer(state, buf, 0);
+
+			/* Get tuple */
+			itemid = PageGetItemId(page, offno);
+			ntup = (HnswNeighborTuple) PageGetItem(page, itemid);
+			ntupSize = ItemIdGetLength(itemid);
+
+			/* Calculate index for update */
+			startIdx = (hc->element->level - lc) * m;
+
+			/* Check for existing connection */
+			if (checkExisting && ConnectionExists(e, ntup, startIdx, lm))
+				idx = -1;
+			else if (idx == -2)
+			{
+				/* Find free offset if still exists */
+				/* TODO Retry updating connections if not */
+				for (int j = 0; j < lm; j++)
+				{
+					if (!ItemPointerIsValid(&ntup->indextids[startIdx + j]))
+					{
+						idx = startIdx + j;
+						break;
+					}
+				}
+			}
+			else
+				idx += startIdx;
+
+			/* Make robust to issues */
+			if (idx >= 0 && idx < ntup->count)
+			{
+				ItemPointer indextid = &ntup->indextids[idx];
+
+				/* Update neighbor */
+				ItemPointerSet(indextid, e->blkno, e->offno);
+
+				/* Overwrite tuple */
+				if (!PageIndexTupleOverwrite(page, offno, (Item) ntup, ntupSize))
+					elog(ERROR, "failed to add index item to \"%s\"", RelationGetRelationName(index));
+
+				/* Commit */
+				MarkBufferDirty(buf);
+				GenericXLogFinish(state);
+			}
+			else
+				GenericXLogAbort(state);
+
+			UnlockReleaseBuffer(buf);
 		}
-		else
-			GenericXLogAbort(state);
-
-		UnlockReleaseBuffer(buf);
 	}
 }
 
@@ -391,8 +456,10 @@ HnswAddDuplicate(Relation index, HnswElement element, HnswElement dup)
  * Write changes to disk
  */
 static void
-WriteElement(Relation index, HnswElement element, int m, List *updates, HnswElement dup, HnswElement entryPoint)
+WriteElement(Relation index, FmgrInfo *procinfo, Oid collation, HnswElement element, int m, int efConstruction, HnswElement dup, HnswElement entryPoint)
 {
+	BlockNumber newInsertPage = InvalidBlockNumber;
+
 	/* Try to add to existing page */
 	if (dup != NULL)
 	{
@@ -400,13 +467,19 @@ WriteElement(Relation index, HnswElement element, int m, List *updates, HnswElem
 			return;
 	}
 
-	/* If fails, take this path */
-	WriteNewElementPages(index, element, m);
-	UpdateNeighborPages(index, element, m, updates);
+	/* Write element and neighbor tuples */
+	WriteNewElementPages(index, element, m, GetInsertPage(index), &newInsertPage);
+
+	/* Update insert page if needed */
+	if (BlockNumberIsValid(newInsertPage))
+		HnswUpdateMetaPage(index, 0, NULL, newInsertPage, MAIN_FORKNUM);
+
+	/* Update neighbors */
+	HnswUpdateNeighborPages(index, procinfo, collation, element, m, false);
 
 	/* Update metapage if needed */
 	if (entryPoint == NULL || element->level > entryPoint->level)
-		HnswUpdateMetaPage(index, true, element, InvalidBlockNumber, MAIN_FORKNUM);
+		HnswUpdateMetaPage(index, HNSW_UPDATE_ENTRY_GREATER, element, InvalidBlockNumber, MAIN_FORKNUM);
 }
 
 /*
@@ -424,8 +497,8 @@ HnswInsertTuple(Relation index, Datum *values, bool *isnull, ItemPointer heap_ti
 	double		ml = HnswGetMl(m);
 	FmgrInfo   *procinfo = index_getprocinfo(index, 1, HNSW_DISTANCE_PROC);
 	Oid			collation = index->rd_indcollation[0];
-	List	   *updates = NIL;
 	HnswElement dup;
+	LOCKMODE	lockmode = ShareLock;
 
 	/* Detoast once for all calls */
 	value = PointerGetDatum(PG_DETOAST_DATUM(values[0]));
@@ -442,14 +515,41 @@ HnswInsertTuple(Relation index, Datum *values, bool *isnull, ItemPointer heap_ti
 	element = HnswInitElement(heap_tid, m, ml, HnswGetMaxLevel(m));
 	element->vec = DatumGetVector(value);
 
+	/*
+	 * Get a shared lock. This allows vacuum to ensure no in-flight inserts
+	 * before repairing graph. Use a page lock so it does not interfere with
+	 * buffer lock (or reads when vacuuming).
+	 */
+	LockPage(index, HNSW_UPDATE_LOCK, lockmode);
+
 	/* Get entry point */
 	entryPoint = HnswGetEntryPoint(index);
 
+	/* Prevent concurrent inserts when likely updating entry point */
+	if (entryPoint == NULL || element->level > entryPoint->level)
+	{
+		/* Release shared lock */
+		UnlockPage(index, HNSW_UPDATE_LOCK, lockmode);
+
+		/* Get exclusive lock */
+		lockmode = ExclusiveLock;
+		LockPage(index, HNSW_UPDATE_LOCK, lockmode);
+
+		/* Get latest entry point after lock is acquired */
+		entryPoint = HnswGetEntryPoint(index);
+	}
+
 	/* Insert element in graph */
-	dup = HnswInsertElement(element, entryPoint, index, procinfo, collation, m, efConstruction, &updates, false);
+	HnswInsertElement(element, entryPoint, index, procinfo, collation, m, efConstruction, false);
+
+	/* Look for duplicate */
+	dup = HnswFindDuplicate(element);
 
 	/* Write to disk */
-	WriteElement(index, element, m, updates, dup, entryPoint);
+	WriteElement(index, procinfo, collation, element, m, efConstruction, dup, entryPoint);
+
+	/* Release lock */
+	UnlockPage(index, HNSW_UPDATE_LOCK, lockmode);
 
 	return true;
 }

src/hnswscan.c

@@ -4,34 +4,35 @@
 #include "hnsw.h"
 #include "pgstat.h"
 #include "storage/bufmgr.h"
+#include "storage/lmgr.h"
 #include "utils/memutils.h"
 
 /*
  * Algorithm 5 from paper
  */
-static void
+static List *
 GetScanItems(IndexScanDesc scan, Datum q)
 {
 	HnswScanOpaque so = (HnswScanOpaque) scan->opaque;
 	Relation	index = scan->indexRelation;
 	FmgrInfo   *procinfo = so->procinfo;
 	Oid			collation = so->collation;
-	List	   *ep = NIL;
+	List	   *ep;
 	List	   *w;
 	HnswElement entryPoint = HnswGetEntryPoint(index);
 
 	if (entryPoint == NULL)
-		return;
+		return NIL;
 
-	ep = lappend(ep, HnswEntryCandidate(entryPoint, q, index, procinfo, collation, false));
+	ep = list_make1(HnswEntryCandidate(entryPoint, q, index, procinfo, collation, false));
 
 	for (int lc = entryPoint->level; lc >= 1; lc--)
 	{
-		w = HnswSearchLayer(q, ep, 1, lc, index, procinfo, collation, false, NULL, NULL);
+		w = HnswSearchLayer(q, ep, 1, lc, index, procinfo, collation, false, NULL);
 		ep = w;
 	}
 
-	so->w = HnswSearchLayer(q, ep, hnsw_ef_search, 0, index, procinfo, collation, false, NULL, NULL);
+	return HnswSearchLayer(q, ep, hnsw_ef_search, 0, index, procinfo, collation, false, NULL);
 }
 
 /*
@@ -144,7 +145,17 @@ hnswgettuple(IndexScanDesc scan, ScanDirection dir)
 				HnswNormValue(so->normprocinfo, so->collation, &value, NULL);
 		}
 
-		GetScanItems(scan, value);
+		/*
+		 * Get a shared lock. This allows vacuum to ensure no in-flight scans
+		 * before marking tuples as deleted.
+		 */
+		LockPage(scan->indexRelation, HNSW_SCAN_LOCK, ShareLock);
+
+		so->w = GetScanItems(scan, value);
+
+		/* Release shared lock */
+		UnlockPage(scan->indexRelation, HNSW_SCAN_LOCK, ShareLock);
+
 		so->first = false;
 	}
 
@@ -174,6 +185,7 @@ hnswgettuple(IndexScanDesc scan, ScanDirection dir)
 		scan->xs_ctup.t_self = *tid;
 #endif
 
+		/* Unpin buffer */
 		if (BufferIsValid(so->buf))
 			ReleaseBuffer(so->buf);
 

src/hnswutils.c

@@ -7,7 +7,7 @@
 #include "vector.h"
 
 /*
- * Get the number of connection in the index
+ * Get the max number of connections in an upper layer for each element in the index
  */
 int
 HnswGetM(Relation index)
@@ -197,8 +197,8 @@ HnswAddHeapTid(HnswElement element, ItemPointer heaptid)
 /*
  * Allocate an element from block and offset numbers
  */
-static HnswElement
-InitElementFromBlock(BlockNumber blkno, OffsetNumber offno)
+HnswElement
+HnswInitElementFromBlock(BlockNumber blkno, OffsetNumber offno)
 {
 	HnswElement element = palloc(sizeof(HnswElementData));
 
@@ -226,7 +226,7 @@ HnswGetEntryPoint(Relation index)
 	metap = HnswPageGetMeta(page);
 
 	if (BlockNumberIsValid(metap->entryBlkno))
-		entryPoint = InitElementFromBlock(metap->entryBlkno, metap->entryOffno);
+		entryPoint = HnswInitElementFromBlock(metap->entryBlkno, metap->entryOffno);
 
 	UnlockReleaseBuffer(buf);
 
@@ -234,22 +234,12 @@ HnswGetEntryPoint(Relation index)
 }
 
 /*
- * Update the metapage
+ * Update the metapage info
  */
-void
-HnswUpdateMetaPage(Relation index, bool updateEntry, HnswElement entryPoint, BlockNumber insertPage, ForkNumber forkNum)
+static void
+HnswUpdateMetaPageInfo(Page page, int updateEntry, HnswElement entryPoint, BlockNumber insertPage)
 {
-	Buffer		buf;
-	Page		page;
-	GenericXLogState *state;
-	HnswMetaPage metap;
-
-	buf = ReadBufferExtended(index, forkNum, HNSW_METAPAGE_BLKNO, RBM_NORMAL, NULL);
-	LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
-	state = GenericXLogStart(index);
-	page = GenericXLogRegisterBuffer(state, buf, 0);
-
-	metap = HnswPageGetMeta(page);
+	HnswMetaPage metap = HnswPageGetMeta(page);
 
 	if (updateEntry)
 	{
@@ -259,7 +249,7 @@ HnswUpdateMetaPage(Relation index, bool updateEntry, HnswElement entryPoint, Blo
 			metap->entryOffno = InvalidOffsetNumber;
 			metap->entryLevel = -1;
 		}
-		else
+		else if (entryPoint->level > metap->entryLevel || updateEntry == HNSW_UPDATE_ENTRY_ALWAYS)
 		{
 			metap->entryBlkno = entryPoint->blkno;
 			metap->entryOffno = entryPoint->offno;
@@ -269,6 +259,24 @@ HnswUpdateMetaPage(Relation index, bool updateEntry, HnswElement entryPoint, Blo
 
 	if (BlockNumberIsValid(insertPage))
 		metap->insertPage = insertPage;
+}
+
+/*
+ * Update the metapage
+ */
+void
+HnswUpdateMetaPage(Relation index, int updateEntry, HnswElement entryPoint, BlockNumber insertPage, ForkNumber forkNum)
+{
+	Buffer		buf;
+	Page		page;
+	GenericXLogState *state;
+
+	buf = ReadBufferExtended(index, forkNum, HNSW_METAPAGE_BLKNO, RBM_NORMAL, NULL);
+	LockBuffer(buf, BUFFER_LOCK_EXCLUSIVE);
+	state = GenericXLogStart(index);
+	page = GenericXLogRegisterBuffer(state, buf, 0);
+
+	HnswUpdateMetaPageInfo(page, updateEntry, entryPoint, insertPage);
 
 	HnswCommitBuffer(buf, state);
 }
@@ -356,7 +364,7 @@ LoadNeighborsFromPage(HnswElement element, Relation index, Page page)
 		if (!ItemPointerIsValid(indextid))
 			continue;
 
-		e = InitElementFromBlock(ItemPointerGetBlockNumber(indextid), ItemPointerGetOffsetNumber(indextid));
+		e = HnswInitElementFromBlock(ItemPointerGetBlockNumber(indextid), ItemPointerGetOffsetNumber(indextid));
 
 		/* Calculate level based on offset */
 		level = element->level - i / m;
@@ -372,8 +380,8 @@ LoadNeighborsFromPage(HnswElement element, Relation index, Page page)
 /*
  * Load neighbors
  */
-static void
-LoadNeighbors(HnswElement element, Relation index)
+void
+HnswLoadNeighbors(HnswElement element, Relation index)
 {
 	Buffer		buf;
 	Page		page;
@@ -387,6 +395,37 @@ LoadNeighbors(HnswElement element, Relation index)
 	UnlockReleaseBuffer(buf);
 }
 
+/*
+ * Load an element from a tuple
+ */
+void
+HnswLoadElementFromTuple(HnswElement element, HnswElementTuple etup, bool loadHeaptids, bool loadVec)
+{
+	element->level = etup->level;
+	element->deleted = etup->deleted;
+	element->neighborPage = ItemPointerGetBlockNumber(&etup->neighbortid);
+	element->neighborOffno = ItemPointerGetOffsetNumber(&etup->neighbortid);
+	element->heaptids = NIL;
+
+	if (loadHeaptids)
+	{
+		for (int i = 0; i < HNSW_HEAPTIDS; i++)
+		{
+			/* Can stop at first invalid */
+			if (!ItemPointerIsValid(&etup->heaptids[i]))
+				break;
+
+			HnswAddHeapTid(element, &etup->heaptids[i]);
+		}
+	}
+
+	if (loadVec)
+	{
+		element->vec = palloc(VECTOR_SIZE(etup->vec.dim));
+		memcpy(element->vec, &etup->vec, VECTOR_SIZE(etup->vec.dim));
+	}
+}
+
 /*
  * Load an element and optionally get its distance from q
  */
@@ -407,25 +446,7 @@ HnswLoadElement(HnswElement element, float *distance, Datum *q, Relation index,
 	Assert(HnswIsElementTuple(etup));
 
 	/* Load element */
-	element->heaptids = NIL;
-	for (int i = 0; i < HNSW_HEAPTIDS; i++)
-	{
-		/* Can stop at first invalid */
-		if (!ItemPointerIsValid(&etup->heaptids[i]))
-			break;
-
-		HnswAddHeapTid(element, &etup->heaptids[i]);
-	}
-	element->level = etup->level;
-	element->neighborPage = ItemPointerGetBlockNumber(&etup->neighbortid);
-	element->neighborOffno = ItemPointerGetOffsetNumber(&etup->neighbortid);
-	element->deleted = etup->deleted;
-
-	if (loadVec)
-	{
-		element->vec = palloc(VECTOR_SIZE(etup->vec.dim));
-		memcpy(element->vec, &etup->vec, VECTOR_SIZE(etup->vec.dim));
-	}
+	HnswLoadElementFromTuple(element, etup, true, loadVec);
 
 	/* Calculate distance */
 	if (distance != NULL)
@@ -447,7 +468,7 @@ GetCandidateDistance(HnswCandidate * hc, Datum q, FmgrInfo *procinfo, Oid collat
  * Create a candidate for the entry point
  */
 HnswCandidate *
-HnswEntryCandidate(HnswElement entryPoint, Datum q, Relation index, FmgrInfo *procinfo, Oid collation, bool loadvec)
+HnswEntryCandidate(HnswElement entryPoint, Datum q, Relation index, FmgrInfo *procinfo, Oid collation, bool loadVec)
 {
 	HnswCandidate *hc = palloc(sizeof(HnswCandidate));
 
@@ -455,7 +476,7 @@ HnswEntryCandidate(HnswElement entryPoint, Datum q, Relation index, FmgrInfo *pr
 	if (index == NULL)
 		hc->distance = GetCandidateDistance(hc, q, procinfo, collation);
 	else
-		HnswLoadElement(hc->element, &hc->distance, &q, index, procinfo, collation, loadvec);
+		HnswLoadElement(hc->element, &hc->distance, &q, index, procinfo, collation, loadVec);
 	return hc;
 }
 
@@ -522,7 +543,7 @@ AddToVisited(HTAB *v, HnswCandidate * hc, Relation index, bool *found)
  * Algorithm 2 from paper
  */
 List *
-HnswSearchLayer(Datum q, List *ep, int ef, int lc, Relation index, FmgrInfo *procinfo, Oid collation, bool inserting, BlockNumber *skipPage, OffsetNumber *skipOffno)
+HnswSearchLayer(Datum q, List *ep, int ef, int lc, Relation index, FmgrInfo *procinfo, Oid collation, bool inserting, HnswElement skipElement)
 {
 	ListCell   *lc2;
 
@@ -558,7 +579,13 @@ HnswSearchLayer(Datum q, List *ep, int ef, int lc, Relation index, FmgrInfo *pro
 		pairingheap_add(C, &(CreatePairingHeapNode(hc)->ph_node));
 		pairingheap_add(W, &(CreatePairingHeapNode(hc)->ph_node));
 
-		wlen++;
+		/*
+		 * Do not count elements being deleted towards ef when vacuuming. It
+		 * would be ideal to do this for inserts as well, but this could
+		 * affect insert performance.
+		 */
+		if (skipElement == NULL || list_length(hc->element->heaptids) != 0)
+			wlen++;
 	}
 
 	while (!pairingheap_is_empty(C))
@@ -571,7 +598,7 @@ HnswSearchLayer(Datum q, List *ep, int ef, int lc, Relation index, FmgrInfo *pro
 			break;
 
 		if (c->element->neighbors == NULL)
-			LoadNeighbors(c->element, index);
+			HnswLoadNeighbors(c->element, index);
 
 		/* Get the neighborhood at layer lc */
 		neighborhood = &c->element->neighbors[lc];
@@ -594,17 +621,7 @@ HnswSearchLayer(Datum q, List *ep, int ef, int lc, Relation index, FmgrInfo *pro
 				else
 					HnswLoadElement(e->element, &eDistance, &q, index, procinfo, collation, inserting);
 
-				/* Skip if fully deleted */
-				if (e->element->deleted)
-					continue;
-
-				/* Skip for inserts if deleting */
-				if (inserting && list_length(e->element->heaptids) == 0)
-					continue;
-
-				/* Skip self for vacuuming update */
-				if (skipPage != NULL && e->element->neighborPage == *skipPage && e->element->neighborOffno == *skipOffno)
-					continue;
+				Assert(!e->element->deleted);
 
 				/* Make robust to issues */
 				if (e->element->level < lc)
@@ -620,11 +637,20 @@ HnswSearchLayer(Datum q, List *ep, int ef, int lc, Relation index, FmgrInfo *pro
 
 					pairingheap_add(C, &(CreatePairingHeapNode(ec)->ph_node));
 					pairingheap_add(W, &(CreatePairingHeapNode(ec)->ph_node));
-					wlen++;
 
-					/* No need to decrement wlen */
-					if (wlen > ef)
-						pairingheap_remove_first(W);
+					/*
+					 * Do not count elements being deleted towards ef when
+					 * vacuuming. It would be ideal to do this for inserts as
+					 * well, but this could affect insert performance.
+					 */
+					if (skipElement == NULL || list_length(e->element->heaptids) != 0)
+					{
+						wlen++;
+
+						/* No need to decrement wlen */
+						if (wlen > ef)
+							pairingheap_remove_first(W);
+					}
 				}
 			}
 		}
@@ -703,7 +729,7 @@ SelectNeighbors(List *c, int m, int lc, FmgrInfo *procinfo, Oid collation, HnswC
 	List	   *w = list_copy(c);
 	pairingheap *wd;
 
-	if (list_length(w) < m)
+	if (list_length(w) <= m)
 		return w;
 
 	wd = pairingheap_allocate(CompareNearestCandidates, NULL);
@@ -743,14 +769,14 @@ SelectNeighbors(List *c, int m, int lc, FmgrInfo *procinfo, Oid collation, HnswC
 /*
  * Find duplicate element
  */
-static HnswElement
-HnswFindDuplicate(HnswElement e, List *neighbors)
+HnswElement
+HnswFindDuplicate(HnswElement e)
 {
-	ListCell   *lc;
+	HnswNeighborArray *neighbors = &e->neighbors[0];
 
-	foreach(lc, neighbors)
+	for (int i = 0; i < neighbors->length; i++)
 	{
-		HnswCandidate *neighbor = lfirst(lc);
+		HnswCandidate *neighbor = &neighbors->items[i];
 
 		/* Exit early since ordered by distance */
 		if (vector_cmp_internal(e->vec, neighbor->element->vec) != 0)
@@ -799,183 +825,168 @@ CompareCandidateDistances(const void *a, const void *b)
 	return 0;
 }
 
-/*
- * Create update
- */
-static HnswUpdate *
-CreateUpdate(HnswCandidate * hc, int level, int index)
-{
-	HnswUpdate *update = palloc(sizeof(HnswUpdate));
-
-	update->hc = *hc;
-	update->level = level;
-	update->index = index;
-	return update;
-}
-
 /*
  * Update connections
  */
-static void
-UpdateConnections(HnswElement element, List *neighbors, int m, int lc, List **updates, Relation index, FmgrInfo *procinfo, Oid collation)
+void
+HnswUpdateConnection(HnswElement element, HnswCandidate * hc, int m, int lc, int *updateIdx, Relation index, FmgrInfo *procinfo, Oid collation)
 {
-	ListCell   *lc2;
+	HnswNeighborArray *currentNeighbors = &hc->element->neighbors[lc];
 
-	foreach(lc2, neighbors)
+	HnswCandidate hc2;
+
+	hc2.element = element;
+	hc2.distance = hc->distance;
+
+	if (currentNeighbors->length < m)
 	{
-		HnswCandidate *hc = (HnswCandidate *) lfirst(lc2);
-		HnswNeighborArray *currentNeighbors = &hc->element->neighbors[lc];
+		currentNeighbors->items[currentNeighbors->length++] = hc2;
 
-		HnswCandidate hc2;
+		/* Track update */
+		if (updateIdx != NULL)
+			*updateIdx = -2;
+	}
+	else
+	{
+		/* Shrink connections */
+		HnswCandidate *pruned = NULL;
 
-		hc2.element = element;
-		hc2.distance = hc->distance;
-
-		if (currentNeighbors->length < m)
+		/* Load elements on insert */
+		if (index != NULL)
 		{
-			currentNeighbors->items[currentNeighbors->length++] = hc2;
+			Datum		q = PointerGetDatum(hc->element->vec);
 
-			/* Track updates */
-			if (updates != NULL)
-				*updates = lappend(*updates, CreateUpdate(hc, lc, currentNeighbors->length - 1));
-		}
-		else
-		{
-			/* Shrink connections */
-			HnswCandidate *pruned = NULL;
-			List	   *c = NIL;
-
-			/* Load elements on insert */
-			if (index != NULL)
-			{
-				Datum		q = PointerGetDatum(hc->element->vec);
-
-				for (int i = 0; i < currentNeighbors->length; i++)
-				{
-					HnswCandidate *hc3 = &currentNeighbors->items[i];
-
-					if (hc3->element->vec == NULL)
-						HnswLoadElement(hc3->element, &hc3->distance, &q, index, procinfo, collation, true);
-					else
-						hc3->distance = GetCandidateDistance(hc3, q, procinfo, collation);
-
-					/* Prune element if being deleted */
-					if (list_length(hc3->element->heaptids) == 0)
-					{
-						pruned = &currentNeighbors->items[i];
-						break;
-					}
-				}
-			}
-
-			if (pruned == NULL)
-			{
-				/* Add and sort candidates */
-				for (int i = 0; i < currentNeighbors->length; i++)
-					c = lappend(c, &currentNeighbors->items[i]);
-				c = lappend(c, &hc2);
-				list_sort(c, CompareCandidateDistances);
-
-				SelectNeighbors(c, m, lc, procinfo, collation, &pruned);
-
-				/* Should not happen */
-				if (pruned == NULL)
-					continue;
-			}
-
-			/* Find and replace the pruned element */
 			for (int i = 0; i < currentNeighbors->length; i++)
 			{
-				if (currentNeighbors->items[i].element == pruned->element)
+				HnswCandidate *hc3 = &currentNeighbors->items[i];
+
+				if (hc3->element->vec == NULL)
+					HnswLoadElement(hc3->element, &hc3->distance, &q, index, procinfo, collation, true);
+				else
+					hc3->distance = GetCandidateDistance(hc3, q, procinfo, collation);
+
+				/* Prune element if being deleted */
+				if (list_length(hc3->element->heaptids) == 0)
 				{
-					currentNeighbors->items[i] = hc2;
-
-					/* Track updates */
-					if (updates != NULL)
-						*updates = lappend(*updates, CreateUpdate(hc, lc, i));
-
+					pruned = &currentNeighbors->items[i];
 					break;
 				}
 			}
 		}
+
+		if (pruned == NULL)
+		{
+			List	   *c = NIL;
+
+			/* Add and sort candidates */
+			for (int i = 0; i < currentNeighbors->length; i++)
+				c = lappend(c, &currentNeighbors->items[i]);
+			c = lappend(c, &hc2);
+			list_sort(c, CompareCandidateDistances);
+
+			SelectNeighbors(c, m, lc, procinfo, collation, &pruned);
+
+			/* Should not happen */
+			if (pruned == NULL)
+				return;
+		}
+
+		/* Find and replace the pruned element */
+		for (int i = 0; i < currentNeighbors->length; i++)
+		{
+			if (currentNeighbors->items[i].element == pruned->element)
+			{
+				currentNeighbors->items[i] = hc2;
+
+				/* Track update */
+				if (updateIdx != NULL)
+					*updateIdx = i;
+
+				break;
+			}
+		}
 	}
 }
 
+/*
+ * Remove elements being deleted or skipped
+ */
+static List *
+RemoveElements(List *w, HnswElement skipElement)
+{
+	ListCell   *lc2;
+	List	   *w2 = NIL;
+
+	foreach(lc2, w)
+	{
+		HnswCandidate *hc = (HnswCandidate *) lfirst(lc2);
+
+		/* Skip self for vacuuming update */
+		if (skipElement != NULL && hc->element->blkno == skipElement->blkno && hc->element->offno == skipElement->offno)
+			continue;
+
+		if (list_length(hc->element->heaptids) != 0)
+			w2 = lappend(w2, hc);
+	}
+
+	return w2;
+}
+
 /*
  * Algorithm 1 from paper
  */
-HnswElement
-HnswInsertElement(HnswElement element, HnswElement entryPoint, Relation index, FmgrInfo *procinfo, Oid collation, int m, int efConstruction, List **updates, bool vacuuming)
+void
+HnswInsertElement(HnswElement element, HnswElement entryPoint, Relation index, FmgrInfo *procinfo, Oid collation, int m, int efConstruction, bool existing)
 {
-	List	   *ep = NIL;
+	List	   *ep;
 	List	   *w;
 	int			level = element->level;
 	int			entryLevel;
-	List	  **newNeighbors = palloc(sizeof(List *) * (level + 1));
 	Datum		q = PointerGetDatum(element->vec);
-	HnswElement dup;
-	BlockNumber *skipPage = vacuuming ? &element->neighborPage : NULL;
-	OffsetNumber *skipOffno = vacuuming ? &element->neighborOffno : NULL;
-	bool		removeEntryPoint;
-	HnswCandidate *entryCandidate;
+	HnswElement skipElement = existing ? element : NULL;
+
+	/* No neighbors if no entry point */
+	if (entryPoint == NULL)
+		return;
 
 	/* Get entry point and level */
-	if (entryPoint != NULL)
-	{
-		entryCandidate = HnswEntryCandidate(entryPoint, q, index, procinfo, collation, true);
-		ep = lappend(ep, entryCandidate);
-		entryLevel = entryPoint->level;
-		removeEntryPoint = vacuuming && list_length(entryPoint->heaptids) == 0;
-	}
-	else
-	{
-		entryLevel = -1;
-		removeEntryPoint = false;
-	}
+	ep = list_make1(HnswEntryCandidate(entryPoint, q, index, procinfo, collation, true));
+	entryLevel = entryPoint->level;
 
 	/* 1st phase: greedy search to insert level */
 	for (int lc = entryLevel; lc >= level + 1; lc--)
 	{
-		w = HnswSearchLayer(q, ep, 1, lc, index, procinfo, collation, true, skipPage, skipOffno);
+		w = HnswSearchLayer(q, ep, 1, lc, index, procinfo, collation, true, skipElement);
 		ep = w;
 	}
 
 	if (level > entryLevel)
 		level = entryLevel;
 
+	/* Add one for existing element */
+	if (existing)
+		efConstruction++;
+
 	/* 2nd phase */
 	for (int lc = level; lc >= 0; lc--)
 	{
 		int			lm = HnswGetLayerM(m, lc);
+		List	   *neighbors;
+		List	   *lw;
 
-		w = HnswSearchLayer(q, ep, efConstruction, lc, index, procinfo, collation, true, skipPage, skipOffno);
+		w = HnswSearchLayer(q, ep, efConstruction, lc, index, procinfo, collation, true, skipElement);
 
-		/* Remove entry point if it's being deleted */
-		if (removeEntryPoint)
-			w = list_delete_ptr(w, entryCandidate);
+		/* Elements being deleted or skipped can help with search */
+		/* but should be removed before selecting neighbors */
+		if (index != NULL)
+			lw = RemoveElements(w, skipElement);
+		else
+			lw = w;
+
+		neighbors = SelectNeighbors(lw, lm, lc, procinfo, collation, NULL);
+
+		AddConnections(element, neighbors, lm, lc);
 
-		newNeighbors[lc] = SelectNeighbors(w, lm, lc, procinfo, collation, NULL);
 		ep = w;
 	}
-
-	/* Look for duplicate */
-	if (level >= 0 && !vacuuming)
-	{
-		dup = HnswFindDuplicate(element, newNeighbors[0]);
-		if (dup != NULL)
-			return dup;
-	}
-
-	/* Update connections */
-	for (int lc = level; lc >= 0; lc--)
-	{
-		int			lm = HnswGetLayerM(m, lc);
-
-		AddConnections(element, newNeighbors[lc], lm, lc);
-
-		if (!vacuuming)
-			UpdateConnections(element, newNeighbors[lc], lm, lc, updates, index, procinfo, collation);
-	}
-
-	return NULL;
 }

src/hnswvacuum.c

@@ -5,10 +5,11 @@
 #include "commands/vacuum.h"
 #include "hnsw.h"
 #include "storage/bufmgr.h"
+#include "storage/lmgr.h"
 #include "utils/memutils.h"
 
 /*
- * Check if deleted list contains an index tid
+ * Check if deleted list contains an index TID
  */
 static bool
 DeletedContains(HTAB *deleted, ItemPointer indextid)
@@ -32,6 +33,7 @@ RemoveHeapTids(HnswVacuumState * vacuumstate)
 	Relation	index = vacuumstate->index;
 	BufferAccessStrategy bas = vacuumstate->bas;
 	HnswElement entryPoint = HnswGetEntryPoint(vacuumstate->index);
+	IndexBulkDeleteResult *stats = vacuumstate->stats;
 
 	/* Store separately since highestPoint.level is uint8 */
 	int			highestLevel = -1;
@@ -77,11 +79,15 @@ RemoveHeapTids(HnswVacuumState * vacuumstate)
 						break;
 
 					if (vacuumstate->callback(&etup->heaptids[i], vacuumstate->callback_state))
+					{
 						itemUpdated = true;
+						stats->tuples_removed++;
+					}
 					else
 					{
 						/* Move to front of list */
 						etup->heaptids[idx++] = etup->heaptids[i];
+						stats->num_index_tuples++;
 					}
 				}
 
@@ -109,7 +115,7 @@ RemoveHeapTids(HnswVacuumState * vacuumstate)
 
 				(void) hash_search(vacuumstate->deleted, &ip, HASH_ENTER, NULL);
 			}
-			else if (etup->level > highestLevel && !(blkno == entryPoint->blkno && offno == entryPoint->offno))
+			else if (etup->level > highestLevel && !(entryPoint != NULL && blkno == entryPoint->blkno && offno == entryPoint->offno))
 			{
 				/* Keep track of highest non-entry point */
 				highestPoint->blkno = blkno;
@@ -169,6 +175,11 @@ NeedsUpdated(HnswVacuumState * vacuumstate, HnswElement element)
 		}
 	}
 
+	/* Also update if layer 0 is not full */
+	/* This could indicate too many candidates being deleted during insert */
+	if (!needsUpdated)
+		needsUpdated = !ItemPointerIsValid(&ntup->indextids[ntup->count - 1]);
+
 	UnlockReleaseBuffer(buf);
 
 	return needsUpdated;
@@ -178,7 +189,7 @@ NeedsUpdated(HnswVacuumState * vacuumstate, HnswElement element)
  * Repair graph for a single element
  */
 static void
-RepairGraphElement(HnswVacuumState * vacuumstate, HnswElement element)
+RepairGraphElement(HnswVacuumState * vacuumstate, HnswElement element, HnswElement entryPoint)
 {
 	Relation	index = vacuumstate->index;
 	Buffer		buf;
@@ -188,42 +199,20 @@ RepairGraphElement(HnswVacuumState * vacuumstate, HnswElement element)
 	int			efConstruction = vacuumstate->efConstruction;
 	FmgrInfo   *procinfo = vacuumstate->procinfo;
 	Oid			collation = vacuumstate->collation;
-	HnswElement entryPoint;
 	BufferAccessStrategy bas = vacuumstate->bas;
 	HnswNeighborTuple ntup = vacuumstate->ntup;
 	Size		ntupSize = HNSW_NEIGHBOR_TUPLE_SIZE(element->level, m);
 
-	/* Check if any neighbors point to deleted values */
-	if (!NeedsUpdated(vacuumstate, element))
+	/* Skip if element is entry point */
+	if (entryPoint != NULL && element->blkno == entryPoint->blkno && element->offno == entryPoint->offno)
 		return;
 
-	/* Refresh entry point for each element */
-	entryPoint = HnswGetEntryPoint(index);
-
-	/* Special case for entry point */
-	if (element->blkno == entryPoint->blkno && element->offno == entryPoint->offno)
-	{
-		if (BlockNumberIsValid(vacuumstate->highestPoint.blkno))
-		{
-			/* Already updated */
-			if (vacuumstate->highestPoint.blkno == element->blkno && vacuumstate->highestPoint.offno == element->offno)
-				return;
-
-			entryPoint = &vacuumstate->highestPoint;
-
-			/* Reset neighbors from previous update */
-			entryPoint->neighbors = NULL;
-		}
-		else
-			entryPoint = NULL;
-	}
-
 	/* Init fields */
 	HnswInitNeighbors(element, m);
 	element->heaptids = NIL;
 
 	/* Add element to graph, skipping itself */
-	HnswInsertElement(element, entryPoint, index, procinfo, collation, m, efConstruction, NULL, true);
+	HnswInsertElement(element, entryPoint, index, procinfo, collation, m, efConstruction, true);
 
 	/* Update neighbor tuple */
 	/* Do this before getting page to minimize locking */
@@ -243,6 +232,9 @@ RepairGraphElement(HnswVacuumState * vacuumstate, HnswElement element)
 	MarkBufferDirty(buf);
 	GenericXLogFinish(state);
 	UnlockReleaseBuffer(buf);
+
+	/* Update neighbors */
+	HnswUpdateNeighborPages(index, procinfo, collation, element, m, true);
 }
 
 /*
@@ -256,16 +248,35 @@ RepairGraphEntryPoint(HnswVacuumState * vacuumstate)
 	HnswElement entryPoint;
 	MemoryContext oldCtx = MemoryContextSwitchTo(vacuumstate->tmpCtx);
 
-	/* Repair graph for highest non-entry point */
-	/* This may not be the highest with new inserts, but should be fine */
-	if (BlockNumberIsValid(highestPoint->blkno))
+	if (!BlockNumberIsValid(highestPoint->blkno))
+		highestPoint = NULL;
+
+	/*
+	 * Repair graph for highest non-entry point. Highest point may be outdated
+	 * due to inserts that happen during and after RemoveHeapTids.
+	 */
+	if (highestPoint != NULL)
 	{
+		/* Get a shared lock */
+		LockPage(index, HNSW_UPDATE_LOCK, ShareLock);
+
+		/* Load element */
 		HnswLoadElement(highestPoint, NULL, NULL, index, vacuumstate->procinfo, vacuumstate->collation, true);
-		RepairGraphElement(vacuumstate, highestPoint);
+
+		/* Repair if needed */
+		if (NeedsUpdated(vacuumstate, highestPoint))
+			RepairGraphElement(vacuumstate, highestPoint, HnswGetEntryPoint(index));
+
+		/* Release lock */
+		UnlockPage(index, HNSW_UPDATE_LOCK, ShareLock);
 	}
 
-	/* See if entry point needs updated */
+	/* Prevent concurrent inserts when possibly updating entry point */
+	LockPage(index, HNSW_UPDATE_LOCK, ExclusiveLock);
+
+	/* Get latest entry point */
 	entryPoint = HnswGetEntryPoint(index);
+
 	if (entryPoint != NULL)
 	{
 		ItemPointerData epData;
@@ -273,15 +284,37 @@ RepairGraphEntryPoint(HnswVacuumState * vacuumstate)
 		ItemPointerSet(&epData, entryPoint->blkno, entryPoint->offno);
 
 		if (DeletedContains(vacuumstate->deleted, &epData))
-			HnswUpdateMetaPage(index, true, highestPoint, InvalidBlockNumber, MAIN_FORKNUM);
+		{
+			/*
+			 * Replace the entry point with the highest point. If highest
+			 * point is outdated and empty, the entry point will be empty
+			 * until an element is repaired.
+			 */
+			HnswUpdateMetaPage(index, HNSW_UPDATE_ENTRY_ALWAYS, highestPoint, InvalidBlockNumber, MAIN_FORKNUM);
+		}
 		else
 		{
-			/* Highest point will be used to repair */
+			/*
+			 * Repair the entry point with the highest point. If highest point
+			 * is outdated, this can remove connections at higher levels in
+			 * the graph until they are repaired, but this should be fine.
+			 */
 			HnswLoadElement(entryPoint, NULL, NULL, index, vacuumstate->procinfo, vacuumstate->collation, true);
-			RepairGraphElement(vacuumstate, entryPoint);
+
+			if (NeedsUpdated(vacuumstate, entryPoint))
+			{
+				/* Reset neighbors from previous update */
+				if (highestPoint != NULL)
+					highestPoint->neighbors = NULL;
+
+				RepairGraphElement(vacuumstate, entryPoint, highestPoint);
+			}
 		}
 	}
 
+	/* Release lock */
+	UnlockPage(index, HNSW_UPDATE_LOCK, ExclusiveLock);
+
 	/* Reset memory context */
 	MemoryContextSwitchTo(oldCtx);
 	MemoryContextReset(vacuumstate->tmpCtx);
@@ -297,6 +330,11 @@ RepairGraph(HnswVacuumState * vacuumstate)
 	BufferAccessStrategy bas = vacuumstate->bas;
 	BlockNumber blkno = HNSW_HEAD_BLKNO;
 
+	/* Wait for inserts to complete */
+	LockPage(index, HNSW_UPDATE_LOCK, ExclusiveLock);
+	UnlockPage(index, HNSW_UPDATE_LOCK, ExclusiveLock);
+
+	/* Repair entry point first */
 	RepairGraphEntryPoint(vacuumstate);
 
 	while (BlockNumberIsValid(blkno))
@@ -333,14 +371,8 @@ RepairGraph(HnswVacuumState * vacuumstate)
 				continue;
 
 			/* Create an element */
-			element = palloc(sizeof(HnswElementData));
-			element->neighborPage = ItemPointerGetBlockNumber(&etup->neighbortid);
-			element->neighborOffno = ItemPointerGetOffsetNumber(&etup->neighbortid);
-			element->level = etup->level;
-			element->blkno = blkno;
-			element->offno = offno;
-			element->vec = palloc(VECTOR_SIZE(etup->vec.dim));
-			memcpy(element->vec, &etup->vec, VECTOR_SIZE(etup->vec.dim));
+			element = HnswInitElementFromBlock(blkno, offno);
+			HnswLoadElementFromTuple(element, etup, false, true);
 
 			elements = lappend(elements, element);
 		}
@@ -351,7 +383,48 @@ RepairGraph(HnswVacuumState * vacuumstate)
 
 		/* Update neighbor pages */
 		foreach(lc2, elements)
-			RepairGraphElement(vacuumstate, (HnswElement) lfirst(lc2));
+		{
+			HnswElement element = (HnswElement) lfirst(lc2);
+			HnswElement entryPoint;
+			LOCKMODE	lockmode = ShareLock;
+
+			/* Check if any neighbors point to deleted values */
+			if (!NeedsUpdated(vacuumstate, element))
+				continue;
+
+			/* Get a shared lock */
+			LockPage(index, HNSW_UPDATE_LOCK, lockmode);
+
+			/* Refresh entry point for each element */
+			entryPoint = HnswGetEntryPoint(index);
+
+			/* Prevent concurrent inserts when likely updating entry point */
+			if (entryPoint == NULL || element->level > entryPoint->level)
+			{
+				/* Release shared lock */
+				UnlockPage(index, HNSW_UPDATE_LOCK, lockmode);
+
+				/* Get exclusive lock */
+				lockmode = ExclusiveLock;
+				LockPage(index, HNSW_UPDATE_LOCK, lockmode);
+
+				/* Get latest entry point after lock is acquired */
+				entryPoint = HnswGetEntryPoint(index);
+			}
+
+			/* Repair connections */
+			RepairGraphElement(vacuumstate, element, entryPoint);
+
+			/*
+			 * Update metapage if needed. Should only happen if entry point
+			 * was replaced and highest point was outdated.
+			 */
+			if (entryPoint == NULL || element->level > entryPoint->level)
+				HnswUpdateMetaPage(index, HNSW_UPDATE_ENTRY_GREATER, element, InvalidBlockNumber, MAIN_FORKNUM);
+
+			/* Release lock */
+			UnlockPage(index, HNSW_UPDATE_LOCK, lockmode);
+		}
 
 		/* Reset memory context */
 		MemoryContextSwitchTo(oldCtx);
@@ -369,7 +442,10 @@ MarkDeleted(HnswVacuumState * vacuumstate)
 	BlockNumber insertPage = InvalidBlockNumber;
 	Relation	index = vacuumstate->index;
 	BufferAccessStrategy bas = vacuumstate->bas;
-	IndexBulkDeleteResult *stats = vacuumstate->stats;
+
+	/* Wait for selects to complete */
+	LockPage(index, HNSW_SCAN_LOCK, ExclusiveLock);
+	UnlockPage(index, HNSW_SCAN_LOCK, ExclusiveLock);
 
 	while (BlockNumberIsValid(blkno))
 	{
@@ -413,17 +489,17 @@ MarkDeleted(HnswVacuumState * vacuumstate)
 
 			/* Skip deleted tuples */
 			if (etup->deleted)
-				continue;
-
-			/* Skip live tuples */
-			if (ItemPointerIsValid(&etup->heaptids[0]))
 			{
-				stats->num_index_tuples++;
+				/* Set to first free page */
+				if (!BlockNumberIsValid(insertPage))
+					insertPage = blkno;
+
 				continue;
 			}
 
-			/* Update stats */
-			stats->tuples_removed++;
+			/* Skip live tuples */
+			if (ItemPointerIsValid(&etup->heaptids[0]))
+				continue;
 
 			/* Calculate sizes */
 			etupSize = HNSW_ELEMENT_TUPLE_SIZE(etup->vec.dim);
@@ -486,7 +562,8 @@ MarkDeleted(HnswVacuumState * vacuumstate)
 		UnlockReleaseBuffer(buf);
 	}
 
-	HnswUpdateMetaPage(index, false, NULL, insertPage, MAIN_FORKNUM);
+	/* Update insert page last, after everything has been marked as deleted */
+	HnswUpdateMetaPage(index, 0, NULL, insertPage, MAIN_FORKNUM);
 }
 
 /*

src/ivfflat.c

@@ -24,7 +24,7 @@ IvfflatInit(void)
 {
 	ivfflat_relopt_kind = add_reloption_kind();
 	add_int_reloption(ivfflat_relopt_kind, "lists", "Number of inverted lists",
-					  IVFFLAT_DEFAULT_LISTS, 1, IVFFLAT_MAX_LISTS
+					  IVFFLAT_DEFAULT_LISTS, IVFFLAT_MIN_LISTS, IVFFLAT_MAX_LISTS
 #if PG_VERSION_NUM >= 130000
 					  ,AccessExclusiveLock
 #endif
@@ -32,7 +32,7 @@ IvfflatInit(void)
 
 	DefineCustomIntVariable("ivfflat.probes", "Sets the number of probes",
 							"Valid range is 1..lists.", &ivfflat_probes,
-							1, 1, IVFFLAT_MAX_LISTS, PGC_USERSET, 0, NULL, NULL, NULL);
+							IVFFLAT_DEFAULT_PROBES, IVFFLAT_MIN_LISTS, IVFFLAT_MAX_LISTS, PGC_USERSET, 0, NULL, NULL, NULL);
 }
 
 /*

src/ivfflat.h

@@ -40,8 +40,11 @@
 #define IVFFLAT_METAPAGE_BLKNO	0
 #define IVFFLAT_HEAD_BLKNO		1	/* first list page */
 
+/* IVFFlat parameters */
 #define IVFFLAT_DEFAULT_LISTS	100
+#define IVFFLAT_MIN_LISTS		1
 #define IVFFLAT_MAX_LISTS		32768
+#define IVFFLAT_DEFAULT_PROBES	1
 
 /* Build phases */
 /* PROGRESS_CREATEIDX_SUBPHASE_INITIALIZE is 1 */

src/ivfscan.c

@@ -343,6 +343,7 @@ ivfflatgettuple(IndexScanDesc scan, ScanDirection dir)
 		scan->xs_ctup.t_self = *tid;
 #endif
 
+		/* Unpin buffer */
 		if (BufferIsValid(so->buf))
 			ReleaseBuffer(so->buf);
 

src/vector.c

@@ -15,6 +15,10 @@
 #include "utils/numeric.h"
 #include "vector.h"
 
+#if PG_VERSION_NUM >= 160000
+#include "varatt.h"
+#endif
+
 #if PG_VERSION_NUM >= 120000
 #include "common/shortest_dec.h"
 #include "utils/float.h"
@@ -35,6 +39,7 @@ PG_MODULE_MAGIC;
 /*
  * Initialize index options and variables
  */
+PGDLLEXPORT void _PG_init(void);
 void
 _PG_init(void)
 {
@@ -100,6 +105,23 @@ CheckElement(float value)
 				 errmsg("infinite value not allowed in vector")));
 }
 
+/*
+ * Allocate and initialize a new vector
+ */
+Vector *
+InitVector(int dim)
+{
+	Vector	   *result;
+	int			size;
+
+	size = VECTOR_SIZE(dim);
+	result = (Vector *) palloc0(size);
+	SET_VARSIZE(result, size);
+	result->dim = dim;
+
+	return result;
+}
+
 /*
  * Check for whitespace, since array_isspace() is static
  */
@@ -733,55 +755,16 @@ vector_dims(PG_FUNCTION_ARGS)
 PGDLLEXPORT PG_FUNCTION_INFO_V1(vector_norm);
 Datum
 vector_norm(PG_FUNCTION_ARGS)
-{
-	Vector	   *a = PG_GETARG_VECTOR_P(0);
-	float	   *ax = a->x;
-	float		norm = 0.0;
-
-	/* Auto-vectorized */
-	for (int i = 0; i < a->dim; i++)
-		norm += ax[i] * ax[i];
-
-	PG_RETURN_FLOAT8(sqrt((double) norm));
-}
-
-/*
- * Normalize a vector with the L2 norm
- */
-PGDLLEXPORT PG_FUNCTION_INFO_V1(normalize_l2);
-Datum
-normalize_l2(PG_FUNCTION_ARGS)
 {
 	Vector	   *a = PG_GETARG_VECTOR_P(0);
 	float	   *ax = a->x;
 	double		norm = 0.0;
-	Vector	   *result;
-	float	   *rx;
-
-	result = InitVector(a->dim);
-	rx = result->x;
 
 	/* Auto-vectorized */
 	for (int i = 0; i < a->dim; i++)
 		norm += (double) ax[i] * (double) ax[i];
 
-	norm = sqrt(norm);
-
-	if (norm > 0)
-	{
-		/* Auto-vectorized */
-		for (int i = 0, imax = a->dim; i < imax; i++)
-			rx[i] = ax[i] / norm;
-
-		/* Check for overflow */
-		for (int i = 0, imax = a->dim; i < imax; i++)
-		{
-			if (isinf(rx[i]))
-				float_overflow_error();
-		}
-	}
-
-	PG_RETURN_POINTER(result);
+	PG_RETURN_FLOAT8(sqrt(norm));
 }
 
 /*

src/vector.h

@@ -1,12 +1,6 @@
 #ifndef VECTOR_H
 #define VECTOR_H
 
-#include "postgres.h"
-
-#if PG_VERSION_NUM >= 160000
-#include "varatt.h"
-#endif
-
 #define VECTOR_MAX_DIM 16000
 
 #define VECTOR_SIZE(_dim)		(offsetof(Vector, x) + sizeof(float)*(_dim))
@@ -14,9 +8,6 @@
 #define PG_GETARG_VECTOR_P(x)	DatumGetVector(PG_GETARG_DATUM(x))
 #define PG_RETURN_VECTOR_P(x)	PG_RETURN_POINTER(x)
 
-/* Exported functions */
-PGDLLEXPORT void _PG_init(void);
-
 typedef struct Vector
 {
 	int32		vl_len_;		/* varlena header (do not touch directly!) */
@@ -25,24 +16,8 @@ typedef struct Vector
 	float		x[FLEXIBLE_ARRAY_MEMBER];
 }			Vector;
 
+Vector	   *InitVector(int dim);
 void		PrintVector(char *msg, Vector * vector);
 int			vector_cmp_internal(Vector * a, Vector * b);
 
-/*
- * Allocate and initialize a new vector
- */
-static inline Vector *
-InitVector(int dim)
-{
-	Vector	   *result;
-	int			size;
-
-	size = VECTOR_SIZE(dim);
-	result = (Vector *) palloc0(size);
-	SET_VARSIZE(result, size);
-	result->dim = dim;
-
-	return result;
-}
-
 #endif

test/expected/functions.out

@@ -48,34 +48,10 @@ SELECT vector_norm('[0,1]');
            1
 (1 row)
 
-SELECT normalize_l2('[3,4]');
- normalize_l2 
---------------
- [0.6,0.8]
-(1 row)
-
-SELECT normalize_l2('[3,0]');
- normalize_l2 
---------------
- [1,0]
-(1 row)
-
-SELECT normalize_l2('[0,0.1]');
- normalize_l2 
---------------
- [0,1]
-(1 row)
-
-SELECT normalize_l2('[0,0]');
- normalize_l2 
---------------
- [0,0]
-(1 row)
-
-SELECT normalize_l2('[3e38]');
- normalize_l2 
---------------
- [1]
+SELECT vector_norm('[3e37,4e37]')::real;
+ vector_norm 
+-------------
+       5e+37
 (1 row)
 
 SELECT l2_distance('[0,0]', '[3,4]');

test/expected/hnsw_l2.out

@@ -27,4 +27,10 @@ SELECT COUNT(*) FROM t;
      5
 (1 row)
 
+TRUNCATE t;
+SELECT * FROM t ORDER BY val <-> '[3,3,3]';
+ val 
+-----
+(0 rows)
+
 DROP TABLE t;

test/expected/hnsw_options.out

@@ -1,25 +1,26 @@
-SET enable_seqscan = off;
 CREATE TABLE t (val vector(3));
-CREATE INDEX ON t USING hnsw (val vector_l2_ops) WITH (m = 3);
-ERROR:  value 3 out of bounds for option "m"
-DETAIL:  Valid values are between "4" and "100".
+CREATE INDEX ON t USING hnsw (val vector_l2_ops) WITH (m = 1);
+ERROR:  value 1 out of bounds for option "m"
+DETAIL:  Valid values are between "2" and "100".
 CREATE INDEX ON t USING hnsw (val vector_l2_ops) WITH (m = 101);
 ERROR:  value 101 out of bounds for option "m"
-DETAIL:  Valid values are between "4" and "100".
-CREATE INDEX ON t USING hnsw (val vector_l2_ops) WITH (ef_construction = 9);
-ERROR:  value 9 out of bounds for option "ef_construction"
-DETAIL:  Valid values are between "10" and "1000".
+DETAIL:  Valid values are between "2" and "100".
+CREATE INDEX ON t USING hnsw (val vector_l2_ops) WITH (ef_construction = 3);
+ERROR:  value 3 out of bounds for option "ef_construction"
+DETAIL:  Valid values are between "4" and "1000".
 CREATE INDEX ON t USING hnsw (val vector_l2_ops) WITH (ef_construction = 1001);
 ERROR:  value 1001 out of bounds for option "ef_construction"
-DETAIL:  Valid values are between "10" and "1000".
+DETAIL:  Valid values are between "4" and "1000".
+CREATE INDEX ON t USING hnsw (val vector_l2_ops) WITH (m = 16, ef_construction = 31);
+ERROR:  ef_construction must be greater than or equal to 2 * m
 SHOW hnsw.ef_search;
  hnsw.ef_search 
 ----------------
  40
 (1 row)
 
-SET hnsw.ef_search = 9;
-ERROR:  9 is outside the valid range for parameter "hnsw.ef_search" (10 .. 1000)
+SET hnsw.ef_search = 0;
+ERROR:  0 is outside the valid range for parameter "hnsw.ef_search" (1 .. 1000)
 SET hnsw.ef_search = 1001;
-ERROR:  1001 is outside the valid range for parameter "hnsw.ef_search" (10 .. 1000)
+ERROR:  1001 is outside the valid range for parameter "hnsw.ef_search" (1 .. 1000)
 DROP TABLE t;

test/expected/ivfflat_l2.out

@@ -1,7 +1,7 @@
 SET enable_seqscan = off;
 CREATE TABLE t (val vector(3));
 INSERT INTO t (val) VALUES ('[0,0,0]'), ('[1,2,3]'), ('[1,1,1]'), (NULL);
-CREATE INDEX ON t USING ivfflat (val) WITH (lists = 1);
+CREATE INDEX ON t USING ivfflat (val vector_l2_ops) WITH (lists = 1);
 INSERT INTO t (val) VALUES ('[1,2,4]');
 SELECT * FROM t ORDER BY val <-> '[3,3,3]';
    val   
@@ -27,4 +27,13 @@ SELECT COUNT(*) FROM t;
      5
 (1 row)
 
+TRUNCATE t;
+NOTICE:  ivfflat index created with little data
+DETAIL:  This will cause low recall.
+HINT:  Drop the index until the table has more data.
+SELECT * FROM t ORDER BY val <-> '[3,3,3]';
+ val 
+-----
+(0 rows)
+
 DROP TABLE t;

test/expected/ivfflat_options.out

@@ -1,9 +1,8 @@
-SET enable_seqscan = off;
 CREATE TABLE t (val vector(3));
-CREATE INDEX ON t USING ivfflat (val) WITH (lists = 0);
+CREATE INDEX ON t USING ivfflat (val vector_l2_ops) WITH (lists = 0);
 ERROR:  value 0 out of bounds for option "lists"
 DETAIL:  Valid values are between "1" and "32768".
-CREATE INDEX ON t USING ivfflat (val) WITH (lists = 32769);
+CREATE INDEX ON t USING ivfflat (val vector_l2_ops) WITH (lists = 32769);
 ERROR:  value 32769 out of bounds for option "lists"
 DETAIL:  Valid values are between "1" and "32768".
 SHOW ivfflat.probes;

test/expected/ivfflat_unlogged.out

@@ -1,7 +1,7 @@
 SET enable_seqscan = off;
 CREATE UNLOGGED TABLE t (val vector(3));
 INSERT INTO t (val) VALUES ('[0,0,0]'), ('[1,2,3]'), ('[1,1,1]'), (NULL);
-CREATE INDEX ON t USING ivfflat (val) WITH (lists = 1);
+CREATE INDEX ON t USING ivfflat (val vector_l2_ops) WITH (lists = 1);
 SELECT * FROM t ORDER BY val <-> '[3,3,3]';
    val   
 ---------

test/sql/functions.sql

@@ -11,12 +11,7 @@ SELECT vector_dims('[1,2,3]');
 SELECT round(vector_norm('[1,1]')::numeric, 5);
 SELECT vector_norm('[3,4]');
 SELECT vector_norm('[0,1]');
-
-SELECT normalize_l2('[3,4]');
-SELECT normalize_l2('[3,0]');
-SELECT normalize_l2('[0,0.1]');
-SELECT normalize_l2('[0,0]');
-SELECT normalize_l2('[3e38]');
+SELECT vector_norm('[3e37,4e37]')::real;
 
 SELECT l2_distance('[0,0]', '[3,4]');
 SELECT l2_distance('[0,0]', '[0,1]');

test/sql/hnsw_l2.sql

@@ -10,4 +10,7 @@ SELECT * FROM t ORDER BY val <-> '[3,3,3]';
 SELECT * FROM t ORDER BY val <-> (SELECT NULL::vector);
 SELECT COUNT(*) FROM t;
 
+TRUNCATE t;
+SELECT * FROM t ORDER BY val <-> '[3,3,3]';
+
 DROP TABLE t;

test/sql/hnsw_options.sql

@@ -1,14 +1,13 @@
-SET enable_seqscan = off;
-
 CREATE TABLE t (val vector(3));
-CREATE INDEX ON t USING hnsw (val vector_l2_ops) WITH (m = 3);
+CREATE INDEX ON t USING hnsw (val vector_l2_ops) WITH (m = 1);
 CREATE INDEX ON t USING hnsw (val vector_l2_ops) WITH (m = 101);
-CREATE INDEX ON t USING hnsw (val vector_l2_ops) WITH (ef_construction = 9);
+CREATE INDEX ON t USING hnsw (val vector_l2_ops) WITH (ef_construction = 3);
 CREATE INDEX ON t USING hnsw (val vector_l2_ops) WITH (ef_construction = 1001);
+CREATE INDEX ON t USING hnsw (val vector_l2_ops) WITH (m = 16, ef_construction = 31);
 
 SHOW hnsw.ef_search;
 
-SET hnsw.ef_search = 9;
+SET hnsw.ef_search = 0;
 SET hnsw.ef_search = 1001;
 
 DROP TABLE t;

test/sql/ivfflat_l2.sql

@@ -2,7 +2,7 @@ SET enable_seqscan = off;
 
 CREATE TABLE t (val vector(3));
 INSERT INTO t (val) VALUES ('[0,0,0]'), ('[1,2,3]'), ('[1,1,1]'), (NULL);
-CREATE INDEX ON t USING ivfflat (val) WITH (lists = 1);
+CREATE INDEX ON t USING ivfflat (val vector_l2_ops) WITH (lists = 1);
 
 INSERT INTO t (val) VALUES ('[1,2,4]');
 
@@ -10,4 +10,7 @@ SELECT * FROM t ORDER BY val <-> '[3,3,3]';
 SELECT * FROM t ORDER BY val <-> (SELECT NULL::vector);
 SELECT COUNT(*) FROM t;
 
+TRUNCATE t;
+SELECT * FROM t ORDER BY val <-> '[3,3,3]';
+
 DROP TABLE t;

test/sql/ivfflat_options.sql

@@ -1,8 +1,6 @@
-SET enable_seqscan = off;
-
 CREATE TABLE t (val vector(3));
-CREATE INDEX ON t USING ivfflat (val) WITH (lists = 0);
-CREATE INDEX ON t USING ivfflat (val) WITH (lists = 32769);
+CREATE INDEX ON t USING ivfflat (val vector_l2_ops) WITH (lists = 0);
+CREATE INDEX ON t USING ivfflat (val vector_l2_ops) WITH (lists = 32769);
 
 SHOW ivfflat.probes;
 

test/sql/ivfflat_unlogged.sql

@@ -2,7 +2,7 @@ SET enable_seqscan = off;
 
 CREATE UNLOGGED TABLE t (val vector(3));
 INSERT INTO t (val) VALUES ('[0,0,0]'), ('[1,2,3]'), ('[1,1,1]'), (NULL);
-CREATE INDEX ON t USING ivfflat (val) WITH (lists = 1);
+CREATE INDEX ON t USING ivfflat (val vector_l2_ops) WITH (lists = 1);
 
 SELECT * FROM t ORDER BY val <-> '[3,3,3]';
 

test/t/001_ivfflat_wal.pl

@@ -26,7 +26,8 @@ sub test_index_replay
 	  or die "Timed out while waiting for replica 1 to catch up";
 
 	my @r = ();
-	for (1 .. $dim) {
+	for (1 .. $dim)
+	{
 		push(@r, rand());
 	}
 	my $sql = join(",", @r);
@@ -52,11 +53,13 @@ my $array_sql = join(",", ('random()') x $dim);
 # Initialize primary node
 $node_primary = get_new_node('primary');
 $node_primary->init(allows_streaming => 1);
-if ($dim > 32) {
+if ($dim > 32)
+{
 	# TODO use wal_keep_segments for Postgres < 13
 	$node_primary->append_conf('postgresql.conf', qq(wal_keep_size = 1GB));
 }
-if ($dim > 1500) {
+if ($dim > 1500)
+{
 	$node_primary->append_conf('postgresql.conf', qq(maintenance_work_mem = 128MB));
 }
 $node_primary->start;
@@ -67,8 +70,7 @@ $node_primary->backup($backup_name);
 
 # Create streaming replica linking to primary
 $node_replica = get_new_node('replica');
-$node_replica->init_from_backup($node_primary, $backup_name,
-	has_streaming => 1);
+$node_replica->init_from_backup($node_primary, $backup_name, has_streaming => 1);
 $node_replica->start;
 
 # Create ivfflat index on primary
@@ -77,7 +79,7 @@ $node_primary->safe_psql("postgres", "CREATE TABLE tst (i int4, v vector($dim));
 $node_primary->safe_psql("postgres",
 	"INSERT INTO tst SELECT i % 10, ARRAY[$array_sql] FROM generate_series(1, 100000) i;"
 );
-$node_primary->safe_psql("postgres", "CREATE INDEX ON tst USING ivfflat (v);");
+$node_primary->safe_psql("postgres", "CREATE INDEX ON tst USING ivfflat (v vector_l2_ops);");
 
 # Test that queries give same result
 test_index_replay('initial');

test/t/002_ivfflat_vacuum.pl

@@ -7,7 +7,8 @@ use Test::More;
 my $dim = 3;
 
 my @r = ();
-for (1 .. $dim) {
+for (1 .. $dim)
+{
 	my $v = int(rand(1000)) + 1;
 	push(@r, "i % $v");
 }
@@ -24,7 +25,7 @@ $node->safe_psql("postgres", "CREATE TABLE tst (i int4, v vector($dim));");
 $node->safe_psql("postgres",
 	"INSERT INTO tst SELECT i % 10, ARRAY[$array_sql] FROM generate_series(1, 100000) i;"
 );
-$node->safe_psql("postgres", "CREATE INDEX ON tst USING ivfflat (v);");
+$node->safe_psql("postgres", "CREATE INDEX ON tst USING ivfflat (v vector_l2_ops);");
 
 # Get size
 my $size = $node->safe_psql("postgres", "SELECT pg_total_relation_size('tst_v_idx');");

test/t/003_ivfflat_recall.pl

@@ -22,7 +22,8 @@ sub test_recall
 	));
 	like($explain, qr/Index Scan using idx on tst/);
 
-	for my $i (0 .. $#queries) {
+	for my $i (0 .. $#queries)
+	{
 		my $actual = $node->safe_psql("postgres", qq(
 			SET enable_seqscan = off;
 			SET ivfflat.probes = $probes;
@@ -33,8 +34,10 @@ sub test_recall
 
 		my @expected_ids = split("\n", $expected[$i]);
 
-		foreach (@expected_ids) {
-			if (exists($actual_set{$_})) {
+		foreach (@expected_ids)
+		{
+			if (exists($actual_set{$_}))
+			{
 				$correct++;
 			}
 			$total++;
@@ -57,7 +60,8 @@ $node->safe_psql("postgres",
 );
 
 # Generate queries
-for (1..20) {
+for (1 .. 20)
+{
 	my $r1 = rand();
 	my $r2 = rand();
 	my $r3 = rand();
@@ -66,26 +70,21 @@ for (1..20) {
 
 # Check each index type
 my @operators = ("<->", "<#>", "<=>");
+my @opclasses = ("vector_l2_ops", "vector_ip_ops", "vector_cosine_ops");
 
-foreach (@operators) {
-	my $operator = $_;
+for my $i (0 .. $#operators)
+{
+	my $operator = $operators[$i];
+	my $opclass = $opclasses[$i];
 
 	# Get exact results
 	@expected = ();
-	foreach (@queries) {
+	foreach (@queries)
+	{
 		my $res = $node->safe_psql("postgres", "SELECT i FROM tst ORDER BY v $operator '$_' LIMIT $limit;");
 		push(@expected, $res);
 	}
 
-	my $opclass;
-	if ($operator eq "<->") {
-		$opclass = "vector_l2_ops";
-	} elsif ($operator eq "<#>") {
-		$opclass = "vector_ip_ops";
-	} else {
-		$opclass = "vector_cosine_ops";
-	}
-
 	# Build index serially
 	$node->safe_psql("postgres", qq(
 		SET max_parallel_maintenance_workers = 0;
@@ -93,13 +92,14 @@ foreach (@operators) {
 	));
 
 	# Test approximate results
-	if ($operator ne "<#>") {
+	if ($operator ne "<#>")
+	{
 		# TODO fix test
-		test_recall(1, 0.75, $operator);
+		test_recall(1, 0.71, $operator);
 		test_recall(10, 0.95, $operator);
 	}
 	# Account for equal distances
-	test_recall(100, 0.995, $operator);
+	test_recall(100, 0.9925, $operator);
 
 	$node->safe_psql("postgres", "DROP INDEX idx;");
 
@@ -113,13 +113,14 @@ foreach (@operators) {
 	like($stderr, qr/using \d+ parallel workers/);
 
 	# Test approximate results
-	if ($operator ne "<#>") {
+	if ($operator ne "<#>")
+	{
 		# TODO fix test
-		test_recall(1, 0.75, $operator);
+		test_recall(1, 0.71, $operator);
 		test_recall(10, 0.95, $operator);
 	}
 	# Account for equal distances
-	test_recall(100, 0.995, $operator);
+	test_recall(100, 0.9925, $operator);
 
 	$node->safe_psql("postgres", "DROP INDEX idx;");
 }

test/t/004_ivfflat_centers.pl

@@ -20,7 +20,7 @@ sub test_centers
 {
 	my ($lists, $min) = @_;
 
-	my ($ret, $stdout, $stderr) = $node->psql("postgres", "CREATE INDEX ON tst USING ivfflat (v) WITH (lists = $lists);");
+	my ($ret, $stdout, $stderr) = $node->psql("postgres", "CREATE INDEX ON tst USING ivfflat (v vector_l2_ops) WITH (lists = $lists);");
 	is($ret, 0, $stderr);
 }
 

test/t/005_ivfflat_query_recall.pl

@@ -18,24 +18,21 @@ $node->safe_psql("postgres",
 
 # Check each index type
 my @operators = ("<->", "<#>", "<=>");
-foreach (@operators) {
-	my $operator = $_;
+my @opclasses = ("vector_l2_ops", "vector_ip_ops", "vector_cosine_ops");
+
+for my $i (0 .. $#operators)
+{
+	my $operator = $operators[$i];
+	my $opclass = $opclasses[$i];
 
 	# Add index
-	my $opclass;
-	if ($operator eq "<->") {
-		$opclass = "vector_l2_ops";
-	} elsif ($operator eq "<#>") {
-		$opclass = "vector_ip_ops";
-	} else {
-		$opclass = "vector_cosine_ops";
-	}
 	$node->safe_psql("postgres", "CREATE INDEX ON tst USING ivfflat (v $opclass);");
 
 	# Test 100% recall
-	for (1..20) {
-		my $i = int(rand() * 100000);
-		my $query = $node->safe_psql("postgres", "SELECT v FROM tst WHERE i = $i;");
+	for (1 .. 20)
+	{
+		my $id = int(rand() * 100000);
+		my $query = $node->safe_psql("postgres", "SELECT v FROM tst WHERE i = $id;");
 		my $res = $node->safe_psql("postgres", qq(
 			SET enable_seqscan = off;
 			SELECT v FROM tst ORDER BY v <-> '$query' LIMIT 1;

test/t/006_ivfflat_lists.pl

@@ -16,8 +16,8 @@ $node->safe_psql("postgres",
 	"INSERT INTO tst SELECT ARRAY[random(), random(), random()] FROM generate_series(1, 100000) i;"
 );
 
-$node->safe_psql("postgres", "CREATE INDEX lists50 ON tst USING ivfflat (v) WITH (lists = 50);");
-$node->safe_psql("postgres", "CREATE INDEX lists100 ON tst USING ivfflat (v) WITH (lists = 100);");
+$node->safe_psql("postgres", "CREATE INDEX lists50 ON tst USING ivfflat (v vector_l2_ops) WITH (lists = 50);");
+$node->safe_psql("postgres", "CREATE INDEX lists100 ON tst USING ivfflat (v vector_l2_ops) WITH (lists = 100);");
 
 # Test prefers more lists
 my $res = $node->safe_psql("postgres", "EXPLAIN SELECT v FROM tst ORDER BY v <-> '[0.5,0.5,0.5]' LIMIT 10;");
@@ -26,7 +26,7 @@ unlike($res, qr/lists50/);
 
 # Test errors with too much memory
 my ($ret, $stdout, $stderr) = $node->psql("postgres",
-	"CREATE INDEX lists10000 ON tst USING ivfflat (v) WITH (lists = 10000);"
+	"CREATE INDEX lists10000 ON tst USING ivfflat (v vector_l2_ops) WITH (lists = 10000);"
 );
 like($stderr, qr/memory required is/);
 

test/t/007_ivfflat_inserts.pl

@@ -19,7 +19,7 @@ $node->safe_psql("postgres", "CREATE TABLE tst (v vector($dim));");
 $node->safe_psql("postgres",
 	"INSERT INTO tst SELECT ARRAY[$array_sql] FROM generate_series(1, 10000) i;"
 );
-$node->safe_psql("postgres", "CREATE INDEX ON tst USING ivfflat (v);");
+$node->safe_psql("postgres", "CREATE INDEX ON tst USING ivfflat (v vector_l2_ops);");
 
 $node->pgbench(
 	"--no-vacuum --client=5 --transactions=100",
@@ -28,7 +28,7 @@ $node->pgbench(
 	[qr{^$}],
 	"concurrent INSERTs",
 	{
-		"007_inserts" => "INSERT INTO tst SELECT ARRAY[$array_sql] FROM generate_series(1, 10) i;"
+		"007_ivfflat_inserts" => "INSERT INTO tst SELECT ARRAY[$array_sql] FROM generate_series(1, 10) i;"
 	}
 );
 

test/t/008_aggregates.pl

@@ -30,7 +30,8 @@ sub test_aggregate
 
 	# Test matches real for avg
 	# Cannot test sum since sum(real) varies between calls
-	if ($agg eq 'avg') {
+	if ($agg eq 'avg')
+	{
 		my $r1 = $node->safe_psql("postgres", "SELECT $agg(r1)::float4 FROM tst;");
 		my $r2 = $node->safe_psql("postgres", "SELECT $agg(r2)::float4 FROM tst;");
 		my $r3 = $node->safe_psql("postgres", "SELECT $agg(r3)::float4 FROM tst;");

test/t/010_hnsw_wal.pl

@@ -26,7 +26,8 @@ sub test_index_replay
 	  or die "Timed out while waiting for replica 1 to catch up";
 
 	my @r = ();
-	for (1 .. $dim) {
+	for (1 .. $dim)
+	{
 		push(@r, rand());
 	}
 	my $sql = join(",", @r);
@@ -52,11 +53,13 @@ my $array_sql = join(",", ('random()') x $dim);
 # Initialize primary node
 $node_primary = get_new_node('primary');
 $node_primary->init(allows_streaming => 1);
-if ($dim > 32) {
+if ($dim > 32)
+{
 	# TODO use wal_keep_segments for Postgres < 13
 	$node_primary->append_conf('postgresql.conf', qq(wal_keep_size = 1GB));
 }
-if ($dim > 1500) {
+if ($dim > 1500)
+{
 	$node_primary->append_conf('postgresql.conf', qq(maintenance_work_mem = 128MB));
 }
 $node_primary->start;
@@ -67,8 +70,7 @@ $node_primary->backup($backup_name);
 
 # Create streaming replica linking to primary
 $node_replica = get_new_node('replica');
-$node_replica->init_from_backup($node_primary, $backup_name,
-	has_streaming => 1);
+$node_replica->init_from_backup($node_primary, $backup_name, has_streaming => 1);
 $node_replica->start;
 
 # Create hnsw index on primary

test/t/011_hnsw_vacuum.pl

@@ -7,7 +7,8 @@ use Test::More;
 my $dim = 3;
 
 my @r = ();
-for (1 .. $dim) {
+for (1 .. $dim)
+{
 	my $v = int(rand(1000)) + 1;
 	push(@r, "i % $v");
 }
@@ -22,7 +23,7 @@ $node->start;
 $node->safe_psql("postgres", "CREATE EXTENSION vector;");
 $node->safe_psql("postgres", "CREATE TABLE tst (i int4, v vector($dim));");
 $node->safe_psql("postgres",
-	"INSERT INTO tst SELECT i % 10, ARRAY[$array_sql] FROM generate_series(1, 10000) i;"
+	"INSERT INTO tst SELECT i, ARRAY[$array_sql] FROM generate_series(1, 10000) i;"
 );
 $node->safe_psql("postgres", "CREATE INDEX ON tst USING hnsw (v vector_l2_ops);");
 
@@ -33,11 +34,21 @@ my $size = $node->safe_psql("postgres", "SELECT pg_total_relation_size('tst_v_id
 $node->safe_psql("postgres", "DELETE FROM tst;");
 $node->safe_psql("postgres", "VACUUM tst;");
 $node->safe_psql("postgres",
-	"INSERT INTO tst SELECT i % 10, ARRAY[$array_sql] FROM generate_series(1, 10000) i;"
+	"INSERT INTO tst SELECT i, ARRAY[$array_sql] FROM generate_series(1, 10000) i;"
 );
 
 # Check size
+# May increase some due to different levels
 my $new_size = $node->safe_psql("postgres", "SELECT pg_total_relation_size('tst_v_idx');");
-cmp_ok($new_size, "<=", $size * 1.01, "size does not increase too much");
+cmp_ok($new_size, "<=", $size * 1.02, "size does not increase too much");
+
+# Delete all but one
+$node->safe_psql("postgres", "DELETE FROM tst WHERE i != 123;");
+$node->safe_psql("postgres", "VACUUM tst;");
+my $res = $node->safe_psql("postgres", qq(
+	SET enable_seqscan = off;
+	SELECT i FROM tst ORDER BY v <-> '[0,0,0]' LIMIT 10;
+));
+is($res, 123);
 
 done_testing();

test/t/012_hnsw_build_recall.pl

@@ -21,7 +21,8 @@ sub test_recall
 	));
 	like($explain, qr/Index Scan/);
 
-	for my $i (0 .. $#queries) {
+	for my $i (0 .. $#queries)
+	{
 		my $actual = $node->safe_psql("postgres", qq(
 			SET enable_seqscan = off;
 			SELECT i FROM tst ORDER BY v $operator '$queries[$i]' LIMIT $limit;
@@ -31,8 +32,10 @@ sub test_recall
 
 		my @expected_ids = split("\n", $expected[$i]);
 
-		foreach (@expected_ids) {
-			if (exists($actual_set{$_})) {
+		foreach (@expected_ids)
+		{
+			if (exists($actual_set{$_}))
+			{
 				$correct++;
 			}
 			$total++;
@@ -55,7 +58,8 @@ $node->safe_psql("postgres",
 );
 
 # Generate queries
-for (1..20) {
+for (1 .. 20)
+{
 	my $r1 = rand();
 	my $r2 = rand();
 	my $r3 = rand();
@@ -64,33 +68,46 @@ for (1..20) {
 
 # Check each index type
 my @operators = ("<->", "<#>", "<=>");
+my @opclasses = ("vector_l2_ops", "vector_ip_ops", "vector_cosine_ops");
 
-foreach (@operators) {
-	my $operator = $_;
+for my $i (0 .. $#operators)
+{
+	my $operator = $operators[$i];
+	my $opclass = $opclasses[$i];
 
 	# Get exact results
 	@expected = ();
-	foreach (@queries) {
+	foreach (@queries)
+	{
 		my $res = $node->safe_psql("postgres", "SELECT i FROM tst ORDER BY v $operator '$_' LIMIT $limit;");
 		push(@expected, $res);
 	}
 
-	# Add index
-	my $opclass;
-	if ($operator eq "<->") {
-		$opclass = "vector_l2_ops";
-	} elsif ($operator eq "<#>") {
-		$opclass = "vector_ip_ops";
-	} else {
-		$opclass = "vector_cosine_ops";
-	}
-	$node->safe_psql("postgres", "CREATE INDEX ON tst USING hnsw (v $opclass);");
+	# Build index serially
+	$node->safe_psql("postgres", qq(
+		SET max_parallel_maintenance_workers = 0;
+		CREATE INDEX idx ON tst USING hnsw (v $opclass);
+	));
 
-	if ($operator eq "<#>") {
-		test_recall(0.80, $operator);
-	} else {
-		test_recall(0.99, $operator);
-	}
+	# Test approximate results
+	my $min = $operator eq "<#>" ? 0.80 : 0.99;
+	test_recall($min, $operator);
+
+	$node->safe_psql("postgres", "DROP INDEX idx;");
+
+	# Build index in parallel
+	my ($ret, $stdout, $stderr) = $node->psql("postgres", qq(
+		SET client_min_messages = DEBUG;
+		SET min_parallel_table_scan_size = 1;
+		CREATE INDEX idx ON tst USING hnsw (v $opclass);
+	));
+	is($ret, 0, $stderr);
+	like($stderr, qr/using \d+ parallel workers/);
+
+	# Test approximate results
+	test_recall($min, $operator);
+
+	$node->safe_psql("postgres", "DROP INDEX idx;");
 }
 
 done_testing();

test/t/013_hnsw_insert_recall.pl

@@ -21,7 +21,8 @@ sub test_recall
 	));
 	like($explain, qr/Index Scan/);
 
-	for my $i (0 .. $#queries) {
+	for my $i (0 .. $#queries)
+	{
 		my $actual = $node->safe_psql("postgres", qq(
 			SET enable_seqscan = off;
 			SELECT i FROM tst ORDER BY v $operator '$queries[$i]' LIMIT $limit;
@@ -31,8 +32,10 @@ sub test_recall
 
 		my @expected_ids = split("\n", $expected[$i]);
 
-		foreach (@expected_ids) {
-			if (exists($actual_set{$_})) {
+		foreach (@expected_ids)
+		{
+			if (exists($actual_set{$_}))
+			{
 				$correct++;
 			}
 			$total++;
@@ -49,10 +52,11 @@ $node->start;
 
 # Create table
 $node->safe_psql("postgres", "CREATE EXTENSION vector;");
-$node->safe_psql("postgres", "CREATE TABLE tst (i int4, v vector(3));");
+$node->safe_psql("postgres", "CREATE TABLE tst (i serial, v vector(3));");
 
 # Generate queries
-for (1..20) {
+for (1 .. 20)
+{
 	my $r1 = rand();
 	my $r2 = rand();
 	my $r3 = rand();
@@ -61,28 +65,32 @@ for (1..20) {
 
 # Check each index type
 my @operators = ("<->", "<#>", "<=>");
+my @opclasses = ("vector_l2_ops", "vector_ip_ops", "vector_cosine_ops");
 
-foreach (@operators) {
-	my $operator = $_;
+for my $i (0 .. $#operators)
+{
+	my $operator = $operators[$i];
+	my $opclass = $opclasses[$i];
 
 	# Add index
-	my $opclass;
-	if ($operator eq "<->") {
-		$opclass = "vector_l2_ops";
-	} elsif ($operator eq "<#>") {
-		$opclass = "vector_ip_ops";
-	} else {
-		$opclass = "vector_cosine_ops";
-	}
 	$node->safe_psql("postgres", "CREATE INDEX idx ON tst USING hnsw (v $opclass);");
 
-	$node->safe_psql("postgres",
-		"INSERT INTO tst SELECT i, ARRAY[random(), random(), random()] FROM generate_series(1, 10000) i;"
+	# Use concurrent inserts
+	$node->pgbench(
+		"--no-vacuum --client=10 --transactions=1000",
+		0,
+		[qr{actually processed}],
+		[qr{^$}],
+		"concurrent INSERTs",
+		{
+			"013_hnsw_insert_recall_$opclass" => "INSERT INTO tst (v) VALUES (ARRAY[random(), random(), random()]);"
+		}
 	);
 
 	# Get exact results
 	@expected = ();
-	foreach (@queries) {
+	foreach (@queries)
+	{
 		my $res = $node->safe_psql("postgres", qq(
 			SET enable_indexscan = off;
 			SELECT i FROM tst ORDER BY v $operator '$_' LIMIT $limit;
@@ -90,11 +98,8 @@ foreach (@operators) {
 		push(@expected, $res);
 	}
 
-	if ($operator eq "<#>") {
-		test_recall(0.80, $operator);
-	} else {
-		test_recall(0.99, $operator);
-	}
+	my $min = $operator eq "<#>" ? 0.80 : 0.99;
+	test_recall($min, $operator);
 
 	$node->safe_psql("postgres", "DROP INDEX idx;");
 	$node->safe_psql("postgres", "TRUNCATE tst;");

test/t/014_hnsw_inserts.pl

@@ -17,22 +17,8 @@ $node->start;
 # Create table and index
 $node->safe_psql("postgres", "CREATE EXTENSION vector;");
 $node->safe_psql("postgres", "CREATE TABLE tst (v vector($dim));");
-$node->safe_psql("postgres",
-	"INSERT INTO tst SELECT ARRAY[$array_sql] FROM generate_series(1, 100) i;"
-);
 $node->safe_psql("postgres", "CREATE INDEX ON tst USING hnsw (v vector_l2_ops);");
 
-$node->pgbench(
-	"--no-vacuum --client=5 --transactions=100",
-	0,
-	[qr{actually processed}],
-	[qr{^$}],
-	"concurrent INSERTs",
-	{
-		"007_inserts" => "INSERT INTO tst SELECT ARRAY[$array_sql] FROM generate_series(1, 10) i;"
-	}
-);
-
 sub idx_scan
 {
 	# Stats do not update instantaneously
@@ -41,18 +27,48 @@ sub idx_scan
 	$node->safe_psql("postgres", "SELECT idx_scan FROM pg_stat_user_indexes WHERE indexrelid = 'tst_v_idx'::regclass;");
 }
 
-my $expected = 100 + 5 * 100 * 10;
+for my $i (1 .. 20)
+{
+	$node->pgbench(
+		"--no-vacuum --client=10 --transactions=1",
+		0,
+		[qr{actually processed}],
+		[qr{^$}],
+		"concurrent INSERTs",
+		{
+			"014_hnsw_inserts_$i" => "INSERT INTO tst VALUES (ARRAY[$array_sql]);"
+		}
+	);
 
-my $count = $node->safe_psql("postgres", "SELECT COUNT(*) FROM tst;");
-is($count, $expected);
-is(idx_scan(), 0);
+	my $count = $node->safe_psql("postgres", qq(
+		SET enable_seqscan = off;
+		SELECT COUNT(*) FROM (SELECT v FROM tst ORDER BY v <-> (SELECT v FROM tst LIMIT 1)) t;
+	));
+	is($count, 10);
 
-$count = $node->safe_psql("postgres", qq(
+	$node->safe_psql("postgres", "TRUNCATE tst;");
+}
+
+$node->pgbench(
+	"--no-vacuum --client=20 --transactions=5",
+	0,
+	[qr{actually processed}],
+	[qr{^$}],
+	"concurrent INSERTs",
+	{
+		"014_hnsw_inserts" => "INSERT INTO tst SELECT ARRAY[$array_sql] FROM generate_series(1, 10) i;"
+	}
+);
+
+my $count = $node->safe_psql("postgres", qq(
 	SET enable_seqscan = off;
-	SET hnsw.ef_search = 400;
+	SET hnsw.ef_search = 1000;
 	SELECT COUNT(*) FROM (SELECT v FROM tst ORDER BY v <-> (SELECT v FROM tst LIMIT 1)) t;
 ));
-is($count, 400);
-is(idx_scan(), 1);
+# Elements may lose all incoming connections with the HNSW algorithm
+# Vacuuming can fix this if one of the elements neighbors is deleted
+cmp_ok($count, ">=", 997);
+
+is(idx_scan(), 21);
 
 done_testing();

test/t/015_hnsw_duplicates.pl

@@ -0,0 +1,58 @@
+use strict;
+use warnings;
+use PostgresNode;
+use TestLib;
+use Test::More;
+
+# Initialize node
+my $node = get_new_node('node');
+$node->init;
+$node->start;
+
+# Create table
+$node->safe_psql("postgres", "CREATE EXTENSION vector;");
+$node->safe_psql("postgres", "CREATE TABLE tst (v vector(3));");
+
+sub insert_vectors
+{
+	for my $i (1 .. 20)
+	{
+		$node->safe_psql("postgres", "INSERT INTO tst VALUES ('[1,1,1]');");
+	}
+}
+
+sub test_duplicates
+{
+	my $res = $node->safe_psql("postgres", qq(
+		SET enable_seqscan = off;
+		SET hnsw.ef_search = 1;
+		SELECT COUNT(*) FROM (SELECT * FROM tst ORDER BY v <-> '[1,1,1]') t;
+	));
+	is($res, 10);
+}
+
+# Test duplicates with build
+insert_vectors();
+$node->safe_psql("postgres", "CREATE INDEX idx ON tst USING hnsw (v vector_l2_ops);");
+test_duplicates();
+
+# Reset
+$node->safe_psql("postgres", "TRUNCATE tst;");
+
+# Test duplicates with inserts
+insert_vectors();
+test_duplicates();
+
+# Test fallback path for inserts
+$node->pgbench(
+	"--no-vacuum --client=5 --transactions=100",
+	0,
+	[qr{actually processed}],
+	[qr{^$}],
+	"concurrent INSERTs",
+	{
+		"015_hnsw_duplicates" => "INSERT INTO tst VALUES ('[1,1,1]');"
+	}
+);
+
+done_testing();

test/t/016_hnsw_vacuum_recall.pl

@@ -0,0 +1,97 @@
+use strict;
+use warnings;
+use PostgresNode;
+use TestLib;
+use Test::More;
+
+my $node;
+my @queries = ();
+my @expected;
+my $limit = 20;
+
+sub test_recall
+{
+	my ($min, $ef_search, $test_name) = @_;
+	my $correct = 0;
+	my $total = 0;
+
+	my $explain = $node->safe_psql("postgres", qq(
+		SET enable_seqscan = off;
+		SET hnsw.ef_search = $ef_search;
+		EXPLAIN ANALYZE SELECT i FROM tst ORDER BY v <-> '$queries[0]' LIMIT $limit;
+	));
+	like($explain, qr/Index Scan/);
+
+	for my $i (0 .. $#queries)
+	{
+		my $actual = $node->safe_psql("postgres", qq(
+			SET enable_seqscan = off;
+			SET hnsw.ef_search = $ef_search;
+			SELECT i FROM tst ORDER BY v <-> '$queries[$i]' LIMIT $limit;
+		));
+		my @actual_ids = split("\n", $actual);
+		my %actual_set = map { $_ => 1 } @actual_ids;
+
+		my @expected_ids = split("\n", $expected[$i]);
+
+		foreach (@expected_ids)
+		{
+			if (exists($actual_set{$_}))
+			{
+				$correct++;
+			}
+			$total++;
+		}
+	}
+
+	cmp_ok($correct / $total, ">=", $min, $test_name);
+}
+
+# Initialize node
+$node = get_new_node('node');
+$node->init;
+$node->start;
+
+# Create table
+$node->safe_psql("postgres", "CREATE EXTENSION vector;");
+$node->safe_psql("postgres", "CREATE TABLE tst (i int4, v vector(3));");
+$node->safe_psql("postgres", "ALTER TABLE tst SET (autovacuum_enabled = false);");
+$node->safe_psql("postgres",
+	"INSERT INTO tst SELECT i, ARRAY[random(), random(), random()] FROM generate_series(1, 10000) i;"
+);
+
+# Add index
+$node->safe_psql("postgres", "CREATE INDEX ON tst USING hnsw (v vector_l2_ops) WITH (m = 4, ef_construction = 8);");
+
+# Delete data
+$node->safe_psql("postgres", "DELETE FROM tst WHERE i > 2500;");
+
+# Generate queries
+for (1 .. 20)
+{
+	my $r1 = rand();
+	my $r2 = rand();
+	my $r3 = rand();
+	push(@queries, "[$r1,$r2,$r3]");
+}
+
+# Get exact results
+@expected = ();
+foreach (@queries)
+{
+	my $res = $node->safe_psql("postgres", qq(
+		SET enable_indexscan = off;
+		SELECT i FROM tst ORDER BY v <-> '$_' LIMIT $limit;
+	));
+	push(@expected, $res);
+}
+
+test_recall(0.20, $limit, "before vacuum");
+test_recall(0.95, 100, "before vacuum");
+
+# TODO test concurrent inserts with vacuum
+$node->safe_psql("postgres", "VACUUM tst;");
+
+test_recall(0.95, $limit, "after vacuum");
+
+done_testing();

vector.control

@@ -1,4 +1,4 @@
 comment = 'vector data type and ivfflat access method'
-default_version = '0.4.4'
+default_version = '0.5.0'
 module_pathname = '$libdir/vector'
 relocatable = true