Compare commits

..

63 Commits

Author SHA1 Message Date
Andrew Kane
63982fef41 Include vectors with zero norm in indexes for cosine distance 2023-11-18 13:40:22 -08:00
Andrew Kane
5b12ae8225 Added note about ef_construction [skip ci] 2023-11-16 18:53:48 -08:00
Japin Li
4549e8aeb1 Fix coredump about HnswFreeElement() (#357)
The HnswInitElement() allocate an element with not initialize value
filed, which may has garbage that lead HnswFreeElement() free an
invalid pointer.
2023-11-15 16:19:59 -08:00
Andrew Kane
3263b350f5 Updated HnswLoadElementFromTuple to be less vector-specific 2023-11-11 21:14:12 -08:00
Andrew Kane
dfee5d4045 Added support for on-disk parallel index builds for HNSW 2023-11-11 19:29:45 -08:00
Andrew Kane
69a2ce0d43 Use datumIsEqual to compare 2023-11-10 10:46:48 -08:00
Andrew Kane
c5e8c46b80 Switched from VECTOR_SIZE to VARSIZE_ANY [skip ci] 2023-11-09 19:41:38 -08:00
Andrew Kane
94f7304ccd Keep vector for now to be overly cautious about packing [skip ci] 2023-11-09 18:43:55 -08:00
Andrew Kane
d078db3d25 Switched HnswElementTuple to generic data and zero full section 2023-11-09 18:28:25 -08:00
Andrew Kane
fbb904ae2f Use pointer for VARSIZE_ANY 2023-11-09 17:50:28 -08:00
Andrew Kane
3cf6f62900 Switched to datum for HnswElement 2023-11-09 17:35:39 -08:00
Andrew Kane
2a69e22ca4 Switched from VECTOR_SIZE to VARSIZE_ANY where possible (less vector-specific) 2023-11-09 17:16:43 -08:00
Andrew Kane
84e073888c Removed vector-specific code from HNSW_ELEMENT_TUPLE_SIZE [skip ci] 2023-11-09 16:57:01 -08:00
Andrew Kane
81a62d55d1 Switched from HNSW_ELEMENT_TUPLE_SIZE to ItemIdGetLength where possible (less vector-specific) 2023-11-09 16:32:00 -08:00
Andrew Kane
3f3463bde5 Improved memory calculation for HNSW and removed vector-specific code 2023-11-09 16:21:26 -08:00
Andrew Kane
a01a72d812 Updated comment [skip ci] 2023-11-05 08:42:06 -08:00
Andrew Kane
0c2fc18a80 Updated comment [skip ci] 2023-11-05 08:40:21 -08:00
Andrew Kane
e860042d3c Improved variable name [skip ci] 2023-11-05 08:35:54 -08:00
Andrew Kane
5986862bd2 Added note about check constraint [skip ci] 2023-11-04 15:01:37 -07:00
Andrew Kane
5d24f5d09a Improved header installation on Windows 2023-11-04 11:16:40 -07:00
Andrew Kane
7c43b0d8ee Updated example [skip ci] 2023-11-03 23:54:50 -07:00
Andrew Kane
7be40036f4 Updated readme [skip ci] 2023-11-03 23:46:23 -07:00
Andrew Kane
9b5a1a69db Updated readme [skip ci] 2023-11-03 23:43:47 -07:00
Andrew Kane
04b96506f5 Added info on storing vectors with more precision [skip ci] 2023-11-03 20:14:28 -07:00
Andrew Kane
35cd7b63cb Updated readme [skip ci] 2023-11-03 17:02:30 -07:00
Andrew Kane
b5416d6f10 Updated readme [skip ci] 2023-11-03 16:48:57 -07:00
Andrew Kane
f361bf2704 Improved docs on indexing vectors with different dimensions [skip ci] 2023-11-03 16:42:14 -07:00
Andrew Kane
3d8c1921aa Improved upgrading docs - #339 [skip ci] 2023-11-03 16:15:06 -07:00
Andrew Kane
154207bc17 Added info on columns with different dimensions [skip ci] 2023-11-03 16:02:00 -07:00
Andrew Kane
8e507f3bf5 Free remaining allocation from deconstruct_array - #332 2023-11-02 21:20:21 -07:00
Andrew Kane
e115773a55 Removed unneeded allocation 2023-11-02 21:16:06 -07:00
Andrew Kane
9333bef046 Added link to setup-pgvector [skip ci] 2023-11-02 13:22:19 -07:00
Andrew Kane
4851e47d9f Added Reciprocal Rank Fusion example to readme [skip ci] 2023-11-01 13:20:49 -07:00
Andrew Kane
12aecfb4f5 Added Nim and Zig to readme [skip ci] 2023-10-31 02:26:18 -07:00
Andrew Kane
800697fb14 Updated column alias [skip ci] 2023-10-29 16:47:55 -07:00
Andrew Kane
de1f2b09dd Improved indexing progress queries [skip ci] 2023-10-29 16:41:39 -07:00
Andrew Kane
bcccb7f5a5 Improved docs for indexing progress - closes #320 and closes #321 [skip ci] 2023-10-29 16:13:12 -07:00
Andrew Kane
bec3d30d68 Added TypeScript to readme [skip ci] 2023-10-29 12:49:01 -07:00
Andrew Kane
588de60445 Added Groovy to readme [skip ci] 2023-10-29 12:39:53 -07:00
Andrew Kane
c599f92b52 Updated readme [skip ci] 2023-10-27 13:22:37 -07:00
Andrew Kane
2a17b335da Added Kotlin to readme [skip ci] 2023-10-26 12:25:58 -07:00
Andrew Kane
6ede6ac301 Added link to pgvector-c [skip ci] 2023-10-26 00:30:06 -07:00
Andrew Kane
3f49b95f01 Added Postgres 17 to CI [skip ci] 2023-10-19 00:37:24 -07:00
Andrew Kane
ef1bea7163 Updated checkout action [skip ci] 2023-10-19 00:36:53 -07:00
Andrew Kane
e630efd195 Version bump to 0.5.1 [skip ci] 2023-10-10 17:40:57 -07:00
Andrew Kane
b5b912906b Added check for MVCC-compliant snapshot and removed marking tuples as dead for IVFFlat index scans - closes #260 2023-10-10 17:28:48 -07:00
Andrew Kane
4b5db94307 Disable closer caching for new elements for now 2023-10-06 14:27:09 -07:00
Andrew Kane
65e70326b8 Updated comment [skip ci] 2023-10-06 14:07:35 -07:00
Andrew Kane
71641ed84e Updated comment [skip ci] 2023-10-06 13:58:07 -07:00
Andrew Kane
f3dba25036 Added comment [skip ci] 2023-10-06 13:56:25 -07:00
Andrew Kane
5588ba6410 Improved variable name [skip ci] 2023-10-06 13:46:19 -07:00
Andrew Kane
ec9fac5456 Improved closerSet logic 2023-10-06 13:39:55 -07:00
Andrew Kane
8085d3e538 Moved sorting logic into SelectNeighbors 2023-10-06 12:56:15 -07:00
Andrew Kane
cae162ffc6 Ensure order is deterministic for SelectNeighbors closer caching 2023-10-06 12:26:53 -07:00
Andrew Kane
62482e3760 Use e for consistency 2023-10-05 16:15:13 -07:00
Heikki Linnakangas
c81302b835 Improve HNSW index build performance more (#295)
This takes the approach from commit a713e2acaa further. Once we have
remove a candidate from the "closer" set, we still don't need to
recalculate everything that follows. Any candidates that were in the
closer set before still only need to be compared with any new
candidates that we have added.
2023-10-05 16:04:50 -07:00
Andrew Kane
a713e2acaa Improved performance of HNSW index builds - closes #292
Co-authored-by: Heikki Linnakangas <heikki.linnakangas@iki.fi>
2023-10-05 13:21:26 -07:00
Andrew Kane
6e1312ddbe DRY max size 2023-10-04 21:43:34 -07:00
Andrew Kane
4ef5bca275 Use BLCKSZ for consistency 2023-10-04 21:37:55 -07:00
Xiaoran Wang
1ecf6ada76 Include ItemIdData when computing the maxSize for the data in a page (#274)
As the data is aligned, for hnsw, the combined size won't be in the range
(8156 (maxSize exlucding `ItemIdData`), 8160]. So even if the
ItemIdData is not included in the maxSize, it works well now, but I
think it's better to make it correct.
2023-10-04 21:26:46 -07:00
Andrew Kane
564a3d45fc Added check for MVCC-compliant snapshot for HNSW index scans - closes #281 2023-10-04 20:14:50 -07:00
Andrew Kane
8d7abb6590 Revert "Fixed locking for index scans for HNSW - #256"
This reverts commit d032726976.
2023-09-26 23:00:14 -07:00
jeff-davis
b247b688a8 No need to MarkBufferDirty(); GenericXLogFinish() does that. (#265) 2023-09-15 13:14:10 -07:00
30 changed files with 807 additions and 507 deletions

View File

@@ -8,6 +8,8 @@ jobs:
fail-fast: false
matrix:
include:
- postgres: 17
os: ubuntu-22.04
- postgres: 16
os: ubuntu-22.04
- postgres: 15
@@ -21,7 +23,7 @@ jobs:
- postgres: 11
os: ubuntu-20.04
steps:
- uses: actions/checkout@v3
- uses: actions/checkout@v4
- uses: ankane/setup-postgres@v1
with:
postgres-version: ${{ matrix.postgres }}
@@ -43,7 +45,7 @@ jobs:
runs-on: macos-latest
if: ${{ !startsWith(github.ref_name, 'windows') }}
steps:
- uses: actions/checkout@v3
- uses: actions/checkout@v4
- uses: ankane/setup-postgres@v1
with:
postgres-version: 14
@@ -65,7 +67,7 @@ jobs:
runs-on: windows-latest
if: ${{ !startsWith(github.ref_name, 'mac') }}
steps:
- uses: actions/checkout@v3
- uses: actions/checkout@v4
- uses: ankane/setup-postgres@v1
with:
postgres-version: 14

View File

@@ -1,7 +1,11 @@
## 0.5.1 (unreleased)
## 0.5.2 (unreleased)
- Improved performance of index scans for IVFFlat after updates and deletes
- Fixed locking for index scans for HNSW
- Added support for on-disk parallel index builds for HNSW
## 0.5.1 (2023-10-10)
- Improved performance of HNSW index builds
- Added check for MVCC-compliant snapshot for index scans
## 0.5.0 (2023-08-28)

View File

@@ -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.5.0",
"version": "0.5.1",
"maintainer": [
"Andrew Kane <andrew@ankane.org>"
],
@@ -20,7 +20,7 @@
"vector": {
"file": "sql/vector.sql",
"docfile": "README.md",
"version": "0.5.0",
"version": "0.5.1",
"abstract": "Open-source vector similarity search for Postgres"
}
},

View File

@@ -1,9 +1,9 @@
EXTENSION = vector
EXTVERSION = 0.5.0
EXTVERSION = 0.5.1
MODULE_big = vector
DATA = $(wildcard sql/*--*.sql)
OBJS = src/float4.o 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
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)

View File

@@ -1,7 +1,7 @@
EXTENSION = vector
EXTVERSION = 0.5.0
EXTVERSION = 0.5.1
OBJS = src\float4.obj 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
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
@@ -56,7 +56,7 @@ install:
copy $(EXTENSION).control "$(SHAREDIR)\extension"
copy sql\$(EXTENSION)--*.sql "$(SHAREDIR)\extension"
mkdir "$(INCLUDEDIR_SERVER)\extension\$(EXTENSION)"
copy $(HEADERS) "$(INCLUDEDIR_SERVER)\extension\$(EXTENSION)"
for %f in ($(HEADERS)) do copy %f "$(INCLUDEDIR_SERVER)\extension\$(EXTENSION)"
installcheck:
"$(BINDIR)\pg_regress" --bindir="$(BINDIR)" $(REGRESS_OPTS) $(REGRESS)

103
README.md
View File

@@ -18,7 +18,7 @@ Compile and install the extension (supports Postgres 11+)
```sh
cd /tmp
git clone --branch v0.5.0 https://github.com/pgvector/pgvector.git
git clone --branch v0.5.1 https://github.com/pgvector/pgvector.git
cd pgvector
make
make install # may need sudo
@@ -26,7 +26,7 @@ make install # may need sudo
See the [installation notes](#installation-notes) if you run into issues
You can also install it with [Docker](#docker), [Homebrew](#homebrew), [PGXN](#pgxn), [APT](#apt), [Yum](#yum), or [conda-forge](#conda-forge), and it comes preinstalled with [Postgres.app](#postgresapp) and many [hosted providers](#hosted-postgres)
You can also install it with [Docker](#docker), [Homebrew](#homebrew), [PGXN](#pgxn), [APT](#apt), [Yum](#yum), or [conda-forge](#conda-forge), and it comes preinstalled with [Postgres.app](#postgresapp) and many [hosted providers](#hosted-postgres). There are also instructions for [GitHub Actions](https://github.com/pgvector/setup-pgvector).
## Getting Started
@@ -215,6 +215,23 @@ SELECT ...
COMMIT;
```
### Indexing Progress
Check [indexing progress](https://www.postgresql.org/docs/current/progress-reporting.html#CREATE-INDEX-PROGRESS-REPORTING) with Postgres 12+
```sql
SELECT phase, round(100.0 * tuples_done / nullif(tuples_total, 0), 1) AS "%" FROM pg_stat_progress_create_index;
```
The phases for IVFFlat are:
1. `initializing`
2. `performing k-means`
3. `assigning tuples`
4. `loading tuples`
Note: `%` is only populated during the `loading tuples` phase
## 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). Theres no training step like IVFFlat, so the index can be created without any data in the table.
@@ -252,6 +269,8 @@ Specify HNSW parameters
CREATE INDEX ON items USING hnsw (embedding vector_l2_ops) WITH (m = 16, ef_construction = 64);
```
A higher value of `ef_construction` provides better recall at the cost of index build time / insert speed.
### Query Options
Specify the size of the dynamic candidate list for search (40 by default)
@@ -271,22 +290,18 @@ SELECT ...
COMMIT;
```
## Indexing Progress
### Indexing Progress
Check [indexing progress](https://www.postgresql.org/docs/current/progress-reporting.html#CREATE-INDEX-PROGRESS-REPORTING) with Postgres 12+
```sql
SELECT phase, tuples_done, tuples_total FROM pg_stat_progress_create_index;
SELECT phase, round(100.0 * blocks_done / nullif(blocks_total, 0), 1) AS "%" FROM pg_stat_progress_create_index;
```
The phases are:
The phases for HNSW are:
1. `initializing`
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
2. `loading tuples`
## Filtering
@@ -317,13 +332,15 @@ CREATE TABLE items (embedding vector(3), category_id int) PARTITION BY LIST(cate
## Hybrid Search
Use together with Postgres [full-text search](https://www.postgresql.org/docs/current/textsearch-intro.html) for hybrid search ([Python example](https://github.com/pgvector/pgvector-python/blob/master/examples/hybrid_search.py)).
Use together with Postgres [full-text search](https://www.postgresql.org/docs/current/textsearch-intro.html) for hybrid search.
```sql
SELECT id, content FROM items, plainto_tsquery('hello search') query
WHERE textsearch @@ query ORDER BY ts_rank_cd(textsearch, query) DESC LIMIT 5;
```
You can use [Reciprocal Rank Fusion](https://github.com/pgvector/pgvector-python/blob/master/examples/hybrid_search_rrf.py) or a [cross-encoder](https://github.com/pgvector/pgvector-python/blob/master/examples/hybrid_search.py) to combine results.
## Performance
Use `EXPLAIN ANALYZE` to debug performance.
@@ -360,6 +377,7 @@ Use pgvector from any language with a Postgres client. You can even generate and
Language | Libraries / Examples
--- | ---
C | [pgvector-c](https://github.com/pgvector/pgvector-c)
C++ | [pgvector-cpp](https://github.com/pgvector/pgvector-cpp)
C# | [pgvector-dotnet](https://github.com/pgvector/pgvector-dotnet)
Crystal | [pgvector-crystal](https://github.com/pgvector/pgvector-crystal)
@@ -367,10 +385,11 @@ Dart | [pgvector-dart](https://github.com/pgvector/pgvector-dart)
Elixir | [pgvector-elixir](https://github.com/pgvector/pgvector-elixir)
Go | [pgvector-go](https://github.com/pgvector/pgvector-go)
Haskell | [pgvector-haskell](https://github.com/pgvector/pgvector-haskell)
Java, Scala | [pgvector-java](https://github.com/pgvector/pgvector-java)
Java, Kotlin, Groovy, Scala | [pgvector-java](https://github.com/pgvector/pgvector-java)
JavaScript, TypeScript | [pgvector-node](https://github.com/pgvector/pgvector-node)
Julia | [pgvector-julia](https://github.com/pgvector/pgvector-julia)
Lua | [pgvector-lua](https://github.com/pgvector/pgvector-lua)
Node.js | [pgvector-node](https://github.com/pgvector/pgvector-node)
Nim | [pgvector-nim](https://github.com/pgvector/pgvector-nim)
Perl | [pgvector-perl](https://github.com/pgvector/pgvector-perl)
PHP | [pgvector-php](https://github.com/pgvector/pgvector-php)
Python | [pgvector-python](https://github.com/pgvector/pgvector-python)
@@ -378,6 +397,7 @@ R | [pgvector-r](https://github.com/pgvector/pgvector-r)
Ruby | [pgvector-ruby](https://github.com/pgvector/pgvector-ruby), [Neighbor](https://github.com/ankane/neighbor)
Rust | [pgvector-rust](https://github.com/pgvector/pgvector-rust)
Swift | [pgvector-swift](https://github.com/pgvector/pgvector-swift)
Zig | [pgvector-zig](https://github.com/pgvector/pgvector-zig)
## Frequently Asked Questions
@@ -393,6 +413,55 @@ Yes, pgvector uses the write-ahead log (WAL), which allows for replication and p
Youll need to use [dimensionality reduction](https://en.wikipedia.org/wiki/Dimensionality_reduction) at the moment.
#### Can I store vectors with different dimensions in the same column?
You can use `vector` as the type (instead of `vector(3)`).
```sql
CREATE TABLE embeddings (model_id bigint, item_id bigint, embedding vector, PRIMARY KEY (model_id, item_id));
```
However, you can only create indexes on rows with the same number of dimensions (using [expression](https://www.postgresql.org/docs/current/indexes-expressional.html) and [partial](https://www.postgresql.org/docs/current/indexes-partial.html) indexing):
```sql
CREATE INDEX ON embeddings USING hnsw ((embedding::vector(3)) vector_l2_ops) WHERE (model_id = 123);
```
and query with:
```sql
SELECT * FROM embeddings WHERE model_id = 123 ORDER BY embedding::vector(3) <-> '[3,1,2]' LIMIT 5;
```
#### Can I store vectors with more precision?
You can use the `double precision[]` or `numeric[]` type to store vectors with more precision.
```sql
CREATE TABLE items (id bigserial PRIMARY KEY, embedding double precision[]);
-- use {} instead of [] for Postgres arrays
INSERT INTO items (embedding) VALUES ('{1,2,3}'), ('{4,5,6}');
```
Optionally, add a [check constraint](https://www.postgresql.org/docs/current/ddl-constraints.html) to ensure data can be converted to the `vector` type and has the expected dimensions.
```sql
ALTER TABLE items ADD CHECK (vector_dims(embedding::vector) = 3);
```
Use [expression indexing](https://www.postgresql.org/docs/current/indexes-expressional.html) to index (at a lower precision):
```sql
CREATE INDEX ON items USING hnsw ((embedding::vector(3)) vector_l2_ops);
```
and query with:
```sql
SELECT * FROM items ORDER BY embedding::vector(3) <-> '[3,1,2]' LIMIT 5;
```
## Troubleshooting
#### Why isnt a query using an index?
@@ -406,6 +475,8 @@ SELECT ...
COMMIT;
```
Also, if the table is small, a table scan may be faster.
#### Why isnt a query using a parallel table scan?
The planner doesnt consider [out-of-line storage](https://www.postgresql.org/docs/current/storage-toast.html) in cost estimates, which can make a serial scan look cheaper. You can reduce the cost of a parallel scan for a query with:
@@ -509,7 +580,7 @@ Then use `nmake` to build:
```cmd
set "PGROOT=C:\Program Files\PostgreSQL\15"
git clone --branch v0.5.0 https://github.com/pgvector/pgvector.git
git clone --branch v0.5.1 https://github.com/pgvector/pgvector.git
cd pgvector
nmake /F Makefile.win
nmake /F Makefile.win install
@@ -530,7 +601,7 @@ 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.5.0 https://github.com/pgvector/pgvector.git
git clone --branch v0.5.1 https://github.com/pgvector/pgvector.git
cd pgvector
docker build --build-arg PG_MAJOR=15 -t myuser/pgvector .
```
@@ -595,7 +666,7 @@ pgvector is available on [these providers](https://github.com/pgvector/pgvector/
## Upgrading
Install the latest version. Then in each database you want to upgrade, run:
[Install](#installation) the latest version (use the same method as the original installation). Then in each database you want to upgrade, run:
```sql
ALTER EXTENSION vector UPDATE;

View File

@@ -0,0 +1,2 @@
-- complain if script is sourced in psql, rather than via CREATE EXTENSION
\echo Use "ALTER EXTENSION vector UPDATE TO '0.5.1'" to load this file. \quit

View File

@@ -34,9 +34,6 @@ CREATE TYPE vector (
CREATE FUNCTION l2_distance(vector, vector) RETURNS float8
AS 'MODULE_PATHNAME' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
CREATE FUNCTION l2_distance(float4[], float4[]) RETURNS float8
AS 'MODULE_PATHNAME', 'float4_l2_distance' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
CREATE FUNCTION inner_product(vector, vector) RETURNS float8
AS 'MODULE_PATHNAME' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
@@ -87,9 +84,6 @@ CREATE FUNCTION vector_cmp(vector, vector) RETURNS int4
CREATE FUNCTION vector_l2_squared_distance(vector, vector) RETURNS float8
AS 'MODULE_PATHNAME' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
CREATE FUNCTION float4_l2_squared_distance(float4[], float4[]) RETURNS float8
AS 'MODULE_PATHNAME' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
CREATE FUNCTION vector_negative_inner_product(vector, vector) RETURNS float8
AS 'MODULE_PATHNAME' LANGUAGE C IMMUTABLE STRICT PARALLEL SAFE;
@@ -170,11 +164,6 @@ CREATE OPERATOR <-> (
COMMUTATOR = '<->'
);
CREATE OPERATOR <-> (
LEFTARG = float4[], RIGHTARG = float4[], PROCEDURE = l2_distance,
COMMUTATOR = '<->'
);
CREATE OPERATOR <#> (
LEFTARG = vector, RIGHTARG = vector, PROCEDURE = vector_negative_inner_product,
COMMUTATOR = '<#>'
@@ -291,11 +280,6 @@ CREATE OPERATOR CLASS vector_l2_ops
OPERATOR 1 <-> (vector, vector) FOR ORDER BY float_ops,
FUNCTION 1 vector_l2_squared_distance(vector, vector);
CREATE OPERATOR CLASS float4_l2_ops
FOR TYPE float4[] USING hnsw AS
OPERATOR 1 <-> (float4[], float4[]) FOR ORDER BY float_ops,
FUNCTION 1 float4_l2_squared_distance(float4[], float4[]);
CREATE OPERATOR CLASS vector_ip_ops
FOR TYPE vector USING hnsw AS
OPERATOR 1 <#> (vector, vector) FOR ORDER BY float_ops,

View File

@@ -1,60 +0,0 @@
#include "postgres.h"
#include <math.h>
#include "utils/array.h"
/*
* Get the L2 distance between vectors
*/
PGDLLEXPORT PG_FUNCTION_INFO_V1(float4_l2_distance);
Datum
float4_l2_distance(PG_FUNCTION_ARGS)
{
ArrayType *a = PG_GETARG_ARRAYTYPE_P(0);
ArrayType *b = PG_GETARG_ARRAYTYPE_P(1);
float *ax = (float *) ARR_DATA_PTR(a);
float *bx = (float *) ARR_DATA_PTR(b);
float distance = 0.0;
float diff;
/* TODO Check rank, dimensions, and nulls */
int dim = ARR_DIMS(a)[0];
/* Auto-vectorized */
for (int i = 0; i < dim; i++)
{
diff = ax[i] - bx[i];
distance += diff * diff;
}
PG_RETURN_FLOAT8(sqrt((double) distance));
}
/*
* Get the L2 squared distance between vectors
* This saves a sqrt calculation
*/
PGDLLEXPORT PG_FUNCTION_INFO_V1(float4_l2_squared_distance);
Datum
float4_l2_squared_distance(PG_FUNCTION_ARGS)
{
ArrayType *a = PG_GETARG_ARRAYTYPE_P(0);
ArrayType *b = PG_GETARG_ARRAYTYPE_P(1);
float *ax = (float *) ARR_DATA_PTR(a);
float *bx = (float *) ARR_DATA_PTR(b);
float distance = 0.0;
float diff;
/* TODO Check rank, dimensions, and nulls */
int dim = ARR_DIMS(a)[0];
/* Auto-vectorized */
for (int i = 0; i < dim; i++)
{
diff = ax[i] - bx[i];
distance += diff * diff;
}
PG_RETURN_FLOAT8((double) distance);
}

View File

@@ -14,6 +14,7 @@
#endif
int hnsw_ef_search;
bool hnsw_enable_parallel_build;
static relopt_kind hnsw_relopt_kind;
/*
@@ -33,18 +34,17 @@ HnswInit(void)
HNSW_DEFAULT_EF_CONSTRUCTION, HNSW_MIN_EF_CONSTRUCTION, HNSW_MAX_EF_CONSTRUCTION
#if PG_VERSION_NUM >= 130000
,AccessExclusiveLock
#endif
);
add_int_reloption(hnsw_relopt_kind, "dimensions", "Number of dimensions",
HNSW_DEFAULT_DIMENSIONS, HNSW_MIN_DIMENSIONS, HNSW_MAX_DIMENSIONS
#if PG_VERSION_NUM >= 130000
,AccessExclusiveLock
#endif
);
DefineCustomIntVariable("hnsw.ef_search", "Sets the size of the dynamic candidate list for search",
"Valid range is 1..1000.", &hnsw_ef_search,
HNSW_DEFAULT_EF_SEARCH, HNSW_MIN_EF_SEARCH, HNSW_MAX_EF_SEARCH, PGC_USERSET, 0, NULL, NULL, NULL);
/* Behind a variable for now since can be slower than building in memory */
DefineCustomBoolVariable("hnsw.enable_parallel_build", "Enables or disables building indexes in parallel",
NULL, &hnsw_enable_parallel_build,
false, PGC_USERSET, 0, NULL, NULL, NULL);
}
/*
@@ -131,7 +131,6 @@ hnswoptions(Datum reloptions, bool validate)
static const relopt_parse_elt tab[] = {
{"m", RELOPT_TYPE_INT, offsetof(HnswOptions, m)},
{"ef_construction", RELOPT_TYPE_INT, offsetof(HnswOptions, efConstruction)},
{"dimensions", RELOPT_TYPE_INT, offsetof(HnswOptions, dimensions)},
};
#if PG_VERSION_NUM >= 130000

View File

@@ -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 */
@@ -42,9 +47,6 @@
#define HNSW_DEFAULT_EF_SEARCH 40
#define HNSW_MIN_EF_SEARCH 1
#define HNSW_MAX_EF_SEARCH 1000
#define HNSW_DEFAULT_DIMENSIONS -1
#define HNSW_MIN_DIMENSIONS 1
#define HNSW_MAX_DIMENSIONS HNSW_MAX_DIM
/* Tuple types */
#define HNSW_ELEMENT_TUPLE_TYPE 1
@@ -60,7 +62,9 @@
/* PROGRESS_CREATEIDX_SUBPHASE_INITIALIZE is 1 */
#define PROGRESS_HNSW_PHASE_LOAD 2
#define HNSW_ELEMENT_TUPLE_SIZE(_datum) MAXALIGN(offsetof(HnswElementTupleData, value) + VARSIZE_ANY(_datum))
#define HNSW_MAX_SIZE (BLCKSZ - MAXALIGN(SizeOfPageHeaderData) - MAXALIGN(sizeof(HnswPageOpaqueData)) - sizeof(ItemIdData))
#define HNSW_ELEMENT_TUPLE_SIZE(size) MAXALIGN(offsetof(HnswElementTupleData, data) + (size))
#define HNSW_NEIGHBOR_TUPLE_SIZE(level, m) MAXALIGN(offsetof(HnswNeighborTupleData, indextids) + ((level) + 2) * (m) * sizeof(ItemPointerData))
#define HnswPageGetOpaque(page) ((HnswPageOpaque) PageGetSpecialPointer(page))
@@ -91,6 +95,7 @@
/* Variables */
extern int hnsw_ef_search;
extern bool hnsw_enable_parallel_build;
typedef struct HnswNeighborArray HnswNeighborArray;
@@ -99,7 +104,6 @@ typedef struct HnswElementData
List *heaptids;
uint8 level;
uint8 deleted;
bool loaded;
HnswNeighborArray *neighbors;
BlockNumber blkno;
OffsetNumber offno;
@@ -114,11 +118,13 @@ typedef struct HnswCandidate
{
HnswElement element;
float distance;
bool closer;
} HnswCandidate;
typedef struct HnswNeighborArray
{
int length;
bool closerSet;
HnswCandidate *items;
} HnswNeighborArray;
@@ -134,9 +140,51 @@ typedef struct HnswOptions
int32 vl_len_; /* varlena header (do not touch directly!) */
int m; /* number of connections */
int efConstruction; /* size of dynamic candidate list */
int dimensions;
} 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 */
@@ -164,12 +212,16 @@ typedef struct HnswBuildState
HnswElement entryPoint;
double ml;
int maxLevel;
double maxInMemoryElements;
long memoryLeft;
bool flushed;
Vector *normvec;
/* Memory */
MemoryContext tmpCtx;
/* Parallel builds */
HnswLeader *hnswleader;
HnswShared *hnswshared;
} HnswBuildState;
typedef struct HnswMetaPageData
@@ -205,7 +257,7 @@ typedef struct HnswElementTupleData
ItemPointerData heaptids[HNSW_HEAPTIDS];
ItemPointerData neighbortid;
uint16 unused2;
char value[FLEXIBLE_ARRAY_MEMBER];
Vector data;
} HnswElementTupleData;
typedef HnswElementTupleData * HnswElementTuple;
@@ -263,9 +315,8 @@ typedef struct HnswVacuumState
/* Methods */
int HnswGetM(Relation index);
int HnswGetEfConstruction(Relation index);
int HnswGetDimensions(Relation index);
FmgrInfo *HnswOptionalProcInfo(Relation index, uint16 procnum);
bool HnswNormValue(FmgrInfo *procinfo, Oid collation, Datum *value, Vector * result);
void HnswNormValue(FmgrInfo *procinfo, Oid collation, Datum *value, Vector * result);
void HnswCommitBuffer(Buffer buf, GenericXLogState *state);
Buffer HnswNewBuffer(Relation index, ForkNumber forkNum);
void HnswInitPage(Buffer buf, Page page);
@@ -291,6 +342,7 @@ void HnswLoadElement(HnswElement element, float *distance, Datum *q, Relation i
void HnswSetElementTuple(HnswElementTuple etup, HnswElement element);
void HnswUpdateConnection(HnswElement element, HnswCandidate * hc, int m, int lc, int *updateIdx, Relation index, FmgrInfo *procinfo, Oid collation);
void HnswLoadNeighbors(HnswElement element, Relation index, int m);
PGDLLEXPORT void HnswParallelBuildMain(dsm_segment *seg, shm_toc *toc);
/* Index access methods */
IndexBuildResult *hnswbuild(Relation heap, Relation index, IndexInfo *indexInfo);

View File

@@ -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/datum.h"
#include "utils/memutils.h"
@@ -36,6 +39,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
*/
@@ -82,7 +102,6 @@ HnswBuildAppendPage(Relation index, Buffer *buf, Page *page, GenericXLogState **
HnswPageGetOpaque(*page)->nextblkno = BufferGetBlockNumber(newbuf);
/* Commit */
MarkBufferDirty(*buf);
GenericXLogFinish(*state);
UnlockReleaseBuffer(*buf);
@@ -107,6 +126,7 @@ CreateElementPages(HnswBuildState * buildstate)
{
Relation index = buildstate->index;
ForkNumber forkNum = buildstate->forkNum;
Size etupAllocSize;
Size maxSize;
HnswElementTuple etup;
HnswNeighborTuple ntup;
@@ -117,11 +137,12 @@ CreateElementPages(HnswBuildState * buildstate)
ListCell *lc;
/* Calculate sizes */
maxSize = BLCKSZ - MAXALIGN(SizeOfPageHeaderData) - MAXALIGN(sizeof(HnswPageOpaqueData));
etupAllocSize = BLCKSZ;
maxSize = HNSW_MAX_SIZE;
/* Allocate once */
etup = palloc0(maxSize);
ntup = palloc0(maxSize);
etup = palloc0(etupAllocSize);
ntup = palloc0(BLCKSZ);
/* Prepare first page */
buf = HnswNewBuffer(index, forkNum);
@@ -136,13 +157,20 @@ CreateElementPages(HnswBuildState * buildstate)
Size ntupSize;
Size combinedSize;
HnswSetElementTuple(etup, element);
/* Zero memory for each element */
MemSet(etup, 0, etupAllocSize);
/* Calculate sizes */
etupSize = HNSW_ELEMENT_TUPLE_SIZE(element->value);
etupSize = HNSW_ELEMENT_TUPLE_SIZE(VARSIZE_ANY(DatumGetPointer(element->value)));
ntupSize = HNSW_NEIGHBOR_TUPLE_SIZE(element->level, buildstate->m);
combinedSize = etupSize + ntupSize + sizeof(ItemIdData);
/* Initial size check */
if (etupSize > etupAllocSize)
elog(ERROR, "index tuple too large");
HnswSetElementTuple(etup, element);
/* Keep element and neighbors on the same page if possible */
if (PageGetFreeSpace(page) < etupSize || (combinedSize <= maxSize && PageGetFreeSpace(page) < combinedSize))
HnswBuildAppendPage(index, &buf, &page, &state, forkNum);
@@ -179,7 +207,6 @@ CreateElementPages(HnswBuildState * buildstate)
insertPage = BufferGetBlockNumber(buf);
/* Commit */
MarkBufferDirty(buf);
GenericXLogFinish(state);
UnlockReleaseBuffer(buf);
@@ -227,7 +254,6 @@ CreateNeighborPages(HnswBuildState * buildstate)
elog(ERROR, "failed to add index item to \"%s\"", RelationGetRelationName(index));
/* Commit */
MarkBufferDirty(buf);
GenericXLogFinish(state);
UnlockReleaseBuffer(buf);
}
@@ -267,23 +293,26 @@ FlushPages(HnswBuildState * buildstate)
* Insert tuple
*/
static bool
InsertTuple(Relation index, Datum *values, HnswElement element, HnswBuildState * buildstate, HnswElement * dup)
InsertTuple(Relation index, Datum *values, HnswElement element, HnswBuildState * buildstate, HnswElement * dup, MemoryContext outerCtx)
{
FmgrInfo *procinfo = buildstate->procinfo;
Oid collation = buildstate->collation;
HnswElement entryPoint = buildstate->entryPoint;
int efConstruction = buildstate->efConstruction;
int m = buildstate->m;
MemoryContext oldCtx;
/* Detoast once for all calls */
element->value = PointerGetDatum(PG_DETOAST_DATUM(values[0]));
Datum value = PointerGetDatum(PG_DETOAST_DATUM(values[0]));
/* Normalize if needed */
if (buildstate->normprocinfo != NULL)
{
if (!HnswNormValue(buildstate->normprocinfo, collation, &element->value, buildstate->normvec))
return false;
}
HnswNormValue(buildstate->normprocinfo, collation, &value, buildstate->normvec);
/* Copy value to element so accessible outside of memory context */
oldCtx = MemoryContextSwitchTo(outerCtx);
element->value = datumCopy(value, false, -1);
MemoryContextSwitchTo(oldCtx);
/* Insert element in graph */
HnswInsertElement(element, entryPoint, NULL, procinfo, collation, m, efConstruction, false);
@@ -313,6 +342,21 @@ InsertTuple(Relation index, Datum *values, HnswElement element, HnswBuildState *
return *dup == NULL;
}
/*
* Get the memory used by an element
*/
static long
HnswElementMemory(HnswElement e, int m)
{
long elementSize = sizeof(HnswElementData);
elementSize += sizeof(HnswNeighborArray) * (e->level + 1);
elementSize += sizeof(HnswCandidate) * (m * (e->level + 2));
elementSize += sizeof(ItemPointerData);
elementSize += VARSIZE_ANY(DatumGetPointer(e->value));
return elementSize;
}
/*
* Callback for table_index_build_scan
*/
@@ -334,7 +378,7 @@ BuildCallback(Relation index, CALLBACK_ITEM_POINTER, Datum *values,
if (isnull[0])
return;
if (buildstate->indtuples >= buildstate->maxInMemoryElements)
if (buildstate->memoryLeft <= 0)
{
if (!buildstate->flushed)
{
@@ -349,7 +393,18 @@ BuildCallback(Relation index, CALLBACK_ITEM_POINTER, Datum *values,
oldCtx = MemoryContextSwitchTo(buildstate->tmpCtx);
if (HnswInsertTuple(buildstate->index, values, isnull, tid, buildstate->heap))
UpdateProgress(PROGRESS_CREATEIDX_TUPLES_DONE, ++buildstate->indtuples);
{
if (buildstate->hnswshared)
{
HnswShared *hnswshared = buildstate->hnswshared;
SpinLockAcquire(&hnswshared->mutex);
UpdateProgress(PROGRESS_CREATEIDX_TUPLES_DONE, ++hnswshared->indtuples);
SpinLockRelease(&hnswshared->mutex);
}
else
UpdateProgress(PROGRESS_CREATEIDX_TUPLES_DONE, ++buildstate->indtuples);
}
/* Reset memory context */
MemoryContextSwitchTo(oldCtx);
@@ -365,44 +420,27 @@ BuildCallback(Relation index, CALLBACK_ITEM_POINTER, Datum *values,
oldCtx = MemoryContextSwitchTo(buildstate->tmpCtx);
/* Insert tuple */
inserted = InsertTuple(index, values, element, buildstate, &dup);
inserted = InsertTuple(index, values, element, buildstate, &dup, oldCtx);
/* Switch memory context */
/* Reset memory context */
MemoryContextSwitchTo(oldCtx);
MemoryContextReset(buildstate->tmpCtx);
/* Add outside memory context */
if (dup != NULL)
{
HnswAddHeapTid(dup, tid);
buildstate->memoryLeft -= sizeof(ItemPointerData);
}
/* Add to buildstate or free */
if (inserted)
{
element->value = datumCopy(element->value, false, -1);
element->loaded = true;
buildstate->elements = lappend(buildstate->elements, element);
buildstate->memoryLeft -= HnswElementMemory(element, buildstate->m);
}
else
HnswFreeElement(element);
/* Reset memory context */
MemoryContextReset(buildstate->tmpCtx);
}
/*
* Get the max number of elements that fit into maintenance_work_mem
*/
static double
HnswGetMaxInMemoryElements(int m, double ml, int dimensions)
{
Size elementSize = sizeof(HnswElementData);
double avgLevel = -log(0.5) * ml;
elementSize += sizeof(HnswNeighborArray) * (avgLevel + 1);
elementSize += sizeof(HnswCandidate) * (m * (avgLevel + 2));
elementSize += sizeof(ItemPointerData);
/* TODO Handle non-vector types */
elementSize += VECTOR_SIZE(dimensions);
return (maintenance_work_mem * 1024L) / elementSize;
}
/*
@@ -418,10 +456,7 @@ InitBuildState(HnswBuildState * buildstate, Relation heap, Relation index, Index
buildstate->m = HnswGetM(index);
buildstate->efConstruction = HnswGetEfConstruction(index);
buildstate->dimensions = HnswGetDimensions(index);
if (buildstate->dimensions < 0)
buildstate->dimensions = TupleDescAttr(index->rd_att, 0)->atttypmod;
buildstate->dimensions = TupleDescAttr(index->rd_att, 0)->atttypmod;
/* Require column to have dimensions to be indexed */
if (buildstate->dimensions < 0)
@@ -445,7 +480,7 @@ InitBuildState(HnswBuildState * buildstate, Relation heap, Relation index, Index
buildstate->entryPoint = NULL;
buildstate->ml = HnswGetMl(buildstate->m);
buildstate->maxLevel = HnswGetMaxLevel(buildstate->m);
buildstate->maxInMemoryElements = HnswGetMaxInMemoryElements(buildstate->m, buildstate->ml, buildstate->dimensions);
buildstate->memoryLeft = maintenance_work_mem * 1024L;
buildstate->flushed = false;
/* Reuse for each tuple */
@@ -454,6 +489,9 @@ InitBuildState(HnswBuildState * buildstate, Relation heap, Relation index, Index
buildstate->tmpCtx = AllocSetContextCreate(CurrentMemoryContext,
"Hnsw build temporary context",
ALLOCSET_DEFAULT_SIZES);
buildstate->hnswleader = NULL;
buildstate->hnswshared = NULL;
}
/*
@@ -466,21 +504,374 @@ FreeBuildState(HnswBuildState * buildstate)
MemoryContextDelete(buildstate->tmpCtx);
}
/*
* Within leader, wait for end of heap scan
*/
static double
ParallelHeapScan(HnswBuildState * buildstate)
{
HnswShared *hnswshared = buildstate->hnswleader->hnswshared;
int nparticipanttuplesorts;
double reltuples;
nparticipanttuplesorts = buildstate->hnswleader->nparticipanttuplesorts;
for (;;)
{
SpinLockAcquire(&hnswshared->mutex);
if (hnswshared->nparticipantsdone == nparticipanttuplesorts)
{
buildstate->indtuples = hnswshared->indtuples;
reltuples = hnswshared->reltuples;
SpinLockRelease(&hnswshared->mutex);
break;
}
SpinLockRelease(&hnswshared->mutex);
ConditionVariableSleep(&hnswshared->workersdonecv,
WAIT_EVENT_PARALLEL_CREATE_INDEX_SCAN);
}
ConditionVariableCancelSleep();
return reltuples;
}
/*
* Perform a worker's portion of a parallel insert
*/
static void
HnswParallelScanAndInsert(HnswSpool * hnswspool, HnswShared * hnswshared, bool progress)
{
HnswBuildState buildstate;
#if PG_VERSION_NUM >= 120000
TableScanDesc scan;
#else
HeapScanDesc scan;
#endif
double reltuples;
IndexInfo *indexInfo;
/* Join parallel scan */
indexInfo = BuildIndexInfo(hnswspool->index);
indexInfo->ii_Concurrent = hnswshared->isconcurrent;
InitBuildState(&buildstate, hnswspool->heap, hnswspool->index, indexInfo, MAIN_FORKNUM);
/* TODO Support in-memory builds */
buildstate.memoryLeft = 0;
buildstate.flushed = true;
buildstate.hnswshared = hnswshared;
#if PG_VERSION_NUM >= 120000
scan = table_beginscan_parallel(hnswspool->heap,
ParallelTableScanFromHnswShared(hnswshared));
reltuples = table_index_build_scan(hnswspool->heap, hnswspool->index, indexInfo,
true, progress, BuildCallback,
(void *) &buildstate, scan);
#else
scan = heap_beginscan_parallel(hnswspool->heap, &hnswshared->heapdesc);
reltuples = IndexBuildHeapScan(hnswspool->heap, hnswspool->index, indexInfo,
true, BuildCallback,
(void *) &buildstate, scan);
#endif
/* Record statistics */
SpinLockAcquire(&hnswshared->mutex);
hnswshared->nparticipantsdone++;
hnswshared->reltuples += reltuples;
SpinLockRelease(&hnswshared->mutex);
/* Log statistics */
if (progress)
ereport(DEBUG1, (errmsg("leader processed " INT64_FORMAT " tuples", (int64) reltuples)));
else
ereport(DEBUG1, (errmsg("worker processed " INT64_FORMAT " tuples", (int64) reltuples)));
/* Notify leader */
ConditionVariableSignal(&hnswshared->workersdonecv);
FreeBuildState(&buildstate);
}
/*
* Perform work within a launched parallel process
*/
void
HnswParallelBuildMain(dsm_segment *seg, shm_toc *toc)
{
char *sharedquery;
HnswSpool *hnswspool;
HnswShared *hnswshared;
Relation heapRel;
Relation indexRel;
LOCKMODE heapLockmode;
LOCKMODE indexLockmode;
/* Set debug_query_string for individual workers first */
sharedquery = shm_toc_lookup(toc, PARALLEL_KEY_QUERY_TEXT, true);
debug_query_string = sharedquery;
/* Report the query string from leader */
pgstat_report_activity(STATE_RUNNING, debug_query_string);
/* Look up shared state */
hnswshared = shm_toc_lookup(toc, PARALLEL_KEY_HNSW_SHARED, false);
/* Open relations using lock modes known to be obtained by index.c */
if (!hnswshared->isconcurrent)
{
heapLockmode = ShareLock;
indexLockmode = AccessExclusiveLock;
}
else
{
heapLockmode = ShareUpdateExclusiveLock;
indexLockmode = RowExclusiveLock;
}
/* Open relations within worker */
#if PG_VERSION_NUM >= 120000
heapRel = table_open(hnswshared->heaprelid, heapLockmode);
#else
heapRel = heap_open(hnswshared->heaprelid, heapLockmode);
#endif
indexRel = index_open(hnswshared->indexrelid, indexLockmode);
/* Initialize worker's own spool */
hnswspool = (HnswSpool *) palloc0(sizeof(HnswSpool));
hnswspool->heap = heapRel;
hnswspool->index = indexRel;
/* Perform inserts */
HnswParallelScanAndInsert(hnswspool, hnswshared, false);
/* Close relations within worker */
index_close(indexRel, indexLockmode);
#if PG_VERSION_NUM >= 120000
table_close(heapRel, heapLockmode);
#else
heap_close(heapRel, heapLockmode);
#endif
}
/*
* End parallel build
*/
static void
HnswEndParallel(HnswLeader * hnswleader)
{
/* Shutdown worker processes */
WaitForParallelWorkersToFinish(hnswleader->pcxt);
/* Free last reference to MVCC snapshot, if one was used */
if (IsMVCCSnapshot(hnswleader->snapshot))
UnregisterSnapshot(hnswleader->snapshot);
DestroyParallelContext(hnswleader->pcxt);
ExitParallelMode();
}
/*
* Return size of shared memory required for parallel index build
*/
static Size
ParallelEstimateShared(Relation heap, Snapshot snapshot)
{
#if PG_VERSION_NUM >= 120000
return add_size(BUFFERALIGN(sizeof(HnswShared)), table_parallelscan_estimate(heap, snapshot));
#else
if (!IsMVCCSnapshot(snapshot))
{
Assert(snapshot == SnapshotAny);
return sizeof(HnswShared);
}
return add_size(offsetof(HnswShared, heapdesc) +
offsetof(ParallelHeapScanDescData, phs_snapshot_data),
EstimateSnapshotSpace(snapshot));
#endif
}
/*
* Within leader, participate as a parallel worker
*/
static void
HnswLeaderParticipateAsWorker(HnswBuildState * buildstate)
{
HnswLeader *hnswleader = buildstate->hnswleader;
HnswSpool *leaderworker;
/* Allocate memory and initialize private spool */
leaderworker = (HnswSpool *) palloc0(sizeof(HnswSpool));
leaderworker->heap = buildstate->heap;
leaderworker->index = buildstate->index;
/* Perform work common to all participants */
HnswParallelScanAndInsert(leaderworker, hnswleader->hnswshared, true);
}
/*
* Begin parallel build
*/
static void
HnswBeginParallel(HnswBuildState * buildstate, bool isconcurrent, int request)
{
ParallelContext *pcxt;
int scantuplesortstates;
Snapshot snapshot;
Size esthnswshared;
HnswShared *hnswshared;
HnswLeader *hnswleader = (HnswLeader *) palloc0(sizeof(HnswLeader));
bool leaderparticipates = true;
int querylen;
#ifdef DISABLE_LEADER_PARTICIPATION
leaderparticipates = false;
#endif
/* Enter parallel mode and create context */
EnterParallelMode();
Assert(request > 0);
#if PG_VERSION_NUM >= 120000
pcxt = CreateParallelContext("vector", "HnswParallelBuildMain", request);
#else
pcxt = CreateParallelContext("vector", "HnswParallelBuildMain", request, true);
#endif
scantuplesortstates = leaderparticipates ? request + 1 : request;
/* Get snapshot for table scan */
if (!isconcurrent)
snapshot = SnapshotAny;
else
snapshot = RegisterSnapshot(GetTransactionSnapshot());
/* Estimate size of workspaces */
esthnswshared = ParallelEstimateShared(buildstate->heap, snapshot);
shm_toc_estimate_chunk(&pcxt->estimator, esthnswshared);
shm_toc_estimate_keys(&pcxt->estimator, 1);
/* Finally, estimate PARALLEL_KEY_QUERY_TEXT space */
if (debug_query_string)
{
querylen = strlen(debug_query_string);
shm_toc_estimate_chunk(&pcxt->estimator, querylen + 1);
shm_toc_estimate_keys(&pcxt->estimator, 1);
}
else
querylen = 0; /* keep compiler quiet */
/* Everyone's had a chance to ask for space, so now create the DSM */
InitializeParallelDSM(pcxt);
/* If no DSM segment was available, back out (do serial build) */
if (pcxt->seg == NULL)
{
if (IsMVCCSnapshot(snapshot))
UnregisterSnapshot(snapshot);
DestroyParallelContext(pcxt);
ExitParallelMode();
return;
}
/* Store shared build state, for which we reserved space */
hnswshared = (HnswShared *) shm_toc_allocate(pcxt->toc, esthnswshared);
/* Initialize immutable state */
hnswshared->heaprelid = RelationGetRelid(buildstate->heap);
hnswshared->indexrelid = RelationGetRelid(buildstate->index);
hnswshared->isconcurrent = isconcurrent;
hnswshared->scantuplesortstates = scantuplesortstates;
ConditionVariableInit(&hnswshared->workersdonecv);
SpinLockInit(&hnswshared->mutex);
/* Initialize mutable state */
hnswshared->nparticipantsdone = 0;
hnswshared->reltuples = 0;
hnswshared->indtuples = 0;
#if PG_VERSION_NUM >= 120000
table_parallelscan_initialize(buildstate->heap,
ParallelTableScanFromHnswShared(hnswshared),
snapshot);
#else
heap_parallelscan_initialize(&hnswshared->heapdesc, buildstate->heap, snapshot);
#endif
shm_toc_insert(pcxt->toc, PARALLEL_KEY_HNSW_SHARED, hnswshared);
/* Store query string for workers */
if (debug_query_string)
{
char *sharedquery;
sharedquery = (char *) shm_toc_allocate(pcxt->toc, querylen + 1);
memcpy(sharedquery, debug_query_string, querylen + 1);
shm_toc_insert(pcxt->toc, PARALLEL_KEY_QUERY_TEXT, sharedquery);
}
/* Launch workers, saving status for leader/caller */
LaunchParallelWorkers(pcxt);
hnswleader->pcxt = pcxt;
hnswleader->nparticipanttuplesorts = pcxt->nworkers_launched;
if (leaderparticipates)
hnswleader->nparticipanttuplesorts++;
hnswleader->hnswshared = hnswshared;
hnswleader->snapshot = snapshot;
/* If no workers were successfully launched, back out (do serial build) */
if (pcxt->nworkers_launched == 0)
{
HnswEndParallel(hnswleader);
return;
}
/* Log participants */
ereport(DEBUG1, (errmsg("using %d parallel workers", pcxt->nworkers_launched)));
/* Save leader state now that it's clear build will be parallel */
buildstate->hnswleader = hnswleader;
/* Join heap scan ourselves */
if (leaderparticipates)
HnswLeaderParticipateAsWorker(buildstate);
/* Wait for all launched workers */
WaitForParallelWorkersToAttach(pcxt);
}
/*
* Build graph
*/
static void
BuildGraph(HnswBuildState * buildstate, ForkNumber forkNum)
{
int parallel_workers = 0;
UpdateProgress(PROGRESS_CREATEIDX_SUBPHASE, PROGRESS_HNSW_PHASE_LOAD);
/* Calculate parallel workers */
if (hnsw_enable_parallel_build)
parallel_workers = plan_create_index_workers(RelationGetRelid(buildstate->heap), RelationGetRelid(buildstate->index));
/* Attempt to launch parallel worker scan when required */
if (parallel_workers > 0)
{
/* TODO Support in-memory builds */
FlushPages(buildstate);
HnswBeginParallel(buildstate, buildstate->indexInfo->ii_Concurrent, parallel_workers);
}
/* Add tuples to sort */
if (buildstate->hnswleader)
buildstate->reltuples = ParallelHeapScan(buildstate);
else
{
#if PG_VERSION_NUM >= 120000
buildstate->reltuples = table_index_build_scan(buildstate->heap, buildstate->index, buildstate->indexInfo,
true, true, BuildCallback, (void *) buildstate, NULL);
buildstate->reltuples = table_index_build_scan(buildstate->heap, buildstate->index, buildstate->indexInfo,
true, true, BuildCallback, (void *) buildstate, NULL);
#else
buildstate->reltuples = IndexBuildHeapScan(buildstate->heap, buildstate->index, buildstate->indexInfo,
true, BuildCallback, (void *) buildstate, NULL);
buildstate->reltuples = IndexBuildHeapScan(buildstate->heap, buildstate->index, buildstate->indexInfo,
true, BuildCallback, (void *) buildstate, NULL);
#endif
}
/* End parallel build */
if (buildstate->hnswleader)
HnswEndParallel(buildstate->hnswleader);
}
/*

View File

@@ -131,10 +131,10 @@ WriteNewElementPages(Relation index, HnswElement e, int m, BlockNumber insertPag
BlockNumber newInsertPage = InvalidBlockNumber;
/* Calculate sizes */
etupSize = HNSW_ELEMENT_TUPLE_SIZE(e->value);
etupSize = HNSW_ELEMENT_TUPLE_SIZE(VARSIZE_ANY(DatumGetPointer(e->value)));
ntupSize = HNSW_NEIGHBOR_TUPLE_SIZE(e->level, m);
combinedSize = etupSize + ntupSize + sizeof(ItemIdData);
maxSize = BLCKSZ - MAXALIGN(SizeOfPageHeaderData) - MAXALIGN(sizeof(HnswPageOpaqueData));
maxSize = HNSW_MAX_SIZE;
minCombinedSize = etupSize + HNSW_NEIGHBOR_TUPLE_SIZE(0, m) + sizeof(ItemIdData);
/* Prepare element tuple */
@@ -201,8 +201,6 @@ WriteNewElementPages(Relation index, HnswElement e, int m, BlockNumber insertPag
HnswInsertAppendPage(index, &newbuf, &newpage, state, page);
/* Commit */
MarkBufferDirty(newbuf);
MarkBufferDirty(buf);
GenericXLogFinish(state);
/* Unlock previous buffer */
@@ -269,9 +267,6 @@ WriteNewElementPages(Relation index, HnswElement e, int m, BlockNumber insertPag
}
/* Commit */
MarkBufferDirty(buf);
if (nbuf != buf)
MarkBufferDirty(nbuf);
GenericXLogFinish(state);
UnlockReleaseBuffer(buf);
if (nbuf != buf)
@@ -390,7 +385,6 @@ HnswUpdateNeighborPages(Relation index, FmgrInfo *procinfo, Oid collation, HnswE
elog(ERROR, "failed to add index item to \"%s\"", RelationGetRelationName(index));
/* Commit */
MarkBufferDirty(buf);
GenericXLogFinish(state);
}
else
@@ -410,8 +404,9 @@ HnswAddDuplicate(Relation index, HnswElement element, HnswElement dup)
Buffer buf;
Page page;
GenericXLogState *state;
Size etupSize = HNSW_ELEMENT_TUPLE_SIZE(dup->value);
ItemId itemid;
HnswElementTuple etup;
Size etupSize;
int i;
/* Read page */
@@ -421,7 +416,9 @@ HnswAddDuplicate(Relation index, HnswElement element, HnswElement dup)
page = GenericXLogRegisterBuffer(state, buf, 0);
/* Find space */
etup = (HnswElementTuple) PageGetItem(page, PageGetItemId(page, dup->offno));
itemid = PageGetItemId(page, dup->offno);
etup = (HnswElementTuple) PageGetItem(page, itemid);
etupSize = ItemIdGetLength(itemid);
for (i = 0; i < HNSW_HEAPTIDS; i++)
{
if (!ItemPointerIsValid(&etup->heaptids[i]))
@@ -444,7 +441,6 @@ HnswAddDuplicate(Relation index, HnswElement element, HnswElement dup)
elog(ERROR, "failed to add index item to \"%s\"", RelationGetRelationName(index));
/* Commit */
MarkBufferDirty(buf);
GenericXLogFinish(state);
UnlockReleaseBuffer(buf);
@@ -504,10 +500,7 @@ HnswInsertTuple(Relation index, Datum *values, bool *isnull, ItemPointer heap_ti
/* Normalize if needed */
normprocinfo = HnswOptionalProcInfo(index, HNSW_NORM_PROC);
if (normprocinfo != NULL)
{
if (!HnswNormValue(normprocinfo, collation, &value, NULL))
return false;
}
HnswNormValue(normprocinfo, collation, &value, NULL);
/*
* Get a shared lock. This allows vacuum to ensure no in-flight inserts

View File

@@ -113,12 +113,6 @@ hnswbeginscan(Relation index, int nkeys, int norderbys)
scan->opaque = so;
/*
* 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);
return scan;
}
@@ -166,11 +160,25 @@ hnswgettuple(IndexScanDesc scan, ScanDirection dir)
if (scan->orderByData == NULL)
elog(ERROR, "cannot scan hnsw index without order");
/* Requires MVCC-compliant snapshot as not able to maintain a pin */
/* https://www.postgresql.org/docs/current/index-locking.html */
if (!IsMVCCSnapshot(scan->xs_snapshot))
elog(ERROR, "non-MVCC snapshots are not supported with hnsw");
/* Get scan value */
value = GetScanValue(scan);
/*
* 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;
}
@@ -198,15 +206,6 @@ hnswgettuple(IndexScanDesc scan, ScanDirection dir)
scan->xs_ctup.t_self = *heaptid;
#endif
/*
* Typically, an index scan must maintain a pin on the index page
* holding the item last returned by amgettuple. However, this is not
* needed with the current vacuum strategy, which ensures scans do not
* visit tuples in danger of being marked as deleted.
*
* https://www.postgresql.org/docs/current/index-locking.html
*/
scan->xs_recheckorderby = false;
return true;
}
@@ -223,9 +222,6 @@ hnswendscan(IndexScanDesc scan)
{
HnswScanOpaque so = (HnswScanOpaque) scan->opaque;
/* Release shared lock */
UnlockPage(scan->indexRelation, HNSW_SCAN_LOCK, ShareLock);
MemoryContextDelete(so->tmpCtx);
pfree(so);

View File

@@ -35,20 +35,6 @@ HnswGetEfConstruction(Relation index)
return HNSW_DEFAULT_EF_CONSTRUCTION;
}
/*
* Get the number of dimensions in the index
*/
int
HnswGetDimensions(Relation index)
{
HnswOptions *opts = (HnswOptions *) index->rd_options;
if (opts)
return opts->dimensions;
return HNSW_DEFAULT_DIMENSIONS;
}
/*
* Get proc
*/
@@ -69,7 +55,7 @@ HnswOptionalProcInfo(Relation index, uint16 procnum)
* The caller needs to free the pointer stored in value
* if it's different than the original value
*/
bool
void
HnswNormValue(FmgrInfo *procinfo, Oid collation, Datum *value, Vector * result)
{
double norm = DatumGetFloat8(FunctionCall1Coll(procinfo, collation, *value));
@@ -85,11 +71,7 @@ HnswNormValue(FmgrInfo *procinfo, Oid collation, Datum *value, Vector * result)
result->x[i] = v->x[i] / norm;
*value = PointerGetDatum(result);
return true;
}
return false;
}
/*
@@ -132,7 +114,6 @@ HnswInitRegisterPage(Relation index, Buffer *buf, Page *page, GenericXLogState *
void
HnswCommitBuffer(Buffer buf, GenericXLogState *state)
{
MarkBufferDirty(buf);
GenericXLogFinish(state);
UnlockReleaseBuffer(buf);
}
@@ -155,6 +136,7 @@ HnswInitNeighbors(HnswElement element, int m)
a = &element->neighbors[lc];
a->length = 0;
a->items = palloc(sizeof(HnswCandidate) * lm);
a->closerSet = false;
}
}
@@ -191,6 +173,8 @@ HnswInitElement(ItemPointer heaptid, int m, double ml, int maxLevel)
HnswInitNeighbors(element, m);
element->value = PointerGetDatum(NULL);
return element;
}
@@ -202,7 +186,7 @@ HnswFreeElement(HnswElement element)
{
HnswFreeNeighbors(element);
list_free_deep(element->heaptids);
if (element->loaded)
if (DatumGetPointer(element->value))
pfree(DatumGetPointer(element->value));
pfree(element);
}
@@ -230,7 +214,7 @@ HnswInitElementFromBlock(BlockNumber blkno, OffsetNumber offno)
element->blkno = blkno;
element->offno = offno;
element->neighbors = NULL;
element->loaded = false;
element->value = PointerGetDatum(NULL);
return element;
}
@@ -340,7 +324,7 @@ HnswSetElementTuple(HnswElementTuple etup, HnswElement element)
else
ItemPointerSetInvalid(&etup->heaptids[i]);
}
memcpy(&etup->value, DatumGetPointer(element->value), VARSIZE_ANY(element->value));
memcpy(&etup->data, DatumGetPointer(element->value), VARSIZE_ANY(DatumGetPointer(element->value)));
}
/*
@@ -462,12 +446,7 @@ HnswLoadElementFromTuple(HnswElement element, HnswElementTuple etup, bool loadHe
}
if (loadVec)
{
Datum value = PointerGetDatum(&etup->value);
element->value = datumCopy(value, false, -1);
element->loaded = true;
}
element->value = datumCopy(PointerGetDatum(&etup->data), false, -1);
}
/*
@@ -494,7 +473,7 @@ HnswLoadElement(HnswElement element, float *distance, Datum *q, Relation index,
/* Calculate distance */
if (distance != NULL)
*distance = (float) DatumGetFloat8(FunctionCall2Coll(procinfo, collation, *q, PointerGetDatum(&etup->value)));
*distance = (float) DatumGetFloat8(FunctionCall2Coll(procinfo, collation, *q, PointerGetDatum(&etup->data)));
UnlockReleaseBuffer(buf);
}
@@ -711,6 +690,34 @@ HnswSearchLayer(Datum q, List *ep, int ef, int lc, Relation index, FmgrInfo *pro
return w;
}
/*
* Compare candidate distances
*/
static int
#if PG_VERSION_NUM >= 130000
CompareCandidateDistances(const ListCell *a, const ListCell *b)
#else
CompareCandidateDistances(const void *a, const void *b)
#endif
{
HnswCandidate *hca = lfirst((ListCell *) a);
HnswCandidate *hcb = lfirst((ListCell *) b);
if (hca->distance < hcb->distance)
return 1;
if (hca->distance > hcb->distance)
return -1;
if (hca->element < hcb->element)
return 1;
if (hca->element > hcb->element)
return -1;
return 0;
}
/*
* Calculate the distance between elements
*/
@@ -767,33 +774,77 @@ CheckElementCloser(HnswCandidate * e, List *r, int lc, FmgrInfo *procinfo, Oid c
* Algorithm 4 from paper
*/
static List *
SelectNeighbors(List *c, int m, int lc, FmgrInfo *procinfo, Oid collation, HnswCandidate * *pruned)
SelectNeighbors(List *c, int m, int lc, FmgrInfo *procinfo, Oid collation, HnswElement e2, HnswCandidate * newCandidate, HnswCandidate * *pruned, bool sortCandidates)
{
List *r = NIL;
List *w = list_copy(c);
pairingheap *wd;
bool mustCalculate = !e2->neighbors[lc].closerSet;
List *added = NIL;
bool removedAny = false;
if (list_length(w) <= m)
return w;
wd = pairingheap_allocate(CompareNearestCandidates, NULL);
/* Ensure order of candidates is deterministic for closer caching */
if (sortCandidates)
list_sort(w, CompareCandidateDistances);
while (list_length(w) > 0 && list_length(r) < m)
{
/* Assumes w is already ordered desc */
HnswCandidate *e = llast(w);
bool closer;
w = list_delete_last(w);
closer = CheckElementCloser(e, r, lc, procinfo, collation);
/* Use previous state of r and wd to skip work when possible */
if (mustCalculate)
e->closer = CheckElementCloser(e, r, lc, procinfo, collation);
else if (list_length(added) > 0)
{
/*
* If the current candidate was closer, we only need to compare it
* with the other candidates that we have added.
*/
if (e->closer)
{
e->closer = CheckElementCloser(e, added, lc, procinfo, collation);
if (closer)
if (!e->closer)
removedAny = true;
}
else
{
/*
* If we have removed any candidates from closer, a candidate
* that was not closer earlier might now be.
*/
if (removedAny)
{
e->closer = CheckElementCloser(e, r, lc, procinfo, collation);
if (e->closer)
added = lappend(added, e);
}
}
}
else if (e == newCandidate)
{
e->closer = CheckElementCloser(e, r, lc, procinfo, collation);
if (e->closer)
added = lappend(added, e);
}
if (e->closer)
r = lappend(r, e);
else
pairingheap_add(wd, &(CreatePairingHeapNode(e)->ph_node));
}
/* Cached value can only be used in future if sorted deterministically */
e2->neighbors[lc].closerSet = sortCandidates;
/* Keep pruned connections */
while (!pairingheap_is_empty(wd) && list_length(r) < m)
r = lappend(r, ((HnswPairingHeapNode *) pairingheap_remove_first(wd))->inner);
@@ -847,28 +898,6 @@ AddConnections(HnswElement element, List *neighbors, int m, int lc)
a->items[a->length++] = *((HnswCandidate *) lfirst(lc2));
}
/*
* Compare candidate distances
*/
static int
#if PG_VERSION_NUM >= 130000
CompareCandidateDistances(const ListCell *a, const ListCell *b)
#else
CompareCandidateDistances(const void *a, const void *b)
#endif
{
HnswCandidate *hca = lfirst((ListCell *) a);
HnswCandidate *hcb = lfirst((ListCell *) b);
if (hca->distance < hcb->distance)
return 1;
if (hca->distance > hcb->distance)
return -1;
return 0;
}
/*
* Update connections
*/
@@ -904,7 +933,7 @@ HnswUpdateConnection(HnswElement element, HnswCandidate * hc, int m, int lc, int
{
HnswCandidate *hc3 = &currentNeighbors->items[i];
if (!hc3->element->loaded)
if (DatumGetPointer(hc3->element->value) == NULL)
HnswLoadElement(hc3->element, &hc3->distance, &q, index, procinfo, collation, true);
else
hc3->distance = GetCandidateDistance(hc3, q, procinfo, collation);
@@ -922,13 +951,12 @@ HnswUpdateConnection(HnswElement element, HnswCandidate * hc, int m, int lc, int
{
List *c = NIL;
/* Add and sort candidates */
/* Add 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);
SelectNeighbors(c, m, lc, procinfo, collation, hc->element, &hc2, &pruned, true);
/* Should not happen */
if (pruned == NULL)
@@ -1027,7 +1055,12 @@ HnswInsertElement(HnswElement element, HnswElement entryPoint, Relation index, F
else
lw = w;
neighbors = SelectNeighbors(lw, lm, lc, procinfo, collation, NULL);
/*
* Candidates are sorted, but not deterministically. Could set
* sortCandidates to true for in-memory builds to enable closer
* caching, but there does not seem to be a difference in performance.
*/
neighbors = SelectNeighbors(lw, lm, lc, procinfo, collation, element, NULL, NULL, false);
AddConnections(element, neighbors, lm, lc);

View File

@@ -62,7 +62,8 @@ RemoveHeapTids(HnswVacuumState * vacuumstate)
/* Iterate over nodes */
for (offno = FirstOffsetNumber; offno <= maxoffno; offno = OffsetNumberNext(offno))
{
HnswElementTuple etup = (HnswElementTuple) PageGetItem(page, PageGetItemId(page, offno));
ItemId itemid = PageGetItemId(page, offno);
HnswElementTuple etup = (HnswElementTuple) PageGetItem(page, itemid);
int idx = 0;
bool itemUpdated = false;
@@ -93,7 +94,7 @@ RemoveHeapTids(HnswVacuumState * vacuumstate)
if (itemUpdated)
{
Size etupSize = HNSW_ELEMENT_TUPLE_SIZE(PointerGetDatum(&etup->value));
Size etupSize = ItemIdGetLength(itemid);
/* Mark rest as invalid */
for (int i = idx; i < HNSW_HEAPTIDS; i++)
@@ -128,10 +129,7 @@ RemoveHeapTids(HnswVacuumState * vacuumstate)
blkno = HnswPageGetOpaque(page)->nextblkno;
if (updated)
{
MarkBufferDirty(buf);
GenericXLogFinish(state);
}
else
GenericXLogAbort(state);
@@ -229,7 +227,6 @@ RepairGraphElement(HnswVacuumState * vacuumstate, HnswElement element, HnswEleme
elog(ERROR, "failed to add index item to \"%s\"", RelationGetRelationName(index));
/* Commit */
MarkBufferDirty(buf);
GenericXLogFinish(state);
UnlockReleaseBuffer(buf);
@@ -481,11 +478,11 @@ MarkDeleted(HnswVacuumState * vacuumstate)
/* Update element and neighbors together */
for (offno = FirstOffsetNumber; offno <= maxoffno; offno = OffsetNumberNext(offno))
{
HnswElementTuple etup = (HnswElementTuple) PageGetItem(page, PageGetItemId(page, offno));
ItemId itemid = PageGetItemId(page, offno);
HnswElementTuple etup = (HnswElementTuple) PageGetItem(page, itemid);
HnswNeighborTuple ntup;
Size etupSize;
Size ntupSize;
Datum value;
Buffer nbuf;
Page npage;
BlockNumber neighborPage;
@@ -509,11 +506,8 @@ MarkDeleted(HnswVacuumState * vacuumstate)
if (ItemPointerIsValid(&etup->heaptids[0]))
continue;
/* Get datum */
value = PointerGetDatum(&etup->value);
/* Calculate sizes */
etupSize = HNSW_ELEMENT_TUPLE_SIZE(value);
etupSize = ItemIdGetLength(itemid);
ntupSize = HNSW_NEIGHBOR_TUPLE_SIZE(etup->level, vacuumstate->m);
/* Get neighbor page */
@@ -536,7 +530,7 @@ MarkDeleted(HnswVacuumState * vacuumstate)
/* Overwrite element */
etup->deleted = 1;
MemSet(&etup->value, 0, VARSIZE_ANY(value));
MemSet(&etup->data, 0, VARSIZE_ANY(&etup->data));
/* Overwrite neighbors */
for (int i = 0; i < ntup->count; i++)
@@ -551,9 +545,6 @@ MarkDeleted(HnswVacuumState * vacuumstate)
elog(ERROR, "failed to add index item to \"%s\"", RelationGetRelationName(index));
/* Commit */
MarkBufferDirty(buf);
if (nbuf != buf)
MarkBufferDirty(nbuf);
GenericXLogFinish(state);
if (nbuf != buf)
UnlockReleaseBuffer(nbuf);

View File

@@ -177,10 +177,7 @@ AddTupleToSort(Relation index, ItemPointer tid, Datum *values, IvfflatBuildState
/* Normalize if needed */
if (buildstate->normprocinfo != NULL)
{
if (!IvfflatNormValue(buildstate->normprocinfo, buildstate->collation, &value, buildstate->normvec))
return;
}
IvfflatNormValue(buildstate->normprocinfo, buildstate->collation, &value, buildstate->normvec);
/* Find the list that minimizes the distance */
for (int i = 0; i < centers->length; i++)
@@ -506,9 +503,11 @@ CreateListPages(Relation index, VectorArray centers, int dimensions,
Buffer buf;
Page page;
GenericXLogState *state;
Size listSize;
IvfflatList list;
list = palloc0(BLCKSZ);
listSize = MAXALIGN(IVFFLAT_LIST_SIZE(dimensions));
list = palloc(listSize);
buf = IvfflatNewBuffer(index, forkNum);
IvfflatInitRegisterPage(index, &buf, &page, &state);
@@ -516,13 +515,11 @@ CreateListPages(Relation index, VectorArray centers, int dimensions,
for (int i = 0; i < lists; i++)
{
OffsetNumber offno;
Datum center = PointerGetDatum(VectorArrayGet(centers, i));
Size listSize = MAXALIGN(IVFFLAT_LIST_SIZE(center));
/* Load list */
list->startPage = InvalidBlockNumber;
list->insertPage = InvalidBlockNumber;
memcpy(&list->center, DatumGetPointer(center), VARSIZE_ANY(center));
memcpy(&list->center, VectorArrayGet(centers, i), VECTOR_SIZE(dimensions));
/* Ensure free space */
if (PageGetFreeSpace(page) < listSize)

View File

@@ -52,7 +52,7 @@
#define PROGRESS_IVFFLAT_PHASE_ASSIGN 3
#define PROGRESS_IVFFLAT_PHASE_LOAD 4
#define IVFFLAT_LIST_SIZE(_datum) (offsetof(IvfflatListData, center) + VARSIZE_ANY(_datum))
#define IVFFLAT_LIST_SIZE(_dim) (offsetof(IvfflatListData, center) + VECTOR_SIZE(_dim))
#define IvfflatPageGetOpaque(page) ((IvfflatPageOpaque) PageGetSpecialPointer(page))
#define IvfflatPageGetMeta(page) ((IvfflatMetaPageData *) PageGetContents(page))
@@ -229,7 +229,7 @@ typedef struct IvfflatListData
{
BlockNumber startPage;
BlockNumber insertPage;
char center[FLEXIBLE_ARRAY_MEMBER];
Vector center;
} IvfflatListData;
typedef IvfflatListData * IvfflatList;
@@ -246,8 +246,6 @@ typedef struct IvfflatScanOpaqueData
int probes;
int dimensions;
bool first;
Buffer buf;
ItemPointerData heaptid;
/* Sorting */
Tuplesortstate *sortstate;

View File

@@ -83,10 +83,7 @@ InsertTuple(Relation index, Datum *values, bool *isnull, ItemPointer heap_tid, R
/* Normalize if needed */
normprocinfo = IvfflatOptionalProcInfo(index, IVFFLAT_NORM_PROC);
if (normprocinfo != NULL)
{
if (!IvfflatNormValue(normprocinfo, index->rd_indcollation[0], &value, NULL))
return;
}
IvfflatNormValue(normprocinfo, index->rd_indcollation[0], &value, NULL);
/* Find the insert page - sets the page and list info */
FindInsertPage(index, values, &insertPage, &listInfo);
@@ -99,7 +96,7 @@ InsertTuple(Relation index, Datum *values, bool *isnull, ItemPointer heap_tid, R
/* Get tuple size */
itemsz = MAXALIGN(IndexTupleSize(itup));
Assert(itemsz <= BLCKSZ - MAXALIGN(SizeOfPageHeaderData) - MAXALIGN(sizeof(IvfflatPageOpaqueData)));
Assert(itemsz <= BLCKSZ - MAXALIGN(SizeOfPageHeaderData) - MAXALIGN(sizeof(IvfflatPageOpaqueData)) - sizeof(ItemIdData));
/* Find a page to insert the item */
for (;;)
@@ -142,8 +139,6 @@ InsertTuple(Relation index, Datum *values, bool *isnull, ItemPointer heap_tid, R
IvfflatPageGetOpaque(page)->nextblkno = insertPage;
/* Commit */
MarkBufferDirty(newbuf);
MarkBufferDirty(buf);
GenericXLogFinish(state);
/* Unlock previous buffer */

View File

@@ -143,10 +143,6 @@ GetScanItems(IndexScanDesc scan, Datum value)
bool isnull;
ItemId itemid = PageGetItemId(page, offno);
/* Skip dead tuples */
if (scan->ignore_killed_tuples && ItemIdIsDead(itemid))
continue;
itup = (IndexTuple) PageGetItem(page, itemid);
datum = index_getattr(itup, 1, tupdesc, &isnull);
@@ -161,8 +157,6 @@ GetScanItems(IndexScanDesc scan, Datum value)
slot->tts_isnull[0] = false;
slot->tts_values[1] = PointerGetDatum(&itup->t_tid);
slot->tts_isnull[1] = false;
slot->tts_values[2] = Int32GetDatum((int) searchPage);
slot->tts_isnull[2] = false;
ExecStoreVirtualTuple(slot);
tuplesort_puttupleslot(so->sortstate, slot);
@@ -187,55 +181,6 @@ GetScanItems(IndexScanDesc scan, Datum value)
tuplesort_performsort(so->sortstate);
}
/*
* Mark prior tuple as dead
*/
static void
MarkPriorTupleDead(IndexScanDesc scan)
{
IvfflatScanOpaque so = (IvfflatScanOpaque) scan->opaque;
Buffer buf = so->buf;
Page page;
OffsetNumber maxoffno;
/* Safety check */
if (!BufferIsValid(so->buf) || !ItemPointerIsValid(&so->heaptid))
return;
/* Only a shared locked is needed for ItemIdMarkDead */
LockBuffer(buf, BUFFER_LOCK_SHARE);
page = BufferGetPage(buf);
maxoffno = PageGetMaxOffsetNumber(page);
for (OffsetNumber offno = FirstOffsetNumber; offno <= maxoffno; offno = OffsetNumberNext(offno))
{
ItemId itemid = PageGetItemId(page, offno);
IndexTuple itup = (IndexTuple) PageGetItem(page, itemid);
/*
* Find tuple. Since buffer has been pinned, tuple cannot have been
* vacuumed (and heap TID reused).
*/
if (ItemPointerEquals(&itup->t_tid, &so->heaptid))
{
/*
* Make sure tuple has not already been marked dead to avoid extra
* WAL if wal_log_hints or data checksums enabled
*/
if (!ItemIdIsDead(itemid))
{
ItemIdMarkDead(itemid);
MarkBufferDirtyHint(buf, true);
}
break;
}
}
/* Unlock buffer */
LockBuffer(buf, BUFFER_LOCK_UNLOCK);
}
/*
* Prepare for an index scan
*/
@@ -261,9 +206,7 @@ ivfflatbeginscan(Relation index, int nkeys, int norderbys)
probes = lists;
so = (IvfflatScanOpaque) palloc(offsetof(IvfflatScanOpaqueData, lists) + probes * sizeof(IvfflatScanList));
so->buf = InvalidBuffer;
so->first = true;
ItemPointerSetInvalid(&so->heaptid);
so->probes = probes;
so->dimensions = dimensions;
@@ -274,13 +217,12 @@ ivfflatbeginscan(Relation index, int nkeys, int norderbys)
/* Create tuple description for sorting */
#if PG_VERSION_NUM >= 120000
so->tupdesc = CreateTemplateTupleDesc(3);
so->tupdesc = CreateTemplateTupleDesc(2);
#else
so->tupdesc = CreateTemplateTupleDesc(3, false);
so->tupdesc = CreateTemplateTupleDesc(2, false);
#endif
TupleDescInitEntry(so->tupdesc, (AttrNumber) 1, "distance", FLOAT8OID, -1, 0);
TupleDescInitEntry(so->tupdesc, (AttrNumber) 2, "heaptid", TIDOID, -1, 0);
TupleDescInitEntry(so->tupdesc, (AttrNumber) 3, "indexblkno", INT4OID, -1, 0);
/* Prep sort */
so->sortstate = tuplesort_begin_heap(so->tupdesc, 1, attNums, sortOperators, sortCollations, nullsFirstFlags, work_mem, NULL, false);
@@ -312,7 +254,6 @@ ivfflatrescan(IndexScanDesc scan, ScanKey keys, int nkeys, ScanKey orderbys, int
#endif
so->first = true;
ItemPointerSetInvalid(&so->heaptid);
pairingheap_reset(so->listQueue);
if (keys && scan->numberOfKeys > 0)
@@ -347,6 +288,11 @@ ivfflatgettuple(IndexScanDesc scan, ScanDirection dir)
if (scan->orderByData == NULL)
elog(ERROR, "cannot scan ivfflat index without order");
/* Requires MVCC-compliant snapshot as not able to pin during sorting */
/* https://www.postgresql.org/docs/current/index-locking.html */
if (!IsMVCCSnapshot(scan->xs_snapshot))
elog(ERROR, "non-MVCC snapshots are not supported with ivfflat");
if (scan->orderByData->sk_flags & SK_ISNULL)
value = PointerGetDatum(InitVector(so->dimensions));
else
@@ -370,17 +316,10 @@ ivfflatgettuple(IndexScanDesc scan, ScanDirection dir)
if (value != scan->orderByData->sk_argument)
pfree(DatumGetPointer(value));
}
else
{
/* Mark prior tuple as dead */
if (scan->kill_prior_tuple)
MarkPriorTupleDead(scan);
}
if (tuplesort_gettupleslot(so->sortstate, true, false, so->slot, NULL))
{
ItemPointer heaptid = (ItemPointer) DatumGetPointer(slot_getattr(so->slot, 2, &so->isnull));
BlockNumber indexblkno = DatumGetInt32(slot_getattr(so->slot, 3, &so->isnull));
#if PG_VERSION_NUM >= 120000
scan->xs_heaptid = *heaptid;
@@ -388,21 +327,6 @@ ivfflatgettuple(IndexScanDesc scan, ScanDirection dir)
scan->xs_ctup.t_self = *heaptid;
#endif
/* Keep track of info needed to mark tuple as dead */
so->heaptid = *heaptid;
/* Unpin buffer */
if (BufferIsValid(so->buf))
ReleaseBuffer(so->buf);
/*
* An index scan must maintain a pin on the index page holding the
* item last returned by amgettuple
*
* https://www.postgresql.org/docs/current/index-locking.html
*/
so->buf = ReadBuffer(scan->indexRelation, indexblkno);
scan->xs_recheckorderby = false;
return true;
}
@@ -418,10 +342,6 @@ ivfflatendscan(IndexScanDesc scan)
{
IvfflatScanOpaque so = (IvfflatScanOpaque) scan->opaque;
/* Release pin */
if (BufferIsValid(so->buf))
ReleaseBuffer(so->buf);
pairingheap_free(so->listQueue);
tuplesort_end(so->sortstate);

View File

@@ -136,7 +136,6 @@ IvfflatInitRegisterPage(Relation index, Buffer *buf, Page *page, GenericXLogStat
void
IvfflatCommitBuffer(Buffer buf, GenericXLogState *state)
{
MarkBufferDirty(buf);
GenericXLogFinish(state);
UnlockReleaseBuffer(buf);
}
@@ -160,8 +159,6 @@ IvfflatAppendPage(Relation index, Buffer *buf, Page *page, GenericXLogState **st
IvfflatInitPage(newbuf, newpage);
/* Commit */
MarkBufferDirty(*buf);
MarkBufferDirty(newbuf);
GenericXLogFinish(*state);
/* Unlock */

View File

@@ -107,7 +107,6 @@ ivfflatbulkdelete(IndexVacuumInfo *info, IndexBulkDeleteResult *stats,
{
/* Delete tuples */
PageIndexMultiDelete(page, deletable, ndeletable);
MarkBufferDirty(buf);
GenericXLogFinish(state);
}
else

View File

@@ -89,7 +89,7 @@ CheckDim(int dim)
}
/*
* Ensure finite elements
* Ensure finite element
*/
static inline void
CheckElement(float value)
@@ -437,17 +437,18 @@ vector_send(PG_FUNCTION_ARGS)
/*
* Convert vector to vector
* This is needed to check the type modifier
*/
PGDLLEXPORT PG_FUNCTION_INFO_V1(vector);
Datum
vector(PG_FUNCTION_ARGS)
{
Vector *arg = PG_GETARG_VECTOR_P(0);
Vector *vec = PG_GETARG_VECTOR_P(0);
int32 typmod = PG_GETARG_INT32(1);
CheckExpectedDim(typmod, arg->dim);
CheckExpectedDim(typmod, vec->dim);
PG_RETURN_POINTER(arg);
PG_RETURN_POINTER(vec);
}
/*
@@ -464,7 +465,6 @@ array_to_vector(PG_FUNCTION_ARGS)
bool typbyval;
char typalign;
Datum *elemsp;
bool *nullsp;
int nelemsp;
if (ARR_NDIM(array) > 1)
@@ -478,7 +478,7 @@ array_to_vector(PG_FUNCTION_ARGS)
errmsg("array must not contain nulls")));
get_typlenbyvalalign(ARR_ELEMTYPE(array), &typlen, &typbyval, &typalign);
deconstruct_array(array, ARR_ELEMTYPE(array), typlen, typbyval, typalign, &elemsp, &nullsp, &nelemsp);
deconstruct_array(array, ARR_ELEMTYPE(array), typlen, typbyval, typalign, &elemsp, NULL, &nelemsp);
CheckDim(nelemsp);
CheckExpectedDim(typmod, nelemsp);
@@ -512,6 +512,12 @@ array_to_vector(PG_FUNCTION_ARGS)
errmsg("unsupported array type")));
}
/*
* Free allocation from deconstruct_array. Do not free individual elements
* when pass-by-reference since they point to original array.
*/
pfree(elemsp);
/* Check elements */
for (int i = 0; i < result->dim; i++)
CheckElement(result->x[i]);

View File

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

View File

@@ -9,18 +9,19 @@ SELECT * FROM t ORDER BY val <=> '[3,3,3]';
[1,1,1]
[1,2,3]
[1,2,4]
(3 rows)
[0,0,0]
(4 rows)
SELECT COUNT(*) FROM (SELECT * FROM t ORDER BY val <=> '[0,0,0]') t2;
count
-------
3
4
(1 row)
SELECT COUNT(*) FROM (SELECT * FROM t ORDER BY val <=> (SELECT NULL::vector)) t2;
count
-------
3
4
(1 row)
DROP TABLE t;

View File

@@ -9,18 +9,19 @@ SELECT * FROM t ORDER BY val <=> '[3,3,3]';
[1,1,1]
[1,2,3]
[1,2,4]
(3 rows)
[0,0,0]
(4 rows)
SELECT COUNT(*) FROM (SELECT * FROM t ORDER BY val <=> '[0,0,0]') t2;
count
-------
3
4
(1 row)
SELECT COUNT(*) FROM (SELECT * FROM t ORDER BY val <=> (SELECT NULL::vector)) t2;
count
-------
3
4
(1 row)
DROP TABLE t;

View File

@@ -13,10 +13,10 @@ SELECT vector_norm('[3,4]');
SELECT vector_norm('[0,1]');
SELECT vector_norm('[3e37,4e37]')::real;
SELECT l2_distance('[0,0]'::vector, '[3,4]');
SELECT l2_distance('[0,0]'::vector, '[0,1]');
SELECT l2_distance('[1,2]'::vector, '[3]');
SELECT l2_distance('[3e38]'::vector, '[-3e38]');
SELECT l2_distance('[0,0]', '[3,4]');
SELECT l2_distance('[0,0]', '[0,1]');
SELECT l2_distance('[1,2]', '[3]');
SELECT l2_distance('[3e38]', '[-3e38]');
SELECT inner_product('[1,2]', '[3,4]');
SELECT inner_product('[1,2]', '[3]');

View File

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

View File

@@ -1,93 +0,0 @@
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, $operator) = @_;
my $correct = 0;
my $total = 0;
my $explain = $node->safe_psql("postgres", qq(
SET enable_seqscan = off;
EXPLAIN ANALYZE SELECT i FROM tst ORDER BY v $operator '$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;
SELECT i FROM tst ORDER BY v $operator '$queries[$i]' LIMIT $limit;
));
my @actual_ids = split("\n", $actual);
my %actual_set = map { $_ => 1 } @actual_ids;
my @expected_ids = split("\n", $expected[$i]);
foreach (@expected_ids)
{
if (exists($actual_set{$_}))
{
$correct++;
}
$total++;
}
}
cmp_ok($correct / $total, ">=", $min, $operator);
}
# Initialize node
$node = get_new_node('node');
$node->init;
$node->start;
# Create table
$node->safe_psql("postgres", "CREATE EXTENSION vector;");
$node->safe_psql("postgres", "CREATE TABLE tst (i int4, v float4[3]);");
$node->safe_psql("postgres",
"INSERT INTO tst SELECT i, ARRAY[random(), random(), random()] FROM generate_series(1, 10000) i;"
);
# Generate queries
for (1 .. 20)
{
my $r1 = rand();
my $r2 = rand();
my $r3 = rand();
push(@queries, "{$r1,$r2,$r3}");
}
# Check each index type
my @operators = ("<->");
my @opclasses = ("float4_l2_ops");
for my $i (0 .. $#operators)
{
my $operator = $operators[$i];
my $opclass = $opclasses[$i];
# Get exact results
@expected = ();
foreach (@queries)
{
my $res = $node->safe_psql("postgres", "SELECT i FROM tst ORDER BY v $operator '$_' LIMIT $limit;");
push(@expected, $res);
}
# Add index
$node->safe_psql("postgres", "CREATE INDEX ON tst USING hnsw (v $opclass) WITH (dimensions = 3);");
my $min = $operator eq "<#>" ? 0.80 : 0.99;
test_recall($min, $operator);
}
done_testing();

View File

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