cuvs

package
v0.0.0-debug-20260702 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 2, 2026 License: Apache-2.0 Imports: 12 Imported by: 0

Documentation

Overview

Package cuvs carries shared cuvs-specific helpers — currently the CDC wire format used by CAGRA and IVF-PQ for tag=1 event chunks.

Lives one directory below pkg/vectorindex so callers can reach these helpers without dragging in the entire pkg/vectorindex surface, and so future cuvs-only utilities have a natural home.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func CdcAppendEventsSql 

func CdcAppendEventsSql(
	tblcfg vectorindex.IndexTableConfig,
	indexId string,
	startChunkId int64,
	records []byte,
	recordSizes []int,
	colMetaJSON string,
) ([]string, error)

CdcAppendEventsSql formats one or more INSERT statements that append the given encoded event-record bytes as new tag=1 chunks. records is the concatenation of records produced by EncodeEventRecord (in temporal order). The caller is responsible for not letting a single record straddle a chunk boundary; this helper accepts a `recordSizes` slice telling it where the record boundaries are so it can pack chunks while respecting them.

Each emitted chunk is wrapped in the CDC frame (see FrameCdcChunk); the payload budget per chunk is MaxChunkSize - cdcFrameOverhead so the on-wire size stays within MaxChunkSize. Empty records → no SQL.

chunkId starts at startChunkId and increments per emitted chunk.

colMetaJSON is the INCLUDE-column metadata (cuvscdc.ResolveIncludeColumns output, e.g. `[{"name":"a","type":1},...]`); when non-empty it is embedded in EVERY emitted chunk's frame header so the search-side can decode the chunk's records (and recover ibpr) without depending on a tag=0 sub-index. Pass "" when the index has no INCLUDE columns.

func CdcIncludeBytesPerRow 

func CdcIncludeBytesPerRow(colMetaJSON string) (int, error)

CdcIncludeBytesPerRow returns the per-row INCLUDE byte size for a given column-meta JSON. Format mirrors what gpu_cagra_set_filter_columns consumes in cgo/cuvs/cagra_c.h: a JSON array of {"name": ..., "type": N}. Returns (0, nil) for empty/whitespace-only JSON or an empty array.

type code → bytes (matches cgo/cuvs/filter.hpp:51-57): 

0 int32   → 4
1 int64   → 8
2 float32 → 4
3 float64 → 8
4 uint64  → 8 (varchar hash)

func CdcLoadEventsSql 

func CdcLoadEventsSql(tblcfg vectorindex.IndexTableConfig, indexId string) string

CdcLoadEventsSql formats the SELECT that returns every tag=1 chunk for the given index_id. No ORDER BY (per repo convention); the caller must sort chunks by chunk_id in Go before replay since record ordering across chunks matters for last-event-wins semantics.

func EncodeEventRecord 

func EncodeEventRecord(
	dst []byte,
	op CdcOp,
	pkid int64,
	vec []float32,
	include []byte,
	dim int,
	includeBytesPerRow int,
) ([]byte, error)

EncodeEventRecord appends one record to dst and returns the new slice. vec is required iff op==CdcOpInsert; include is required iff op==CdcOpInsert AND includeBytesPerRow > 0. dim must be the index's dimensionality.

func EncodeIncludeRow 

func EncodeIncludeRow(bindings []IncludeBinding, row []any, includeBytesPerRow int) ([]byte, error)

EncodeIncludeRow encodes one row's INCLUDE column values into the row-major byte layout consumed by IncludeColSizes / SplitIncludeBytes: 

col0_value || col1_value || ... || null_mask

where null_mask is ceil(ncols/8) bytes, LSB-first within each byte (bit i = 1 means binding i is NULL). row[binding.Pos] supplies the native Go value (int32 / int64 / uint64 / float32 / float64) or nil.

Returns nil when bindings is empty.

func FrameCdcChunk 

func FrameCdcChunk(records, header []byte, nInserts, nDeletes, nUpserts uint32) []byte

FrameCdcChunk wraps the given record bytes (plus an optional header, typically colMetaJSON) into the on-wire chunk frame described above. nInserts / nDeletes / nUpserts are the per-op record counts contained in `records`, written into the frame header so downstream consumers (idxcron Updatable, replay) can read counts without walking every record.

The returned slice is exactly cdcFrameOverhead + len(header) + len(records) bytes. Exposed so tests can construct framed chunks directly.

Pass header=nil when the chunk has no INCLUDE-column metadata.

func IncludeColSizes 

func IncludeColSizes(colMetaJSON string) ([]int, error)

IncludeColSizes returns the per-column elem size in bytes for a colMetaJSON. Returns nil for empty / no INCLUDE columns. Used both for size accounting and for demuxing row-major include bytes into per-column slices.

func MarshalColMetaJSON 

func MarshalColMetaJSON(entries []ColMetaEntry) (string, error)

MarshalColMetaJSON encodes the canonical colMetaJSON for a list of INCLUDE-column descriptors. Single producer used by both the iscp writer (via ResolveIncludeColumns) and the build-time table function (via colMetaJSONFromCols) so the wire shape is identical and column names containing `"` or `\` escape correctly.

Returns "" when entries is empty so callers can skip embedding a header without a special case.

func NextChunkIdSql 

func NextChunkIdSql(tblcfg vectorindex.IndexTableConfig, indexId string, tag vectorindex.ChunkTag) string

NextChunkIdSql formats a SELECT that returns the next available chunk_id for (index_id, tag). Returns 0 when no rows exist for the given tag.

Caller typically does: 

res, _ := runSql(sqlproc, NextChunkIdSql(...))
defer res.Close()
next := ParseNextChunkId(res)  // 0 if empty

func PeekColMetaJSON 

func PeekColMetaJSON(chunks []EventChunk) (string, error)

PeekColMetaJSON returns the colMetaJSON embedded in the first chunk's frame header section. Every chunk carries the header (writer-side invariant of CdcAppendEventsSql), so chunks[0] is always sufficient — no need to sort or scan. Returns "" when the header section is empty (the index has no INCLUDE columns) or chunks is empty.

Used by the search side when no tag=0 sub-index has loaded: peek the header to compute includeBytesPerRow BEFORE calling ReplayEventLog (which itself needs ibpr to decode INSERT records).

func SaveSmallTailAsCdc 

func SaveSmallTailAsCdc(
	tblcfg vectorindex.IndexTableConfig,
	rows []PendingRecord,
	dim int,
	includeBytesPerRow int,
	colMetaJSON string,
) ([]string, error)

SaveSmallTailAsCdc encodes rows as cuvs CDC tag=1 INSERT records under tblcfg.IndexTable with index_id = vectorindex.CdcTailId. Used by cagra_create / ivfpq_create when the trailing partial chunk is smaller than the cuvs minimum — those rows land in the event log so brute-force replay can serve them until the next rebuild lifts the tail back above threshold.

Returns the INSERT SQL strings the caller must run inside the build txn. chunk_id starts at 0; the build txn already wipes the storage table for this index slice (ALTER REINDEX) or is run once at CREATE INDEX with the table empty.

colMetaJSON (cuvscdc.ResolveIncludeColumns output) is embedded in the frame header section of every emitted chunk so the search side can recover the INCLUDE-column layout for tag=1 replay even when no tag=0 sub-index exists. Pass "" for indexes with no INCLUDE columns.

When rows is empty, returns nil. Empty source + INCLUDE columns is handled by the first ongoing CDC iteration (CagraSync.Save / IvfpqSync.Save), which also embeds colMetaJSON in its frames.

func SortChunks 

func SortChunks(chunks []EventChunk)

SortChunks sorts chunks ascending by chunk_id in place.

func SplitIncludeBytes 

func SplitIncludeBytes(
	colMetaJSON string,
	includeBytes []byte,
	nrows uint64,
	includeBytesPerRow int,
) ([][]byte, [][]uint32, error)

SplitIncludeBytes demuxes the row-major INCLUDE byte stream produced by EncodeEventRecord (or a CDC writer) into one row-major byte slice per column plus a packed uint32 null bitmap per column in the shape that gpu_*_add_filter_chunk consumes (LSB-first; bit i = 1 means row i IS NULL).

includeBytes layout per row: col0_value || col1_value || ... || null_mask where null_mask is ceil(ncols/8) bytes, LSB-first within each byte.

Returned colNulls[i] is nil when column i has no nulls in this batch (matches the AddFilterChunk convention of skipping the null-mask path).

func UnframeCdcChunk 

func UnframeCdcChunk(framed []byte) (records, header []byte, nInserts, nDeletes, nUpserts uint32, err error)

UnframeCdcChunk validates the frame and returns the record bytes plus the header bytes (both aliased into framed) and the per-op counts (inserts, deletes, upserts) recorded at frame time. Returns an error on any framing inconsistency: short input, wrong magic, unknown version, length overrun, or CRC mismatch.

header is nil when the chunk has no INCLUDE-column metadata (H=0).

Types

type CdcEventRecord 

type CdcEventRecord struct {
	Op      CdcOp
	Pkid    int64
	Vec     []float32 // populated only for CdcOpInsert
	Include []byte    // populated only for CdcOpInsert (and only when includeBytesPerRow > 0)
}

CdcEventRecord is the decoded form of one tag=1 record.

func DecodeEventRecord 

func DecodeEventRecord(
	src []byte,
	dim int,
	includeBytesPerRow int,
) (rec CdcEventRecord, n int, ok bool)

DecodeEventRecord decodes the next record at src[0:]. Returns the record and the number of bytes consumed. Returns ok=false when src cannot start a valid record (e.g. unknown op byte, or fewer bytes than the record needs) — the caller treats this as the end of the stream within the chunk.

Vec and Include in the returned record alias into src; copy if you need to retain them past the next call.

type CdcOp 

type CdcOp byte

CdcOp is the op code stored in the leading byte of each event record. 

const (
	CdcOpDelete CdcOp = 0
	CdcOpInsert CdcOp = 1
	// CdcOpUpsert has the same payload as CdcOpInsert (op|pkid|vec|include)
	// but is encoded distinctly so downstream consumers can ignore UPSERTs
	// when computing reliable new-row counts (see package comment).
	CdcOpUpsert CdcOp = 2
)

type ColMetaEntry 

type ColMetaEntry struct {
	Name string `json:"name"`
	Type int    `json:"type"`
}

ColMetaEntry is the canonical on-wire shape of one INCLUDE-column descriptor in colMetaJSON. The JSON shape is `[{"name":"foo","type":1},...]` — the C++ FilterStore + Go decoders expect exactly that. Producers Marshal a []ColMetaEntry via MarshalColMetaJSON.

type EventChunk 

type EventChunk struct {
	ChunkId int64
	Data    []byte
}

EventChunk is one row from CdcLoadEventsSql, wired up so the caller can sort by chunk_id before replay.

type IncludeBinding 

type IncludeBinding struct {
	Name      string // canonical column name
	Pos       int32  // index into the row []any delivered by ISCP
	TypeCode  int    // colmeta type code: 0=int32, 1=int64, 2=float32, 3=float64, 4=uint64
	SizeBytes int    // element size derived from TypeCode (4 or 8)
}

IncludeBinding is one INCLUDE column's resolved binding to a source-table column. Built once at CDC writer construction via ResolveIncludeColumns and re-used per row to encode the include byte stream.

func ResolveIncludeColumns 

func ResolveIncludeColumns(
	includedColumns string,
	nameToPos map[string]int32,
	colTypeID func(pos int32) int32,
) ([]IncludeBinding, string, int, error)

ResolveIncludeColumns parses the comma-separated names in includedColumns and resolves each against the source table's column layout. nameToPos maps a column name to its position in the row []any delivered by ISCP; colTypeID returns the source-table type ID (matching pkg/container/types.T values) for that position.

Returns:

  • bindings: per-column resolved (Pos, TypeCode, SizeBytes), declaration order
  • colMetaJSON: shape that IncludeColSizes / SplitIncludeBytes consume (so writer and sync agree on layout by construction)
  • includeBytesPerRow: total bytes per row (per-col data + null-mask)

All zero when includedColumns is empty.

type OverflowEntry 

type OverflowEntry struct {
	Pkid    int64
	Vec     []float32
	Include []byte
}

OverflowEntry is one row in the brute-force overflow.

type PendingRecord 

type PendingRecord struct {
	Pkid    int64
	Vec     []float32
	Include []byte
}

PendingRecord is one source-table row buffered for CDC tag=1 emission when the cuvs build can't take it (small dataset or partial trailing chunk below intermediate_graph_degree / lists). Vec must be a self-owned copy — the table-function row buffer underneath argVecs is reused per call. Include is the already-encoded INCLUDE-column payload (matches includeBytesPerRow); pass nil when the index has no INCLUDE columns.

type ReplayState 

type ReplayState struct {
	Deleted  []int64
	Overflow []OverflowEntry

	// ColMetaJSON is the payload of a CdcOpHeader record observed during
	// replay, when one was present (small-tail emit path writes it as
	// the first record of chunk_id=0). Empty otherwise. Callers that
	// need the INCLUDE-column layout but have no tag=0 sub-index read
	// this back here.
	ColMetaJSON string
}

ReplayState is the post-replay output: pkids deleted in the main index and pkids living in the brute-force overflow with their vec + include bytes. The two are mutually exclusive — INSERT-after-DELETE moves the pkid out of Deleted, DELETE-after-INSERT moves it out of Overflow.

func ReplayEventLog 

func ReplayEventLog(
	chunks []EventChunk,
	dim int,
	includeBytesPerRow int,
) (ReplayState, error)

ReplayEventLog walks the chunks (assumed sorted by chunk_id) and applies each record in order, returning the final (deleted, overflow) state. dim and includeBytesPerRow describe the INSERT record layout. Replay is O(n) in event count.

Source Files

View all Source files

Directories

Path Synopsis
Package idxcron carries the shared cuvs idxcron Updatable body.
 Package idxcron carries the shared cuvs idxcron Updatable body.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.