HNSW index update with CDC

pkg/cdc/sinker.go

@@ -19,11 +19,12 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
-	"go.uber.org/zap"
 	"strings"
 	"sync/atomic"
 	"time"
 
+	"go.uber.org/zap"
+
 	"github.com/matrixorigin/matrixone/pkg/catalog"
 	"github.com/matrixorigin/matrixone/pkg/common/moerr"
 	"github.com/matrixorigin/matrixone/pkg/common/mpool"
@@ -53,6 +54,7 @@ var (
 )
 
 var NewSinker = func(
+	cnUUID string,
 	sinkUri UriInfo,
 	dbTblInfo *DbTableInfo,
 	watermarkUpdater IWatermarkUpdater,
@@ -68,6 +70,10 @@ var NewSinker = func(
 		return NewConsoleSinker(dbTblInfo, watermarkUpdater), nil
 	}
 
+	if sinkUri.SinkTyp == CDCSinkType_HnswSync {
+		return NewHnswSyncSinker(cnUUID, sinkUri, dbTblInfo, watermarkUpdater, tableDef, retryTimes, retryDuration, ar, maxSqlLength, sendSqlTimeout)
+	}
+
 	var (
 		err  error
 		sink Sink
@@ -445,7 +451,8 @@ func (s *mysqlSinker) SendDummy() {
 }
 
 func (s *mysqlSinker) Error() error {
-	if errPtr := s.err.Load().(*error); *errPtr != nil {
+	if ptr := s.err.Load(); ptr != nil {
+		errPtr := ptr.(*error)
 		if moErr, ok := (*errPtr).(*moerr.Error); !ok {
 			return moerr.ConvertGoError(context.Background(), *errPtr)
 		} else {

pkg/cdc/sinker_test.go

@@ -119,7 +119,7 @@ func TestNewSinker(t *testing.T) {
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got, err := NewSinker(tt.args.sinkUri, tt.args.dbTblInfo, tt.args.watermarkUpdater, tt.args.tableDef, tt.args.retryTimes, tt.args.retryDuration, tt.args.ar, CDCDefaultTaskExtra_MaxSQLLen, CDCDefaultSendSqlTimeout)
+			got, err := NewSinker("", tt.args.sinkUri, tt.args.dbTblInfo, tt.args.watermarkUpdater, tt.args.tableDef, tt.args.retryTimes, tt.args.retryDuration, tt.args.ar, CDCDefaultTaskExtra_MaxSQLLen, CDCDefaultSendSqlTimeout)
 			if !tt.wantErr(t, err, fmt.Sprintf("NewSinker(%v, %v, %v, %v, %v, %v)", tt.args.sinkUri, tt.args.dbTblInfo, tt.args.watermarkUpdater, tt.args.tableDef, tt.args.retryTimes, tt.args.retryDuration)) {
 				return
 			}

pkg/cdc/types.go

@@ -52,9 +52,10 @@ const (
 )
 
 const (
-	CDCSinkType_MySQL   = "mysql"
-	CDCSinkType_MO      = "matrixone"
-	CDCSinkType_Console = "console"
+	CDCSinkType_MySQL    = "mysql"
+	CDCSinkType_MO       = "matrixone"
+	CDCSinkType_Console  = "console"
+	CDCSinkType_HnswSync = "hnswsync"
 )
 
 const (

pkg/frontend/cdc_exector.go

@@ -451,6 +451,7 @@ func (exec *CDCTaskExecutor) addExecPipelineForTable(ctx context.Context, info *
 
 	// step 2. new sinker
 	sinker, err := cdc.NewSinker(
+		exec.cnUUID,
 		exec.sinkUri,
 		info,
 		exec.watermarkUpdater,

pkg/frontend/cdc_options.go

@@ -123,7 +123,7 @@ func (opts *CDCCreateTaskOptions) ValidateAndFill(
 		if cdc.EnableConsoleSink && opts.SinkType == cdc.CDCSinkType_Console {
 			opts.UseConsole = true
 		}
-		if !opts.UseConsole && opts.SinkType != cdc.CDCSinkType_MySQL && opts.SinkType != cdc.CDCSinkType_MO {
+		if !opts.UseConsole && opts.SinkType != cdc.CDCSinkType_MySQL && opts.SinkType != cdc.CDCSinkType_MO && opts.SinkType != cdc.CDCSinkType_HnswSync {
 			err = moerr.NewInternalErrorf(ctx, "unsupported sink type: %s", req.SinkType)
 			return
 		}

pkg/frontend/cdc_test.go

@@ -2891,7 +2891,7 @@ func TestCdcTask_addExecPipelineForTable(t *testing.T) {
 	})
 	defer stubGetTableDef.Reset()
 
-	stubSinker := gostub.Stub(&cdc.NewSinker, func(cdc.UriInfo, *cdc.DbTableInfo, cdc.IWatermarkUpdater,
+	stubSinker := gostub.Stub(&cdc.NewSinker, func(string, cdc.UriInfo, *cdc.DbTableInfo, cdc.IWatermarkUpdater,
 		*plan.TableDef, int, time.Duration, *cdc.ActiveRoutine, uint64, string) (cdc.Sinker, error) {
 		return &mockSinker{}, nil
 	})

pkg/sql/compile/ddl.go

@@ -2108,6 +2108,13 @@ func (s *Scope) DropIndex(c *Compile) error {
 	if err != nil {
 		return err
 	}
+
+	// TODO: HNSWCDC delete cdc task for vector, fulltext index
+	// cdc task name = __mo_cdc_{qry.Database}_{qry.Table}_{qry.IndexName}
+	// pitr name = __mo_table_pitr_{qry.Database}_{qry.Table}
+	// DROP PITR IF EXISTS `__mo_table_pitr_${qry.Database}_${srctable}`
+	// DROP CDC TASK __mo_cdc_${qry.Database}_${srctable}_${qry.IndexName}
+
 	return nil
 }
 
@@ -2668,6 +2675,8 @@ func (s *Scope) DropTable(c *Compile) error {
 		}
 	}
 
+	// TODO: HSNWCDC delete cdc task of the vector and fulltext index here
+
 	if isTemp {
 		if err := dbSource.Delete(c.proc.Ctx, engine.GetTempTableName(dbName, tblName)); err != nil {
 			return err

pkg/sql/compile/ddl_index_algo.go

@@ -556,5 +556,18 @@ func (s *Scope) handleVectorHnswIndex(
 		}
 	}
 
+	// TODO: HNSWCDC 4. register CDC update
+	sqls, err = genCdcHnswIndex(c.proc, indexDefs, qryDatabase, originalTableDef)
+	if err != nil {
+		return err
+	}
+
+	for _, sql := range sqls {
+		err = c.runSql(sql)
+		if err != nil {
+			return err
+		}
+	}
+
 	return nil
 }

pkg/sql/compile/util.go

@@ -561,3 +561,41 @@ func genBuildHnswIndex(proc *process.Process, indexDefs map[string]*plan.IndexDe
 
 	return []string{sql}, nil
 }
+
+// TODO: HNSWCDC 4. register CDC update
+// DROP PITR IF EXISTS `__mo_table_pitr_${db}_${srctable}`
+// DROP CDC IF EXISTS TASK __mo_cdc_${db}_${srctable}_${indexInfo.IndexName} NOTE: IF EXISTS is not valid SQL for DROP CDC
+// CREATE PITR __mo_table_pitr_${db}_${srctable} for table ${db} ${srctable) range 2 'h';
+// CREATE CDC __mo_cdc_${db}_${srctable}_${indexInfo.IndexName} 'mysql://root:111@127.0.0.1:6001' 'hnswsync' 'mysql://root:111@127.0.0.1:6001' '${db}.${srctable}' {'Level'='table'}'
+func genCdcHnswIndex(proc *process.Process, indexDefs map[string]*plan.IndexDef, qryDatabase string, originalTableDef *plan.TableDef) ([]string, error) {
+
+	idxdef_meta, ok := indexDefs[catalog.Hnsw_TblType_Metadata]
+	if !ok {
+		return nil, moerr.NewInternalErrorNoCtx("hnsw_meta index definition not found")
+	}
+	srctbl := originalTableDef.Name
+	pitrname := fmt.Sprintf("__mo_index_pitr_%s_%s", qryDatabase, srctbl)
+	cdcname := fmt.Sprintf("__mo_index_cdc_%s_%s_%s", qryDatabase, srctbl, idxdef_meta.IndexName)
+
+	var sql string
+
+	sqls := make([]string, 0, 3)
+
+	// CREATE PITR
+	sql = fmt.Sprintf("DROP PITR IF EXISTS `%s`;", pitrname)
+	sqls = append(sqls, sql)
+
+	sql = fmt.Sprintf("CREATE PITR `%s` FOR TABLE `%s` `%s` range 2 'h';", pitrname, qryDatabase, srctbl)
+	sqls = append(sqls, sql)
+
+	// CREATE CDC TASK
+	dummyurl := "mysql://root:111@127.0.0.1:6001"
+	sql = fmt.Sprintf("CREATE CDC `%s` '%s' 'hnswsync' '%s' '%s.%s' {'Level'='table'};", cdcname, dummyurl, dummyurl, qryDatabase, srctbl)
+	sqls = append(sqls, sql)
+
+	//os.Stderr.WriteString(fmt.Sprintf("%v\n", sqls))
+	// TODO: HNSWCDC remove the line below to run the above SQLs
+	sqls = sqls[:0]
+
+	return sqls, nil
+}

pkg/sql/plan/function/func_cast.go

@@ -4408,6 +4408,8 @@ func strToArray[T types.RealNumbers](
 	from vector.FunctionParameterWrapper[types.Varlena],
 	to *vector.FunctionResult[types.Varlena], length int, _ types.Type) error {
 
+	toType := to.GetType()
+
 	var i uint64
 	var l = uint64(length)
 	for i = 0; i < l; i++ {
@@ -4417,11 +4419,18 @@ func strToArray[T types.RealNumbers](
 				return err
 			}
 		} else {
-
-			b, err := types.StringToArrayToBytes[T](convertByteSliceToString(v))
+			arr, err := types.StringToArray[T](convertByteSliceToString(v))
 			if err != nil {
 				return err
 			}
+
+			// bypass the dimension check if width is max dimension
+			if int(toType.Width) != types.MaxArrayDimension && int(toType.Width) != len(arr) {
+				return moerr.NewArrayDefMismatchNoCtx(int(toType.Width), len(arr))
+			}
+
+			b := types.ArrayToBytes[T](arr)
+
 			if err = to.AppendBytes(b, false); err != nil {
 				return err
 			}

pkg/sql/plan/function/func_hnsw.go

@@ -0,0 +1,77 @@
+// Copyright 2021 - 2022 Matrix Origin
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package function
+
+import (
+	"encoding/json"
+
+	"github.com/matrixorigin/matrixone/pkg/common/moerr"
+	"github.com/matrixorigin/matrixone/pkg/container/vector"
+	"github.com/matrixorigin/matrixone/pkg/logutil"
+	"github.com/matrixorigin/matrixone/pkg/vectorindex"
+	"github.com/matrixorigin/matrixone/pkg/vectorindex/hnsw"
+	"github.com/matrixorigin/matrixone/pkg/vm/process"
+)
+
+func hnswCdcUpdate(ivecs []*vector.Vector, result vector.FunctionResultWrapper, proc *process.Process, length int, selectList *FunctionSelectList) error {
+
+	if len(ivecs) != 4 {
+		return moerr.NewInvalidInput(proc.Ctx, "number of arguments != 4")
+	}
+
+	dbVec := vector.GenerateFunctionStrParameter(ivecs[0])
+	tblVec := vector.GenerateFunctionStrParameter(ivecs[1])
+	dimVec := vector.GenerateFunctionFixedTypeParameter[int32](ivecs[2])
+	cdcVec := vector.GenerateFunctionStrParameter(ivecs[3])
+
+	for i := uint64(0); i < uint64(length); i++ {
+		dbname, isnull := dbVec.GetStrValue(i)
+		if isnull {
+			return moerr.NewInvalidInput(proc.Ctx, "dbname is null")
+		}
+
+		tblname, isnull := tblVec.GetStrValue(i)
+		if isnull {
+			return moerr.NewInvalidInput(proc.Ctx, "table name is null")
+
+		}
+
+		dim, isnull := dimVec.GetValue(i)
+		if isnull {
+			return moerr.NewInvalidInput(proc.Ctx, "dimension is null")
+		}
+
+		cdcstr, isnull := cdcVec.GetStrValue(i)
+		if isnull {
+			return moerr.NewInvalidInput(proc.Ctx, "cdc is null")
+		}
+
+		var cdc vectorindex.VectorIndexCdc[float32]
+		err := json.Unmarshal([]byte(cdcstr), &cdc)
+		if err != nil {
+			return moerr.NewInvalidInput(proc.Ctx, "cdc is not json object")
+		}
+		logutil.Infof("hnsw_cdc_update: START db=%s, table=%s\n", dbname, tblname)
+		// hnsw sync
+		//os.Stderr.WriteString(fmt.Sprintf("db=%s, table=%s, dim=%d, json=%s\n", dbname, tblname, dim, cdcstr))
+		err = hnsw.CdcSync(proc, string(dbname), string(tblname), dim, &cdc)
+		if err != nil {
+			return err
+		}
+		logutil.Infof("hnsw_cdc_update: END db=%s, table=%s\n", dbname, tblname)
+	}
+
+	return nil
+}