Cross-Cluster Physical Replication

.gitignore

@@ -1,5 +1,6 @@
 .idea/
-.kiro
+.kiro/
+.cursor/
 *.o
 *.a
 *.exe

pkg/bootstrap/service.go

@@ -103,6 +103,9 @@ func init() {
 	sql = predefine.GenInitISCPTaskSQL()
 	initSQLs = append(initSQLs, sql)
 
+	sql = predefine.GenInitPublicationTaskSQL()
+	initSQLs = append(initSQLs, sql)
+
 	initSQLs = append(initSQLs, trace.InitSQLs...)
 
 	initSQLs = append(initSQLs, shardservice.InitSQLs...)

pkg/bootstrap/versions/v4_0_0/cluster_upgrade_list.go

@@ -27,6 +27,10 @@ import (
 var clusterUpgEntries = []versions.UpgradeEntry{
 	upg_mo_iscp_log_new,
 	upg_mo_iscp_task,
+	upg_mo_publication_task,
+	upg_mo_ccpr_log_new,
+	upg_mo_ccpr_tables_new,
+	upg_mo_ccpr_dbs_new,
 	upg_mo_index_update_new,
 	upg_create_mo_branch_metadata,
 	upg_rename_system_stmt_info_4000,
@@ -60,6 +64,47 @@ var upg_mo_iscp_task = versions.UpgradeEntry{
 	},
 }
 
+var upg_mo_ccpr_log_new = versions.UpgradeEntry{
+	Schema:    catalog.MO_CATALOG,
+	TableName: catalog.MO_CCPR_LOG,
+	UpgType:   versions.CREATE_NEW_TABLE,
+	UpgSql:    frontend.MoCatalogMoCcprLogDDL,
+	CheckFunc: func(txn executor.TxnExecutor, accountId uint32) (bool, error) {
+		return versions.CheckTableDefinition(txn, accountId, catalog.MO_CATALOG, catalog.MO_CCPR_LOG)
+	},
+}
+
+var upg_mo_ccpr_tables_new = versions.UpgradeEntry{
+	Schema:    catalog.MO_CATALOG,
+	TableName: catalog.MO_CCPR_TABLES,
+	UpgType:   versions.CREATE_NEW_TABLE,
+	UpgSql:    frontend.MoCatalogMoCcprTablesDDL,
+	CheckFunc: func(txn executor.TxnExecutor, accountId uint32) (bool, error) {
+		return versions.CheckTableDefinition(txn, accountId, catalog.MO_CATALOG, catalog.MO_CCPR_TABLES)
+	},
+}
+
+var upg_mo_ccpr_dbs_new = versions.UpgradeEntry{
+	Schema:    catalog.MO_CATALOG,
+	TableName: catalog.MO_CCPR_DBS,
+	UpgType:   versions.CREATE_NEW_TABLE,
+	UpgSql:    frontend.MoCatalogMoCcprDbsDDL,
+	CheckFunc: func(txn executor.TxnExecutor, accountId uint32) (bool, error) {
+		return versions.CheckTableDefinition(txn, accountId, catalog.MO_CATALOG, catalog.MO_CCPR_DBS)
+	},
+}
+
+var upg_mo_publication_task = versions.UpgradeEntry{
+	Schema:    catalog.MOTaskDB,
+	TableName: catalog.MOSysDaemonTask,
+	UpgType:   versions.CREATE_NEW_TABLE,
+	UpgSql:    predefine.GenInitPublicationTaskSQL(),
+	CheckFunc: func(txn executor.TxnExecutor, accountId uint32) (bool, error) {
+		ok, err := versions.CheckTableDataExist(txn, accountId, predefine.GenPublicationTaskCheckSQL())
+		return ok, err
+	},
+}
+
 var upg_mo_index_update_new = versions.UpgradeEntry{
 	Schema:    catalog.MO_CATALOG,
 	TableName: catalog.MO_INDEX_UPDATE,

pkg/catalog/types.go

@@ -29,7 +29,8 @@ import (
 
 const (
 	// for schema
-	PropSchemaExtra = "schema_extra"
+	PropSchemaExtra     = "schema_extra"
+	PropFromPublication = "from_publication"
 
 	Row_ID           = objectio.PhysicalAddr_Attr
 	PrefixPriColName = "__mo_cpkey_"
@@ -174,11 +175,16 @@ const (
 	MO_ISCP_LOG         = "mo_iscp_log"
 	MO_STORED_PROCEDURE = "mo_stored_procedure"
 
+	MO_CCPR_LOG = "mo_ccpr_log"
+
 	MO_INDEX_UPDATE = "mo_index_update"
 
 	MO_BRANCH_METADATA  = "mo_branch_metadata"
 	MO_FEATURE_LIMIT    = "mo_feature_limit"
 	MO_FEATURE_REGISTRY = "mo_feature_registry"
+
+	MO_CCPR_TABLES = "mo_ccpr_tables"
+	MO_CCPR_DBS    = "mo_ccpr_dbs"
 )
 
 func IsSystemTable(id uint64) bool {
@@ -793,6 +799,36 @@ var (
 		types.New(types.T_uuid, 0, 0),                      // segment_id
 		types.New(types.T_TS, 0, 0),                        // flush_point
 	}
+
+	// mo_ccpr_tables schema: tableid (pk), taskid, dbname, tablename, account_id
+	MoCCPRTablesSchema = []string{
+		"tableid",
+		"taskid",
+		"dbname",
+		"tablename",
+		"account_id",
+	}
+	MoCCPRTablesTypes = []types.Type{
+		types.New(types.T_uint64, 0, 0),    // tableid (primary key)
+		types.New(types.T_uuid, 0, 0),      // taskid
+		types.New(types.T_varchar, 256, 0), // dbname
+		types.New(types.T_varchar, 256, 0), // tablename
+		types.New(types.T_uint32, 0, 0),    // account_id
+	}
+
+	// mo_ccpr_dbs schema: dbid (pk), taskid, dbname, account_id
+	MoCCPRDbsSchema = []string{
+		"dbid",
+		"taskid",
+		"dbname",
+		"account_id",
+	}
+	MoCCPRDbsTypes = []types.Type{
+		types.New(types.T_uint64, 0, 0),    // dbid (primary key)
+		types.New(types.T_uuid, 0, 0),      // taskid
+		types.New(types.T_varchar, 256, 0), // dbname
+		types.New(types.T_uint32, 0, 0),    // account_id
+	}
 )
 
 var (

pkg/cnservice/server_task.go

@@ -30,6 +30,7 @@ import (
 	logservicepb "github.com/matrixorigin/matrixone/pkg/pb/logservice"
 	"github.com/matrixorigin/matrixone/pkg/pb/task"
 	"github.com/matrixorigin/matrixone/pkg/proxy"
+	"github.com/matrixorigin/matrixone/pkg/publication"
 	moconnector "github.com/matrixorigin/matrixone/pkg/stream/connector"
 	"github.com/matrixorigin/matrixone/pkg/taskservice"
 	"github.com/matrixorigin/matrixone/pkg/util"
@@ -340,6 +341,17 @@ func (s *service) registerExecutorsLocked() {
 		),
 	)
 
+	s.task.runner.RegisterExecutor(task.TaskCode_PublicationExecutor,
+		publication.PublicationTaskExecutorFactory(
+			s.storeEngine,
+			s._txnClient,
+			s.task.runner.Attach,
+			s.cfg.UUID,
+			common.PublicationAllocator,
+			nil,  // upstreamSQLHelperFactory can be nil for now
+			s.pu, // pass ParameterUnit from service
+		),
+	)
 	s.task.runner.RegisterExecutor(task.TaskCode_IndexUpdateTaskExecutor,
 		idxcron.IndexUpdateTaskExecutorFactory(
 			s.cfg.UUID,

pkg/common/moerr/error.go

@@ -251,6 +251,15 @@ const (
 	// ErrSchedulerClosed scheduler has been closed, cannot schedule new jobs
 	ErrSchedulerClosed uint16 = 20641
 
+	// GC sync protection errors
+	ErrGCIsRunning              uint16 = 20642
+	ErrSyncProtectionNotFound   uint16 = 20643
+	ErrSyncProtectionExists     uint16 = 20644
+	ErrSyncProtectionMaxCount   uint16 = 20645
+	ErrSyncProtectionSoftDelete uint16 = 20646
+	ErrSyncProtectionInvalid    uint16 = 20647
+	ErrSyncProtectionExpired    uint16 = 20648
+
 	// Group 7: lock service
 	// ErrDeadLockDetected lockservice has detected a deadlock and should abort the transaction if it receives this error
 	ErrDeadLockDetected uint16 = 20701
@@ -323,6 +332,8 @@ const (
 
 	// Group 15: Vector Search
 	ErrVectorNeedRetryWithPreMode uint16 = 22301
+	// Group 16: CCPR
+	ErrCCPRReadOnly uint16 = 22401
 
 	// ErrEnd, the max value of MOErrorCode
 	ErrEnd uint16 = 65535
@@ -508,6 +519,15 @@ var errorMsgRefer = map[uint16]moErrorMsgItem{
 	ErrOfflineTxnWrite:            {ER_UNKNOWN_ERROR, []string{MySQLDefaultSqlState}, "write offline txn: %s"},
 	ErrSchedulerClosed:            {ER_UNKNOWN_ERROR, []string{MySQLDefaultSqlState}, "scheduler closed"},
 
+	// GC sync protection errors
+	ErrGCIsRunning:              {ER_UNKNOWN_ERROR, []string{MySQLDefaultSqlState}, "GC is running, please retry later"},
+	ErrSyncProtectionNotFound:   {ER_UNKNOWN_ERROR, []string{MySQLDefaultSqlState}, "sync protection not found: %s"},
+	ErrSyncProtectionExists:     {ER_UNKNOWN_ERROR, []string{MySQLDefaultSqlState}, "sync protection already exists: %s"},
+	ErrSyncProtectionMaxCount:   {ER_UNKNOWN_ERROR, []string{MySQLDefaultSqlState}, "sync protection max count reached: %d"},
+	ErrSyncProtectionSoftDelete: {ER_UNKNOWN_ERROR, []string{MySQLDefaultSqlState}, "sync protection is soft deleted: %s"},
+	ErrSyncProtectionInvalid:    {ER_UNKNOWN_ERROR, []string{MySQLDefaultSqlState}, "invalid sync protection request"},
+	ErrSyncProtectionExpired:    {ER_UNKNOWN_ERROR, []string{MySQLDefaultSqlState}, "sync protection expired: job %s validTS %d < prepareTS %d"},
+
 	// Group 7: lock service
 	ErrDeadLockDetected:        {ER_UNKNOWN_ERROR, []string{MySQLDefaultSqlState}, "deadlock detected"},
 	ErrLockTableBindChanged:    {ER_UNKNOWN_ERROR, []string{MySQLDefaultSqlState}, "lock table bind changed"},
@@ -573,6 +593,9 @@ var errorMsgRefer = map[uint16]moErrorMsgItem{
 	// Group 15: Vector Search
 	ErrVectorNeedRetryWithPreMode: {ER_UNKNOWN_ERROR, []string{MySQLDefaultSqlState}, "vector search need retry with pre mode"},
 
+	// Group 16: CCPR
+	ErrCCPRReadOnly: {ER_UNKNOWN_ERROR, []string{MySQLDefaultSqlState}, "ccpr shared object is read-only"},
+
 	// Group End: max value of MOErrorCode
 	ErrEnd: {ER_UNKNOWN_ERROR, []string{MySQLDefaultSqlState}, "internal error: end of errcode code"},
 }
@@ -1141,6 +1164,17 @@ func IsRPCClientClosed(err error) bool {
 	return IsMoErrCode(err, ErrClientClosed)
 }
 
+// IsSyncProtectionValidationError checks if error is any sync protection validation error.
+// This allows CN to easily distinguish sync protection validation errors from other commit errors.
+func IsSyncProtectionValidationError(err error) bool {
+	if err == nil {
+		return false
+	}
+	return IsMoErrCode(err, ErrSyncProtectionNotFound) ||
+		IsMoErrCode(err, ErrSyncProtectionSoftDelete) ||
+		IsMoErrCode(err, ErrSyncProtectionExpired)
+}
+
 func NewTxnClosed(ctx context.Context, txnID []byte) *Error {
 	id := "unknown"
 	if len(txnID) > 0 {
@@ -1690,6 +1724,10 @@ func NewErrTooBigPrecision(ctx context.Context, precision int32, funcName string
 	return newError(ctx, ErrTooBigPrecision, precision, funcName, maxPrecision)
 }
 
+func NewCCPRReadOnly(ctx context.Context) *Error {
+	return newError(ctx, ErrCCPRReadOnly)
+}
+
 var contextFunc atomic.Value
 
 // noReportCtx is a cached context that suppresses error reporting.

pkg/common/moerr/error_no_ctx.go

@@ -503,3 +503,32 @@ func NewSchedulerClosedNoCtx() *Error {
 func NewVectorNeedRetryWithPreModeNoCtx() *Error {
 	return newError(Context(), ErrVectorNeedRetryWithPreMode)
 }
+
+// GC sync protection errors
+func NewGCIsRunningNoCtx() *Error {
+	return newError(Context(), ErrGCIsRunning)
+}
+
+func NewSyncProtectionNotFoundNoCtx(jobID string) *Error {
+	return newError(Context(), ErrSyncProtectionNotFound, jobID)
+}
+
+func NewSyncProtectionExistsNoCtx(jobID string) *Error {
+	return newError(Context(), ErrSyncProtectionExists, jobID)
+}
+
+func NewSyncProtectionMaxCountNoCtx(maxCount int) *Error {
+	return newError(Context(), ErrSyncProtectionMaxCount, maxCount)
+}
+
+func NewSyncProtectionSoftDeleteNoCtx(jobID string) *Error {
+	return newError(Context(), ErrSyncProtectionSoftDelete, jobID)
+}
+
+func NewSyncProtectionInvalidNoCtx() *Error {
+	return newError(Context(), ErrSyncProtectionInvalid)
+}
+
+func NewSyncProtectionExpiredNoCtx(jobID string, validTS, prepareTS int64) *Error {
+	return newError(Context(), ErrSyncProtectionExpired, jobID, validTS, prepareTS)
+}

pkg/common/moerr/sync_protection_error_test.go

@@ -0,0 +1,51 @@
+// 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 moerr
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestIsSyncProtectionValidationError_Nil(t *testing.T) {
+	assert.False(t, IsSyncProtectionValidationError(nil))
+}
+
+func TestIsSyncProtectionValidationError_NotFound(t *testing.T) {
+	err := NewSyncProtectionNotFoundNoCtx("job1")
+	assert.True(t, IsSyncProtectionValidationError(err))
+}
+
+func TestIsSyncProtectionValidationError_SoftDelete(t *testing.T) {
+	err := NewSyncProtectionSoftDeleteNoCtx("job1")
+	assert.True(t, IsSyncProtectionValidationError(err))
+}
+
+func TestIsSyncProtectionValidationError_Expired(t *testing.T) {
+	err := NewSyncProtectionExpiredNoCtx("job1", 100, 200)
+	assert.True(t, IsSyncProtectionValidationError(err))
+}
+
+func TestIsSyncProtectionValidationError_OtherError(t *testing.T) {
+	err := NewInternalErrorNoCtx("some other error")
+	assert.False(t, IsSyncProtectionValidationError(err))
+}
+
+func TestNewSyncProtectionExpiredNoCtx(t *testing.T) {
+	err := NewSyncProtectionExpiredNoCtx("job1", 100, 200)
+	assert.NotNil(t, err)
+	assert.True(t, IsMoErrCode(err, ErrSyncProtectionExpired))
+}

pkg/defines/type.go

@@ -261,6 +261,10 @@ type IvfReaderParam struct{}
 // PkCheckByTN whether TN does primary key uniqueness check against transaction's workspace or not.
 type PkCheckByTN struct{}
 
+// SkipTransferKey is used to indicate that the delete operation should skip transfer processing.
+// Used by CCPR for cross-cluster tombstones.
+type SkipTransferKey struct{}
+
 // StartTS is the start timestamp of a statement.
 type StartTS struct{}
 

pkg/fileservice/minio_sdk_test.go

@@ -40,7 +40,9 @@ func TestMinioSDK(t *testing.T) {
 			if cmd == nil {
 				t.SkipNow()
 			}
-			defer cmd.Process.Kill()
+			t.Cleanup(func() {
+				cmd.Process.Kill()
+			})
 
 			ret, err := NewMinioSDK(
 				context.Background(),
@@ -87,7 +89,9 @@ func TestMinioSDK(t *testing.T) {
 				true,
 				false,
 			)
-			assert.Nil(t, err)
+			if err != nil {
+				t.Fatal(err)
+			}
 			return fs
 		})
 	})
@@ -120,22 +124,27 @@ func startMinio(dir string) (*exec.Cmd, error) {
 	}
 
 	// create bucket
-	for {
+	var lastErr error
+	for i := 0; i < 30; i++ {
 		endpoint := "localhost:9007"
 		client, err := minio.New(endpoint, &minio.Options{
 			Creds: credentials.NewStaticV4("minioadmin", "minioadmin", ""),
 		})
 		if err != nil {
 			return nil, err
 		}
-		err = client.MakeBucket(context.Background(), "test", minio.MakeBucketOptions{})
-		if err != nil {
-			logutil.Warn("minio error", zap.Any("error", err))
+		lastErr = client.MakeBucket(context.Background(), "test", minio.MakeBucketOptions{})
+		if lastErr != nil {
+			logutil.Warn("minio error", zap.Any("error", lastErr))
 			time.Sleep(time.Second)
 			continue
 		}
 		break
 	}
+	if lastErr != nil {
+		cmd.Process.Kill()
+		return nil, lastErr
+	}
 
 	return cmd, nil
 }