IGNOREME: Iscp integration

pkg/iscp/index_sqlwriter.go

@@ -339,8 +339,9 @@ func NewGenericHnswSqlWriter[T types.RealNumbers](algo string, jobID JobID, info
 
 	// get the first indexdef as they are the same
 	idxdef := indexdef[0]
+	writer_capacity := 8192
 
-	w := &HnswSqlWriter[T]{tabledef: tabledef, indexdef: indexdef, jobID: jobID, info: info, cdc: vectorindex.NewVectorIndexCdc[T]()}
+	w := &HnswSqlWriter[T]{tabledef: tabledef, indexdef: indexdef, jobID: jobID, info: info, cdc: vectorindex.NewVectorIndexCdc[T](writer_capacity)}
 
 	paramstr := idxdef.IndexAlgoParams
 	var meta, storage string

pkg/sql/compile/alter.go

@@ -49,6 +49,7 @@ func convertDBEOB(ctx context.Context, e error, name string) error {
 func (s *Scope) AlterTableCopy(c *Compile) error {
 	qry := s.Plan.GetDdl().GetAlterTable()
 	dbName := qry.Database
+
 	if dbName == "" {
 		dbName = c.db
 	}
@@ -149,25 +150,34 @@ func (s *Scope) AlterTableCopy(c *Compile) error {
 	if qry.Options.SkipPkDedup || len(qry.Options.SkipUniqueIdxDedup) > 0 {
 		opt = opt.WithAlterCopyOpt(qry.Options)
 	}
-	// 4. copy the original table data to the temporary replica table
-	err = c.runSqlWithOptions(qry.InsertTmpDataSql, opt)
+
+	//4. obtain relation for new tables
+	newRel, err := dbSource.Relation(c.proc.Ctx, qry.CopyTableDef.Name, nil)
 	if err != nil {
-		c.proc.Error(c.proc.Ctx, "insert data to copy table for alter table",
+		c.proc.Error(c.proc.Ctx, "obtain new relation for copy table for alter table",
 			zap.String("databaseName", dbName),
 			zap.String("origin tableName", qry.GetTableDef().Name),
-			zap.String("copy tableName", qry.CopyTableDef.Name),
-			zap.String("InsertTmpDataSql", qry.InsertTmpDataSql),
+			zap.String("copy table name", qry.CopyTableDef.Name),
 			zap.Error(err))
 		return err
 	}
 
-	//5. obtain relation for new tables
-	newRel, err := dbSource.Relation(c.proc.Ctx, qry.CopyTableDef.Name, nil)
+	//5. ISCP: temp table already created pitr and iscp job with temp table name
+	// and we don't want iscp to run with temp table so drop pitr and iscp job with the temp table here
+	newTmpTableDef := newRel.CopyTableDef(c.proc.Ctx)
+	err = DropAllIndexCdcTasks(c, newTmpTableDef, dbName, qry.CopyTableDef.Name)
 	if err != nil {
-		c.proc.Error(c.proc.Ctx, "obtain new relation for copy table for alter table",
+		return err
+	}
+
+	// 6. copy the original table data to the temporary replica table
+	err = c.runSqlWithOptions(qry.InsertTmpDataSql, opt)
+	if err != nil {
+		c.proc.Error(c.proc.Ctx, "insert data to copy table for alter table",
 			zap.String("databaseName", dbName),
 			zap.String("origin tableName", qry.GetTableDef().Name),
-			zap.String("copy table name", qry.CopyTableDef.Name),
+			zap.String("copy tableName", qry.CopyTableDef.Name),
+			zap.String("InsertTmpDataSql", qry.InsertTmpDataSql),
 			zap.Error(err))
 		return err
 	}
@@ -179,7 +189,8 @@ func (s *Scope) AlterTableCopy(c *Compile) error {
 		return err
 	}
 
-	// 7. drop original table
+	// 7. drop original table.
+	// ISCP: That will also drop ISCP related jobs and pitr of the original table.
 	dropSql := fmt.Sprintf("drop table `%s`.`%s`", dbName, tblName)
 	if err := c.runSqlWithOptions(
 		dropSql,
@@ -193,26 +204,6 @@ func (s *Scope) AlterTableCopy(c *Compile) error {
 		return err
 	}
 
-	// 7.1 delete all index objects of the table in mo_catalog.mo_indexes
-	if qry.Database != catalog.MO_CATALOG && qry.TableDef.Name != catalog.MO_INDEXES {
-		if qry.GetTableDef().Pkey != nil || len(qry.GetTableDef().Indexes) > 0 {
-			deleteSql := fmt.Sprintf(
-				deleteMoIndexesWithTableIdFormat,
-				qry.GetTableDef().TblId,
-			)
-			err = c.runSql(deleteSql)
-			if err != nil {
-				c.proc.Error(c.proc.Ctx, "delete all index meta data of origin table in `mo_indexes` for alter table",
-					zap.String("databaseName", dbName),
-					zap.String("origin tableName", qry.GetTableDef().Name),
-					zap.String("delete all index sql", deleteSql),
-					zap.Error(err))
-
-				return err
-			}
-		}
-	}
-
 	newId := newRel.GetTableID(c.proc.Ctx)
 	//-------------------------------------------------------------------------
 	// 8. rename temporary replica table into the original table(Table Id remains unchanged)
@@ -237,11 +228,11 @@ func (s *Scope) AlterTableCopy(c *Compile) error {
 		return err
 	}
 
+	newTableDef := newRel.CopyTableDef(c.proc.Ctx)
 	//--------------------------------------------------------------------------------------------------------------
 	{
 		// 9. invoke reindex for the new table, if it contains ivf index.
 		multiTableIndexes := make(map[string]*MultiTableIndex)
-		newTableDef := newRel.CopyTableDef(c.proc.Ctx)
 		extra := newRel.GetExtraInfo()
 		id := newRel.GetTableID(c.proc.Ctx)
 
@@ -311,6 +302,12 @@ func (s *Scope) AlterTableCopy(c *Compile) error {
 		}
 	}
 
+	// 10. register ISCP job again
+	err = CreateAllIndexCdcTasks(c, newTableDef.Indexes, dbName, tblName)
+	if err != nil {
+		return err
+	}
+
 	// get and update the change mapping information of table colIds
 	if err = updateNewTableColId(c, newRel, qry.ChangeTblColIdMap); err != nil {
 		c.proc.Error(c.proc.Ctx, "get and update the change mapping information of table colIds for alter table",

pkg/sql/compile/ddl.go

@@ -27,6 +27,7 @@ import (
 
 	moruntime "github.com/matrixorigin/matrixone/pkg/common/runtime"
 	"github.com/matrixorigin/matrixone/pkg/config"
+	"github.com/matrixorigin/matrixone/pkg/iscp"
 	"github.com/matrixorigin/matrixone/pkg/pb/task"
 
 	"github.com/matrixorigin/matrixone/pkg/cdc"
@@ -239,6 +240,13 @@ func (s *Scope) DropDatabase(c *Compile) error {
 			return err
 		}
 	}
+
+	// 5.unregister iscp jobs
+	err = iscp.UnregisterJobsByDBName(c.proc.Ctx, c.proc.GetService(), c.proc.GetTxnOperator(), dbName)
+	if err != nil {
+		return err
+	}
+
 	return err
 }
 
@@ -589,6 +597,13 @@ func (s *Scope) AlterTableInplace(c *Compile) error {
 						newIndexes = append(newIndexes, extra.IndexTables[idx])
 					}
 				}
+
+				// drop index cdc task
+				err = DropIndexCdcTask(c, oTableDef, dbName, tblName, constraintName)
+				if err != nil {
+					return err
+				}
+
 				// Avoid modifying slice directly during iteration
 				oTableDef.Indexes = notDroppedIndex
 				extra.IndexTables = newIndexes
@@ -746,9 +761,8 @@ func (s *Scope) AlterTableInplace(c *Compile) error {
 						if err != nil {
 							return err
 						}
-
 					case catalog.MoIndexHnswAlgo.ToString():
-						// PASS: keep option unchange for incremental update
+						// PASS
 					default:
 						return moerr.NewInternalError(c.proc.Ctx, "invalid index algo type for alter reindex")
 					}
@@ -890,6 +904,25 @@ func (s *Scope) AlterTableInplace(c *Compile) error {
 		return err
 	}
 
+	// post alter table rename -- AlterKind_RenameTable to update iscp job
+	for _, req := range reqs {
+		if req.Kind == api.AlterKind_RenameTable {
+			op, ok := req.Operation.(*api.AlterTableReq_RenameTable)
+			if ok {
+				err = iscp.RenameSrcTable(c.proc.Ctx,
+					c.proc.GetService(),
+					c.proc.GetTxnOperator(),
+					req.DbId,
+					req.TableId,
+					op.RenameTable.OldName,
+					op.RenameTable.NewName)
+				if err != nil {
+					return err
+				}
+			}
+		}
+	}
+
 	// remove refChildTbls for drop foreign key clause
 	//remove the child table id -- tblId from the parent table -- fkTblId
 	for _, fkTblId := range removeRefChildTbls {
@@ -1468,6 +1501,21 @@ func (s *Scope) CreateTable(c *Compile) error {
 			)
 			return err
 		}
+
+		// create iscp jobs for index async update
+		ct, err := GetConstraintDef(c.proc.Ctx, newRelation)
+		if err != nil {
+			return err
+		}
+		for _, constraint := range ct.Cts {
+			if idxdef, ok := constraint.(*engine.IndexDef); ok && len(idxdef.Indexes) > 0 {
+				err = CreateAllIndexCdcTasks(c, idxdef.Indexes, dbName, tblName)
+				if err != nil {
+					return err
+				}
+			}
+		}
+
 	}
 
 	if c.keepAutoIncrement == 0 {
@@ -2078,6 +2126,21 @@ func (s *Scope) handleVectorIvfFlatIndex(
 		return err
 	}
 
+	async, err := catalog.IsIndexAsync(indexDefs[catalog.SystemSI_IVFFLAT_TblType_Metadata].IndexAlgoParams)
+	if err != nil {
+		return err
+	}
+
+	// create ISCP job when Async is true
+	if async {
+		logutil.Infof("Ivfflat index Async is true")
+		sinker_type := getSinkerTypeFromAlgo(catalog.MoIndexIvfFlatAlgo.ToString())
+		err = CreateIndexCdcTask(c, qryDatabase, originalTableDef.Name,
+			indexDefs[catalog.SystemSI_IVFFLAT_TblType_Metadata].IndexName, sinker_type)
+		if err != nil {
+			return err
+		}
+	}
 	return nil
 
 }
@@ -2102,6 +2165,9 @@ func (s *Scope) DropIndex(c *Compile) error {
 		return err
 	}
 
+	// old tabledef
+	oldTableDef := r.GetTableDef(c.proc.Ctx)
+
 	//1. build and update constraint def
 	oldCt, err := GetConstraintDef(c.proc.Ctx, r)
 	if err != nil {
@@ -2130,14 +2196,20 @@ func (s *Scope) DropIndex(c *Compile) error {
 			return err
 		}
 
+		//3. delete iscp job for vector, fulltext index
+		err = DropIndexCdcTask(c, oldTableDef, qry.Database, qry.Table, qry.IndexName)
+		if err != nil {
+			return err
+		}
 	}
 
-	//3. delete index object from mo_catalog.mo_indexes
+	//4. delete index object from mo_catalog.mo_indexes
 	deleteSql := fmt.Sprintf(deleteMoIndexesWithTableIdAndIndexNameFormat, r.GetTableID(c.proc.Ctx), qry.IndexName)
 	err = c.runSql(deleteSql)
 	if err != nil {
 		return err
 	}
+
 	return nil
 }
 
@@ -2651,6 +2723,12 @@ func (s *Scope) DropTable(c *Compile) error {
 		}
 	}
 
+	// delete cdc task of the vector and fulltext index here
+	err = DropAllIndexCdcTasks(c, rel.GetTableDef(c.proc.Ctx), qry.Database, qry.Table)
+	if err != nil {
+		return err
+	}
+
 	// delete all index objects record of the table in mo_catalog.mo_indexes
 	if !qry.IsView && qry.Database != catalog.MO_CATALOG && qry.Table != catalog.MO_INDEXES {
 		if qry.GetTableDef().Pkey != nil || len(qry.GetTableDef().Indexes) > 0 {

pkg/sql/compile/ddl_index_algo.go

@@ -24,6 +24,7 @@ import (
 	"github.com/matrixorigin/matrixone/pkg/catalog"
 	"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/pb/api"
 	"github.com/matrixorigin/matrixone/pkg/pb/plan"
 	"github.com/matrixorigin/matrixone/pkg/util/executor"
@@ -140,6 +141,7 @@ func (s *Scope) handleFullTextIndexTable(
 		return moerr.NewInternalErrorNoCtx("FullText index is not enabled")
 	}
 
+	// create hidden tables
 	if indexInfo != nil {
 		if len(indexInfo.GetIndexTables()) != 1 {
 			return moerr.NewInternalErrorNoCtx("index table count not equal to 1")
@@ -152,13 +154,33 @@ func (s *Scope) handleFullTextIndexTable(
 		}
 	}
 
-	insertSQLs := genInsertIndexTableSqlForFullTextIndex(originalTableDef, indexDef, qryDatabase)
+	insertSQLs, err := genInsertIndexTableSqlForFullTextIndex(originalTableDef, indexDef, qryDatabase)
+	if err != nil {
+		return err
+	}
+
 	for _, insertSQL := range insertSQLs {
 		err = c.runSql(insertSQL)
 		if err != nil {
 			return err
 		}
 	}
+
+	async, err := catalog.IsIndexAsync(indexDef.IndexAlgoParams)
+	if err != nil {
+		return err
+	}
+	// create ISCP job for Async fulltext index
+	if async {
+		logutil.Infof("fulltext index Async is true")
+		sinker_type := getSinkerTypeFromAlgo(catalog.MOIndexFullTextAlgo.ToString())
+		err = CreateIndexCdcTask(c, qryDatabase, originalTableDef.Name,
+			indexDef.IndexName, sinker_type)
+		if err != nil {
+			return err
+		}
+	}
+
 	return nil
 }
 
@@ -580,5 +602,12 @@ func (s *Scope) handleVectorHnswIndex(
 		}
 	}
 
+	// 4. register ISCP job for async update
+	sinker_type := getSinkerTypeFromAlgo(catalog.MoIndexHnswAlgo.ToString())
+	err = CreateIndexCdcTask(c, qryDatabase, originalTableDef.Name, indexDefs[catalog.Hnsw_TblType_Metadata].IndexName, sinker_type)
+	if err != nil {
+		return err
+	}
+
 	return nil
 }