iscp

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: 47 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	ISCPDataType_Snapshot int8 = iota
	ISCPDataType_Tail
)
View Source 
const (
	DefaultGCInterval             = time.Hour
	DefaultGCTTL                  = time.Hour
	DefaultSyncTaskInterval       = time.Second * 10
	DefaultFlushWatermarkInterval = time.Hour
	DefaultFlushWatermarkTTL      = time.Hour

	DefaultRetryTimes    = 5
	DefaultRetryInterval = time.Second
	DefaultRetryDuration = time.Minute * 10
)
View Source 
const (
	TriggerType_Invalid uint16 = iota
	TriggerType_Default
	TriggerType_AlwaysUpdate
	TriggerType_Timed
)
View Source 
const (
	ISCPJobState_Invalid int8 = iota
	ISCPJobState_Pending
	ISCPJobState_Running
	ISCPJobState_Completed
	ISCPJobState_Error
	ISCPJobState_Canceled
)
View Source 
const (
	JobStage_Init int8 = iota
	JobStage_Running
)
View Source 
const (
	ISCPWorkerThread = 10

	SubmitRetryTimes    = 1000
	SubmitRetryDuration = time.Hour
)
View Source 
const (
	DefaultMaxChangeInterval = time.Minute * 30
)
View Source 
const (
	MAX_CDC_DATA_SIZE = 8192
)
View Source 
const (
	MOISCPLogTableName = catalog.MO_ISCP_LOG
)
View Source 
const (
	PermanentErrorThreshold = 10000
)
View Source 
const (
	TargetDbName = "test_intra_system_change_propagation"
)

Variables

View Source 
var CheckLeaseWithRetry = func(
	ctx context.Context,
	cnUUID string,
	txnEngine engine.Engine,
	cnTxnClient client.TxnClient,
) (ok bool, err error) {
	defer func() {
		if err != nil || !ok {
			logutil.Errorf(
				"ISCP-Task check lease failed, err=%v, ok=%v, cnUUID=%v",
				zap.Error(err),
				zap.Bool("ok", ok),
				zap.String("cnUUID", cnUUID),
			)
		}
	}()
	err = retry(
		ctx,
		func() error {
			ok, err = checkLease(ctx, cnUUID, txnEngine, cnTxnClient)
			return err
		},
		DefaultRetryTimes,
		DefaultRetryInterval,
		DefaultRetryDuration,
	)
	return
}
View Source 
var CollectChanges = func(ctx context.Context, rel engine.Relation, fromTs, toTs types.TS, mp *mpool.MPool) (engine.ChangesHandle, error) {
	return rel.CollectChanges(ctx, fromTs, toTs, false, mp)
}
View Source 
var ExecWithResult = func(
	ctx context.Context,
	sql string,
	cnUUID string,
	txn client.TxnOperator,
) (executor.Result, error) {
	v, ok := moruntime.ServiceRuntime(cnUUID).GetGlobalVariables(moruntime.InternalSQLExecutor)
	if !ok {
		panic("missing lock service")
	}

	exec := v.(executor.SQLExecutor)
	opts := executor.Options{}.
		WithDisableIncrStatement().
		WithTxn(txn)

	return exec.Exec(ctx, sql, opts)
}
View Source 
var FlushJobStatusOnIterationState = func(
	ctx context.Context,
	cnUUID string,
	cnEngine engine.Engine,
	cnTxnClient client.TxnClient,
	accountID uint32,
	tableID uint64,
	jobNames []string,
	jobIDs []uint64,
	lsns []uint64,
	jobStatuses []*JobStatus,
	watermark types.TS,
	state int8,
	prevLSN []uint64,
) (err error) {

	ctx = context.WithValue(ctx, defines.TenantIDKey{}, catalog.System_Account)
	ctx, cancel := context.WithTimeout(ctx, time.Minute*5)
	defer cancel()
	txnWriter, err := getTxn(ctx, cnEngine, cnTxnClient, "iscp iteration")
	if err != nil {
		return
	}
	defer func() {
		if err != nil {
			err = errors.Join(err, txnWriter.Rollback(ctx))
		} else {
			err = txnWriter.Commit(ctx)
		}
		if err != nil {
			logutil.Error(
				"ISCP-Task flush job status failed",
				zap.Uint32("tenantID", accountID),
				zap.Uint64("tableID", tableID),
				zap.Strings("jobNames", jobNames),
				zap.Int8("state", state),
				zap.Error(err),
			)
		}
	}()
	for i := range jobNames {
		jobStatuses[i].LSN = lsns[i]
		err = FlushStatus(
			ctx,
			cnUUID,
			txnWriter,
			accountID,
			tableID,
			jobNames[i],
			jobIDs[i],
			jobStatuses[i],
			watermark,
			state,
			prevLSN[i],
		)
		if err != nil {
			return
		}
	}
	return
}
View Source 
var GetJobSpecs = func(
	ctx context.Context,
	cnUUID string,
	cnTxnClient client.TxnClient,
	cnEngine engine.Engine,
	txn client.TxnOperator,
	tenantId uint32,
	tableID uint64,
	jobName []string,
	lsns []uint64,
	watermark types.TS,
	jobStatuses []*JobStatus,
	jobIDs []uint64,
) (jobSpec []*JobSpec, prevStatus []*JobStatus, err error) {
	var buf bytes.Buffer
	buf.WriteString("SELECT job_spec, job_status FROM mo_catalog.mo_iscp_log WHERE")
	for i, jobName := range jobName {
		if i > 0 {
			buf.WriteString(" OR")
		}
		buf.WriteString(fmt.Sprintf(" (account_id = %d AND table_id = %d AND job_name = '%s' AND job_id = %d)",
			tenantId, tableID, jobName, jobIDs[i]))
	}
	sql := buf.String()
	execResult, err := ExecWithResult(ctx, sql, cnUUID, txn)
	if err != nil {
		return
	}
	prevLSNs := make([]uint64, len(jobName))
	for i := range jobName {
		prevLSNs[i] = lsns[i] - 1
	}
	defer execResult.Close()
	jobSpec = make([]*JobSpec, len(jobName))
	prevStatus = make([]*JobStatus, len(jobName))
	execResult.ReadRows(func(rows int, cols []*vector.Vector) bool {
		if rows != len(jobName) {
			errMsg := fmt.Sprintf("invalid rows %d, expected %d", rows, len(jobName))
			FlushPermanentErrorMessage(
				ctx,
				cnUUID,
				cnEngine,
				cnTxnClient,
				tenantId,
				tableID,
				jobName,
				jobIDs,
				lsns,
				jobStatuses,
				types.MaxTs(),
				errMsg,
				prevLSNs,
			)
		}
		for i := 0; i < rows; i++ {
			jobSpec[i], err = UnmarshalJobSpec([]byte(cols[0].GetStringAt(i)))
			if err != nil {
				return false
			}
			prevStatus[i], err = UnmarshalJobStatus([]byte(cols[1].GetStringAt(i)))
			if err != nil {
				return false
			}
		}
		return true
	})
	return
}

Functions

func ExecuteIteration 

func ExecuteIteration(
	ctx context.Context,
	cnUUID string,
	cnEngine engine.Engine,
	cnTxnClient client.TxnClient,
	iterCtx *IterationContext,
	mp *mpool.MPool,
) (err error)

func FlushPermanentErrorMessage 

func FlushPermanentErrorMessage(
	ctx context.Context,
	cnUUID string,
	cnEngine engine.Engine,
	cnTxnClient client.TxnClient,
	accountID uint32,
	tableID uint64,
	jobNames []string,
	jobIDs []uint64,
	lsns []uint64,
	jobStatuses []*JobStatus,
	watermark types.TS,
	errMsg string,
	prevLSN []uint64,
) (err error)

func FlushStatus 

func FlushStatus(
	ctx context.Context,
	cnUUID string,
	txn client.TxnOperator,
	tenantId uint32,
	tableID uint64,
	jobName string,
	jobID uint64,
	jobStatus *JobStatus,
	watermark types.TS,
	state int8,
	prevLSN uint64,
) (err error)

func HasHooks 

func HasHooks(algo string) bool

HasHooks reports whether an algorithm has a registered ISCP hook.

func ISCPTaskExecutorFactory 

func ISCPTaskExecutorFactory(
	txnEngine engine.Engine,
	cnTxnClient client.TxnClient,
	attachToTask func(context.Context, uint64, taskservice.ActiveRoutine) error,
	cdUUID string,
	mp *mpool.MPool,
) func(ctx context.Context, task task.Task) (err error)

func MarshalJobSpec 

func MarshalJobSpec(jobSpec *JobSpec) (string, error)

func MarshalJobStatus 

func MarshalJobStatus(status *JobStatus) (string, error)

func ProcessInitSQL 

func ProcessInitSQL(
	ctx context.Context,
	cnUUID string,
	cnEngine engine.Engine,
	cnTxnClient client.TxnClient,
	sql string,
	dbName string,
	tableName string,
	indexName string,
) (err error)

func Register 

func Register(algo string, h Hooks)

Register installs an ISCP hook for an algorithm. Algo names are case-insensitive (lower-cased on store). Intended to be called from pkg/indexplugin/all init() bodies; panics on duplicate registration.

func RegisterJob 

func RegisterJob(
	ctx context.Context,
	cnUUID string,
	txn client.TxnOperator,
	jobSpec *JobSpec,
	jobID *JobID,
	startFromNow bool,
) (ok bool, err error)

RegisterJob create jobs. return true if create, return false if task already exists, return error when error JobID.DBName: required, the name of the database. JobID.TableName: required, the name of the table. JobID.JobName: required, used to distinguish jobs on the same table. 

Each job on the same table must have a unique job name.
If a duplicate job name is encountered, the returned 'ok' will be false,
and the original job will be kept without creating a new one.

JobSpec.ConsumerInfo.ConsumerType: required, the type of downstream consumer. 

The consumer will be generated according to this type during synchronization.

JobSpec.ConsumerInfo.ColumnNames: optional, currently not effective (TODO). 

Specifies the columns needed by the downstream consumer.
If empty, the consumer will receive all columns of the table.

JobSpec.Priority: optional, the priority for acquiring a worker during execution, currently not effective (TODO). JobSpec.TriggerSpec.JobType: optional, if not set, it will be set to TriggerType_Default. JobSpec.TriggerSpec.Schedule: should be filled according to the triggertype. 

For example, for TriggerType_Default, no need to fill;
for TriggerType_Timed, Interval and share should be specified.

func RenameSrcTable 

func RenameSrcTable(
	ctx context.Context,
	cnUUID string,
	txn client.TxnOperator,
	dbID, tbID uint64,
	oldTableName, newTableName string,
) (err error)

func RunCuvs 

func RunCuvs(c *IndexConsumer, ctx context.Context, errch chan error, r DataRetriever, factory CuvsSyncFactory)

RunCuvs drives one ISCP consumer iteration for cuvs-backed vector indexes (CAGRA, IVF-PQ). It reads raw EncodeEventRecord byte chunks from the consumer's send channel — no JSON marshal/unmarshal — and appends them straight into the sync's pending buffer via AppendRecords. On channel close it flushes via Save and updates the tail watermark.

Used as the Run implementation by CAGRA and IVF-PQ plugin Hooks. HNSW stays on RunHnsw[T] / VectorIndexCdc[T] — different sync architecture (in-memory graph mutation, no event log).

func RunHnsw 

func RunHnsw[T types.RealNumbers](c *IndexConsumer, ctx context.Context, errch chan error, r DataRetriever)

RunHnsw drives one ISCP consumer iteration for HNSW indexes. It reads JSON CDC blobs from the writer's send channel, applies them to an in-memory HnswSync graph via Update, then flushes the model to the hidden tables via Save when the channel closes. Used as the Run implementation by HNSW's plugin Hooks.

The generic parameter T is the vector element type (float32 or float64), matched against the writer type at the call site.

func RunIndex 

func RunIndex(c *IndexConsumer, ctx context.Context, errch chan error, r DataRetriever)

RunIndex drives one ISCP consumer iteration for SQL-based index algorithms (fulltext, IVF-FLAT). It drains SQL statements from the consumer's send channel and executes them against the hidden tables, committing each statement individually for snapshot data and under one long-lived transaction for tail data. Used as the default Run implementation by plugin Hooks whose writer produces SQL rather than a JSON CDC blob.

func UnregisterJob 

func UnregisterJob(
	ctx context.Context,
	cnUUID string,
	txn client.TxnOperator,
	jobID *JobID,
) (ok bool, err error)

return true if delete success, return false if no task found, return error when delete failed.

func UnregisterJobsByDBName 

func UnregisterJobsByDBName(
	ctx context.Context,
	cnUUID string,
	txn client.TxnOperator,
	dbName string,
) (err error)

func UpdateJobSpec 

func UpdateJobSpec(
	ctx context.Context,
	cnUUID string,
	txn client.TxnOperator,
	jobID *JobID,
	jobSpec *JobSpec,
) (err error)

Types

type AtomicBatch 

type AtomicBatch struct {
	Mp      *mpool.MPool
	Batches []*batch.Batch
	Rows    *btree.BTreeG[AtomicBatchRow]
}

AtomicBatch holds batches from [Tail_wip,...,Tail_done] or [Tail_done]. These batches have atomicity

func NewAtomicBatch 

func NewAtomicBatch(mp *mpool.MPool) *AtomicBatch

func (*AtomicBatch) Append 

func (bat *AtomicBatch) Append(
	packer *types.Packer,
	batch *batch.Batch,
	tsColIdx, compositedPkColIdx int,
)

func (*AtomicBatch) Close 

func (bat *AtomicBatch) Close()

func (*AtomicBatch) GetRowIterator 

func (bat *AtomicBatch) GetRowIterator() RowIterator

type AtomicBatchRow 

type AtomicBatchRow struct {
	Ts     types.TS
	Pk     []byte
	Offset int
	Src    *batch.Batch
}

func (AtomicBatchRow) Less 

func (row AtomicBatchRow) Less(other AtomicBatchRow) bool

type BaseIndexSqlWriter 

type BaseIndexSqlWriter struct {
	// contains filtered or unexported fields
}

Base implementation of IVFFLAT and FULLTEXT. Their implementation are simliar.

func (*BaseIndexSqlWriter) CheckLastOp 

func (w *BaseIndexSqlWriter) CheckLastOp(op string) bool

return true when last op is empty or last op == current op

func (*BaseIndexSqlWriter) Delete 

func (w *BaseIndexSqlWriter) Delete(ctx context.Context, row []any) error

func (*BaseIndexSqlWriter) Empty 

func (w *BaseIndexSqlWriter) Empty() bool

func (*BaseIndexSqlWriter) Full 

func (w *BaseIndexSqlWriter) Full() bool

Implementation of Base Index SqlWriter

func (*BaseIndexSqlWriter) Insert 

func (w *BaseIndexSqlWriter) Insert(ctx context.Context, row []any) error

func (*BaseIndexSqlWriter) Reset 

func (w *BaseIndexSqlWriter) Reset()

func (*BaseIndexSqlWriter) Upsert 

func (w *BaseIndexSqlWriter) Upsert(ctx context.Context, row []any) error

type Consumer 

type Consumer interface {
	Consume(context.Context, DataRetriever) error
}

func NewConsumer 

func NewConsumer(
	cnUUID string,
	cnEngine engine.Engine,
	cnTxnClient client.TxnClient,
	tableDef *plan.TableDef,
	jobID JobID,
	info *ConsumerInfo,
) (Consumer, error)

func NewIndexConsumer 

func NewIndexConsumer(cnUUID string,
	cnEngine engine.Engine,
	cnTxnClient client.TxnClient,
	tableDef *plan.TableDef,
	jobID JobID,
	info *ConsumerInfo) (Consumer, error)

func NewInteralSqlConsumer 

func NewInteralSqlConsumer(
	cnUUID string,
	cnEngine engine.Engine,
	cnTxnClient client.TxnClient,
	tableDef *plan.TableDef,
	jobID JobID,
	info *ConsumerInfo,
) (Consumer, error)

type ConsumerInfo 

type ConsumerInfo struct {
	ConsumerType int8
	IndexName    string
	TableName    string
	DBName       string
	Columns      []string
	SrcTable     TableInfo
	InitSQL      string
}

type ConsumerType 

type ConsumerType int8
const (
	ConsumerType_IndexSync ConsumerType = iota
	ConsumerType_CNConsumer

	ConsumerType_CustomizedStart = 1000
)

type CuvsCdcWriter 

type CuvsCdcWriter struct {
	// contains filtered or unexported fields
}

CuvsCdcWriter is the shared CDC writer for cuvs-backed vector indexes (CAGRA, IVF-PQ). One instance per (table, index) pair, constructed once at NewIndexConsumer time.

Buffers CDC row events as a sequence of EncodeEventRecord byte chunks — the same binary format CagraSync.Save / IvfpqSync.Save eventually flushes to the storage table. No JSON, no VectorIndexCdc[T] round-trip.

Despite the "Writer" name (and the IndexSqlWriter interface it satisfies), the output is not SQL — it's the binary event-record stream consumed by RunCuvs → sync.AppendRecords.

INSERT and UPSERT are encoded with distinct op codes (CdcOpInsert / CdcOpUpsert) sharing the same payload layout. Replay handles them identically (idempotent overflow-map write — safe under MO's duplicate / replay-from-corruption UPSERT semantics), but the chunk frame's n_inserts counter tallies only CdcOpInsert so the idxcron gate sees a strict lower bound on growth. This mirrors HNSW's pattern (HnswSqlWriter preserves UPSERT distinctly all the way through). The synchronous in-process callers (CagraSync.Update) still emit DELETE+INSERT pairs as their own thing — that's a different code path from this CDC writer.

func NewCuvsCdcWriter 

func NewCuvsCdcWriter(algoName, dbName, tblName, indexName string,
	tabledef *plan.TableDef, indexdefs []*plan.IndexDef) (*CuvsCdcWriter, error)

NewCuvsCdcWriter constructs a CuvsCdcWriter from the per-(table, index) def. algoName is for diagnostics only (the writer does not switch on it); dbName/tblName/indexName are typically pulled from the ISCP ConsumerInfo at the call site.

Both CAGRA and IVF-PQ on cuvs are fp32-only with a bigint PK; this constructor enforces both shapes.

func (*CuvsCdcWriter) CheckLastOp 

func (w *CuvsCdcWriter) CheckLastOp(_ string) bool

func (*CuvsCdcWriter) ColMetaJSON 

func (w *CuvsCdcWriter) ColMetaJSON() string

func (*CuvsCdcWriter) DbName 

func (w *CuvsCdcWriter) DbName() string

Accessors used by per-algo Hooks.Run impls to build their algo's sync object (CagraSync / IvfpqSync) — passed the same dbName etc. the writer was constructed with so writer and sync agree on layout (especially the colMetaJSON for INCLUDE columns).

func (*CuvsCdcWriter) Delete 

func (w *CuvsCdcWriter) Delete(ctx context.Context, row []any) error

Delete encodes a DELETE event. Only the primary key is consulted — the delete-row payload from ISCP carries just row[0]=pk.

func (*CuvsCdcWriter) Dimension 

func (w *CuvsCdcWriter) Dimension() int32

func (*CuvsCdcWriter) Empty 

func (w *CuvsCdcWriter) Empty() bool

func (*CuvsCdcWriter) Full 

func (w *CuvsCdcWriter) Full() bool

func (*CuvsCdcWriter) IndexDef 

func (w *CuvsCdcWriter) IndexDef() []*plan.IndexDef

func (*CuvsCdcWriter) IndexName 

func (w *CuvsCdcWriter) IndexName() string

func (*CuvsCdcWriter) Insert 

func (w *CuvsCdcWriter) Insert(ctx context.Context, row []any) error

func (*CuvsCdcWriter) Reset 

func (w *CuvsCdcWriter) Reset()

func (*CuvsCdcWriter) TblName 

func (w *CuvsCdcWriter) TblName() string

func (*CuvsCdcWriter) ToSql 

func (w *CuvsCdcWriter) ToSql() ([]byte, error)

ToSql returns a copy of the accumulated event-record bytes. The naming is historical (IndexSqlWriter interface). Caller takes ownership of the returned slice; framing immediately calls Reset() so the writer's internal buffer is reused for the next batch.

func (*CuvsCdcWriter) Upsert 

func (w *CuvsCdcWriter) Upsert(ctx context.Context, row []any) error

Upsert encodes with CdcOpUpsert so the idxcron count gate can ignore it — only true INSERTs are guaranteed-new-row events (see package comment).

type CuvsSync 

type CuvsSync interface {
	AppendRecords(sqlproc *sqlexec.SqlProcess, recordBytes []byte) error
	Save(sqlproc *sqlexec.SqlProcess) error
	Destroy()
}

CuvsSync is the algorithm-side interface that RunCuvs drives. cuvs vector indexes (CAGRA, IVF-PQ) implement this via their per-package CagraSync / IvfpqSync types (under //go:build gpu — pkg/iscp is CPU-safe).

Semantics:

  • AppendRecords appends a writer-produced byte chunk (one or more EncodeEventRecord records concatenated, no JSON wrapping) to the sync's pending buffer.
  • Save flushes the pending records to the storage table as tag=1 event chunks. Called once at channel close.
  • Destroy releases any cuvs-side resources. Called via defer.

type CuvsSyncFactory 

type CuvsSyncFactory func(sqlproc *sqlexec.SqlProcess) (CuvsSync, error)

CuvsSyncFactory constructs a CuvsSync inside the txn provided by the runner. Plugin Hooks.Run impls pass a closure that calls cagra.NewCagraSync / ivfpq.NewIvfpqSync; the factory is what keeps the GPU-tagged types out of pkg/iscp.

type DataRetriever 

type DataRetriever interface {
	Next() (iscpData *ISCPData)
	UpdateWatermark(ctx context.Context, cnUUID string, txn client.TxnOperator) error
	GetDataType() int8
	GetAccountID() uint32
	GetTableID() uint64
}

type DataRetrieverConsumer 

type DataRetrieverConsumer interface {
	DataRetriever
	SetNextBatch(*ISCPData)
	SetError(error)
	Close()
}

type DataRetrieverImpl 

type DataRetrieverImpl struct {
	// contains filtered or unexported fields
}

func NewDataRetriever 

func NewDataRetriever(
	ctx context.Context,
	accountID uint32,
	tableID uint64,
	jobName string,
	jobID uint64,
	status *JobStatus,
	lsn uint64,
	dataType int8,
) *DataRetrieverImpl

func (*DataRetrieverImpl) Close 

func (r *DataRetrieverImpl) Close()

func (*DataRetrieverImpl) GetAccountID 

func (r *DataRetrieverImpl) GetAccountID() uint32

func (*DataRetrieverImpl) GetDataType 

func (r *DataRetrieverImpl) GetDataType() int8

func (*DataRetrieverImpl) GetTableID 

func (r *DataRetrieverImpl) GetTableID() uint64

func (*DataRetrieverImpl) Next 

func (r *DataRetrieverImpl) Next() *ISCPData

func (*DataRetrieverImpl) SetError 

func (r *DataRetrieverImpl) SetError(err error)

after error occurs, the data retriever won't consume any more data

func (*DataRetrieverImpl) SetNextBatch 

func (r *DataRetrieverImpl) SetNextBatch(data *ISCPData)

func (*DataRetrieverImpl) UpdateWatermark 

func (r *DataRetrieverImpl) UpdateWatermark(ctx context.Context,
	cnUUID string,
	txn client.TxnOperator) error

type FulltextSqlWriter 

type FulltextSqlWriter struct {
	BaseIndexSqlWriter
	// contains filtered or unexported fields
}

Fulltext Sql Writer. Only one hidden secondary index table

func (*FulltextSqlWriter) ToSql 

func (w *FulltextSqlWriter) ToSql() ([]byte, error)

with src as (select cast(serial(cast(column_0 as bigint), cast(column_1 as bigint)) as varchar) as id, column_2 as body, column_3 as title from (values row(1, 2, 'body', 'title'), row(2, 3, 'body is heavy', 'I do not know'))) select f.* from src cross apply fulltext_index_tokenize('{"parser":"ngram"}', 61, id, body, title) as f;

type HnswSqlWriter 

type HnswSqlWriter[T types.RealNumbers] struct {
	// contains filtered or unexported fields
}

Hnsw Sql Writer. Use the vectorindex.VectorIndeXCdc JSON format

func (*HnswSqlWriter[T]) CheckLastOp 

func (w *HnswSqlWriter[T]) CheckLastOp(op string) bool

func (*HnswSqlWriter[T]) Delete 

func (w *HnswSqlWriter[T]) Delete(ctx context.Context, row []any) error

func (*HnswSqlWriter[T]) Empty 

func (w *HnswSqlWriter[T]) Empty() bool

func (*HnswSqlWriter[T]) Full 

func (w *HnswSqlWriter[T]) Full() bool

func (*HnswSqlWriter[T]) Insert 

func (w *HnswSqlWriter[T]) Insert(ctx context.Context, row []any) error

func (*HnswSqlWriter[T]) NewSync 

func (w *HnswSqlWriter[T]) NewSync(sqlproc *sqlexec.SqlProcess) (*hnsw.HnswSync[T], error)

func (*HnswSqlWriter[T]) Reset 

func (w *HnswSqlWriter[T]) Reset()

func (*HnswSqlWriter[T]) ToSql 

func (w *HnswSqlWriter[T]) ToSql() ([]byte, error)

func (*HnswSqlWriter[T]) Upsert 

func (w *HnswSqlWriter[T]) Upsert(ctx context.Context, row []any) error

type Hooks 

type Hooks interface {
	// NewSqlWriter constructs the per-(table,index) writer that turns
	// CDC row events into either a SQL statement (fulltext / ivfflat)
	// or a JSON CDC blob (hnsw / cagra / ivfpq). The returned writer
	// satisfies IndexSqlWriter; the caller takes ownership.
	NewSqlWriter(jobID JobID, info *ConsumerInfo,
		tabledef *plan.TableDef, indexdefs []*plan.IndexDef) (IndexSqlWriter, error)

	// Run drives one consumer iteration. SQL-based algorithms delegate
	// to the shared RunIndex helper; HNSW / CAGRA / IVF-PQ run their
	// algorithm-specific Update+Save loop driven by the writer's CDC
	// JSON output. The consumer goroutine calls this once per iteration
	// and the implementation is responsible for draining sqlBufSendCh
	// until close, then updating watermarks (for tail data) and
	// reporting errors via errch.
	Run(c *IndexConsumer, ctx context.Context, errch chan error, r DataRetriever)
}

Hooks is the per-algorithm ISCP hook layer. One implementation per index type (HNSW, IVF-FLAT, fulltext, CAGRA, IVF-PQ). Hooks are registered from pkg/indexplugin/all via Register; pkg/iscp looks them up by algo string from NewIndexConsumer / NewIndexSqlWriter.

Implementations live alongside the algorithm's other plugin hooks at pkg/<algopkg>/plugin/iscp/ (e.g. pkg/vectorindex/hnsw/plugin/iscp/), not in pkg/iscp — keeps algorithm-specific code (and its build-tag constraints, where applicable) out of the ISCP framework.

func GetHooks 

func GetHooks(algo string) (Hooks, bool)

GetHooks returns the Hooks for an algorithm, or (nil, false) if no hook is registered.

type ISCPData 

type ISCPData struct {
	// contains filtered or unexported fields
}

func NewISCPData 

func NewISCPData(
	noMoreData bool,
	insertBatch *AtomicBatch,
	deleteBatch *AtomicBatch,
	err error,
) *ISCPData

func (*ISCPData) Done 

func (d *ISCPData) Done()

func (*ISCPData) Set 

func (d *ISCPData) Set(cnt int)

type ISCPExecutorOption 

type ISCPExecutorOption struct {
	GCInterval             time.Duration
	GCTTL                  time.Duration
	SyncTaskInterval       time.Duration
	FlushWatermarkInterval time.Duration
	FlushWatermarkTTL      time.Duration
	RetryTimes             int
}

type ISCPTaskExecutor 

type ISCPTaskExecutor struct {
	// contains filtered or unexported fields
}

Intra-System Change Propagation Task Executor iscp executor manages iterations, updates watermarks, and updates the iscp table.

func NewISCPTaskExecutor 

func NewISCPTaskExecutor(
	ctx context.Context,
	txnEngine engine.Engine,
	cnTxnClient client.TxnClient,
	cdUUID string,
	option *ISCPExecutorOption,
	mp *mpool.MPool,
) (exec *ISCPTaskExecutor, err error)

func (*ISCPTaskExecutor) Cancel 

func (exec *ISCPTaskExecutor) Cancel() error

func (*ISCPTaskExecutor) FlushWatermarkForAllTables 

func (exec *ISCPTaskExecutor) FlushWatermarkForAllTables(ttl time.Duration) error

func (*ISCPTaskExecutor) GC 

func (exec *ISCPTaskExecutor) GC(cleanupThreshold time.Duration) (err error)

func (*ISCPTaskExecutor) GCInMemoryJob 

func (exec *ISCPTaskExecutor) GCInMemoryJob(threshold time.Duration)

func (*ISCPTaskExecutor) GetJobType 

func (exec *ISCPTaskExecutor) GetJobType(accountID uint32, srcTableID uint64, jobName string) (jobType uint16, ok bool)

For UT

func (*ISCPTaskExecutor) GetWatermark 

func (exec *ISCPTaskExecutor) GetWatermark(accountID uint32, srcTableID uint64, jobName string) (watermark types.TS, ok bool)

For UT

func (*ISCPTaskExecutor) IsRunning 

func (exec *ISCPTaskExecutor) IsRunning() bool

func (*ISCPTaskExecutor) Pause 

func (exec *ISCPTaskExecutor) Pause() error

func (*ISCPTaskExecutor) Restart 

func (exec *ISCPTaskExecutor) Restart() error

func (*ISCPTaskExecutor) Resume 

func (exec *ISCPTaskExecutor) Resume() error

func (*ISCPTaskExecutor) SetRpcHandleFn 

func (exec *ISCPTaskExecutor) SetRpcHandleFn(fn RpcHandleFn)

func (*ISCPTaskExecutor) Start 

func (exec *ISCPTaskExecutor) Start() error

func (*ISCPTaskExecutor) Stop 

func (exec *ISCPTaskExecutor) Stop()

func (*ISCPTaskExecutor) String 

func (exec *ISCPTaskExecutor) String() string

type IndexConsumer 

type IndexConsumer struct {
	// contains filtered or unexported fields
}

IndexConsumer

func (*IndexConsumer) Algo 

func (c *IndexConsumer) Algo() string

Algo returns the algorithm string this consumer's writer was built for. Useful for diagnostics inside plugin Hooks.

func (*IndexConsumer) Consume 

func (c *IndexConsumer) Consume(ctx context.Context, r DataRetriever) error

func (*IndexConsumer) RunTxn 

func (c *IndexConsumer) RunTxn(
	ctx context.Context,
	r DataRetriever,
	timeout time.Duration,
	cb func(sqlproc *sqlexec.SqlProcess) error,
) error

RunTxn wraps sqlexec.RunTxnWithSqlContext bound to this consumer's connection plumbing (engine, txn client, CN UUID) and the retriever's account ID. Plugin Hooks.Run impls living outside pkg/iscp use this to execute SQL / drive a sync object inside a properly scoped txn without needing access to the consumer's private fields.

func (*IndexConsumer) SqlBufSendCh 

func (c *IndexConsumer) SqlBufSendCh() <-chan []byte

SqlBufSendCh exposes the channel on which the framing layer delivers writer SQL / CDC JSON blobs. Plugin Hooks.Run impls read from this channel and stop when it closes (signalling end-of-iteration).

func (*IndexConsumer) SqlWriter 

func (c *IndexConsumer) SqlWriter() IndexSqlWriter

SqlWriter returns the writer this consumer is paired with. Used by plugin Hooks.Run impls that need to dispatch on the writer's concrete type (e.g. HNSW picking RunHnsw[float32] vs [float64]).

type IndexEntry 

type IndexEntry struct {
	// contains filtered or unexported fields
}

IndexConsumer

type IndexSqlWriter 

type IndexSqlWriter interface {
	CheckLastOp(op string) bool
	Upsert(ctx context.Context, row []any) error
	Insert(ctx context.Context, row []any) error
	Delete(ctx context.Context, row []any) error
	Full() bool
	ToSql() ([]byte, error)
	Reset()
	Empty() bool
}

IndexSqlWriter interface

func NewFulltextSqlWriter 

func NewFulltextSqlWriter(algo string, jobID JobID, info *ConsumerInfo, tabledef *plan.TableDef, indexdef []*plan.IndexDef) (IndexSqlWriter, error)

New Fulltext Sql Writer

func NewGenericHnswSqlWriter 

func NewGenericHnswSqlWriter[T types.RealNumbers](algo string, jobID JobID, info *ConsumerInfo, tabledef *plan.TableDef, indexdef []*plan.IndexDef) (IndexSqlWriter, error)

func NewHnswSqlWriter 

func NewHnswSqlWriter(algo string, jobID JobID, info *ConsumerInfo, tabledef *plan.TableDef, indexdef []*plan.IndexDef) (IndexSqlWriter, error)

Implementation of HNSW Sql writer

func NewIndexSqlWriter 

func NewIndexSqlWriter(algo string, jobID JobID, info *ConsumerInfo, tabledef *plan.TableDef, indexdef []*plan.IndexDef) (IndexSqlWriter, error)

NewIndexSqlWriter dispatches to the per-algorithm writer via the iscp Hooks registry. Replaces the hardcoded fulltext / ivfflat / hnsw switch — new algorithms register a Hooks impl (see pkg/sql/compile/iscp_register.go) and slot in automatically.

func NewIvfflatSqlWriter 

func NewIvfflatSqlWriter(algo string, jobID JobID, info *ConsumerInfo, tabledef *plan.TableDef, indexdef []*plan.IndexDef) (IndexSqlWriter, error)

Implementation of Ivfflat Sql writer

type InitWatermark 

type InitWatermark struct {
	AccountID uint32
	TableID   uint64
	JobName   string
	JobID     uint64
	Watermark types.TS
}

type IterationContext 

type IterationContext struct {
	// contains filtered or unexported fields
}

func NewIterationContext 

func NewIterationContext(accountID uint32, tableID uint64, jobNames []string, jobIDs []uint64, lsn []uint64, fromTS types.TS, toTS types.TS) *IterationContext

func (*IterationContext) String 

func (iterCtx *IterationContext) String() string

type IvfflatSqlWriter 

type IvfflatSqlWriter struct {
	BaseIndexSqlWriter
	// contains filtered or unexported fields
}

Ivfflat Sql writer. Three hidden secondary index tables

func (*IvfflatSqlWriter) ToSql 

func (w *IvfflatSqlWriter) ToSql() ([]byte, error)

REPLACE INTO __mo_index_secondary_0197786c-285f-70cb-9337-e484a3ff92c4(__mo_index_centroid_fk_version, __mo_index_centroid_fk_id, __mo_index_pri_col, __mo_index_centroid_fk_entry) with centroid as (select * from __mo_index_secondary_0197786c-285f-70bb-b277-2cef56da590a where __mo_index_centroid_version = 0), src as (select column_0 as id, cast(column_1 as vecf32(3)) as embed from (values row(2005,'[0.4532634, 0.7297859, 0.48885703]'), row(2009, '[0.68150306, 0.6950923, 0.16590895] '))) select __mo_index_centroid_version, __mo_index_centroid_id, id, embed from src centroidx('vector_l2_ops') join centroid using (__mo_index_centroid, embed);

type JobEntry 

type JobEntry struct {
	// contains filtered or unexported fields
}

Intra-System Change Propagation Job Entry

func NewJobEntry 

func NewJobEntry(
	tableInfo *TableEntry,
	jobName string,
	jobSpec *JobSpec,
	jobID uint64,
	watermark types.TS,
	state int8,
	dropAt types.Timestamp,
) *JobEntry

func (*JobEntry) IsInitedAndFinished 

func (jobEntry *JobEntry) IsInitedAndFinished() bool

func (*JobEntry) StringLocked 

func (jobEntry *JobEntry) StringLocked() string

func (*JobEntry) UpdateWatermark 

func (jobEntry *JobEntry) UpdateWatermark(
	from, to types.TS,
	watermarkFlushThreshold time.Duration,
)

type JobID 

type JobID struct {
	DBName    string
	TableName string
	JobName   string
}

type JobKey 

type JobKey struct {
	JobName string
	JobID   uint64
}

type JobSpec 

type JobSpec struct {
	Priority int8
	ConsumerInfo
	TriggerSpec
}

func UnmarshalJobSpec 

func UnmarshalJobSpec(jsonByte []byte) (*JobSpec, error)

type JobStatus 

type JobStatus struct {
	LSN       uint64
	Stage     int8
	TaskID    uint64
	From      types.TS
	To        types.TS
	StartAt   types.TS
	EndAt     types.TS
	ErrorCode int
	ErrorMsg  string
}

func UnmarshalJobStatus 

func UnmarshalJobStatus(jsonByte []byte) (*JobStatus, error)

func (*JobStatus) PermanentlyFailed 

func (status *JobStatus) PermanentlyFailed() bool

func (*JobStatus) SetError 

func (status *JobStatus) SetError(err error)

type OutputType 

type OutputType int8
const (
	OutputTypeSnapshot OutputType = iota
	OutputTypeTail
)

func (OutputType) String 

func (t OutputType) String() string

type RowIterator 

type RowIterator interface {
	Next() bool
	Row(ctx context.Context, row []any) error
	Close()
}

type RowType 

type RowType int
const (
	NoOp RowType = iota
	InsertRow
	DeleteRow
	UpsertRow
)

type RpcHandleFn 

type RpcHandleFn func(
	ctx context.Context,
	meta txn.TxnMeta,
	req *cmd_util.GetChangedTableListReq,
	resp *cmd_util.GetChangedTableListResp,
) (func(), error)

type Schedule 

type Schedule struct {
	Interval time.Duration
	Share    bool
}

type TableEntry 

type TableEntry struct {
	// contains filtered or unexported fields
}

Intra-System Change Propagation Table Entry

func NewTableEntry 

func NewTableEntry(
	exec *ISCPTaskExecutor,
	accountID uint32,
	dbID, tableID uint64,
	dbName, tableName string,
) *TableEntry

func (*TableEntry) AddOrUpdateSinker 

func (t *TableEntry) AddOrUpdateSinker(
	ctx context.Context,
	jobName string,
	jobSpec *JobSpec,
	jobStatus *JobStatus,
	jobID uint64,
	watermark types.TS,
	state int8,
	dropAt types.Timestamp,
) (newCreate bool)

func (*TableEntry) GetWatermark 

func (t *TableEntry) GetWatermark(jobName string) (watermark types.TS, ok bool)

for UT

func (*TableEntry) IsEmpty 

func (t *TableEntry) IsEmpty() bool

func (*TableEntry) String 

func (t *TableEntry) String() string

func (*TableEntry) UpdateWatermark 

func (t *TableEntry) UpdateWatermark(iter *IterationContext)

type TableInfo 

type TableInfo struct {
	DBName    string
	DBID      uint64
	TableName string
	TableID   uint64
}

type TriggerSpec 

type TriggerSpec struct {
	JobType uint16
	Schedule
}

func (*TriggerSpec) Check 

func (triggerSpec *TriggerSpec) Check(
	otherConsumers []*JobEntry,
	consumer *JobEntry,
	now types.TS,
) (
	ok bool, from, to types.TS, shareIteration bool,
)

func (*TriggerSpec) GetType 

func (triggerSpec *TriggerSpec) GetType() uint16

func (*TriggerSpec) Init 

func (triggerSpec *TriggerSpec) Init()

type Worker 

type Worker interface {
	Submit(iteration *IterationContext) error
	Stop()
}

func NewWorker 

func NewWorker(cnUUID string, cnEngine engine.Engine, cnTxnClient client.TxnClient, mp *mpool.MPool) Worker

Source Files

View all Source files

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.