Compare commits

..

10 Commits

Author SHA1 Message Date
Andrew Kane
89dc41ce2b Fixed crash with debugging disabled [skip ci] 2026-06-10 14:31:54 -07:00
Andrew Kane
06ab41c094 Removed extra whitespace [skip ci] 2026-06-10 13:49:39 -07:00
Andrew Kane
8a8bc90d6d Added support for amgetbatch for IVFFlat [skip ci] 2026-06-10 13:47:36 -07:00
Andrew Kane
91717c81b3 Improved batch size flexibility [skip ci] 2026-06-10 12:10:20 -07:00
Andrew Kane
98a7f94608 Added comment [skip ci] 2026-06-10 11:59:38 -07:00
Andrew Kane
7db3d67b2d Re-added check for iterative scans [skip ci] 2026-06-10 11:58:19 -07:00
Andrew Kane
1dd39f66de Minimized diff [skip ci] 2026-06-10 11:56:52 -07:00
Andrew Kane
cd431c51ca Removed unneeded check [skip ci] 2026-06-10 11:55:31 -07:00
Andrew Kane
73890f2f70 Fixed comment [skip ci] 2026-06-10 11:46:15 -07:00
Andrew Kane
18980387ec Added support for amgetbatch [skip ci] 2026-06-10 11:31:57 -07:00
13 changed files with 138 additions and 274 deletions

View File

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

View File

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

View File

@@ -1329,7 +1329,7 @@ make clean && PG_CFLAGS="-DUSE_ASSERT_CHECKING" make && make install
To enable benchmarking: To enable benchmarking:
```sh ```sh
make clean && PG_CFLAGS="-DHNSW_BENCH -DIVFFLAT_BENCH" make && make install make clean && PG_CFLAGS="-DIVFFLAT_BENCH" make && make install
``` ```
To show memory usage: To show memory usage:

View File

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

View File

@@ -258,6 +258,11 @@ hnswvalidate(Oid opclassoid)
return true; return true;
} }
static void
hnswunguardbatch(IndexScanDesc scan, IndexScanBatch batch)
{
}
/* /*
* Define index handler * Define index handler
* *
@@ -279,6 +284,7 @@ hnswhandler(PG_FUNCTION_ARGS)
.amconsistentequality = false, .amconsistentequality = false,
.amconsistentordering = false, .amconsistentordering = false,
.amcanbackward = false, .amcanbackward = false,
.amcanmarkpos = false,
.amcanunique = false, .amcanunique = false,
.amcanmulticol = false, .amcanmulticol = false,
.amoptionalkey = true, .amoptionalkey = true,
@@ -290,11 +296,7 @@ hnswhandler(PG_FUNCTION_ARGS)
.amcanparallel = false, .amcanparallel = false,
.amcanbuildparallel = true, .amcanbuildparallel = true,
.amcaninclude = false, .amcaninclude = false,
#if PG_VERSION_NUM >= 170000
.amusemaintenanceworkmem = true,
#else
.amusemaintenanceworkmem = false, .amusemaintenanceworkmem = false,
#endif
.amsummarizing = false, .amsummarizing = false,
.amparallelvacuumoptions = VACUUM_OPTION_PARALLEL_BULKDEL, .amparallelvacuumoptions = VACUUM_OPTION_PARALLEL_BULKDEL,
.amkeytype = InvalidOid, .amkeytype = InvalidOid,
@@ -315,11 +317,14 @@ hnswhandler(PG_FUNCTION_ARGS)
.amadjustmembers = NULL, .amadjustmembers = NULL,
.ambeginscan = hnswbeginscan, .ambeginscan = hnswbeginscan,
.amrescan = hnswrescan, .amrescan = hnswrescan,
.amgettuple = hnswgettuple, .amgettuple = NULL,
.amgetbatch = hnswgetbatch,
.amunguardbatch = hnswunguardbatch,
.amkillitemsbatch = NULL,
.amgettransform = NULL,
.amgetbitmap = NULL, .amgetbitmap = NULL,
.amendscan = hnswendscan, .amendscan = hnswendscan,
.ammarkpos = NULL, .amposreset = NULL,
.amrestrpos = NULL,
.amestimateparallelscan = NULL, .amestimateparallelscan = NULL,
.aminitparallelscan = NULL, .aminitparallelscan = NULL,
.amparallelrescan = NULL, .amparallelrescan = NULL,
@@ -355,11 +360,7 @@ hnswhandler(PG_FUNCTION_ARGS)
amroutine->amcanbuildparallel = true; amroutine->amcanbuildparallel = true;
#endif #endif
amroutine->amcaninclude = false; amroutine->amcaninclude = false;
#if PG_VERSION_NUM >= 170000 amroutine->amusemaintenanceworkmem = false; /* not used during VACUUM */
amroutine->amusemaintenanceworkmem = true;
#else
amroutine->amusemaintenanceworkmem = false;
#endif
#if PG_VERSION_NUM >= 160000 #if PG_VERSION_NUM >= 160000
amroutine->amsummarizing = false; amroutine->amsummarizing = false;
#endif #endif

View File

@@ -18,20 +18,10 @@
#include "utils/sampling.h" #include "utils/sampling.h"
#include "vector.h" #include "vector.h"
#ifdef HNSW_BENCH
#include "portability/instr_time.h"
#endif
#if PG_VERSION_NUM >= 190000 #if PG_VERSION_NUM >= 190000
typedef Pointer Item; typedef Pointer Item;
#endif #endif
#if PG_VERSION_NUM >= 170000
#define HnswTidStore TidStore
#else
#define HnswTidStore tidhash_hash
#endif
#define HNSW_MAX_DIM 2000 #define HNSW_MAX_DIM 2000
#define HNSW_MAX_NNZ 1000 #define HNSW_MAX_NNZ 1000
@@ -88,21 +78,6 @@ typedef Pointer Item;
#define HnswPageGetOpaque(page) ((HnswPageOpaque) PageGetSpecialPointer(page)) #define HnswPageGetOpaque(page) ((HnswPageOpaque) PageGetSpecialPointer(page))
#define HnswPageGetMeta(page) ((HnswMetaPageData *) PageGetContents(page)) #define HnswPageGetMeta(page) ((HnswMetaPageData *) PageGetContents(page))
#ifdef HNSW_BENCH
#define HnswBench(name, code) \
do { \
instr_time start; \
instr_time duration; \
INSTR_TIME_SET_CURRENT(start); \
(code); \
INSTR_TIME_SET_CURRENT(duration); \
INSTR_TIME_SUBTRACT(duration, start); \
elog(INFO, "%s: %.3f ms", name, INSTR_TIME_GET_MILLISEC(duration)); \
} while (0)
#else
#define HnswBench(name, code) (code)
#endif
#if PG_VERSION_NUM >= 150000 #if PG_VERSION_NUM >= 150000
#define RandomDouble() pg_prng_double(&pg_global_prng_state) #define RandomDouble() pg_prng_double(&pg_global_prng_state)
#define SeedRandom(seed) pg_prng_seed(&pg_global_prng_state, seed) #define SeedRandom(seed) pg_prng_seed(&pg_global_prng_state, seed)
@@ -433,11 +408,10 @@ typedef struct HnswVacuumState
HnswSupport support; HnswSupport support;
/* Variables */ /* Variables */
struct HnswTidStore *deleted; struct tidhash_hash *deleted;
BufferAccessStrategy bas; BufferAccessStrategy bas;
HnswNeighborTuple ntup; HnswNeighborTuple ntup;
HnswElementData highestPoint; HnswElementData highestPoint;
HnswElementData fallbackPoint;
/* Memory */ /* Memory */
MemoryContext tmpCtx; MemoryContext tmpCtx;
@@ -491,7 +465,7 @@ IndexBulkDeleteResult *hnswbulkdelete(IndexVacuumInfo *info, IndexBulkDeleteResu
IndexBulkDeleteResult *hnswvacuumcleanup(IndexVacuumInfo *info, IndexBulkDeleteResult *stats); IndexBulkDeleteResult *hnswvacuumcleanup(IndexVacuumInfo *info, IndexBulkDeleteResult *stats);
IndexScanDesc hnswbeginscan(Relation index, int nkeys, int norderbys); IndexScanDesc hnswbeginscan(Relation index, int nkeys, int norderbys);
void hnswrescan(IndexScanDesc scan, ScanKey keys, int nkeys, ScanKey orderbys, int norderbys); void hnswrescan(IndexScanDesc scan, ScanKey keys, int nkeys, ScanKey orderbys, int norderbys);
bool hnswgettuple(IndexScanDesc scan, ScanDirection dir); IndexScanBatch hnswgetbatch(IndexScanDesc scan, IndexScanBatch priorbatch, ScanDirection dir);
void hnswendscan(IndexScanDesc scan); void hnswendscan(IndexScanDesc scan);
static inline HnswNeighborArray * static inline HnswNeighborArray *

View File

@@ -3,6 +3,7 @@
#include <limits.h> #include <limits.h>
#include "access/genam.h" #include "access/genam.h"
#include "access/indexbatch.h"
#include "access/relscan.h" #include "access/relscan.h"
#include "hnsw.h" #include "hnsw.h"
#include "lib/pairingheap.h" #include "lib/pairingheap.h"
@@ -135,6 +136,11 @@ hnswbeginscan(Relation index, int nkeys, int norderbys)
double maxMemory; double maxMemory;
scan = RelationGetIndexScan(index, nkeys, norderbys); scan = RelationGetIndexScan(index, nkeys, norderbys);
scan->maxitemsbatch = hnsw_ef_search * HNSW_HEAPTIDS;
/* unused but must be > 0 */
scan->batch_index_opaque_static = MAXALIGN(1);
scan->batch_index_opaque_dyn = 0;
scan->batch_tuples_workspace = 0;
so = (HnswScanOpaque) palloc(sizeof(HnswScanOpaqueData)); so = (HnswScanOpaque) palloc(sizeof(HnswScanOpaqueData));
so->typeInfo = HnswGetTypeInfo(index); so->typeInfo = HnswGetTypeInfo(index);
@@ -184,12 +190,13 @@ hnswrescan(IndexScanDesc scan, ScanKey keys, int nkeys, ScanKey orderbys, int no
} }
/* /*
* Fetch the next tuple in the given scan * Fetch the next batch in the given scan
*/ */
bool IndexScanBatch
hnswgettuple(IndexScanDesc scan, ScanDirection dir) hnswgetbatch(IndexScanDesc scan, IndexScanBatch priorbatch, ScanDirection dir)
{ {
HnswScanOpaque so = (HnswScanOpaque) scan->opaque; HnswScanOpaque so = (HnswScanOpaque) scan->opaque;
IndexScanBatch batch = indexam_util_alloc_batch(scan);
MemoryContext oldCtx = MemoryContextSwitchTo(so->tmpCtx); MemoryContext oldCtx = MemoryContextSwitchTo(so->tmpCtx);
/* /*
@@ -245,6 +252,7 @@ hnswgettuple(IndexScanDesc scan, ScanDirection dir)
HnswSearchCandidate *sc; HnswSearchCandidate *sc;
HnswElement element; HnswElement element;
ItemPointer heaptid; ItemPointer heaptid;
int nitems = 0;
if (list_length(so->w) == 0) if (list_length(so->w) == 0)
{ {
@@ -290,6 +298,8 @@ hnswgettuple(IndexScanDesc scan, ScanDirection dir)
break; break;
} }
while (list_length(so->w) != 0)
{
sc = llast(so->w); sc = llast(so->w);
element = HnswPtrAccess(base, sc->element); element = HnswPtrAccess(base, sc->element);
@@ -318,16 +328,34 @@ hnswgettuple(IndexScanDesc scan, ScanDirection dir)
so->previousDistance = sc->distance; so->previousDistance = sc->distance;
} }
batch->items[nitems].tableTid = *heaptid;
batch->items[nitems].indexOffset = -1;
batch->items[nitems].tupleOffset = 0;
nitems++;
/* Keep batch size flexible */
if (nitems == scan->maxitemsbatch)
break;
}
/* Needed for strict iterative scans */
if (nitems == 0)
continue;
MemoryContextSwitchTo(oldCtx); MemoryContextSwitchTo(oldCtx);
scan->xs_heaptid = *heaptid;
scan->xs_recheck = false; scan->xs_recheck = false;
scan->xs_recheckorderby = false; scan->xs_recheckorderby = false;
return true;
batch->firstItem = 0;
batch->lastItem = nitems - 1;
batch->dir = ForwardScanDirection;
return batch;
} }
MemoryContextSwitchTo(oldCtx); MemoryContextSwitchTo(oldCtx);
return false; indexam_util_release_batch(scan, batch);
return NULL;
} }
/* /*

View File

@@ -546,9 +546,6 @@ HnswLoadElementImpl(BlockNumber blkno, OffsetNumber offno, double *distance, Hns
Assert(HnswIsElementTuple(etup)); Assert(HnswIsElementTuple(etup));
if (unlikely(etup->deleted))
elog(ERROR, "cannot load deleted element");
/* Calculate distance */ /* Calculate distance */
if (distance != NULL) if (distance != NULL)
{ {

View File

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

View File

@@ -399,7 +399,6 @@ InitBuildState(IvfflatBuildState * buildstate, Relation heap, Relation index, In
buildstate->slot = MakeSingleTupleTableSlot(buildstate->sortdesc, &TTSOpsVirtual); buildstate->slot = MakeSingleTupleTableSlot(buildstate->sortdesc, &TTSOpsVirtual);
/* TODO Ensure within maintenance_work_mem */
buildstate->centers = VectorArrayInit(buildstate->lists, buildstate->dimensions, buildstate->typeInfo->itemSize(buildstate->dimensions)); buildstate->centers = VectorArrayInit(buildstate->lists, buildstate->dimensions, buildstate->typeInfo->itemSize(buildstate->dimensions));
buildstate->listInfo = palloc(sizeof(ListInfo) * buildstate->lists); buildstate->listInfo = palloc(sizeof(ListInfo) * buildstate->lists);

View File

@@ -175,6 +175,11 @@ ivfflatvalidate(Oid opclassoid)
return true; return true;
} }
static void
ivfflatunguardbatch(IndexScanDesc scan, IndexScanBatch batch)
{
}
/* /*
* Define index handler * Define index handler
* *
@@ -196,6 +201,7 @@ ivfflathandler(PG_FUNCTION_ARGS)
.amconsistentequality = false, .amconsistentequality = false,
.amconsistentordering = false, .amconsistentordering = false,
.amcanbackward = false, .amcanbackward = false,
.amcanmarkpos = false,
.amcanunique = false, .amcanunique = false,
.amcanmulticol = false, .amcanmulticol = false,
.amoptionalkey = true, .amoptionalkey = true,
@@ -228,11 +234,14 @@ ivfflathandler(PG_FUNCTION_ARGS)
.amadjustmembers = NULL, .amadjustmembers = NULL,
.ambeginscan = ivfflatbeginscan, .ambeginscan = ivfflatbeginscan,
.amrescan = ivfflatrescan, .amrescan = ivfflatrescan,
.amgettuple = ivfflatgettuple, .amgettuple = NULL,
.amgetbatch = ivfflatgetbatch,
.amunguardbatch = ivfflatunguardbatch,
.amkillitemsbatch = NULL,
.amgettransform = NULL,
.amgetbitmap = NULL, .amgetbitmap = NULL,
.amendscan = ivfflatendscan, .amendscan = ivfflatendscan,
.ammarkpos = NULL, .amposreset = NULL,
.amrestrpos = NULL,
.amestimateparallelscan = NULL, .amestimateparallelscan = NULL,
.aminitparallelscan = NULL, .aminitparallelscan = NULL,
.amparallelrescan = NULL, .amparallelrescan = NULL,

View File

@@ -344,7 +344,7 @@ IndexBulkDeleteResult *ivfflatbulkdelete(IndexVacuumInfo *info, IndexBulkDeleteR
IndexBulkDeleteResult *ivfflatvacuumcleanup(IndexVacuumInfo *info, IndexBulkDeleteResult *stats); IndexBulkDeleteResult *ivfflatvacuumcleanup(IndexVacuumInfo *info, IndexBulkDeleteResult *stats);
IndexScanDesc ivfflatbeginscan(Relation index, int nkeys, int norderbys); IndexScanDesc ivfflatbeginscan(Relation index, int nkeys, int norderbys);
void ivfflatrescan(IndexScanDesc scan, ScanKey keys, int nkeys, ScanKey orderbys, int norderbys); void ivfflatrescan(IndexScanDesc scan, ScanKey keys, int nkeys, ScanKey orderbys, int norderbys);
bool ivfflatgettuple(IndexScanDesc scan, ScanDirection dir); IndexScanBatch ivfflatgetbatch(IndexScanDesc scan, IndexScanBatch priorbatch, ScanDirection dir);
void ivfflatendscan(IndexScanDesc scan); void ivfflatendscan(IndexScanDesc scan);
#endif #endif

View File

@@ -3,6 +3,7 @@
#include <float.h> #include <float.h>
#include "access/genam.h" #include "access/genam.h"
#include "access/indexbatch.h"
#include "access/itup.h" #include "access/itup.h"
#include "access/relscan.h" #include "access/relscan.h"
#include "access/tupdesc.h" #include "access/tupdesc.h"
@@ -261,6 +262,11 @@ ivfflatbeginscan(Relation index, int nkeys, int norderbys)
MemoryContext oldCtx; MemoryContext oldCtx;
scan = RelationGetIndexScan(index, nkeys, norderbys); scan = RelationGetIndexScan(index, nkeys, norderbys);
scan->maxitemsbatch = 1000;
/* unused but must be > 0 */
scan->batch_index_opaque_static = MAXALIGN(1);
scan->batch_index_opaque_dyn = 0;
scan->batch_tuples_workspace = 0;
/* Get lists and dimensions from metapage */ /* Get lists and dimensions from metapage */
IvfflatGetMetaPageInfo(index, &lists, &dimensions); IvfflatGetMetaPageInfo(index, &lists, &dimensions);
@@ -348,14 +354,16 @@ ivfflatrescan(IndexScanDesc scan, ScanKey keys, int nkeys, ScanKey orderbys, int
} }
/* /*
* Fetch the next tuple in the given scan * Fetch the next batch in the given scan
*/ */
bool IndexScanBatch
ivfflatgettuple(IndexScanDesc scan, ScanDirection dir) ivfflatgetbatch(IndexScanDesc scan, IndexScanBatch priorbatch, ScanDirection dir)
{ {
IvfflatScanOpaque so = (IvfflatScanOpaque) scan->opaque; IvfflatScanOpaque so = (IvfflatScanOpaque) scan->opaque;
IndexScanBatch batch = indexam_util_alloc_batch(scan);
ItemPointer heaptid; ItemPointer heaptid;
bool isnull; bool isnull;
int nitems = 0;
/* /*
* Index can be used to scan backward, but Postgres doesn't support * Index can be used to scan backward, but Postgres doesn't support
@@ -393,17 +401,37 @@ ivfflatgettuple(IndexScanDesc scan, ScanDirection dir)
while (!tuplesort_gettupleslot(so->sortstate, true, false, so->mslot, NULL)) while (!tuplesort_gettupleslot(so->sortstate, true, false, so->mslot, NULL))
{ {
if (so->listIndex == so->maxProbes) if (so->listIndex == so->maxProbes)
return false; {
indexam_util_release_batch(scan, batch);
return NULL;
}
IvfflatBench("GetScanItems", GetScanItems(scan, so->value)); IvfflatBench("GetScanItems", GetScanItems(scan, so->value));
} }
for (;;)
{
heaptid = (ItemPointer) DatumGetPointer(slot_getattr(so->mslot, 2, &isnull)); heaptid = (ItemPointer) DatumGetPointer(slot_getattr(so->mslot, 2, &isnull));
scan->xs_heaptid = *heaptid; batch->items[nitems].tableTid = *heaptid;
batch->items[nitems].indexOffset = -1;
batch->items[nitems].tupleOffset = 0;
nitems++;
if (nitems == scan->maxitemsbatch)
break;
if (!tuplesort_gettupleslot(so->sortstate, true, false, so->mslot, NULL))
break;
}
scan->xs_recheck = false; scan->xs_recheck = false;
scan->xs_recheckorderby = false; scan->xs_recheckorderby = false;
return true;
batch->firstItem = 0;
batch->lastItem = nitems - 1;
batch->dir = ForwardScanDirection;
return batch;
} }
/* /*